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..ae0ecdffe35d --- /dev/null +++ b/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 { + #[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 { + 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; + } + } +} 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..816d15ec41fb 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; @@ -18,6 +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, 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}; @@ -1675,103 +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 -} - -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")] {