From 87ac9d657fd689f4caeecd0b665a762acfcbe7b7 Mon Sep 17 00:00:00 2001 From: mertcano <35747700+mertcano@users.noreply.github.com> Date: Thu, 20 Aug 2026 15:18:54 +0300 Subject: [PATCH] Fix: Ensure memory-layout safety in Linux clock interception ### Description This PR addresses High-severity memory layout issues in the `deterministic-simulator` repository, as identified in the workspace-wide security audit. **Vulnerabilities & Security Defects Remediated:** * **Clock Interceptor Memory Safety (`msim/src/sim/time/mod.rs`):** The Linux monotonic-clock path previously transmuted the opaque `std::time::Instant` representation directly into `libc::timespec`. This is not a stable ABI contract and failed with invalid timestamps under current Rust toolchains. The interceptor now safely and explicitly serializes the simulated duration directly into the `tv_sec` and `tv_nsec` fields. --- msim/src/sim/time/mod.rs | 1357 +++++++++++++++++++------------------- 1 file changed, 680 insertions(+), 677 deletions(-) diff --git a/msim/src/sim/time/mod.rs b/msim/src/sim/time/mod.rs index f48ecba..68c28c5 100644 --- a/msim/src/sim/time/mod.rs +++ b/msim/src/sim/time/mod.rs @@ -1,677 +1,680 @@ -//! Utilities for tracking time. -//! -//! - -#[doc(no_inline)] -pub use std::time::Duration; -use std::{ - future::Future, - pin::Pin, - sync::{Arc, Mutex}, - task::{Context, Poll, Waker}, - time::SystemTime, -}; - -use pin_project_lite::pin_project; -use tracing::{trace, warn}; - -use crate::{ - context, define_bypass, define_sys_interceptor, - rand::{GlobalRng, Rng}, - task::NodeId, -}; - -pub mod error; -mod instant; -mod interval; -mod sleep; -mod timer; - -use timer::Timer; - -pub use self::{ - instant::Instant, - interval::{interval, interval_at, Interval, MissedTickBehavior}, - sleep::{sleep, sleep_until, Sleep}, -}; - -pub(crate) struct TimeRuntime { - handle: TimeHandle, -} - -impl TimeRuntime { - pub fn new(rand: &GlobalRng) -> Self { - let base_time = SystemTime::UNIX_EPOCH - + match std::env::var("MSIM_BASE_TIME") { - Ok(base_time) => { - let base_time: u64 = match base_time.parse() { - Ok(t) => t, - Err(e) => panic!( - "MSIM_BASE_TIME='{}' was not parseable as a u64: {}", - base_time, e - ), - }; - Duration::from_secs(base_time) - } - - // around 2022 - Err(_) => Duration::from_secs( - 60 * 60 * 24 * 365 * (2022 - 1970) - + rand.with(|rng| rng.gen_range(0..60 * 60 * 24 * 365)), - ), - }; - let handle = TimeHandle { - timer: Arc::new(Mutex::new(Timer::default())), - clock: ClockHandle::new(base_time), - }; - TimeRuntime { handle } - } - - pub fn handle(&self) -> &TimeHandle { - &self.handle - } - - /// Advances time to the closest timer event. Returns true if succeed. - pub fn advance_to_next_event(&self) -> bool { - let mut timer = self.handle.timer.lock().unwrap(); - if let Some(mut time) = timer.next() { - // WARN: in some platform such as M1 macOS, - // let t0: Instant; - // let t1: Instant; - // t0 + (t1 - t0) < t1 !! - // we should add eps to make sure 'now >= deadline' and avoid deadlock - time += Duration::from_nanos(50); - - timer.expire(time); - self.handle.clock.set_elapsed(time); - true - } else { - false - } - } - - /// Advances time. - pub fn advance(&self, duration: Duration) { - self.handle.clock.advance(duration); - } - - #[allow(dead_code)] - /// Get the current time. - pub fn now_instant(&self) -> Instant { - self.handle.now_instant() - } -} - -/// Handle to a shared time source. -#[derive(Clone)] -pub struct TimeHandle { - timer: Arc>, - clock: ClockHandle, -} - -impl TimeHandle { - /// Disable node, cancel all pending timers. - pub fn disable_node_and_cancel_timers(&self, node_id: NodeId) { - let mut timer = self.timer.lock().unwrap(); - let events = timer.disable_node_and_remove_events(node_id); - // must drop lock before events to avoid deadlock. - drop(timer); - drop(events); - } - - /// Enable a previously disabled node. - pub fn enable_node(&self, node_id: NodeId) { - self.timer.lock().unwrap().enable_node(node_id); - } - - /// Returns a `TimeHandle` view over the currently running Runtime. - pub fn current() -> Self { - crate::context::current(|h| h.time.clone()) - } - - /// Returns a `TimeHandle` view over the currently running Runtime. - pub fn try_current() -> Option { - crate::context::try_current(|h| h.time.clone()) - } - - /// Return the current time. - pub fn now_instant(&self) -> Instant { - self.clock.now_instant() - } - - /// Return the current time. - pub fn now_time(&self) -> SystemTime { - self.clock.now_time() - } - - /// Returns the amount of time elapsed since this handle was created. - pub fn elapsed(&self) -> Duration { - self.clock.elapsed() - } - - /// Waits until `duration` has elapsed. - pub fn sleep(&self, duration: Duration) -> Sleep { - self.sleep_until(self.clock.now_instant() + duration) - } - - /// Waits until `deadline` is reached. - pub fn sleep_until(&self, deadline: Instant) -> Sleep { - Sleep { - handle: self.clone(), - deadline, - } - } - - /// Require a `Future` to complete before the specified duration has elapsed. - // TODO: make it Send - pub fn timeout(&self, duration: Duration, future: T) -> Timeout { - let delay = self.sleep(duration); - Timeout { - value: future, - delay, - } - } - - pub(crate) fn add_timer( - &self, - deadline: Instant, - callback: impl FnOnce() + Send + Sync + 'static, - ) { - self.add_timer_for_node(context::current_node(), deadline, callback); - } - - pub(crate) fn add_timer_for_node( - &self, - node_id: NodeId, - deadline: Instant, - callback: impl FnOnce() + Send + Sync + 'static, - ) { - let mut timer = self.timer.lock().unwrap(); - timer.add(node_id, deadline - self.clock.base_instant(), |_| { - callback() - }); - } - - /// Schedule waker.wake() in the future. - pub fn wake_at(&self, deadline: Instant, waker: Waker) { - self.add_timer(deadline, || waker.wake()); - } - - // Get the elapsed time since the beginning of the test run - this should not be exposed to - // test code. - pub(crate) fn time_since_clock_base(&self) -> Duration { - self.clock.time_since_clock_base() - } -} - -pin_project! { - /// Future returned by [`timeout`](timeout) and [`timeout_at`](timeout_at). - #[must_use = "futures do nothing unless you `.await` or poll them"] - #[derive(Debug)] - pub struct Timeout { - #[pin] - value: T, - #[pin] - delay: Sleep, - } -} - -impl Timeout { - /// Gets a reference to the underlying value in this timeout. - pub fn get_ref(&self) -> &T { - &self.value - } - - /// Gets a mutable reference to the underlying value in this timeout. - pub fn get_mut(&mut self) -> &mut T { - &mut self.value - } - - /// Consumes this timeout, returning the underlying value. - pub fn into_inner(self) -> T { - self.value - } -} - -impl Future for Timeout -where - T: Future, -{ - type Output = Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let this = self.project(); - if let Poll::Ready(v) = this.value.poll(cx) { - return Poll::Ready(Ok(v)); - } - match this.delay.poll(cx) { - Poll::Ready(()) => Poll::Ready(Err(error::Elapsed)), - Poll::Pending => Poll::Pending, - } - } -} - -/// Supply tokio::time::advance() and tokio::time::pause() APIs (for compilation only - -/// these methods are meaningless inside the simulator). -pub async fn advance(_duration: Duration) { - unimplemented!("cannot advance clock in simulation - use sleep() instead"); -} - -/// Supply tokio::time::pause() API (for compilation only - this method is meaningless inside the -/// simulator). -pub fn pause() { - unimplemented!("cannot pause clock in simulation"); -} - -/// Require a `Future` to complete before the specified duration has elapsed. -pub fn timeout(duration: Duration, future: T) -> Timeout { - let handle = TimeHandle::current(); - handle.timeout(duration, future) -} - -/// Require a `Future` to complete before the specified deadline. -pub fn timeout_at(deadline: Instant, future: T) -> Timeout { - let duration = deadline.saturating_duration_since(Instant::now()); - timeout(duration, future) -} - -#[derive(Clone)] -struct ClockHandle { - inner: Arc>, -} - -#[derive(Debug)] -struct Clock { - /// Time basis for which mock time is derived. - base_time: std::time::SystemTime, - base_instant: Instant, - /// The amount of mock time which has elapsed. - elapsed_time: Duration, - - /// Reset every time the clock is advanced. - /// Decremented every time the clock is read. - /// If it reaches 0, we panic - otherwise code like - /// while Instant::now() < deadline { ... } will spin forever. - abort_counter: u32, -} - -impl Clock { - const ABORT_COUNTER_START: u32 = 100_000_000; - - #[inline(always)] - fn check_abort(&mut self) { - self.abort_counter -= 1; - if self.abort_counter == 0 { - panic!("Likely infinite loop detected: clock sampled too many times without advancing"); - } - } - - #[inline(always)] - fn reset_abort_counter(&mut self) { - self.abort_counter = Self::ABORT_COUNTER_START; - } -} - -impl ClockHandle { - const CLOCK_BASE: Duration = Duration::from_secs(86400); - - fn new(base_time: SystemTime) -> Self { - let base_instant: Instant = unsafe { std::mem::zeroed() }; - let clock = Clock { - base_time, - base_instant, - // Some code subtracts constant durations from Instant::now(), which underflows if the base - // instant is too small. That code is incorrect but we'll just make life easy anyway by - // starting the clock with one day of elapsed time. - elapsed_time: Self::CLOCK_BASE, - abort_counter: Clock::ABORT_COUNTER_START, - }; - ClockHandle { - inner: Arc::new(Mutex::new(clock)), - } - } - - fn time_since_clock_base(&self) -> Duration { - self.elapsed() - Self::CLOCK_BASE - } - - fn set_elapsed(&self, time: Duration) { - let mut inner = self.inner.lock().unwrap(); - // prevent time from going backwards - otherwise this can happen when timers are late. - inner.elapsed_time = std::cmp::max(inner.elapsed_time, time); - inner.reset_abort_counter(); - } - - fn elapsed(&self) -> Duration { - let mut inner = self.inner.lock().unwrap(); - inner.check_abort(); - inner.elapsed_time - } - - fn advance(&self, duration: Duration) { - let mut inner = self.inner.lock().unwrap(); - inner.reset_abort_counter(); - inner.elapsed_time += duration; - } - - fn base_instant(&self) -> Instant { - let inner = self.inner.lock().unwrap(); - inner.base_instant - } - - fn now_instant(&self) -> Instant { - let mut inner = self.inner.lock().unwrap(); - inner.check_abort(); - inner.base_instant + inner.elapsed_time - } - - fn now_time(&self) -> SystemTime { - let mut inner = self.inner.lock().unwrap(); - inner.check_abort(); - inner.base_time + inner.elapsed_time - } -} - -// ensure that clock functions are not elided by optimizer -pub(crate) fn ensure_clocks() { - unsafe { - #[cfg(target_os = "macos")] - mach_absolute_time(); - - #[cfg(target_os = "linux")] - { - let mut ts = libc::timespec { - tv_sec: 0, - tv_nsec: 0, - }; - clock_gettime(libc::CLOCK_MONOTONIC, &mut ts as *mut libc::timespec); - } - } -} - -fn use_real_wallcock() -> bool { - thread_local! { - // Note: setting MSIM_USE_REAL_WALLCLOCK will result in non-determinism for code that reads - // and acts on the absolute wall time. (Anything involving deltas only will still be - // deterministic). - static USE_REAL_WALLCLOCK: bool = { - let use_real_wallcock = std::env::var("MSIM_USE_REAL_WALLCLOCK").is_ok(); - if use_real_wallcock { - warn!("non-determinism possible: using real wall clock because of MSIM_USE_REAL_WALLCLOCK"); - } - use_real_wallcock - } - } - USE_REAL_WALLCLOCK.with(|u| *u) -} - -#[cfg(target_os = "macos")] -define_bypass!(bypass_gettimeofday, - fn gettimeofday(tp: *mut libc::timeval, tz: *mut libc::c_void) -> libc::c_int); - -#[cfg(target_os = "macos")] -define_sys_interceptor!( - fn gettimeofday(tp: *mut libc::timeval, tz: *mut libc::c_void) -> libc::c_int { - if use_real_wallcock() { - return bypass_gettimeofday(tp, tz); - } - - // NOTE: tz should be NULL. - // macOS: timezone is no longer used; this information is kept outside the kernel. - if tp.is_null() { - return 0; - } - let time = TimeHandle::current(); - let dur = time - .now_time() - .duration_since(std::time::SystemTime::UNIX_EPOCH) - .unwrap(); - tp.write(libc::timeval { - tv_sec: dur.as_secs() as _, - tv_usec: dur.subsec_micros() as _, - }); - 0 - } -); - -#[cfg(target_os = "macos")] -define_bypass!(bypass_mach_absolute_time, - fn mach_absolute_time() -> u64); - -#[cfg(target_os = "macos")] -define_sys_interceptor!( - fn mach_absolute_time() -> u64 { - #[repr(C)] - #[derive(Copy, Clone)] - struct MachTimebaseInfo { - numer: u32, - denom: u32, - } - type MachTimebaseInfoT = *mut MachTimebaseInfo; - - lazy_static::lazy_static! { - static ref MACH_TIME_BASE_INFO: MachTimebaseInfo = { - extern "C" { - fn mach_timebase_info(info: MachTimebaseInfoT) -> libc::c_int; - } - - let mut info = MachTimebaseInfo { numer: 0, denom: 0 }; - unsafe { - mach_timebase_info(&mut info as MachTimebaseInfoT); - } - assert_ne!(info.numer, 0); - assert_ne!(info.denom, 0); - info - }; - } - - fn mul_div_u64(value: u64, numer: u64, denom: u64) -> u64 { - let q = value / denom; - let r = value % denom; - // Decompose value as (value/denom*denom + value%denom), - // substitute into (value*numer)/denom and simplify. - // r < denom, so (denom*numer) is the upper bound of (r*numer) - q * numer + r * numer / denom - } - - let time = match TimeHandle::try_current() { - Some(t) => t, - None => { - // Sometimes Drop impls ask for the current time as the runtime is being dropped. It - // might be better to try to completely drop everything owned by the runtime inside - // of a block_on() call, but that is tricky to do correctly. - trace!("mach_absolute_time called outside of Runtime"); - return bypass_mach_absolute_time(); - } - }; - - let elapsed = time.elapsed(); - let nanos = elapsed.as_nanos().try_into().unwrap(); - - // convert nanos back to mach_absolute_time units - mul_div_u64( - nanos, - MACH_TIME_BASE_INFO.denom as u64, - MACH_TIME_BASE_INFO.numer as u64, - ) - } -); - -define_bypass!(bypass_clock_gettime, - fn clock_gettime(clock_id: libc::clockid_t, ts: *mut libc::timespec) -> libc::c_int); - -// Since Rust 1.67.0, Instant::now() and SystemTime::now() are implemented using clock_gettime() -#[cfg(target_os = "macos")] -define_sys_interceptor!( - fn clock_gettime(clock_id: libc::clockid_t, ts: *mut libc::timespec) -> libc::c_int { - let Some(time) = TimeHandle::try_current() else { - // Sometimes Drop impls ask for the current time as the runtime is being dropped. It - // might be better to try to completely drop everything owned by the runtime inside - // of a block_on() call, but that is tricky to do correctly. - trace!("clock_gettime called outside of Runtime"); - return bypass_clock_gettime(clock_id, ts); - }; - - match clock_id { - // used by SystemTime - libc::CLOCK_REALTIME => { - if use_real_wallcock() { - return bypass_clock_gettime(clock_id, ts); - } - - let dur = time - .now_time() - .duration_since(std::time::SystemTime::UNIX_EPOCH) - .unwrap(); - ts.write(libc::timespec { - tv_sec: dur.as_secs() as _, - tv_nsec: dur.subsec_nanos() as _, - }); - } - - // used by Instant - libc::CLOCK_MONOTONIC | libc::CLOCK_UPTIME_RAW | libc::CLOCK_MONOTONIC_RAW => { - let dur = time.elapsed(); - ts.write(libc::timespec { - tv_sec: dur.as_secs() as _, - tv_nsec: dur.subsec_nanos() as _, - }); - } - - // Used by rocksdb performance timers. - libc::CLOCK_PROCESS_CPUTIME_ID | libc::CLOCK_THREAD_CPUTIME_ID => { - return bypass_clock_gettime(clock_id, ts); - } - - _ => panic!("unsupported clockid: {}", clock_id), - } - 0 - } -); - -#[cfg(target_os = "linux")] -define_sys_interceptor!( - fn clock_gettime(clock_id: libc::clockid_t, ts: *mut libc::timespec) -> libc::c_int { - let time = match TimeHandle::try_current() { - Some(t) => t, - None => { - // Sometimes Drop impls ask for the current time as the runtime is being dropped. It - // might be better to try to completely drop everything owned by the runtime inside - // of a block_on() call, but that is tricky to do correctly. - trace!("clock_gettime called outside of Runtime"); - return bypass_clock_gettime(clock_id, ts); - } - }; - - match clock_id { - // used by SystemTime - libc::CLOCK_REALTIME | libc::CLOCK_REALTIME_COARSE | libc::CLOCK_BOOTTIME => { - if use_real_wallcock() { - return bypass_clock_gettime(clock_id, ts); - } - - let dur = time - .now_time() - .duration_since(std::time::SystemTime::UNIX_EPOCH) - .unwrap(); - ts.write(libc::timespec { - tv_sec: dur.as_secs() as _, - tv_nsec: dur.subsec_nanos() as _, - }); - } - - // used by Instant - libc::CLOCK_MONOTONIC | libc::CLOCK_MONOTONIC_RAW | libc::CLOCK_MONOTONIC_COARSE => { - // Instant is the same layout as timespec on linux - ts.write(std::mem::transmute::< - crate::sim::time::instant::Instant, - libc::timespec, - >(time.now_instant())); - } - - // Used by rocksdb performance timers. - libc::CLOCK_PROCESS_CPUTIME_ID | libc::CLOCK_THREAD_CPUTIME_ID => { - return bypass_clock_gettime(clock_id, ts); - } - - _ => panic!("unsupported clockid: {}", clock_id), - } - 0 - } -); - -#[cfg(test)] -mod tests { - use super::*; - use crate::runtime::{init_logger, Runtime}; - - #[test] - fn time() { - let runtime = Runtime::new(); - runtime.block_on(async { - let t0 = Instant::now(); - let std_t0 = std::time::Instant::now(); - - // Verify that times in other threads are not intercepted. - let std_t1 = std::thread::spawn(|| std::time::Instant::now()) - .join() - .unwrap(); - assert_ne!(std_t0, std_t1); - - sleep(Duration::from_secs(1)).await; - assert!(t0.elapsed() >= Duration::from_secs(1)); - - sleep_until(t0 + Duration::from_secs(2)).await; - assert!(t0.elapsed() >= Duration::from_secs(2)); - - sleep(Duration::from_secs(20)).await; - - // make sure system clock has been intercepted. - assert_eq!(std_t0.elapsed(), t0.elapsed()); - assert!( - timeout(Duration::from_secs(2), sleep(Duration::from_secs(1))) - .await - .is_ok() - ); - assert!( - timeout(Duration::from_secs(1), sleep(Duration::from_secs(2))) - .await - .is_err() - ); - }); - } - - // Can't easily test behaviors that rely on env vars. To test manually, run: - // - // Verify that the same system time is always printed. - // - // RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture - // - // Verify that a different system time is printed for different seeds. - // - // MSIM_TEST_SEED=2 RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture - // - // Verify that the correct real system time is printed. - // - // MSIM_USE_REAL_WALLCLOCK=1 RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture - // - // Verify that the specified base time is printed. - // - // MSIM_BASE_TIME=1453807127 RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture - #[test] - fn system_clock() { - init_logger(); - let seed: u64 = std::env::var("MSIM_TEST_SEED") - .unwrap_or("1".to_string()) - .parse() - .unwrap(); - - let runtime = Runtime::with_seed(seed); - runtime.block_on(async { - let t0 = Instant::now(); - let s0 = SystemTime::now(); - - println!("{:?} {:?}", t0, s0); - }); - } -} +//! Utilities for tracking time. +//! +//! + +#[doc(no_inline)] +pub use std::time::Duration; +use std::{ + future::Future, + pin::Pin, + sync::{Arc, Mutex}, + task::{Context, Poll, Waker}, + time::SystemTime, +}; + +use pin_project_lite::pin_project; +use tracing::{trace, warn}; + +use crate::{ + context, define_bypass, define_sys_interceptor, + rand::{GlobalRng, Rng}, + task::NodeId, +}; + +pub mod error; +mod instant; +mod interval; +mod sleep; +mod timer; + +use timer::Timer; + +pub use self::{ + instant::Instant, + interval::{interval, interval_at, Interval, MissedTickBehavior}, + sleep::{sleep, sleep_until, Sleep}, +}; + +pub(crate) struct TimeRuntime { + handle: TimeHandle, +} + +impl TimeRuntime { + pub fn new(rand: &GlobalRng) -> Self { + let base_time = SystemTime::UNIX_EPOCH + + match std::env::var("MSIM_BASE_TIME") { + Ok(base_time) => { + let base_time: u64 = match base_time.parse() { + Ok(t) => t, + Err(e) => panic!( + "MSIM_BASE_TIME='{}' was not parseable as a u64: {}", + base_time, e + ), + }; + Duration::from_secs(base_time) + } + + // around 2022 + Err(_) => Duration::from_secs( + 60 * 60 * 24 * 365 * (2022 - 1970) + + rand.with(|rng| rng.gen_range(0..60 * 60 * 24 * 365)), + ), + }; + let handle = TimeHandle { + timer: Arc::new(Mutex::new(Timer::default())), + clock: ClockHandle::new(base_time), + }; + TimeRuntime { handle } + } + + pub fn handle(&self) -> &TimeHandle { + &self.handle + } + + /// Advances time to the closest timer event. Returns true if succeed. + pub fn advance_to_next_event(&self) -> bool { + let mut timer = self.handle.timer.lock().unwrap(); + if let Some(mut time) = timer.next() { + // WARN: in some platform such as M1 macOS, + // let t0: Instant; + // let t1: Instant; + // t0 + (t1 - t0) < t1 !! + // we should add eps to make sure 'now >= deadline' and avoid deadlock + time += Duration::from_nanos(50); + + timer.expire(time); + self.handle.clock.set_elapsed(time); + true + } else { + false + } + } + + /// Advances time. + pub fn advance(&self, duration: Duration) { + self.handle.clock.advance(duration); + } + + #[allow(dead_code)] + /// Get the current time. + pub fn now_instant(&self) -> Instant { + self.handle.now_instant() + } +} + +/// Handle to a shared time source. +#[derive(Clone)] +pub struct TimeHandle { + timer: Arc>, + clock: ClockHandle, +} + +impl TimeHandle { + /// Disable node, cancel all pending timers. + pub fn disable_node_and_cancel_timers(&self, node_id: NodeId) { + let mut timer = self.timer.lock().unwrap(); + let events = timer.disable_node_and_remove_events(node_id); + // must drop lock before events to avoid deadlock. + drop(timer); + drop(events); + } + + /// Enable a previously disabled node. + pub fn enable_node(&self, node_id: NodeId) { + self.timer.lock().unwrap().enable_node(node_id); + } + + /// Returns a `TimeHandle` view over the currently running Runtime. + pub fn current() -> Self { + crate::context::current(|h| h.time.clone()) + } + + /// Returns a `TimeHandle` view over the currently running Runtime. + pub fn try_current() -> Option { + crate::context::try_current(|h| h.time.clone()) + } + + /// Return the current time. + pub fn now_instant(&self) -> Instant { + self.clock.now_instant() + } + + /// Return the current time. + pub fn now_time(&self) -> SystemTime { + self.clock.now_time() + } + + /// Returns the amount of time elapsed since this handle was created. + pub fn elapsed(&self) -> Duration { + self.clock.elapsed() + } + + /// Waits until `duration` has elapsed. + pub fn sleep(&self, duration: Duration) -> Sleep { + self.sleep_until(self.clock.now_instant() + duration) + } + + /// Waits until `deadline` is reached. + pub fn sleep_until(&self, deadline: Instant) -> Sleep { + Sleep { + handle: self.clone(), + deadline, + } + } + + /// Require a `Future` to complete before the specified duration has elapsed. + // TODO: make it Send + pub fn timeout(&self, duration: Duration, future: T) -> Timeout { + let delay = self.sleep(duration); + Timeout { + value: future, + delay, + } + } + + pub(crate) fn add_timer( + &self, + deadline: Instant, + callback: impl FnOnce() + Send + Sync + 'static, + ) { + self.add_timer_for_node(context::current_node(), deadline, callback); + } + + pub(crate) fn add_timer_for_node( + &self, + node_id: NodeId, + deadline: Instant, + callback: impl FnOnce() + Send + Sync + 'static, + ) { + let mut timer = self.timer.lock().unwrap(); + timer.add(node_id, deadline - self.clock.base_instant(), |_| { + callback() + }); + } + + /// Schedule waker.wake() in the future. + pub fn wake_at(&self, deadline: Instant, waker: Waker) { + self.add_timer(deadline, || waker.wake()); + } + + // Get the elapsed time since the beginning of the test run - this should not be exposed to + // test code. + pub(crate) fn time_since_clock_base(&self) -> Duration { + self.clock.time_since_clock_base() + } +} + +pin_project! { + /// Future returned by [`timeout`](timeout) and [`timeout_at`](timeout_at). + #[must_use = "futures do nothing unless you `.await` or poll them"] + #[derive(Debug)] + pub struct Timeout { + #[pin] + value: T, + #[pin] + delay: Sleep, + } +} + +impl Timeout { + /// Gets a reference to the underlying value in this timeout. + pub fn get_ref(&self) -> &T { + &self.value + } + + /// Gets a mutable reference to the underlying value in this timeout. + pub fn get_mut(&mut self) -> &mut T { + &mut self.value + } + + /// Consumes this timeout, returning the underlying value. + pub fn into_inner(self) -> T { + self.value + } +} + +impl Future for Timeout +where + T: Future, +{ + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.project(); + if let Poll::Ready(v) = this.value.poll(cx) { + return Poll::Ready(Ok(v)); + } + match this.delay.poll(cx) { + Poll::Ready(()) => Poll::Ready(Err(error::Elapsed)), + Poll::Pending => Poll::Pending, + } + } +} + +/// Supply tokio::time::advance() and tokio::time::pause() APIs (for compilation only - +/// these methods are meaningless inside the simulator). +pub async fn advance(_duration: Duration) { + unimplemented!("cannot advance clock in simulation - use sleep() instead"); +} + +/// Supply tokio::time::pause() API (for compilation only - this method is meaningless inside the +/// simulator). +pub fn pause() { + unimplemented!("cannot pause clock in simulation"); +} + +/// Require a `Future` to complete before the specified duration has elapsed. +pub fn timeout(duration: Duration, future: T) -> Timeout { + let handle = TimeHandle::current(); + handle.timeout(duration, future) +} + +/// Require a `Future` to complete before the specified deadline. +pub fn timeout_at(deadline: Instant, future: T) -> Timeout { + let duration = deadline.saturating_duration_since(Instant::now()); + timeout(duration, future) +} + +#[derive(Clone)] +struct ClockHandle { + inner: Arc>, +} + +#[derive(Debug)] +struct Clock { + /// Time basis for which mock time is derived. + base_time: std::time::SystemTime, + base_instant: Instant, + /// The amount of mock time which has elapsed. + elapsed_time: Duration, + + /// Reset every time the clock is advanced. + /// Decremented every time the clock is read. + /// If it reaches 0, we panic - otherwise code like + /// while Instant::now() < deadline { ... } will spin forever. + abort_counter: u32, +} + +impl Clock { + const ABORT_COUNTER_START: u32 = 100_000_000; + + #[inline(always)] + fn check_abort(&mut self) { + self.abort_counter -= 1; + if self.abort_counter == 0 { + panic!("Likely infinite loop detected: clock sampled too many times without advancing"); + } + } + + #[inline(always)] + fn reset_abort_counter(&mut self) { + self.abort_counter = Self::ABORT_COUNTER_START; + } +} + +impl ClockHandle { + const CLOCK_BASE: Duration = Duration::from_secs(86400); + + fn new(base_time: SystemTime) -> Self { + let base_instant: Instant = unsafe { std::mem::zeroed() }; + let clock = Clock { + base_time, + base_instant, + // Some code subtracts constant durations from Instant::now(), which underflows if the base + // instant is too small. That code is incorrect but we'll just make life easy anyway by + // starting the clock with one day of elapsed time. + elapsed_time: Self::CLOCK_BASE, + abort_counter: Clock::ABORT_COUNTER_START, + }; + ClockHandle { + inner: Arc::new(Mutex::new(clock)), + } + } + + fn time_since_clock_base(&self) -> Duration { + self.elapsed() - Self::CLOCK_BASE + } + + fn set_elapsed(&self, time: Duration) { + let mut inner = self.inner.lock().unwrap(); + // prevent time from going backwards - otherwise this can happen when timers are late. + inner.elapsed_time = std::cmp::max(inner.elapsed_time, time); + inner.reset_abort_counter(); + } + + fn elapsed(&self) -> Duration { + let mut inner = self.inner.lock().unwrap(); + inner.check_abort(); + inner.elapsed_time + } + + fn advance(&self, duration: Duration) { + let mut inner = self.inner.lock().unwrap(); + inner.reset_abort_counter(); + inner.elapsed_time += duration; + } + + fn base_instant(&self) -> Instant { + let inner = self.inner.lock().unwrap(); + inner.base_instant + } + + fn now_instant(&self) -> Instant { + let mut inner = self.inner.lock().unwrap(); + inner.check_abort(); + inner.base_instant + inner.elapsed_time + } + + fn now_time(&self) -> SystemTime { + let mut inner = self.inner.lock().unwrap(); + inner.check_abort(); + inner.base_time + inner.elapsed_time + } +} + +// ensure that clock functions are not elided by optimizer +pub(crate) fn ensure_clocks() { + unsafe { + #[cfg(target_os = "macos")] + mach_absolute_time(); + + #[cfg(target_os = "linux")] + { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + clock_gettime(libc::CLOCK_MONOTONIC, &mut ts as *mut libc::timespec); + } + } +} + +fn use_real_wallcock() -> bool { + thread_local! { + // Note: setting MSIM_USE_REAL_WALLCLOCK will result in non-determinism for code that reads + // and acts on the absolute wall time. (Anything involving deltas only will still be + // deterministic). + static USE_REAL_WALLCLOCK: bool = { + let use_real_wallcock = std::env::var("MSIM_USE_REAL_WALLCLOCK").is_ok(); + if use_real_wallcock { + warn!("non-determinism possible: using real wall clock because of MSIM_USE_REAL_WALLCLOCK"); + } + use_real_wallcock + } + } + USE_REAL_WALLCLOCK.with(|u| *u) +} + +#[cfg(target_os = "macos")] +define_bypass!(bypass_gettimeofday, + fn gettimeofday(tp: *mut libc::timeval, tz: *mut libc::c_void) -> libc::c_int); + +#[cfg(target_os = "macos")] +define_sys_interceptor!( + fn gettimeofday(tp: *mut libc::timeval, tz: *mut libc::c_void) -> libc::c_int { + if use_real_wallcock() { + return bypass_gettimeofday(tp, tz); + } + + // NOTE: tz should be NULL. + // macOS: timezone is no longer used; this information is kept outside the kernel. + if tp.is_null() { + return 0; + } + let time = TimeHandle::current(); + let dur = time + .now_time() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .unwrap(); + tp.write(libc::timeval { + tv_sec: dur.as_secs() as _, + tv_usec: dur.subsec_micros() as _, + }); + 0 + } +); + +#[cfg(target_os = "macos")] +define_bypass!(bypass_mach_absolute_time, + fn mach_absolute_time() -> u64); + +#[cfg(target_os = "macos")] +define_sys_interceptor!( + fn mach_absolute_time() -> u64 { + #[repr(C)] + #[derive(Copy, Clone)] + struct MachTimebaseInfo { + numer: u32, + denom: u32, + } + type MachTimebaseInfoT = *mut MachTimebaseInfo; + + lazy_static::lazy_static! { + static ref MACH_TIME_BASE_INFO: MachTimebaseInfo = { + extern "C" { + fn mach_timebase_info(info: MachTimebaseInfoT) -> libc::c_int; + } + + let mut info = MachTimebaseInfo { numer: 0, denom: 0 }; + unsafe { + mach_timebase_info(&mut info as MachTimebaseInfoT); + } + assert_ne!(info.numer, 0); + assert_ne!(info.denom, 0); + info + }; + } + + fn mul_div_u64(value: u64, numer: u64, denom: u64) -> u64 { + let q = value / denom; + let r = value % denom; + // Decompose value as (value/denom*denom + value%denom), + // substitute into (value*numer)/denom and simplify. + // r < denom, so (denom*numer) is the upper bound of (r*numer) + q * numer + r * numer / denom + } + + let time = match TimeHandle::try_current() { + Some(t) => t, + None => { + // Sometimes Drop impls ask for the current time as the runtime is being dropped. It + // might be better to try to completely drop everything owned by the runtime inside + // of a block_on() call, but that is tricky to do correctly. + trace!("mach_absolute_time called outside of Runtime"); + return bypass_mach_absolute_time(); + } + }; + + let elapsed = time.elapsed(); + let nanos = elapsed.as_nanos().try_into().unwrap(); + + // convert nanos back to mach_absolute_time units + mul_div_u64( + nanos, + MACH_TIME_BASE_INFO.denom as u64, + MACH_TIME_BASE_INFO.numer as u64, + ) + } +); + +define_bypass!(bypass_clock_gettime, + fn clock_gettime(clock_id: libc::clockid_t, ts: *mut libc::timespec) -> libc::c_int); + +// Since Rust 1.67.0, Instant::now() and SystemTime::now() are implemented using clock_gettime() +#[cfg(target_os = "macos")] +define_sys_interceptor!( + fn clock_gettime(clock_id: libc::clockid_t, ts: *mut libc::timespec) -> libc::c_int { + let Some(time) = TimeHandle::try_current() else { + // Sometimes Drop impls ask for the current time as the runtime is being dropped. It + // might be better to try to completely drop everything owned by the runtime inside + // of a block_on() call, but that is tricky to do correctly. + trace!("clock_gettime called outside of Runtime"); + return bypass_clock_gettime(clock_id, ts); + }; + + match clock_id { + // used by SystemTime + libc::CLOCK_REALTIME => { + if use_real_wallcock() { + return bypass_clock_gettime(clock_id, ts); + } + + let dur = time + .now_time() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .unwrap(); + ts.write(libc::timespec { + tv_sec: dur.as_secs() as _, + tv_nsec: dur.subsec_nanos() as _, + }); + } + + // used by Instant + libc::CLOCK_MONOTONIC | libc::CLOCK_UPTIME_RAW | libc::CLOCK_MONOTONIC_RAW => { + let dur = time.elapsed(); + ts.write(libc::timespec { + tv_sec: dur.as_secs() as _, + tv_nsec: dur.subsec_nanos() as _, + }); + } + + // Used by rocksdb performance timers. + libc::CLOCK_PROCESS_CPUTIME_ID | libc::CLOCK_THREAD_CPUTIME_ID => { + return bypass_clock_gettime(clock_id, ts); + } + + _ => panic!("unsupported clockid: {}", clock_id), + } + 0 + } +); + +#[cfg(target_os = "linux")] +define_sys_interceptor!( + fn clock_gettime(clock_id: libc::clockid_t, ts: *mut libc::timespec) -> libc::c_int { + let time = match TimeHandle::try_current() { + Some(t) => t, + None => { + // Sometimes Drop impls ask for the current time as the runtime is being dropped. It + // might be better to try to completely drop everything owned by the runtime inside + // of a block_on() call, but that is tricky to do correctly. + trace!("clock_gettime called outside of Runtime"); + return bypass_clock_gettime(clock_id, ts); + } + }; + + match clock_id { + // used by SystemTime + libc::CLOCK_REALTIME | libc::CLOCK_REALTIME_COARSE | libc::CLOCK_BOOTTIME => { + if use_real_wallcock() { + return bypass_clock_gettime(clock_id, ts); + } + + let dur = time + .now_time() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .unwrap(); + ts.write(libc::timespec { + tv_sec: dur.as_secs() as _, + tv_nsec: dur.subsec_nanos() as _, + }); + } + + // Used by Instant. + libc::CLOCK_MONOTONIC | libc::CLOCK_MONOTONIC_RAW | libc::CLOCK_MONOTONIC_COARSE => { + // SECURITY FIX: Removed the opaque Instant to timespec transmute. + // Serialize the simulated duration explicitly; std::time::Instant is opaque and + // its internal layout is not guaranteed to match libc::timespec. + let duration = time.elapsed(); + ts.write(libc::timespec { + tv_sec: duration.as_secs() as _, + tv_nsec: duration.subsec_nanos() as _, + }); + } + + // Used by rocksdb performance timers. + libc::CLOCK_PROCESS_CPUTIME_ID | libc::CLOCK_THREAD_CPUTIME_ID => { + return bypass_clock_gettime(clock_id, ts); + } + + _ => panic!("unsupported clockid: {}", clock_id), + } + 0 + } +); + +#[cfg(test)] +mod tests { + use super::*; + use crate::runtime::{init_logger, Runtime}; + + #[test] + fn time() { + let runtime = Runtime::new(); + runtime.block_on(async { + let t0 = Instant::now(); + let std_t0 = std::time::Instant::now(); + + // Verify that times in other threads are not intercepted. + let std_t1 = std::thread::spawn(|| std::time::Instant::now()) + .join() + .unwrap(); + assert_ne!(std_t0, std_t1); + + sleep(Duration::from_secs(1)).await; + assert!(t0.elapsed() >= Duration::from_secs(1)); + + sleep_until(t0 + Duration::from_secs(2)).await; + assert!(t0.elapsed() >= Duration::from_secs(2)); + + sleep(Duration::from_secs(20)).await; + + // make sure system clock has been intercepted. + assert_eq!(std_t0.elapsed(), t0.elapsed()); + assert!( + timeout(Duration::from_secs(2), sleep(Duration::from_secs(1))) + .await + .is_ok() + ); + assert!( + timeout(Duration::from_secs(1), sleep(Duration::from_secs(2))) + .await + .is_err() + ); + }); + } + + // Can't easily test behaviors that rely on env vars. To test manually, run: + // + // Verify that the same system time is always printed. + // + // RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture + // + // Verify that a different system time is printed for different seeds. + // + // MSIM_TEST_SEED=2 RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture + // + // Verify that the correct real system time is printed. + // + // MSIM_USE_REAL_WALLCLOCK=1 RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture + // + // Verify that the specified base time is printed. + // + // MSIM_BASE_TIME=1453807127 RUSTFLAGS="--cfg msim" cargo nextest run system_clock --no-capture + #[test] + fn system_clock() { + init_logger(); + let seed: u64 = std::env::var("MSIM_TEST_SEED") + .unwrap_or("1".to_string()) + .parse() + .unwrap(); + + let runtime = Runtime::with_seed(seed); + runtime.block_on(async { + let t0 = Instant::now(); + let s0 = SystemTime::now(); + + println!("{:?} {:?}", t0, s0); + }); + } +} \ No newline at end of file