From db620b42ec8330ed33996edd7ce00d61ad947a4b Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 6 Jun 2018 16:04:39 -0700 Subject: [PATCH] Another attempt at abstracting `Instant::now` (#381) Currently, the timer uses a `Now` trait to abstract the source of time. This allows time to be mocked out. However, the current implementation has a number of limitations as represented by #288 and #296. The main issues are that `Now` requires `&mut self` which prevents a value from being easily used in a concurrent environment. Also, when wanting to write code that is abstract over the source of time, generics get out of hand. This patch provides an alternate solution. A new type, `Clock` is provided which defaults to `Instant::now` as the source of time, but allows configuring the actual source using a new iteration of the `Now` trait. This time, `Now` is `Send + Sync + 'static`. Internally, `Clock` stores the now value in an `Arc` value, which introduces dynamism and allows `Clock` values to be cloned and be `Sync`. Also, the current clock can be set for the current execution context using the `with_default` pattern. Because using the `Instant::now` will be the most common case by far, it is special cased in order to avoid the need to allocate an `Arc` and use dynamic dispatch. --- src/clock.rs | 15 +++ src/lib.rs | 1 + src/runtime/builder.rs | 29 +++++- src/runtime/current_thread/builder.rs | 88 ++++++++++++++++ src/runtime/current_thread/mod.rs | 2 + src/runtime/current_thread/runtime.rs | 62 +++++++----- tests/clock.rs | 69 +++++++++++++ tokio-timer/src/clock/clock.rs | 138 ++++++++++++++++++++++++++ tokio-timer/src/clock/mod.rs | 22 ++++ tokio-timer/src/clock/now.rs | 13 +++ tokio-timer/src/lib.rs | 1 + tokio-timer/src/timer/mod.rs | 3 + tokio-timer/src/timer/now.rs | 23 +---- tokio-timer/tests/clock.rs | 51 ++++++++++ tokio-timer/tests/support/mod.rs | 2 +- 15 files changed, 467 insertions(+), 52 deletions(-) create mode 100644 src/clock.rs create mode 100644 src/runtime/current_thread/builder.rs create mode 100644 tests/clock.rs create mode 100644 tokio-timer/src/clock/clock.rs create mode 100644 tokio-timer/src/clock/mod.rs create mode 100644 tokio-timer/src/clock/now.rs create mode 100644 tokio-timer/tests/clock.rs diff --git a/src/clock.rs b/src/clock.rs new file mode 100644 index 000000000..313416690 --- /dev/null +++ b/src/clock.rs @@ -0,0 +1,15 @@ +//! A configurable source of time. +//! +//! This module provides the [`now`][n] function, which returns an `Instant` +//! representing "now". The source of time used by this function is configurable +//! (via the [`tokio-timer`] crate) and allows mocking out the source of time in +//! tests or performing caching operations to reduce the number of syscalls. +//! +//! Note that, because the source of time is configurable, it is possible to +//! observe non-monotonic behavior when calling [`now`] from different +//! executors. +//! +//! [n]: fn.now.html +//! [`tokio-timer`]: https://docs.rs/tokio-timer/0.2/tokio_timer/clock/index.html + +pub use tokio_timer::clock::now; diff --git a/src/lib.rs b/src/lib.rs index 96449bae8..7ff0b2fd1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -82,6 +82,7 @@ extern crate tokio_udp; #[cfg(feature = "unstable-futures")] extern crate futures2; +pub mod clock; pub mod executor; pub mod fs; pub mod net; diff --git a/src/runtime/builder.rs b/src/runtime/builder.rs index 6953426dd..c4e048b43 100644 --- a/src/runtime/builder.rs +++ b/src/runtime/builder.rs @@ -7,6 +7,7 @@ use std::io; use tokio_reactor; use tokio_threadpool::Builder as ThreadPoolBuilder; use tokio_threadpool::park::DefaultPark; +use tokio_timer::clock::{self, Clock}; use tokio_timer::timer::{self, Timer}; /// Builds Tokio Runtime with custom configuration values. @@ -48,6 +49,9 @@ use tokio_timer::timer::{self, Timer}; pub struct Builder { /// Thread pool specific builder threadpool_builder: ThreadPoolBuilder, + + /// The clock to use + clock: Clock, } impl Builder { @@ -59,7 +63,16 @@ impl Builder { let mut threadpool_builder = ThreadPoolBuilder::new(); threadpool_builder.name_prefix("tokio-runtime-worker-"); - Builder { threadpool_builder } + Builder { + threadpool_builder, + clock: Clock::new(), + } + } + + /// Set the `Clock` instance that will be used by the runtime. + pub fn clock(&mut self, clock: Clock) -> &mut Self { + self.clock = clock; + self } /// Set builder to set up the thread pool instance. @@ -87,6 +100,10 @@ impl Builder { use std::collections::HashMap; use std::sync::{Arc, Mutex}; + // Get a handle to the clock for the runtime. + let clock1 = self.clock.clone(); + let clock2 = clock1.clone(); + let timers = Arc::new(Mutex::new(HashMap::<_, timer::Handle>::new())); let t1 = timers.clone(); @@ -103,14 +120,16 @@ impl Builder { .clone(); tokio_reactor::with_default(&reactor_handle, enter, |enter| { - timer::with_default(&timer_handle, enter, |_| { - w.run(); - }); + clock::with_default(&clock1, enter, |enter| { + timer::with_default(&timer_handle, enter, |_| { + w.run(); + }); + }) }); }) .custom_park(move |worker_id| { // Create a new timer - let timer = Timer::new(DefaultPark::new()); + let timer = Timer::new_with_now(DefaultPark::new(), clock2.clone()); timers.lock().unwrap() .insert(worker_id.clone(), timer.handle()); diff --git a/src/runtime/current_thread/builder.rs b/src/runtime/current_thread/builder.rs new file mode 100644 index 000000000..72960fadf --- /dev/null +++ b/src/runtime/current_thread/builder.rs @@ -0,0 +1,88 @@ +use executor::current_thread::CurrentThread; +use runtime::current_thread::Runtime; + +use tokio_reactor::Reactor; +use tokio_timer::clock::Clock; +use tokio_timer::timer::Timer; + +use std::io; + +/// Builds a Single-threaded runtime with custom configuration values. +/// +/// Methods can be chained in order to set the configuration values. The +/// Runtime is constructed by calling [`build`]. +/// +/// New instances of `Builder` are obtained via [`Builder::new`]. +/// +/// See function level documentation for details on the various configuration +/// settings. +/// +/// [`build`]: #method.build +/// [`Builder::new`]: #method.new +/// +/// # Examples +/// +/// ``` +/// extern crate tokio; +/// extern crate tokio_timer; +/// +/// use tokio::runtime::current_thread::Builder; +/// use tokio_timer::clock::Clock; +/// +/// # pub fn main() { +/// // build Runtime +/// let runtime = Builder::new() +/// .clock(Clock::new()) +/// .build(); +/// // ... call runtime.run(...) +/// # let _ = runtime; +/// # } +/// ``` +#[derive(Debug)] +pub struct Builder { + /// The clock to use + clock: Clock, +} + +impl Builder { + /// Returns a new runtime builder initialized with default configuration + /// values. + /// + /// Configuration methods can be chained on the return value. + pub fn new() -> Builder { + Builder { + clock: Clock::new(), + } + } + + /// Set the `Clock` instance that will be used by the runtime. + pub fn clock(&mut self, clock: Clock) -> &mut Self { + self.clock = clock; + self + } + + /// Create the configured `Runtime`. + pub fn build(&mut self) -> io::Result { + // We need a reactor to receive events about IO objects from kernel + let reactor = Reactor::new()?; + let reactor_handle = reactor.handle(); + + // Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the + // reactor pick up some new external events. + let timer = Timer::new_with_now(reactor, self.clock.clone()); + let timer_handle = timer.handle(); + + // And now put a single-threaded executor on top of the timer. When there are no futures ready + // to do something, it'll let the timer or the reactor to generate some new stimuli for the + // futures to continue in their life. + let executor = CurrentThread::new_with_park(timer); + + let runtime = Runtime::new2( + reactor_handle, + timer_handle, + self.clock.clone(), + executor); + + Ok(runtime) + } +} diff --git a/src/runtime/current_thread/mod.rs b/src/runtime/current_thread/mod.rs index 90a90672c..30bd8f6dd 100644 --- a/src/runtime/current_thread/mod.rs +++ b/src/runtime/current_thread/mod.rs @@ -63,6 +63,8 @@ //! [concurrent-rt]: ../struct.Runtime.html //! [chan]: https://docs.rs/futures/0.1/futures/sync/mpsc/fn.channel.html +mod builder; mod runtime; +pub use self::builder::Builder; pub use self::runtime::{Runtime, Handle}; diff --git a/src/runtime/current_thread/runtime.rs b/src/runtime/current_thread/runtime.rs index 3e31f1006..8939f08cc 100644 --- a/src/runtime/current_thread/runtime.rs +++ b/src/runtime/current_thread/runtime.rs @@ -1,7 +1,9 @@ use executor::current_thread::{self, CurrentThread}; use executor::current_thread::Handle as ExecutorHandle; +use runtime::current_thread::Builder; use tokio_reactor::{self, Reactor}; +use tokio_timer::clock::{self, Clock}; use tokio_timer::timer::{self, Timer}; use tokio_executor; @@ -19,6 +21,7 @@ use std::io; pub struct Runtime { reactor_handle: tokio_reactor::Handle, timer_handle: timer::Handle, + clock: Clock, executor: CurrentThread>, } @@ -48,22 +51,21 @@ pub struct RunError { impl Runtime { /// Returns a new runtime initialized with default configuration values. pub fn new() -> io::Result { - // We need a reactor to receive events about IO objects from kernel - let reactor = Reactor::new()?; - let reactor_handle = reactor.handle(); + Builder::new().build() + } - // Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the - // reactor pick up some new external events. - let timer = Timer::new(reactor); - let timer_handle = timer.handle(); - - // And now put a single-threaded executor on top of the timer. When there are no futures ready - // to do something, it'll let the timer or the reactor to generate some new stimuli for the - // futures to continue in their life. - let executor = CurrentThread::new_with_park(timer); - - let runtime = Runtime { reactor_handle, timer_handle, executor }; - Ok(runtime) + pub(super) fn new2( + reactor_handle: tokio_reactor::Handle, + timer_handle: timer::Handle, + clock: Clock, + executor: CurrentThread>) -> Runtime + { + Runtime { + reactor_handle, + timer_handle, + clock, + executor, + } } /// Get a new handle to spawn futures on the single-threaded Tokio runtime @@ -150,7 +152,13 @@ impl Runtime { fn enter(&mut self, f: F) -> R where F: FnOnce(&mut current_thread::Entered>) -> R { - let Runtime { ref reactor_handle, ref timer_handle, ref mut executor, .. } = *self; + let Runtime { + ref reactor_handle, + ref timer_handle, + ref clock, + ref mut executor, + .. + } = *self; // Binds an executor to this thread let mut enter = tokio_executor::enter().expect("Multiple executors at once"); @@ -158,16 +166,18 @@ impl Runtime { // This will set the default handle and timer to use inside the closure // and run the future. tokio_reactor::with_default(&reactor_handle, &mut enter, |enter| { - timer::with_default(&timer_handle, enter, |enter| { - // The TaskExecutor is a fake executor that looks into the - // current single-threaded executor when used. This is a trick, - // because we need two mutable references to the executor (one - // to run the provided future, another to install as the default - // one). We use the fake one here as the default one. - let mut default_executor = current_thread::TaskExecutor::current(); - tokio_executor::with_default(&mut default_executor, enter, |enter| { - let mut executor = executor.enter(enter); - f(&mut executor) + clock::with_default(clock, enter, |enter| { + timer::with_default(&timer_handle, enter, |enter| { + // The TaskExecutor is a fake executor that looks into the + // current single-threaded executor when used. This is a trick, + // because we need two mutable references to the executor (one + // to run the provided future, another to install as the default + // one). We use the fake one here as the default one. + let mut default_executor = current_thread::TaskExecutor::current(); + tokio_executor::with_default(&mut default_executor, enter, |enter| { + let mut executor = executor.enter(enter); + f(&mut executor) + }) }) }) }) diff --git a/tests/clock.rs b/tests/clock.rs new file mode 100644 index 000000000..ee99329f0 --- /dev/null +++ b/tests/clock.rs @@ -0,0 +1,69 @@ +extern crate futures; +extern crate tokio; +extern crate tokio_timer; +extern crate env_logger; + +use tokio::prelude::*; +use tokio::runtime::{self, current_thread}; +use tokio::timer::*; +use tokio_timer::clock::Clock; + +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +struct MockNow(Instant); + +impl tokio_timer::clock::Now for MockNow { + fn now(&self) -> Instant { + self.0 + } +} + +#[test] +fn clock_and_timer_concurrent() { + let _ = env_logger::init(); + + let when = Instant::now() + Duration::from_millis(5_000); + let clock = Clock::new_with_now(MockNow(when)); + + let mut rt = runtime::Builder::new() + .clock(clock) + .build() + .unwrap(); + + let (tx, rx) = mpsc::channel(); + + rt.spawn({ + Delay::new(when) + .map_err(|e| panic!("unexpected error; err={:?}", e)) + .and_then(move |_| { + assert!(Instant::now() < when); + tx.send(()).unwrap(); + Ok(()) + }) + }); + + rx.recv().unwrap(); +} + +#[test] +fn clock_and_timer_single_threaded() { + let _ = env_logger::init(); + + let when = Instant::now() + Duration::from_millis(5_000); + let clock = Clock::new_with_now(MockNow(when)); + + let mut rt = current_thread::Builder::new() + .clock(clock) + .build() + .unwrap(); + + rt.block_on({ + Delay::new(when) + .map_err(|e| panic!("unexpected error; err={:?}", e)) + .and_then(move |_| { + assert!(Instant::now() < when); + Ok(()) + }) + }).unwrap(); +} diff --git a/tokio-timer/src/clock/clock.rs b/tokio-timer/src/clock/clock.rs new file mode 100644 index 000000000..76330c75f --- /dev/null +++ b/tokio-timer/src/clock/clock.rs @@ -0,0 +1,138 @@ +use clock::Now; +use timer; + +use tokio_executor::Enter; + +use std::cell::Cell; +use std::fmt; +use std::sync::Arc; +use std::time::Instant; + +/// A handle to a source of time. +/// +/// `Clock` instances return `Instant` values corresponding to "now". The source +/// of these values is configurable. The default source is `Instant::now()`. +#[derive(Default, Clone)] +pub struct Clock { + now: Option>, +} + +/// Thread-local tracking the current clock +thread_local!(static CLOCK: Cell> = Cell::new(None)); + +/// Returns an `Instant` corresponding to "now". +/// +/// This function delegates to the source of time configured for the current +/// execution context. By default, this is `Instant::now()`. +/// +/// Note that, because the source of time is configurable, it is possible to +/// observe non-monotonic behavior when calling [`now`] from different +/// executors. +/// +/// See [module](index.html) level documentation for more details. +/// +/// # Examples +/// +/// ``` +/// # use tokio_timer::clock; +/// let now = clock::now(); +/// ``` +pub fn now() -> Instant { + CLOCK.with(|current| { + match current.get() { + Some(ptr) => { + unsafe { (*ptr).now() } + } + None => Instant::now(), + } + }) +} + +impl Clock { + /// Return a new `Clock` instance that uses the current execution context's + /// source of time. + pub fn new() -> Clock { + CLOCK.with(|current| { + match current.get() { + Some(ptr) => { + unsafe { (*ptr).clone() } + } + None => Clock::system(), + } + }) + } + + /// Return a new `Clock` instance that uses `now` as the source of time. + pub fn new_with_now(now: T) -> Clock { + Clock { + now: Some(Arc::new(now)), + } + } + + /// Return a new `Clock` instance that uses `Instant::now()` as the source + /// of time. + pub fn system() -> Clock { + Clock { + now: None, + } + } + + /// Returns an instant corresponding to "now" by using the instance's source + /// of time. + pub fn now(&self) -> Instant { + match self.now { + Some(ref now) => now.now(), + None => Instant::now(), + } + } +} + +#[allow(deprecated)] +impl timer::Now for Clock { + fn now(&mut self) -> Instant { + Clock::now(self) + } +} + +impl fmt::Debug for Clock { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Clock") + .field("now", { + if self.now.is_some() { + &"Some(Arc)" + } else { + &"None" + } + }) + .finish() + } +} + +/// Set the default clock for the duration of the closure. +/// +/// # Panics +/// +/// This function panics if there already is a default clock set. +pub fn with_default(clock: &Clock, enter: &mut Enter, f: F) -> R +where F: FnOnce(&mut Enter) -> R +{ + CLOCK.with(|cell| { + assert!(cell.get().is_none(), "default clock already set for execution context"); + + // Ensure that the clock is removed from the thread-local context + // when leaving the scope. This handles cases that involve panicking. + struct Reset<'a>(&'a Cell>); + + impl<'a> Drop for Reset<'a> { + fn drop(&mut self) { + self.0.set(None); + } + } + + let _reset = Reset(cell); + + cell.set(Some(clock as *const Clock)); + + f(enter) + }) +} diff --git a/tokio-timer/src/clock/mod.rs b/tokio-timer/src/clock/mod.rs new file mode 100644 index 000000000..09c577949 --- /dev/null +++ b/tokio-timer/src/clock/mod.rs @@ -0,0 +1,22 @@ +//! A configurable source of time. +//! +//! This module provides an API to get the current instant in such a way that +//! the source of time may be configured. This allows mocking out the source of +//! time in tests. +//! +//! The [`now`][n] function returns the current `Instant`. By default, it delegates +//! to [`Instant::now`][std]. +//! +//! The source of time used by [`now`] can be configured by implementing the +//! [`Now`] trait and passing an instance to [`with_default`]. +//! +//! [n]: fn.now.html +//! [`Now`]: trait.Now.html +//! [std]: https://doc.rust-lang.org/std/time/struct.Instant.html +//! [`with_default`]: fn.with_default.html + +mod clock; +mod now; + +pub use self::clock::{Clock, now, with_default}; +pub use self::now::Now; diff --git a/tokio-timer/src/clock/now.rs b/tokio-timer/src/clock/now.rs new file mode 100644 index 000000000..65472543c --- /dev/null +++ b/tokio-timer/src/clock/now.rs @@ -0,0 +1,13 @@ +use std::time::Instant; + +/// Returns `Instant` values representing the current instant in time. +/// +/// This allows customizing the source of time which is especially useful for +/// testing. +/// +/// Implementations must ensure that calls to `now` return monotonically +/// increasing `Instant` values. +pub trait Now: Send + Sync + 'static { + /// Returns an instant corresponding to "now". + fn now(&self) -> Instant; +} diff --git a/tokio-timer/src/lib.rs b/tokio-timer/src/lib.rs index 1509423d7..1679ded25 100644 --- a/tokio-timer/src/lib.rs +++ b/tokio-timer/src/lib.rs @@ -26,6 +26,7 @@ extern crate tokio_executor; #[macro_use] extern crate futures; +pub mod clock; pub mod timer; mod atomic; diff --git a/tokio-timer/src/timer/mod.rs b/tokio-timer/src/timer/mod.rs index 60b1f0f82..9043d476e 100644 --- a/tokio-timer/src/timer/mod.rs +++ b/tokio-timer/src/timer/mod.rs @@ -27,6 +27,9 @@ //! [`Now`]: trait.Now.html //! [`Now::now`]: trait.Now.html#method.now +// This allows the usage of the old `Now` trait. +#![allow(deprecated)] + mod entry; mod handle; mod level; diff --git a/tokio-timer/src/timer/now.rs b/tokio-timer/src/timer/now.rs index 244cb650a..bc8ca7808 100644 --- a/tokio-timer/src/timer/now.rs +++ b/tokio-timer/src/timer/now.rs @@ -1,27 +1,10 @@ use std::time::Instant; -/// Returns `Instant` values representing the current instant in time. -/// -/// This allows customizing the source of time which is especially useful for -/// testing. +#[doc(hidden)] +#[deprecated(since = "0.2.4", note = "use clock::Now instead")] pub trait Now { /// Returns an instant corresponding to "now". fn now(&mut self) -> Instant; } -/// Returns the instant corresponding to now using a monotonic clock. -#[derive(Debug)] -pub struct SystemNow(()); - -impl SystemNow { - /// Create a new `SystemNow`. - pub fn new() -> SystemNow { - SystemNow(()) - } -} - -impl Now for SystemNow { - fn now(&mut self) -> Instant { - Instant::now() - } -} +pub use ::clock::Clock as SystemNow; diff --git a/tokio-timer/tests/clock.rs b/tokio-timer/tests/clock.rs new file mode 100644 index 000000000..d8bcaaebd --- /dev/null +++ b/tokio-timer/tests/clock.rs @@ -0,0 +1,51 @@ +extern crate tokio_executor; +extern crate tokio_timer; + +use tokio_timer::clock; +use tokio_timer::clock::*; + +use std::time::Instant; + +struct ConstNow(Instant); + +impl Now for ConstNow { + fn now(&self) -> Instant { + self.0 + } +} + +#[test] +fn default_clock() { + let a = Instant::now(); + let b = clock::now(); + let c = Clock::new().now(); + + assert!(a <= b); + assert!(b <= c); +} + +#[test] +fn custom_clock() { + let now = ConstNow(Instant::now()); + let clock = Clock::new_with_now(now); + + let a = Instant::now(); + let b = clock.now(); + + assert!(b <= a); +} + +#[test] +fn execution_context() { + let now = ConstNow(Instant::now()); + let clock = Clock::new_with_now(now); + + let mut enter = tokio_executor::enter().unwrap(); + + with_default(&clock, &mut enter, |_| { + let a = Instant::now(); + let b = clock::now(); + + assert!(b <= a); + }); +} diff --git a/tokio-timer/tests/support/mod.rs b/tokio-timer/tests/support/mod.rs index 8afeb0d90..a79b0dab0 100644 --- a/tokio-timer/tests/support/mod.rs +++ b/tokio-timer/tests/support/mod.rs @@ -1,4 +1,4 @@ -#![allow(unused_macros, unused_imports, dead_code)] +#![allow(unused_macros, unused_imports, dead_code, deprecated)] use tokio_executor::park::{Park, Unpark}; use tokio_timer::timer::{Timer, Now};