//! Completely mock runtime use std::fmt::{Debug, Display}; use amplify::Getters; use futures::FutureExt as _; use itertools::chain; use strum::IntoEnumIterator as _; use void::{ResultVoidExt as _, Void}; use crate::util::impl_runtime_prelude::*; use crate::net::MockNetProvider; use crate::task::{MockExecutor, SchedulingPolicy}; use crate::time::MockSleepProvider; /// Completely mock runtime /// /// Suitable for test cases that wish to completely control /// the environment experienced by the code under test. /// /// ### Restrictions /// /// The test case must advance the mock time explicitly as desired. /// There is not currently any facility for automatically /// making progress by advancing the mock time by the right amounts /// for the timeouts set by the futures under test. // ^ I think such a facility could be provided. `MockSleepProvider` would have to // provide a method to identify the next interesting time event. // The waitfor machinery in MockSleepProvider and MockSleepRuntime doesn't seem suitable. /// /// Tests that use this runtime *must not* interact with the outside world; /// everything must go through this runtime (and its pieces). /// /// #### Allowed /// /// * Inter-future communication facilities from `futures` /// or other runtime-agnostic crates. /// /// * Fast synchronous operations that will complete "immediately" or "quickly". /// E.g.: filesystem calls. /// /// * `std::sync::Mutex` (assuming the use is deadlock-free in a single-threaded /// executor, as it should be in all of Arti). /// /// * Slower operations that are run synchronously (without futures `await`) /// provided their completion doesn't depend on any of the futures we're running. /// (These kind of operations are often discouraged in async contexts, /// because they block the async runtime or its worker threads. /// But they are often OK in tests.) /// /// * All facilities provided by this `MockExecutor` and its trait impls. /// /// #### Not allowed /// /// * Direct access to the real-world clock (`SystemTime::now`, `Instant::now`). /// Including `coarsetime`, which is not mocked. /// Exception: CPU use measurements. /// /// * Anything that spawns threads and then communicates with those threads /// using async Rust facilities (futures). /// /// * Async sockets, or async use of other kernel-based IPC or network mechanisms. /// /// * Anything provided by a Rust runtime/executor project (eg anything from Tokio), /// unless it is definitively established that it's runtime-agnostic. #[derive(Debug, Default, Clone, Getters, Adhoc)] #[derive_adhoc(SomeMockRuntime)] #[getter(prefix = "mock_")] pub struct MockRuntime { /// Tasks #[adhoc(mock(task))] task: MockExecutor, /// Time provider #[adhoc(mock(sleep))] sleep: MockSleepProvider, /// Net provider #[adhoc(mock(net))] net: MockNetProvider, } /// Builder for a manually-configured `MockRuntime` #[derive(Debug, Default, Clone)] pub struct MockRuntimeBuilder { /// scheduling policy scheduling: SchedulingPolicy, /// starting wall clock time starting_wallclock: Option, } impl MockRuntime { /// Create a new `MockRuntime` with default parameters pub fn new() -> Self { Self::default() } /// Return a builder, for creating a `MockRuntime` with some parameters manually configured pub fn builder() -> MockRuntimeBuilder { Default::default() } /// Run a test case with a variety of runtime parameters, to try to find bugs /// /// `test_case` is an async closure which receives a `MockRuntime`. /// It will be run with a number of differently configured executors. /// /// ### Variations /// /// The only variation currently implemented is this: /// /// Both FIFO and LIFO scheduling policies are tested, /// in the hope that this will help discover ordering-dependent bugs. pub fn test_with_various(mut test_case: TC) where TC: FnMut(MockRuntime) -> FUT, FUT: Future, { Self::try_test_with_various(|runtime| test_case(runtime).map(|()| Ok::<_, Void>(()))) .void_unwrap(); } /// Run a faillible test case with a variety of runtime parameters, to try to find bugs /// /// `test_case` is an async closure which receives a `MockRuntime`. /// It will be run with a number of differently configured executors. /// /// This function accepts a fallible closure, /// and returns the first `Err` to the caller. /// /// See [`test_with_various()`](MockRuntime::test_with_various) for more details. pub fn try_test_with_various(mut test_case: TC) -> Result<(), E> where TC: FnMut(MockRuntime) -> FUT, FUT: Future>, { for scheduling in SchedulingPolicy::iter() { let runtime = MockRuntime::builder().scheduling(scheduling).build(); runtime.block_on(test_case(runtime.clone()))?; } Ok(()) } /// Spawn a task and return something to identify it /// /// See [`MockExecutor::spawn_identified()`] pub fn spawn_identified( &self, desc: impl Display, fut: impl Future + Send + 'static, ) -> impl Debug + Clone + Send + 'static { // MSRV: 1.65 cannot cope and erroneously claims that desc has to be 'static let desc = desc.to_string(); self.task.spawn_identified(desc, fut) } /// Run tasks and advance time, until every task except this one is waiting /// /// On return the other tasks won't be waiting on timeouts, /// since time will be advanced as needed. /// /// Therefore the other tasks (if any) will be waiting for something /// that won't happen by itself, /// such as a provocation via their APIs from this task. /// /// # Panics /// /// See [`progress_until_stalled`](MockRuntime::progress_until_stalled) pub async fn advance_until_stalled(&self) { loop { self.progress_until_stalled().await; let Some(timeout) = self.time_until_next_timeout() else { // Nothing is waiting on timeouts return; }; assert_ne!(timeout, Duration::ZERO); self.sleep.advance(timeout).await; } } /// Run tasks in the current executor until every task except this one is waiting /// /// Calls [`MockExecutor::progress_until_stalled()`]. /// /// # Restriction - no automatic time advance /// /// The mocked time will *not* be automatically advanced. /// /// Usually /// (and especially if the tasks under test are waiting for timeouts or periodic events) /// you must use /// [`advance()`](MockRuntime::advance) /// or /// [`jump_to()`](MockRuntime::jump_to) /// to ensure the simulated time progresses as required. /// /// # Panics /// /// Might malfunction or panic if more than one such call is running at once. /// /// (Ie, you must `.await` or drop the returned `Future` /// before calling this method again.) /// /// Must be called and awaited within a future being run by `self`. pub async fn progress_until_stalled(&self) { self.task.progress_until_stalled().await; } /// Run tasks and advance time up to at most `limit` /// /// Will return when all other tasks are either: /// * Waiting on a timeout that will fire strictly after `limit`, /// (return value is the time until the earliest such) /// * Waiting for something else that won't happen by itself. /// (return value is `None`) /// /// Like [`advance_until_stalled`](MockRuntime::advance_until_stalled) /// but stops when the mock time reaches `limit`. /// /// # Panics /// /// Panics if the time somehow advances beyond `limit`. /// (This function won't do that, but maybe it was beyond `limit` on entry, /// or another task advanced the clock.) /// /// And, see [`progress_until_stalled`](MockRuntime::progress_until_stalled) pub async fn advance_until(&self, limit: Instant) -> Option { loop { self.task.progress_until_stalled().await; let timeout = self.time_until_next_timeout(); let limit = limit .checked_duration_since(self.now()) .expect("MockRuntime::advance_until: time advanced beyond `limit`!"); if limit == Duration::ZERO { // Time has reached `limit` return timeout; } let advance = chain!(timeout, [limit]).min().expect("empty!"); assert_ne!(advance, Duration::ZERO); self.sleep.advance(advance).await; } } /// Advances time by `dur`, firing time events and other tasks in order /// /// Prefer this to [`MockSleepProvider::advance()`]; /// it works more correctly. /// /// Specifically, it advances time in successive stages, /// so that timeouts occur sequentially, in the right order. /// /// # Panics /// /// Can panic if the mock time is advanced by other tasks. /// /// And, see [`progress_until_stalled`](MockRuntime::progress_until_stalled) pub async fn advance_by(&self, dur: Duration) -> Option { let limit = self .now() .checked_add(dur) .expect("MockRuntime::advance: time overflow"); self.advance_until(limit).await } /// See [`MockSleepProvider::jump_to()`] pub fn jump_to(&self, new_wallclock: SystemTime) { self.sleep.jump_to(new_wallclock); } /// Return the amount of virtual time until the next timeout /// should elapse. /// /// If there are no more timeouts, return None. /// /// If the next /// timeout should elapse right now, return Some(0). /// However, if other tasks are proceeding, /// typically in that situation those other tasks will wake, /// so a `Some(0)` return won't be visible. /// In test cases, detect immediate timeouts by detecting /// what your task does after the timeout occurs. /// /// Likewise whether this function returns `None` or `Some(...)` /// can depend on whether tasks have actually yet polled various futures. /// The answer should be correct after /// [`progress_until_stalled`](Self::progress_until_stalled). // // The corresponding method on MockSleepProvider is still only pub(crate), // mostly because I think's very hard (or maybe impossible) to use correctly, // other than as part of the implementation, without `progress_until_stalled`. pub fn time_until_next_timeout(&self) -> Option { self.sleep.time_until_next_timeout() } /// Dispatches to [`MockSleepProvider::advance()`] /// /// ### Deprecated /// /// Usually, you will want `MockRuntime::advance_by`, /// which advances the time in stages, /// ensuring that all timeouts trigger in the expected sequence. /// Changing from `advance` to `advance_by` should be correct, /// but it can expose new bugs. /// /// If you want to step the time in one go, /// use `runtime.mock_sleep().advance()`. #[deprecated(note = "use MockRuntime::advance_by, or MockSleepProvider::advance()")] pub async fn advance(&self, dur: Duration) { self.sleep.advance(dur).await; } } impl MockRuntimeBuilder { /// Set the scheduling policy pub fn scheduling(mut self, scheduling: SchedulingPolicy) -> Self { self.scheduling = scheduling; self } /// Set the starting wall clock time pub fn starting_wallclock(mut self, starting_wallclock: SystemTime) -> Self { self.starting_wallclock = Some(starting_wallclock); self } /// Build the runtime pub fn build(self) -> MockRuntime { let MockRuntimeBuilder { scheduling, starting_wallclock, } = self; let sleep = if let Some(starting_wallclock) = starting_wallclock { MockSleepProvider::new(starting_wallclock) } else { MockSleepProvider::default() }; let task = MockExecutor::with_scheduling(scheduling); MockRuntime { sleep, task, ..Default::default() } } } #[cfg(test)] mod test { // @@ begin test lint list maintained by maint/add_warning @@ #![allow(clippy::bool_assert_comparison)] #![allow(clippy::clone_on_copy)] #![allow(clippy::dbg_macro)] #![allow(clippy::print_stderr)] #![allow(clippy::print_stdout)] #![allow(clippy::single_char_pattern)] #![allow(clippy::unwrap_used)] #![allow(clippy::unchecked_duration_subtraction)] #![allow(clippy::useless_vec)] #![allow(clippy::needless_pass_by_value)] //! use super::*; use futures::channel::mpsc; use futures::{SinkExt as _, StreamExt as _}; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering::SeqCst; use std::sync::Arc; use tracing::trace; use tracing_test::traced_test; //---------- helper alias ---------- fn ms(i: u64) -> Duration { Duration::from_millis(i) } //---------- set up some test tasks ---------- struct TestTasks { runtime: MockRuntime, start: Instant, tx: mpsc::Sender<()>, signals: Vec>, } impl TestTasks { fn spawn(runtime: &MockRuntime) -> TestTasks { let start = runtime.now(); let mut signals = vec![]; let mut new_signal = || { let signal = Arc::new(AtomicBool::new(false)); signals.push(signal.clone()); signal }; let (tx, mut rx) = mpsc::channel(0); runtime.spawn_identified("rx", { let signal = new_signal(); async move { trace!("task rx starting..."); let _: Option<()> = rx.next().await; signal.store(true, SeqCst); trace!("task rx finished."); } }); for i in 1..=3 { let signal = new_signal(); runtime.spawn_identified(i, { let runtime = runtime.clone(); async move { trace!("task {i} starting..."); runtime.sleep(ms(i * 1000)).await; signal.store(true, SeqCst); trace!("task {i} finished."); } }); } let runtime = runtime.clone(); TestTasks { runtime, start, tx, signals, } } fn signals_list(&self) -> String { self.signals .iter() .map(|s| if s.load(SeqCst) { 't' } else { 'f' }) .collect() } } //---------- test advance_until_stalled ---------- impl TestTasks { async fn advance_until_stalled(&self, exp_offset_from_start: Duration, exp_signals: &str) { self.runtime.advance_until_stalled().await; assert_eq!(self.runtime.now() - self.start, exp_offset_from_start); assert_eq!(self.signals_list(), exp_signals); } } #[traced_test] #[test] fn advance_until_stalled() { MockRuntime::test_with_various(|runtime| async move { let mut tt = TestTasks::spawn(&runtime); tt.advance_until_stalled(ms(3000), "fttt").await; tt.tx.send(()).await.unwrap(); tt.advance_until_stalled(ms(3000), "tttt").await; }); } //---------- test advance_until ---------- impl TestTasks { async fn advance_until( &self, offset_from_start: Duration, exp_signals: &str, exp_got: Option, ) { let limit = self.start + offset_from_start; eprintln!("===> advance_until {}ms", offset_from_start.as_millis()); let got = self.runtime.advance_until(limit).await; assert_eq!(self.runtime.now() - self.start, offset_from_start); assert_eq!(self.signals_list(), exp_signals); assert_eq!(got, exp_got); } } #[traced_test] #[test] fn advance_until() { MockRuntime::test_with_various(|runtime| async move { let mut tt = TestTasks::spawn(&runtime); tt.advance_until(ms(1100), "ftff", Some(ms(900))).await; tt.advance_until(ms(2000), "fttf", Some(ms(1000))).await; tt.tx.send(()).await.unwrap(); tt.advance_until(ms(2000), "tttf", Some(ms(1000))).await; tt.advance_until(ms(3300), "tttt", None).await; }); } //---------- test advance_by ---------- impl TestTasks { async fn advance_by( &self, advance: Duration, exp_offset_from_start: Duration, exp_signals: &str, exp_got: Option, ) { eprintln!("===> advance {}ms", advance.as_millis()); let got = self.runtime.advance_by(advance).await; assert_eq!(self.runtime.now() - self.start, exp_offset_from_start); assert_eq!(self.signals_list(), exp_signals); assert_eq!(got, exp_got); } } #[traced_test] #[test] fn advance_by() { MockRuntime::test_with_various(|runtime| async move { let mut tt = TestTasks::spawn(&runtime); tt.advance_by(ms(1100), ms(1100), "ftff", Some(ms(900))) .await; tt.advance_by(ms(900), ms(2000), "fttf", Some(ms(1000))) .await; tt.tx.send(()).await.unwrap(); tt.advance_by(ms(1300), ms(3300), "tttt", None).await; }); } }