diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 9bf58b4a9..dd4972ed0 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -260,6 +260,7 @@ jobs:
           - loom_pool::group_b
           - loom_pool::group_c
           - loom_pool::group_d
+          - time::driver
     steps:
       - uses: actions/checkout@v2
       - name: Install Rust
diff --git a/tokio-util/src/time/delay_queue.rs b/tokio-util/src/time/delay_queue.rs
index 000b44237..4edd5cd64 100644
--- a/tokio-util/src/time/delay_queue.rs
+++ b/tokio-util/src/time/delay_queue.rs
@@ -14,7 +14,7 @@ use std::cmp;
 use std::future::Future;
 use std::marker::PhantomData;
 use std::pin::Pin;
-use std::task::{self, Poll};
+use std::task::{self, Poll, Waker};
 
 /// A queue of delayed elements.
 ///
@@ -145,6 +145,11 @@ pub struct DelayQueue<T> {
 
     /// Instant at which the timer starts
     start: Instant,
+
+    /// Waker that is invoked when we potentially need to reset the timer.
+    /// Because we lazily create the timer when the first entry is created, we
+    /// need to awaken any poller that polled us before that point.
+    waker: Option<Waker>,
 }
 
 /// An entry in `DelayQueue` that has expired and removed.
@@ -253,6 +258,7 @@ impl<T> DelayQueue<T> {
             delay: None,
             wheel_now: 0,
             start: Instant::now(),
+            waker: None,
         }
     }
 
