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<Now>` 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

@@ -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;

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;

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());

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<Runtime> {
+        // 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)
+    }
+}

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};

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<Timer<Reactor>>,
 }
 
@@ -48,22 +51,21 @@ pub struct RunError {
 impl Runtime {
     /// Returns a new runtime initialized with default configuration values.
     pub fn new() -> io::Result<Runtime> {
-        // 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<Timer<Reactor>>) -> 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<F, R>(&mut self, f: F) -> R
     where F: FnOnce(&mut current_thread::Entered<Timer<Reactor>>) -> 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)
+                    })
                 })
             })
         })

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();
+}

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<Arc<Now>>,
+}
+
+/// Thread-local tracking the current clock
+thread_local!(static CLOCK: Cell<Option<*const Clock>> = 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<T: 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<Now>)"
+                } 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<F, R>(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<Option<*const Clock>>);
+
+        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)
+    })
+}

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;

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;
+}

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;

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;

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;

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);
+    });
+}

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};