refactor(bench): share the KV-router replay deadline clock

lib/bench/kv_router/active_sequences_open_loop.rs

@@ -6,8 +6,9 @@ use std::collections::{BTreeMap, HashMap};
 use std::hint::black_box;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, mpsc};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
+use dynamo_bench::kv_router_common::clock::BenchmarkClock;
 use dynamo_bench::kv_router_common::issuer::pin_current_thread;
 use dynamo_bench::kv_router_common::replay::NoopSequencePublisher;
 use dynamo_bench::kv_router_common::trace_gen::WorkerTimelines;
@@ -1608,69 +1609,6 @@ fn nearest_rank(values: &[u64], numerator: usize, denominator: usize) -> u64 {
     values[rank.saturating_sub(1).min(values.len() - 1)]
 }
 
-struct BenchmarkClock {
-    epoch: Instant,
-    spin_ns: u64,
-    #[cfg(target_os = "linux")]
-    monotonic_epoch_ns: u64,
-}
-
-impl BenchmarkClock {
-    fn new(spin_ns: u64) -> anyhow::Result<Self> {
-        #[cfg(target_os = "linux")]
-        let monotonic_epoch_ns = monotonic_now_ns()?;
-        Ok(Self {
-            epoch: Instant::now(),
-            spin_ns,
-            #[cfg(target_os = "linux")]
-            monotonic_epoch_ns,
-        })
-    }
-
-    fn now_ns(&self) -> u64 {
-        self.epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64
-    }
-
-    fn wait_until(&self, target_ns: u64) {
-        let sleep_target = target_ns.saturating_sub(self.spin_ns);
-        let now = self.now_ns();
-        #[cfg(target_os = "linux")]
-        if sleep_target > now {
-            sleep_until_monotonic(self.monotonic_epoch_ns.saturating_add(sleep_target));
-        }
-        #[cfg(not(target_os = "linux"))]
-        if sleep_target > now {
-            std::thread::sleep(Duration::from_nanos(sleep_target - now));
-        }
-        while self.now_ns() < target_ns {
-            std::hint::spin_loop();
-        }
-    }
-
-    fn timer_name(&self) -> &'static str {
-        if cfg!(target_os = "linux") {
-            "clock_nanosleep_monotonic_absolute"
-        } else {
-            "portable_sleep_spin_non_authoritative"
-        }
-    }
-}
-
-#[cfg(target_os = "linux")]
-fn monotonic_now_ns() -> anyhow::Result<u64> {
-    let mut timestamp = libc::timespec {
-        tv_sec: 0,
-        tv_nsec: 0,
-    };
-    let rc = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut timestamp) };
-    if rc != 0 {
-        return Err(std::io::Error::last_os_error().into());
-    }
-    Ok((timestamp.tv_sec as u64)
-        .saturating_mul(1_000_000_000)
-        .saturating_add(timestamp.tv_nsec as u64))
-}
-
 fn process_cpu_time_ns() -> Option<u64> {
     cpu_time_ns(CpuClock::Process)
 }
@@ -1707,30 +1645,6 @@ fn cpu_time_ns(_clock: CpuClock) -> Option<u64> {
     None
 }
 
-#[cfg(target_os = "linux")]
-fn sleep_until_monotonic(target_ns: u64) {
-    let request = libc::timespec {
-        tv_sec: (target_ns / 1_000_000_000) as libc::time_t,
-        tv_nsec: (target_ns % 1_000_000_000) as libc::c_long,
-    };
-    loop {
-        let rc = unsafe {
-            libc::clock_nanosleep(
-                libc::CLOCK_MONOTONIC,
-                libc::TIMER_ABSTIME,
-                &request,
-                std::ptr::null_mut(),
-            )
-        };
-        if rc == 0 {
-            return;
-        }
-        if rc != libc::EINTR {
-            return;
-        }
-    }
-}
-
 #[cfg(test)]
 #[allow(dead_code, unused_imports)]
 mod tests {

lib/bench/kv_router/approximate_lru_bench.rs

@@ -5,11 +5,12 @@ use std::collections::{HashMap, HashSet};
 use std::hint::black_box;
 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, mpsc};
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use anyhow::Context;
 use clap::{Parser, ValueEnum};
 use dynamo_bench::kv_router_common::args::CommonArgs;
+use dynamo_bench::kv_router_common::clock::BenchmarkClock;
 use dynamo_bench::kv_router_common::issuer::{
     parse_cpu_list, pin_current_thread, pin_current_thread_to_cpus,
 };
@@ -1161,54 +1162,6 @@ fn push_failure(failures: &mut Vec<String>, failure: &str) {
     }
 }
 
