From 914c8d6b1e4bb869c6db65d824ddd486bfbd8437 Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Tue, 3 Nov 2026 00:02:09 +0000 Subject: [PATCH 1/3] refactor(bench): share the KV-router replay deadline clock Move the duplicated BenchmarkClock and its Linux absolute CLOCK_MONOTONIC sleep from the active-sequences, approximate-LRU and Mooncake replay benchmarks into kv_router/common/clock.rs. Timing behavior is unchanged. Refs DYN-5211 Signed-off-by: svc-glamr@nvidia.com --- .../kv_router/active_sequences_open_loop.rs | 90 +----------- lib/bench/kv_router/approximate_lru_bench.rs | 72 +--------- lib/bench/kv_router/common/clock.rs | 135 ++++++++++++++++++ lib/bench/kv_router/common/mod.rs | 6 + lib/bench/kv_router/mooncake_open_loop.rs | 94 +----------- 5 files changed, 146 insertions(+), 251 deletions(-) create mode 100644 lib/bench/kv_router/common/clock.rs diff --git a/lib/bench/kv_router/active_sequences_open_loop.rs b/lib/bench/kv_router/active_sequences_open_loop.rs index baa9335d62d0..ab33da90b2b1 100644 --- a/lib/bench/kv_router/active_sequences_open_loop.rs +++ b/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 { - #[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 { - 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 { cpu_time_ns(CpuClock::Process) } @@ -1707,30 +1645,6 @@ fn cpu_time_ns(_clock: CpuClock) -> Option { 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 { diff --git a/lib/bench/kv_router/approximate_lru_bench.rs b/lib/bench/kv_router/approximate_lru_bench.rs index 740b24e7c5dc..0dfae4bb4aaf 100644 --- a/lib/bench/kv_router/approximate_lru_bench.rs +++ b/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, 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 { - #[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 { let mut timestamp = libc::timespec { @@ -1244,27 +1197,6 @@ fn thread_cpu_time_ns() -> Option { 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 { diff --git a/lib/bench/kv_router/common/clock.rs b/lib/bench/kv_router/common/clock.rs new file mode 100644 index 000000000000..6a164c91e14d --- /dev/null +++ b/lib/bench/kv_router/common/clock.rs @@ -0,0 +1,135 @@ +// 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 { + #[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 + } + + // `now_ns` and `wait_until` run on the issuer's spin path from other + // crates (bench and test binaries); `#[inline]` keeps them inlinable there. + #[inline] + pub fn now_ns(&self) -> u64 { + self.epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64 + } + + #[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" + } + } +} + +#[cfg(target_os = "linux")] +fn monotonic_now_ns() -> anyhow::Result { + 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 == 0 { + return; + } + if rc != libc::EINTR { + return; + } + } +} + +#[cfg(test)] +mod tests { + use super::BenchmarkClock; + + // Only lower bounds are asserted; upper bounds are flaky on shared CI. + #[test] + fn wait_until_never_returns_before_target() { + let spin_ns = 200_000; + let clock = BenchmarkClock::new(spin_ns).unwrap(); + + // Past the spin window: sleeps, then spins. + let sleep_target = clock.now_ns() + 5_000_000; + clock.wait_until(sleep_target); + let after_sleep = clock.now_ns(); + assert!(after_sleep >= sleep_target); + + // Inside the spin window: spins only. + let spin_target = clock.now_ns() + spin_ns / 2; + clock.wait_until(spin_target); + let after_spin = clock.now_ns(); + assert!(after_spin >= spin_target); + assert!(after_spin >= after_sleep); + + // A target already in the past returns without waiting. + clock.wait_until(0); + assert!(clock.now_ns() >= after_spin); + } +} diff --git a/lib/bench/kv_router/common/mod.rs b/lib/bench/kv_router/common/mod.rs index 371924b72a50..d5bf16cc28c8 100644 --- a/lib/bench/kv_router/common/mod.rs +++ b/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")] diff --git a/lib/bench/kv_router/mooncake_open_loop.rs b/lib/bench/kv_router/mooncake_open_loop.rs index 1693ce5a2c72..12d1b676b17c 100644 --- a/lib/bench/kv_router/mooncake_open_loop.rs +++ b/lib/bench/kv_router/mooncake_open_loop.rs @@ -18,6 +18,7 @@ use dynamo_kv_router::protocols::{KvCacheEventData, RouterEvent, StorageTier}; use serde::Serialize; use tokio::sync::{Notify, oneshot}; +use dynamo_bench::kv_router_common::clock::BenchmarkClock; pub use dynamo_bench::kv_router_common::issuer::parse_cpu_list; use dynamo_bench::kv_router_common::issuer::{contiguous_worker_issuer, pin_current_thread}; @@ -1679,99 +1680,6 @@ fn elapsed_ns(epoch: Instant) -> u64 { epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64 } -struct BenchmarkClock { - epoch: Instant, - spin_ns: u64, - #[cfg(target_os = "linux")] - monotonic_epoch_ns: u64, -} - -impl BenchmarkClock { - fn new(spin_ns: u64) -> anyhow::Result { - #[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 epoch(&self) -> Instant { - self.epoch - } - - fn now_ns(&self) -> u64 { - elapsed_ns(self.epoch) - } - - 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 { - 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 == 0 { - return; - } - if rc != libc::EINTR { - return; - } - } -} - fn thread_cpu_time_ns() -> u64 { #[cfg(target_os = "linux")] { From 1177f1d392b756e3510685b7ad9638eb0aa955d1 Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Tue, 3 Nov 2026 00:40:46 +0000 Subject: [PATCH 2/3] fix(bench): drop the unused non-Linux Duration import in the Mooncake bench The shared BenchmarkClock took over the only non-Linux use of Duration in mooncake_open_loop.rs, so the cfg-gated import is now unused on every platform. Refs DYN-5211 Signed-off-by: svc-glamr@nvidia.com --- lib/bench/kv_router/mooncake_open_loop.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/lib/bench/kv_router/mooncake_open_loop.rs b/lib/bench/kv_router/mooncake_open_loop.rs index 12d1b676b17c..ff860536d94e 100644 --- a/lib/bench/kv_router/mooncake_open_loop.rs +++ b/lib/bench/kv_router/mooncake_open_loop.rs @@ -5,8 +5,6 @@ use std::collections::BTreeMap; use std::hint::black_box; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Barrier}; -#[cfg(not(target_os = "linux"))] -use std::time::Duration; use std::time::Instant; use dynamo_kv_router::LocalBlockHash; From ee45f5461b34e8a1b7752fcddbdca34e8019943c Mon Sep 17 00:00:00 2001 From: GLAMR Date: Wed, 4 Nov 2026 19:01:26 +0000 Subject: [PATCH 3/3] refactor(bench): address shared deadline clock review Signed-off-by: GLAMR --- lib/bench/kv_router/common/clock.rs | 41 ++++------------------- lib/bench/kv_router/mooncake_open_loop.rs | 6 +--- 2 files changed, 7 insertions(+), 40 deletions(-) diff --git a/lib/bench/kv_router/common/clock.rs b/lib/bench/kv_router/common/clock.rs index 6a164c91e14d..ae0ecdffe35d 100644 --- a/lib/bench/kv_router/common/clock.rs +++ b/lib/bench/kv_router/common/clock.rs @@ -33,11 +33,9 @@ impl BenchmarkClock { self.epoch } - // `now_ns` and `wait_until` run on the issuer's spin path from other - // crates (bench and test binaries); `#[inline]` keeps them inlinable there. #[inline] pub fn now_ns(&self) -> u64 { - self.epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64 + elapsed_ns(self.epoch) } #[inline] @@ -66,6 +64,11 @@ impl BenchmarkClock { } } +#[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 { let mut timestamp = libc::timespec { @@ -96,40 +99,8 @@ fn sleep_until_monotonic(target_ns: u64) { std::ptr::null_mut(), ) }; - if rc == 0 { - return; - } if rc != libc::EINTR { return; } } } - -#[cfg(test)] -mod tests { - use super::BenchmarkClock; - - // Only lower bounds are asserted; upper bounds are flaky on shared CI. - #[test] - fn wait_until_never_returns_before_target() { - let spin_ns = 200_000; - let clock = BenchmarkClock::new(spin_ns).unwrap(); - - // Past the spin window: sleeps, then spins. - let sleep_target = clock.now_ns() + 5_000_000; - clock.wait_until(sleep_target); - let after_sleep = clock.now_ns(); - assert!(after_sleep >= sleep_target); - - // Inside the spin window: spins only. - let spin_target = clock.now_ns() + spin_ns / 2; - clock.wait_until(spin_target); - let after_spin = clock.now_ns(); - assert!(after_spin >= spin_target); - assert!(after_spin >= after_sleep); - - // A target already in the past returns without waiting. - clock.wait_until(0); - assert!(clock.now_ns() >= after_spin); - } -} diff --git a/lib/bench/kv_router/mooncake_open_loop.rs b/lib/bench/kv_router/mooncake_open_loop.rs index ff860536d94e..816d15ec41fb 100644 --- a/lib/bench/kv_router/mooncake_open_loop.rs +++ b/lib/bench/kv_router/mooncake_open_loop.rs @@ -16,7 +16,7 @@ use dynamo_kv_router::protocols::{KvCacheEventData, RouterEvent, StorageTier}; use serde::Serialize; use tokio::sync::{Notify, oneshot}; -use dynamo_bench::kv_router_common::clock::BenchmarkClock; +use dynamo_bench::kv_router_common::clock::{BenchmarkClock, elapsed_ns}; pub use dynamo_bench::kv_router_common::issuer::parse_cpu_list; use dynamo_bench::kv_router_common::issuer::{contiguous_worker_issuer, pin_current_thread}; @@ -1674,10 +1674,6 @@ fn nearest_rank(values: &[u64], numerator: usize, denominator: usize) -> u64 { values[rank.saturating_sub(1).min(values.len() - 1)] } -fn elapsed_ns(epoch: Instant) -> u64 { - epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64 -} - fn thread_cpu_time_ns() -> u64 { #[cfg(target_os = "linux")] {