@@ -330,6 +336,10 @@ impl<T> DelayQueue<T> {
         };
 
         if should_set_delay {
+            if let Some(waker) = self.waker.take() {
+                waker.wake();
+            }
+
             let delay_time = self.start + Duration::from_millis(when);
             if let Some(ref mut delay) = &mut self.delay {
                 delay.reset(delay_time);
@@ -348,6 +358,15 @@ impl<T> DelayQueue<T> {
         &mut self,
         cx: &mut task::Context<'_>,
     ) -> Poll<Option<Result<Expired<T>, Error>>> {
+        if !self
+            .waker
+            .as_ref()
+            .map(|w| w.will_wake(cx.waker()))
+            .unwrap_or(false)
+        {
+            self.waker = Some(cx.waker().clone());
+        }
+
         let item = ready!(self.poll_idx(cx));
         Poll::Ready(item.map(|result| {
             result.map(|idx| {
@@ -533,6 +552,7 @@ impl<T> DelayQueue<T> {
 
         let next_deadline = self.next_deadline();
         if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) {
+            // This should awaken us if necessary (ie, if already expired)
             delay.reset(deadline);
         }
     }
diff --git a/tokio-util/tests/time_delay_queue.rs b/tokio-util/tests/time_delay_queue.rs
index 42a56b8b3..d42dca87d 100644
--- a/tokio-util/tests/time_delay_queue.rs
+++ b/tokio-util/tests/time_delay_queue.rs
@@ -2,7 +2,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
-use tokio::time::{self, sleep, Duration, Instant};
+use tokio::time::{self, sleep, sleep_until, Duration, Instant};
 use tokio_test::{assert_ok, assert_pending, assert_ready, task};
 use tokio_util::time::DelayQueue;
 
@@ -107,9 +107,10 @@ async fn multi_delay_at_start() {
     assert_pending!(poll!(queue));
     assert!(!queue.is_woken());
 
+    let start = Instant::now();
     for elapsed in 0..1200 {
-        sleep(ms(1)).await;
         let elapsed = elapsed + 1;
+        tokio::time::sleep_until(start + ms(elapsed)).await;
 
         if delays.contains(&elapsed) {
             assert!(queue.is_woken());
@@ -117,7 +118,12 @@ async fn multi_delay_at_start() {
             assert_pending!(poll!(queue));
         } else if queue.is_woken() {
             let cascade = &[192, 960];
-            assert!(cascade.contains(&elapsed), "elapsed={}", elapsed);
+            assert!(
+                cascade.contains(&elapsed),
+                "elapsed={} dt={:?}",
+                elapsed,
+                Instant::now() - start
+            );
 
             assert_pending!(poll!(queue));
         }
@@ -205,7 +211,7 @@ async fn reset_much_later() {
 
     sleep(ms(3)).await;
 
-    queue.reset_at(&key, now + ms(5));
+    queue.reset_at(&key, now + ms(10));
 
     sleep(ms(20)).await;
 
@@ -402,7 +408,7 @@ async fn insert_before_first_after_poll() {
 
     sleep(ms(99)).await;
 
-    assert!(!queue.is_woken());
+    assert_pending!(poll!(queue));
 
     sleep(ms(1)).await;
 
@@ -457,7 +463,7 @@ async fn reset_later_after_slot_starts() {
 
     assert_pending!(poll!(queue));
 
-    sleep(ms(80)).await;
+    sleep_until(now + Duration::from_millis(80)).await;
 
     assert!(!queue.is_woken());
 
@@ -472,7 +478,7 @@ async fn reset_later_after_slot_starts() {
 
     assert_pending!(poll!(queue));
 
-    sleep(ms(39)).await;
+    sleep_until(now + Duration::from_millis(119)).await;
     assert!(!queue.is_woken());
 
     sleep(ms(1)).await;
@@ -515,7 +521,7 @@ async fn reset_earlier_after_slot_starts() {
 
     assert_pending!(poll!(queue));
 
-    sleep(ms(80)).await;
+    sleep_until(now + Duration::from_millis(80)).await;
 
     assert!(!queue.is_woken());
 
@@ -530,7 +536,7 @@ async fn reset_earlier_after_slot_starts() {
 
     assert_pending!(poll!(queue));
 
-    sleep(ms(39)).await;
+    sleep_until(now + Duration::from_millis(119)).await;
     assert!(!queue.is_woken());
 
     sleep(ms(1)).await;
diff --git a/tokio/src/loom/std/mod.rs b/tokio/src/loom/std/mod.rs
index 414ef9062..c3f74efb7 100644
--- a/tokio/src/loom/std/mod.rs
+++ b/tokio/src/loom/std/mod.rs
@@ -47,7 +47,7 @@ pub(crate) mod rand {
 }
 
 pub(crate) mod sync {
-    pub(crate) use std::sync::Arc;
+    pub(crate) use std::sync::{Arc, Weak};
 
     // Below, make sure all the feature-influenced types are exported for
     // internal use. Note however that some are not _currently_ named by
diff --git a/tokio/src/stream/throttle.rs b/tokio/src/stream/throttle.rs
index 8f4a256d1..ff1fbf01f 100644
--- a/tokio/src/stream/throttle.rs
+++ b/tokio/src/stream/throttle.rs
@@ -17,7 +17,7 @@ where
     let delay = if duration == Duration::from_millis(0) {
         None
     } else {
-        Some(Sleep::new_timeout(Instant::now() + duration, duration))
+        Some(Sleep::new_timeout(Instant::now() + duration))
     };
 
     Throttle {
diff --git a/tokio/src/stream/timeout.rs b/tokio/src/stream/timeout.rs
index 669973ffa..61154da05 100644
--- a/tokio/src/stream/timeout.rs
+++ b/tokio/src/stream/timeout.rs
@@ -23,7 +23,7 @@ pin_project! {
 impl<S: Stream> Timeout<S> {
     pub(super) fn new(stream: S, duration: Duration) -> Self {
         let next = Instant::now() + duration;
-        let deadline = Sleep::new_timeout(next, duration);
+        let deadline = Sleep::new_timeout(next);
 
         Timeout {
             stream: Fuse::new(stream),
diff --git a/tokio/src/time/driver/atomic_stack.rs b/tokio/src/time/driver/atomic_stack.rs
deleted file mode 100644
index 5dcc4726e..000000000
--- a/tokio/src/time/driver/atomic_stack.rs
+++ /dev/null
@@ -1,124 +0,0 @@
-use crate::time::driver::Entry;
-use crate::time::error::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()),
-        }
-    }
-
-    /// Pushes 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)
-    }
-
-    /// Takes all entries from the stack
-    pub(crate) fn take(&self) -> AtomicStackEntries {
-        let ptr = self.head.swap(ptr::null_mut(), SeqCst);
-        AtomicStackEntries { ptr }
-    }
-
-    /// Drains 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() || self.ptr == SHUTDOWN {
-            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(Error::shutdown());
-        }
-    }
-}
diff --git a/tokio/src/time/driver/entry.rs b/tokio/src/time/driver/entry.rs
index b40cae739..e0926797f 100644
--- a/tokio/src/time/driver/entry.rs
+++ b/tokio/src/time/driver/entry.rs
@@ -1,362 +1,684 @@
-use crate::loom::sync::atomic::AtomicU64;
+//! Timer state structures.
+//!
+//! This module contains the heart of the intrusive timer implementation, and as
+//! such the structures inside are full of tricky concurrency and unsafe code.
+//!
+//! # Ground rules
+//!
+//! The heart of the timer implementation here is the `TimerShared` structure,
+//! shared between the `TimerEntry` and the driver. Generally, we permit access
+//! to `TimerShared` ONLY via either 1) a mutable reference to `TimerEntry` or
+//! 2) a held driver lock.
+//!
+//! It follows from this that any changes made while holding BOTH 1 and 2 will
+//! be reliably visible, regardless of ordering. This is because of the acq/rel
+//! fences on the driver lock ensuring ordering with 2, and rust mutable
+//! reference rules for 1 (a mutable reference to an object can't be passed
+//! between threads without an acq/rel barrier, and same-thread we have local
+//! happens-before ordering).
+//!
+//! # State field
+//!
+//! Each timer has a state field associated with it. This field contains either
+//! the current scheduled time, or a special flag value indicating its state.
+//! This state can either indicate that the timer is on the 'pending' queue (and
+//! thus will be fired with an `Ok(())` result soon) or that it has already been
+//! fired/deregistered.
+//!
+//! This single state field allows for code that is firing the timer to
+//! synchronize with any racing `reset` calls reliably.
+//!
+//! # Cached vs true timeouts
+//!
+//! To allow for the use case of a timeout that is periodically reset before
+//! expiration to be as lightweight as possible, we support optimistically
+//! lock-free timer resets, in the case where a timer is rescheduled to a later
+//! point than it was originally scheduled for.
+//!
+//! This is accomplished by lazily rescheduling timers. That is, we update the
+//! state field field with the true expiration of the timer from the holder of
+//! the [`TimerEntry`]. When the driver services timers (ie, whenever it's
+//! walking lists of timers), it checks this "true when" value, and reschedules
+//! based on it.
+//!
+//! We do, however, also need to track what the expiration time was when we
+//! originally registered the timer; this is used to locate the right linked
+//! list when the timer is being cancelled. This is referred to as the "cached
+//! when" internally.
+//!
+//! There is of course a race condition between timer reset and timer
+//! expiration. If the driver fails to observe the updated expiration time, it
+//! could trigger expiration of the timer too early. However, because
+//! `mark_pending` performs a compare-and-swap, it will identify this race and
+//! refuse to mark the timer as pending.
+
+use crate::loom::cell::UnsafeCell;
+use crate::loom::sync::atomic::Ordering;
+
 use crate::sync::AtomicWaker;
-use crate::time::driver::{Handle, Inner};
-use crate::time::{error::Error, Duration, Instant};
+use crate::time::Instant;
+use crate::util::linked_list;
 
-use std::cell::UnsafeCell;
-use std::ptr;
-use std::sync::atomic::Ordering::SeqCst;
-use std::sync::atomic::{AtomicBool, AtomicU8};
-use std::sync::{Arc, Weak};
-use std::task::{self, Poll};
-use std::u64;
+use super::Handle;
 
-/// Internal state shared between a `Sleep` instance and the timer.
+use std::cell::UnsafeCell as StdUnsafeCell;
+use std::task::{Context, Poll, Waker};
+use std::{marker::PhantomPinned, pin::Pin, ptr::NonNull};
+
+type TimerResult = Result<(), crate::time::error::Error>;
+
+const STATE_DEREGISTERED: u64 = u64::max_value();
+const STATE_PENDING_FIRE: u64 = STATE_DEREGISTERED - 1;
+const STATE_MIN_VALUE: u64 = STATE_PENDING_FIRE;
+
+/// Not all platforms support 64-bit compare-and-swap. This hack replaces the
+/// AtomicU64 with a mutex around a u64 on platforms that don't. This is slow,
+/// unfortunately, but 32-bit platforms are a bit niche so it'll do for now.
 ///
-/// 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.
+/// Note: We use "x86 or 64-bit pointers" as the condition here because
+/// target_has_atomic is not stable.
+#[cfg(all(
+    not(tokio_force_time_entry_locked),
+    any(target_arch = "x86", target_pointer_width = "64")
+))]
+type AtomicU64 = crate::loom::sync::atomic::AtomicU64;
+
+#[cfg(not(all(
+    not(tokio_force_time_entry_locked),
+    any(target_arch = "x86", target_pointer_width = "64")
+)))]
 #[derive(Debug)]
-pub(crate) struct Entry {
-    /// Only accessed from `Registration`.
-    time: CachePadded<UnsafeCell<Time>>,
+struct AtomicU64 {
+    inner: crate::loom::sync::Mutex<u64>,
+}
 
-    /// Timer internals. Using a weak pointer allows the timer to shutdown
-    /// without all `Sleep` instances having completed.
-    ///
-    /// When empty, it means that the entry has not yet been linked with a
-    /// timer instance.
-    inner: Weak<Inner>,
+#[cfg(not(all(
+    not(tokio_force_time_entry_locked),
+    any(target_arch = "x86", target_pointer_width = "64")
+)))]
+impl AtomicU64 {
+    fn new(v: u64) -> Self {
+        Self {
+            inner: crate::loom::sync::Mutex::new(v),
+        }
+    }
 
-    /// 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.
+    fn load(&self, _order: Ordering) -> u64 {
+        debug_assert_ne!(_order, Ordering::SeqCst); // we only provide AcqRel with the lock
+        *self.inner.lock()
+    }
+
+    fn store(&self, v: u64, _order: Ordering) {
+        debug_assert_ne!(_order, Ordering::SeqCst); // we only provide AcqRel with the lock
+        *self.inner.lock() = v;
+    }
+
+    fn compare_exchange(
+        &self,
+        current: u64,
+        new: u64,
+        _success: Ordering,
+        _failure: Ordering,
+    ) -> Result<u64, u64> {
+        debug_assert_ne!(_success, Ordering::SeqCst); // we only provide AcqRel with the lock
+        debug_assert_ne!(_failure, Ordering::SeqCst);
+
+        let mut lock = self.inner.lock();
+
+        if *lock == current {
+            *lock = new;
+            Ok(current)
+        } else {
+            Err(*lock)
+        }
+    }
+
+    fn compare_exchange_weak(
+        &self,
+        current: u64,
+        new: u64,
+        success: Ordering,
+        failure: Ordering,
+    ) -> Result<u64, u64> {
+        self.compare_exchange(current, new, success, failure)
+    }
+}
+
+/// This structure holds the current shared state of the timer - its scheduled
+/// time (if registered), or otherwise the result of the timer completing, as
+/// well as the registered waker.
+///
+/// Generally, the StateCell is only permitted to be accessed from two contexts:
+/// Either a thread holding the corresponding &mut TimerEntry, or a thread
+/// holding the timer driver lock. The write actions on the StateCell amount to
+/// passing "ownership" of the StateCell between these contexts; moving a timer
+/// from the TimerEntry to the driver requires _both_ holding the &mut
+/// TimerEntry and the driver lock, while moving it back (firing the timer)
+/// requires only the driver lock.
+pub(super) struct StateCell {
+    /// Holds either the scheduled expiration time for this timer, or (if the
+    /// timer has been fired and is unregistered), [`u64::max_value()`].
     state: AtomicU64,
-
-    /// Stores the actual error. If `state` indicates that an error occurred,
-    /// this is guaranteed to be a non-zero value representing the first error
-    /// that occurred. Otherwise its value is undefined.
-    error: AtomicU8,
-
-    /// 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 `Sleep` instance can be reset to a different deadline by the thread
-    /// that owns the `Sleep` 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(crate) 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(crate) prev_stack: UnsafeCell<*const Entry>,
+    /// If the timer is fired (an Acquire order read on state shows
+    /// `u64::max_value()`), holds the result that should be returned from
+    /// polling the timer. Otherwise, the contents are unspecified and reading
+    /// without holding the driver lock is undefined behavior.
+    result: UnsafeCell<TimerResult>,
+    /// The currently-registered waker
+    waker: CachePadded<AtomicWaker>,
 }
 
-/// Stores the info for `Sleep`.
+impl Default for StateCell {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl std::fmt::Debug for StateCell {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "StateCell({:?})", self.read_state())
+    }
+}
+
+impl StateCell {
+    fn new() -> Self {
+        Self {
+            state: AtomicU64::new(STATE_DEREGISTERED),
+            result: UnsafeCell::new(Ok(())),
+            waker: CachePadded(AtomicWaker::new()),
+        }
+    }
+
+    fn is_pending(&self) -> bool {
+        self.state.load(Ordering::Relaxed) == STATE_PENDING_FIRE
+    }
+
+    /// Returns the current expiration time, or None if not currently scheduled.
+    fn when(&self) -> Option<u64> {
+        let cur_state = self.state.load(Ordering::Relaxed);
+
+        if cur_state == u64::max_value() {
+            None
+        } else {
+            Some(cur_state)
+        }
+    }
+
+    /// If the timer is completed, returns the result of the timer. Otherwise,
+    /// returns None and registers the waker.
+    fn poll(&self, waker: &Waker) -> Poll<TimerResult> {
+        // We must register first. This ensures that either `fire` will
+        // observe the new waker, or we will observe a racing fire to have set
+        // the state, or both.
+        self.waker.0.register_by_ref(waker);
+
+        self.read_state()
+    }
+
+    fn read_state(&self) -> Poll<TimerResult> {
+        let cur_state = self.state.load(Ordering::Acquire);
+
+        if cur_state == STATE_DEREGISTERED {
+            // SAFETY: The driver has fired this timer; this involves writing
+            // the result, and then writing (with release ordering) the state
+            // field.
+            Poll::Ready(unsafe { self.result.with(|p| *p) })
+        } else {
+            Poll::Pending
+        }
+    }
+
+    /// Marks this timer as being moved to the pending list, if its scheduled
+    /// time is not after `not_after`.
+    ///
+    /// If the timer is scheduled for a time after not_after, returns an Err
+    /// containing the current scheduled time.
+    ///
+    /// SAFETY: Must hold the driver lock.
+    unsafe fn mark_pending(&self, not_after: u64) -> Result<(), u64> {
+        // Quick initial debug check to see if the timer is already fired. Since
+        // firing the timer can only happen with the driver lock held, we know
+        // we shouldn't be able to "miss" a transition to a fired state, even
+        // with relaxed ordering.
+        let mut cur_state = self.state.load(Ordering::Relaxed);
+
+        loop {
+            debug_assert!(cur_state < STATE_MIN_VALUE);
+
+            if cur_state > not_after {
+                break Err(cur_state);
+            }
+
+            match self.state.compare_exchange(
+                cur_state,
+                STATE_PENDING_FIRE,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => {
+                    break Ok(());
+                }
+                Err(actual_state) => {
+                    cur_state = actual_state;
+                }
+            }
+        }
+    }
+
+    /// Fires the timer, setting the result to the provided result.
+    ///
+    /// Returns:
+    /// * `Some(waker) - if fired and a waker needs to be invoked once the
+    ///   driver lock is released
+    /// * `None` - if fired and a waker does not need to be invoked, or if
+    ///   already fired
+    ///
+    /// SAFETY: The driver lock must be held.
+    unsafe fn fire(&self, result: TimerResult) -> Option<Waker> {
+        // Quick initial check to see if the timer is already fired. Since
+        // firing the timer can only happen with the driver lock held, we know
+        // we shouldn't be able to "miss" a transition to a fired state, even
+        // with relaxed ordering.
+        let cur_state = self.state.load(Ordering::Relaxed);
+        if cur_state == STATE_DEREGISTERED {
+            return None;
+        }
+
+        // SAFETY: We assume the driver lock is held and the timer is not
+        // fired, so only the driver is accessing this field.
+        //
+        // We perform a release-ordered store to state below, to ensure this
+        // write is visible before the state update is visible.
+        unsafe { self.result.with_mut(|p| *p = result) };
+
+        self.state.store(STATE_DEREGISTERED, Ordering::Release);
+
+        self.waker.0.take_waker()
+    }
+
+    /// Marks the timer as registered (poll will return None) and sets the
+    /// expiration time.
+    ///
+    /// While this function is memory-safe, it should only be called from a
+    /// context holding both `&mut TimerEntry` and the driver lock.    
+    fn set_expiration(&self, timestamp: u64) {
+        debug_assert!(timestamp < STATE_MIN_VALUE);
+
+        // We can use relaxed ordering because we hold the driver lock and will
+        // fence when we release the lock.
+        self.state.store(timestamp, Ordering::Relaxed);
+    }
+
+    /// Attempts to adjust the timer to a new timestamp.
+    ///
+    /// If the timer has already been fired, is pending firing, or the new
+    /// timestamp is earlier than the old timestamp, (or occasionally
+    /// spuriously) returns Err without changing the timer's state. In this
+    /// case, the timer must be deregistered and re-registered.
+    fn extend_expiration(&self, new_timestamp: u64) -> Result<(), ()> {
+        let mut prior = self.state.load(Ordering::Relaxed);
+        loop {
+            if new_timestamp < prior || prior >= STATE_MIN_VALUE {
+                return Err(());
+            }
+
+            match self.state.compare_exchange_weak(
+                prior,
+                new_timestamp,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => {
+                    return Ok(());
+                }
+                Err(true_prior) => {
+                    prior = true_prior;
+                }
+            }
+        }
+    }
+
+    /// Returns true if the state of this timer indicates that the timer might
+    /// be registered with the driver. This check is performed with relaxed
+    /// ordering, but is conservative - if it returns false, the timer is
+    /// definitely _not_ registered.
+    pub(super) fn might_be_registered(&self) -> bool {
+        self.state.load(Ordering::Relaxed) != u64::max_value()
+    }
+}
+
+/// A timer entry.
+///
+/// This is the handle to a timer that is controlled by the requester of the
+/// timer. As this participates in intrusive data structures, it must be pinned
+/// before polling.
 #[derive(Debug)]
-pub(crate) struct Time {
-    pub(crate) deadline: Instant,
-    pub(crate) duration: Duration,
+pub(super) struct TimerEntry {
+    /// Arc reference to the driver. We can only free the driver after
+    /// deregistering everything from their respective timer wheels.
+    driver: Handle,
+    /// Shared inner structure; this is part of an intrusive linked list, and
+    /// therefore other references can exist to it while mutable references to
+    /// Entry exist.
+    ///
+    /// This is manipulated only under the inner mutex. TODO: Can we use loom
+    /// cells for this?
+    inner: StdUnsafeCell<TimerShared>,
+    /// Initial deadline for the timer. This is used to register on the first
+    /// poll, as we can't register prior to being pinned.
+    initial_deadline: Option<Instant>,
 }
 
-/// Flag indicating a timer entry has elapsed
-const ELAPSED: u64 = 1 << 63;
+unsafe impl Send for TimerEntry {}
+unsafe impl Sync for TimerEntry {}
 
-/// Flag indicating a timer entry has reached an error state
-const ERROR: u64 = u64::MAX;
+/// An TimerHandle is the (non-enforced) "unique" pointer from the driver to the
+/// timer entry. Generally, at most one TimerHandle exists for a timer at a time
+/// (enforced by the timer state machine).
+///
+/// SAFETY: An TimerHandle is essentially a raw pointer, and the usual caveats
+/// of pointer safety apply. In particular, TimerHandle does not itself enforce
+/// that the timer does still exist; however, normally an TimerHandle is created
+/// immediately before registering the timer, and is consumed when firing the
+/// timer, to help minimize mistakes. Still, because TimerHandle cannot enforce
+/// memory safety, all operations are unsafe.
+#[derive(Debug)]
+pub(crate) struct TimerHandle {
+    inner: NonNull<TimerShared>,
+}
+
+pub(super) type EntryList = crate::util::linked_list::LinkedList<TimerShared, TimerShared>;
+
+/// The shared state structure of a timer. This structure is shared between the
+/// frontend (`Entry`) and driver backend.
+///
+/// Note that this structure is located inside the `TimerEntry` structure.
+#[derive(Debug)]
+pub(crate) struct TimerShared {
+    /// Current state. This records whether the timer entry is currently under
+    /// the ownership of the driver, and if not, its current state (not
+    /// complete, fired, error, etc).
+    state: StateCell,
+
+    /// Data manipulated by the driver thread itself, only.
+    driver_state: CachePadded<TimerSharedPadded>,
+
+    _p: PhantomPinned,
+}
+
+impl TimerShared {
+    pub(super) fn new() -> Self {
+        Self {
+            state: StateCell::default(),
+            driver_state: CachePadded(TimerSharedPadded::new()),
+            _p: PhantomPinned,
+        }
+    }
+
+    /// Gets the cached time-of-expiration value
+    pub(super) fn cached_when(&self) -> u64 {
+        // Cached-when is only accessed under the driver lock, so we can use relaxed
+        self.driver_state.0.cached_when.load(Ordering::Relaxed)
+    }
+
+    /// Gets the true time-of-expiration value, and copies it into the cached
+    /// time-of-expiration value.
+    ///
+    /// SAFETY: Must be called with the driver lock held, and when this entry is
+    /// not in any timer wheel lists.
+    pub(super) unsafe fn sync_when(&self) -> u64 {
+        let true_when = self.true_when();
+
+        self.driver_state
+            .0
+            .cached_when
+            .store(true_when, Ordering::Relaxed);
+
+        true_when
+    }
+
+    /// Returns the true time-of-expiration value, with relaxed memory ordering.
+    pub(super) fn true_when(&self) -> u64 {
+        self.state.when().expect("Timer already fired")
+    }
+
+    /// Sets the true time-of-expiration value, even if it is less than the
+    /// current expiration or the timer is deregistered.
+    ///
+    /// SAFETY: Must only be called with the driver lock held and the entry not
+    /// in the timer wheel.
+    pub(super) unsafe fn set_expiration(&self, t: u64) {
+        self.state.set_expiration(t);
+        self.driver_state.0.cached_when.store(t, Ordering::Relaxed);
+    }
+
+    /// Sets the true time-of-expiration only if it is after the current.
+    pub(super) fn extend_expiration(&self, t: u64) -> Result<(), ()> {
+        self.state.extend_expiration(t)
+    }
+
+    /// Returns a TimerHandle for this timer.
+    pub(super) fn handle(&self) -> TimerHandle {
+        TimerHandle {
+            inner: NonNull::from(self),
+        }
+    }
+
+    /// Returns true if the state of this timer indicates that the timer might
+    /// be registered with the driver. This check is performed with relaxed
+    /// ordering, but is conservative - if it returns false, the timer is
+    /// definitely _not_ registered.
+    pub(super) fn might_be_registered(&self) -> bool {
+        self.state.might_be_registered()
+    }
+}
+
+/// Additional shared state between the driver and the timer which is cache
+/// padded. This contains the information that the driver thread accesses most
+/// frequently to minimize contention. In particular, we move it away from the
+/// waker, as the waker is updated on every poll.
+struct TimerSharedPadded {
+    /// The expiration time for which this entry is currently registered.
+    /// Generally owned by the driver, but is accessed by the entry when not
+    /// registered.
+    cached_when: AtomicU64,
+
+    /// The true expiration time. Set by the timer future, read by the driver.
+    true_when: AtomicU64,
+
+    /// A link within the doubly-linked list of timers on a particular level and
+    /// slot. Valid only if state is equal to Registered.
+    ///
+    /// Only accessed under the entry lock.
+    pointers: StdUnsafeCell<linked_list::Pointers<TimerShared>>,
+}
+
+impl std::fmt::Debug for TimerSharedPadded {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("TimerSharedPadded")
+            .field("when", &self.true_when.load(Ordering::Relaxed))
+            .field("cached_when", &self.cached_when.load(Ordering::Relaxed))
+            .finish()
+    }
+}
+
+impl TimerSharedPadded {
+    fn new() -> Self {
+        Self {
+            cached_when: AtomicU64::new(0),
+            true_when: AtomicU64::new(0),
+            pointers: StdUnsafeCell::new(linked_list::Pointers::new()),
+        }
+    }
+}
+
+unsafe impl Send for TimerShared {}
+unsafe impl Sync for TimerShared {}
+
+unsafe impl linked_list::Link for TimerShared {
+    type Handle = TimerHandle;
+
+    type Target = TimerShared;
+
+    fn as_raw(handle: &Self::Handle) -> NonNull<Self::Target> {
+        handle.inner
+    }
+
+    unsafe fn from_raw(ptr: NonNull<Self::Target>) -> Self::Handle {
+        TimerHandle { inner: ptr }
+    }
+
+    unsafe fn pointers(
+        target: NonNull<Self::Target>,
+    ) -> NonNull<linked_list::Pointers<Self::Target>> {
+        unsafe { NonNull::new(target.as_ref().driver_state.0.pointers.get()).unwrap() }
+    }
+}
 
 // ===== impl Entry =====
 
-impl Entry {
-    pub(crate) fn new(handle: &Handle, deadline: Instant, duration: Duration) -> Arc<Entry> {
-        let inner = handle.inner().unwrap();
+impl TimerEntry {
+    pub(crate) fn new(handle: &Handle, deadline: Instant) -> Self {
+        let driver = handle.clone();
 
-        // Attempt to increment the number of active timeouts
-        let entry = if let Err(err) = inner.increment() {
-            let entry = Entry::new2(deadline, duration, Weak::new(), ERROR);
-            entry.error(err);
-            entry
-        } else {
-            let when = inner.normalize_deadline(deadline);
-            let state = if when <= inner.elapsed() {
-                ELAPSED
-            } else {
-                when
-            };
-            Entry::new2(deadline, duration, Arc::downgrade(&inner), state)
-        };
-
-        let entry = Arc::new(entry);
-        if let Err(err) = inner.queue(&entry) {
-            entry.error(err);
-        }
-
-        entry
-    }
-
-    /// 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()
-    }
-
-    pub(crate) fn when(&self) -> u64 {
-        self.when_internal().expect("invalid internal state")
-    }
-
-    /// 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;
+        Self {
+            driver,
+            inner: StdUnsafeCell::new(TimerShared::new()),
+            initial_deadline: Some(deadline),
         }
     }
 
-    /// 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)
-        }
+    fn inner(&self) -> &TimerShared {
+        unsafe { &*self.inner.get() }
     }
 
     pub(crate) fn is_elapsed(&self) -> bool {
-        let state = self.state.load(SeqCst);
-        is_elapsed(state)
+        !self.inner().state.might_be_registered() && self.initial_deadline.is_none()
     }
 
-    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();
+    /// Cancels and deregisters the timer. This operation is irreversible.
+    pub(crate) fn cancel(self: Pin<&mut Self>) {
+        // We need to perform an acq/rel fence with the driver thread, and the
+        // simplest way to do so is to grab the driver lock.
+        //
+        // Why is this necessary? We're about to release this timer's memory for
+        // some other non-timer use. However, we've been doing a bunch of
+        // relaxed (or even non-atomic) writes from the driver thread, and we'll
+        // be doing more from _this thread_ (as this memory is interpreted as
+        // something else).
+        //
+        // It is critical to ensure that, from the point of view of the driver,
+        // those future non-timer writes happen-after the timer is fully fired,
+        // and from the purpose of this thread, the driver's writes all
+        // happen-before we drop the timer. This in turn requires us to perform
+        // an acquire-release barrier in _both_ directions between the driver
+        // and dropping thread.
+        //
+        // The lock acquisition in clear_entry serves this purpose. All of the
+        // driver manipulations happen with the lock held, so we can just take
+        // the lock and be sure that this drop happens-after everything the
+        // driver did so far and happens-before everything the driver does in
+        // the future. While we have the lock held, we also go ahead and
+        // deregister the entry if necessary.
+        unsafe { self.driver.clear_entry(NonNull::from(self.inner())) };
     }
 
-    pub(crate) fn error(&self, error: Error) {
-        // Record the precise nature of the error, if there isn't already an
-        // error present. If we don't actually transition to the error state
-        // below, that's fine, as the error details we set here will be ignored.
-        self.error.compare_and_swap(0, error.as_u8(), SeqCst);
+    pub(crate) fn reset(mut self: Pin<&mut Self>, new_time: Instant) {
+        unsafe { self.as_mut().get_unchecked_mut() }.initial_deadline = None;
 
-        // Only transition to the error state if not currently elapsed
-        let mut curr = self.state.load(SeqCst);
+        let tick = self.driver.time_source().deadline_to_tick(new_time);
 
-        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
+        if self.inner().extend_expiration(tick).is_ok() {
             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::from_u8(self.error.load(SeqCst)))
-            } 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::from_u8(self.error.load(SeqCst)))
-            } else {
-                Ok(())
-            });
-        }
-
-        Poll::Pending
-    }
-
-    /// Only called by `Registration`
-    pub(crate) fn reset(entry: &mut Arc<Entry>) {
-        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 next = if when <= elapsed { ELAPSED } else { when };
-
-        let mut curr = entry.state.load(SeqCst);
-
-        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 actual = entry.state.compare_and_swap(curr, next, SeqCst);
-
-            if curr == actual {
-                break;
-            }
-
-            curr = actual;
-        }
-
-        // If the state has transitioned to 'elapsed' then wake the task as
-        // this entry is ready to be polled.
-        if !is_elapsed(curr) && is_elapsed(next) {
-            entry.waker.wake();
-        }
-
-        // The driver tracks all non-elapsed entries; notify the driver that it
-        // should update its state for this entry unless the entry had already
-        // elapsed and remains elapsed.
-        if !is_elapsed(curr) || !is_elapsed(next) {
-            let _ = inner.queue(entry);
+        unsafe {
+            self.driver.reregister(tick, self.inner().into());
         }
     }
 
