timer: move tokio-timer into tokio crate (#1674)

A step towards collapsing Tokio sub crates into a single `tokio`
crate (#1318).

The `timer` implementation is now provided by the main `tokio` crate.
The `timer` functionality may still be excluded from the build by
skipping the `timer` feature flag.
This commit is contained in:
Carl Lerche
2019-10-21 16:45:13 -07:00
committed by GitHub
parent c9bcbe77b9
commit b8cee1a60a
53 changed files with 216 additions and 771 deletions
+5 -2
View File
@@ -59,7 +59,7 @@ rt-full = [
signal = ["tokio-net/signal"]
sync = ["tokio-sync"]
tcp = ["io", "tokio-net/tcp"]
timer = ["tokio-timer"]
timer = ["crossbeam-utils", "slab"]
tracing = ["tracing-core"]
udp = ["io", "tokio-net/udp"]
uds = ["io", "tokio-net/uds"]
@@ -72,14 +72,16 @@ futures-util-preview = { version = "=0.3.0-alpha.19", features = ["sink"] }
# Everything else is optional...
bytes = { version = "0.4", optional = true }
crossbeam-utils = { version = "0.6.0", optional = true }
num_cpus = { version = "1.8.0", optional = true }
# Backs `DelayQueue`
slab = { version = "0.4.1", optional = true }
tokio-codec = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-codec" }
tokio-io = { version = "=0.2.0-alpha.6", optional = true, features = ["util"], path = "../tokio-io" }
tokio-executor = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-executor" }
tokio-macros = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-macros" }
tokio-net = { version = "=0.2.0-alpha.6", optional = true, features = ["async-traits"], path = "../tokio-net" }
tokio-sync = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-sync", features = ["async-traits"] }
tokio-timer = { version = "=0.3.0-alpha.6", optional = true, path = "../tokio-timer", features = ["async-traits"] }
tracing-core = { version = "0.1", optional = true }
[target.'cfg(feature = "tracing")'.dependencies]
@@ -98,6 +100,7 @@ http = "0.1"
httparse = "1.0"
libc = "0.2"
num_cpus = "1.0"
rand = "0.7.2"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
tempfile = "3.1.0"
+3 -4
View File
@@ -2,14 +2,13 @@
//!
//! This module provides the [`now`][n] function, which returns an `Instant`
//! representing "now". The source of time used by this function is configurable
//! (via the [`tokio-timer`] crate) and allows mocking out the source of time in
//! tests or performing caching operations to reduce the number of syscalls.
//! and allows mocking out the source of time in tests or performing caching
//! operations to reduce the number of syscalls.
//!
//! Note that, because the source of time is configurable, it is possible to
//! observe non-monotonic behavior when calling [`now`][n] from different
//! executors.
//!
//! [n]: fn.now.html
//! [`tokio-timer`]: https://docs.rs/tokio-timer/0.2/tokio_timer/clock/index.html
pub use tokio_timer::clock::now;
pub use crate::timer::clock::now;
+1 -1
View File
@@ -1,7 +1,7 @@
//! Asynchronous values.
#[cfg(feature = "timer")]
use tokio_timer::Timeout;
use crate::timer::Timeout;
#[cfg(feature = "timer")]
use std::time::Duration;
+4 -4
View File
@@ -1,9 +1,9 @@
use crate::runtime::current_thread::Runtime;
use crate::timer::clock::Clock;
use crate::timer::timer::Timer;
use tokio_executor::current_thread::CurrentThread;
use tokio_net::driver::Reactor;
use tokio_timer::clock::Clock;
use tokio_timer::timer::Timer;
use std::io;
@@ -24,7 +24,7 @@ use std::io;
///
/// ```
/// use tokio::runtime::current_thread::Builder;
/// use tokio_timer::clock::Clock;
/// use tokio::timer::clock::Clock;
///
/// # pub fn main() {
/// // build Runtime
@@ -66,7 +66,7 @@ impl Builder {
// Place a timer wheel on top of the reactor. If there are no timeouts to fire, it'll let the
// reactor pick up some new external events.
let timer = Timer::new_with_now(reactor, self.clock.clone());
let timer = Timer::new_with_clock(reactor, self.clock.clone());
let timer_handle = timer.handle();
// And now put a single-threaded executor on top of the timer. When there are no futures ready
+2 -2
View File
@@ -1,10 +1,10 @@
use crate::runtime::current_thread::Builder;
use crate::timer::clock::{self, Clock};
use crate::timer::timer::{self, Timer};
use tokio_executor::current_thread::Handle as ExecutorHandle;
use tokio_executor::current_thread::{self, CurrentThread};
use tokio_net::driver::{self, Reactor};
use tokio_timer::clock::{self, Clock};
use tokio_timer::timer::{self, Timer};
use std::error::Error;
use std::fmt;
+1 -2
View File
@@ -20,7 +20,7 @@
//!
//! * Spawn a background thread running a [`Reactor`] instance.
//! * Start a [`ThreadPool`] for executing futures.
//! * Run an instance of [`Timer`] **per** thread pool worker thread.
//! * Run an instance of `Timer` **per** thread pool worker thread.
//!
//! The thread pool uses a work-stealing strategy and is configured to start a
//! worker thread for each CPU core available on the system. This tends to be
@@ -127,7 +127,6 @@
//! [`ThreadPool`]: https://docs.rs/tokio-executor/0.2.0-alpha.2/tokio_executor/threadpool/struct.ThreadPool.html
//! [`run`]: fn.run.html
//! [`tokio::spawn`]: ../executor/fn.spawn.html
//! [`Timer`]: https://docs.rs/tokio-timer/0.2/tokio_timer/timer/struct.Timer.html
//! [`tokio::main`]: ../../tokio_macros/attr.main.html
pub mod current_thread;
+4 -4
View File
@@ -1,9 +1,9 @@
use super::{Inner, Runtime};
use crate::timer::clock::{self, Clock};
use crate::timer::timer::{self, Timer};
use tokio_executor::thread_pool;
use tokio_net::driver::{self, Reactor};
use tokio_timer::clock::{self, Clock};
use tokio_timer::timer::{self, Timer};
use tracing_core as trace;
use std::{fmt, io};
@@ -26,7 +26,7 @@ use std::sync::{Arc, Mutex};
///
/// ```
/// use tokio::runtime::Builder;
/// use tokio_timer::clock::Clock;
/// use tokio::timer::clock::Clock;
///
/// fn main() {
/// // build Runtime
@@ -233,7 +233,7 @@ impl Builder {
reactor_handles.push(reactor.handle());
// Create a new timer.
let timer = Timer::new_with_now(reactor, self.clock.clone());
let timer = Timer::new_with_clock(reactor, self.clock.clone());
timer_handles.push(timer.handle());
timers.push(Mutex::new(Some(timer)));
}
+2 -1
View File
@@ -9,9 +9,10 @@ pub use self::spawner::Spawner;
#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
pub use tokio_executor::thread_pool::JoinHandle;
use crate::timer::timer;
use tokio_executor::thread_pool::ThreadPool;
use tokio_net::driver;
use tokio_timer::timer;
use tracing_core as trace;
use std::future::Future;
+1 -1
View File
@@ -4,7 +4,7 @@
use std::time::Duration;
#[cfg(feature = "timer")]
use tokio_timer::{throttle::Throttle, Timeout};
use crate::timer::{throttle::Throttle, Timeout};
#[doc(inline)]
pub use futures_core::Stream;
+60
View File
@@ -0,0 +1,60 @@
//! Implementation of an atomic u64 cell. On 64 bit platforms, this is a
//! re-export of `AtomicU64`. On 32 bit platforms, this is implemented using a
//! `Mutex`.
pub(crate) use self::imp::AtomicU64;
// `AtomicU64` can only be used on targets with `target_has_atomic` is 64 or greater.
// Once `cfg_target_has_atomic` feature is stable, we can replace it with
// `#[cfg(target_has_atomic = "64")]`.
// Refs: https://github.com/rust-lang/rust/tree/master/src/librustc_target
#[cfg(not(any(target_arch = "arm", target_arch = "mips", target_arch = "powerpc")))]
mod imp {
pub(crate) use std::sync::atomic::AtomicU64;
}
#[cfg(any(target_arch = "arm", target_arch = "mips", target_arch = "powerpc"))]
mod imp {
use std::sync::atomic::Ordering;
use std::sync::Mutex;
#[derive(Debug)]
pub(crate) struct AtomicU64 {
inner: Mutex<u64>,
}
impl AtomicU64 {
pub(crate) fn new(val: u64) -> AtomicU64 {
AtomicU64 {
inner: Mutex::new(val),
}
}
pub(crate) fn load(&self, _: Ordering) -> u64 {
*self.inner.lock().unwrap()
}
pub(crate) fn store(&self, val: u64, _: Ordering) {
*self.inner.lock().unwrap() = val;
}
pub(crate) fn fetch_or(&self, val: u64, _: Ordering) -> u64 {
let mut lock = self.inner.lock().unwrap();
let prev = *lock;
*lock = prev | val;
prev
}
pub(crate) fn compare_and_swap(&self, old: u64, new: u64, _: Ordering) -> u64 {
let mut lock = self.inner.lock().unwrap();
let prev = *lock;
if prev != old {
return prev;
}
*lock = new;
prev
}
}
}
+149
View File
@@ -0,0 +1,149 @@
//! A configurable source of time.
//!
//! This module provides an API to get the current instant in such a way that
//! the source of time may be configured. This allows mocking out the source of
//! time in tests.
//!
//! The [`now`][n] function returns the current [`Instant`]. By default, it delegates
//! to [`Instant::now`].
//!
//! The source of time used by [`now`][n] can be configured by implementing the
//! [`Now`] trait and passing an instance to [`with_default`].
//!
//! [n]: fn.now.html
//! [`Now`]: trait.Now.html
//! [`Instant`]: std::time::Instant
//! [`Instant::now`]: std::time::Instant::now
//! [`with_default`]: fn.with_default.html
mod now;
pub use self::now::Now;
use std::cell::Cell;
use std::fmt;
use std::sync::Arc;
use std::time::Instant;
/// A handle to a source of time.
///
/// `Clock` instances return [`Instant`] values corresponding to "now". The source
/// of these values is configurable. The default source is [`Instant::now`].
///
/// [`Instant`]: std::time::Instant
/// [`Instant::now`]: std::time::Instant::now
#[derive(Default, Clone)]
pub struct Clock {
now: Option<Arc<dyn Now>>,
}
thread_local! {
/// Thread-local tracking the current clock
static CLOCK: Cell<Option<*const Clock>> = Cell::new(None)
}
/// Returns an `Instant` corresponding to "now".
///
/// This function delegates to the source of time configured for the current
/// execution context. By default, this is `Instant::now()`.
///
/// Note that, because the source of time is configurable, it is possible to
/// observe non-monotonic behavior when calling `now` from different
/// executors.
///
/// See [module](index.html) level documentation for more details.
///
/// # Examples
///
/// ```
/// # use tokio::timer::clock;
/// let now = clock::now();
/// ```
pub fn now() -> Instant {
CLOCK.with(|current| match current.get() {
Some(ptr) => unsafe { (*ptr).now() },
None => Instant::now(),
})
}
impl Clock {
/// Return a new `Clock` instance that uses the current execution context's
/// source of time.
pub fn new() -> Clock {
CLOCK.with(|current| match current.get() {
Some(ptr) => unsafe { (*ptr).clone() },
None => Clock::system(),
})
}
/// Return a new `Clock` instance that uses `now` as the source of time.
pub fn new_with_now<T: Now>(now: T) -> Clock {
Clock {
now: Some(Arc::new(now)),
}
}
/// Return a new `Clock` instance that uses [`Instant::now`] as the source
/// of time.
///
/// [`Instant::now`]: std::time::Instant::now
pub fn system() -> Clock {
Clock { now: None }
}
/// Returns an instant corresponding to "now" by using the instance's source
/// of time.
pub fn now(&self) -> Instant {
match self.now {
Some(ref now) => now.now(),
None => Instant::now(),
}
}
}
impl fmt::Debug for Clock {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Clock")
.field("now", {
if self.now.is_some() {
&"Some(Arc<Now>)"
} else {
&"None"
}
})
.finish()
}
}
/// Set the default clock for the duration of the closure.
///
/// # Panics
///
/// This function panics if there already is a default clock set.
pub fn with_default<F, R>(clock: &Clock, f: F) -> R
where
F: FnOnce() -> R,
{
CLOCK.with(|cell| {
assert!(
cell.get().is_none(),
"default clock already set for execution context"
);
// Ensure that the clock is removed from the thread-local context
// when leaving the scope. This handles cases that involve panicking.
struct Reset<'a>(&'a Cell<Option<*const Clock>>);
impl Drop for Reset<'_> {
fn drop(&mut self) {
self.0.set(None);
}
}
let _reset = Reset(cell);
cell.set(Some(clock as *const Clock));
f()
})
}
+15
View File
@@ -0,0 +1,15 @@
use std::time::Instant;
/// Returns [`Instant`] values representing the current instant in time.
///
/// This allows customizing the source of time which is especially useful for
/// testing.
///
/// Implementations must ensure that calls to `now` return monotonically
/// increasing [`Instant`] values.
///
/// [`Instant`]: std::time::Instant
pub trait Now: Send + Sync + 'static {
/// Returns an instant corresponding to "now".
fn now(&self) -> Instant;
}
+162
View File
@@ -0,0 +1,162 @@
#![allow(deprecated)]
use crate::Delay;
use futures::{Async, Future, Poll};
use std::error;
use std::fmt;
use std::time::Instant;
#[deprecated(since = "0.2.6", note = "use Timeout instead")]
#[doc(hidden)]
#[derive(Debug)]
pub struct Deadline<T> {
future: T,
delay: Delay,
}
#[deprecated(since = "0.2.6", note = "use Timeout instead")]
#[doc(hidden)]
#[derive(Debug)]
pub struct DeadlineError<T>(Kind<T>);
/// Deadline error variants
#[derive(Debug)]
enum Kind<T> {
/// Inner future returned an error
Inner(T),
/// The deadline elapsed.
Elapsed,
/// Timer returned an error.
Timer(crate::Error),
}
impl<T> Deadline<T> {
/// Create a new `Deadline` that completes when `future` completes or when
/// `deadline` is reached.
pub fn new(future: T, deadline: Instant) -> Deadline<T> {
Deadline::new_with_delay(future, Delay::new(deadline))
}
pub(crate) fn new_with_delay(future: T, delay: Delay) -> Deadline<T> {
Deadline { future, delay }
}
/// Gets a reference to the underlying future in this deadline.
pub fn get_ref(&self) -> &T {
&self.future
}
/// Gets a mutable reference to the underlying future in this deadline.
pub fn get_mut(&mut self) -> &mut T {
&mut self.future
}
/// Consumes this deadline, returning the underlying future.
pub fn into_inner(self) -> T {
self.future
}
}
impl<T> Future for Deadline<T>
where
T: Future,
{
type Item = T::Item;
type Error = DeadlineError<T::Error>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
// First, try polling the future
match self.future.poll() {
Ok(Async::Ready(v)) => return Ok(Async::Ready(v)),
Ok(Async::NotReady) => {}
Err(e) => return Err(DeadlineError::inner(e)),
}
// Now check the timer
match self.delay.poll() {
Ok(Async::NotReady) => Ok(Async::NotReady),
Ok(Async::Ready(_)) => Err(DeadlineError::elapsed()),
Err(e) => Err(DeadlineError::timer(e)),
}
}
}
// ===== impl DeadlineError =====
impl<T> DeadlineError<T> {
/// Create a new `DeadlineError` representing the inner future completing
/// with `Err`.
pub fn inner(err: T) -> DeadlineError<T> {
DeadlineError(Kind::Inner(err))
}
/// Returns `true` if the error was caused by the inner future completing
/// with `Err`.
pub fn is_inner(&self) -> bool {
match self.0 {
Kind::Inner(_) => true,
_ => false,
}
}
/// Consumes `self`, returning the inner future error.
pub fn into_inner(self) -> Option<T> {
match self.0 {
Kind::Inner(err) => Some(err),
_ => None,
}
}
/// Create a new `DeadlineError` representing the inner future not
/// completing before the deadline is reached.
pub fn elapsed() -> DeadlineError<T> {
DeadlineError(Kind::Elapsed)
}
/// Returns `true` if the error was caused by the inner future not
/// completing before the deadline is reached.
pub fn is_elapsed(&self) -> bool {
match self.0 {
Kind::Elapsed => true,
_ => false,
}
}
/// Creates a new `DeadlineError` representing an error encountered by the
/// timer implementation
pub fn timer(err: crate::Error) -> DeadlineError<T> {
DeadlineError(Kind::Timer(err))
}
/// Returns `true` if the error was caused by the timer.
pub fn is_timer(&self) -> bool {
match self.0 {
Kind::Timer(_) => true,
_ => false,
}
}
/// Consumes `self`, returning the error raised by the timer implementation.
pub fn into_timer(self) -> Option<crate::Error> {
match self.0 {
Kind::Timer(err) => Some(err),
_ => None,
}
}
}
impl<T: error::Error> error::Error for DeadlineError<T> {}
impl<T: fmt::Display> fmt::Display for DeadlineError<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
use self::Kind::*;
match self.0 {
Inner(ref e) => e.fmt(fmt),
Elapsed => "deadline has elapsed".fmt(fmt),
Timer(ref e) => e.fmt(fmt),
}
}
}
+114
View File
@@ -0,0 +1,114 @@
use crate::timer::timer::{HandlePriv, Registration};
use futures_core::ready;
use std::future::Future;
use std::pin::Pin;
use std::task::{self, Poll};
use std::time::{Duration, Instant};
/// A future that completes at a specified instant in time.
///
/// Instances of `Delay` perform no work and complete with `()` once the
/// specified deadline has been reached.
///
/// `Delay` has a resolution of one millisecond and should not be used for tasks
/// that require high-resolution timers.
///
/// # Cancellation
///
/// Canceling a `Delay` is done by dropping the value. No additional cleanup or
/// other work is required.
///
/// [`new`]: #method.new
#[derive(Debug)]
pub struct Delay {
/// The link between the `Delay` instance at the timer that drives it.
///
/// This also stores the `deadline` value.
registration: Registration,
}
impl Delay {
/// Create a new `Delay` instance that elapses at `deadline`.
///
/// Only millisecond level resolution is guaranteed. There is no guarantee
/// as to how the sub-millisecond portion of `deadline` will be handled.
/// `Delay` should not be used for high-resolution timer use cases.
pub(crate) fn new(deadline: Instant) -> Delay {
let registration = Registration::new(deadline, Duration::from_millis(0));
Delay { registration }
}
pub(crate) fn new_timeout(deadline: Instant, duration: Duration) -> Delay {
let registration = Registration::new(deadline, duration);
Delay { registration }
}
pub(crate) fn new_with_handle(
deadline: Instant,
duration: Duration,
handle: HandlePriv,
) -> Delay {
let mut registration = Registration::new(deadline, duration);
registration.register_with(handle);
Delay { registration }
}
/// Returns the instant at which the future will complete.
pub fn deadline(&self) -> Instant {
self.registration.deadline()
}
/// Returns true if the `Delay` has elapsed
///
/// A `Delay` is elapsed when the requested duration has elapsed.
pub fn is_elapsed(&self) -> bool {
self.registration.is_elapsed()
}
/// Reset the `Delay` instance to a new deadline.
///
/// Calling this function allows changing the instant at which the `Delay`
/// future completes without having to create new associated state.
///
/// This function can be called both before and after the future has
/// completed.
pub fn reset(&mut self, deadline: Instant) {
self.registration.reset(deadline);
}
pub(crate) fn reset_timeout(&mut self) {
self.registration.reset_timeout();
}
/// Register the delay with the timer instance for the current execution
/// context.
fn register(&mut self) {
self.registration.register();
}
}
impl Future for Delay {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
// Ensure the `Delay` instance is associated with a timer.
self.register();
// `poll_elapsed` can return an error in two cases:
//
// - AtCapacity: this is a pathlogical case where far too many
// delays have been scheduled.
// - Shutdown: No timer has been setup, which is a mis-use error.
//
// 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.registration.poll_elapsed(cx)) {
Ok(()) => Poll::Ready(()),
Err(e) => panic!("timer error: {}", e),
}
}
}
+855
View File
@@ -0,0 +1,855 @@
//! A queue of delayed elements.
//!
//! See [`DelayQueue`] for more details.
//!
//! [`DelayQueue`]: struct.DelayQueue.html
use crate::timer::clock::now;
use crate::timer::timer::Handle;
use crate::timer::wheel::{self, Wheel};
use crate::timer::{Delay, Error};
use futures_core::ready;
use slab::Slab;
use std::cmp;
use std::future::Future;
use std::marker::PhantomData;
use std::pin::Pin;
use std::task::{self, Poll};
use std::time::{Duration, Instant};
/// A queue of delayed elements.
///
/// Once an element is inserted into the `DelayQueue`, it is yielded once the
/// specified deadline has been reached.
///
/// # Usage
///
/// Elements are inserted into `DelayQueue` using the [`insert`] or
/// [`insert_at`] methods. A deadline is provided with the item and a [`Key`] is
/// returned. The key is used to remove the entry or to change the deadline at
/// which it should be yielded back.
///
/// Once delays have been configured, the `DelayQueue` is used via its
/// [`Stream`] implementation. [`poll`] is called. If an entry has reached its
/// deadline, it is returned. If not, `Async::NotReady` indicating that the
/// current task will be notified once the deadline has been reached.
///
/// # `Stream` implementation
///
/// Items are retrieved from the queue via [`Stream::poll`]. If no delays have
/// expired, no items are returned. In this case, `NotReady` is returned and the
/// current task is registered to be notified once the next item's delay has
/// expired.
///
/// If no items are in the queue, i.e. `is_empty()` returns `true`, then `poll`
/// returns `Ready(None)`. This indicates that the stream has reached an end.
/// However, if a new item is inserted *after*, `poll` will once again start
/// returning items or `NotReady.
///
/// Items are returned ordered by their expirations. Items that are configured
/// to expire first will be returned first. There are no ordering guarantees
/// for items configured to expire the same instant. Also note that delays are
/// rounded to the closest millisecond.
///
/// # Implementation
///
/// The `DelayQueue` is backed by the same hashed timing wheel implementation as
/// [`Timer`] as such, it offers the same performance benefits. See [`Timer`]
/// for further implementation notes.
///
/// State associated with each entry is stored in a [`slab`]. This allows
/// amortizing the cost of allocation. Space created for expired entries is
/// reused when inserting new entries.
///
/// Capacity can be checked using [`capacity`] and allocated preemptively by using
/// the [`reserve`] method.
///
/// # Usage
///
/// Using `DelayQueue` to manage cache entries.
///
/// ```rust,no_run
/// use tokio::timer::{delay_queue, DelayQueue, Error};
///
/// use futures_core::ready;
/// use std::collections::HashMap;
/// use std::task::{Context, Poll};
/// use std::time::Duration;
/// # type CacheKey = String;
/// # type Value = String;
///
/// struct Cache {
/// entries: HashMap<CacheKey, (Value, delay_queue::Key)>,
/// expirations: DelayQueue<CacheKey>,
/// }
///
/// const TTL_SECS: u64 = 30;
///
/// impl Cache {
/// fn insert(&mut self, key: CacheKey, value: Value) {
/// let delay = self.expirations
/// .insert(key.clone(), Duration::from_secs(TTL_SECS));
///
/// self.entries.insert(key, (value, delay));
/// }
///
/// fn get(&self, key: &CacheKey) -> Option<&Value> {
/// self.entries.get(key)
/// .map(|&(ref v, _)| v)
/// }
///
/// fn remove(&mut self, key: &CacheKey) {
/// if let Some((_, cache_key)) = self.entries.remove(key) {
/// self.expirations.remove(&cache_key);
/// }
/// }
///
/// fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
/// while let Some(res) = ready!(self.expirations.poll_next(cx)) {
/// let entry = res?;
/// self.entries.remove(entry.get_ref());
/// }
///
/// Poll::Ready(Ok(()))
/// }
/// }
/// ```
///
/// [`insert`]: #method.insert
/// [`insert_at`]: #method.insert_at
/// [`Key`]: struct.Key.html
/// [`Stream`]: https://docs.rs/futures/0.1/futures/stream/trait.Stream.html
/// [`poll`]: #method.poll
/// [`Stream::poll`]: #method.poll
/// [`Timer`]: ../struct.Timer.html
/// [`slab`]: https://docs.rs/slab
/// [`capacity`]: #method.capacity
/// [`reserve`]: #method.reserve
#[derive(Debug)]
pub struct DelayQueue<T> {
/// Handle to the timer driving the `DelayQueue`
handle: Handle,
/// Stores data associated with entries
slab: Slab<Data<T>>,
/// Lookup structure tracking all delays in the queue
wheel: Wheel<Stack<T>>,
/// Delays that were inserted when already expired. These cannot be stored
/// in the wheel
expired: Stack<T>,
/// Delay expiring when the *first* item in the queue expires
delay: Option<Delay>,
/// Wheel polling state
poll: wheel::Poll,
/// Instant at which the timer starts
start: Instant,
}
/// An entry in `DelayQueue` that has expired and removed.
///
/// Values are returned by [`DelayQueue::poll`].
///
/// [`DelayQueue::poll`]: struct.DelayQueue.html#method.poll
#[derive(Debug)]
pub struct Expired<T> {
/// The data stored in the queue
data: T,
/// The expiration time
deadline: Instant,
/// The key associated with the entry
key: Key,
}
/// Token to a value stored in a `DelayQueue`.
///
/// Instances of `Key` are returned by [`DelayQueue::insert`]. See [`DelayQueue`]
/// documentation for more details.
///
/// [`DelayQueue`]: struct.DelayQueue.html
/// [`DelayQueue::insert`]: struct.DelayQueue.html#method.insert
#[derive(Debug, Clone)]
pub struct Key {
index: usize,
}
#[derive(Debug)]
struct Stack<T> {
/// Head of the stack
head: Option<usize>,
_p: PhantomData<fn() -> T>,
}
#[derive(Debug)]
struct Data<T> {
/// The data being stored in the queue and will be returned at the requested
/// instant.
inner: T,
/// The instant at which the item is returned.
when: u64,
/// Set to true when stored in the `expired` queue
expired: bool,
/// Next entry in the stack
next: Option<usize>,
/// Previous entry in the stack
prev: Option<usize>,
}
/// Maximum number of entries the queue can handle
const MAX_ENTRIES: usize = (1 << 30) - 1;
impl<T> DelayQueue<T> {
/// Create a new, empty, `DelayQueue`
///
/// The queue will not allocate storage until items are inserted into it.
///
/// # Examples
///
/// ```rust
/// # use tokio::timer::DelayQueue;
/// let delay_queue: DelayQueue<u32> = DelayQueue::new();
/// ```
pub fn new() -> DelayQueue<T> {
DelayQueue::with_capacity(0)
}
/// Create a new, empty, `DelayQueue` backed by the specified timer.
///
/// The queue will not allocate storage until items are inserted into it.
///
/// # Examples
///
/// ```rust,no_run
/// # use tokio::timer::DelayQueue;
/// use tokio::timer::timer::Handle;
///
/// let handle = Handle::default();
/// let delay_queue: DelayQueue<u32> = DelayQueue::with_capacity_and_handle(0, &handle);
/// ```
pub fn with_capacity_and_handle(capacity: usize, handle: &Handle) -> DelayQueue<T> {
DelayQueue {
handle: handle.clone(),
wheel: Wheel::new(),
slab: Slab::with_capacity(capacity),
expired: Stack::default(),
delay: None,
poll: wheel::Poll::new(0),
start: now(),
}
}
/// Create a new, empty, `DelayQueue` with the specified capacity.
///
/// The queue will be able to hold at least `capacity` elements without
/// reallocating. If `capacity` is 0, the queue will not allocate for
/// storage.
///
/// # Examples
///
/// ```rust
/// # use tokio::timer::DelayQueue;
/// # use std::time::Duration;
/// let mut delay_queue = DelayQueue::with_capacity(10);
///
/// // These insertions are done without further allocation
/// for i in 0..10 {
/// delay_queue.insert(i, Duration::from_secs(i));
/// }
///
/// // This will make the queue allocate additional storage
/// delay_queue.insert(11, Duration::from_secs(11));
/// ```
pub fn with_capacity(capacity: usize) -> DelayQueue<T> {
DelayQueue::with_capacity_and_handle(capacity, &Handle::default())
}
/// Insert `value` into the queue set to expire at a specific instant in
/// time.
///
/// This function is identical to `insert`, but takes an `Instant` instead
/// of a `Duration`.
///
/// `value` is stored in the queue until `when` is reached. At which point,
/// `value` will be returned from [`poll`]. If `when` has already been
/// reached, then `value` is immediately made available to poll.
///
/// The return value represents the insertion and is used at an argument to
/// [`remove`] and [`reset`]. Note that [`Key`] is token and is reused once
/// `value` is removed from the queue either by calling [`poll`] after
/// `when` is reached or by calling [`remove`]. At this point, the caller
/// must take care to not use the returned [`Key`] again as it may reference
/// a different item in the queue.
///
/// See [type] level documentation for more details.
///
/// # Panics
///
/// This function panics if `when` is too far in the future.
///
/// # Examples
///
/// Basic usage
///
/// ```rust
/// use tokio::timer::DelayQueue;
/// use std::time::{Instant, Duration};
///
/// let mut delay_queue = DelayQueue::new();
/// let key = delay_queue.insert_at(
/// "foo", Instant::now() + Duration::from_secs(5));
///
/// // Remove the entry
/// let item = delay_queue.remove(&key);
/// assert_eq!(*item.get_ref(), "foo");
/// ```
///
/// [`poll`]: #method.poll
/// [`remove`]: #method.remove
/// [`reset`]: #method.reset
/// [`Key`]: struct.Key.html
/// [type]: #
pub fn insert_at(&mut self, value: T, when: Instant) -> Key {
assert!(self.slab.len() < MAX_ENTRIES, "max entries exceeded");
// Normalize the deadline. Values cannot be set to expire in the past.
let when = self.normalize_deadline(when);
// Insert the value in the store
let key = self.slab.insert(Data {
inner: value,
when,
expired: false,
next: None,
prev: None,
});
self.insert_idx(when, key);
// Set a new delay if the current's deadline is later than the one of the new item
let should_set_delay = if let Some(ref delay) = self.delay {
let current_exp = self.normalize_deadline(delay.deadline());
current_exp > when
} else {
true
};
if should_set_delay {
self.delay = Some(self.handle.delay(self.start + Duration::from_millis(when)));
}
Key::new(key)
}
/// Attempt to pull out the next value of the delay queue, registering the
/// current task for wakeup if the value is not yet available, and returning
/// None if the queue is exhausted.
pub fn poll_next(
&mut self,
cx: &mut task::Context<'_>,
) -> Poll<Option<Result<Expired<T>, Error>>> {
let item = ready!(self.poll_idx(cx));
Poll::Ready(item.map(|result| {
result.map(|idx| {
let data = self.slab.remove(idx);
debug_assert!(data.next.is_none());
debug_assert!(data.prev.is_none());
Expired {
key: Key::new(idx),
data: data.inner,
deadline: self.start + Duration::from_millis(data.when),
}
})
}))
}
/// Insert `value` into the queue set to expire after the requested duration
/// elapses.
///
/// This function is identical to `insert_at`, but takes a `Duration`
/// instead of an `Instant`.
///
/// `value` is stored in the queue until `when` is reached. At which point,
/// `value` will be returned from [`poll`]. If `when` has already been
/// reached, then `value` is immediately made available to poll.
///
/// The return value represents the insertion and is used at an argument to
/// [`remove`] and [`reset`]. Note that [`Key`] is token and is reused once
/// `value` is removed from the queue either by calling [`poll`] after
/// `when` is reached or by calling [`remove`]. At this point, the caller
/// must take care to not use the returned [`Key`] again as it may reference
/// a different item in the queue.
///
/// See [type] level documentation for more details.
///
/// # Panics
///
/// This function panics if `timeout` is greater than the maximum supported
/// duration.
///
/// # Examples
///
/// Basic usage
///
/// ```rust
/// use tokio::timer::DelayQueue;
/// use std::time::Duration;
///
/// let mut delay_queue = DelayQueue::new();
/// let key = delay_queue.insert("foo", Duration::from_secs(5));
///
/// // Remove the entry
/// let item = delay_queue.remove(&key);
/// assert_eq!(*item.get_ref(), "foo");
/// ```
///
/// [`poll`]: #method.poll
/// [`remove`]: #method.remove
/// [`reset`]: #method.reset
/// [`Key`]: struct.Key.html
/// [type]: #
pub fn insert(&mut self, value: T, timeout: Duration) -> Key {
self.insert_at(value, now() + timeout)
}
fn insert_idx(&mut self, when: u64, key: usize) {
use self::wheel::{InsertError, Stack};
// Register the deadline with the timer wheel
match self.wheel.insert(when, key, &mut self.slab) {
Ok(_) => {}
Err((_, InsertError::Elapsed)) => {
self.slab[key].expired = true;
// The delay is already expired, store it in the expired queue
self.expired.push(key, &mut self.slab);
}
Err((_, err)) => panic!("invalid deadline; err={:?}", err),
}
}
/// Remove the item associated with `key` from the queue.
///
/// There must be an item associated with `key`. The function returns the
/// removed item as well as the `Instant` at which it will the delay will
/// have expired.
///
/// # Panics
///
/// The function panics if `key` is not contained by the queue.
///
/// # Examples
///
/// Basic usage
///
/// ```rust
/// use tokio::timer::DelayQueue;
/// use std::time::Duration;
///
/// let mut delay_queue = DelayQueue::new();
/// let key = delay_queue.insert("foo", Duration::from_secs(5));
///
/// // Remove the entry
/// let item = delay_queue.remove(&key);
/// assert_eq!(*item.get_ref(), "foo");
/// ```
pub fn remove(&mut self, key: &Key) -> Expired<T> {
use crate::timer::wheel::Stack;
// Special case the `expired` queue
if self.slab[key.index].expired {
self.expired.remove(&key.index, &mut self.slab);
} else {
self.wheel.remove(&key.index, &mut self.slab);
}
let data = self.slab.remove(key.index);
Expired {
key: Key::new(key.index),
data: data.inner,
deadline: self.start + Duration::from_millis(data.when),
}
}
/// Sets the delay of the item associated with `key` to expire at `when`.
///
/// This function is identical to `reset` but takes an `Instant` instead of
/// a `Duration`.
///
/// The item remains in the queue but the delay is set to expire at `when`.
/// If `when` is in the past, then the item is immediately made available to
/// the caller.
///
/// # Panics
///
/// This function panics if `when` is too far in the future or if `key` is
/// not contained by the queue.
///
/// # Examples
///
/// Basic usage
///
/// ```rust
/// use tokio::timer::DelayQueue;
/// use std::time::{Duration, Instant};
///
/// let mut delay_queue = DelayQueue::new();
/// let key = delay_queue.insert("foo", Duration::from_secs(5));
///
/// // "foo" is scheduled to be returned in 5 seconds
///
/// delay_queue.reset_at(&key, Instant::now() + Duration::from_secs(10));
///
/// // "foo"is now scheduled to be returned in 10 seconds
/// ```
pub fn reset_at(&mut self, key: &Key, when: Instant) {
self.wheel.remove(&key.index, &mut self.slab);
// Normalize the deadline. Values cannot be set to expire in the past.
let when = self.normalize_deadline(when);
self.slab[key.index].when = when;
self.insert_idx(when, key.index);
let next_deadline = self.next_deadline();
if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) {
delay.reset(deadline);
}
}
/// Returns the next time poll as determined by the wheel
fn next_deadline(&mut self) -> Option<Instant> {
self.wheel
.poll_at()
.map(|poll_at| self.start + Duration::from_millis(poll_at))
}
/// Sets the delay of the item associated with `key` to expire after
/// `timeout`.
///
/// This function is identical to `reset_at` but takes a `Duration` instead
/// of an `Instant`.
///
/// The item remains in the queue but the delay is set to expire after
/// `timeout`. If `timeout` is zero, then the item is immediately made
/// available to the caller.
///
/// # Panics
///
/// This function panics if `timeout` is greater than the maximum supported
/// duration or if `key` is not contained by the queue.
///
/// # Examples
///
/// Basic usage
///
/// ```rust
/// use tokio::timer::DelayQueue;
/// use std::time::Duration;
///
/// let mut delay_queue = DelayQueue::new();
/// let key = delay_queue.insert("foo", Duration::from_secs(5));
///
/// // "foo" is scheduled to be returned in 5 seconds
///
/// delay_queue.reset(&key, Duration::from_secs(10));
///
/// // "foo"is now scheduled to be returned in 10 seconds
/// ```
pub fn reset(&mut self, key: &Key, timeout: Duration) {
self.reset_at(key, now() + timeout);
}
/// Clears the queue, removing all items.
///
/// After calling `clear`, [`poll`] will return `Ok(Ready(None))`.
///
/// Note that this method has no effect on the allocated capacity.
///
/// [`poll`]: #method.poll
///
/// # Examples
///
/// ```rust
/// use tokio::timer::DelayQueue;
/// use std::time::Duration;
///
/// let mut delay_queue = DelayQueue::new();
///
/// delay_queue.insert("foo", Duration::from_secs(5));
///
/// assert!(!delay_queue.is_empty());
///
/// delay_queue.clear();
///
/// assert!(delay_queue.is_empty());
/// ```
pub fn clear(&mut self) {
self.slab.clear();
self.expired = Stack::default();
self.wheel = Wheel::new();
self.delay = None;
}
/// Returns the number of elements the queue can hold without reallocating.
///
/// # Examples
///
/// ```rust
/// use tokio::timer::DelayQueue;
///
/// let delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
/// assert_eq!(delay_queue.capacity(), 10);
/// ```
pub fn capacity(&self) -> usize {
self.slab.capacity()
}
/// Reserve capacity for at least `additional` more items to be queued
/// without allocating.
///
/// `reserve` does nothing if the queue already has sufficient capacity for
/// `additional` more values. If more capacity is required, a new segment of
/// memory will be allocated and all existing values will be copied into it.
/// As such, if the queue is already very large, a call to `reserve` can end
/// up being expensive.
///
/// The queue may reserve more than `additional` extra space in order to
/// avoid frequent reallocations.
///
/// # Panics
///
/// Panics if the new capacity exceeds the maximum number of entries the
/// queue can contain.
///
/// # Examples
///
/// ```
/// use tokio::timer::DelayQueue;
/// use std::time::Duration;
///
/// let mut delay_queue = DelayQueue::new();
///
/// delay_queue.insert("hello", Duration::from_secs(10));
/// delay_queue.reserve(10);
///
/// assert!(delay_queue.capacity() >= 11);
/// ```
pub fn reserve(&mut self, additional: usize) {
self.slab.reserve(additional);
}
/// Returns `true` if there are no items in the queue.
///
/// Note that this function returns `false` even if all items have not yet
/// expired and a call to `poll` will return `NotReady`.
///
/// # Examples
///
/// ```
/// use tokio::timer::DelayQueue;
/// use std::time::Duration;
///
/// let mut delay_queue = DelayQueue::new();
/// assert!(delay_queue.is_empty());
///
/// delay_queue.insert("hello", Duration::from_secs(5));
/// assert!(!delay_queue.is_empty());
/// ```
pub fn is_empty(&self) -> bool {
self.slab.is_empty()
}
/// Polls the queue, returning the index of the next slot in the slab that
/// should be returned.
///
/// A slot should be returned when the associated deadline has been reached.
fn poll_idx(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Result<usize, Error>>> {
use self::wheel::Stack;
let expired = self.expired.pop(&mut self.slab);
if expired.is_some() {
return Poll::Ready(expired.map(Ok));
}
loop {
if let Some(ref mut delay) = self.delay {
if !delay.is_elapsed() {
ready!(Pin::new(&mut *delay).poll(cx));
}
let now =
crate::timer::ms(delay.deadline() - self.start, crate::timer::Round::Down);
self.poll = wheel::Poll::new(now);
}
self.delay = None;
if let Some(idx) = self.wheel.poll(&mut self.poll, &mut self.slab) {
return Poll::Ready(Some(Ok(idx)));
}
if let Some(deadline) = self.next_deadline() {
self.delay = Some(self.handle.delay(deadline));
} else {
return Poll::Ready(None);
}
}
}
fn normalize_deadline(&self, when: Instant) -> u64 {
let when = if when < self.start {
0
} else {
crate::timer::ms(when - self.start, crate::timer::Round::Up)
};
cmp::max(when, self.wheel.elapsed())
}
}
// We never put `T` in a `Pin`...
impl<T> Unpin for DelayQueue<T> {}
impl<T> futures_core::Stream for DelayQueue<T> {
// DelayQueue seems much more specific, where a user may care that it
// has reached capacity, so return those errors instead of panicking.
type Item = Result<Expired<T>, Error>;
fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
DelayQueue::poll_next(self.get_mut(), cx)
}
}
impl<T> Default for DelayQueue<T> {
fn default() -> DelayQueue<T> {
DelayQueue::new()
}
}
impl<T> wheel::Stack for Stack<T> {
type Owned = usize;
type Borrowed = usize;
type Store = Slab<Data<T>>;
fn is_empty(&self) -> bool {
self.head.is_none()
}
fn push(&mut self, item: Self::Owned, store: &mut Self::Store) {
// Ensure the entry is not already in a stack.
debug_assert!(store[item].next.is_none());
debug_assert!(store[item].prev.is_none());
// Remove the old head entry
let old = self.head.take();
if let Some(idx) = old {
store[idx].prev = Some(item);
}
store[item].next = old;
self.head = Some(item)
}
fn pop(&mut self, store: &mut Self::Store) -> Option<Self::Owned> {
if let Some(idx) = self.head {
self.head = store[idx].next;
if let Some(idx) = self.head {
store[idx].prev = None;
}
store[idx].next = None;
debug_assert!(store[idx].prev.is_none());
Some(idx)
} else {
None
}
}
fn remove(&mut self, item: &Self::Borrowed, store: &mut Self::Store) {
assert!(store.contains(*item));
// Ensure that the entry is in fact contained by the stack
debug_assert!({
// This walks the full linked list even if an entry is found.
let mut next = self.head;
let mut contains = false;
while let Some(idx) = next {
if idx == *item {
debug_assert!(!contains);
contains = true;
}
next = store[idx].next;
}
contains
});
if let Some(next) = store[*item].next {
store[next].prev = store[*item].prev;
}
if let Some(prev) = store[*item].prev {
store[prev].next = store[*item].next;
} else {
self.head = store[*item].next;
}
store[*item].next = None;
store[*item].prev = None;
}
fn when(item: &Self::Borrowed, store: &Self::Store) -> u64 {
store[*item].when
}
}
impl<T> Default for Stack<T> {
fn default() -> Stack<T> {
Stack {
head: None,
_p: PhantomData,
}
}
}
impl Key {
pub(crate) fn new(index: usize) -> Key {
Key { index }
}
}
impl<T> Expired<T> {
/// Returns a reference to the inner value.
pub fn get_ref(&self) -> &T {
&self.data
}
/// Returns a mutable reference to the inner value.
pub fn get_mut(&mut self) -> &mut T {
&mut self.data
}
/// Consumes `self` and returns the inner value.
pub fn into_inner(self) -> T {
self.data
}
}
+72
View File
@@ -0,0 +1,72 @@
use self::Kind::*;
use std::error;
use std::fmt;
/// Errors encountered by the timer implementation.
///
/// Currently, there are two different errors that can occur:
///
/// * `shutdown` occurs when a timer operation is attempted, but the timer
/// instance has been dropped. In this case, the operation will never be able
/// to complete and the `shutdown` error is returned. This is a permanent
/// error, i.e., once this error is observed, timer operations will never
/// succeed in the future.
///
/// * `at_capacity` occurs when a timer operation is attempted, but the timer
/// instance is currently handling its maximum number of outstanding delays.
/// In this case, the operation is not able to be performed at the current
/// moment, and `at_capacity` is returned. This is a transient error, i.e., at
/// some point in the future, if the operation is attempted again, it might
/// succeed. Callers that observe this error should attempt to [shed load]. One
/// 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)]
pub struct Error(Kind);
#[derive(Debug)]
enum Kind {
Shutdown,
AtCapacity,
}
impl Error {
/// Create an error representing a shutdown timer.
pub fn shutdown() -> Error {
Error(Shutdown)
}
/// Returns `true` if the error was caused by the timer being shutdown.
pub fn is_shutdown(&self) -> bool {
match self.0 {
Kind::Shutdown => true,
_ => false,
}
}
/// Create an error representing a timer at capacity.
pub fn at_capacity() -> Error {
Error(AtCapacity)
}
/// Returns `true` if the error was caused by the timer being at capacity.
pub fn is_at_capacity(&self) -> bool {
match self.0 {
Kind::AtCapacity => true,
_ => false,
}
}
}
impl error::Error for Error {}
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
use self::Kind::*;
let descr = match self.0 {
Shutdown => "timer is shutdown",
AtCapacity => "timer is at capacity and cannot create a new entry",
};
write!(fmt, "{}", descr)
}
}
+112
View File
@@ -0,0 +1,112 @@
use crate::timer::{clock, Delay};
use futures_core::ready;
use futures_util::future::poll_fn;
use std::future::Future;
use std::pin::Pin;
use std::task::{self, Poll};
use std::time::{Duration, Instant};
/// A stream representing notifications at fixed interval
#[derive(Debug)]
pub struct Interval {
/// Future that completes the next time the `Interval` yields a value.
delay: Delay,
/// The duration between values yielded by `Interval`.
duration: Duration,
}
impl Interval {
/// Create a new `Interval` that starts at `at` and yields every `duration`
/// interval after that.
///
/// Note that when it starts, it produces item too.
///
/// The `duration` argument must be a non-zero duration.
///
/// # Panics
///
/// This function panics if `duration` is zero.
pub fn new(at: Instant, duration: Duration) -> Interval {
assert!(
duration > Duration::new(0, 0),
"`duration` must be non-zero."
);
Interval::new_with_delay(Delay::new(at), duration)
}
/// Creates new `Interval` that yields with interval of `duration`.
///
/// The function is shortcut for `Interval::new(tokio::timer::clock::now() + duration, duration)`.
///
/// The `duration` argument must be a non-zero duration.
///
/// # Panics
///
/// This function panics if `duration` is zero.
pub fn new_interval(duration: Duration) -> Interval {
Interval::new(clock::now() + duration, duration)
}
pub(crate) fn new_with_delay(delay: Delay, duration: Duration) -> Interval {
Interval { delay, duration }
}
#[doc(hidden)] // TODO: remove
pub fn poll_next(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Instant>> {
// Wait for the delay to be done
ready!(Pin::new(&mut self.delay).poll(cx));
// Get the `now` by looking at the `delay` deadline
let now = self.delay.deadline();
// The next interval value is `duration` after the one that just
// yielded.
let next = now + self.duration;
self.delay.reset(next);
// Return the current instant
Poll::Ready(Some(now))
}
/// Completes when the next instant in the interval has been reached.
///
/// # Examples
///
/// ```
/// use tokio::timer::Interval;
///
/// use std::time::Duration;
///
/// #[tokio::main]
/// async fn main() {
/// let mut interval = Interval::new_interval(Duration::from_millis(10));
///
/// interval.next().await;
/// interval.next().await;
/// interval.next().await;
///
/// // approximately 30ms have elapsed.
/// }
/// ```
#[allow(clippy::should_implement_trait)] // TODO: rename (tokio-rs/tokio#1261)
pub async fn next(&mut self) -> Option<Instant> {
poll_fn(|cx| self.poll_next(cx)).await
}
}
impl futures_core::FusedStream for Interval {
fn is_terminated(&self) -> bool {
false
}
}
impl futures_core::Stream for Interval {
type Item = Instant;
fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
Interval::poll_next(self.get_mut(), cx)
}
}
+72 -3
View File
@@ -74,6 +74,75 @@
//! [Interval]: struct.Interval.html
//! [`DelayQueue`]: struct.DelayQueue.html
pub use tokio_timer::{
delay, delay_for, delay_queue, timeout, Delay, DelayQueue, Error, Interval, Timeout,
};
pub mod clock;
pub mod delay_queue;
#[doc(inline)]
pub use self::delay_queue::DelayQueue;
pub mod throttle;
// TODO: clean this up
#[allow(clippy::module_inception)]
pub mod timer;
pub use timer::{set_default, Timer};
pub mod timeout;
#[doc(inline)]
pub use timeout::Timeout;
mod atomic;
mod delay;
pub use self::delay::Delay;
mod error;
pub use error::Error;
mod interval;
pub use interval::Interval;
mod wheel;
use std::time::{Duration, Instant};
/// Create a Future that completes at `deadline`.
pub fn delay(deadline: Instant) -> Delay {
Delay::new(deadline)
}
/// Create a Future that completes in `duration` from now.
///
/// Equivalent to `delay(tokio::timer::clock::now() + duration)`. Analogous to `std::thread::sleep`.
pub fn delay_for(duration: Duration) -> Delay {
delay(clock::now() + 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))
}
+85
View File
@@ -0,0 +1,85 @@
//! Slow down a stream by enforcing a delay between items.
use crate::timer::{clock, Delay};
use futures_core::ready;
use futures_core::Stream;
use std::{
future::Future,
marker::Unpin,
pin::Pin,
task::{self, Poll},
time::Duration,
};
/// Slow down a stream by enforcing a delay between items.
#[derive(Debug)]
#[must_use = "streams do nothing unless polled"]
pub struct Throttle<T> {
delay: Delay,
/// Set to true when `delay` has returned ready, but `stream` hasn't.
has_delayed: bool,
stream: T,
}
impl<T> Throttle<T> {
/// Slow down a stream by enforcing a delay between items.
pub fn new(stream: T, duration: Duration) -> Self {
Self {
delay: Delay::new_timeout(clock::now() + duration, duration),
has_delayed: true,
stream,
}
}
}
// XXX: are these safe if `T: !Unpin`?
impl<T: Unpin> Throttle<T> {
/// Acquires a reference to the underlying stream that this combinator is
/// pulling from.
pub fn get_ref(&self) -> &T {
&self.stream
}
/// Acquires a mutable reference to the underlying stream that this combinator
/// is pulling from.
///
/// Note that care must be taken to avoid tampering with the state of the stream
/// which may otherwise confuse this combinator.
pub fn get_mut(&mut self) -> &mut T {
&mut self.stream
}
/// Consumes this combinator, returning the underlying stream.
///
/// Note that this may discard intermediate state of this combinator, so care
/// should be taken to avoid losing resources when this is called.
pub fn into_inner(self) -> T {
self.stream
}
}
impl<T: Stream> Stream for Throttle<T> {
type Item = T::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
unsafe {
if !self.has_delayed {
ready!(self.as_mut().map_unchecked_mut(|me| &mut me.delay).poll(cx));
self.as_mut().get_unchecked_mut().has_delayed = true;
}
let value = ready!(self
.as_mut()
.map_unchecked_mut(|me| &mut me.stream)
.poll_next(cx));
if value.is_some() {
self.as_mut().get_unchecked_mut().delay.reset_timeout();
self.as_mut().get_unchecked_mut().has_delayed = false;
}
Poll::Ready(value)
}
}
}
+232
View File
@@ -0,0 +1,232 @@
//! Allows a future or stream to execute for a maximum amount of time.
//!
//! See [`Timeout`] documentation for more details.
//!
//! [`Timeout`]: struct.Timeout.html
use crate::timer::clock::now;
use crate::timer::Delay;
use futures_core::ready;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::task::{self, Poll};
use std::time::{Duration, Instant};
/// Allows a `Future` or `Stream` to execute for a limited amount of time.
///
/// If the future or stream completes before the timeout has expired, then
/// `Timeout` returns the completed value. Otherwise, `Timeout` returns an
/// [`Error`].
///
/// # Futures and Streams
///
/// The exact behavor depends on if the inner value is a `Future` or a `Stream`.
/// In the case of a `Future`, `Timeout` will require the future to complete by
/// a fixed deadline. In the case of a `Stream`, `Timeout` will allow each item
/// to take the entire timeout before returning an error.
///
/// In order to set an upper bound on the processing of the *entire* stream,
/// then a timeout should be set on the future that processes the stream. For
/// example:
///
/// ```rust,no_run
/// use tokio::prelude::*;
/// use tokio::sync::mpsc;
///
/// use std::thread;
/// use std::time::Duration;
///
/// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let (mut tx, rx) = mpsc::unbounded_channel();
///
/// thread::spawn(move || {
/// tx.try_send(()).unwrap();
/// thread::sleep(Duration::from_millis(10));
/// tx.try_send(()).unwrap();
/// });
///
/// let process = rx.for_each(|item| {
/// // do something with `item`
/// # drop(item);
/// # tokio::future::ready(())
/// });
///
/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
/// process.timeout(Duration::from_millis(10)).await?;
/// # Ok(())
/// # }
/// ```
///
/// # Cancelation
///
/// Cancelling a `Timeout` is done by dropping the value. No additional cleanup
/// or other work is required.
///
/// The original future or stream may be obtained by calling [`Timeout::into_inner`]. This
/// consumes the `Timeout`.
///
/// [`Error`]: struct.Error.html
/// [`Timeout::into_inner`]: struct.Timeout.html#method.into_iter
#[must_use = "futures do nothing unless you `.await` or poll them"]
#[derive(Debug)]
pub struct Timeout<T> {
value: T,
delay: Delay,
}
/// Error returned by `Timeout`.
#[derive(Debug)]
pub struct Elapsed(());
impl<T> Timeout<T> {
/// Create a new `Timeout` that allows `value` to execute for a duration of
/// at most `timeout`.
///
/// The exact behavior depends on if `value` is a `Future` or a `Stream`.
///
/// See [type] level documentation for more details.
///
/// [type]: #
///
/// # Examples
///
/// Create a new `Timeout` set to expire in 10 milliseconds.
///
/// ```rust
/// use tokio::timer::Timeout;
/// use tokio::sync::oneshot;
///
/// use std::time::Duration;
///
/// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let (tx, rx) = oneshot::channel();
/// # tx.send(()).unwrap();
///
/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds.
/// Timeout::new(rx, Duration::from_millis(10)).await??;
/// # Ok(())
/// # }
/// ```
pub fn new(value: T, timeout: Duration) -> Timeout<T> {
let delay = Delay::new_timeout(now() + timeout, timeout);
Timeout::new_with_delay(value, delay)
}
pub(crate) fn new_with_delay(value: T, delay: Delay) -> Timeout<T> {
Timeout { value, delay }
}
/// Gets a reference to the underlying value in this timeout.
pub fn get_ref(&self) -> &T {
&self.value
}
/// Gets a mutable reference to the underlying value in this timeout.
pub fn get_mut(&mut self) -> &mut T {
&mut self.value
}
/// Consumes this timeout, returning the underlying value.
pub fn into_inner(self) -> T {
self.value
}
}
impl<T: Future> Timeout<T> {
/// Create a new `Timeout` that completes when `future` completes or when
/// `deadline` is reached.
///
/// This function differs from `new` in that:
///
/// * It only accepts `Future` arguments.
/// * It sets an explicit `Instant` at which the timeout expires.
pub fn new_at(future: T, deadline: Instant) -> Timeout<T> {
let delay = Delay::new(deadline);
Timeout {
value: future,
delay,
}
}
}
impl<T> Future for Timeout<T>
where
T: Future,
{
type Output = Result<T::Output, Elapsed>;
fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
// First, try polling the future
// Safety: we never move `self.value`
unsafe {
let p = self.as_mut().map_unchecked_mut(|me| &mut me.value);
if let Poll::Ready(v) = p.poll(cx) {
return Poll::Ready(Ok(v));
}
}
// Now check the timer
// Safety: X_X!
unsafe {
match self.map_unchecked_mut(|me| &mut me.delay).poll(cx) {
Poll::Ready(()) => Poll::Ready(Err(Elapsed(()))),
Poll::Pending => Poll::Pending,
}
}
}
}
impl<T> futures_core::Stream for Timeout<T>
where
T: futures_core::Stream,
{
type Item = Result<T::Item, Elapsed>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
// Safety: T might be !Unpin, but we never move neither `value`
// nor `delay`.
//
// ... X_X
unsafe {
// First, try polling the future
let v = self
.as_mut()
.map_unchecked_mut(|me| &mut me.value)
.poll_next(cx);
if let Poll::Ready(v) = v {
if v.is_some() {
self.as_mut().get_unchecked_mut().delay.reset_timeout();
}
return Poll::Ready(v.map(Ok));
}
// Now check the timer
ready!(self.as_mut().map_unchecked_mut(|me| &mut me.delay).poll(cx));
// if delay was ready, timeout elapsed!
self.as_mut().get_unchecked_mut().delay.reset_timeout();
Poll::Ready(Some(Err(Elapsed(()))))
}
}
}
// ===== impl Elapsed =====
impl fmt::Display for Elapsed {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
"deadline has elapsed".fmt(fmt)
}
}
impl std::error::Error for Elapsed {}
impl From<Elapsed> for std::io::Error {
fn from(_err: Elapsed) -> std::io::Error {
std::io::ErrorKind::TimedOut.into()
}
}
+124
View File
@@ -0,0 +1,124 @@
use super::Entry;
use crate::timer::Error;
use std::ptr;
use std::sync::atomic::AtomicPtr;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
/// A stack of `Entry` nodes
#[derive(Debug)]
pub(crate) struct AtomicStack {
/// Stack head
head: AtomicPtr<Entry>,
}
/// Entries that were removed from the stack
#[derive(Debug)]
pub(crate) struct AtomicStackEntries {
ptr: *mut Entry,
}
/// Used to indicate that the timer has shutdown.
const SHUTDOWN: *mut Entry = 1 as *mut _;
impl AtomicStack {
pub(crate) fn new() -> AtomicStack {
AtomicStack {
head: AtomicPtr::new(ptr::null_mut()),
}
}
/// Push an entry onto the stack.
///
/// Returns `true` if the entry was pushed, `false` if the entry is already
/// on the stack, `Err` if the timer is shutdown.
pub(crate) fn push(&self, entry: &Arc<Entry>) -> Result<bool, Error> {
// First, set the queued bit on the entry
let queued = entry.queued.fetch_or(true, SeqCst);
if queued {
// Already queued, nothing more to do
return Ok(false);
}
let ptr = Arc::into_raw(entry.clone()) as *mut _;
let mut curr = self.head.load(SeqCst);
loop {
if curr == SHUTDOWN {
// Don't leak the entry node
let _ = unsafe { Arc::from_raw(ptr) };
return Err(Error::shutdown());
}
// Update the `next` pointer. This is safe because setting the queued
// bit is a "lock" on this field.
unsafe {
*(entry.next_atomic.get()) = curr;
}
let actual = self.head.compare_and_swap(curr, ptr, SeqCst);
if actual == curr {
break;
}
curr = actual;
}
Ok(true)
}
/// Take all entries from the stack
pub(crate) fn take(&self) -> AtomicStackEntries {
let ptr = self.head.swap(ptr::null_mut(), SeqCst);
AtomicStackEntries { ptr }
}
/// Drain all remaining nodes in the stack and prevent any new nodes from
/// being pushed onto the stack.
pub(crate) fn shutdown(&self) {
// Shutdown the processing queue
let ptr = self.head.swap(SHUTDOWN, SeqCst);
// Let the drop fn of `AtomicStackEntries` handle draining the stack
drop(AtomicStackEntries { ptr });
}
}
// ===== impl AtomicStackEntries =====
impl Iterator for AtomicStackEntries {
type Item = Arc<Entry>;
fn next(&mut self) -> Option<Self::Item> {
if self.ptr.is_null() {
return None;
}
// Convert the pointer to an `Arc<Entry>`
let entry = unsafe { Arc::from_raw(self.ptr) };
// Update `self.ptr` to point to the next element of the stack
self.ptr = unsafe { (*entry.next_atomic.get()) };
// Unset the queued flag
let res = entry.queued.fetch_and(false, SeqCst);
debug_assert!(res);
// Return the entry
Some(entry)
}
}
impl Drop for AtomicStackEntries {
fn drop(&mut self) {
for entry in self {
// Flag the entry as errored
entry.error();
}
}
}
+393
View File
@@ -0,0 +1,393 @@
use crate::timer::atomic::AtomicU64;
use crate::timer::timer::{HandlePriv, Inner};
use crate::timer::Error;
use tokio_sync::AtomicWaker;
use crossbeam_utils::CachePadded;
use std::cell::UnsafeCell;
use std::ptr;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering::{Relaxed, SeqCst};
use std::sync::{Arc, Weak};
use std::task::{self, Poll};
use std::time::{Duration, Instant};
use std::u64;
/// Internal state shared between a `Delay` instance and the timer.
///
/// This struct is used as a node in two intrusive data structures:
///
/// * An atomic stack used to signal to the timer thread that the entry state
/// has changed. The timer thread will observe the entry on this stack and
/// perform any actions as necessary.
///
/// * A doubly linked list used **only** by the timer thread. Each slot in the
/// timer wheel is a head pointer to the list of entries that must be
/// processed during that timer tick.
#[derive(Debug)]
pub(crate) struct Entry {
/// Only accessed from `Registration`.
time: CachePadded<UnsafeCell<Time>>,
/// Timer internals. Using a weak pointer allows the timer to shutdown
/// without all `Delay` instances having completed.
///
/// When `None`, the entry has not yet been linked with a timer instance.
inner: Option<Weak<Inner>>,
/// Tracks the entry state. This value contains the following information:
///
/// * The deadline at which the entry must be "fired".
/// * A flag indicating if the entry has already been fired.
/// * Whether or not the entry transitioned to the error state.
///
/// When an `Entry` is created, `state` is initialized to the instant at
/// which the entry must be fired. When a timer is reset to a different
/// instant, this value is changed.
state: AtomicU64,
/// Task to notify once the deadline is reached.
waker: AtomicWaker,
/// True when the entry is queued in the "process" stack. This value
/// is set before pushing the value and unset after popping the value.
///
/// TODO: This could possibly be rolled up into `state`.
pub(super) queued: AtomicBool,
/// Next entry in the "process" linked list.
///
/// Access to this field is coordinated by the `queued` flag.
///
/// Represents a strong Arc ref.
pub(super) next_atomic: UnsafeCell<*mut Entry>,
/// When the entry expires, relative to the `start` of the timer
/// (Inner::start). This is only used by the timer.
///
/// A `Delay` instance can be reset to a different deadline by the thread
/// that owns the `Delay` instance. In this case, the timer thread will not
/// immediately know that this has happened. The timer thread must know the
/// last deadline that it saw as it uses this value to locate the entry in
/// its wheel.
///
/// Once the timer thread observes that the instant has changed, it updates
/// the wheel and sets this value. The idea is that this value eventually
/// converges to the value of `state` as the timer thread makes updates.
when: UnsafeCell<Option<u64>>,
/// Next entry in the State's linked list.
///
/// This is only accessed by the timer
pub(super) next_stack: UnsafeCell<Option<Arc<Entry>>>,
/// Previous entry in the State's linked list.
///
/// This is only accessed by the timer and is used to unlink a canceled
/// entry.
///
/// This is a weak reference.
pub(super) prev_stack: UnsafeCell<*const Entry>,
}
/// Stores the info for `Delay`.
#[derive(Debug)]
pub(crate) struct Time {
pub(crate) deadline: Instant,
pub(crate) duration: Duration,
}
/// Flag indicating a timer entry has elapsed
const ELAPSED: u64 = 1 << 63;
/// Flag indicating a timer entry has reached an error state
const ERROR: u64 = u64::MAX;
// ===== impl Entry =====
impl Entry {
pub(crate) fn new(deadline: Instant, duration: Duration) -> Entry {
Entry {
time: CachePadded::new(UnsafeCell::new(Time { deadline, duration })),
inner: None,
waker: AtomicWaker::new(),
state: AtomicU64::new(0),
queued: AtomicBool::new(false),
next_atomic: UnsafeCell::new(ptr::null_mut()),
when: UnsafeCell::new(None),
next_stack: UnsafeCell::new(None),
prev_stack: UnsafeCell::new(ptr::null_mut()),
}
}
/// Only called by `Registration`
pub(crate) fn time_ref(&self) -> &Time {
unsafe { &*self.time.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.get()
}
/// Returns `true` if the `Entry` is currently associated with a timer
/// instance.
pub(crate) fn is_registered(&self) -> bool {
self.inner.is_some()
}
/// Only called by `Registration`
pub(crate) fn register(me: &mut Arc<Self>) {
let handle = match HandlePriv::try_current() {
Ok(handle) => handle,
Err(_) => {
// Could not associate the entry with a timer, transition the
// state to error
Arc::get_mut(me).unwrap().transition_to_error();
return;
}
};
Entry::register_with(me, handle)
}
/// Only called by `Registration`
pub(crate) fn register_with(me: &mut Arc<Self>, handle: HandlePriv) {
assert!(!me.is_registered(), "only register an entry once");
let deadline = me.time_ref().deadline;
let inner = match handle.inner() {
Some(inner) => inner,
None => {
// Could not associate the entry with a timer, transition the
// state to error
Arc::get_mut(me).unwrap().transition_to_error();
return;
}
};
// Increment the number of active timeouts
if inner.increment().is_err() {
Arc::get_mut(me).unwrap().transition_to_error();
return;
}
// Associate the entry with the timer
Arc::get_mut(me).unwrap().inner = Some(handle.into_inner());
let when = inner.normalize_deadline(deadline);
// Relaxed OK: At this point, there are no other threads that have
// access to this entry.
if when <= inner.elapsed() {
me.state.store(ELAPSED, Relaxed);
return;
} else {
me.state.store(when, Relaxed);
}
if inner.queue(me).is_err() {
// The timer has shutdown, transition the entry to the error state.
me.error();
}
}
fn transition_to_error(&mut self) {
self.inner = Some(Weak::new());
self.state = AtomicU64::new(ERROR);
}
/// The current entry state as known by the timer. This is not the value of
/// `state`, but lets the timer know how to converge its state to `state`.
pub(crate) fn when_internal(&self) -> Option<u64> {
unsafe { (*self.when.get()) }
}
pub(crate) fn set_when_internal(&self, when: Option<u64>) {
unsafe {
(*self.when.get()) = when;
}
}
/// Called by `Timer` to load the current value of `state` for processing
pub(crate) fn load_state(&self) -> Option<u64> {
let state = self.state.load(SeqCst);
if is_elapsed(state) {
None
} else {
Some(state)
}
}
pub(crate) fn is_elapsed(&self) -> bool {
let state = self.state.load(SeqCst);
is_elapsed(state)
}
pub(crate) fn fire(&self, when: u64) {
let mut curr = self.state.load(SeqCst);
loop {
if is_elapsed(curr) || curr > when {
return;
}
let next = ELAPSED | curr;
let actual = self.state.compare_and_swap(curr, next, SeqCst);
if curr == actual {
break;
}
curr = actual;
}
self.waker.wake();
}
pub(crate) fn error(&self) {
// Only transition to the error state if not currently elapsed
let mut curr = self.state.load(SeqCst);
loop {
if is_elapsed(curr) {
return;
}
let next = ERROR;
let actual = self.state.compare_and_swap(curr, next, SeqCst);
if curr == actual {
break;
}
curr = actual;
}
self.waker.wake();
}
pub(crate) fn cancel(entry: &Arc<Entry>) {
let state = entry.state.fetch_or(ELAPSED, SeqCst);
if is_elapsed(state) {
// Nothing more to do
return;
}
// If registered with a timer instance, try to upgrade the Arc.
let inner = match entry.upgrade_inner() {
Some(inner) => inner,
None => return,
};
let _ = inner.queue(entry);
}
pub(crate) fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
let mut curr = self.state.load(SeqCst);
if is_elapsed(curr) {
return Poll::Ready(if curr == ERROR {
Err(Error::shutdown())
} else {
Ok(())
});
}
self.waker.register_by_ref(cx.waker());
curr = self.state.load(SeqCst);
if is_elapsed(curr) {
return Poll::Ready(if curr == ERROR {
Err(Error::shutdown())
} else {
Ok(())
});
}
Poll::Pending
}
/// Only called by `Registration`
pub(crate) fn reset(entry: &mut Arc<Entry>) {
if !entry.is_registered() {
return;
}
let inner = match entry.upgrade_inner() {
Some(inner) => inner,
None => return,
};
let deadline = entry.time_ref().deadline;
let when = inner.normalize_deadline(deadline);
let elapsed = inner.elapsed();
let mut curr = entry.state.load(SeqCst);
let mut notify;
loop {
// In these two cases, there is no work to do when resetting the
// timer. If the `Entry` is in an error state, then it cannot be
// used anymore. If resetting the entry to the current value, then
// the reset is a noop.
if curr == ERROR || curr == when {
return;
}
let next;
if when <= elapsed {
next = ELAPSED;
notify = !is_elapsed(curr);
} else {
next = when;
notify = true;
}
let actual = entry.state.compare_and_swap(curr, next, SeqCst);
if curr == actual {
break;
}
curr = actual;
}
if notify {
let _ = inner.queue(entry);
}
}
fn upgrade_inner(&self) -> Option<Arc<Inner>> {
self.inner.as_ref().and_then(|inner| inner.upgrade())
}
}
fn is_elapsed(state: u64) -> bool {
state & ELAPSED == ELAPSED
}
impl Drop for Entry {
fn drop(&mut self) {
let inner = match self.upgrade_inner() {
Some(inner) => inner,
None => return,
};
inner.decrement();
}
}
unsafe impl Send for Entry {}
unsafe impl Sync for Entry {}
+187
View File
@@ -0,0 +1,187 @@
use crate::timer::clock::now;
use crate::timer::timer::Inner;
use crate::timer::{Delay, Error, Timeout};
use std::cell::RefCell;
use std::fmt;
use std::marker::PhantomData;
use std::sync::{Arc, Weak};
use std::time::{Duration, Instant};
/// Handle to timer instance.
///
/// The `Handle` allows creating `Delay` instances that are driven by the
/// associated timer.
///
/// A `Handle` is obtained by calling [`Timer::handle`], [`Handle::current`], or
/// [`Handle::default`].
///
/// * [`Timer::handle`]: returns a handle associated with the specific timer.
/// The handle will always reference the same timer.
///
/// * [`Handle::current`]: returns a handle to the timer for the execution
/// context **at the time the function is called**. This function must be
/// called from a runtime that has an associated timer or it will panic.
/// The handle will always reference the same timer.
///
/// * [`Handle::default`]: returns a handle to the timer for the execution
/// context **at the time the handle is used**. This function is safe to call
/// at any time. The handle may reference different specific timer instances.
/// Calling `Handle::default().delay(...)` is always equivalent to
/// `Delay::new(...)`.
///
/// [`Timer::handle`]: struct.Timer.html#method.handle
/// [`Handle::current`]: #method.current
/// [`Handle::default`]: #method.default
#[derive(Debug, Clone)]
pub struct Handle {
inner: Option<HandlePriv>,
}
/// Like `Handle` but never `None`.
#[derive(Clone)]
pub(crate) struct HandlePriv {
inner: Weak<Inner>,
}
thread_local! {
/// Tracks the timer for the current execution context.
static CURRENT_TIMER: RefCell<Option<HandlePriv>> = RefCell::new(None)
}
#[derive(Debug)]
///Unsets default timer handler on drop.
pub struct DefaultGuard<'a> {
_lifetime: PhantomData<&'a u8>,
}
impl Drop for DefaultGuard<'_> {
fn drop(&mut self) {
CURRENT_TIMER.with(|current| {
let mut current = current.borrow_mut();
*current = None;
})
}
}
///Sets handle to default timer, returning guard that unsets it on drop.
///
/// # Panics
///
/// This function panics if there already is a default timer set.
pub fn set_default(handle: &Handle) -> DefaultGuard<'_> {
CURRENT_TIMER.with(|current| {
let mut current = current.borrow_mut();
assert!(
current.is_none(),
"default Tokio timer already set \
for execution context"
);
let handle = handle
.as_priv()
.unwrap_or_else(|| panic!("`handle` does not reference a timer"));
*current = Some(handle.clone());
});
DefaultGuard {
_lifetime: PhantomData,
}
}
impl Handle {
pub(crate) fn new(inner: Weak<Inner>) -> Handle {
let inner = HandlePriv { inner };
Handle { inner: Some(inner) }
}
/// Returns a handle to the current timer.
///
/// The current timer is the timer that is currently set as default using
/// [`with_default`].
///
/// This function should only be called from within the context of
/// [`with_default`]. Calling this function from outside of this context
/// will return a `Handle` that does not reference a timer. `Delay`
/// instances created with this handle will error.
///
/// See [type] level documentation for more ways to obtain a `Handle` value.
///
/// [`with_default`]: fn.with_default
/// [type]: #
pub fn current() -> Handle {
let private =
HandlePriv::try_current().unwrap_or_else(|_| HandlePriv { inner: Weak::new() });
Handle {
inner: Some(private),
}
}
/// Create a `Delay` driven by this handle's associated `Timer`.
pub fn delay(&self, deadline: Instant) -> Delay {
self.delay_timeout(deadline, Duration::from_secs(0))
}
fn delay_timeout(&self, deadline: Instant, duration: Duration) -> Delay {
match self.inner {
Some(ref handle_priv) => {
Delay::new_with_handle(deadline, Duration::from_secs(0), handle_priv.clone())
}
None => Delay::new_timeout(deadline, duration),
}
}
/// Create a `Timeout` driven by this handle's associated `Timer`.
pub fn timeout<T>(&self, value: T, timeout: Duration) -> Timeout<T> {
Timeout::new_with_delay(value, self.delay_timeout(now() + timeout, timeout))
}
/*
/// Create a new `Interval` that starts at `at` and yields every `duration`
/// interval after that.
pub fn interval(&self, at: Instant, duration: Duration) -> Interval {
Interval::new_with_delay(self.delay(at), duration)
}
*/
fn as_priv(&self) -> Option<&HandlePriv> {
self.inner.as_ref()
}
}
impl Default for Handle {
fn default() -> Handle {
Handle { inner: None }
}
}
impl HandlePriv {
/// Try to get a handle to the current timer.
///
/// Returns `Err` if no handle is found.
pub(crate) fn try_current() -> Result<HandlePriv, Error> {
CURRENT_TIMER.with(|current| match *current.borrow() {
Some(ref handle) => Ok(handle.clone()),
None => Err(Error::shutdown()),
})
}
/// Try to return a strong ref to the inner
pub(crate) fn inner(&self) -> Option<Arc<Inner>> {
self.inner.upgrade()
}
/// Consume the handle, returning the weak Inner ref.
pub(crate) fn into_inner(self) -> Weak<Inner> {
self.inner
}
}
impl fmt::Debug for HandlePriv {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "HandlePriv")
}
}
+484
View File
@@ -0,0 +1,484 @@
//! Timer implementation.
//!
//! This module contains the types needed to run a timer.
//!
//! The [`Timer`] type runs the timer logic. It holds all the necessary state
//! to track all associated [`Delay`] instances and delivering notifications
//! once the deadlines are reached.
//!
//! The [`Handle`] type is a reference to a [`Timer`] instance. This type is
//! `Clone`, `Send`, and `Sync`. This type is used to create instances of
//! [`Delay`].
//!
//! [`Timer`] is generic over [`Now`]. This allows the source of time to be
//! customized. This ability is especially useful in tests and any environment
//! where determinism is necessary.
//!
//! Note, when using the Tokio runtime, the [`Timer`] does not need to be manually
//! setup as the runtime comes pre-configured with a [`Timer`] instance.
//!
//! [`Timer`]: struct.Timer.html
//! [`Handle`]: struct.Handle.html
//! [`Delay`]: Delay
//! [`Now`]: clock::Now
//! [`Now::now`]: clock::Now::now
//! [`Instant`]: std::time::Instant
//! [`Instant::now`]: std::time::Instant::now
mod atomic_stack;
use self::atomic_stack::AtomicStack;
mod entry;
use self::entry::Entry;
mod handle;
pub(crate) use self::handle::HandlePriv;
pub use self::handle::{set_default, Handle};
mod registration;
pub(crate) use self::registration::Registration;
mod stack;
use self::stack::Stack;
use crate::timer::atomic::AtomicU64;
use crate::timer::clock::Clock;
use crate::timer::wheel;
use crate::timer::Error;
use tokio_executor::park::{Park, ParkThread, Unpark};
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::usize;
use std::{cmp, fmt};
/// Timer implementation that drives [`Delay`], [`Interval`], and [`Timeout`].
///
/// A `Timer` instance tracks the state necessary for managing time and
/// notifying the [`Delay`] instances once their deadlines are reached.
///
/// It is expected that a single `Timer` instance manages many individual
/// [`Delay`] instances. The `Timer` implementation is thread-safe and, as such,
/// is able to handle callers from across threads.
///
/// Callers do not use `Timer` directly to create [`Delay`] instances. Instead,
/// [`Handle`][Handle.struct] is used. A handle for the timer instance is obtained by calling
/// [`handle`]. [`Handle`][Handle.struct] is the type that implements `Clone` and is `Send +
/// Sync`.
///
/// After creating the `Timer` instance, the caller must repeatedly call
/// [`turn`]. The timer will perform no work unless [`turn`] is called
/// repeatedly.
///
/// The `Timer` has a resolution of one millisecond. Any unit of time that falls
/// between milliseconds are rounded up to the next millisecond.
///
/// When the `Timer` instance is dropped, any outstanding [`Delay`] instance that
/// has not elapsed will be notified with an error. At this point, calling
/// `poll` on the [`Delay`] instance will result in `Err` being returned.
///
/// # Implementation
///
/// `Timer` is based on the [paper by Varghese and Lauck][paper].
///
/// A hashed timing wheel is a vector of slots, where each slot handles a time
/// slice. As time progresses, the timer walks over the slot for the current
/// instant, and processes each entry for that slot. When the timer reaches the
/// end of the wheel, it starts again at the beginning.
///
/// The `Timer` implementation maintains six wheels arranged in a set of levels.
/// As the levels go up, the slots of the associated wheel represent larger
/// intervals of time. At each level, the wheel has 64 slots. Each slot covers a
/// range of time equal to the wheel at the lower level. At level zero, each
/// slot represents one millisecond of time.
///
/// The wheels are:
///
/// * Level 0: 64 x 1 millisecond slots.
/// * Level 1: 64 x 64 millisecond slots.
/// * Level 2: 64 x ~4 second slots.
/// * Level 3: 64 x ~4 minute slots.
/// * Level 4: 64 x ~4 hour slots.
/// * Level 5: 64 x ~12 day slots.
///
/// When the timer processes entries at level zero, it will notify all the
/// [`Delay`] instances as their deadlines have been reached. For all higher
/// levels, all entries will be redistributed across the wheel at the next level
/// down. Eventually, as time progresses, entries will [`Delay`] instances will
/// either be canceled (dropped) or their associated entries will reach level
/// zero and be notified.
///
/// [`Delay`]: struct.Delay.html
/// [`Interval`]: struct.Interval.html
/// [`Timeout`]: struct.Timeout.html
/// [paper]: http://www.cs.columbia.edu/~nahum/w6998/papers/ton97-timing-wheels.pdf
/// [`handle`]: #method.handle
/// [`turn`]: #method.turn
/// [Handle.struct]: struct.Handle.html
#[derive(Debug)]
pub struct Timer<T> {
/// Shared state
inner: Arc<Inner>,
/// Timer wheel
wheel: wheel::Wheel<Stack>,
/// Thread parker. The `Timer` park implementation delegates to this.
park: T,
/// Source of "now" instances
clock: Clock,
}
/// Return value from the `turn` method on `Timer`.
///
/// Currently this value doesn't actually provide any functionality, but it may
/// in the future give insight into what happened during `turn`.
#[derive(Debug)]
pub struct Turn(());
/// Timer state shared between `Timer`, `Handle`, and `Registration`.
pub(crate) struct Inner {
/// The instant at which the timer started running.
start: Instant,
/// The last published timer `elapsed` value.
elapsed: AtomicU64,
/// Number of active timeouts
num: AtomicUsize,
/// Head of the "process" linked list.
process: AtomicStack,
/// Unparks the timer thread.
unpark: Box<dyn Unpark>,
}
/// Maximum number of timeouts the system can handle concurrently.
const MAX_TIMEOUTS: usize = usize::MAX >> 1;
// ===== impl Timer =====
impl<T> Timer<T>
where
T: Park,
{
/// Create a new `Timer` instance that uses `park` to block the current
/// thread.
///
/// Once the timer has been created, a handle can be obtained using
/// [`handle`]. The handle is used to create `Delay` instances.
///
/// Use `default` when constructing a `Timer` using the default `park`
/// instance.
///
/// [`handle`]: #method.handle
pub fn new(park: T) -> Self {
Timer::new_with_clock(park, Clock::new())
}
}
impl<T> Timer<T> {
/// Returns a reference to the underlying `Park` instance.
pub fn get_park(&self) -> &T {
&self.park
}
/// Returns a mutable reference to the underlying `Park` instance.
pub fn get_park_mut(&mut self) -> &mut T {
&mut self.park
}
}
impl<T> Timer<T>
where
T: Park,
{
/// Create a new `Timer` instance that uses `park` to block the current
/// thread and `now` to get the current `Instant`.
///
/// Specifying the source of time is useful when testing.
pub fn new_with_clock(park: T, clock: Clock) -> Self {
let unpark = Box::new(park.unpark());
Timer {
inner: Arc::new(Inner::new(clock.now(), unpark)),
wheel: wheel::Wheel::new(),
park,
clock,
}
}
/// Returns a handle to the timer.
///
/// The `Handle` is how `Delay` instances are created. The `Delay` instances
/// can either be created directly or the `Handle` instance can be passed to
/// `with_default`, setting the timer as the default timer for the execution
/// context.
pub fn handle(&self) -> Handle {
Handle::new(Arc::downgrade(&self.inner))
}
/// Performs one iteration of the timer loop.
///
/// This function must be called repeatedly in order for the `Timer`
/// instance to make progress. This is where the work happens.
///
/// The `Timer` will use the `Park` instance that was specified in [`new`]
/// to block the current thread until the next `Delay` instance elapses. One
/// call to `turn` results in at most one call to `park.park()`.
///
/// # Return
///
/// On success, `Ok(Turn)` is returned, where `Turn` is a placeholder type
/// that currently does nothing but may, in the future, have functions add
/// to provide information about the call to `turn`.
///
/// If the call to `park.park()` fails, then `Err` is returned with the
/// error.
///
/// [`new`]: #method.new
pub fn turn(&mut self, max_wait: Option<Duration>) -> Result<Turn, T::Error> {
match max_wait {
Some(timeout) => self.park_timeout(timeout)?,
None => self.park()?,
}
Ok(Turn(()))
}
/// Converts an `Expiration` to an `Instant`.
fn expiration_instant(&self, when: u64) -> Instant {
self.inner.start + Duration::from_millis(when)
}
/// Run timer related logic
fn process(&mut self) {
let now = crate::timer::ms(
self.clock.now() - self.inner.start,
crate::timer::Round::Down,
);
let mut poll = wheel::Poll::new(now);
while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
let when = entry.when_internal().expect("invalid internal entry state");
// Fire the entry
entry.fire(when);
// Track that the entry has been fired
entry.set_when_internal(None);
}
// Update the elapsed cache
self.inner.elapsed.store(self.wheel.elapsed(), SeqCst);
}
/// Process the entry queue
///
/// This handles adding and canceling timeouts.
fn process_queue(&mut self) {
for entry in self.inner.process.take() {
match (entry.when_internal(), entry.load_state()) {
(None, None) => {
// Nothing to do
}
(Some(_), None) => {
// Remove the entry
self.clear_entry(&entry);
}
(None, Some(when)) => {
// Queue the entry
self.add_entry(entry, when);
}
(Some(_), Some(next)) => {
self.clear_entry(&entry);
self.add_entry(entry, next);
}
}
}
}
fn clear_entry(&mut self, entry: &Arc<Entry>) {
self.wheel.remove(entry, &mut ());
entry.set_when_internal(None);
}
/// Fire the entry if it needs to, otherwise queue it to be processed later.
///
/// Returns `None` if the entry was fired.
fn add_entry(&mut self, entry: Arc<Entry>, when: u64) {
use crate::timer::wheel::InsertError;
entry.set_when_internal(Some(when));
match self.wheel.insert(when, entry, &mut ()) {
Ok(_) => {}
Err((entry, InsertError::Elapsed)) => {
// The entry's deadline has elapsed, so fire it and update the
// internal state accordingly.
entry.set_when_internal(None);
entry.fire(when);
}
Err((entry, InsertError::Invalid)) => {
// The entry's deadline is invalid, so error it and update the
// internal state accordingly.
entry.set_when_internal(None);
entry.error();
}
}
}
}
impl Default for Timer<ParkThread> {
fn default() -> Self {
Timer::new(ParkThread::new())
}
}
impl<T> Park for Timer<T>
where
T: Park,
{
type Unpark = T::Unpark;
type Error = T::Error;
fn unpark(&self) -> Self::Unpark {
self.park.unpark()
}
fn park(&mut self) -> Result<(), Self::Error> {
self.process_queue();
match self.wheel.poll_at() {
Some(when) => {
let now = self.clock.now();
let deadline = self.expiration_instant(when);
if deadline > now {
self.park.park_timeout(deadline - now)?;
} else {
self.park.park_timeout(Duration::from_secs(0))?;
}
}
None => {
self.park.park()?;
}
}
self.process();
Ok(())
}
fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
self.process_queue();
match self.wheel.poll_at() {
Some(when) => {
let now = self.clock.now();
let deadline = self.expiration_instant(when);
if deadline > now {
self.park.park_timeout(cmp::min(deadline - now, duration))?;
} else {
self.park.park_timeout(Duration::from_secs(0))?;
}
}
None => {
self.park.park_timeout(duration)?;
}
}
self.process();
Ok(())
}
}
impl<T> Drop for Timer<T> {
fn drop(&mut self) {
use std::u64;
// Shutdown the stack of entries to process, preventing any new entries
// from being pushed.
self.inner.process.shutdown();
// Clear the wheel, using u64::MAX allows us to drain everything
let mut poll = wheel::Poll::new(u64::MAX);
while let Some(entry) = self.wheel.poll(&mut poll, &mut ()) {
entry.error();
}
}
}
// ===== impl Inner =====
impl Inner {
fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
Inner {
num: AtomicUsize::new(0),
elapsed: AtomicU64::new(0),
process: AtomicStack::new(),
start,
unpark,
}
}
fn elapsed(&self) -> u64 {
self.elapsed.load(SeqCst)
}
/// Increment the number of active timeouts
fn increment(&self) -> Result<(), Error> {
let mut curr = self.num.load(SeqCst);
loop {
if curr == MAX_TIMEOUTS {
return Err(Error::at_capacity());
}
let actual = self.num.compare_and_swap(curr, curr + 1, SeqCst);
if curr == actual {
return Ok(());
}
curr = actual;
}
}
/// Decrement the number of active timeouts
fn decrement(&self) {
let prev = self.num.fetch_sub(1, SeqCst);
debug_assert!(prev <= MAX_TIMEOUTS);
}
fn queue(&self, entry: &Arc<Entry>) -> Result<(), Error> {
if self.process.push(entry)? {
// The timer is notified so that it can process the timeout
self.unpark.unpark();
}
Ok(())
}
fn normalize_deadline(&self, deadline: Instant) -> u64 {
if deadline < self.start {
return 0;
}
crate::timer::ms(deadline - self.start, crate::timer::Round::Up)
}
}
impl fmt::Debug for Inner {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Inner").finish()
}
}
+11
View File
@@ -0,0 +1,11 @@
use std::time::Instant;
#[doc(hidden)]
#[deprecated(since = "0.2.4", note = "use clock::Now instead")]
pub trait Now {
/// Returns an instant corresponding to "now".
fn now(&mut self) -> Instant;
}
#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
pub use crate::clock::Clock as SystemNow;
+70
View File
@@ -0,0 +1,70 @@
use crate::timer::timer::{Entry, HandlePriv};
use crate::timer::Error;
use std::sync::Arc;
use std::task::{self, Poll};
use std::time::{Duration, Instant};
/// Registration with a timer.
///
/// The association between a `Delay` instance and a timer is done lazily in
/// `poll`
#[derive(Debug)]
pub(crate) struct Registration {
entry: Arc<Entry>,
}
impl Registration {
pub(crate) fn new(deadline: Instant, duration: Duration) -> Registration {
fn is_send<T: Send + Sync>() {}
is_send::<Registration>();
Registration {
entry: Arc::new(Entry::new(deadline, duration)),
}
}
pub(crate) fn deadline(&self) -> Instant {
self.entry.time_ref().deadline
}
pub(crate) fn register(&mut self) {
if !self.entry.is_registered() {
Entry::register(&mut self.entry)
}
}
pub(crate) fn register_with(&mut self, handle: HandlePriv) {
Entry::register_with(&mut self.entry, handle)
}
pub(crate) fn reset(&mut self, deadline: Instant) {
unsafe {
self.entry.time_mut().deadline = deadline;
}
Entry::reset(&mut self.entry);
}
// Used by `Timeout<Stream>`
pub(crate) fn reset_timeout(&mut self) {
let deadline = crate::clock::now() + self.entry.time_ref().duration;
unsafe {
self.entry.time_mut().deadline = deadline;
}
Entry::reset(&mut self.entry);
}
pub(crate) fn is_elapsed(&self) -> bool {
self.entry.is_elapsed()
}
pub(crate) fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
self.entry.poll_elapsed(cx)
}
}
impl Drop for Registration {
fn drop(&mut self) {
Entry::cancel(&self.entry);
}
}
+121
View File
@@ -0,0 +1,121 @@
use crate::timer::timer::Entry;
use crate::timer::wheel;
use std::ptr;
use std::sync::Arc;
/// A doubly linked stack
#[derive(Debug)]
pub(crate) struct Stack {
head: Option<Arc<Entry>>,
}
impl Default for Stack {
fn default() -> Stack {
Stack { head: None }
}
}
impl wheel::Stack for Stack {
type Owned = Arc<Entry>;
type Borrowed = Entry;
type Store = ();
fn is_empty(&self) -> bool {
self.head.is_none()
}
fn push(&mut self, entry: Self::Owned, _: &mut Self::Store) {
// Get a pointer to the entry to for the prev link
let ptr: *const Entry = &*entry as *const _;
// Remove the old head entry
let old = self.head.take();
unsafe {
// Ensure the entry is not already in a stack.
debug_assert!((*entry.next_stack.get()).is_none());
debug_assert!((*entry.prev_stack.get()).is_null());
if let Some(ref entry) = old.as_ref() {
debug_assert!({
// The head is not already set to the entry
ptr != &***entry as *const _
});
// Set the previous link on the old head
*entry.prev_stack.get() = ptr;
}
// Set this entry's next pointer
*entry.next_stack.get() = old;
}
// Update the head pointer
self.head = Some(entry);
}
/// Pop an item from the stack
fn pop(&mut self, _: &mut ()) -> Option<Arc<Entry>> {
let entry = self.head.take();
unsafe {
if let Some(entry) = entry.as_ref() {
self.head = (*entry.next_stack.get()).take();
if let Some(entry) = self.head.as_ref() {
*entry.prev_stack.get() = ptr::null();
}
*entry.prev_stack.get() = ptr::null();
}
}
entry
}
fn remove(&mut self, entry: &Entry, _: &mut ()) {
unsafe {
// Ensure that the entry is in fact contained by the stack
debug_assert!({
// This walks the full linked list even if an entry is found.
let mut next = self.head.as_ref();
let mut contains = false;
while let Some(n) = next {
if entry as *const _ == &**n as *const _ {
debug_assert!(!contains);
contains = true;
}
next = (*n.next_stack.get()).as_ref();
}
contains
});
// Unlink `entry` from the next node
let next = (*entry.next_stack.get()).take();
if let Some(next) = next.as_ref() {
(*next.prev_stack.get()) = *entry.prev_stack.get();
}
// Unlink `entry` from the prev node
if let Some(prev) = (*entry.prev_stack.get()).as_ref() {
*prev.next_stack.get() = next;
} else {
// It is the head
self.head = next;
}
// Unset the prev pointer
*entry.prev_stack.get() = ptr::null();
}
}
fn when(item: &Entry, _: &()) -> u64 {
item.when_internal().expect("invalid internal state")
}
}
+255
View File
@@ -0,0 +1,255 @@
use crate::timer::wheel::Stack;
use std::fmt;
/// Wheel for a single level in the timer. This wheel contains 64 slots.
pub(crate) struct Level<T> {
level: usize,
/// Bit field tracking which slots currently contain entries.
///
/// Using a bit field to track slots that contain entries allows avoiding a
/// scan to find entries. This field is updated when entries are added or
/// removed from a slot.
///
/// The least-significant bit represents slot zero.
occupied: u64,
/// Slots
slot: [T; LEVEL_MULT],
}
/// Indicates when a slot must be processed next.
#[derive(Debug)]
pub(crate) struct Expiration {
/// The level containing the slot.
pub(crate) level: usize,
/// The slot index.
pub(crate) slot: usize,
/// The instant at which the slot needs to be processed.
pub(crate) deadline: u64,
}
/// Level multiplier.
///
/// Being a power of 2 is very important.
const LEVEL_MULT: usize = 64;
impl<T: Stack> Level<T> {
pub(crate) fn new(level: usize) -> Level<T> {
// Rust's derived implementations for arrays require that the value
// contained by the array be `Copy`. So, here we have to manually
// initialize every single slot.
macro_rules! s {
() => {
T::default()
};
};
Level {
level,
occupied: 0,
slot: [
// It does not look like the necessary traits are
// derived for [T; 64].
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
s!(),
],
}
}
/// Finds the slot that needs to be processed next and returns the slot and
/// `Instant` at which this slot must be processed.
pub(crate) fn next_expiration(&self, now: u64) -> Option<Expiration> {
// Use the `occupied` bit field to get the index of the next slot that
// needs to be processed.
let slot = match self.next_occupied_slot(now) {
Some(slot) => slot,
None => return None,
};
// From the slot index, calculate the `Instant` at which it needs to be
// processed. This value *must* be in the future with respect to `now`.
let level_range = level_range(self.level);
let slot_range = slot_range(self.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;
debug_assert!(
deadline >= now,
"deadline={}; now={}; level={}; slot={}; occupied={:b}",
deadline,
now,
self.level,
slot,
self.occupied
);
Some(Expiration {
level: self.level,
slot,
deadline,
})
}
fn next_occupied_slot(&self, now: u64) -> Option<usize> {
if self.occupied == 0 {
return None;
}
// Get the slot for now using Maths
let now_slot = (now / slot_range(self.level)) as usize;
let occupied = self.occupied.rotate_right(now_slot as u32);
let zeros = occupied.trailing_zeros() as usize;
let slot = (zeros + now_slot) % 64;
Some(slot)
}
pub(crate) fn add_entry(&mut self, when: u64, item: T::Owned, store: &mut T::Store) {
let slot = slot_for(when, self.level);
self.slot[slot].push(item, store);
self.occupied |= occupied_bit(slot);
}
pub(crate) fn remove_entry(&mut self, when: u64, item: &T::Borrowed, store: &mut T::Store) {
let slot = slot_for(when, self.level);
self.slot[slot].remove(item, store);
if self.slot[slot].is_empty() {
// The bit is currently set
debug_assert!(self.occupied & occupied_bit(slot) != 0);
// Unset the bit
self.occupied ^= occupied_bit(slot);
}
}
pub(crate) fn pop_entry_slot(&mut self, slot: usize, store: &mut T::Store) -> Option<T::Owned> {
let ret = self.slot[slot].pop(store);
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
}
}
impl<T> fmt::Debug for Level<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("Level")
.field("occupied", &self.occupied)
.finish()
}
}
fn occupied_bit(slot: usize) -> u64 {
(1 << slot)
}
fn slot_range(level: usize) -> u64 {
LEVEL_MULT.pow(level as u32) as u64
}
fn level_range(level: usize) -> u64 {
LEVEL_MULT as u64 * slot_range(level)
}
/// Convert a duration (milliseconds) and a level to a slot position
fn slot_for(duration: u64, level: usize) -> usize {
((duration >> (level * 6)) % LEVEL_MULT as u64) as usize
}
/*
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_slot_for() {
for pos in 1..64 {
assert_eq!(pos as usize, slot_for(pos, 0));
}
for level in 1..5 {
for pos in level..64 {
let a = pos * 64_usize.pow(level as u32);
assert_eq!(pos as usize, slot_for(a as u64, level));
}
}
}
}
*/
+311
View File
@@ -0,0 +1,311 @@
mod level;
pub(crate) use self::level::Expiration;
use self::level::Level;
mod stack;
pub(crate) use self::stack::Stack;
use std::borrow::Borrow;
use std::usize;
/// Timing wheel implementation.
///
/// This type provides the hashed timing wheel implementation that backs `Timer`
/// and `DelayQueue`.
///
/// The structure is generic over `T: Stack`. This allows handling timeout data
/// being stored on the heap or in a slab. In order to support the latter case,
/// the slab must be passed into each function allowing the implementation to
/// lookup timer entries.
///
/// See `Timer` documentation for some implementation notes.
#[derive(Debug)]
pub(crate) struct Wheel<T> {
/// The number of milliseconds elapsed since the wheel started.
elapsed: u64,
/// Timer wheel.
///
/// Levels:
///
/// * 1 ms slots / 64 ms range
/// * 64 ms slots / ~ 4 sec range
/// * ~ 4 sec slots / ~ 4 min range
/// * ~ 4 min slots / ~ 4 hr range
/// * ~ 4 hr slots / ~ 12 day range
/// * ~ 12 day slots / ~ 2 yr range
levels: Vec<Level<T>>,
}
/// Number of levels. Each level has 64 slots. By using 6 levels with 64 slots
/// each, the timer is able to track time up to 2 years into the future with a
/// precision of 1 millisecond.
const NUM_LEVELS: usize = 6;
/// The maximum duration of a delay
const MAX_DURATION: u64 = 1 << (6 * NUM_LEVELS);
#[derive(Debug)]
pub(crate) enum InsertError {
Elapsed,
Invalid,
}
/// Poll expirations from the wheel
#[derive(Debug, Default)]
pub(crate) struct Poll {
now: u64,
expiration: Option<Expiration>,
}
impl<T> Wheel<T>
where
T: Stack,
{
/// Create a new timing wheel
pub(crate) fn new() -> Wheel<T> {
let levels = (0..NUM_LEVELS).map(Level::new).collect();
Wheel { elapsed: 0, levels }
}
/// Return the number of milliseconds that have elapsed since the timing
/// wheel's creation.
pub(crate) fn elapsed(&self) -> u64 {
self.elapsed
}
/// Insert an entry into the timing 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.
///
/// `Err(Elapsed)` indicates that `when` represents an instant that has
/// already passed. In this case, the caller should fire the timeout
/// immediately.
///
/// `Err(Invalid)` indicates an invalid `when` argument as been supplied.
pub(crate) fn insert(
&mut self,
when: u64,
item: T::Owned,
store: &mut T::Store,
) -> Result<(), (T::Owned, InsertError)> {
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, store);
debug_assert!({
self.levels[level]
.next_expiration(self.elapsed)
.map(|e| e.deadline >= self.elapsed)
.unwrap_or(true)
});
Ok(())
}
/// Remove `item` from thee timing wheel.
pub(crate) fn remove(&mut self, item: &T::Borrowed, store: &mut T::Store) {
let when = T::when(item, store);
let level = self.level_for(when);
self.levels[level].remove_entry(when, item, store);
}
/// Instant at which to poll
pub(crate) fn poll_at(&self) -> Option<u64> {
self.next_expiration().map(|expiration| expiration.deadline)
}
pub(crate) fn poll(&mut self, poll: &mut Poll, store: &mut T::Store) -> Option<T::Owned> {
loop {
if poll.expiration.is_none() {
poll.expiration = self.next_expiration().and_then(|expiration| {
if expiration.deadline > poll.now {
None
} else {
Some(expiration)
}
});
}
match poll.expiration {
Some(ref expiration) => {
if let Some(item) = self.poll_expiration(expiration, store) {
return Some(item);
}
self.set_elapsed(expiration.deadline);
}
None => {
self.set_elapsed(poll.now);
return None;
}
}
poll.expiration = None;
}
}
/// Returns the instant at which the next timeout expires.
fn next_expiration(&self) -> Option<Expiration> {
// Check all levels
for level in 0..NUM_LEVELS {
if let Some(expiration) = self.levels[level].next_expiration(self.elapsed) {
// There cannot be any expirations at a higher level that happen
// before this one.
debug_assert!({
let mut res = true;
for l2 in (level + 1)..NUM_LEVELS {
if let Some(e2) = self.levels[l2].next_expiration(self.elapsed) {
if e2.deadline < expiration.deadline {
res = false;
}
}
}
res
});
return Some(expiration);
}
}
None
}
pub(crate) fn poll_expiration(
&mut self,
expiration: &Expiration,
store: &mut T::Store,
) -> Option<T::Owned> {
while let Some(item) = self.pop_entry(expiration, store) {
if expiration.level == 0 {
debug_assert_eq!(T::when(item.borrow(), store), expiration.deadline);
return Some(item);
} else {
let when = T::when(item.borrow(), store);
let next_level = expiration.level - 1;
self.levels[next_level].add_entry(when, item, store);
}
}
None
}
fn set_elapsed(&mut self, when: u64) {
assert!(
self.elapsed <= when,
"elapsed={:?}; when={:?}",
self.elapsed,
when
);
if when > self.elapsed {
self.elapsed = when;
}
}
fn pop_entry(&mut self, expiration: &Expiration, store: &mut T::Store) -> Option<T::Owned> {
self.levels[expiration.level].pop_entry_slot(expiration.slot, store)
}
fn level_for(&self, when: u64) -> usize {
level_for(self.elapsed, when)
}
}
fn level_for(elapsed: u64, when: u64) -> usize {
let masked = elapsed ^ when;
assert!(masked != 0, "elapsed={}; when={}", elapsed, when);
let leading_zeros = masked.leading_zeros() as usize;
let significant = 63 - leading_zeros;
significant / 6
}
impl Poll {
pub(crate) fn new(now: u64) -> Poll {
Poll {
now,
expiration: None,
}
}
}
#[cfg(test)]
mod test {
use super::*;
#[test]
fn test_level_for() {
for pos in 1..64 {
assert_eq!(
0,
level_for(0, pos),
"level_for({}) -- binary = {:b}",
pos,
pos
);
}
for level in 1..5 {
for pos in level..64 {
let a = pos * 64_usize.pow(level as u32);
assert_eq!(
level,
level_for(0, a as u64),
"level_for({}) -- binary = {:b}",
a,
a
);
if pos > level {
let a = a - 1;
assert_eq!(
level,
level_for(0, a as u64),
"level_for({}) -- binary = {:b}",
a,
a
);
}
if pos < 64 {
let a = a + 1;
assert_eq!(
level,
level_for(0, a as u64),
"level_for({}) -- binary = {:b}",
a,
a
);
}
}
}
}
}
+26
View File
@@ -0,0 +1,26 @@
use std::borrow::Borrow;
/// Abstracts the stack operations needed to track timeouts.
pub(crate) trait Stack: Default {
/// Type of the item stored in the stack
type Owned: Borrow<Self::Borrowed>;
/// Borrowed item
type Borrowed;
/// Item storage, this allows a slab to be used instead of just the heap
type Store;
/// Returns `true` if the stack is empty
fn is_empty(&self) -> bool;
/// Push an item onto the stack
fn push(&mut self, item: Self::Owned, store: &mut Self::Store);
/// Pop an item from the stack
fn pop(&mut self, store: &mut Self::Store) -> Option<Self::Owned>;
fn remove(&mut self, item: &Self::Borrowed, store: &mut Self::Store);
fn when(item: &Self::Borrowed, store: &Self::Store) -> u64;
}
+2 -4
View File
@@ -1,17 +1,15 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "default")]
use tokio::runtime::{self, current_thread};
use tokio::timer::clock::Clock;
use tokio::timer::*;
use tokio_timer;
use tokio_timer::clock::Clock;
use std::sync::mpsc;
use std::time::{Duration, Instant};
struct MockNow(Instant);
impl tokio_timer::clock::Now for MockNow {
impl tokio::timer::clock::Now for MockNow {
fn now(&self) -> Instant {
self.0
}
+48
View File
@@ -0,0 +1,48 @@
#![warn(rust_2018_idioms)]
use tokio::timer::clock;
use tokio::timer::clock::*;
use std::time::Instant;
struct ConstNow(Instant);
impl Now for ConstNow {
fn now(&self) -> Instant {
self.0
}
}
#[test]
fn default_clock() {
let a = Instant::now();
let b = clock::now();
let c = Clock::new().now();
assert!(a <= b);
assert!(b <= c);
}
#[test]
fn custom_clock() {
let now = ConstNow(Instant::now());
let clock = Clock::new_with_now(now);
let a = Instant::now();
let b = clock.now();
assert!(b <= a);
}
#[test]
fn execution_context() {
let now = ConstNow(Instant::now());
let clock = Clock::new_with_now(now);
with_default(&clock, || {
let a = Instant::now();
let b = clock::now();
assert!(b <= a);
});
}
+538
View File
@@ -0,0 +1,538 @@
#![warn(rust_2018_idioms)]
use tokio::timer::delay;
use tokio::timer::timer::Handle;
use tokio_test::task::MockTask;
use tokio_test::{assert_pending, assert_ready, clock};
use std::time::{Duration, Instant};
#[test]
fn immediate_delay() {
let mut task = MockTask::new();
clock::mock(|clock| {
// Create `Delay` that elapsed immediately.
let mut fut = delay(clock.now());
// Ready!
assert_ready!(task.poll(&mut fut));
// Turn the timer, it runs for the elapsed time
clock.turn_for(ms(1000));
// The time has not advanced. The `turn` completed immediately.
assert_eq!(clock.advanced(), ms(1000));
});
}
#[test]
fn delayed_delay_level_0() {
let mut task = MockTask::new();
for &i in &[1, 10, 60] {
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(i));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_eq!(clock.advanced(), ms(i));
assert_ready!(task.poll(&mut fut));
});
}
}
#[test]
fn sub_ms_delayed_delay() {
let mut task = MockTask::new();
clock::mock(|clock| {
for _ in 0..5 {
let deadline = clock.now() + Duration::from_millis(1) + Duration::new(0, 1);
let mut fut = delay(deadline);
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_ready!(task.poll(&mut fut));
assert!(clock.now() >= deadline);
clock.advance(Duration::new(0, 1));
}
});
}
#[test]
fn delayed_delay_wrapping_level_0() {
let mut task = MockTask::new();
clock::mock(|clock| {
clock.turn_for(ms(5));
assert_eq!(clock.advanced(), ms(5));
let mut fut = delay(clock.now() + ms(60));
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_eq!(clock.advanced(), ms(64));
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_eq!(clock.advanced(), ms(65));
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn timer_wrapping_with_higher_levels() {
let mut task = MockTask::new();
clock::mock(|clock| {
// Set delay to hit level 1
let mut s1 = delay(clock.now() + ms(64));
assert_pending!(task.poll(&mut s1));
// Turn a bit
clock.turn_for(ms(5));
// Set timeout such that it will hit level 0, but wrap
let mut s2 = delay(clock.now() + ms(60));
assert_pending!(task.poll(&mut s2));
// This should result in s1 firing
clock.turn();
assert_eq!(clock.advanced(), ms(64));
assert_ready!(task.poll(&mut s1));
assert_pending!(task.poll(&mut s2));
clock.turn();
assert_eq!(clock.advanced(), ms(65));
assert_ready!(task.poll(&mut s2));
});
}
#[test]
fn delay_with_deadline_in_past() {
let mut task = MockTask::new();
clock::mock(|clock| {
// Create `Delay` that elapsed immediately.
let mut fut = delay(clock.now() - ms(100));
// Even though the delay expires in the past, it is not ready yet
// because the timer must observe it.
assert_ready!(task.poll(&mut fut));
// Turn the timer, it runs for the elapsed time
clock.turn_for(ms(1000));
// The time has not advanced. The `turn` completed immediately.
assert_eq!(clock.advanced(), ms(1000));
});
}
#[test]
fn delayed_delay_level_1() {
let mut task = MockTask::new();
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(234));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
// Turn the timer, this will wake up to cascade the timer down.
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(192));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
// Turn the timer again
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(234));
// The delay has elapsed.
assert_ready!(task.poll(&mut fut));
});
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(234));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
// Turn the timer with a smaller timeout than the cascade.
clock.turn_for(ms(100));
assert_eq!(clock.advanced(), ms(100));
assert_pending!(task.poll(&mut fut));
// Turn the timer, this will wake up to cascade the timer down.
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(192));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
// Turn the timer again
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(234));
// The delay has elapsed.
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn creating_delay_outside_of_context() {
let now = Instant::now();
// This creates a delay outside of the context of a mock timer. This tests
// that it will still expire.
let mut fut = delay(now + ms(500));
let mut task = MockTask::new();
clock::mock_at(now, |clock| {
// This registers the delay with the timer
assert_pending!(task.poll(&mut fut));
// Wait some time... the timer is cascading
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(448));
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(500));
// The delay has elapsed
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn concurrently_set_two_timers_second_one_shorter() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
clock::mock(|clock| {
let mut fut1 = delay(clock.now() + ms(500));
let mut fut2 = delay(clock.now() + ms(200));
// The delay has not elapsed
assert_pending!(t1.poll(&mut fut1));
assert_pending!(t2.poll(&mut fut2));
// Delay until a cascade
clock.turn();
assert_eq!(clock.advanced(), ms(192));
// Delay until the second timer.
clock.turn();
assert_eq!(clock.advanced(), ms(200));
// The shorter delay fires
assert_ready!(t2.poll(&mut fut2));
assert_pending!(t1.poll(&mut fut1));
clock.turn();
assert_eq!(clock.advanced(), ms(448));
assert_pending!(t1.poll(&mut fut1));
// Turn again, this time the time will advance to the second delay
clock.turn();
assert_eq!(clock.advanced(), ms(500));
assert_ready!(t1.poll(&mut fut1));
})
}
#[test]
fn short_delay() {
let mut task = MockTask::new();
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(1));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
// Turn the timer, but not enough time will go by.
clock.turn();
// The delay has elapsed.
assert_ready!(task.poll(&mut fut));
// The time has advanced to the point of the delay elapsing.
assert_eq!(clock.advanced(), ms(1));
})
}
#[test]
fn sorta_long_delay() {
const MIN_5: u64 = 5 * 60 * 1000;
let mut task = MockTask::new();
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(MIN_5));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
let cascades = &[262_144, 262_144 + 9 * 4096, 262_144 + 9 * 4096 + 15 * 64];
for &elapsed in cascades {
clock.turn();
assert_eq!(clock.advanced(), ms(elapsed));
assert_pending!(task.poll(&mut fut));
}
clock.turn();
assert_eq!(clock.advanced(), ms(MIN_5));
// The delay has elapsed.
assert_ready!(task.poll(&mut fut));
})
}
#[test]
fn very_long_delay() {
const MO_5: u64 = 5 * 30 * 24 * 60 * 60 * 1000;
let mut task = MockTask::new();
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(MO_5));
// The delay has not elapsed.
assert_pending!(task.poll(&mut fut));
let cascades = &[
12_884_901_888,
12_952_010_752,
12_959_875_072,
12_959_997_952,
];
for &elapsed in cascades {
clock.turn();
assert_eq!(clock.advanced(), ms(elapsed));
assert_pending!(task.poll(&mut fut));
}
// Turn the timer, but not enough time will go by.
clock.turn();
// The time has advanced to the point of the delay elapsing.
assert_eq!(clock.advanced(), ms(MO_5));
// The delay has elapsed.
assert_ready!(task.poll(&mut fut));
})
}
#[test]
#[should_panic]
fn greater_than_max() {
const YR_5: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
let mut task = MockTask::new();
clock::mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = delay(clock.now() + ms(YR_5));
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(0));
// boom
let _ = task.poll(&mut fut);
})
}
#[test]
fn unpark_is_delayed() {
let mut t1 = MockTask::new();
let mut t2 = MockTask::new();
let mut t3 = MockTask::new();
clock::mock(|clock| {
let mut fut1 = delay(clock.now() + ms(100));
let mut fut2 = delay(clock.now() + ms(101));
let mut fut3 = delay(clock.now() + ms(200));
assert_pending!(t1.poll(&mut fut1));
assert_pending!(t2.poll(&mut fut2));
assert_pending!(t3.poll(&mut fut3));
clock.park_for(ms(500));
assert_eq!(clock.advanced(), ms(500));
assert_ready!(t1.poll(&mut fut1));
assert_ready!(t2.poll(&mut fut2));
assert_ready!(t3.poll(&mut fut3));
})
}
#[test]
fn set_timeout_at_deadline_greater_than_max_timer() {
const YR_1: u64 = 365 * 24 * 60 * 60 * 1000;
const YR_5: u64 = 5 * YR_1;
let mut task = MockTask::new();
clock::mock(|clock| {
for _ in 0..5 {
clock.turn_for(ms(YR_1));
}
let mut fut = delay(clock.now() + ms(1));
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(YR_5) + ms(1));
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn reset_future_delay_before_fire() {
let mut task = MockTask::new();
clock::mock(|clock| {
let mut fut = delay(clock.now() + ms(100));
assert_pending!(task.poll(&mut fut));
fut.reset(clock.now() + ms(200));
clock.turn();
assert_eq!(clock.advanced(), ms(192));
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_eq!(clock.advanced(), ms(200));
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn reset_past_delay_before_turn() {
let mut task = MockTask::new();
clock::mock(|clock| {
let mut fut = delay(clock.now() + ms(100));
assert_pending!(task.poll(&mut fut));
fut.reset(clock.now() + ms(80));
clock.turn();
assert_eq!(clock.advanced(), ms(64));
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_eq!(clock.advanced(), ms(80));
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn reset_past_delay_before_fire() {
let mut task = MockTask::new();
clock::mock(|clock| {
let mut fut = delay(clock.now() + ms(100));
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(10));
assert_pending!(task.poll(&mut fut));
fut.reset(clock.now() + ms(80));
clock.turn();
assert_eq!(clock.advanced(), ms(64));
assert_pending!(task.poll(&mut fut));
clock.turn();
assert_eq!(clock.advanced(), ms(90));
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn reset_future_delay_after_fire() {
let mut task = MockTask::new();
clock::mock(|clock| {
let mut fut = delay(clock.now() + ms(100));
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(64));
clock.turn();
assert_eq!(clock.advanced(), ms(100));
assert_ready!(task.poll(&mut fut));
fut.reset(clock.now() + ms(10));
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(110));
assert_ready!(task.poll(&mut fut));
});
}
#[test]
fn delay_with_default_handle() {
let handle = Handle::default();
let now = Instant::now();
let mut task = MockTask::new();
let mut fut = handle.delay(now + ms(1));
clock::mock_at(now, |clock| {
assert_pending!(task.poll(&mut fut));
clock.turn_for(ms(1));
assert_ready!(task.poll(&mut fut));
});
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
+243
View File
@@ -0,0 +1,243 @@
#![warn(rust_2018_idioms)]
use tokio::timer::{Delay, Timer};
use tokio_executor::current_thread::CurrentThread;
use tokio_executor::park::{Park, Unpark, UnparkThread};
use rand;
use rand::Rng;
use std::cmp;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::{Arc, Barrier};
use std::task::{Context, Poll};
use std::thread;
use std::time::{Duration, Instant};
struct Signal {
rem: AtomicUsize,
unpark: UnparkThread,
}
#[test]
fn hammer_complete() {
const ITERS: usize = 5;
const THREADS: usize = 4;
const PER_THREAD: usize = 40;
const MIN_DELAY: u64 = 1;
const MAX_DELAY: u64 = 5_000;
for _ in 0..ITERS {
let mut timer = Timer::default();
let handle = timer.handle();
let barrier = Arc::new(Barrier::new(THREADS));
let done = Arc::new(Signal {
rem: AtomicUsize::new(THREADS),
unpark: timer.get_park().unpark(),
});
for _ in 0..THREADS {
let handle = handle.clone();
let barrier = barrier.clone();
let done = done.clone();
thread::spawn(move || {
let mut exec = CurrentThread::new();
let mut rng = rand::thread_rng();
barrier.wait();
for _ in 0..PER_THREAD {
let deadline =
Instant::now() + Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY));
let delay = handle.delay(deadline);
exec.spawn(async move {
delay.await;
let now = Instant::now();
assert!(now >= deadline, "deadline greater by {:?}", deadline - now);
});
}
// Run the logic
exec.run().unwrap();
if 1 == done.rem.fetch_sub(1, SeqCst) {
done.unpark.unpark();
}
});
}
while done.rem.load(SeqCst) > 0 {
timer.turn(None).unwrap();
}
}
}
#[test]
fn hammer_cancel() {
const ITERS: usize = 5;
const THREADS: usize = 4;
const PER_THREAD: usize = 40;
const MIN_DELAY: u64 = 1;
const MAX_DELAY: u64 = 5_000;
for _ in 0..ITERS {
let mut timer = Timer::default();
let handle = timer.handle();
let barrier = Arc::new(Barrier::new(THREADS));
let done = Arc::new(Signal {
rem: AtomicUsize::new(THREADS),
unpark: timer.get_park().unpark(),
});
for _ in 0..THREADS {
let handle = handle.clone();
let barrier = barrier.clone();
let done = done.clone();
thread::spawn(move || {
let mut exec = CurrentThread::new();
let mut rng = rand::thread_rng();
barrier.wait();
for _ in 0..PER_THREAD {
let timeout1 = Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY));
let timeout2 = Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY));
let deadline = Instant::now() + cmp::min(timeout1, timeout2);
let delay = handle.delay(Instant::now() + timeout1);
let join = handle.timeout(delay, timeout2);
exec.spawn(async move {
let _ = join.await;
let now = Instant::now();
assert!(now >= deadline, "deadline greater by {:?}", deadline - now);
});
}
// Run the logic
exec.run().unwrap();
if 1 == done.rem.fetch_sub(1, SeqCst) {
done.unpark.unpark();
}
});
}
while done.rem.load(SeqCst) > 0 {
timer.turn(None).unwrap();
}
}
}
#[test]
fn hammer_reset() {
const ITERS: usize = 5;
const THREADS: usize = 4;
const PER_THREAD: usize = 40;
const MIN_DELAY: u64 = 1;
const MAX_DELAY: u64 = 250;
for _ in 0..ITERS {
let mut timer = Timer::default();
let handle = timer.handle();
let barrier = Arc::new(Barrier::new(THREADS));
let done = Arc::new(Signal {
rem: AtomicUsize::new(THREADS),
unpark: timer.get_park().unpark(),
});
for _ in 0..THREADS {
let handle = handle.clone();
let barrier = barrier.clone();
let done = done.clone();
thread::spawn(move || {
let mut exec = CurrentThread::new();
let mut rng = rand::thread_rng();
barrier.wait();
for _ in 0..PER_THREAD {
let deadline1 =
Instant::now() + Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY));
let deadline2 =
deadline1 + Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY));
let deadline3 =
deadline2 + Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY));
struct Select {
a: Option<Delay>,
b: Option<Delay>,
}
impl Future for Select {
type Output = Delay;
fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Self::Output> {
let res = Pin::new(self.a.as_mut().unwrap()).poll(cx);
if res.is_ready() {
return Poll::Ready(self.a.take().unwrap());
}
let res = Pin::new(self.b.as_mut().unwrap()).poll(cx);
if res.is_ready() {
return Poll::Ready(self.b.take().unwrap());
}
Poll::Pending
}
}
let s = Select {
a: Some(handle.delay(deadline1)),
b: Some(handle.delay(deadline2)),
};
exec.spawn(async move {
let mut delay = s.await;
let now = Instant::now();
assert!(
now >= deadline1,
"deadline greater by {:?}",
deadline1 - now
);
delay.reset(deadline3);
delay.await;
});
}
// Run the logic
exec.run().unwrap();
if 1 == done.rem.fetch_sub(1, SeqCst) {
done.unpark.unpark();
}
});
}
while done.rem.load(SeqCst) > 0 {
timer.turn(None).unwrap();
}
}
}
+54
View File
@@ -0,0 +1,54 @@
#![warn(rust_2018_idioms)]
use tokio::timer::*;
use tokio_test::task::MockTask;
use tokio_test::{assert_pending, assert_ready_eq, clock};
use std::time::Duration;
#[test]
#[should_panic]
fn interval_zero_duration() {
clock::mock(|clock| {
let _ = Interval::new(clock.now(), ms(0));
});
}
#[test]
fn usage() {
let mut task = MockTask::new();
clock::mock(|clock| {
let start = clock.now();
let mut int = Interval::new(start, ms(300));
macro_rules! poll {
() => {
task.enter(|cx| int.poll_next(cx))
};
}
assert_ready_eq!(poll!(), Some(start));
assert_pending!(poll!());
clock.advance(ms(100));
assert_pending!(poll!());
clock.advance(ms(200));
assert_ready_eq!(poll!(), Some(start + ms(300)));
assert_pending!(poll!());
clock.advance(ms(400));
assert_ready_eq!(poll!(), Some(start + ms(600)));
assert_pending!(poll!());
clock.advance(ms(500));
assert_ready_eq!(poll!(), Some(start + ms(900)));
assert_ready_eq!(poll!(), Some(start + ms(1200)));
assert_pending!(poll!());
});
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
+401
View File
@@ -0,0 +1,401 @@
#![warn(rust_2018_idioms)]
use tokio::timer::*;
use tokio_test::task::MockTask;
use tokio_test::{assert_ok, assert_pending, assert_ready, clock};
use std::time::Duration;
macro_rules! poll {
($task:ident, $queue:ident) => {
$task.enter(|cx| $queue.poll_next(cx))
};
}
macro_rules! assert_ready_ok {
($e:expr) => {{
assert_ok!(match assert_ready!($e) {
Some(v) => v,
None => panic!("None"),
})
}};
}
#[test]
fn single_immediate_delay() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let _key = queue.insert_at("foo", clock.now());
let entry = assert_ready_ok!(poll!(t, queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(t, queue));
assert!(entry.is_none())
});
}
#[test]
fn multi_immediate_delays() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let _k = queue.insert_at("1", clock.now());
let _k = queue.insert_at("2", clock.now());
let _k = queue.insert_at("3", clock.now());
let mut res = vec![];
while res.len() < 3 {
let entry = assert_ready_ok!(poll!(t, queue));
res.push(entry.into_inner());
}
let entry = assert_ready!(poll!(t, queue));
assert!(entry.is_none());
res.sort();
assert_eq!("1", res[0]);
assert_eq!("2", res[1]);
assert_eq!("3", res[2]);
});
}
#[test]
fn single_short_delay() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let _key = queue.insert_at("foo", clock.now() + ms(5));
assert_pending!(poll!(t, queue));
clock.turn_for(ms(1));
assert!(!t.is_woken());
clock.turn_for(ms(5));
assert!(t.is_woken());
let entry = assert_ready_ok!(poll!(t, queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(t, queue));
assert!(entry.is_none());
});
}
#[test]
fn multi_delay_at_start() {
let long = 262_144 + 9 * 4096;
let delays = &[1000, 2, 234, long, 60, 10];
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
// Setup the delays
for &i in delays {
let _key = queue.insert_at(i, clock.now() + ms(i));
}
assert_pending!(poll!(t, queue));
assert!(!t.is_woken());
for elapsed in 0..1200 {
clock.turn_for(ms(1));
let elapsed = elapsed + 1;
if delays.contains(&elapsed) {
assert!(t.is_woken());
assert_ready!(poll!(t, queue));
assert_pending!(poll!(t, queue));
} else {
if t.is_woken() {
let cascade = &[192, 960];
assert!(cascade.contains(&elapsed), "elapsed={}", elapsed);
assert_pending!(poll!(t, queue));
}
}
}
});
}
#[test]
fn insert_in_past_fires_immediately() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let now = clock.now();
clock.turn_for(ms(10));
queue.insert_at("foo", now);
assert_ready!(poll!(t, queue));
});
}
#[test]
fn remove_entry() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let key = queue.insert_at("foo", clock.now() + ms(5));
assert_pending!(poll!(t, queue));
let entry = queue.remove(&key);
assert_eq!(entry.into_inner(), "foo");
clock.turn_for(ms(10));
let entry = assert_ready!(poll!(t, queue));
assert!(entry.is_none());
});
}
#[test]
fn reset_entry() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let now = clock.now();
let key = queue.insert_at("foo", now + ms(5));
assert_pending!(poll!(t, queue));
clock.turn_for(ms(1));
queue.reset_at(&key, now + ms(10));
assert_pending!(poll!(t, queue));
clock.turn_for(ms(7));
assert!(!t.is_woken());
assert_pending!(poll!(t, queue));
clock.turn_for(ms(3));
assert!(t.is_woken());
let entry = assert_ready_ok!(poll!(t, queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(t, queue));
assert!(entry.is_none())
});
}
#[test]
fn reset_much_later() {
let mut t = MockTask::new();
// Reproduces tokio-rs/tokio#849.
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
clock.turn_for(ms(1));
let key = queue.insert_at("foo", epoch + ms(200));
assert_pending!(poll!(t, queue));
clock.turn_for(ms(3));
queue.reset_at(&key, epoch + ms(5));
clock.turn_for(ms(20));
assert!(t.is_woken());
});
}
#[test]
fn reset_twice() {
let mut t = MockTask::new();
// Reproduces tokio-rs/tokio#849.
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
clock.turn_for(ms(1));
let key = queue.insert_at("foo", epoch + ms(200));
assert_pending!(poll!(t, queue));
clock.turn_for(ms(3));
queue.reset_at(&key, epoch + ms(50));
clock.turn_for(ms(20));
queue.reset_at(&key, epoch + ms(40));
clock.turn_for(ms(20));
assert!(t.is_woken());
});
}
#[test]
fn remove_expired_item() {
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let now = clock.now();
clock.turn_for(ms(10));
let key = queue.insert_at("foo", now);
let entry = queue.remove(&key);
assert_eq!(entry.into_inner(), "foo");
})
}
#[test]
fn expires_before_last_insert() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
queue.insert_at("foo", epoch + ms(10_000));
// Delay should be set to 8.192s here.
assert_pending!(poll!(t, queue));
// Delay should be set to the delay of the new item here
queue.insert_at("bar", epoch + ms(600));
assert_pending!(poll!(t, queue));
clock.advance(ms(600));
assert!(t.is_woken());
let entry = assert_ready_ok!(poll!(t, queue)).into_inner();
assert_eq!(entry, "bar");
})
}
#[test]
fn multi_reset() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
let foo = queue.insert_at("foo", epoch + ms(200));
let bar = queue.insert_at("bar", epoch + ms(250));
assert_pending!(poll!(t, queue));
queue.reset_at(&foo, epoch + ms(300));
queue.reset_at(&bar, epoch + ms(350));
queue.reset_at(&foo, epoch + ms(400));
})
}
#[test]
fn expire_first_key_when_reset_to_expire_earlier() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
let foo = queue.insert_at("foo", epoch + ms(200));
queue.insert_at("bar", epoch + ms(250));
assert_pending!(poll!(t, queue));
queue.reset_at(&foo, epoch + ms(100));
clock.advance(ms(100));
assert!(t.is_woken());
let entry = assert_ready_ok!(poll!(t, queue)).into_inner();
assert_eq!(entry, "foo");
})
}
#[test]
fn expire_second_key_when_reset_to_expire_earlier() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
queue.insert_at("foo", epoch + ms(200));
let bar = queue.insert_at("bar", epoch + ms(250));
assert_pending!(poll!(t, queue));
queue.reset_at(&bar, epoch + ms(100));
clock.advance(ms(100));
assert!(t.is_woken());
let entry = assert_ready_ok!(poll!(t, queue)).into_inner();
assert_eq!(entry, "bar");
})
}
#[test]
fn reset_first_expiring_item_to_expire_later() {
let mut t = MockTask::new();
clock::mock(|clock| {
let mut queue = DelayQueue::new();
let epoch = clock.now();
let foo = queue.insert_at("foo", epoch + ms(200));
let _bar = queue.insert_at("bar", epoch + ms(250));
assert_pending!(poll!(t, queue));
queue.reset_at(&foo, epoch + ms(300));
clock.advance(ms(250));
assert!(t.is_woken());
let entry = assert_ready_ok!(poll!(t, queue)).into_inner();
assert_eq!(entry, "bar");
})
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
@@ -1,5 +1,4 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "default")]
use tokio::prelude::*;
use tokio::timer::*;
+67
View File
@@ -0,0 +1,67 @@
#![warn(rust_2018_idioms)]
use tokio::sync::mpsc;
use tokio::timer::throttle::Throttle;
use tokio_test::task::MockTask;
use tokio_test::{assert_pending, assert_ready_eq, clock};
use futures_core::Stream;
use std::time::Duration;
macro_rules! poll {
($task:ident, $stream:ident) => {{
use std::pin::Pin;
$task.enter(|cx| Pin::new(&mut $stream).poll_next(cx))
}};
}
#[test]
fn throttle() {
let mut t = MockTask::new();
clock::mock(|clock| {
let (mut tx, rx) = mpsc::unbounded_channel();
let mut stream = Throttle::new(rx, ms(1));
assert_pending!(poll!(t, stream));
for i in 0..3 {
tx.try_send(i).unwrap();
}
for i in 0..3 {
assert_ready_eq!(poll!(t, stream), Some(i));
assert_pending!(poll!(t, stream));
clock.advance(ms(1));
}
assert_pending!(poll!(t, stream));
});
}
#[test]
fn throttle_dur_0() {
let mut t = MockTask::new();
clock::mock(|_| {
let (mut tx, rx) = mpsc::unbounded_channel();
let mut stream = Throttle::new(rx, ms(0));
assert_pending!(poll!(t, stream));
for i in 0..3 {
tx.try_send(i).unwrap();
}
for i in 0..3 {
assert_ready_eq!(poll!(t, stream), Some(i));
}
assert_pending!(poll!(t, stream));
});
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
+205
View File
@@ -0,0 +1,205 @@
#![warn(rust_2018_idioms)]
use tokio::sync::oneshot;
use tokio::timer::*;
use tokio_test::task::MockTask;
use tokio_test::{
assert_err, assert_pending, assert_ready, assert_ready_err, assert_ready_ok, clock,
};
use std::time::Duration;
#[test]
fn simultaneous_deadline_future_completion() {
let mut t = MockTask::new();
clock::mock(|clock| {
// Create a future that is immediately ready
let fut = Box::pin(Timeout::new_at(async {}, clock.now()));
// Ready!
assert_ready_ok!(t.poll(fut));
});
}
#[test]
fn completed_future_past_deadline() {
let mut t = MockTask::new();
clock::mock(|clock| {
// Wrap it with a deadline
let fut = Timeout::new_at(async {}, clock.now() - ms(1000));
let fut = Box::pin(fut);
// Ready!
assert_ready_ok!(t.poll(fut));
});
}
#[test]
fn future_and_deadline_in_future() {
let mut t = MockTask::new();
clock::mock(|clock| {
// Not yet complete
let (tx, rx) = oneshot::channel();
// Wrap it with a deadline
let mut fut = Timeout::new_at(rx, clock.now() + ms(100));
assert_pending!(t.poll(&mut fut));
// Turn the timer, it runs for the elapsed time
clock.advance(ms(90));
assert_pending!(t.poll(&mut fut));
// Complete the future
tx.send(()).unwrap();
assert_ready_ok!(t.poll(&mut fut)).unwrap();
});
}
#[test]
fn future_and_timeout_in_future() {
let mut t = MockTask::new();
clock::mock(|clock| {
// Not yet complete
let (tx, rx) = oneshot::channel();
// Wrap it with a deadline
let mut fut = Timeout::new(rx, ms(100));
// Ready!
assert_pending!(t.poll(&mut fut));
// Turn the timer, it runs for the elapsed time
clock.advance(ms(90));
assert_pending!(t.poll(&mut fut));
// Complete the future
tx.send(()).unwrap();
assert_ready_ok!(t.poll(&mut fut)).unwrap();
});
}
struct Empty;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
impl Future for Empty {
type Output = ();
fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
Poll::Pending
}
}
#[test]
fn deadline_now_elapses() {
let mut t = MockTask::new();
clock::mock(|clock| {
// Wrap it with a deadline
let mut fut = Timeout::new_at(Empty, clock.now());
assert_ready_err!(t.poll(&mut fut));
});
}
#[test]
fn deadline_future_elapses() {
let mut t = MockTask::new();
clock::mock(|clock| {
// Wrap it with a deadline
let mut fut = Timeout::new_at(Empty, clock.now() + ms(300));
assert_pending!(t.poll(&mut fut));
clock.advance(ms(300));
assert_ready_err!(t.poll(&mut fut));
});
}
#[cfg(feature = "async-traits")]
macro_rules! poll {
($task:ident, $stream:ident) => {{
use futures_core::Stream;
$task.enter(|cx| Pin::new(&mut $stream).poll_next(cx))
}};
}
#[test]
#[cfg(feature = "async-traits")]
fn stream_and_timeout_in_future() {
use tokio_sync::mpsc;
let mut t = MockTask::new();
clock::mock(|clock| {
// Not yet complete
let (mut tx, rx) = mpsc::unbounded_channel();
// Wrap it with a deadline
let mut stream = Timeout::new(rx, ms(100));
// Not ready
assert_pending!(poll!(t, stream));
// Turn the timer, it runs for the elapsed time
clock.advance(ms(90));
assert_pending!(poll!(t, stream));
// Complete the future
tx.try_send(()).unwrap();
let item = assert_ready!(poll!(t, stream));
assert!(item.is_some());
});
}
#[test]
#[cfg(feature = "async-traits")]
fn idle_stream_timesout_periodically() {
use tokio_sync::mpsc;
let mut t = MockTask::new();
clock::mock(|clock| {
// Not yet complete
let (_tx, rx) = mpsc::unbounded_channel::<()>();
// Wrap it with a deadline
let mut stream = Timeout::new(rx, ms(100));
// Not ready
assert_pending!(poll!(t, stream));
// Turn the timer, it runs for the elapsed time
clock.advance(ms(100));
let v = assert_ready!(poll!(t, stream)).unwrap();
assert_err!(v);
// Stream's timeout should reset
assert_pending!(poll!(t, stream));
// Turn the timer, it runs for the elapsed time
clock.advance(ms(100));
let v = assert_ready!(poll!(t, stream)).unwrap();
assert_err!(v)
});
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}