diff --git a/crates/flux-profiler/src/drainer.rs b/crates/flux-profiler/src/drainer.rs index 878bbaf..eb14390 100644 --- a/crates/flux-profiler/src/drainer.rs +++ b/crates/flux-profiler/src/drainer.rs @@ -67,6 +67,9 @@ pub struct ThreadEvents<'a> { pub perf: &'a [PerfSample], pub alloc: &'a [AllocSample], pub loss: Loss, + /// Newest time already written for this thread. A new clock sample can move + /// time back a little, and each thread's events have to stay in order. + pub last_written_ns: u64, } fn split_token(token: &str) -> (&str, u64) { @@ -137,6 +140,7 @@ impl EventsDrainer { perf: &t.events.perf, alloc: &t.events.alloc, loss: t.loss(), + last_written_ns: t.last_written_ns, }) }) } @@ -172,11 +176,13 @@ impl EventsDrainer { perf: &t.events.perf[..n.min(t.events.perf.len())], alloc: &t.events.alloc[..n.min(t.events.alloc.len())], loss, + last_written_ns: t.last_written_ns, } }) }); fxt::write(dumped, &self.meta, &self.clocks, out)?; self.release(); + self.clocks.recalibrate(); Ok(()) } @@ -184,7 +190,7 @@ impl EventsDrainer { /// loss once. pub fn release(&mut self) { for thread in self.threads.values_mut().flatten() { - thread.release_dumped(); + thread.release_dumped(&self.clocks); } } } @@ -258,6 +264,9 @@ struct ThreadDrainer { /// Loss totals already reported by earlier dumps; a dump reports the /// delta so no interval's loss is reported twice. dumped_loss: Loss, + /// Newest time written for this thread, so a new clock sample cannot put a + /// later mark before an earlier one. + last_written_ns: u64, } impl ThreadDrainer { @@ -271,6 +280,7 @@ impl ThreadDrainer { unmatched_closes: 0, expected_seq: 0, dumped_loss: Loss::default(), + last_written_ns: 0, }) } @@ -370,8 +380,12 @@ impl ThreadDrainer { if self.open_ids.is_empty() { self.events.marks.len() } else { self.frame_start } } - fn release_dumped(&mut self) { - self.events.release(self.completed_len()); + fn release_dumped(&mut self, clocks: &SocketClocks) { + let n = self.completed_len(); + if let Some(last) = self.events.marks[..n].last() { + self.last_written_ns = self.last_written_ns.max(clocks.resolve_ns(last.ts)); + } + self.events.release(n); // The in-flight top-level frame's open (if any) is now at index 0. self.frame_start = 0; self.dumped_loss = Loss { missed: self.rings.missed(), dropped: self.unmatched_closes }; diff --git a/crates/flux-profiler/src/fxt.rs b/crates/flux-profiler/src/fxt.rs index 3a0ec2a..0bb071b 100644 --- a/crates/flux-profiler/src/fxt.rs +++ b/crates/flux-profiler/src/fxt.rs @@ -89,7 +89,7 @@ pub(super) fn write<'a>( for (j, mark) in t.marks.iter().enumerate() { let ty = if mark.is_open() { DURATION_BEGIN } else { DURATION_END }; let name = fxt.intern_frame(mark.id, names); - let ts = clocks.resolve_ns(mark.ts); + let ts = clocks.resolve_ns(mark.ts).max(t.last_written_ns); fxt.event(ty, index, name, ts); if let Some(&a) = t.alloc.get(j) && @@ -283,8 +283,16 @@ mod tests { perf: &[PerfSample], schema: Schema, ) -> Vec { - let thread = - ThreadEvents { name: "t", tid: 0, id: 1, marks, alloc, perf, loss: Loss::default() }; + let thread = ThreadEvents { + name: "t", + tid: 0, + id: 1, + marks, + alloc, + perf, + loss: Loss::default(), + last_written_ns: 0, + }; trace( [thread].into_iter(), &FlamegraphMeta { names: names(), schema }, @@ -292,6 +300,37 @@ mod tests { ) } + /// A new clock sample can move time back a little. Each thread keeps its + /// own last written time, so a busy thread cannot drag a quiet one forward. + #[test] + fn marks_never_go_before_last_written() { + let last_written_ns = 5_000; + let quiet = ThreadEvents { + name: "quiet", + tid: 0, + id: 1, + marks: &frames(), + alloc: &[], + perf: &[], + loss: Loss::default(), + last_written_ns, + }; + // A busy thread far ahead in time must not drag the quiet one forward. + let busy = + ThreadEvents { name: "busy", tid: 0, id: 2, last_written_ns: 9_000_000, ..quiet }; + let buf = trace( + [quiet, busy].into_iter(), + &FlamegraphMeta { names: names(), schema: Schema::empty() }, + &SocketClocks::identity(), + ); + + let ts = event_timestamps(&buf); + assert!(!ts.is_empty()); + let quiet_range = last_written_ns..9_000_000; + assert!(ts.iter().any(|&t| quiet_range.contains(&t)), "{ts:?} all dragged forward"); + assert!(ts.iter().all(|&t| t >= last_written_ns), "{ts:?} went before the last written"); + } + #[test] fn alloc_emits_memory_counter() { let buf = render(&frames(), &[alloc(0), alloc(4096)], &[], Schema::empty()); @@ -343,6 +382,7 @@ mod tests { marks: &marks, alloc: &[], perf: &[], + last_written_ns: 0, loss: Loss::default(), }; let buf = trace( diff --git a/crates/flux-profiler/src/socket_clock.rs b/crates/flux-profiler/src/socket_clock.rs index 0a7911c..68ba1ff 100644 --- a/crates/flux-profiler/src/socket_clock.rs +++ b/crates/flux-profiler/src/socket_clock.rs @@ -67,6 +67,18 @@ impl SocketClocks { Self { nodes } } + /// ntpd corrects the wall clock but not the TSC, so an anchor drifts. One + /// sample re-anchors every node: the gaps between sockets' TSCs are fixed + /// at boot, so the drift is shared. + pub fn recalibrate(&mut self) { + let (wall_ns, tsc, node) = sample(); + let now = ((node as u64) << SOCKET_SHIFT) | tsc; + let by = wall_ns as i64 - self.resolve_ns(now) as i64; + for node in &mut self.nodes { + node.wall_ns = node.wall_ns.saturating_add_signed(by); + } + } + pub fn resolve_ns(&self, packed: u64) -> u64 { let node = self.nodes[(packed >> SOCKET_SHIFT) as usize & (MAX_NODES - 1)]; let tsc = packed & TSC_MASK; @@ -112,6 +124,8 @@ mod tests { use super::{MAX_NODES, Node, SocketClocks}; + const SECOND: u64 = 1_000_000_000; + fn packed(node: u64, tsc: u64) -> u64 { (node << SOCKET_SHIFT) | tsc } @@ -139,6 +153,36 @@ mod tests { assert_eq!(adv0, adv1); } + /// The whole point: a clock that has drifted must come back to the wall + /// clock. An earlier version cancelled its own correction and this caught + /// it. + #[test] + fn recalibrate_corrects_a_fast_clock() { + let (wall_ns, tsc, node) = super::sample(); + let mut c = SocketClocks { nodes: [Node { tsc, wall_ns: wall_ns + SECOND }; MAX_NODES] }; + let now = packed(node as u64, tsc); + assert!(c.resolve_ns(now).abs_diff(wall_ns) > SECOND / 2, "the clock starts wrong"); + + c.recalibrate(); + + let (wall_ns, tsc, node) = super::sample(); + let err = c.resolve_ns(packed(node as u64, tsc)).abs_diff(wall_ns); + assert!(err < 1_000_000, "still {err}ns out after recalibrating"); + } + + /// Sockets differ by a fixed amount, so one sample moves them all together. + #[test] + fn recalibrate_keeps_the_gap_between_sockets() { + let (wall_ns, tsc, _) = super::sample(); + let mut nodes = [Node { tsc, wall_ns: wall_ns + SECOND }; MAX_NODES]; + nodes[1].wall_ns += 500; + let mut c = SocketClocks { nodes }; + + c.recalibrate(); + + assert_eq!(c.nodes[1].wall_ns - c.nodes[0].wall_ns, 500); + } + #[test] fn calibrate_wall_clock() { let before = Nanos::now().0; diff --git a/crates/flux-timing/src/duration.rs b/crates/flux-timing/src/duration.rs index 33bb5df..18e0e2b 100644 --- a/crates/flux-timing/src/duration.rs +++ b/crates/flux-timing/src/duration.rs @@ -5,7 +5,7 @@ use type_hash_derive::TypeHash; use crate::{ Nanos, - global_clock::{global_clock_not_mocked, ticks_per_micro, ticks_per_milli, ticks_per_sec}, + global_clock::{nanos_to_ticks, ticks_to_nanos}, }; #[derive(Copy, Clone, Debug, Default, Serialize, Deserialize, TypeHash)] @@ -35,13 +35,11 @@ impl Duration { Self(self.0.saturating_add(rhs.0)) } - /// Overflows at ~182 years. #[inline] pub fn from_secs(s: u64) -> Self { - Self(s * ticks_per_sec()) + Self(nanos_to_ticks(s * 1_000_000_000)) } - /// Overflows at ~182 years. #[inline] pub fn from_mins(s: u64) -> Self { Self::from_secs(s * 60) @@ -52,68 +50,70 @@ impl Duration { Self::from_nanos((s * 1_000_000_000.0).round() as u64) } - /// Overflows at ~182 years. #[inline] pub fn from_millis(s: u64) -> Self { - Self(s * ticks_per_milli()) + Self(nanos_to_ticks(s * 1_000_000)) } - /// Overflows at ~66 days. #[inline] pub fn from_micros(s: u64) -> Self { - Self(s * ticks_per_milli() / 1_000) + Self(nanos_to_ticks(s * 1_000)) } - /// Overflows at ~66 days. #[inline] pub fn from_nanos(s: u64) -> Self { - Self(s * ticks_per_micro() / 1000) + Self(nanos_to_ticks(s)) + } + + #[inline] + fn nanos(self) -> u64 { + ticks_to_nanos(self.0) } #[inline] pub fn as_secs(&self) -> f64 { - self.0 as f64 / ticks_per_sec() as f64 + self.nanos() as f64 / 1e9 } #[inline] pub fn as_secs_u64(&self) -> u64 { - self.0 / ticks_per_sec() + self.nanos() / 1_000_000_000 } #[inline] pub fn as_millis(&self) -> f64 { - self.0 as f64 / ticks_per_milli() as f64 + self.nanos() as f64 / 1e6 } #[inline] pub fn as_millis_u64(&self) -> u64 { - self.0 / ticks_per_milli() + self.nanos() / 1_000_000 } #[inline] pub fn as_micros(&self) -> f64 { - self.0 as f64 * 1_000.0 / ticks_per_milli() as f64 + self.nanos() as f64 / 1e3 } #[inline] pub fn as_micros_u64(&self) -> u64 { - self.0 / ticks_per_micro() + self.nanos() / 1_000 } #[inline] pub fn as_micros_u128(&self) -> u128 { - (self.0 / ticks_per_micro()) as u128 + u128::from(self.as_micros_u64()) } #[inline] pub fn as_nanos(&self) -> f64 { - self.0 as f64 * 1000.0 / ticks_per_micro() as f64 + self.nanos() as f64 } } impl std::fmt::Display for Duration { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - Nanos(global_clock_not_mocked().delta_as_nanos(0, self.0)).fmt(f) + Nanos(self.nanos()).fmt(f) } } @@ -374,7 +374,7 @@ impl From for i64 { impl From for std::time::Duration { #[inline] fn from(value: Duration) -> Self { - Self::from_nanos(global_clock_not_mocked().delta_as_nanos(0, value.0)) + Self::from_nanos(value.nanos()) } } @@ -386,10 +386,9 @@ impl From for Duration { } impl From for Duration { - /// Overflows at ~66 days. #[inline] fn from(value: Nanos) -> Self { - Self(value.0 * ticks_per_micro() / 1000) + Self(nanos_to_ticks(value.0)) } } @@ -431,20 +430,31 @@ mod tests { std::time::Duration::from(d).as_nanos() } - /// Assert drift < 500ppm of expected, minimum 10ns absolute. - /// `from_nanos` path uses `ticks_per_micro` (~3200) so truncation can reach - /// ~200ppm. + /// Assert drift < 1ppm of expected, minimum 10ns absolute. fn check(ours: Duration, expected: std::time::Duration) { let actual_ns = to_ns(ours); let expected_ns = expected.as_nanos(); let diff = (actual_ns as i128 - expected_ns as i128).unsigned_abs(); - let max_err = (expected_ns / 2_000).max(10); // 500ppm or 10ns + let max_err = (expected_ns / 1_000_000).max(10); // 1ppm or 10ns assert!( diff <= max_err, - "drift {diff}ns > {max_err}ns (500ppm): actual={actual_ns}ns expected={expected_ns}ns" + "drift {diff}ns > {max_err}ns (1ppm): actual={actual_ns}ns expected={expected_ns}ns" ); } + #[test] + fn as_nanos_scale_within_1ppm() { + for secs in [1u64, 60, 3600, 86_400, 604_800] { + let d = Duration::from_secs(secs); + let expected = secs as f64 * 1e9; + let ppm = (d.as_nanos() / expected - 1.0).abs() * 1e6; + assert!(ppm < 1.0, "{secs}s: as_nanos off by {ppm:.4}ppm ({})", d.as_nanos()); + // one truncated microsecond on top of the 1ppm scale budget + let drift = (secs * 1_000_000).abs_diff(d.as_micros_u64()); + assert!(drift <= 1 + secs, "{secs}s: as_micros_u64 off by {drift}us"); + } + } + #[test] fn from_secs_matches_std() { for s in [0, 1, 5, 60, 3600, 86400, 604_800] { diff --git a/crates/flux-timing/src/global_clock.rs b/crates/flux-timing/src/global_clock.rs index d456369..224e98f 100644 --- a/crates/flux-timing/src/global_clock.rs +++ b/crates/flux-timing/src/global_clock.rs @@ -40,7 +40,6 @@ impl GovernorClock for OurClockForNanos { } } -static GLOBAL_NANOS_FOR_MULTIPLIER: OnceLock = OnceLock::new(); // might be mocked static GLOBAL_CLOCK: OnceLock = OnceLock::new(); // never mocked @@ -67,32 +66,49 @@ pub fn global_clock_not_mocked() -> &'static Clock { GLOBAL_CLOCK_NON_MOCKED.get_or_init(Clock::new) } -const MULTIPLIER: u64 = 100_000_000; +static NANOS_PER_TICK: OnceLock = OnceLock::new(); +static TICKS_PER_NANO: OnceLock = OnceLock::new(); +/// A tick is a fraction of a nanosecond - 0.2330078 of one at 4.3GHz - and an +/// integer cannot hold that, so the rate is kept multiplied by +/// `2^FRACTION_BITS` and shifted back down after each conversion. Whatever is +/// too small to fit in those bits is lost, and a lost fraction is a clock that +/// runs slow: +/// +/// whole number (old `ticks_per_micro`: 3792 of a true 3792.87) 20 s/day +/// 16 bits 5.7 s/day +/// 24 bits 22 ms/day +/// 32 bits 86 us/day +const FRACTION_BITS: u32 = 32; +const ONE: u128 = 1 << FRACTION_BITS; -fn nanos_for_multiplier() -> u64 { - *GLOBAL_NANOS_FOR_MULTIPLIER - .get_or_init(|| global_clock_not_mocked().delta_as_nanos(0, MULTIPLIER)) +#[inline] +fn nanos_per_tick() -> u64 { + *NANOS_PER_TICK.get_or_init(|| global_clock_not_mocked().delta_as_nanos(0, ONE as u64)) } -static TICKS_PER_SEC: OnceLock = OnceLock::new(); -static TICKS_PER_MILLI: OnceLock = OnceLock::new(); -static TICKS_PER_MICRO: OnceLock = OnceLock::new(); +/// Rates are stored times `ONE`, so `nanos_per_tick()` holds `0.233 * ONE`. +/// The reverse rate has to come out stored the same way: +/// +/// ```text +/// ONE * ONE / (0.233 * ONE) = 4.29 * ONE +/// ``` +#[inline] +fn ticks_per_nano() -> u64 { + *TICKS_PER_NANO.get_or_init(|| (ONE * ONE / u128::from(nanos_per_tick())) as u64) +} -/// Overflow: `s * ticks_per_sec()` wraps at ~182 years. #[inline] -pub(super) fn ticks_per_sec() -> u64 { - *TICKS_PER_SEC.get_or_init(|| 1_000_000_000 * MULTIPLIER / nanos_for_multiplier()) +fn scale(value: u64, rate: u64) -> u64 { + ((u128::from(value) * u128::from(rate)) >> FRACTION_BITS) as u64 } -/// Overflow: `ms * ticks_per_milli()` wraps at ~182 years. #[inline] -pub(super) fn ticks_per_milli() -> u64 { - *TICKS_PER_MILLI.get_or_init(|| 1_000_000 * MULTIPLIER / nanos_for_multiplier()) +pub(super) fn ticks_to_nanos(ticks: u64) -> u64 { + scale(ticks, nanos_per_tick()) } -/// Overflow: `us * ticks_per_micro()` wraps at ~182 years. -/// For nanos: `ns * ticks_per_micro() / 1000` wraps at ~66 days. +/// A u64 of ticks runs out after ~136 years at 4GHz, and so does this. #[inline] -pub(super) fn ticks_per_micro() -> u64 { - *TICKS_PER_MICRO.get_or_init(|| 1_000 * MULTIPLIER / nanos_for_multiplier()) +pub(super) fn nanos_to_ticks(nanos: u64) -> u64 { + scale(nanos, ticks_per_nano()) } diff --git a/crates/flux-timing/src/instant.rs b/crates/flux-timing/src/instant.rs index 1e64c77..12e8ce7 100644 --- a/crates/flux-timing/src/instant.rs +++ b/crates/flux-timing/src/instant.rs @@ -5,7 +5,7 @@ use type_hash_derive::TypeHash; use crate::{ Duration, Nanos, - global_clock::{global_clock_not_mocked, ticks_per_micro}, + global_clock::{global_clock_not_mocked, nanos_to_ticks, ticks_to_nanos}, }; pub const SOCKET_SHIFT: u32 = 62; @@ -64,7 +64,7 @@ impl Instant { #[inline] pub fn as_delta_nanos(&self) -> Nanos { - Nanos(global_clock_not_mocked().delta_as_nanos(0, self.remove_socket().0)) + Nanos(ticks_to_nanos(self.remove_socket().0)) } #[inline] @@ -99,7 +99,7 @@ impl Add for Instant { type Output = Self; fn add(self, rhs: Nanos) -> Self::Output { - Self(self.0 + rhs.0 * ticks_per_micro() / 1000) + Self(self.0 + nanos_to_ticks(rhs.0)) } } @@ -107,7 +107,7 @@ impl Sub for Instant { type Output = Self; fn sub(self, rhs: Nanos) -> Self::Output { - Self(self.0.saturating_sub(rhs.0 * ticks_per_micro() / 1000)) + Self(self.0.saturating_sub(nanos_to_ticks(rhs.0))) } } diff --git a/crates/flux-timing/src/nanos.rs b/crates/flux-timing/src/nanos.rs index a8643b8..876a394 100644 --- a/crates/flux-timing/src/nanos.rs +++ b/crates/flux-timing/src/nanos.rs @@ -12,7 +12,7 @@ use type_hash_derive::TypeHash; use crate::{ Duration, - global_clock::{global_clock, global_clock_not_mocked}, + global_clock::{global_clock, ticks_to_nanos}, }; /// Nanos since unix epoch, good till 2554 or so @@ -480,7 +480,7 @@ impl FromStr for Nanos { impl From for Nanos { #[inline] fn from(value: Duration) -> Self { - Self(global_clock_not_mocked().delta_as_nanos(0, value.0)) + Self(ticks_to_nanos(value.0)) } }