Eliminate long running perf collector clock drift

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,19 +176,21 @@ 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(())
     }
 
     /// Loss stays a per-interval delta, so an accumulating caller sees each
     /// 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 };

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,15 +283,54 @@ mod tests {
         perf: &[PerfSample],
         schema: Schema,
     ) -> Vec<u8> {
-        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 },
             &SocketClocks::identity(),
         )
     }
 
+    /// 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(

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;

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<Duration> for i64 {
 impl From<Duration> 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<std::time::Duration> for Duration {
 }
 
 impl From<Nanos> 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] {