reorganize modules (#1766)

This patch started as an effort to make `time::Timer` private. However, in an
effort to get the build compiling again, more and more changes were made. This
probably should have been broken up, but here we are. I will attempt to
summarize the changes here.

* Feature flags are reorganized to make clearer. `net-driver` becomes
  `io-driver`. `rt-current-thread` becomes `rt-core`.

* The `Runtime` can be created without any executor. This replaces `enter`. It
  also allows creating I/O / time drivers that are standalone.

* `tokio::timer` is renamed to `tokio::time`. This brings it in line with `std`.

* `tokio::timer::Timer` is renamed to `Driver` and made private.

* The `clock` module is removed. Instead, an `Instant` type is provided. This
  type defaults to calling `std::time::Instant`. A `test-util` feature flag can
  be used to enable hooking into time.

* The `blocking` module is moved to the top level and is cleaned up.

* The `task` module is moved to the top level.

* The thread-pool's in-place blocking implementation is cleaned up.

* `runtime::Spawner` is renamed to `runtime::Handle` and can be used to "enter"
  a runtime context.

6 files changed, 1190 insertions, 0 deletions

diff --git a/tokio/src/time/driver/atomic_stack.rs b/tokio/src/time/driver/atomic_stack.rs
new file mode 100644
index 00000000..036d283d
--- /dev/null
+++ b/tokio/src/time/driver/atomic_stack.rs
@@ -0,0 +1,124 @@
+use crate::time::driver::Entry;
+use crate::time::Error;
+
+use std::ptr;
+use std::sync::atomic::AtomicPtr;
+use std::sync::atomic::Ordering::SeqCst;
+use std::sync::Arc;
+
+/// A stack of `Entry` nodes
+#[derive(Debug)]
+pub(crate) struct AtomicStack {
+    /// Stack head
+    head: AtomicPtr<Entry>,
+}
+
+/// Entries that were removed from the stack
+#[derive(Debug)]
+pub(crate) struct AtomicStackEntries {
+    ptr: *mut Entry,
+}
+
+/// Used to indicate that the timer has shutdown.
+const SHUTDOWN: *mut Entry = 1 as *mut _;
+
+impl AtomicStack {
+    pub(crate) fn new() -> AtomicStack {
+        AtomicStack {
+            head: AtomicPtr::new(ptr::null_mut()),
+        }
+    }
+
+    /// Push an entry onto the stack.
+    ///
+    /// Returns `true` if the entry was pushed, `false` if the entry is already
+    /// on the stack, `Err` if the timer is shutdown.
+    pub(crate) fn push(&self, entry: &Arc<Entry>) -> Result<bool, Error> {
+        // First, set the queued bit on the entry
+        let queued = entry.queued.fetch_or(true, SeqCst);
+
+        if queued {
+            // Already queued, nothing more to do
+            return Ok(false);
+        }
+
+        let ptr = Arc::into_raw(entry.clone()) as *mut _;
+
+        let mut curr = self.head.load(SeqCst);
+
+        loop {
+            if curr == SHUTDOWN {
+                // Don't leak the entry node
+                let _ = unsafe { Arc::from_raw(ptr) };
+
+                return Err(Error::shutdown());
+            }
+
+            // Update the `next` pointer. This is safe because setting the queued
+            // bit is a "lock" on this field.
+            unsafe {
+                *(entry.next_atomic.get()) = curr;
+            }
+
+            let actual = self.head.compare_and_swap(curr, ptr, SeqCst);
+
+            if actual == curr {
+                break;
+            }
+
+            curr = actual;
+        }
+
+        Ok(true)
+    }
+
+    /// Take all entries from the stack
+    pub(crate) fn take(&self) -> AtomicStackEntries {
+        let ptr = self.head.swap(ptr::null_mut(), SeqCst);
+        AtomicStackEntries { ptr }
+    }
+
+    /// Drain all remaining nodes in the stack and prevent any new nodes from
+    /// being pushed onto the stack.
+    pub(crate) fn shutdown(&self) {
+        // Shutdown the processing queue
+        let ptr = self.head.swap(SHUTDOWN, SeqCst);
+
+        // Let the drop fn of `AtomicStackEntries` handle draining the stack
+        drop(AtomicStackEntries { ptr });
+    }
+}
+
+// ===== impl AtomicStackEntries =====
+
+impl Iterator for AtomicStackEntries {
+    type Item = Arc<Entry>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.ptr.is_null() {
+            return None;
+        }
+
+        // Convert the pointer to an `Arc<Entry>`
+        let entry = unsafe { Arc::from_raw(self.ptr) };
+
+        // Update `self.ptr` to point to the next element of the stack
+        self.ptr = unsafe { (*entry.next_atomic.get()) };
+
+        // Unset the queued flag
+        let res = entry.queued.fetch_and(false, SeqCst);
+        debug_assert!(res);
+
+        // Return the entry
+        Some(entry)
+    }
+}
+
+impl Drop for AtomicStackEntries {
+    fn drop(&mut self) {
+        for entry in self {
+            // Flag the entry as errored
+            entry.error();
+        }
+    }
+}
diff --git a/tokio/src/time/driver/entry.rs b/tokio/src/time/driver/entry.rs
new file mode 100644
index 00000000..97ce34de
--- /dev/null
+++ b/tokio/src/time/driver/entry.rs
@@ -0,0 +1,395 @@
+use crate::loom::sync::atomic::AtomicU64;
+use crate::sync::AtomicWaker;
+use crate::time::driver::{HandlePriv, Inner};
+use crate::time::{Duration, Error, Instant};
+
+use std::cell::UnsafeCell;
+use std::ptr;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering::{Relaxed, SeqCst};
+use std::sync::{Arc, Weak};
+use std::task::{self, Poll};
+use std::u64;
+
+/// Internal state shared between a `Delay` instance and the timer.
+///
+/// This struct is used as a node in two intrusive data structures:
+///
+/// * An atomic stack used to signal to the timer thread that the entry state
+///   has changed. The timer thread will observe the entry on this stack and
+///   perform any actions as necessary.
+///
+/// * A doubly linked list used **only** by the timer thread. Each slot in the
+///   timer wheel is a head pointer to the list of entries that must be
+///   processed during that timer tick.
+#[derive(Debug)]
+pub(crate) struct Entry {
+    /// Only accessed from `Registration`.
+    time: CachePadded<UnsafeCell<Time>>,
+
+    /// Timer internals. Using a weak pointer allows the timer to shutdown
+    /// without all `Delay` instances having completed.
+    ///
+    /// When `None`, the entry has not yet been linked with a timer instance.
+    inner: Option<Weak<Inner>>,
+
+    /// Tracks the entry state. This value contains the following information:
+    ///
+    /// * The deadline at which the entry must be "fired".
+    /// * A flag indicating if the entry has already been fired.
+    /// * Whether or not the entry transitioned to the error state.
+    ///
+    /// When an `Entry` is created, `state` is initialized to the instant at
+    /// which the entry must be fired. When a timer is reset to a different
+    /// instant, this value is changed.
+    state: AtomicU64,
+
+    /// Task to notify once the deadline is reached.
+    waker: AtomicWaker,
+
+    /// True when the entry is queued in the "process" stack. This value
+    /// is set before pushing the value and unset after popping the value.
+    ///
+    /// TODO: This could possibly be rolled up into `state`.
+    pub(super) queued: AtomicBool,
+
+    /// Next entry in the "process" linked list.
+    ///
+    /// Access to this field is coordinated by the `queued` flag.
+    ///
+    /// Represents a strong Arc ref.
+    pub(super) next_atomic: UnsafeCell<*mut Entry>,
+
+    /// When the entry expires, relative to the `start` of the timer
+    /// (Inner::start). This is only used by the timer.
+    ///
+    /// A `Delay` instance can be reset to a different deadline by the thread
+    /// that owns the `Delay` instance. In this case, the timer thread will not
+    /// immediately know that this has happened. The timer thread must know the
+    /// last deadline that it saw as it uses this value to locate the entry in
+    /// its wheel.
+    ///
+    /// Once the timer thread observes that the instant has changed, it updates
+    /// the wheel and sets this value. The idea is that this value eventually
+    /// converges to the value of `state` as the timer thread makes updates.
+    when: UnsafeCell<Option<u64>>,
+
+    /// Next entry in the State's linked list.
+    ///
+    /// This is only accessed by the timer
+    pub(super) next_stack: UnsafeCell<Option<Arc<Entry>>>,
+
+    /// Previous entry in the State's linked list.
+    ///
+    /// This is only accessed by the timer and is used to unlink a canceled
+    /// entry.
+    ///
+    /// This is a weak reference.
+    pub(super) prev_stack: UnsafeCell<*const Entry>,
+}
+
+/// Stores the info for `Delay`.
+#[derive(Debug)]
+pub(crate) struct Time {
+    pub(crate) deadline: Instant,
+    pub(crate) duration: Duration,
+}
+
+/// Flag indicating a timer entry has elapsed
+const ELAPSED: u64 = 1 << 63;
+
+/// Flag indicating a timer entry has reached an error state
+const ERROR: u64 = u64::MAX;
+
+// ===== impl Entry =====
+
+impl Entry {
+    pub(crate) fn new(deadline: Instant, duration: Duration) -> Entry {
+        Entry {
+            time: CachePadded(UnsafeCell::new(Time { deadline, duration })),
+            inner: None,
+            waker: AtomicWaker::new(),
+            state: AtomicU64::new(0),
+            queued: AtomicBool::new(false),
+            next_atomic: UnsafeCell::new(ptr::null_mut()),
+            when: UnsafeCell::new(None),
+            next_stack: UnsafeCell::new(None),
+            prev_stack: UnsafeCell::new(ptr::null_mut()),
+        }
+    }
+
+    /// Only called by `Registration`
+    pub(crate) fn time_ref(&self) -> &Time {
+        unsafe { &*self.time.0.get() }
+    }
+
+    /// Only called by `Registration`
+    #[allow(clippy::mut_from_ref)] // https://github.com/rust-lang/rust-clippy/issues/4281
+    pub(crate) unsafe fn time_mut(&self) -> &mut Time {
+        &mut *self.time.0.get()
+    }
+
+    /// Returns `true` if the `Entry` is currently associated with a timer
+    /// instance.
+    pub(crate) fn is_registered(&self) -> bool {
+        self.inner.is_some()
+    }
+
+    /// Only called by `Registration`
+    pub(crate) fn register(me: &mut Arc<Self>) {
+        let handle = match HandlePriv::try_current() {
+            Ok(handle) => handle,
+            Err(_) => {
+                // Could not associate the entry with a timer, transition the
+                // state to error
+                Arc::get_mut(me).unwrap().transition_to_error();
+
+                return;
+            }
+        };
+
+        Entry::register_with(me, handle)
+    }
+
+    /// Only called by `Registration`
+    pub(crate) fn register_with(me: &mut Arc<Self>, handle: HandlePriv) {
+        assert!(!me.is_registered(), "only register an entry once");
+
+        let deadline = me.time_ref().deadline;
+
+        let inner = match handle.inner() {
+            Some(inner) => inner,
+            None => {
+                // Could not associate the entry with a timer, transition the
+                // state to error
+                Arc::get_mut(me).unwrap().transition_to_error();
+
+                return;
+            }
+        };
+
+        // Increment the number of active timeouts
+        if inner.increment().is_err() {
+            Arc::get_mut(me).unwrap().transition_to_error();
+
+            return;
+        }
+
+        // Associate the entry with the timer
+        Arc::get_mut(me).unwrap().inner = Some(handle.into_inner());
+
+        let when = inner.normalize_deadline(deadline);
+
+        // Relaxed OK: At this point, there are no other threads that have
+        // access to this entry.
+        if when <= inner.elapsed() {
+            me.state.store(ELAPSED, Relaxed);
+            return;
+        } else {
+            me.state.store(when, Relaxed);
+        }
+
+        if inner.queue(me).is_err() {
+            // The timer has shutdown, transition the entry to the error state.
+            me.error();
+        }
+    }
+
+    fn transition_to_error(&mut self) {
+        self.inner = Some(Weak::new());
+        self.state = AtomicU64::new(ERROR);
+    }
+
+    /// The current entry state as known by the timer. This is not the value of
+    /// `state`, but lets the timer know how to converge its state to `state`.
+    pub(crate) fn when_internal(&self) -> Option<u64> {
+        unsafe { (*self.when.get()) }
+    }
+
+    pub(crate) fn set_when_internal(&self, when: Option<u64>) {
+        unsafe {
+            (*self.when.get()) = when;
+        }
+    }
+
+    /// Called by `Timer` to load the current value of `state` for processing
+    pub(crate) fn load_state(&self) -> Option<u64> {
+        let state = self.state.load(SeqCst);
+
+        if is_elapsed(state) {
+            None
+        } else {
+            Some(state)
+        }
+    }
+
+    pub(crate) fn is_elapsed(&self) -> bool {
+        let state = self.state.load(SeqCst);
+        is_elapsed(state)
+    }
+
+    pub(crate) fn fire(&self, when: u64) {
+        let mut curr = self.state.load(SeqCst);
+
+        loop {
+            if is_elapsed(curr) || curr > when {
+                return;
+            }
+
+            let next = ELAPSED | curr;
+            let actual = self.state.compare_and_swap(curr, next, SeqCst);
+
+            if curr == actual {
+                break;
+            }
+
+            curr = actual;
+        }
+
+        self.waker.wake();
+    }
+
+    pub(crate) fn error(&self) {
+        // Only transition to the error state if not currently elapsed
+        let mut curr = self.state.load(SeqCst);
+
+        loop {
+            if is_elapsed(curr) {
+                return;
+            }
+
+            let next = ERROR;
+
+            let actual = self.state.compare_and_swap(curr, next, SeqCst);
+
+            if curr == actual {
+                break;
+            }
+
+            curr = actual;
+        }
+
+        self.waker.wake();
+    }
+
+    pub(crate) fn cancel(entry: &Arc<Entry>) {
+        let state = entry.state.fetch_or(ELAPSED, SeqCst);
+
+        if is_elapsed(state) {
+            // Nothing more to do
+            return;
+        }
+
+        // If registered with a timer instance, try to upgrade the Arc.
+        let inner = match entry.upgrade_inner() {
+            Some(inner) => inner,
+            None => return,
+        };
+
+        let _ = inner.queue(entry);
+    }
+
+    pub(crate) fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
+        let mut curr = self.state.load(SeqCst);
+
+        if is_elapsed(curr) {
+            return Poll::Ready(if curr == ERROR {
+                Err(Error::shutdown())
+            } else {
+                Ok(())
+            });
+        }
+
+        self.waker.register_by_ref(cx.waker());
+
+        curr = self.state.load(SeqCst);
+
+        if is_elapsed(curr) {
+            return Poll::Ready(if curr == ERROR {
+                Err(Error::shutdown())
+            } else {
+                Ok(())
+            });
+        }
+
+        Poll::Pending
+    }
+
+    /// Only called by `Registration`
+    pub(crate) fn reset(entry: &mut Arc<Entry>) {
+        if !entry.is_registered() {
+            return;
+        }
+
+        let inner = match entry.upgrade_inner() {
+            Some(inner) => inner,
+            None => return,
+        };
+
+        let deadline = entry.time_ref().deadline;
+        let when = inner.normalize_deadline(deadline);
+        let elapsed = inner.elapsed();
+
+        let mut curr = entry.state.load(SeqCst);
+        let mut notify;
+
+        loop {
+            // In these two cases, there is no work to do when resetting the
+            // timer. If the `Entry` is in an error state, then it cannot be
+            // used anymore. If resetting the entry to the current value, then
+            // the reset is a noop.
+            if curr == ERROR || curr == when {
+                return;
+            }
+
+            let next;
+
+            if when <= elapsed {
+                next = ELAPSED;
+                notify = !is_elapsed(curr);
+            } else {
+                next = when;
+                notify = true;
+            }
+
+            let actual = entry.state.compare_and_swap(curr, next, SeqCst);
+
+            if curr == actual {
+                break;
+            }
+
+            curr = actual;
+        }
+
+        if notify {
+            let _ = inner.queue(entry);
+        }
+    }
+
+    fn upgrade_inner(&self) -> Option<Arc<Inner>> {
+        self.inner.as_ref().and_then(|inner| inner.upgrade())
+    }
+}
+
+fn is_elapsed(state: u64) -> bool {
+    state & ELAPSED == ELAPSED
+}
+
+impl Drop for Entry {
+    fn drop(&mut self) {
+        let inner = match self.upgrade_inner() {
+            Some(inner) => inner,
+            None => return,
+        };
+
+        inner.decrement();
+    }
+}
+
+unsafe impl Send for Entry {}
+unsafe impl Sync for Entry {}
+
+#[cfg_attr(target_arch = "x86_64", repr(align(128)))]
+#[cfg_attr(not(target_arch = "x86_64"), repr(align(64)))]
+#[derive(Debug)]
+struct CachePadded<T>(T);
diff --git a/tokio/src/time/driver/handle.rs b/tokio/src/time/driver/handle.rs
new file mode 100644
index 00000000..5d2e6b9b
--- /dev/null
+++ b/tokio/src/time/driver/handle.rs
@@ -0,0 +1,108 @@
+use crate::time::driver::Inner;
+use crate::time::Error;
+
+use std::cell::RefCell;
+use std::fmt;
+use std::marker::PhantomData;
+use std::sync::{Arc, Weak};
+
+/// Handle to time driver instance.
+#[derive(Debug, Clone)]
+pub(crate) struct Handle {
+    inner: Option<HandlePriv>,
+}
+
+/// Like `Handle` but never `None`.
+#[derive(Clone)]
+pub(crate) struct HandlePriv {
+    inner: Weak<Inner>,
+}
+
+thread_local! {
+    /// Tracks the timer for the current execution context.
+    static CURRENT_TIMER: RefCell<Option<HandlePriv>> = RefCell::new(None)
+}
+
+#[derive(Debug)]
+///Unsets default timer handler on drop.
+pub(crate) struct DefaultGuard<'a> {
+    prev: Option<HandlePriv>,
+    _lifetime: PhantomData<&'a u8>,
+}
+
+impl Drop for DefaultGuard<'_> {
+    fn drop(&mut self) {
+        CURRENT_TIMER.with(|current| {
+            let mut current = current.borrow_mut();
+            *current = self.prev.take();
+        })
+    }
+}
+
+///Sets handle to default timer, returning guard that unsets it on drop.
+///
+/// # Panics
+///
+/// This function panics if there already is a default timer set.
+pub(crate) fn set_default(handle: &Handle) -> DefaultGuard<'_> {
+    CURRENT_TIMER.with(|current| {
+        let mut current = current.borrow_mut();
+        let prev = current.take();
+
+        let handle = handle
+            .as_priv()
+            .unwrap_or_else(|| panic!("`handle` does not reference a timer"));
+
+        *current = Some(handle.clone());
+
+        DefaultGuard {
+            prev,
+            _lifetime: PhantomData,
+        }
+    })
+}
+
+impl Handle {
+    pub(crate) fn new(inner: Weak<Inner>) -> Handle {
+        let inner = HandlePriv { inner };
+        Handle { inner: Some(inner) }
+    }
+
+    fn as_priv(&self) -> Option<&HandlePriv> {
+        self.inner.as_ref()
+    }
+}
+
+impl Default for Handle {
+    fn default() -> Handle {
+        Handle { inner: None }
+    }
+}
+
+impl HandlePriv {
+    /// Try to get a handle to the current timer.
+    ///
+    /// Returns `Err` if no handle is found.
+    pub(crate) fn try_current() -> Result<HandlePriv, Error> {
+        CURRENT_TIMER.with(|current| match *current.borrow() {
+            Some(ref handle) => Ok(handle.clone()),
+            None => Err(Error::shutdown()),
+        })
+    }
+
+    /// Try to return a strong ref to the inner
+    pub(crate) fn inner(&self) -> Option<Arc<Inner>> {
+        self.inner.upgrade()
+    }
+
+    /// Consume the handle, returning the weak Inner ref.
+    pub(crate) fn into_inner(self) -> Weak<Inner> {
+        self.inner
+    }
+}
+
+impl fmt::Debug for HandlePriv {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "HandlePriv")
+    }
+}
diff --git a/tokio/src/time/driver/mod.rs b/tokio/src/time/driver/mod.rs
new file mode 100644
index 00000000..74309524
--- /dev/null
+++ b/tokio/src/time/driver/mod.rs
@@ -0,0 +1,377 @@
+//! Time driver
+
+mod atomic_stack;
+use self::atomic_stack::AtomicStack;
+
+mod entry;
+use self::entry::Entry;
+
+mod handle;
+pub(crate) use self::handle::{set_default, Handle, HandlePriv};
+
+mod registration;
+pub(crate) use self::registration::Registration;
+
+mod stack;
+use self::stack::Stack;
+
+use crate::loom::sync::atomic::{AtomicU64, AtomicUsize};
+use crate::runtime::{Park, Unpark};
+use crate::time::{wheel, Error};
+use crate::time::{Clock, Duration, Instant};
+
+use std::sync::atomic::Ordering::SeqCst;
+use std::sync::Arc;
+use std::usize;
+use std::{cmp, fmt};
+
+/// Time implementation that drives [`Delay`], [`Interval`], and [`Timeout`].
+///
+/// A `Driver` instance tracks the state necessary for managing time and
+/// notifying the [`Delay`] instances once their deadlines are reached.
+///
+/// It is expected that a single instance manages many individual [`Delay`]
+/// instances. The `Driver` implementation is thread-safe and, as such, is able
+/// to handle callers from across threads.
+///
+/// After creating the `Driver` instance, the caller must repeatedly call
+/// [`turn`]. The time driver will perform no work unless [`turn`] is called
+/// repeatedly.
+///
+/// The driver has a resolution of one millisecond. Any unit of time that falls
+/// between milliseconds are rounded up to the next millisecond.
+///
+/// When an instance is dropped, any outstanding [`Delay`] instance that has not
+/// elapsed will be notified with an error. At this point, calling `poll` on the
+/// [`Delay`] instance will result in `Err` being returned.
+///
+/// # Implementation
+///
+/// THe time driver is based on the [paper by Varghese and Lauck][paper].
+///
+/// A hashed timing wheel is a vector of slots, where each slot handles a time
+/// slice. As time progresses, the timer walks over the slot for the current
+/// instant, and processes each entry for that slot. When the timer reaches the
+/// end of the wheel, it starts again at the beginning.
+///
+/// The implementation maintains six wheels arranged in a set of levels.  As the
+/// levels go up, the slots of the associated wheel represent larger intervals
+/// of time. At each level, the wheel has 64 slots. Each slot covers a range of
+/// time equal to the wheel at the lower level. At level zero, each slot
+/// represents one millisecond of time.
+///
+/// The wheels are:
+///
+/// * Level 0: 64 x 1 millisecond slots.
+/// * Level 1: 64 x 64 millisecond slots.
+/// * Level 2: 64 x ~4 second slots.
+/// * Level 3: 64 x ~4 minute slots.
+/// * Level 4: 64 x ~4 hour slots.
+/// * Level 5: 64 x ~12 day slots.
+///
+/// When the timer processes entries at level zero, it will notify all the
+/// [`Delay`] instances as their deadlines have been reached. For all higher
+/// levels, all entries will be redistributed across the wheel at the next level
+/// down. Eventually, as time progresses, entries will [`Delay`] instances will
+/// either be canceled (dropped) or their associated entries will reach level
+/// zero and be notified.
+///
+/// [`Delay`]: struct.Delay.html
+/// [`Interval`]: struct.Interval.html
+/// [`Timeout`]: struct.Timeout.html
+/// [paper]: http://www.cs.columbia.edu/~nahum/w6998/papers/ton97-timing-wheels.pdf
+/// [`handle`]: #method.handle
+/// [`turn`]: #method.turn
+/// [Handle.struct]: struct.Handle.html
+#[derive(Debug)]
+pub(crate) struct Driver<T> {
+    /// Shared state
+    inner: Arc<Inner>,
+
+    /// Timer wheel
+    wheel: wheel::Wheel<Stack>,
+
+    /// Thread parker. The `Driver` park implementation delegates to this.
+    park: T,
+
+    /// Source of "now" instances
+    clock: Clock,
+}
+
+/// Timer state shared between `Driver`, `Handle`, and `Registration`.
+pub(crate) struct Inner {
+    /// The instant at which the timer started running.
+    start: Instant,
+
+    /// The last published timer `elapsed` value.
+    elapsed: AtomicU64,
+
+    /// Number of active timeouts
+    num: AtomicUsize,
+
+    /// Head of the "process" linked list.
+    process: AtomicStack,
+
+    /// Unparks the timer thread.
+    unpark: Box<dyn Unpark>,
+}
+
+/// Maximum number of timeouts the system can handle concurrently.
+const MAX_TIMEOUTS: usize = usize::MAX >> 1;
+
+// ===== impl Driver =====
+
+impl<T> Driver<T>
+where
+    T: Park,
+{
+    /// Create a new `Driver` instance that uses `park` to block the current
+    /// thread and `now` to get the current `Instant`.
+    ///
+    /// Specifying the source of time is useful when testing.
+    pub(crate) fn new(park: T, clock: Clock) -> Driver<T> {
+        let unpark = Box::new(park.unpark());
+
+        Driver {
+            inner: Arc::new(Inner::new(clock.now(), unpark)),
+            wheel: wheel::Wheel::new(),
+            park,
+            clock,
+        }
+    }
+
+    /// Returns a handle to the timer.
+    ///
+    /// The `Handle` is how `Delay` instances are created. The `Delay` instances
+    /// can either be created directly or the `Handle` instance can be passed to
+    /// `with_default`, setting the timer as the default timer for the execution
+    /// context.
+    pub(crate) fn handle(&self) -> Handle {
+        Handle::new(Arc::downgrade(&self.inner))
+    }
+
+    /// Converts an `Expiration` to an `Instant`.
+    fn expiration_instant(&self, when: u64) -> Instant {
+        self.inner.start + Duration::from_millis(when)
+    }
+
+    /// Run timer related logic
+    fn process(&mut self) {
+        let now = crate::time::ms(
+            self.clock.now() - self.inner.start,
+            crate::time::Round::Down,
+        );
+        let mut poll = wheel::Poll::new(now);
+
+        while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
+            let when = entry.when_internal().expect("invalid internal entry state");
+
+            // Fire the entry
+            entry.fire(when);
+
+            // Track that the entry has been fired
+            entry.set_when_internal(None);
+        }
+
+        // Update the elapsed cache
+        self.inner.elapsed.store(self.wheel.elapsed(), SeqCst);
+    }
+
+    /// Process the entry queue
+    ///
+    /// This handles adding and canceling timeouts.
+    fn process_queue(&mut self) {
+        for entry in self.inner.process.take() {
+            match (entry.when_internal(), entry.load_state()) {
+                (None, None) => {
+                    // Nothing to do
+                }
+                (Some(_), None) => {
+                    // Remove the entry
+                    self.clear_entry(&entry);
+                }
+                (None, Some(when)) => {
+                    // Queue the entry
+                    self.add_entry(entry, when);
+                }
+                (Some(_), Some(next)) => {
+                    self.clear_entry(&entry);
+                    self.add_entry(entry, next);
+                }
+            }
+        }
+    }
+
+    fn clear_entry(&mut self, entry: &Arc<Entry>) {
+        self.wheel.remove(entry, &mut ());
+        entry.set_when_internal(None);
+    }
+
+    /// Fire the entry if it needs to, otherwise queue it to be processed later.
+    ///
+    /// Returns `None` if the entry was fired.
+    fn add_entry(&mut self, entry: Arc<Entry>, when: u64) {
+        use crate::time::wheel::InsertError;
+
+        entry.set_when_internal(Some(when));
+
+        match self.wheel.insert(when, entry, &mut ()) {
+            Ok(_) => {}
+            Err((entry, InsertError::Elapsed)) => {
+                // The entry's deadline has elapsed, so fire it and update the
+                // internal state accordingly.
+                entry.set_when_internal(None);
+                entry.fire(when);
+            }
+            Err((entry, InsertError::Invalid)) => {
+                // The entry's deadline is invalid, so error it and update the
+                // internal state accordingly.
+                entry.set_when_internal(None);
+                entry.error();
+            }
+        }
+    }
+}
+
+impl<T> Park for Driver<T>
+where
+    T: Park,
+{
+    type Unpark = T::Unpark;
+    type Error = T::Error;
+
+    fn unpark(&self) -> Self::Unpark {
+        self.park.unpark()
+    }
+
+    fn park(&mut self) -> Result<(), Self::Error> {
+        self.process_queue();
+
+        match self.wheel.poll_at() {
+            Some(when) => {
+                let now = self.clock.now();
+                let deadline = self.expiration_instant(when);
+
+                if deadline > now {
+                    self.park.park_timeout(deadline - now)?;
+                } else {
+                    self.park.park_timeout(Duration::from_secs(0))?;
+                }
+            }
+            None => {
+                self.park.park()?;
+            }
+        }
+
+        self.process();
+
+        Ok(())
+    }
+
+    fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
+        self.process_queue();
+
+        match self.wheel.poll_at() {
+            Some(when) => {
+                let now = self.clock.now();
+                let deadline = self.expiration_instant(when);
+
+                if deadline > now {
+                    self.park.park_timeout(cmp::min(deadline - now, duration))?;
+                } else {
+                    self.park.park_timeout(Duration::from_secs(0))?;
+                }
+            }
+            None => {
+                self.park.park_timeout(duration)?;
+            }
+        }
+
+        self.process();
+
+        Ok(())
+    }
+}
+
+impl<T> Drop for Driver<T> {
+    fn drop(&mut self) {
+        use std::u64;
+
+        // Shutdown the stack of entries to process, preventing any new entries
+        // from being pushed.
+        self.inner.process.shutdown();
+
+        // Clear the wheel, using u64::MAX allows us to drain everything
+        let mut poll = wheel::Poll::new(u64::MAX);
+
+        while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
+            entry.error();
+        }
+    }
+}
+
+// ===== impl Inner =====
+
+impl Inner {
+    fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
+        Inner {
+            num: AtomicUsize::new(0),
+            elapsed: AtomicU64::new(0),
+            process: AtomicStack::new(),
+            start,
+            unpark,
+        }
+    }
+
+    fn elapsed(&self) -> u64 {
+        self.elapsed.load(SeqCst)
+    }
+
+    /// Increment the number of active timeouts
+    fn increment(&self) -> Result<(), Error> {
+        let mut curr = self.num.load(SeqCst);
+
+        loop {
+            if curr == MAX_TIMEOUTS {
+                return Err(Error::at_capacity());
+            }
+
+            let actual = self.num.compare_and_swap(curr, curr + 1, SeqCst);
+
+            if curr == actual {
+                return Ok(());
+            }
+
+            curr = actual;
+        }
+    }
+
+    /// Decrement the number of active timeouts
+    fn decrement(&self) {
+        let prev = self.num.fetch_sub(1, SeqCst);
+        debug_assert!(prev <= MAX_TIMEOUTS);
+    }
+
+    fn queue(&self, entry: &Arc<Entry>) -> Result<(), Error> {
+        if self.process.push(entry)? {
+            // The timer is notified so that it can process the timeout
+            self.unpark.unpark();
+        }
+
+        Ok(())
+    }
+
+    fn normalize_deadline(&self, deadline: Instant) -> u64 {
+        if deadline < self.start {
+            return 0;
+        }
+
+        crate::time::ms(deadline - self.start, crate::time::Round::Up)
+    }
+}
+
+impl fmt::Debug for Inner {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("Inner").finish()
+    }
+}
diff --git a/tokio/src/time/driver/registration.rs b/tokio/src/time/driver/registration.rs
new file mode 100644
index 00000000..3641e549
--- /dev/null
+++ b/tokio/src/time/driver/registration.rs
@@ -0,0 +1,65 @@
+use crate::time::driver::Entry;
+use crate::time::{Duration, Error, Instant};
+
+use std::sync::Arc;
+use std::task::{self, Poll};
+
+/// Registration with a timer.
+///
+/// The association between a `Delay` instance and a timer is done lazily in
+/// `poll`
+#[derive(Debug)]
+pub(crate) struct Registration {
+    entry: Arc<Entry>,
+}
+
+impl Registration {
+    pub(crate) fn new(deadline: Instant, duration: Duration) -> Registration {
+        fn is_send<T: Send + Sync>() {}
+        is_send::<Registration>();
+