-struct BenchmarkClock {
-    epoch: Instant,
-    spin_ns: u64,
-    #[cfg(target_os = "linux")]
-    monotonic_epoch_ns: u64,
-}
-
-impl BenchmarkClock {
-    fn new(spin_ns: u64) -> anyhow::Result<Self> {
-        #[cfg(target_os = "linux")]
-        let monotonic_epoch_ns = clock_time_ns(libc::CLOCK_MONOTONIC)?;
-        Ok(Self {
-            epoch: Instant::now(),
-            spin_ns,
-            #[cfg(target_os = "linux")]
-            monotonic_epoch_ns,
-        })
-    }
-
-    fn now_ns(&self) -> u64 {
-        self.epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64
-    }
-
-    fn wait_until(&self, target_ns: u64) {
-        let sleep_target = target_ns.saturating_sub(self.spin_ns);
-        let now = self.now_ns();
-        #[cfg(target_os = "linux")]
-        if sleep_target > now {
-            sleep_until_monotonic(self.monotonic_epoch_ns.saturating_add(sleep_target));
-        }
-        #[cfg(not(target_os = "linux"))]
-        if sleep_target > now {
-            std::thread::sleep(Duration::from_nanos(sleep_target - now));
-        }
-        while self.now_ns() < target_ns {
-            std::hint::spin_loop();
-        }
-    }
-
-    fn timer_name(&self) -> &'static str {
-        if cfg!(target_os = "linux") {
-            "clock_nanosleep_monotonic_absolute"
-        } else {
-            "portable_sleep_spin_non_authoritative"
-        }
-    }
-}
-
 #[cfg(target_os = "linux")]
 fn clock_time_ns(clock_id: libc::clockid_t) -> anyhow::Result<u64> {
     let mut timestamp = libc::timespec {
@@ -1244,27 +1197,6 @@ fn thread_cpu_time_ns() -> Option<u64> {
     None
 }
 
-#[cfg(target_os = "linux")]
-fn sleep_until_monotonic(target_ns: u64) {
-    let request = libc::timespec {
-        tv_sec: (target_ns / 1_000_000_000) as libc::time_t,
-        tv_nsec: (target_ns % 1_000_000_000) as libc::c_long,
-    };
-    loop {
-        let rc = unsafe {
-            libc::clock_nanosleep(
-                libc::CLOCK_MONOTONIC,
-                libc::TIMER_ABSTIME,
-                &request,
-                std::ptr::null_mut(),
-            )
-        };
-        if rc == 0 || rc != libc::EINTR {
-            return;
-        }
-    }
-}
-
 #[cfg(test)]
 #[allow(dead_code)]
 mod tests {

lib/bench/kv_router/common/clock.rs

@@ -0,0 +1,106 @@
+// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+//! Deadline clock shared by the KV-router open-loop replay benchmarks.
+//!
+//! On Linux, `wait_until` sleeps to an absolute `CLOCK_MONOTONIC` deadline and
+//! then spins for the final `spin_ns`. Elsewhere it falls back to a relative
+//! sleep plus spin, which is not authoritative for timing results.
+
+use std::time::Instant;
+
+pub struct BenchmarkClock {
+    epoch: Instant,
+    spin_ns: u64,
+    #[cfg(target_os = "linux")]
+    monotonic_epoch_ns: u64,
+}
+
+impl BenchmarkClock {
+    pub fn new(spin_ns: u64) -> anyhow::Result<Self> {
+        #[cfg(target_os = "linux")]
+        let monotonic_epoch_ns = monotonic_now_ns()?;
+        Ok(Self {
+            epoch: Instant::now(),
+            spin_ns,
+            #[cfg(target_os = "linux")]
+            monotonic_epoch_ns,
+        })
+    }
+
+    #[inline]
+    pub fn epoch(&self) -> Instant {
+        self.epoch
+    }
+
+    #[inline]
+    pub fn now_ns(&self) -> u64 {
+        elapsed_ns(self.epoch)
+    }
+
+    #[inline]
+    pub fn wait_until(&self, target_ns: u64) {
+        let sleep_target = target_ns.saturating_sub(self.spin_ns);
+        let now = self.now_ns();
+        #[cfg(target_os = "linux")]
+        if sleep_target > now {
+            sleep_until_monotonic(self.monotonic_epoch_ns.saturating_add(sleep_target));
+        }
+        #[cfg(not(target_os = "linux"))]
+        if sleep_target > now {
+            std::thread::sleep(std::time::Duration::from_nanos(sleep_target - now));
+        }
+        while self.now_ns() < target_ns {
+            std::hint::spin_loop();
+        }
+    }
+
+    pub fn timer_name(&self) -> &'static str {
+        if cfg!(target_os = "linux") {
+            "clock_nanosleep_monotonic_absolute"
+        } else {
+            "portable_sleep_spin_non_authoritative"
+        }
+    }
+}
+
+#[inline]
+pub fn elapsed_ns(epoch: Instant) -> u64 {
+    epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64
+}
+
+#[cfg(target_os = "linux")]
+fn monotonic_now_ns() -> anyhow::Result<u64> {
+    let mut timestamp = libc::timespec {
+        tv_sec: 0,
+        tv_nsec: 0,
+    };
+    let rc = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut timestamp) };
+    if rc != 0 {
+        return Err(std::io::Error::last_os_error().into());
+    }
+    Ok((timestamp.tv_sec as u64)
+        .saturating_mul(1_000_000_000)
+        .saturating_add(timestamp.tv_nsec as u64))
+}
+
+#[cfg(target_os = "linux")]
+fn sleep_until_monotonic(target_ns: u64) {
+    let request = libc::timespec {
+        tv_sec: (target_ns / 1_000_000_000) as libc::time_t,
+        tv_nsec: (target_ns % 1_000_000_000) as libc::c_long,
+    };
+    loop {
+        let rc = unsafe {
+            libc::clock_nanosleep(
+                libc::CLOCK_MONOTONIC,
+                libc::TIMER_ABSTIME,
+                &request,
+                std::ptr::null_mut(),
+            )
+        };
+        if rc != libc::EINTR {
+            return;
+        }
+    }
+}

lib/bench/kv_router/common/mod.rs

@@ -7,6 +7,12 @@
     feature = "router-test-support"
 ))]
 pub mod args;
+#[cfg(any(
+    feature = "active-sequences",
+    feature = "mooncake",
+    feature = "router-test-support"
+))]
+pub mod clock;
 #[cfg(any(feature = "dc-ckf-consumer", feature = "dc-ckf-relay"))]
 pub mod dc_ckf_metadata;
 #[cfg(feature = "dc-ckf-relay")]