tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-18 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

time: use intrusive lists for timer tracking (#3080)

Browse Source 
More-or-less a half-rewrite of the current time driver, supporting the
use of intrusive futures for timer registration.

Fixes: #3028, #3069
This commit is contained in:
bdonlan
2020-11-23 10:42:50 -08:00
committed by GitHub
parent f927f01a34
commit ae67851f11
28 changed files with 1558 additions and 919 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+1
View File
@@ -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
tokio-util/src/time/delay_queue.rs
+21 -1
View File
@@ -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);
}
}
tokio-util/tests/time_delay_queue.rs
+15 -9
View File
@@ -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;
tokio/src/loom/std/mod.rs
+1 -1
View File
@@ -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
tokio/src/stream/throttle.rs
+1 -1
View File
@@ -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 {
tokio/src/stream/timeout.rs
+1 -1
View File
@@ -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),
tokio/src/time/driver/atomic_stack.rs
-124
View File
@@ -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());
}
}
}
tokio/src/time/driver/entry.rs
+634 -312
View File
@@ -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);
tokio/src/time/driver/handle.rs
+24 -17
View File
@@ -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 \
tokio/src/time/driver/mod.rs
+287 -253
View File
@@ -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;
tokio/src/time/sleep.rs → tokio/src/time/driver/sleep.rs
+18 -26
View File
@@ -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);
}
}
tokio/src/time/driver/tests/mod.rs
+239 -7
View File
@@ -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);
})
}
*/
tokio/src/time/wheel/level.rs → tokio/src/time/driver/wheel/level.rs
+43 -25
View File
@@ -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])
}
}
tokio/src/time/wheel/mod.rs → tokio/src/time/driver/wheel/mod.rs
+105 -50
View File
@@ -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
}
tokio/src/time/wheel/stack.rs → tokio/src/time/driver/wheel/stack.rs
View File
tokio/src/time/error.rs
+9 -17
View File
@@ -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 {}
tokio/src/time/mod.rs
+7 -36
View File
@@ -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))
}
tokio/src/time/tests/mod.rs
+1 -1
View File
@@ -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>();
tokio/src/time/tests/test_sleep.rs
+3 -9
View File
@@ -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)
}
*/
tokio/src/time/timeout.rs
+1 -1
View File
@@ -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)
}
tokio/src/util/mod.rs
+1
View File
@@ -10,6 +10,7 @@ cfg_io_driver! {
feature = "rt",
feature = "sync",
feature = "signal",
feature = "time",
))]
pub(crate) mod linked_list;
tokio/tests/macros_select.rs
+5 -3
View File
@@ -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 } => {
}
}
}
tokio/tests/stream_timeout.rs
+1 -1
View File
@@ -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!
tokio/tests/sync_mutex.rs
+4 -3
View File
@@ -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();
tokio/tests/sync_mutex_owned.rs
+4 -3
View File
@@ -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();
tokio/tests/time_interval.rs
+2 -1
View File
@@ -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();
tokio/tests/time_rt.rs
+1 -1
View File
@@ -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);
tokio/tests/time_sleep.rs
+129 -16
View File
@@ -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;
}