-    fn new2(deadline: Instant, duration: Duration, inner: Weak<Inner>, state: u64) -> Self {
-        Self {
-            time: CachePadded(UnsafeCell::new(Time { deadline, duration })),
-            inner,
-            waker: AtomicWaker::new(),
-            state: AtomicU64::new(state),
-            queued: AtomicBool::new(false),
-            error: AtomicU8::new(0),
-            next_atomic: UnsafeCell::new(ptr::null_mut()),
-            when: UnsafeCell::new(None),
-            next_stack: UnsafeCell::new(None),
-            prev_stack: UnsafeCell::new(ptr::null_mut()),
+    pub(crate) fn poll_elapsed(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Result<(), super::Error>> {
+        if let Some(deadline) = self.initial_deadline {
+            self.as_mut().reset(deadline);
         }
-    }
 
-    fn upgrade_inner(&self) -> Option<Arc<Inner>> {
-        self.inner.upgrade()
+        let this = unsafe { self.get_unchecked_mut() };
+
+        this.inner().state.poll(cx.waker())
     }
 }
 
-fn is_elapsed(state: u64) -> bool {
-    state & ELAPSED == ELAPSED
+impl TimerHandle {
+    pub(super) unsafe fn cached_when(&self) -> u64 {
+        unsafe { self.inner.as_ref().cached_when() }
+    }
+
+    pub(super) unsafe fn sync_when(&self) -> u64 {
+        unsafe { self.inner.as_ref().sync_when() }
+    }
+
+    pub(super) unsafe fn is_pending(&self) -> bool {
+        unsafe { self.inner.as_ref().state.is_pending() }
+    }
+
+    /// Forcibly sets the true and cached expiration times to the given tick.
+    ///
+    /// SAFETY: The caller must ensure that the handle remains valid, the driver
+    /// lock is held, and that the timer is not in any wheel linked lists.
+    pub(super) unsafe fn set_expiration(&self, tick: u64) {
+        self.inner.as_ref().set_expiration(tick);
+    }
+
+    /// Attempts to mark this entry as pending. If the expiration time is after
+    /// `not_after`, however, returns an Err with the current expiration time.
+    ///
+    /// If an `Err` is returned, the `cached_when` value will be updated to this
+    /// new expiration time.
+    ///
+    /// SAFETY: The caller must ensure that the handle remains valid, the driver
+    /// lock is held, and that the timer is not in any wheel linked lists.
+    /// After returning Ok, the entry must be added to the pending list.
+    pub(super) unsafe fn mark_pending(&self, not_after: u64) -> Result<(), u64> {
+        match self.inner.as_ref().state.mark_pending(not_after) {
+            Ok(()) => Ok(()),
+            Err(tick) => {
+                self.inner
+                    .as_ref()
+                    .driver_state
+                    .0
+                    .cached_when
+                    .store(tick, Ordering::Relaxed);
+                Err(tick)
+            }
+        }
+    }
+
+    /// Attempts to transition to a terminal state. If the state is already a
+    /// terminal state, does nothing.
+    ///
+    /// Because the entry might be dropped after the state is moved to a
+    /// terminal state, this function consumes the handle to ensure we don't
+    /// access the entry afterwards.
+    ///
+    /// Returns the last-registered waker, if any.
+    ///
+    /// SAFETY: The driver lock must be held while invoking this function, and
+    /// the entry must not be in any wheel linked lists.
+    pub(super) unsafe fn fire(self, completed_state: TimerResult) -> Option<Waker> {
+        self.inner.as_ref().state.fire(completed_state)
+    }
 }
 
-impl Drop for Entry {
+impl Drop for TimerEntry {
     fn drop(&mut self) {
-        let inner = match self.upgrade_inner() {
-            Some(inner) => inner,
-            None => return,
-        };
-
-        inner.decrement();
+        unsafe { Pin::new_unchecked(self) }.as_mut().cancel()
     }
 }
 
-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)]
+#[derive(Debug, Default)]
 struct CachePadded<T>(T);
diff --git a/tokio/src/time/driver/handle.rs b/tokio/src/time/driver/handle.rs
index 54b8a8bdf..d4d315dcd 100644
--- a/tokio/src/time/driver/handle.rs
+++ b/tokio/src/time/driver/handle.rs
@@ -1,22 +1,29 @@
-use crate::time::driver::Inner;
+use crate::loom::sync::{Arc, Mutex};
+use crate::time::driver::ClockTime;
 use std::fmt;
-use std::sync::{Arc, Weak};
 
 /// Handle to time driver instance.
 #[derive(Clone)]
 pub(crate) struct Handle {
-    inner: Weak<Inner>,
+    time_source: ClockTime,
+    inner: Arc<Mutex<super::Inner>>,
 }
 
 impl Handle {
     /// Creates a new timer `Handle` from a shared `Inner` timer state.
-    pub(crate) fn new(inner: Weak<Inner>) -> Self {
-        Handle { inner }
+    pub(super) fn new(inner: Arc<Mutex<super::Inner>>) -> Self {
+        let time_source = inner.lock().time_source.clone();
+        Handle { time_source, inner }
     }
 
-    /// Tries to return a strong ref to the inner
-    pub(crate) fn inner(&self) -> Option<Arc<Inner>> {
-        self.inner.upgrade()
+    /// Returns the time source associated with this handle
+    pub(super) fn time_source(&self) -> &ClockTime {
+        &self.time_source
+    }
+
+    /// Locks the driver's inner structure
+    pub(super) fn lock(&self) -> crate::loom::sync::MutexGuard<'_, super::Inner> {
+        self.inner.lock()
     }
 }
 
@@ -31,12 +38,12 @@ cfg_rt! {
         /// It can be triggered when `Builder::enable_time()` or
         /// `Builder::enable_all()` are not included in the builder.
         ///
-        /// It can also panic whenever a timer is created outside of a Tokio
-        /// runtime. That is why `rt.block_on(delay_for(...))` will panic,
-        /// since the function is executed outside of the runtime.
-        /// Whereas `rt.block_on(async {delay_for(...).await})` doesn't
-        /// panic. And this is because wrapping the function on an async makes it
-        /// lazy, and so gets executed inside the runtime successfuly without
+        /// It can also panic whenever a timer is created ouClockTimeide of a
+        /// Tokio runtime. That is why `rt.block_on(delay_for(...))` will panic,
+        /// since the function is executed ouClockTimeide of the runtime.
+        /// Whereas `rt.block_on(async {delay_for(...).await})` doesn't panic.
+        /// And this is because wrapping the function on an async makes it lazy,
+        /// and so gets executed inside the runtime successfuly without
         /// panicking.
         pub(crate) fn current() -> Self {
             crate::runtime::context::time_handle()
@@ -56,12 +63,12 @@ cfg_not_rt! {
         /// It can be triggered when `Builder::enable_time()` or
         /// `Builder::enable_all()` are not included in the builder.
         ///
-        /// It can also panic whenever a timer is created outside of a Tokio
+        /// It can also panic whenever a timer is created ouClockTimeide of a Tokio
         /// runtime. That is why `rt.block_on(delay_for(...))` will panic,
-        /// since the function is executed outside of the runtime.
+        /// since the function is executed ouClockTimeide of the runtime.
         /// Whereas `rt.block_on(async {delay_for(...).await})` doesn't
         /// panic. And this is because wrapping the function on an async makes it
-        /// lazy, and so gets executed inside the runtime successfuly without
+        /// lazy, and so geClockTime executed inside the runtime successfuly without
         /// panicking.
         pub(crate) fn current() -> Self {
             panic!("there is no timer running, must be called from the context of Tokio runtime or \
diff --git a/tokio/src/time/driver/mod.rs b/tokio/src/time/driver/mod.rs
index 8532c551f..917078efb 100644
--- a/tokio/src/time/driver/mod.rs
+++ b/tokio/src/time/driver/mod.rs
@@ -1,26 +1,29 @@
+// Currently, rust warns when an unsafe fn contains an unsafe {} block. However,
+// in the future, this will change to the reverse. For now, suppress this
+// warning and generally stick with being explicit about unsafety.
+#![allow(unused_unsafe)]
 #![cfg_attr(not(feature = "rt"), allow(dead_code))]
 
 //! Time driver
 
-mod atomic_stack;
-use self::atomic_stack::AtomicStack;
-
 mod entry;
-pub(super) use self::entry::Entry;
+pub(self) use self::entry::{EntryList, TimerEntry, TimerHandle, TimerShared};
 
 mod handle;
 pub(crate) use self::handle::Handle;
 
-use crate::loom::sync::atomic::{AtomicU64, AtomicUsize};
+mod wheel;
+
+pub(super) mod sleep;
+
+use crate::loom::sync::{Arc, Mutex};
 use crate::park::{Park, Unpark};
-use crate::time::{error::Error, wheel};
+use crate::time::error::Error;
 use crate::time::{Clock, Duration, Instant};
 
-use std::sync::atomic::Ordering::{Acquire, Relaxed, Release, SeqCst};
-
-use std::sync::Arc;
-use std::usize;
-use std::{cmp, fmt};
+use std::convert::TryInto;
+use std::fmt;
+use std::{num::NonZeroU64, ptr::NonNull, task::Waker};
 
 /// Time implementation that drives [`Sleep`][sleep], [`Interval`][interval], and [`Timeout`][timeout].
 ///
@@ -78,63 +81,96 @@ use std::{cmp, fmt};
 /// [timeout]: crate::time::Timeout
 /// [interval]: crate::time::Interval
 #[derive(Debug)]
-pub(crate) struct Driver<T: Park> {
+pub(crate) struct Driver<P: Park + 'static> {
+    /// Timing backend in use
+    time_source: ClockTime,
+
     /// Shared state
-    inner: Arc<Inner>,
+    inner: Handle,
+
+    /// Parker to delegate to
+    park: P,
+}
+
+/// A structure which handles conversion from Instants to u64 timestamps.
+#[derive(Debug, Clone)]
+pub(self) struct ClockTime {
+    clock: super::clock::Clock,
+    start_time: Instant,
+}
+
+impl ClockTime {
+    pub(self) fn new(clock: Clock) -> Self {
+        Self {
+            clock,
+            start_time: super::clock::now(),
+        }
+    }
+
+    pub(self) fn deadline_to_tick(&self, t: Instant) -> u64 {
+        // Round up to the end of a ms
+        self.instant_to_tick(t + Duration::from_nanos(999_999))
+    }
+
+    pub(self) fn instant_to_tick(&self, t: Instant) -> u64 {
+        // round up
+        let dur: Duration = t
+            .checked_duration_since(self.start_time)
+            .unwrap_or_else(|| Duration::from_secs(0));
+        let ms = dur.as_millis();
+
+        ms.try_into().expect("Duration too far into the future")
+    }
+
+    pub(self) fn tick_to_duration(&self, t: u64) -> Duration {
+        Duration::from_millis(t)
+    }
+
+    pub(self) fn now(&self) -> u64 {
+        self.instant_to_tick(self.clock.now())
+    }
+}
+
+/// Timer state shared between `Driver`, `Handle`, and `Registration`.
+pub(self) struct Inner {
+    /// Timing backend in use
+    time_source: ClockTime,
+
+    /// The last published timer `elapsed` value.
+    elapsed: u64,
+
+    /// The earliest time at which we promise to wake up without unparking
+    next_wake: Option<NonZeroU64>,
 
     /// Timer wheel
     wheel: wheel::Wheel,
 
-    /// Thread parker. The `Driver` park implementation delegates to this.
-    park: T,
-
-    /// Source of "now" instances
-    clock: Clock,
-
     /// True if the driver is being shutdown
     is_shutdown: bool,
-}
 
-/// 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.
+    /// Unparker that can be used to wake the time driver
     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>
+impl<P> Driver<P>
 where
-    T: Park,
+    P: Park + 'static,
 {
     /// Creates a new `Driver` instance that uses `park` to block the current
-    /// thread and `clock` to get the current `Instant`.
+    /// thread and `time_source` to get the current time and convert to ticks.
     ///
     /// 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());
+    pub(crate) fn new(park: P, clock: Clock) -> Driver<P> {
+        let time_source = ClockTime::new(clock);
+
+        let inner = Inner::new(time_source.clone(), Box::new(park.unpark()));
 
         Driver {
-            inner: Arc::new(Inner::new(clock.now(), unpark)),
-            wheel: wheel::Wheel::new(),
+            time_source,
+            inner: Handle::new(Arc::new(Mutex::new(inner))),
             park,
-            clock,
-            is_shutdown: false,
         }
     }
 
@@ -145,145 +181,38 @@ where
     /// `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))
+        self.inner.clone()
     }
 
-    /// Converts an `Expiration` to an `Instant`.
-    fn expiration_instant(&self, when: u64) -> Instant {
-        self.inner.start + Duration::from_millis(when)
-    }
+    fn park_internal(&mut self, limit: Option<Duration>) -> Result<(), P::Error> {
+        let clock = &self.time_source.clock;
 
-    /// Runs timer related logic
-    fn process(&mut self) {
-        let now = crate::time::ms(
-            self.clock.now() - self.inner.start,
-            crate::time::Round::Down,
-        );
+        let mut lock = self.inner.lock();
 
-        while let Some(entry) = self.wheel.poll(now) {
-            let when = entry.when_internal().expect("invalid internal entry state");
+        let next_wake = lock.wheel.next_expiration_time();
+        lock.next_wake =
+            next_wake.map(|t| NonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap()));
 
-            // Fire the entry
-            entry.fire(when);
+        drop(lock);
 
-            // Track that the entry has been fired
-            entry.set_when_internal(None);
-        }
-
-        // Update the elapsed cache
-        self.inner.elapsed.store(self.wheel.elapsed(), SeqCst);
-    }
-
-    /// Processes 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)) => {
-                    // Add the entry to the timer wheel
-                    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);
-        entry.set_when_internal(None);
-    }
-
-    /// Fires the entry if it needs to, otherwise queue it to be processed later.
-    fn add_entry(&mut self, entry: Arc<Entry>, when: u64) {
-        use crate::time::error::InsertError;
-
-        entry.set_when_internal(Some(when));
-
-        match self.wheel.insert(when, entry) {
-            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(Error::invalid());
-            }
-        }
-    }
-}
-
-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() {
+        match next_wake {
             Some(when) => {
-                let now = self.clock.now();
-                let deadline = self.expiration_instant(when);
+                let now = self.time_source.now();
+                // Note that we effectively round up to 1ms here - this avoids
+                // very short-duration microsecond-resolution sleeps that the OS
+                // might treat as zero-length.
+                let mut duration = self.time_source.tick_to_duration(when.saturating_sub(now));
 
-                if deadline > now {
-                    let dur = deadline - now;
-
-                    if self.clock.is_paused() {
-                        self.park.park_timeout(Duration::from_secs(0))?;
-                        self.clock.advance(dur);
-                    } else {
-                        self.park.park_timeout(dur)?;
+                if duration > Duration::from_millis(0) {
+                    if let Some(limit) = limit {
+                        duration = std::cmp::min(limit, duration);
                     }
-                } 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 {
-                    let duration = cmp::min(deadline - now, duration);
-
-                    if self.clock.is_paused() {
+                    if clock.is_paused() {
                         self.park.park_timeout(Duration::from_secs(0))?;
-                        self.clock.advance(duration);
+
+                        // Simulate advancing time
+                        clock.advance(duration);
                     } else {
                         self.park.park_timeout(duration)?;
                     }
@@ -292,42 +221,200 @@ where
                 }
             }
             None => {
-                self.park.park_timeout(duration)?;
+                if let Some(duration) = limit {
+                    if clock.is_paused() {
+                        self.park.park_timeout(Duration::from_secs(0))?;
+                        clock.advance(duration);
+                    } else {
+                        self.park.park_timeout(duration)?;
+                    }
+                } else {
+                    self.park.park()?;
+                }
             }
         }
 
-        self.process();
+        // Process pending timers after waking up
+        self.inner.process();
 
         Ok(())
     }
+}
 
-    fn shutdown(&mut self) {
-        if self.is_shutdown {
-            return;
+impl Handle {
+    /// Runs timer related logic, and returns the next wakeup time
+    pub(self) fn process(&self) {
+        let now = self.time_source().now();
+
+        self.process_at_time(now)
+    }
+
+    pub(self) fn process_at_time(&self, now: u64) {
+        let mut waker_list: [Option<Waker>; 32] = Default::default();
+        let mut waker_idx = 0;
+
+        let mut lock = self.lock();
+
+        assert!(now >= lock.elapsed);
+
+        while let Some(entry) = lock.wheel.poll(now) {
+            debug_assert!(unsafe { entry.is_pending() });
+
+            // SAFETY: We hold the driver lock, and just removed the entry from any linked lists.
+            if let Some(waker) = unsafe { entry.fire(Ok(())) } {
+                waker_list[waker_idx] = Some(waker);
+
+                waker_idx += 1;
+
+                if waker_idx == waker_list.len() {
+                    // Wake a batch of wakers. To avoid deadlock, we must do this with the lock temporarily dropped.
+                    drop(lock);
+
+                    for waker in waker_list.iter_mut() {
+                        waker.take().unwrap().wake();
+                    }
+
+                    waker_idx = 0;
+
+                    lock = self.lock();
+                }
+            }
         }
 
-        use std::u64;
+        // Update the elapsed cache
+        lock.elapsed = lock.wheel.elapsed();
+        lock.next_wake = lock
+            .wheel
+            .poll_at()
+            .map(|t| NonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap()));
 
-        // Shutdown the stack of entries to process, preventing any new entries
-        // from being pushed.
-        self.inner.process.shutdown();
+        drop(lock);
 
-        // Clear the wheel, using u64::MAX allows us to drain everything
-        let end_of_time = u64::MAX;
-
-        while let Some(entry) = self.wheel.poll(end_of_time) {
-            entry.error(Error::shutdown());
+        for waker in waker_list[0..waker_idx].iter_mut() {
+            waker.take().unwrap().wake();
         }
+    }
 
-        self.park.shutdown();
+    /// Removes a registered timer from the driver.
+    ///
+    /// The timer will be moved to the cancelled state. Wakers will _not_ be
+    /// invoked. If the timer is already completed, this function is a no-op.
+    ///
+    /// This function always acquires the driver lock, even if the entry does
+    /// not appear to be registered.
+    ///
+    /// SAFETY: The timer must not be registered with some other driver, and
+    /// `add_entry` must not be called concurrently.
+    pub(self) unsafe fn clear_entry(&self, entry: NonNull<TimerShared>) {
+        unsafe {
+            let mut lock = self.lock();
 
-        self.is_shutdown = true;
+            if entry.as_ref().might_be_registered() {
+                lock.wheel.remove(entry);
+            }
+
+            entry.as_ref().handle().fire(Ok(()));
+        }
+    }
+
+    /// Removes and re-adds an entry to the driver.
+    ///
+    /// SAFETY: The timer must be either unregistered, or registered with this
+    /// driver. No other threads are allowed to concurrently manipulate the
+    /// timer at all (the current thread should hold an exclusive reference to
+    /// the `TimerEntry`)
+    pub(self) unsafe fn reregister(&self, new_tick: u64, entry: NonNull<TimerShared>) {
+        let waker = unsafe {
+            let mut lock = self.lock();
+
+            // We may have raced with a firing/deregistration, so check before
+            // deregistering.
+            if unsafe { entry.as_ref().might_be_registered() } {
+                lock.wheel.remove(entry);
+            }
+
+            // Now that we have exclusive control of this entry, mint a handle to reinsert it.
+            let entry = entry.as_ref().handle();
+
+            if lock.is_shutdown {
+                unsafe { entry.fire(Err(crate::time::error::Error::shutdown())) }
+            } else {
+                entry.set_expiration(new_tick);
+
+                // Note: We don't have to worry about racing with some other resetting
+                // thread, because add_entry and reregister require exclusive control of
+                // the timer entry.
+                match unsafe { lock.wheel.insert(entry) } {
+                    Ok(when) => {
+                        if lock
+                            .next_wake
+                            .map(|next_wake| when < next_wake.get())
+                            .unwrap_or(true)
+                        {
+                            lock.unpark.unpark();
+                        }
+
+                        None
+                    }
+                    Err((entry, super::error::InsertError::Elapsed)) => unsafe {
+                        entry.fire(Ok(()))
+                    },
+                }
+            }
+
+            // Must release lock before invoking waker to avoid the risk of deadlock.
+        };
+
+        // The timer was fired synchronously as a result of the reregistration.
+        // Wake the waker; this is needed because we might reset _after_ a poll,
+        // and otherwise the task won't be awoken to poll again.
+        if let Some(waker) = waker {
+            waker.wake();
+        }
     }
 }
 
-impl<T> Drop for Driver<T>
+impl<P> Park for Driver<P>
 where
-    T: Park,
+    P: Park + 'static,
+{
+    type Unpark = P::Unpark;
+    type Error = P::Error;
+
+    fn unpark(&self) -> Self::Unpark {
+        self.park.unpark()
+    }
+
+    fn park(&mut self) -> Result<(), Self::Error> {
+        self.park_internal(None)
+    }
+
+    fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
+        self.park_internal(Some(duration))
+    }
+
+    fn shutdown(&mut self) {
+        let mut lock = self.inner.lock();
+
+        if lock.is_shutdown {
+            return;
+        }
+
+        lock.is_shutdown = true;
+
+        drop(lock);
+
+        // Advance time forward to the end of time.
+
+        self.inner.process_at_time(u64::MAX);
+
+        self.park.shutdown();
+    }
+}
+
+impl<P> Drop for Driver<P>
+where
+    P: Park + 'static,
 {
     fn drop(&mut self) {
         self.shutdown();
@@ -337,69 +424,16 @@ where
 // ===== impl Inner =====
 
 impl Inner {
-    fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
+    pub(self) fn new(time_source: ClockTime, unpark: Box<dyn Unpark>) -> Self {
         Inner {
-            num: AtomicUsize::new(0),
-            elapsed: AtomicU64::new(0),
-            process: AtomicStack::new(),
-            start,
+            time_source,
+            elapsed: 0,
+            next_wake: None,
             unpark,
+            wheel: wheel::Wheel::new(),
+            is_shutdown: false,
         }
     }
-
-    fn elapsed(&self) -> u64 {
-        self.elapsed.load(SeqCst)
-    }
-
-    #[cfg(all(test, loom))]
-    fn num(&self, ordering: std::sync::atomic::Ordering) -> usize {
-        self.num.load(ordering)
-    }
-
-    /// Increments the number of active timeouts
-    fn increment(&self) -> Result<(), Error> {
-        let mut curr = self.num.load(Relaxed);
-        loop {
-            if curr == MAX_TIMEOUTS {
-                return Err(Error::at_capacity());
-            }
-
-            match self
-                .num
-                .compare_exchange_weak(curr, curr + 1, Release, Relaxed)
-            {
-                Ok(_) => return Ok(()),
-                Err(next) => curr = next,
-            }
-        }
-    }
-
-    /// Decrements the number of active timeouts
-    fn decrement(&self) {
-        let prev = self.num.fetch_sub(1, Acquire);
-        debug_assert!(prev <= MAX_TIMEOUTS);
-    }
-
-    /// add the entry to the "process queue".  entries are not immediately
-    /// pushed into the timer wheel but are instead pushed into the
-    /// process queue and then moved from the process queue into the timer
-    /// wheel on next `process`
-    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 {
@@ -408,5 +442,5 @@ impl fmt::Debug for Inner {
     }
 }
 
-#[cfg(all(test, loom))]
+#[cfg(test)]
 mod tests;
diff --git a/tokio/src/time/sleep.rs b/tokio/src/time/driver/sleep.rs
similarity index 77%
rename from tokio/src/time/sleep.rs
rename to tokio/src/time/driver/sleep.rs
index 2bd4eb1a4..9f358c34e 100644
--- a/tokio/src/time/sleep.rs
+++ b/tokio/src/time/driver/sleep.rs
@@ -1,9 +1,9 @@
-use crate::time::driver::{Entry, Handle};
+use crate::time::driver::{Handle, TimerEntry};
 use crate::time::{error::Error, Duration, Instant};
 
 use std::future::Future;
 use std::pin::Pin;
-use std::sync::Arc;
+
 use std::task::{self, Poll};
 
 /// Waits until `deadline` is reached.
@@ -17,7 +17,7 @@ use std::task::{self, Poll};
 /// Canceling a sleep instance is done by dropping the returned future. No additional
 /// cleanup work is required.
 pub fn sleep_until(deadline: Instant) -> Sleep {
-    Sleep::new_timeout(deadline, Duration::from_millis(0))
+    Sleep::new_timeout(deadline)
 }
 
 /// Waits until `duration` has elapsed.
@@ -62,23 +62,24 @@ pub fn sleep(duration: Duration) -> Sleep {
 #[derive(Debug)]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct Sleep {
-    /// The link between the `Sleep` instance and the timer that drives it.
-    ///
-    /// This also stores the `deadline` value.
-    entry: Arc<Entry>,
+    deadline: Instant,
+
+    // The link between the `Sleep` instance and the timer that drives it.
+    // This will be unboxed in tokio 1.0
+    entry: Pin<Box<TimerEntry>>,
 }
 
 impl Sleep {
-    pub(crate) fn new_timeout(deadline: Instant, duration: Duration) -> Sleep {
+    pub(crate) fn new_timeout(deadline: Instant) -> Sleep {
         let handle = Handle::current();
-        let entry = Entry::new(&handle, deadline, duration);
+        let entry = Box::pin(TimerEntry::new(&handle, deadline));
 
-        Sleep { entry }
+        Sleep { deadline, entry }
     }
 
     /// Returns the instant at which the future will complete.
     pub fn deadline(&self) -> Instant {
-        self.entry.time_ref().deadline
+        self.deadline
     }
 
     /// Returns `true` if `Sleep` has elapsed.
@@ -96,18 +97,15 @@ impl Sleep {
     /// This function can be called both before and after the future has
     /// completed.
     pub fn reset(&mut self, deadline: Instant) {
-        unsafe {
-            self.entry.time_mut().deadline = deadline;
-        }
-
-        Entry::reset(&mut self.entry);
+        self.entry.as_mut().reset(deadline);
+        self.deadline = deadline;
     }
 
-    fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
+    fn poll_elapsed(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
         // Keep track of task budget
         let coop = ready!(crate::coop::poll_proceed(cx));
 
-        self.entry.poll_elapsed(cx).map(move |r| {
+        self.entry.as_mut().poll_elapsed(cx).map(move |r| {
             coop.made_progress();
             r
         })
@@ -117,7 +115,7 @@ impl Sleep {
 impl Future for Sleep {
     type Output = ();
 
-    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
         // `poll_elapsed` can return an error in two cases:
         //
         // - AtCapacity: this is a pathological case where far too many
@@ -127,15 +125,9 @@ impl Future for Sleep {
         // Both cases are extremely rare, and pretty accurately fit into
         // "logic errors", so we just panic in this case. A user couldn't
         // really do much better if we passed the error onwards.
-        match ready!(self.poll_elapsed(cx)) {
+        match ready!(self.as_mut().poll_elapsed(cx)) {
             Ok(()) => Poll::Ready(()),
             Err(e) => panic!("timer error: {}", e),
         }
     }
 }
-
-impl Drop for Sleep {
-    fn drop(&mut self) {
-        Entry::cancel(&self.entry);
-    }
-}
diff --git a/tokio/src/time/driver/tests/mod.rs b/tokio/src/time/driver/tests/mod.rs
index 88ff5525d..e6af798f0 100644
--- a/tokio/src/time/driver/tests/mod.rs
+++ b/tokio/src/time/driver/tests/mod.rs
@@ -1,18 +1,249 @@
-use crate::park::Unpark;
-use crate::time::driver::Inner;
-use crate::time::Instant;
+use std::{task::Context, time::Duration};
 
-use loom::thread;
+#[cfg(not(loom))]
+use futures::task::noop_waker_ref;
 
-use std::sync::atomic::Ordering;
-use std::sync::Arc;
+use crate::loom::sync::{Arc, Mutex};
+use crate::loom::thread;
+use crate::{
+    loom::sync::atomic::{AtomicBool, Ordering},
+    park::Unpark,
+};
 
-struct MockUnpark;
+use super::{Handle, TimerEntry};
 
+struct MockUnpark {}
 impl Unpark for MockUnpark {
     fn unpark(&self) {}
 }
+impl MockUnpark {
+    fn mock() -> Box<dyn Unpark> {
+        Box::new(Self {})
+    }
+}
 
+fn block_on<T>(f: impl std::future::Future<Output = T>) -> T {
+    #[cfg(loom)]
+    return loom::future::block_on(f);
+
+    #[cfg(not(loom))]
+    return futures::executor::block_on(f);
+}
+
+fn model(f: impl Fn() + Send + Sync + 'static) {
+    #[cfg(loom)]
+    loom::model(f);
+
+    #[cfg(not(loom))]
+    f();
+}
+
+#[test]
+fn single_timer() {
+    model(|| {
+        let clock = crate::time::clock::Clock::new();
+        let time_source = super::ClockTime::new(clock.clone());
+
+        let inner = super::Inner::new(time_source.clone(), MockUnpark::mock());
+        let handle = Handle::new(Arc::new(Mutex::new(inner)));
+
+        let handle_ = handle.clone();
+        let jh = thread::spawn(move || {
+            let entry = TimerEntry::new(&handle_, clock.now() + Duration::from_secs(1));
+            pin!(entry);
+
+            block_on(futures::future::poll_fn(|cx| {
+                entry.as_mut().poll_elapsed(cx)
+            }))
+            .unwrap();
+        });
+
+        thread::yield_now();
+
+        // This may or may not return Some (depending on how it races with the
+        // thread). If it does return None, however, the timer should complete
+        // synchronously.
+        handle.process_at_time(time_source.now() + 2_000_000_000);
+
+        jh.join().unwrap();
+    })
+}
+
+#[test]
+fn drop_timer() {
+    model(|| {
+        let clock = crate::time::clock::Clock::new();
+        let time_source = super::ClockTime::new(clock.clone());
+
+        let inner = super::Inner::new(time_source.clone(), MockUnpark::mock());
+        let handle = Handle::new(Arc::new(Mutex::new(inner)));
+
+        let handle_ = handle.clone();
+        let jh = thread::spawn(move || {
+            let entry = TimerEntry::new(&handle_, clock.now() + Duration::from_secs(1));
+            pin!(entry);
+
+            let _ = entry
+                .as_mut()
+                .poll_elapsed(&mut Context::from_waker(futures::task::noop_waker_ref()));
+            let _ = entry
+                .as_mut()
+                .poll_elapsed(&mut Context::from_waker(futures::task::noop_waker_ref()));
+        });
+
+        thread::yield_now();
+
+        // advance 2s in the future.
+        handle.process_at_time(time_source.now() + 2_000_000_000);
+
+        jh.join().unwrap();
+    })
+}
+
+#[test]
+fn change_waker() {
+    model(|| {
+        let clock = crate::time::clock::Clock::new();
+        let time_source = super::ClockTime::new(clock.clone());
+
+        let inner = super::Inner::new(time_source.clone(), MockUnpark::mock());
+        let handle = Handle::new(Arc::new(Mutex::new(inner)));
+
+        let handle_ = handle.clone();
+        let jh = thread::spawn(move || {
+            let entry = TimerEntry::new(&handle_, clock.now() + Duration::from_secs(1));
+            pin!(entry);
+
+            let _ = entry
+                .as_mut()
+                .poll_elapsed(&mut Context::from_waker(futures::task::noop_waker_ref()));
+
+            block_on(futures::future::poll_fn(|cx| {
+                entry.as_mut().poll_elapsed(cx)
+            }))
+            .unwrap();
+        });
+
+        thread::yield_now();
+
+        // advance 2s
+        handle.process_at_time(time_source.now() + 2_000_000_000);
+
+        jh.join().unwrap();
+    })
+}
+
+#[test]
+fn reset_future() {
+    model(|| {
+        let finished_early = Arc::new(AtomicBool::new(false));
+
+        let clock = crate::time::clock::Clock::new();
+        let time_source = super::ClockTime::new(clock.clone());
+
+        let inner = super::Inner::new(time_source.clone(), MockUnpark::mock());
+        let handle = Handle::new(Arc::new(Mutex::new(inner)));
+
+        let handle_ = handle.clone();
+        let finished_early_ = finished_early.clone();
+        let start = clock.now();
+
+        let jh = thread::spawn(move || {
+            let entry = TimerEntry::new(&handle_, start + Duration::from_secs(1));
+            pin!(entry);
+
+            let _ = entry
+                .as_mut()
+                .poll_elapsed(&mut Context::from_waker(futures::task::noop_waker_ref()));
+
+            entry.as_mut().reset(start + Duration::from_secs(2));
+
+            // shouldn't complete before 2s
+            block_on(futures::future::poll_fn(|cx| {
+                entry.as_mut().poll_elapsed(cx)
+            }))
+            .unwrap();
+
+            finished_early_.store(true, Ordering::Relaxed);
+        });
+
+        thread::yield_now();
+
+        // This may or may not return a wakeup time.
+        handle.process_at_time(time_source.instant_to_tick(start + Duration::from_millis(1500)));
+
+        assert!(!finished_early.load(Ordering::Relaxed));
+
+        handle.process_at_time(time_source.instant_to_tick(start + Duration::from_millis(2500)));
+
+        jh.join().unwrap();
+
+        assert!(finished_early.load(Ordering::Relaxed));
+    })
+}
+
+#[test]
+#[cfg(not(loom))]
+fn poll_process_levels() {
+    let clock = crate::time::clock::Clock::new();
+    clock.pause();
+
+    let time_source = super::ClockTime::new(clock.clone());
+
+    let inner = super::Inner::new(time_source, MockUnpark::mock());
+    let handle = Handle::new(Arc::new(Mutex::new(inner)));
+
+    let mut entries = vec![];
+
+    for i in 0..1024 {
+        let mut entry = Box::pin(TimerEntry::new(
+            &handle,
+            clock.now() + Duration::from_millis(i),
+        ));
+
+        let _ = entry
+            .as_mut()
+            .poll_elapsed(&mut Context::from_waker(noop_waker_ref()));
+
+        entries.push(entry);
+    }
+
+    for t in 1..1024 {
+        handle.process_at_time(t as u64);
+        for (deadline, future) in entries.iter_mut().enumerate() {
+            let mut context = Context::from_waker(noop_waker_ref());
+            if deadline <= t {
+                assert!(future.as_mut().poll_elapsed(&mut context).is_ready());
+            } else {
+                assert!(future.as_mut().poll_elapsed(&mut context).is_pending());
+            }
+        }
+    }
+}
+
+#[test]
+#[cfg(not(loom))]
+fn poll_process_levels_targeted() {
+    let mut context = Context::from_waker(noop_waker_ref());
+
+    let clock = crate::time::clock::Clock::new();
+    clock.pause();
+
+    let time_source = super::ClockTime::new(clock.clone());
+
+    let inner = super::Inner::new(time_source, MockUnpark::mock());
+    let handle = Handle::new(Arc::new(Mutex::new(inner)));
+
+    let e1 = TimerEntry::new(&handle, clock.now() + Duration::from_millis(193));
+    pin!(e1);
+
+    handle.process_at_time(62);
+    assert!(e1.as_mut().poll_elapsed(&mut context).is_pending());
+    handle.process_at_time(192);
+    handle.process_at_time(192);
+}
+
+/*
 #[test]
 fn balanced_incr_and_decr() {
     const OPS: usize = 5;
@@ -53,3 +284,4 @@ fn balanced_incr_and_decr() {
         assert_eq!(inner.num(Ordering::SeqCst), 0);
     })
 }
+*/
diff --git a/tokio/src/time/wheel/level.rs b/tokio/src/time/driver/wheel/level.rs
similarity index 73%
rename from tokio/src/time/wheel/level.rs
rename to tokio/src/time/driver/wheel/level.rs
index d51d26a03..58280b10a 100644
--- a/tokio/src/time/wheel/level.rs
+++ b/tokio/src/time/driver/wheel/level.rs
@@ -1,7 +1,8 @@
-use super::{Item, OwnedItem};
-use crate::time::wheel::Stack;
+use crate::time::driver::TimerHandle;
 
-use std::fmt;
+use crate::time::driver::{EntryList, TimerShared};
+
+use std::{fmt, ptr::NonNull};
 
 /// Wheel for a single level in the timer. This wheel contains 64 slots.
 pub(crate) struct Level {
@@ -16,8 +17,8 @@ pub(crate) struct Level {
     /// The least-significant bit represents slot zero.
     occupied: u64,
 
-    /// Slots
-    slot: [Stack; LEVEL_MULT],
+    /// Slots. We access these via the EntryInner `current_list` as well, so this needs to be an UnsafeCell.
+    slot: [EntryList; LEVEL_MULT],
 }
 
 /// Indicates when a slot must be processed next.
@@ -52,7 +53,7 @@ impl Level {
         // However, that is only supported for arrays of size
         // 32 or fewer.  So in our case we have to explicitly
         // invoke the constructor for each array element.
-        let ctor = Stack::default;
+        let ctor = || EntryList::default();
 
         Level {
             level,
@@ -144,14 +145,38 @@ impl Level {
 
         // TODO: This can probably be simplified w/ power of 2 math
         let level_start = now - (now % level_range);
-        let deadline = level_start + slot as u64 * slot_range;
+        let mut deadline = level_start + slot as u64 * slot_range;
+
+        if deadline <= now {
+            // A timer is in a slot "prior" to the current time. This can occur
+            // because we do not have an infinite hierarchy of timer levels, and
+            // eventually a timer scheduled for a very distant time might end up
+            // being placed in a slot that is beyond the end of all of the
+            // arrays.
+            //
+            // To deal with this, we first limit timers to being scheduled no
+            // more than MAX_DURATION ticks in the future; that is, they're at
+            // most one rotation of the top level away. Then, we force timers
+            // that logically would go into the top+1 level, to instead go into
+            // the top level's slots.
+            //
+            // What this means is that the top level's slots act as a
+            // pseudo-ring buffer, and we rotate around them indefinitely. If we
+            // compute a deadline before now, and it's the top level, it
+            // therefore means we're actually looking at a slot in the future.
+            debug_assert_eq!(self.level, super::NUM_LEVELS - 1);
+
+            deadline += level_range;
+        }
 
         debug_assert!(
             deadline >= now,
-            "deadline={}; now={}; level={}; slot={}; occupied={:b}",
+            "deadline={:016X}; now={:016X}; level={}; lr={:016X}, sr={:016X}, slot={}; occupied={:b}",
             deadline,
             now,
             self.level,
+            level_range,
+            slot_range,
             slot,
             self.occupied
         );
@@ -177,18 +202,18 @@ impl Level {
         Some(slot)
     }
 
-    pub(crate) fn add_entry(&mut self, when: u64, item: OwnedItem) {
-        let slot = slot_for(when, self.level);
+    pub(crate) unsafe fn add_entry(&mut self, item: TimerHandle) {
+        let slot = slot_for(item.cached_when(), self.level);
+
+        self.slot[slot].push_front(item);
 
-        self.slot[slot].push(item);
         self.occupied |= occupied_bit(slot);
     }
 
-    pub(crate) fn remove_entry(&mut self, when: u64, item: &Item) {
-        let slot = slot_for(when, self.level);
-
-        self.slot[slot].remove(item);
+    pub(crate) unsafe fn remove_entry(&mut self, item: NonNull<TimerShared>) {
+        let slot = slot_for(unsafe { item.as_ref().cached_when() }, self.level);
 
+        unsafe { self.slot[slot].remove(item) };
         if self.slot[slot].is_empty() {
             // The bit is currently set
             debug_assert!(self.occupied & occupied_bit(slot) != 0);
@@ -198,17 +223,10 @@ impl Level {
         }
     }
 
-    pub(crate) fn pop_entry_slot(&mut self, slot: usize) -> Option<OwnedItem> {
-        let ret = self.slot[slot].pop();
+    pub(crate) fn take_slot(&mut self, slot: usize) -> EntryList {
+        self.occupied &= !occupied_bit(slot);
 
-        if ret.is_some() && self.slot[slot].is_empty() {
-            // The bit is currently set
-            debug_assert!(self.occupied & occupied_bit(slot) != 0);
-
-            self.occupied ^= occupied_bit(slot);
-        }
-
-        ret
+        std::mem::take(&mut self.slot[slot])
     }
 }
 
diff --git a/tokio/src/time/wheel/mod.rs b/tokio/src/time/driver/wheel/mod.rs
similarity index 62%
rename from tokio/src/time/wheel/mod.rs
rename to tokio/src/time/driver/wheel/mod.rs
index 85ed2f113..e9df87afa 100644
--- a/tokio/src/time/wheel/mod.rs
+++ b/tokio/src/time/driver/wheel/mod.rs
@@ -1,17 +1,13 @@
-use crate::time::{driver::Entry, error::InsertError};
+use crate::time::driver::{TimerHandle, TimerShared};
+use crate::time::error::InsertError;
 
 mod level;
 pub(crate) use self::level::Expiration;
 use self::level::Level;
 
-mod stack;
-pub(crate) use self::stack::Stack;
+use std::ptr::NonNull;
 
-use std::sync::Arc;
-use std::usize;
-
-pub(super) type Item = Entry;
-pub(super) type OwnedItem = Arc<Item>;
+use super::EntryList;
 
 /// Timing wheel implementation.
 ///
@@ -40,6 +36,9 @@ pub(crate) struct Wheel {
     /// * ~ 4 hr slots / ~ 12 day range
     /// * ~ 12 day slots / ~ 2 yr range
     levels: Vec<Level>,
+
+    /// Entries queued for firing
+    pending: EntryList,
 }
 
 /// Number of levels. Each level has 64 slots. By using 6 levels with 64 slots
@@ -48,14 +47,18 @@ pub(crate) struct Wheel {
 const NUM_LEVELS: usize = 6;
 
 /// The maximum duration of a `Sleep`
-const MAX_DURATION: u64 = (1 << (6 * NUM_LEVELS)) - 1;
+pub(super) const MAX_DURATION: u64 = (1 << (6 * NUM_LEVELS)) - 1;
 
 impl Wheel {
     /// Create a new timing wheel
     pub(crate) fn new() -> Wheel {
         let levels = (0..NUM_LEVELS).map(Level::new).collect();
 
-        Wheel { elapsed: 0, levels }
+        Wheel {
+            elapsed: 0,
+            levels,
+            pending: EntryList::new(),
+        }
     }
 
     /// Return the number of milliseconds that have elapsed since the timing
@@ -68,14 +71,8 @@ impl Wheel {
     ///
     /// # Arguments
     ///
-    /// * `when`: is the instant at which the entry should be fired. It is
-    ///           represented as the number of milliseconds since the creation
-    ///           of the timing wheel.
-    ///
     /// * `item`: The item to insert into the wheel.
     ///
-    /// * `store`: The slab or `()` when using heap storage.
-    ///
     /// # Return
     ///
     /// Returns `Ok` when the item is successfully inserted, `Err` otherwise.
@@ -85,21 +82,28 @@ impl Wheel {
     /// immediately.
     ///
     /// `Err(Invalid)` indicates an invalid `when` argument as been supplied.
-    pub(crate) fn insert(
+    ///
+    /// # Safety
+    ///
+    /// This function registers item into an intrusive linked list. The caller
+    /// must ensure that `item` is pinned and will not be dropped without first
+    /// being deregistered.
+    pub(crate) unsafe fn insert(
         &mut self,
-        when: u64,
-        item: OwnedItem,
-    ) -> Result<(), (OwnedItem, InsertError)> {
+        item: TimerHandle,
+    ) -> Result<u64, (TimerHandle, InsertError)> {
+        let when = item.sync_when();
+
         if when <= self.elapsed {
             return Err((item, InsertError::Elapsed));
-        } else if when - self.elapsed > MAX_DURATION {
-            return Err((item, InsertError::Invalid));
         }
 
         // Get the level at which the entry should be stored
         let level = self.level_for(when);
 
-        self.levels[level].add_entry(when, item);
+        unsafe {
+            self.levels[level].add_entry(item);
+        }
 
         debug_assert!({
             self.levels[level]
@@ -108,15 +112,21 @@ impl Wheel {
                 .unwrap_or(true)
         });
 
-        Ok(())
+        Ok(when)
     }
 
-    /// Remove `item` from thee timing wheel.
-    pub(crate) fn remove(&mut self, item: &Item) {
-        let when = item.when();
-        let level = self.level_for(when);
+    /// Remove `item` from the timing wheel.
+    pub(crate) unsafe fn remove(&mut self, item: NonNull<TimerShared>) {
+        unsafe {
+            if !item.as_ref().might_be_registered() {
+                self.pending.remove(item);
+            } else {
+                let when = item.as_ref().cached_when();
+                let level = self.level_for(when);
 
-        self.levels[level].remove_entry(when, item);
+                self.levels[level].remove_entry(item);
+            }
+        }
     }
 
     /// Instant at which to poll
@@ -125,8 +135,12 @@ impl Wheel {
     }
 
     /// Advances the timer up to the instant represented by `now`.
-    pub(crate) fn poll(&mut self, now: u64) -> Option<OwnedItem> {
+    pub(crate) fn poll(&mut self, now: u64) -> Option<TimerHandle> {
         loop {
+            if let Some(handle) = self.pending.pop_back() {
+                return Some(handle);
+            }
+
             // under what circumstances is poll.expiration Some vs. None?
             let expiration = self.next_expiration().and_then(|expiration| {
                 if expiration.deadline > now {
@@ -137,10 +151,9 @@ impl Wheel {
             });
 
             match expiration {
+                Some(ref expiration) if expiration.deadline > now => return None,
                 Some(ref expiration) => {
-                    if let Some(item) = self.poll_expiration(expiration) {
-                        return Some(item);
-                    }
+                    self.process_expiration(expiration);
 
                     self.set_elapsed(expiration.deadline);
                 }
@@ -150,14 +163,25 @@ impl Wheel {
                     // the current list of timers.  advance to the poll's
                     // current time and do nothing else.
                     self.set_elapsed(now);
-                    return None;
+                    break;
                 }
             }
         }
+
+        self.pending.pop_back()
     }
 
     /// Returns the instant at which the next timeout expires.
     fn next_expiration(&self) -> Option<Expiration> {
+        if !self.pending.is_empty() {
+            // Expire immediately as we have things pending firing
+            return Some(Expiration {
+                level: 0,
+                slot: 0,
+                deadline: self.elapsed,
+            });
+        }
+
         // Check all levels
         for level in 0..NUM_LEVELS {
             if let Some(expiration) = self.levels[level].next_expiration(self.elapsed) {
@@ -172,6 +196,12 @@ impl Wheel {
         None
     }
 
+    /// Returns the tick at which this timer wheel next needs to perform some
+    /// processing, or None if there are no timers registered.
+    pub(super) fn next_expiration_time(&self) -> Option<u64> {
+        self.next_expiration().map(|ex| ex.deadline)
+    }
+
     /// Used for debug assertions
     fn no_expirations_before(&self, start_level: usize, before: u64) -> bool {
         let mut res = true;
@@ -189,24 +219,41 @@ impl Wheel {
 
     /// iteratively find entries that are between the wheel's current
     /// time and the expiration time.  for each in that population either
-    /// return it for notification (in the case of the last level) or tier
+    /// queue it for notification (in the case of the last level) or tier
     /// it down to the next level (in all other cases).
-    pub(crate) fn poll_expiration(&mut self, expiration: &Expiration) -> Option<OwnedItem> {
-        while let Some(item) = self.pop_entry(expiration) {
+    pub(crate) fn process_expiration(&mut self, expiration: &Expiration) {
+        // Note that we need to take _all_ of the entries off the list before
+        // processing any of them. This is important because it's possible that
+        // those entries might need to be reinserted into the same slot.
+        //
+        // This happens only on the highest level, when an entry is inserted
+        // more than MAX_DURATION into the future. When this happens, we wrap
+        // around, and process some entries a multiple of MAX_DURATION before
+        // they actually need to be dropped down a level. We then reinsert them
+        // back into the same position; we must make sure we don't then process
+        // those entries again or we'll end up in an infinite loop.
+        let mut entries = self.take_entries(expiration);
+
+        while let Some(item) = entries.pop_back() {
             if expiration.level == 0 {
-                debug_assert_eq!(item.when(), expiration.deadline);
+                debug_assert_eq!(unsafe { item.cached_when() }, expiration.deadline);
+            }
 
-                return Some(item);
-            } else {
-                let when = item.when();
-
-                let next_level = expiration.level - 1;
-
-                self.levels[next_level].add_entry(when, item);
+            // Try to expire the entry; this is cheap (doesn't synchronize) if
+            // the timer is not expired, and updates cached_when.
+            match unsafe { item.mark_pending(expiration.deadline) } {
+                Ok(()) => {
+                    // Item was expired
+                    self.pending.push_front(item);
+                }
+                Err(expiration_tick) => {
+                    let level = level_for(expiration.deadline, expiration_tick);
+                    unsafe {
+                        self.levels[level].add_entry(item);
+                    }
+                }
             }
         }
-
-        None
     }
 
     fn set_elapsed(&mut self, when: u64) {
@@ -222,8 +269,10 @@ impl Wheel {
         }
     }
 
-    fn pop_entry(&mut self, expiration: &Expiration) -> Option<OwnedItem> {
-        self.levels[expiration.level].pop_entry_slot(expiration.slot)
+    /// Obtains the list of entries that need processing for the given expiration.
+    ///
+    fn take_entries(&mut self, expiration: &Expiration) -> EntryList {
+        self.levels[expiration.level].take_slot(expiration.slot)
     }
 
     fn level_for(&self, when: u64) -> usize {
@@ -232,12 +281,18 @@ impl Wheel {
 }
 
 fn level_for(elapsed: u64, when: u64) -> usize {
-    let masked = elapsed ^ when;
+    let mut masked = elapsed ^ when;
+
+    if masked >= MAX_DURATION {
+        // Fudge the timer into the top level
+        masked = MAX_DURATION - 1;
+    }
 
     assert!(masked != 0, "elapsed={}; when={}", elapsed, when);
 
     let leading_zeros = masked.leading_zeros() as usize;
     let significant = 63 - leading_zeros;
+
     significant / 6
 }
 
diff --git a/tokio/src/time/wheel/stack.rs b/tokio/src/time/driver/wheel/stack.rs
similarity index 100%
rename from tokio/src/time/wheel/stack.rs
rename to tokio/src/time/driver/wheel/stack.rs
diff --git a/tokio/src/time/error.rs b/tokio/src/time/error.rs
index 24395c47f..8674febe9 100644
--- a/tokio/src/time/error.rs
+++ b/tokio/src/time/error.rs
@@ -23,17 +23,23 @@ use std::fmt;
 ///   way to do this would be dropping the future that issued the timer operation.
 ///
 /// [shed load]: https://en.wikipedia.org/wiki/Load_Shedding
-#[derive(Debug)]
+#[derive(Debug, Copy, Clone)]
 pub struct Error(Kind);
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
 #[repr(u8)]
-enum Kind {
+pub(crate) enum Kind {
     Shutdown = 1,
     AtCapacity = 2,
     Invalid = 3,
 }
 
+impl From<Kind> for Error {
+    fn from(k: Kind) -> Self {
+        Error(k)
+    }
+}
+
 /// Error returned by `Timeout`.
 #[derive(Debug, PartialEq)]
 pub struct Elapsed(());
@@ -41,7 +47,6 @@ pub struct Elapsed(());
 #[derive(Debug)]
 pub(crate) enum InsertError {
     Elapsed,
-    Invalid,
 }
 
 // ===== impl Error =====
@@ -76,19 +81,6 @@ impl Error {
     pub fn is_invalid(&self) -> bool {
         matches!(self.0, Kind::Invalid)
     }
-
-    pub(crate) fn as_u8(&self) -> u8 {
-        self.0 as u8
-    }
-
-    pub(crate) fn from_u8(n: u8) -> Self {
-        Error(match n {
-            1 => Shutdown,
-            2 => AtCapacity,
-            3 => Invalid,
-            _ => panic!("u8 does not correspond to any time error variant"),
-        })
-    }
 }
 
 impl error::Error for Error {}
diff --git a/tokio/src/time/mod.rs b/tokio/src/time/mod.rs
index 29af7175d..6f200dccb 100644
--- a/tokio/src/time/mod.rs
+++ b/tokio/src/time/mod.rs
@@ -77,9 +77,11 @@
 //!
 //! #[tokio::main]
 //! async fn main() {
-//!     let mut interval = time::interval(time::Duration::from_secs(2));
+//!     let interval = time::interval(time::Duration::from_secs(2));
+//!     tokio::pin!(interval);
+//!
 //!     for _i in 0..5 {
-//!         interval.tick().await;
+//!         interval.as_mut().tick().await;
 //!         task_that_takes_a_second().await;
 //!     }
 //! }
@@ -93,11 +95,11 @@ pub(crate) use self::clock::Clock;
 #[cfg(feature = "test-util")]
 pub use clock::{advance, pause, resume};
 
-mod sleep;
-pub use sleep::{sleep, sleep_until, Sleep};
-
 pub(crate) mod driver;
 
+#[doc(inline)]
+pub use driver::sleep::{sleep, sleep_until, Sleep};
+
 pub mod error;
 
 mod instant;
@@ -110,8 +112,6 @@ mod timeout;
 #[doc(inline)]
 pub use timeout::{timeout, timeout_at, Timeout};
 
-mod wheel;
-
 #[cfg(test)]
 #[cfg(not(loom))]
 mod tests;
@@ -119,32 +119,3 @@ mod tests;
 // Re-export for convenience
 #[doc(no_inline)]
 pub use std::time::Duration;
-
-// ===== Internal utils =====
-
-enum Round {
-    Up,
-    Down,
-}
-
-/// Convert a `Duration` to milliseconds, rounding up and saturating at
-/// `u64::MAX`.
-///
-/// The saturating is fine because `u64::MAX` milliseconds are still many
-/// million years.
-#[inline]
-fn ms(duration: Duration, round: Round) -> u64 {
-    const NANOS_PER_MILLI: u32 = 1_000_000;
-    const MILLIS_PER_SEC: u64 = 1_000;
-
-    // Round up.
-    let millis = match round {
-        Round::Up => (duration.subsec_nanos() + NANOS_PER_MILLI - 1) / NANOS_PER_MILLI,
-        Round::Down => duration.subsec_millis(),
-    };
-
-    duration
-        .as_secs()
-        .saturating_mul(MILLIS_PER_SEC)
-        .saturating_add(u64::from(millis))
-}
diff --git a/tokio/src/time/tests/mod.rs b/tokio/src/time/tests/mod.rs
index fae67da98..35e1060ac 100644
--- a/tokio/src/time/tests/mod.rs
+++ b/tokio/src/time/tests/mod.rs
@@ -8,7 +8,7 @@ fn assert_sync<T: Sync>() {}
 
 #[test]
 fn registration_is_send_and_sync() {
-    use crate::time::sleep::Sleep;
+    use crate::time::Sleep;
 
     assert_send::<Sleep>();
     assert_sync::<Sleep>();
diff --git a/tokio/src/time/tests/test_sleep.rs b/tokio/src/time/tests/test_sleep.rs
index c8d931a86..77ca07e31 100644
--- a/tokio/src/time/tests/test_sleep.rs
+++ b/tokio/src/time/tests/test_sleep.rs
@@ -1,13 +1,6 @@
-use crate::park::{Park, Unpark};
-use crate::time::driver::{Driver, Entry, Handle};
-use crate::time::Clock;
-use crate::time::{Duration, Instant};
-
-use tokio_test::task;
-use tokio_test::{assert_ok, assert_pending, assert_ready_ok};
-
-use std::sync::Arc;
+//use crate::time::driver::{Driver, Entry, Handle};
 
+/*
 macro_rules! poll {
     ($e:expr) => {
         $e.enter(|cx, e| e.poll_elapsed(cx))
@@ -447,3 +440,4 @@ impl Unpark for MockUnpark {
 fn ms(n: u64) -> Duration {
     Duration::from_millis(n)
 }
+*/
diff --git a/tokio/src/time/timeout.rs b/tokio/src/time/timeout.rs
index cf09b0711..9d15a7205 100644
--- a/tokio/src/time/timeout.rs
+++ b/tokio/src/time/timeout.rs
@@ -49,7 +49,7 @@ pub fn timeout<T>(duration: Duration, future: T) -> Timeout<T>
 where
     T: Future,
 {
-    let delay = Sleep::new_timeout(Instant::now() + duration, duration);
+    let delay = Sleep::new_timeout(Instant::now() + duration);
     Timeout::new_with_delay(future, delay)
 }
 
diff --git a/tokio/src/util/mod.rs b/tokio/src/util/mod.rs
index b2043dd61..03c2f6bc2 100644
--- a/tokio/src/util/mod.rs
+++ b/tokio/src/util/mod.rs
@@ -10,6 +10,7 @@ cfg_io_driver! {
     feature = "rt",
     feature = "sync",
     feature = "signal",
+    feature = "time",
 ))]
 pub(crate) mod linked_list;
 
diff --git a/tokio/tests/macros_select.rs b/tokio/tests/macros_select.rs
index cc214bbba..3359849db 100644
--- a/tokio/tests/macros_select.rs
+++ b/tokio/tests/macros_select.rs
@@ -359,12 +359,14 @@ async fn join_with_select() {
 async fn use_future_in_if_condition() {
     use tokio::time::{self, Duration};
 
-    let mut sleep = time::sleep(Duration::from_millis(50));
+    let sleep = time::sleep(Duration::from_millis(50));
+    tokio::pin!(sleep);
 
     tokio::select! {
-        _ = &mut sleep, if !sleep.is_elapsed() => {
+        _ = time::sleep(Duration::from_millis(50)), if false => {
+            panic!("if condition ignored")
         }
-        _ = async { 1 } => {
+        _ = async { 1u32 } => {
         }
     }
 }
diff --git a/tokio/tests/stream_timeout.rs b/tokio/tests/stream_timeout.rs
index a787bba39..216b5f75d 100644
--- a/tokio/tests/stream_timeout.rs
+++ b/tokio/tests/stream_timeout.rs
@@ -78,7 +78,7 @@ async fn return_elapsed_errors_only_once() {
     // error is returned.
     assert_pending!(stream.poll_next());
     //
-    time::advance(ms(50)).await;
+    time::advance(ms(51)).await;
     let v = assert_ready!(stream.poll_next());
     assert!(v.unwrap().is_err()); // timeout!
 
diff --git a/tokio/tests/sync_mutex.rs b/tokio/tests/sync_mutex.rs
index 96194b31d..0ddb203d8 100644
--- a/tokio/tests/sync_mutex.rs
+++ b/tokio/tests/sync_mutex.rs
@@ -91,10 +91,11 @@ async fn aborted_future_1() {
         let m2 = m1.clone();
         // Try to lock mutex in a future that is aborted prematurely
         timeout(Duration::from_millis(1u64), async move {
-            let mut iv = interval(Duration::from_millis(1000));
+            let iv = interval(Duration::from_millis(1000));
+            tokio::pin!(iv);
             m2.lock().await;
-            iv.tick().await;
-            iv.tick().await;
+            iv.as_mut().tick().await;
+            iv.as_mut().tick().await;
         })
         .await
         .unwrap_err();
diff --git a/tokio/tests/sync_mutex_owned.rs b/tokio/tests/sync_mutex_owned.rs
index 394a6708b..0f1399c43 100644
--- a/tokio/tests/sync_mutex_owned.rs
+++ b/tokio/tests/sync_mutex_owned.rs
@@ -58,10 +58,11 @@ async fn aborted_future_1() {
         let m2 = m1.clone();
         // Try to lock mutex in a future that is aborted prematurely
         timeout(Duration::from_millis(1u64), async move {
-            let mut iv = interval(Duration::from_millis(1000));
+            let iv = interval(Duration::from_millis(1000));
+            tokio::pin!(iv);
             m2.lock_owned().await;
-            iv.tick().await;
-            iv.tick().await;
+            iv.as_mut().tick().await;
+            iv.as_mut().tick().await;
         })
         .await
         .unwrap_err();
diff --git a/tokio/tests/time_interval.rs b/tokio/tests/time_interval.rs
index 5ac6ae69d..a07871576 100644
--- a/tokio/tests/time_interval.rs
+++ b/tokio/tests/time_interval.rs
@@ -49,7 +49,8 @@ async fn usage_stream() {
     use tokio::stream::StreamExt;
 
     let start = Instant::now();
-    let mut interval = time::interval(ms(10));
+    let interval = time::interval(ms(10));
+    tokio::pin!(interval);
 
     for _ in 0..3 {
         interval.next().await.unwrap();
diff --git a/tokio/tests/time_rt.rs b/tokio/tests/time_rt.rs
index 85db78db8..077534352 100644
--- a/tokio/tests/time_rt.rs
+++ b/tokio/tests/time_rt.rs
@@ -68,7 +68,7 @@ async fn starving() {
     }
 
     let when = Instant::now() + Duration::from_millis(20);
-    let starve = Starve(sleep_until(when), 0);
+    let starve = Starve(Box::pin(sleep_until(when)), 0);
 
     starve.await;
     assert!(Instant::now() >= when);
diff --git a/tokio/tests/time_sleep.rs b/tokio/tests/time_sleep.rs
index 955d833bd..d110ec27a 100644
--- a/tokio/tests/time_sleep.rs
+++ b/tokio/tests/time_sleep.rs
@@ -1,6 +1,11 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
+use std::future::Future;
+use std::task::Context;
+
+use futures::task::noop_waker_ref;
+
 use tokio::time::{self, Duration, Instant};
 use tokio_test::{assert_pending, assert_ready, task};
 
@@ -30,6 +35,25 @@ async fn immediate_sleep() {
     assert_elapsed!(now, 0);
 }
 
+#[tokio::test]
+async fn is_elapsed() {
+    time::pause();
+
+    let sleep = time::sleep(Duration::from_millis(50));
+
+    tokio::pin!(sleep);
+
+    assert!(!sleep.is_elapsed());
+
+    assert!(futures::poll!(sleep.as_mut()).is_pending());
+
+    assert!(!sleep.is_elapsed());
+
+    sleep.as_mut().await;
+
+    assert!(sleep.is_elapsed());
+}
+
 #[tokio::test]
 async fn delayed_sleep_level_0() {
     time::pause();
@@ -75,12 +99,12 @@ async fn reset_future_sleep_before_fire() {
 
     let now = Instant::now();
 
-    let mut sleep = task::spawn(time::sleep_until(now + ms(100)));
+    let mut sleep = task::spawn(Box::pin(time::sleep_until(now + ms(100))));
     assert_pending!(sleep.poll());
 
     let mut sleep = sleep.into_inner();
 
-    sleep.reset(Instant::now() + ms(200));
+    sleep.as_mut().reset(Instant::now() + ms(200));
     sleep.await;
 
     assert_elapsed!(now, 200);
@@ -92,12 +116,12 @@ async fn reset_past_sleep_before_turn() {
 
     let now = Instant::now();
 
-    let mut sleep = task::spawn(time::sleep_until(now + ms(100)));
+    let mut sleep = task::spawn(Box::pin(time::sleep_until(now + ms(100))));
     assert_pending!(sleep.poll());
 
     let mut sleep = sleep.into_inner();
 
-    sleep.reset(now + ms(80));
+    sleep.as_mut().reset(now + ms(80));
     sleep.await;
 
     assert_elapsed!(now, 80);
@@ -109,14 +133,14 @@ async fn reset_past_sleep_before_fire() {
 
     let now = Instant::now();
 
-    let mut sleep = task::spawn(time::sleep_until(now + ms(100)));
+    let mut sleep = task::spawn(Box::pin(time::sleep_until(now + ms(100))));
     assert_pending!(sleep.poll());
 
     let mut sleep = sleep.into_inner();
 
     time::sleep(ms(10)).await;
 
-    sleep.reset(now + ms(80));
+    sleep.as_mut().reset(now + ms(80));
     sleep.await;
 
     assert_elapsed!(now, 80);
@@ -127,12 +151,12 @@ async fn reset_future_sleep_after_fire() {
     time::pause();
 
     let now = Instant::now();
-    let mut sleep = time::sleep_until(now + ms(100));
+    let mut sleep = Box::pin(time::sleep_until(now + ms(100)));
 
-    (&mut sleep).await;
+    sleep.as_mut().await;
     assert_elapsed!(now, 100);
 
-    sleep.reset(now + ms(110));
+    sleep.as_mut().reset(now + ms(110));
     sleep.await;
     assert_elapsed!(now, 110);
 }
@@ -143,16 +167,17 @@ async fn reset_sleep_to_past() {
 
     let now = Instant::now();
 
-    let mut sleep = task::spawn(time::sleep_until(now + ms(100)));
+    let mut sleep = task::spawn(Box::pin(time::sleep_until(now + ms(100))));
     assert_pending!(sleep.poll());
 
     time::sleep(ms(50)).await;
 
     assert!(!sleep.is_woken());
 
-    sleep.reset(now + ms(40));
+    sleep.as_mut().reset(now + ms(40));
 
-    assert!(sleep.is_woken());
+    // TODO: is this required?
+    //assert!(sleep.is_woken());
 
     assert_ready!(sleep.poll());
 }
@@ -167,22 +192,110 @@ fn creating_sleep_outside_of_context() {
     let _fut = time::sleep_until(now + ms(500));
 }
 
-#[should_panic]
 #[tokio::test]
 async fn greater_than_max() {
     const YR_5: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
 
+    time::pause();
     time::sleep_until(Instant::now() + ms(YR_5)).await;
 }
 
+#[tokio::test]
+async fn short_sleeps() {
+    for i in 0..10000 {
+        if (i % 10) == 0 {
+            eprintln!("=== {}", i);
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(0)).await;
+    }
+}
+
+#[tokio::test]
+async fn multi_long_sleeps() {
+    tokio::time::pause();
+
+    for _ in 0..5u32 {
+        tokio::time::sleep(Duration::from_secs(
+            // about a year
+            365 * 24 * 3600,
+        ))
+        .await;
+    }
+
+    let deadline = tokio::time::Instant::now()
+        + Duration::from_secs(
+            // about 10 years
+            10 * 365 * 24 * 3600,
+        );
+
+    tokio::time::sleep_until(deadline).await;
+
+    assert!(tokio::time::Instant::now() >= deadline);
+}
+
+#[tokio::test]
+async fn long_sleeps() {
+    tokio::time::pause();
+
+    let deadline = tokio::time::Instant::now()
+        + Duration::from_secs(
+            // about 10 years
+            10 * 365 * 24 * 3600,
+        );
+
+    tokio::time::sleep_until(deadline).await;
+
+    assert!(tokio::time::Instant::now() >= deadline);
+    assert!(tokio::time::Instant::now() <= deadline + Duration::from_millis(1));
+}
+
+#[tokio::test]
+#[should_panic(expected = "Duration too far into the future")]
+async fn very_long_sleeps() {
+    tokio::time::pause();
+
+    // Some platforms (eg macos) can't represent times this far in the future
+    if let Some(deadline) = tokio::time::Instant::now().checked_add(Duration::from_secs(1u64 << 62))
+    {
+        tokio::time::sleep_until(deadline).await;
+    } else {
+        // make it pass anyway (we can't skip/ignore the test based on the
+        // result of checked_add)
+        panic!("Duration too far into the future (test ignored)")
+    }
+}
+
+#[tokio::test]
+async fn reset_after_firing() {
+    let timer = tokio::time::sleep(std::time::Duration::from_millis(1));
+    tokio::pin!(timer);
+
+    let deadline = timer.deadline();
+
+    timer.as_mut().await;
+    assert_ready!(timer
+        .as_mut()
+        .poll(&mut Context::from_waker(noop_waker_ref())));
+    timer
+        .as_mut()
+        .reset(tokio::time::Instant::now() + std::time::Duration::from_secs(600));
+
+    assert_ne!(deadline, timer.deadline());
+
+    assert_pending!(timer
+        .as_mut()
+        .poll(&mut Context::from_waker(noop_waker_ref())));
+    assert_pending!(timer
+        .as_mut()
+        .poll(&mut Context::from_waker(noop_waker_ref())));
+}
+
 const NUM_LEVELS: usize = 6;
 const MAX_DURATION: u64 = (1 << (6 * NUM_LEVELS)) - 1;
 
-#[should_panic]
 #[tokio::test]
 async fn exactly_max() {
-    // TODO: this should not panic but `time::ms()` is acting up
-    // If fixed, make sure to update documentation on `time::sleep` too.
+    time::pause();
     time::sleep(ms(MAX_DURATION)).await;
 }