diff options
Diffstat (limited to 'third_party/rust/tokio-timer')
36 files changed, 6144 insertions, 0 deletions
diff --git a/third_party/rust/tokio-timer/.cargo-checksum.json b/third_party/rust/tokio-timer/.cargo-checksum.json new file mode 100644 index 0000000000..d9e39669de --- /dev/null +++ b/third_party/rust/tokio-timer/.cargo-checksum.json @@ -0,0 +1 @@ +{"files":{"CHANGELOG.md":"f74e4c4af8f7858910c0dc3875ddcc2f5e6689771a05e79c7afedc41cc98a426","Cargo.toml":"e6950768269656266f50c038e428f67a206462eda5ce024c94b3ceb11bae04a8","LICENSE":"898b1ae9821e98daf8964c8d6c7f61641f5f5aa78ad500020771c0939ee0dea1","README.md":"62860ef13cb86f22cea46c2910a0ce3cc463e9ebcf79d3133694ea24ec9553f0","src/atomic.rs":"4fc9d8c28f76b29be3e6ecd8487e82299b859b6eb86f569503c751ccb5382b34","src/clock/clock.rs":"79e89d13e33d63825f78a3be91b9b1b27de0a2dcf40cc86d59a24794415a763c","src/clock/mod.rs":"b4503672784d133e23a95c0e87237ac6959c87bd249f9a8c38ac71c9db963baa","src/clock/now.rs":"ba278929aab20358545f111228ef32b6b69203bb4ba45483ef35111b0f765f2b","src/deadline.rs":"520661e3d3e7f05d6946cae501e3ad4a0363812243abde12f0098159a74c4d74","src/delay.rs":"745c46b296d372653cabfd8ec19e74dae3e4c224f497778f570035feabf3fb0d","src/delay_queue.rs":"626bf9174eb7b989b4341ef2d30f87871b6cac5f5fab61d2b439122c686798e2","src/error.rs":"542c6364dfc81d710a5ba9da51156610c1d64a9467c62eda4f473c715e7e997b","src/interval.rs":"0a65da9bedfc1489c49867ac0d0059bc84941dfa2886905d3f1a364af57d78ab","src/lib.rs":"4ba9bfac3d0deb3f5f95201de34e5d6cd2ef2552ba4d135055bfdd842627296e","src/throttle.rs":"f1ac99ab5feec26c3e1ced0f442741397d8ef175bba987730e24e9b25a75b5dd","src/timeout.rs":"1a999edaa35d53340292d43cd768f292ac43177b256c7304418f417fb5a0c242","src/timer/atomic_stack.rs":"89edcaba3117c97d1a7e7a52a83f85a4f85422d743b159f31fa6ce9a6468ea0c","src/timer/entry.rs":"7626934cb4ecdcb2ed341a5a525a50af180af246e3d530fd5f40aa213a34d7d9","src/timer/handle.rs":"690110473d5e08f9d2481eb1c854c6c4b50b37acabd9ccddef45b3db27369b73","src/timer/mod.rs":"347993a46808b1bf2135ab5678aacc9e65d9d89d67a44e246cc8fa3bc6445005","src/timer/now.rs":"e876ff3dde4bc2849f375a2362e0298d72803cc32dc333a05696ac780488240e","src/timer/registration.rs":"9f5c7ad5a04e09dbee6446914c26c64582e40c3d01f4256836f3ddafda327393","src/timer/stack.rs":"8ef886ea48f1416256c591c3025f0cac3281393db5fb64d5997b27ef113ac07f","src/wheel/level.rs":"7d202187a7c2afac49dbd7c8ebec0d76dae62e4aa7d5fd6d24d3be0393ab9b6c","src/wheel/mod.rs":"91de01757843b09788841270a63ded7132ad26084e2fed0b1fa62299955cae90","src/wheel/stack.rs":"ac5fd0f6cac1bfdd9497b6eaa4f64beb816bc5605ecae463bd10a5e4c48e78c0","tests/clock.rs":"102180344fae71730d4628a30ec7d298a07a4fae5c1a711f0a8b274219eab362","tests/deadline.rs":"8364b7d8aae7a37b253e27ceeb9a5d80e2e46fbc887e16d48960b732d96166e5","tests/delay.rs":"3b3dba3eeac4d6a39b7c60af3efa63d93f26e1385dc765818732af29cabb1a19","tests/hammer.rs":"1e54fa8f91f9bfe78c66f4ba9a03832a0f8488fe2e521d2be48449e1ce3ce0c0","tests/interval.rs":"dfda478559278a46afb309393ab1fe0ede205d39bb0805ae3eec994c30c46d0c","tests/queue.rs":"9d5594b3e37feecb764187a992a19d418a8e10e4f8603f66d2057750190bc172","tests/support/mod.rs":"19efb8cdf608f3d1cbf9efd3d1608d6197f7fee386d00288a9dfe222fc89114d","tests/throttle.rs":"83e6977d7be434a20082a8cdb04c1dce87bd3e91aa9abfbd793912f257391e87","tests/timeout.rs":"0255896397215ef83764cd8f3b9da1aa10f2e453acf51d04efeee80427e418e4"},"package":"93044f2d313c95ff1cb7809ce9a7a05735b012288a888b62d4434fd58c94f296"}
\ No newline at end of file diff --git a/third_party/rust/tokio-timer/CHANGELOG.md b/third_party/rust/tokio-timer/CHANGELOG.md new file mode 100644 index 0000000000..63eaaf7752 --- /dev/null +++ b/third_party/rust/tokio-timer/CHANGELOG.md @@ -0,0 +1,84 @@ +# 0.2.13 (February 4, 2020) + +* Add `tokio 0.2.x` deprecation notice. + +# 0.2.12 (November 27, 2019) + +### Added +- `timer::set_default`, which functions like `timer::with_default`, but + returns a drop guard (#1725). +- `clock::set_default`, which functions like `clock::with_default`, but + returns a drop guard (#1725). + +# 0.2.11 (May 14, 2019) + +### Added +- `Handle::timeout` API, replacing the deprecated `Handle::deadline` (#1074). + +# 0.2.10 (February 4, 2019) + +### Fixed +- `DelayQueue` when multiple delays are reset (#871). + +# 0.2.9 (January 24, 2019) + +### Fixed +- `DelayQueue` timing logic when inserting / resetting a delay (#851, #863). +- Documentation links (#842, #844, #845) + +# 0.2.8 (November 21, 2018) + +* Implement throttle combinator (#736). +* Derive `Clone` for `delay_queue::Key` (#730). +* Bump internal dependencies (#753). + +# 0.2.7 (September 27, 2018) + +* Fix `Timeout` on error bug (#648). +* Miscellaneous documentation improvements. + +# 0.2.6 (August 23, 2018) + +* Implement `Default` for `timer::Handle` (#553) +* Provide `DelayQueue` utility (#550) +* Reduce size of `Delay` struct (#554) +* Introduce `Timeout`, deprecate `Deadline` (#558) + +# 0.2.5 (August 6, 2018) + +* Add `Interval::interval` shortcut (#492). + +# 0.2.4 (June 6, 2018) + +* Add `sleep` function for easy interval delays (#347). +* Provide `clock::now()`, a configurable source of time (#381). + +# 0.2.3 (May 2, 2018) + +* Improve parking semantics (#327). + +# 0.2.2 (Skipped due to failure in counting module) + +# 0.2.1 (April 2, 2018) + +* Fix build on 32-bit systems (#274). + +# 0.2.0 (March 30, 2018) + +* Rewrite from scratch using a hierarchical wheel strategy (#249). + +# 0.1.2 (Jun 27, 2017) + +* Allow naming timer thread. +* Track changes in dependencies. + +# 0.1.1 (Apr 6, 2017) + +* Set Rust v1.14 as the minimum supported version. +* Fix bug related to intervals. +* Impl `PartialEq + Eq` for TimerError. +* Add `Debug` implementations. + +# 0.1.0 (Jan 11, 2017) + +* Initial Release diff --git a/third_party/rust/tokio-timer/Cargo.toml b/third_party/rust/tokio-timer/Cargo.toml new file mode 100644 index 0000000000..61588bbdc9 --- /dev/null +++ b/third_party/rust/tokio-timer/Cargo.toml @@ -0,0 +1,41 @@ +# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO +# +# When uploading crates to the registry Cargo will automatically +# "normalize" Cargo.toml files for maximal compatibility +# with all versions of Cargo and also rewrite `path` dependencies +# to registry (e.g., crates.io) dependencies +# +# If you believe there's an error in this file please file an +# issue against the rust-lang/cargo repository. If you're +# editing this file be aware that the upstream Cargo.toml +# will likely look very different (and much more reasonable) + +[package] +name = "tokio-timer" +version = "0.2.13" +authors = ["Carl Lerche <me@carllerche.com>"] +description = "Timer facilities for Tokio\n" +homepage = "https://github.com/tokio-rs/tokio" +documentation = "https://docs.rs/tokio-timer/0.2.13/tokio_timer" +readme = "README.md" +license = "MIT" +repository = "https://github.com/tokio-rs/tokio" +[dependencies.crossbeam-utils] +version = "0.7.0" + +[dependencies.futures] +version = "0.1.19" + +[dependencies.slab] +version = "0.4.1" + +[dependencies.tokio-executor] +version = "0.1.1" +[dev-dependencies.rand] +version = "0.7" + +[dev-dependencies.tokio] +version = "0.1.7" + +[dev-dependencies.tokio-mock-task] +version = "0.1.0" diff --git a/third_party/rust/tokio-timer/LICENSE b/third_party/rust/tokio-timer/LICENSE new file mode 100644 index 0000000000..cdb28b4b56 --- /dev/null +++ b/third_party/rust/tokio-timer/LICENSE @@ -0,0 +1,25 @@ +Copyright (c) 2019 Tokio Contributors + +Permission is hereby granted, free of charge, to any +person obtaining a copy of this software and associated +documentation files (the "Software"), to deal in the +Software without restriction, including without +limitation the rights to use, copy, modify, merge, +publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software +is furnished to do so, subject to the following +conditions: + +The above copyright notice and this permission notice +shall be included in all copies or substantial portions +of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF +ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED +TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A +PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT +SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY +CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION +OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR +IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER +DEALINGS IN THE SOFTWARE. diff --git a/third_party/rust/tokio-timer/README.md b/third_party/rust/tokio-timer/README.md new file mode 100644 index 0000000000..618309fd6d --- /dev/null +++ b/third_party/rust/tokio-timer/README.md @@ -0,0 +1,25 @@ +# tokio-timer + +Timer facilities for Tokio + +> **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved into +> [`tokio::time`] behind the `time` [feature flag]. + +[`tokio::time`]: https://docs.rs/tokio/latest/tokio/time/index.html +[feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags + +[Documentation](https://docs.rs/tokio-timer/0.2.12/tokio_timer/) + +## Overview + +This crate provides timer facilities for usage with Tokio. + +## License + +This project is licensed under the [MIT license](LICENSE). + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, shall be licensed as MIT, without any additional +terms or conditions. diff --git a/third_party/rust/tokio-timer/src/atomic.rs b/third_party/rust/tokio-timer/src/atomic.rs new file mode 100644 index 0000000000..d60bd7645d --- /dev/null +++ b/third_party/rust/tokio-timer/src/atomic.rs @@ -0,0 +1,88 @@ +//! Implementation of an atomic u64 cell. On 64 bit platforms, this is a wrapper +//! around `AtomicUsize`. On 32 bit platforms, this is implemented using a +//! `Mutex`. +//! +//! This file can be removed if/when `AtomicU64` lands in `std`. + +pub use self::imp::AtomicU64; + +#[cfg(target_pointer_width = "64")] +mod imp { + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[derive(Debug)] + pub struct AtomicU64 { + inner: AtomicUsize, + } + + impl AtomicU64 { + pub fn new(val: u64) -> AtomicU64 { + AtomicU64 { + inner: AtomicUsize::new(val as usize), + } + } + + pub fn load(&self, ordering: Ordering) -> u64 { + self.inner.load(ordering) as u64 + } + + pub fn store(&self, val: u64, ordering: Ordering) { + self.inner.store(val as usize, ordering) + } + + pub fn fetch_or(&self, val: u64, ordering: Ordering) -> u64 { + self.inner.fetch_or(val as usize, ordering) as u64 + } + + pub fn compare_and_swap(&self, old: u64, new: u64, ordering: Ordering) -> u64 { + self.inner + .compare_and_swap(old as usize, new as usize, ordering) as u64 + } + } +} + +#[cfg(not(target_pointer_width = "64"))] +mod imp { + use std::sync::atomic::Ordering; + use std::sync::Mutex; + + #[derive(Debug)] + pub struct AtomicU64 { + inner: Mutex<u64>, + } + + impl AtomicU64 { + pub fn new(val: u64) -> AtomicU64 { + AtomicU64 { + inner: Mutex::new(val), + } + } + + pub fn load(&self, _: Ordering) -> u64 { + *self.inner.lock().unwrap() + } + + pub fn store(&self, val: u64, _: Ordering) { + *self.inner.lock().unwrap() = val; + } + + pub fn fetch_or(&self, val: u64, _: Ordering) -> u64 { + let mut lock = self.inner.lock().unwrap(); + let prev = *lock; + *lock = prev | val; + prev + } + + pub 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 + } + } +} diff --git a/third_party/rust/tokio-timer/src/clock/clock.rs b/third_party/rust/tokio-timer/src/clock/clock.rs new file mode 100644 index 0000000000..2920e2817f --- /dev/null +++ b/third_party/rust/tokio-timer/src/clock/clock.rs @@ -0,0 +1,150 @@ +use clock::Now; +use timer; + +use tokio_executor::Enter; + +use std::cell::RefCell; +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`]: https://doc.rust-lang.org/std/time/struct.Instant.html +/// [`Instant::now`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.now +#[derive(Default, Clone)] +pub struct Clock { + now: Option<Arc<dyn Now>>, +} + +/// A guard that resets the current `Clock` to `None` when dropped. +#[derive(Debug)] +pub struct DefaultGuard { + _p: (), +} + +thread_local! { + /// Thread-local tracking the current clock + static CLOCK: RefCell<Option<Clock>> = RefCell::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.borrow().as_ref() { + Some(c) => c.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.borrow().as_ref() { + Some(c) => c.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`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.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(), + } + } +} + +#[allow(deprecated)] +impl timer::Now for Clock { + fn now(&mut self) -> Instant { + Clock::now(self) + } +} + +impl fmt::Debug for Clock { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Clock") + .field("now", { + if self.now.is_some() { + &"Some(Arc<Now>)" + } else { + &"None" + } + }) + .finish() + } +} + +/// Set the default clock for the duration of the closure. +/// +/// # Panics +/// +/// This function panics if there already is a default clock set. +pub fn with_default<F, R>(clock: &Clock, enter: &mut Enter, f: F) -> R +where + F: FnOnce(&mut Enter) -> R, +{ + let _guard = set_default(clock); + + f(enter) +} + +/// Sets `clock` as the default clock, returning a guard that unsets it on drop. +/// +/// # Panics +/// +/// This function panics if there already is a default clock set. +pub fn set_default(clock: &Clock) -> DefaultGuard { + CLOCK.with(|cell| { + assert!( + cell.borrow().is_none(), + "default clock already set for execution context" + ); + + *cell.borrow_mut() = Some(clock.clone()); + + DefaultGuard { _p: () } + }) +} + +impl Drop for DefaultGuard { + fn drop(&mut self) { + let _ = CLOCK.try_with(|cell| cell.borrow_mut().take()); + } +} diff --git a/third_party/rust/tokio-timer/src/clock/mod.rs b/third_party/rust/tokio-timer/src/clock/mod.rs new file mode 100644 index 0000000000..8acb831798 --- /dev/null +++ b/third_party/rust/tokio-timer/src/clock/mod.rs @@ -0,0 +1,23 @@ +//! 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`]: https://doc.rust-lang.org/std/time/struct.Instant.html +//! [`Instant::now`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.now +//! [`with_default`]: fn.with_default.html + +mod clock; +mod now; + +pub use self::clock::{now, set_default, with_default, Clock, DefaultGuard}; +pub use self::now::Now; diff --git a/third_party/rust/tokio-timer/src/clock/now.rs b/third_party/rust/tokio-timer/src/clock/now.rs new file mode 100644 index 0000000000..18450c8302 --- /dev/null +++ b/third_party/rust/tokio-timer/src/clock/now.rs @@ -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`]: https://doc.rust-lang.org/std/time/struct.Instant.html +pub trait Now: Send + Sync + 'static { + /// Returns an instant corresponding to "now". + fn now(&self) -> Instant; +} diff --git a/third_party/rust/tokio-timer/src/deadline.rs b/third_party/rust/tokio-timer/src/deadline.rs new file mode 100644 index 0000000000..c4c19b8bbc --- /dev/null +++ b/third_party/rust/tokio-timer/src/deadline.rs @@ -0,0 +1,174 @@ +#![allow(deprecated)] + +use 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(::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: ::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<::Error> { + match self.0 { + Kind::Timer(err) => Some(err), + _ => None, + } + } +} + +impl<T: error::Error> error::Error for DeadlineError<T> { + fn description(&self) -> &str { + use self::Kind::*; + + match self.0 { + Inner(ref e) => e.description(), + Elapsed => "deadline has elapsed", + Timer(ref e) => e.description(), + } + } +} + +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), + } + } +} diff --git a/third_party/rust/tokio-timer/src/delay.rs b/third_party/rust/tokio-timer/src/delay.rs new file mode 100644 index 0000000000..c48adccd57 --- /dev/null +++ b/third_party/rust/tokio-timer/src/delay.rs @@ -0,0 +1,98 @@ +use timer::{HandlePriv, Registration}; +use Error; + +use futures::{Future, 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 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, handle: HandlePriv) -> Delay { + let mut registration = Registration::new(deadline, Duration::from_millis(0)); + 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 Item = (); + type Error = Error; + + fn poll(&mut self) -> Poll<Self::Item, Self::Error> { + // Ensure the `Delay` instance is associated with a timer. + self.register(); + + self.registration.poll_elapsed() + } +} diff --git a/third_party/rust/tokio-timer/src/delay_queue.rs b/third_party/rust/tokio-timer/src/delay_queue.rs new file mode 100644 index 0000000000..1f4b0a17af --- /dev/null +++ b/third_party/rust/tokio-timer/src/delay_queue.rs @@ -0,0 +1,845 @@ +//! A queue of delayed elements. +//! +//! See [`DelayQueue`] for more details. +//! +//! [`DelayQueue`]: struct.DelayQueue.html + +use clock::now; +use timer::Handle; +use wheel::{self, Wheel}; +use {Delay, Error}; + +use futures::{Future, Poll, Stream}; +use slab::Slab; + +use std::cmp; +use std::marker::PhantomData; +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 +/// #[macro_use] +/// extern crate futures; +/// extern crate tokio; +/// # type CacheKey = String; +/// # type Value = String; +/// use tokio::timer::{delay_queue, DelayQueue, Error}; +/// use futures::{Async, Poll, Stream}; +/// use std::collections::HashMap; +/// use std::time::Duration; +/// +/// 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) -> Poll<(), Error> { +/// while let Some(entry) = try_ready!(self.expirations.poll()) { +/// self.entries.remove(entry.get_ref()); +/// } +/// +/// Ok(Async::Ready(())) +/// } +/// } +/// # fn main() {} +/// ``` +/// +/// [`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<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 + /// # extern crate tokio; + /// use tokio::timer::DelayQueue; + /// use std::time::{Instant, Duration}; + /// + /// # fn main() { + /// 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) + } + + /// 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 + /// # extern crate tokio; + /// use tokio::timer::DelayQueue; + /// use std::time::Duration; + /// + /// # fn main() { + /// 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 + /// # extern crate tokio; + /// use tokio::timer::DelayQueue; + /// use std::time::Duration; + /// + /// # fn main() { + /// 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 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 + /// # extern crate tokio; + /// use tokio::timer::DelayQueue; + /// use std::time::{Duration, Instant}; + /// + /// # fn main() { + /// 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 + /// # extern crate tokio; + /// use tokio::timer::DelayQueue; + /// use std::time::Duration; + /// + /// # fn main() { + /// 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 + /// # extern crate tokio; + /// use tokio::timer::DelayQueue; + /// use std::time::Duration; + /// + /// # fn main() { + /// 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) -> Poll<Option<usize>, Error> { + use self::wheel::Stack; + + let expired = self.expired.pop(&mut self.slab); + + if expired.is_some() { + return Ok(expired.into()); + } + + loop { + if let Some(ref mut delay) = self.delay { + if !delay.is_elapsed() { + try_ready!(delay.poll()); + } + + let now = ::ms(delay.deadline() - self.start, ::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 Ok(Some(idx).into()); + } + + if let Some(deadline) = self.next_deadline() { + self.delay = Some(self.handle.delay(deadline)); + } else { + return Ok(None.into()); + } + } + } + + fn normalize_deadline(&self, when: Instant) -> u64 { + let when = if when < self.start { + 0 + } else { + ::ms(when - self.start, ::Round::Up) + }; + + cmp::max(when, self.wheel.elapsed()) + } +} + +impl<T> Stream for DelayQueue<T> { + type Item = Expired<T>; + type Error = Error; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Error> { + let item = try_ready!(self.poll_idx()).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), + } + }); + + Ok(item.into()) + } +} + +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 + } +} diff --git a/third_party/rust/tokio-timer/src/error.rs b/third_party/rust/tokio-timer/src/error.rs new file mode 100644 index 0000000000..a6136d8b55 --- /dev/null +++ b/third_party/rust/tokio-timer/src/error.rs @@ -0,0 +1,78 @@ +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 { + fn description(&self) -> &str { + use self::Kind::*; + + match self.0 { + Shutdown => "timer is shutdown", + AtCapacity => "timer is at capacity and cannot create a new entry", + } + } +} + +impl fmt::Display for Error { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + use std::error::Error; + self.description().fmt(fmt) + } +} diff --git a/third_party/rust/tokio-timer/src/interval.rs b/third_party/rust/tokio-timer/src/interval.rs new file mode 100644 index 0000000000..019efe6a2d --- /dev/null +++ b/third_party/rust/tokio-timer/src/interval.rs @@ -0,0 +1,75 @@ +use Delay; + +use clock; + +use futures::{Future, Poll, Stream}; + +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(Instant::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 } + } +} + +impl Stream for Interval { + type Item = Instant; + type Error = ::Error; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { + // Wait for the delay to be done + let _ = try_ready!(self.delay.poll()); + + // 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. + self.delay.reset(now + self.duration); + + // Return the current instant + Ok(Some(now).into()) + } +} diff --git a/third_party/rust/tokio-timer/src/lib.rs b/third_party/rust/tokio-timer/src/lib.rs new file mode 100644 index 0000000000..a141dd2796 --- /dev/null +++ b/third_party/rust/tokio-timer/src/lib.rs @@ -0,0 +1,105 @@ +#![doc(html_root_url = "https://docs.rs/tokio-timer/0.2.13")] +#![deny(missing_docs, missing_debug_implementations)] + +//! Utilities for tracking time. +//! +//! > **Note:** This crate is **deprecated in tokio 0.2.x** and has been moved +//! > into [`tokio::time`] behind the `time` [feature flag]. +//! +//! [`tokio::time`]: https://docs.rs/tokio/latest/tokio/time/index.html +//! [feature flag]: https://docs.rs/tokio/latest/tokio/index.html#feature-flags +//! +//! This crate provides a number of utilities for working with periods of time: +//! +//! * [`Delay`]: A future that completes at a specified instant in time. +//! +//! * [`Interval`] A stream that yields at fixed time intervals. +//! +//! * [`Throttle`]: Throttle down a stream by enforcing a fixed delay between items. +//! +//! * [`Timeout`]: Wraps a future or stream, setting an upper bound to the +//! amount of time it is allowed to execute. If the future or stream does not +//! complete in time, then it is canceled and an error is returned. +//! +//! * [`DelayQueue`]: A queue where items are returned once the requested delay +//! has expired. +//! +//! These three types are backed by a [`Timer`] instance. In order for +//! [`Delay`], [`Interval`], and [`Timeout`] to function, the associated +//! [`Timer`] instance must be running on some thread. +//! +//! [`Delay`]: struct.Delay.html +//! [`DelayQueue`]: struct.DelayQueue.html +//! [`Throttle`]: throttle/struct.Throttle.html +//! [`Timeout`]: struct.Timeout.html +//! [`Interval`]: struct.Interval.html +//! [`Timer`]: timer/struct.Timer.html + +extern crate tokio_executor; + +extern crate crossbeam_utils; +#[macro_use] +extern crate futures; +extern crate slab; + +pub mod clock; +pub mod delay_queue; +pub mod throttle; +pub mod timeout; +pub mod timer; + +mod atomic; +mod deadline; +mod delay; +mod error; +mod interval; +mod wheel; + +#[deprecated(since = "0.2.6", note = "use Timeout instead")] +#[doc(hidden)] +#[allow(deprecated)] +pub use self::deadline::{Deadline, DeadlineError}; +pub use self::delay::Delay; +#[doc(inline)] +pub use self::delay_queue::DelayQueue; +pub use self::error::Error; +pub use self::interval::Interval; +#[doc(inline)] +pub use self::timeout::Timeout; +pub use self::timer::{with_default, Timer}; + +use std::time::{Duration, Instant}; + +/// Create a Future that completes in `duration` from now. +pub fn sleep(duration: Duration) -> Delay { + Delay::new(Instant::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_nanos() / NANOS_PER_MILLI, + }; + + duration + .as_secs() + .saturating_mul(MILLIS_PER_SEC) + .saturating_add(millis as u64) +} diff --git a/third_party/rust/tokio-timer/src/throttle.rs b/third_party/rust/tokio-timer/src/throttle.rs new file mode 100644 index 0000000000..97d313313e --- /dev/null +++ b/third_party/rust/tokio-timer/src/throttle.rs @@ -0,0 +1,167 @@ +//! Slow down a stream by enforcing a delay between items. + +use {clock, Delay, Error}; + +use futures::future::Either; +use futures::{Async, Future, Poll, Stream}; + +use std::{ + error::Error as StdError, + fmt::{Display, Formatter, Result as FmtResult}, + 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: Option<Delay>, + duration: Duration, + stream: T, +} + +/// Either the error of the underlying stream, or an error within +/// tokio's timing machinery. +#[derive(Debug)] +pub struct ThrottleError<T>(Either<T, Error>); + +impl<T> Throttle<T> { + /// Slow down a stream by enforcing a delay between items. + pub fn new(stream: T, duration: Duration) -> Self { + Self { + delay: None, + duration: duration, + stream: stream, + } + } + + /// 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; + type Error = ThrottleError<T::Error>; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { + if let Some(ref mut delay) = self.delay { + try_ready!({ delay.poll().map_err(ThrottleError::from_timer_err) }); + } + + self.delay = None; + let value = try_ready!({ self.stream.poll().map_err(ThrottleError::from_stream_err) }); + + if value.is_some() { + self.delay = Some(Delay::new(clock::now() + self.duration)); + } + + Ok(Async::Ready(value)) + } +} + +impl<T> ThrottleError<T> { + /// Creates a new `ThrottleError` from the given stream error. + pub fn from_stream_err(err: T) -> Self { + ThrottleError(Either::A(err)) + } + + /// Creates a new `ThrottleError` from the given tokio timer error. + pub fn from_timer_err(err: Error) -> Self { + ThrottleError(Either::B(err)) + } + + /// Attempts to get the underlying stream error, if it is present. + pub fn get_stream_error(&self) -> Option<&T> { + match self.0 { + Either::A(ref x) => Some(x), + _ => None, + } + } + + /// Attempts to get the underlying timer error, if it is present. + pub fn get_timer_error(&self) -> Option<&Error> { + match self.0 { + Either::B(ref x) => Some(x), + _ => None, + } + } + + /// Attempts to extract the underlying stream error, if it is present. + pub fn into_stream_error(self) -> Option<T> { + match self.0 { + Either::A(x) => Some(x), + _ => None, + } + } + + /// Attempts to extract the underlying timer error, if it is present. + pub fn into_timer_error(self) -> Option<Error> { + match self.0 { + Either::B(x) => Some(x), + _ => None, + } + } + + /// Returns whether the throttle error has occured because of an error + /// in the underlying stream. + pub fn is_stream_error(&self) -> bool { + !self.is_timer_error() + } + + /// Returns whether the throttle error has occured because of an error + /// in tokio's timer system. + pub fn is_timer_error(&self) -> bool { + match self.0 { + Either::A(_) => false, + Either::B(_) => true, + } + } +} + +impl<T: StdError> Display for ThrottleError<T> { + fn fmt(&self, f: &mut Formatter) -> FmtResult { + match self.0 { + Either::A(ref err) => write!(f, "stream error: {}", err), + Either::B(ref err) => write!(f, "timer error: {}", err), + } + } +} + +impl<T: StdError + 'static> StdError for ThrottleError<T> { + fn description(&self) -> &str { + match self.0 { + Either::A(_) => "stream error", + Either::B(_) => "timer error", + } + } + + // FIXME(taiki-e): When the minimum support version of tokio reaches Rust 1.30, + // replace this with Error::source. + #[allow(deprecated)] + fn cause(&self) -> Option<&dyn StdError> { + match self.0 { + Either::A(ref err) => Some(err), + Either::B(ref err) => Some(err), + } + } +} diff --git a/third_party/rust/tokio-timer/src/timeout.rs b/third_party/rust/tokio-timer/src/timeout.rs new file mode 100644 index 0000000000..1c02a68675 --- /dev/null +++ b/third_party/rust/tokio-timer/src/timeout.rs @@ -0,0 +1,311 @@ +//! Allows a future or stream to execute for a maximum amount of time. +//! +//! See [`Timeout`] documentation for more details. +//! +//! [`Timeout`]: struct.Timeout.html + +use clock::now; +use Delay; + +use futures::{Async, Future, Poll, Stream}; + +use std::error; +use std::fmt; +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 +/// # extern crate futures; +/// # extern crate tokio; +/// // import the `timeout` function, usually this is done +/// // with `use tokio::prelude::*` +/// use tokio::prelude::FutureExt; +/// use futures::Stream; +/// use futures::sync::mpsc; +/// use std::time::Duration; +/// +/// # fn main() { +/// let (tx, rx) = mpsc::unbounded(); +/// # tx.unbounded_send(()).unwrap(); +/// # drop(tx); +/// +/// let process = rx.for_each(|item| { +/// // do something with `item` +/// # drop(item); +/// # Ok(()) +/// }); +/// +/// # tokio::runtime::current_thread::block_on_all( +/// // Wrap the future with a `Timeout` set to expire in 10 milliseconds. +/// process.timeout(Duration::from_millis(10)) +/// # ).unwrap(); +/// # } +/// ``` +/// +/// # 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 polled"] +#[derive(Debug)] +pub struct Timeout<T> { + value: T, + delay: Delay, +} + +/// Error returned by `Timeout`. +#[derive(Debug)] +pub struct Error<T>(Kind<T>); + +/// Timeout error variants +#[derive(Debug)] +enum Kind<T> { + /// Inner value returned an error + Inner(T), + + /// The timeout elapsed. + Elapsed, + + /// Timer returned an error. + Timer(::Error), +} + +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 + /// # extern crate futures; + /// # extern crate tokio; + /// use tokio::timer::Timeout; + /// use futures::Future; + /// use futures::sync::oneshot; + /// use std::time::Duration; + /// + /// # fn main() { + /// let (tx, rx) = oneshot::channel(); + /// # tx.send(()).unwrap(); + /// + /// # tokio::runtime::current_thread::block_on_all( + /// // Wrap the future with a `Timeout` set to expire in 10 milliseconds. + /// Timeout::new(rx, Duration::from_millis(10)) + /// # ).unwrap(); + /// # } + /// ``` + 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 Item = T::Item; + type Error = Error<T::Error>; + + fn poll(&mut self) -> Poll<Self::Item, Self::Error> { + // First, try polling the future + match self.value.poll() { + Ok(Async::Ready(v)) => return Ok(Async::Ready(v)), + Ok(Async::NotReady) => {} + Err(e) => return Err(Error::inner(e)), + } + + // Now check the timer + match self.delay.poll() { + Ok(Async::NotReady) => Ok(Async::NotReady), + Ok(Async::Ready(_)) => Err(Error::elapsed()), + Err(e) => Err(Error::timer(e)), + } + } +} + +impl<T> Stream for Timeout<T> +where + T: Stream, +{ + type Item = T::Item; + type Error = Error<T::Error>; + + fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { + // First, try polling the future + match self.value.poll() { + Ok(Async::Ready(v)) => { + if v.is_some() { + self.delay.reset_timeout(); + } + return Ok(Async::Ready(v)); + } + Ok(Async::NotReady) => {} + Err(e) => return Err(Error::inner(e)), + } + + // Now check the timer + match self.delay.poll() { + Ok(Async::NotReady) => Ok(Async::NotReady), + Ok(Async::Ready(_)) => { + self.delay.reset_timeout(); + Err(Error::elapsed()) + } + Err(e) => Err(Error::timer(e)), + } + } +} + +// ===== impl Error ===== + +impl<T> Error<T> { + /// Create a new `Error` representing the inner value completing with `Err`. + pub fn inner(err: T) -> Error<T> { + Error(Kind::Inner(err)) + } + + /// Returns `true` if the error was caused by the inner value 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 `Error` representing the inner value not completing before + /// the deadline is reached. + pub fn elapsed() -> Error<T> { + Error(Kind::Elapsed) + } + + /// Returns `true` if the error was caused by the inner value not completing + /// before the deadline is reached. + pub fn is_elapsed(&self) -> bool { + match self.0 { + Kind::Elapsed => true, + _ => false, + } + } + + /// Creates a new `Error` representing an error encountered by the timer + /// implementation + pub fn timer(err: ::Error) -> Error<T> { + Error(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<::Error> { + match self.0 { + Kind::Timer(err) => Some(err), + _ => None, + } + } +} + +impl<T: error::Error> error::Error for Error<T> { + fn description(&self) -> &str { + use self::Kind::*; + + match self.0 { + Inner(ref e) => e.description(), + Elapsed => "deadline has elapsed", + Timer(ref e) => e.description(), + } + } +} + +impl<T: fmt::Display> fmt::Display for Error<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), + } + } +} diff --git a/third_party/rust/tokio-timer/src/timer/atomic_stack.rs b/third_party/rust/tokio-timer/src/timer/atomic_stack.rs new file mode 100644 index 0000000000..4e7d8ed6ec --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/atomic_stack.rs @@ -0,0 +1,124 @@ +use super::Entry; +use 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 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 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).into(); + + 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 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 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) { + while let Some(entry) = self.next() { + // Flag the entry as errored + entry.error(); + } + } +} diff --git a/third_party/rust/tokio-timer/src/timer/entry.rs b/third_party/rust/tokio-timer/src/timer/entry.rs new file mode 100644 index 0000000000..40979afaec --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/entry.rs @@ -0,0 +1,394 @@ +use atomic::AtomicU64; +use timer::{HandlePriv, Inner}; +use Error; + +use crossbeam_utils::CachePadded; +use futures::task::AtomicTask; +use futures::Poll; + +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::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. + task: AtomicTask, + + /// 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 fn new(deadline: Instant, duration: Duration) -> Entry { + Entry { + time: CachePadded::new(UnsafeCell::new(Time { deadline, duration })), + inner: None, + task: AtomicTask::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 fn time_ref(&self) -> &Time { + unsafe { &*self.time.get() } + } + + /// Only called by `Registration` + pub fn time_mut(&self) -> &mut Time { + unsafe { &mut *self.time.get() } + } + + /// Returns `true` if the `Entry` is currently associated with a timer + /// instance. + pub fn is_registered(&self) -> bool { + self.inner.is_some() + } + + /// Only called by `Registration` + pub 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 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 fn when_internal(&self) -> Option<u64> { + unsafe { (*self.when.get()) } + } + + pub 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 fn load_state(&self) -> Option<u64> { + let state = self.state.load(SeqCst); + + if is_elapsed(state) { + None + } else { + Some(state) + } + } + + pub fn is_elapsed(&self) -> bool { + let state = self.state.load(SeqCst); + is_elapsed(state) + } + + pub 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.task.notify(); + } + + pub 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.task.notify(); + } + + pub 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 fn poll_elapsed(&self) -> Poll<(), Error> { + use futures::Async::NotReady; + + let mut curr = self.state.load(SeqCst); + + if is_elapsed(curr) { + if curr == ERROR { + return Err(Error::shutdown()); + } else { + return Ok(().into()); + } + } + + self.task.register(); + + curr = self.state.load(SeqCst).into(); + + if is_elapsed(curr) { + if curr == ERROR { + return Err(Error::shutdown()); + } else { + return Ok(().into()); + } + } + + Ok(NotReady) + } + + /// Only called by `Registration` + pub 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 {} diff --git a/third_party/rust/tokio-timer/src/timer/handle.rs b/third_party/rust/tokio-timer/src/timer/handle.rs new file mode 100644 index 0000000000..4c444d8a66 --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/handle.rs @@ -0,0 +1,201 @@ +use timer::Inner; +use {Deadline, Delay, Error, Interval, Timeout}; + +use tokio_executor::Enter; + +use std::cell::RefCell; +use std::fmt; +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>, +} + +/// A guard that resets the current timer to `None` when dropped. +#[derive(Debug)] +pub struct DefaultGuard { + _p: (), +} + +thread_local! { + /// Tracks the timer for the current execution context. + static CURRENT_TIMER: RefCell<Option<HandlePriv>> = RefCell::new(None) +} + +/// Set the default timer for the duration of the closure. +/// +/// From within the closure, [`Delay`] instances that are created via +/// [`Delay::new`] can be used. +/// +/// # Panics +/// +/// This function panics if there already is a default timer set. +/// +/// [`Delay`]: ../struct.Delay.html +/// [`Delay::new`]: ../struct.Delay.html#method.new +pub fn with_default<F, R>(handle: &Handle, enter: &mut Enter, f: F) -> R +where + F: FnOnce(&mut Enter) -> R, +{ + let _guard = set_default(handle); + f(enter) +} + +/// Sets `handle` as the default timer, returning a 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 { _p: () } +} + +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.html + /// [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 { + match self.inner { + Some(ref handle_priv) => Delay::new_with_handle(deadline, handle_priv.clone()), + None => Delay::new(deadline), + } + } + + #[doc(hidden)] + #[deprecated(since = "0.2.11", note = "use timeout instead")] + pub fn deadline<T>(&self, future: T, deadline: Instant) -> Deadline<T> { + Deadline::new_with_delay(future, self.delay(deadline)) + } + + /// Create a `Timeout` driven by this handle's associated `Timer`. + pub fn timeout<T>(&self, value: T, deadline: Instant) -> Timeout<T> { + Timeout::new_with_delay(value, self.delay(deadline)) + } + + /// 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") + } +} + +impl Drop for DefaultGuard { + fn drop(&mut self) { + let _ = CURRENT_TIMER.try_with(|current| { + let mut current = current.borrow_mut(); + *current = None; + }); + } +} diff --git a/third_party/rust/tokio-timer/src/timer/mod.rs b/third_party/rust/tokio-timer/src/timer/mod.rs new file mode 100644 index 0000000000..31eb0afbb1 --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/mod.rs @@ -0,0 +1,490 @@ +//! 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`]. +//! +//! The [`Now`] trait describes how to get an [`Instant`] representing the +//! current moment in time. [`SystemNow`] is the default implementation, where +//! [`Now::now`] is implemented by calling [`Instant::now`]. +//! +//! [`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`]: ../struct.Delay.html +//! [`Now`]: ../clock/trait.Now.html +//! [`Now::now`]: ../clock/trait.Now.html#method.now +//! [`SystemNow`]: struct.SystemNow.html +//! [`Instant`]: https://doc.rust-lang.org/std/time/struct.Instant.html +//! [`Instant::now`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.now + +// This allows the usage of the old `Now` trait. +#![allow(deprecated)] + +mod atomic_stack; +mod entry; +mod handle; +mod now; +mod registration; +mod stack; + +use self::atomic_stack::AtomicStack; +use self::entry::Entry; +use self::stack::Stack; + +pub(crate) use self::handle::HandlePriv; +pub use self::handle::{set_default, with_default, DefaultGuard, Handle}; +pub use self::now::{Now, SystemNow}; +pub(crate) use self::registration::Registration; + +use atomic::AtomicU64; +use wheel; +use 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, N = SystemNow> { + /// 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 + now: N, +} + +/// 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_now(park, SystemNow::new()) + } +} + +impl<T, N> Timer<T, N> { + /// 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, N> Timer<T, N> +where + T: Park, + N: Now, +{ + /// 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_now(park: T, mut now: N) -> Self { + let unpark = Box::new(park.unpark()); + + Timer { + inner: Arc::new(Inner::new(now.now(), unpark)), + wheel: wheel::Wheel::new(), + park, + now, + } + } + + /// 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 = ::ms(self.now.now() - self.inner.start, ::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 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, SystemNow> { + fn default() -> Self { + Timer::new(ParkThread::new()) + } +} + +impl<T, N> Park for Timer<T, N> +where + T: Park, + N: Now, +{ + 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.now.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.now.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, N> Drop for Timer<T, N> { + 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; + } + + ::ms(deadline - self.start, ::Round::Up) + } +} + +impl fmt::Debug for Inner { + fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result { + fmt.debug_struct("Inner").finish() + } +} diff --git a/third_party/rust/tokio-timer/src/timer/now.rs b/third_party/rust/tokio-timer/src/timer/now.rs new file mode 100644 index 0000000000..9f23bad711 --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/now.rs @@ -0,0 +1,10 @@ +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; +} + +pub use clock::Clock as SystemNow; diff --git a/third_party/rust/tokio-timer/src/timer/registration.rs b/third_party/rust/tokio-timer/src/timer/registration.rs new file mode 100644 index 0000000000..dad1355dcd --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/registration.rs @@ -0,0 +1,67 @@ +use clock::now; +use timer::{Entry, HandlePriv}; +use Error; + +use futures::Poll; + +use std::sync::Arc; +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 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 fn deadline(&self) -> Instant { + self.entry.time_ref().deadline + } + + pub fn register(&mut self) { + if !self.entry.is_registered() { + Entry::register(&mut self.entry) + } + } + + pub fn register_with(&mut self, handle: HandlePriv) { + Entry::register_with(&mut self.entry, handle) + } + + pub fn reset(&mut self, deadline: Instant) { + self.entry.time_mut().deadline = deadline; + Entry::reset(&mut self.entry); + } + + pub fn reset_timeout(&mut self) { + let deadline = now() + self.entry.time_ref().duration; + self.entry.time_mut().deadline = deadline; + Entry::reset(&mut self.entry); + } + + pub fn is_elapsed(&self) -> bool { + self.entry.is_elapsed() + } + + pub fn poll_elapsed(&self) -> Poll<(), Error> { + self.entry.poll_elapsed() + } +} + +impl Drop for Registration { + fn drop(&mut self) { + Entry::cancel(&self.entry); + } +} diff --git a/third_party/rust/tokio-timer/src/timer/stack.rs b/third_party/rust/tokio-timer/src/timer/stack.rs new file mode 100644 index 0000000000..c63eed971b --- /dev/null +++ b/third_party/rust/tokio-timer/src/timer/stack.rs @@ -0,0 +1,121 @@ +use super::Entry; +use 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") + } +} diff --git a/third_party/rust/tokio-timer/src/wheel/level.rs b/third_party/rust/tokio-timer/src/wheel/level.rs new file mode 100644 index 0000000000..6bbd128e0d --- /dev/null +++ b/third_party/rust/tokio-timer/src/wheel/level.rs @@ -0,0 +1,255 @@ +use 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 level: usize, + + /// The slot index. + pub slot: usize, + + /// The instant at which the slot needs to be processed. + pub deadline: u64, +} + +/// Level multiplier. +/// +/// Being a power of 2 is very important. +const LEVEL_MULT: usize = 64; + +impl<T: Stack> Level<T> { + pub 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 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 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 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 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)); + } + } + } +} +*/ diff --git a/third_party/rust/tokio-timer/src/wheel/mod.rs b/third_party/rust/tokio-timer/src/wheel/mod.rs new file mode 100644 index 0000000000..81f92cd886 --- /dev/null +++ b/third_party/rust/tokio-timer/src/wheel/mod.rs @@ -0,0 +1,311 @@ +mod level; +mod stack; + +pub(crate) use self::level::Expiration; +use self::level::Level; +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 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 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 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 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 fn poll_at(&self) -> Option<u64> { + self.next_expiration().map(|expiration| expiration.deadline) + } + + pub 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 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 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 + ); + } + } + } + } +} diff --git a/third_party/rust/tokio-timer/src/wheel/stack.rs b/third_party/rust/tokio-timer/src/wheel/stack.rs new file mode 100644 index 0000000000..6e55c38ccd --- /dev/null +++ b/third_party/rust/tokio-timer/src/wheel/stack.rs @@ -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; +} diff --git a/third_party/rust/tokio-timer/tests/clock.rs b/third_party/rust/tokio-timer/tests/clock.rs new file mode 100644 index 0000000000..d8bcaaebd7 --- /dev/null +++ b/third_party/rust/tokio-timer/tests/clock.rs @@ -0,0 +1,51 @@ +extern crate tokio_executor; +extern crate tokio_timer; + +use tokio_timer::clock; +use tokio_timer::clock::*; + +use std::time::Instant; + +struct ConstNow(Instant); + +impl Now for ConstNow { + fn now(&self) -> Instant { + self.0 + } +} + +#[test] +fn default_clock() { + let a = Instant::now(); + let b = clock::now(); + let c = Clock::new().now(); + + assert!(a <= b); + assert!(b <= c); +} + +#[test] +fn custom_clock() { + let now = ConstNow(Instant::now()); + let clock = Clock::new_with_now(now); + + let a = Instant::now(); + let b = clock.now(); + + assert!(b <= a); +} + +#[test] +fn execution_context() { + let now = ConstNow(Instant::now()); + let clock = Clock::new_with_now(now); + + let mut enter = tokio_executor::enter().unwrap(); + + with_default(&clock, &mut enter, |_| { + let a = Instant::now(); + let b = clock::now(); + + assert!(b <= a); + }); +} diff --git a/third_party/rust/tokio-timer/tests/deadline.rs b/third_party/rust/tokio-timer/tests/deadline.rs new file mode 100644 index 0000000000..8eec79ae7f --- /dev/null +++ b/third_party/rust/tokio-timer/tests/deadline.rs @@ -0,0 +1,106 @@ +#![allow(deprecated)] +extern crate futures; +extern crate tokio_executor; +extern crate tokio_timer; + +#[macro_use] +mod support; +use support::*; + +use tokio_timer::*; + +use futures::sync::oneshot; +use futures::{future, Future}; + +#[test] +fn simultaneous_deadline_future_completion() { + mocked(|_, time| { + // Create a future that is immediately ready + let fut = future::ok::<_, ()>(()); + + // Wrap it with a deadline + let mut fut = Deadline::new(fut, time.now()); + + // Ready! + assert_ready!(fut); + }); +} + +#[test] +fn completed_future_past_deadline() { + mocked(|_, time| { + // Create a future that is immediately ready + let fut = future::ok::<_, ()>(()); + + // Wrap it with a deadline + let mut fut = Deadline::new(fut, time.now() - ms(1000)); + + // Ready! + assert_ready!(fut); + }); +} + +#[test] +fn future_and_deadline_in_future() { + mocked(|timer, time| { + // Not yet complete + let (tx, rx) = oneshot::channel(); + + // Wrap it with a deadline + let mut fut = Deadline::new(rx, time.now() + ms(100)); + + // Ready! + assert_not_ready!(fut); + + // Turn the timer, it runs for the elapsed time + advance(timer, ms(90)); + + assert_not_ready!(fut); + + // Complete the future + tx.send(()).unwrap(); + + assert_ready!(fut); + }); +} + +#[test] +fn deadline_now_elapses() { + mocked(|_, time| { + let fut = future::empty::<(), ()>(); + + // Wrap it with a deadline + let mut fut = Deadline::new(fut, time.now()); + + assert_elapsed!(fut); + }); +} + +#[test] +fn deadline_future_elapses() { + mocked(|timer, time| { + let fut = future::empty::<(), ()>(); + + // Wrap it with a deadline + let mut fut = Deadline::new(fut, time.now() + ms(300)); + + assert_not_ready!(fut); + + advance(timer, ms(300)); + + assert_elapsed!(fut); + }); +} + +#[test] +fn future_errors_first() { + mocked(|_, time| { + let fut = future::err::<(), ()>(()); + + // Wrap it with a deadline + let mut fut = Deadline::new(fut, time.now() + ms(100)); + + // Ready! + assert!(fut.poll().unwrap_err().is_inner()); + }); +} diff --git a/third_party/rust/tokio-timer/tests/delay.rs b/third_party/rust/tokio-timer/tests/delay.rs new file mode 100644 index 0000000000..d21106c630 --- /dev/null +++ b/third_party/rust/tokio-timer/tests/delay.rs @@ -0,0 +1,499 @@ +extern crate futures; +extern crate tokio_executor; +extern crate tokio_timer; + +#[macro_use] +mod support; +use support::*; + +use tokio_timer::timer::Handle; +use tokio_timer::*; + +use futures::Future; + +use std::time::{Duration, Instant}; + +#[test] +fn immediate_delay() { + mocked(|timer, time| { + // Create `Delay` that elapsed immediately. + let mut delay = Delay::new(time.now()); + + // Ready! + assert_ready!(delay); + + // Turn the timer, it runs for the elapsed time + turn(timer, ms(1000)); + + // The time has not advanced. The `turn` completed immediately. + assert_eq!(time.advanced(), ms(1000)); + }); +} + +#[test] +fn delayed_delay_level_0() { + for &i in &[1, 10, 60] { + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(i)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(i)); + + assert_ready!(delay); + }); + } +} + +#[test] +fn sub_ms_delayed_delay() { + mocked(|timer, time| { + for _ in 0..5 { + let deadline = time.now() + Duration::from_millis(1) + Duration::new(0, 1); + + let mut delay = Delay::new(deadline); + + assert_not_ready!(delay); + + turn(timer, None); + assert_ready!(delay); + + assert!(time.now() >= deadline); + + time.advance(Duration::new(0, 1)); + } + }); +} + +#[test] +fn delayed_delay_wrapping_level_0() { + mocked(|timer, time| { + turn(timer, ms(5)); + assert_eq!(time.advanced(), ms(5)); + + let mut delay = Delay::new(time.now() + ms(60)); + + assert_not_ready!(delay); + + turn(timer, None); + assert_eq!(time.advanced(), ms(64)); + assert_not_ready!(delay); + + turn(timer, None); + assert_eq!(time.advanced(), ms(65)); + + assert_ready!(delay); + }); +} + +#[test] +fn timer_wrapping_with_higher_levels() { + mocked(|timer, time| { + // Set delay to hit level 1 + let mut s1 = Delay::new(time.now() + ms(64)); + assert_not_ready!(s1); + + // Turn a bit + turn(timer, ms(5)); + + // Set timeout such that it will hit level 0, but wrap + let mut s2 = Delay::new(time.now() + ms(60)); + assert_not_ready!(s2); + + // This should result in s1 firing + turn(timer, None); + assert_eq!(time.advanced(), ms(64)); + + assert_ready!(s1); + assert_not_ready!(s2); + + turn(timer, None); + assert_eq!(time.advanced(), ms(65)); + + assert_ready!(s2); + }); +} + +#[test] +fn delay_with_deadline_in_past() { + mocked(|timer, time| { + // Create `Delay` that elapsed immediately. + let mut delay = Delay::new(time.now() - ms(100)); + + // Even though the delay expires in the past, it is not ready yet + // because the timer must observe it. + assert_ready!(delay); + + // Turn the timer, it runs for the elapsed time + turn(timer, ms(1000)); + + // The time has not advanced. The `turn` completed immediately. + assert_eq!(time.advanced(), ms(1000)); + }); +} + +#[test] +fn delayed_delay_level_1() { + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(234)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + // Turn the timer, this will wake up to cascade the timer down. + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(192)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + // Turn the timer again + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(234)); + + // The delay has elapsed. + assert_ready!(delay); + }); + + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(234)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + // Turn the timer with a smaller timeout than the cascade. + turn(timer, ms(100)); + assert_eq!(time.advanced(), ms(100)); + + assert_not_ready!(delay); + + // Turn the timer, this will wake up to cascade the timer down. + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(192)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + // Turn the timer again + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(234)); + + // The delay has elapsed. + assert_ready!(delay); + }); +} + +#[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 delay = Delay::new(now + ms(500)); + + mocked_with_now(now, |timer, time| { + // This registers the delay with the timer + assert_not_ready!(delay); + + // Wait some time... the timer is cascading + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(448)); + + assert_not_ready!(delay); + + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(500)); + + // The delay has elapsed + assert_ready!(delay); + }); +} + +#[test] +fn concurrently_set_two_timers_second_one_shorter() { + mocked(|timer, time| { + let mut delay1 = Delay::new(time.now() + ms(500)); + let mut delay2 = Delay::new(time.now() + ms(200)); + + // The delay has not elapsed + assert_not_ready!(delay1); + assert_not_ready!(delay2); + + // Delay until a cascade + turn(timer, None); + assert_eq!(time.advanced(), ms(192)); + + // Delay until the second timer. + turn(timer, None); + assert_eq!(time.advanced(), ms(200)); + + // The shorter delay fires + assert_ready!(delay2); + assert_not_ready!(delay1); + + turn(timer, None); + assert_eq!(time.advanced(), ms(448)); + + assert_not_ready!(delay1); + + // Turn again, this time the time will advance to the second delay + turn(timer, None); + assert_eq!(time.advanced(), ms(500)); + + assert_ready!(delay1); + }) +} + +#[test] +fn short_delay() { + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(1)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + // Turn the timer, but not enough time will go by. + turn(timer, None); + + // The delay has elapsed. + assert_ready!(delay); + + // The time has advanced to the point of the delay elapsing. + assert_eq!(time.advanced(), ms(1)); + }) +} + +#[test] +fn sorta_long_delay() { + const MIN_5: u64 = 5 * 60 * 1000; + + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(MIN_5)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + let cascades = &[262_144, 262_144 + 9 * 4096, 262_144 + 9 * 4096 + 15 * 64]; + + for &elapsed in cascades { + turn(timer, None); + assert_eq!(time.advanced(), ms(elapsed)); + + assert_not_ready!(delay); + } + + turn(timer, None); + assert_eq!(time.advanced(), ms(MIN_5)); + + // The delay has elapsed. + assert_ready!(delay); + }) +} + +#[test] +fn very_long_delay() { + const MO_5: u64 = 5 * 30 * 24 * 60 * 60 * 1000; + + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(MO_5)); + + // The delay has not elapsed. + assert_not_ready!(delay); + + let cascades = &[ + 12_884_901_888, + 12_952_010_752, + 12_959_875_072, + 12_959_997_952, + ]; + + for &elapsed in cascades { + turn(timer, None); + assert_eq!(time.advanced(), ms(elapsed)); + + assert_not_ready!(delay); + } + + // Turn the timer, but not enough time will go by. + turn(timer, None); + + // The time has advanced to the point of the delay elapsing. + assert_eq!(time.advanced(), ms(MO_5)); + + // The delay has elapsed. + assert_ready!(delay); + }) +} + +#[test] +fn greater_than_max() { + const YR_5: u64 = 5 * 365 * 24 * 60 * 60 * 1000; + + mocked(|timer, time| { + // Create a `Delay` that elapses in the future + let mut delay = Delay::new(time.now() + ms(YR_5)); + + assert_not_ready!(delay); + + turn(timer, ms(0)); + + assert!(delay.poll().is_err()); + }) +} + +#[test] +fn unpark_is_delayed() { + mocked(|timer, time| { + let mut delay1 = Delay::new(time.now() + ms(100)); + let mut delay2 = Delay::new(time.now() + ms(101)); + let mut delay3 = Delay::new(time.now() + ms(200)); + + assert_not_ready!(delay1); + assert_not_ready!(delay2); + assert_not_ready!(delay3); + + time.park_for(ms(500)); + + turn(timer, None); + + assert_eq!(time.advanced(), ms(500)); + + assert_ready!(delay1); + assert_ready!(delay2); + assert_ready!(delay3); + }) +} + +#[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; + + mocked(|timer, time| { + for _ in 0..5 { + turn(timer, ms(YR_1)); + } + + let mut delay = Delay::new(time.now() + ms(1)); + assert_not_ready!(delay); + + turn(timer, ms(1000)); + assert_eq!(time.advanced(), Duration::from_millis(YR_5) + ms(1)); + + assert_ready!(delay); + }); +} + +#[test] +fn reset_future_delay_before_fire() { + mocked(|timer, time| { + let mut delay = Delay::new(time.now() + ms(100)); + + assert_not_ready!(delay); + + delay.reset(time.now() + ms(200)); + + turn(timer, None); + assert_eq!(time.advanced(), ms(192)); + + assert_not_ready!(delay); + + turn(timer, None); + assert_eq!(time.advanced(), ms(200)); + + assert_ready!(delay); + }); +} + +#[test] +fn reset_past_delay_before_turn() { + mocked(|timer, time| { + let mut delay = Delay::new(time.now() + ms(100)); + + assert_not_ready!(delay); + + delay.reset(time.now() + ms(80)); + + turn(timer, None); + assert_eq!(time.advanced(), ms(64)); + + assert_not_ready!(delay); + + turn(timer, None); + assert_eq!(time.advanced(), ms(80)); + + assert_ready!(delay); + }); +} + +#[test] +fn reset_past_delay_before_fire() { + mocked(|timer, time| { + let mut delay = Delay::new(time.now() + ms(100)); + + assert_not_ready!(delay); + turn(timer, ms(10)); + + assert_not_ready!(delay); + delay.reset(time.now() + ms(80)); + + turn(timer, None); + assert_eq!(time.advanced(), ms(64)); + + assert_not_ready!(delay); + + turn(timer, None); + assert_eq!(time.advanced(), ms(90)); + + assert_ready!(delay); + }); +} + +#[test] +fn reset_future_delay_after_fire() { + mocked(|timer, time| { + let mut delay = Delay::new(time.now() + ms(100)); + + assert_not_ready!(delay); + + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(64)); + + turn(timer, None); + assert_eq!(time.advanced(), ms(100)); + + assert_ready!(delay); + + delay.reset(time.now() + ms(10)); + assert_not_ready!(delay); + + turn(timer, ms(1000)); + assert_eq!(time.advanced(), ms(110)); + + assert_ready!(delay); + }); +} + +#[test] +fn delay_with_default_handle() { + let handle = Handle::default(); + let now = Instant::now(); + + let mut delay = handle.delay(now + ms(1)); + + mocked_with_now(now, |timer, _time| { + assert_not_ready!(delay); + + turn(timer, ms(1)); + + assert_ready!(delay); + }); +} diff --git a/third_party/rust/tokio-timer/tests/hammer.rs b/third_party/rust/tokio-timer/tests/hammer.rs new file mode 100644 index 0000000000..9986a9e659 --- /dev/null +++ b/third_party/rust/tokio-timer/tests/hammer.rs @@ -0,0 +1,241 @@ +extern crate futures; +extern crate rand; +extern crate tokio_executor; +extern crate tokio_timer; + +use tokio_executor::park::{Park, Unpark, UnparkThread}; +use tokio_timer::*; + +use futures::stream::FuturesUnordered; +use futures::{Future, Stream}; +use rand::Rng; + +use std::cmp; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering::SeqCst; +use std::sync::{Arc, Barrier}; +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 = FuturesUnordered::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)); + + exec.push({ + handle.delay(deadline).and_then(move |_| { + let now = Instant::now(); + assert!(now >= deadline, "deadline greater by {:?}", deadline - now); + Ok(()) + }) + }); + } + + // Run the logic + exec.for_each(|_| Ok(())).wait().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 = FuturesUnordered::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 = + Instant::now() + Duration::from_millis(rng.gen_range(MIN_DELAY, MAX_DELAY)); + + let deadline = cmp::min(deadline1, deadline2); + + let delay = handle.delay(deadline1); + let join = handle.timeout(delay, deadline2); + + exec.push({ + join.and_then(move |_| { + let now = Instant::now(); + assert!(now >= deadline, "deadline greater by {:?}", deadline - now); + Ok(()) + }) + }); + } + + // Run the logic + exec.or_else(|e| { + assert!(e.is_elapsed()); + Ok::<_, ()>(()) + }) + .for_each(|_| Ok(())) + .wait() + .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 = FuturesUnordered::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)); + + exec.push({ + handle + .delay(deadline1) + // Select over a second delay + .select2(handle.delay(deadline2)) + .map_err(|e| panic!("boom; err={:?}", e)) + .and_then(move |res| { + use futures::future::Either::*; + + let now = Instant::now(); + assert!( + now >= deadline1, + "deadline greater by {:?}", + deadline1 - now + ); + + let mut other = match res { + A((_, other)) => other, + B((_, other)) => other, + }; + + other.reset(deadline3); + other + }) + .and_then(move |_| { + let now = Instant::now(); + assert!( + now >= deadline3, + "deadline greater by {:?}", + deadline3 - now + ); + Ok(()) + }) + }); + } + + // Run the logic + exec.for_each(|_| Ok(())).wait().unwrap(); + + if 1 == done.rem.fetch_sub(1, SeqCst) { + done.unpark.unpark(); + } + }); + } + + while done.rem.load(SeqCst) > 0 { + timer.turn(None).unwrap(); + } + } +} diff --git a/third_party/rust/tokio-timer/tests/interval.rs b/third_party/rust/tokio-timer/tests/interval.rs new file mode 100644 index 0000000000..27828fc83a --- /dev/null +++ b/third_party/rust/tokio-timer/tests/interval.rs @@ -0,0 +1,46 @@ +extern crate futures; +extern crate tokio_executor; +extern crate tokio_timer; + +#[macro_use] +mod support; +use support::*; + +use tokio_timer::*; + +use futures::Stream; + +#[test] +#[should_panic] +fn interval_zero_duration() { + mocked(|_, time| { + let _ = Interval::new(time.now(), ms(0)); + }); +} + +#[test] +fn usage() { + mocked(|timer, time| { + let start = time.now(); + let mut int = Interval::new(start, ms(300)); + + assert_ready_eq!(int, Some(start)); + assert_not_ready!(int); + + advance(timer, ms(100)); + assert_not_ready!(int); + + advance(timer, ms(200)); + assert_ready_eq!(int, Some(start + ms(300))); + assert_not_ready!(int); + + advance(timer, ms(400)); + assert_ready_eq!(int, Some(start + ms(600))); + assert_not_ready!(int); + + advance(timer, ms(500)); + assert_ready_eq!(int, Some(start + ms(900))); + assert_ready_eq!(int, Some(start + ms(1200))); + assert_not_ready!(int); + }); +} diff --git a/third_party/rust/tokio-timer/tests/queue.rs b/third_party/rust/tokio-timer/tests/queue.rs new file mode 100644 index 0000000000..613b6c6ddd --- /dev/null +++ b/third_party/rust/tokio-timer/tests/queue.rs @@ -0,0 +1,406 @@ +extern crate futures; +extern crate tokio_executor; +extern crate tokio_mock_task; +extern crate tokio_timer; + +#[macro_use] +mod support; +use support::*; + +use tokio_mock_task::MockTask; +use tokio_timer::*; + +use futures::Stream; + +#[test] +fn single_immediate_delay() { + mocked(|_timer, time| { + let mut queue = DelayQueue::new(); + let _key = queue.insert_at("foo", time.now()); + + let entry = assert_ready!(queue).unwrap(); + assert_eq!(*entry.get_ref(), "foo"); + + let entry = assert_ready!(queue); + assert!(entry.is_none()) + }); +} + +#[test] +fn multi_immediate_delays() { + mocked(|_timer, time| { + let mut queue = DelayQueue::new(); + + let _k = queue.insert_at("1", time.now()); + let _k = queue.insert_at("2", time.now()); + let _k = queue.insert_at("3", time.now()); + + let mut res = vec![]; + + while res.len() < 3 { + let entry = assert_ready!(queue).unwrap(); + res.push(entry.into_inner()); + } + + let entry = assert_ready!(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() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let _key = queue.insert_at("foo", time.now() + ms(5)); + + let mut task = MockTask::new(); + + task.enter(|| { + assert_not_ready!(queue); + }); + + turn(timer, ms(1)); + + assert!(!task.is_notified()); + + turn(timer, ms(5)); + + assert!(task.is_notified()); + + let entry = assert_ready!(queue).unwrap(); + assert_eq!(*entry.get_ref(), "foo"); + + let entry = assert_ready!(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]; + + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + // Setup the delays + for &i in delays { + let _key = queue.insert_at(i, time.now() + ms(i)); + } + + task.enter(|| { + assert_not_ready!(queue); + }); + + assert!(!task.is_notified()); + + for elapsed in 0..1200 { + turn(timer, ms(1)); + let elapsed = elapsed + 1; + + if delays.contains(&elapsed) { + assert!(task.is_notified()); + + task.enter(|| { + assert_ready!(queue); + assert_not_ready!(queue); + }); + } else { + if task.is_notified() { + let cascade = &[192, 960]; + assert!(cascade.contains(&elapsed), "elapsed={}", elapsed); + + task.enter(|| { + assert_not_ready!(queue, "elapsed={}", elapsed); + }); + } + } + } + }); +} + +#[test] +fn insert_in_past_fires_immediately() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + + let now = time.now(); + + turn(timer, ms(10)); + + queue.insert_at("foo", now); + + assert_ready!(queue); + }); +} + +#[test] +fn remove_entry() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let key = queue.insert_at("foo", time.now() + ms(5)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + let entry = queue.remove(&key); + assert_eq!(entry.into_inner(), "foo"); + + turn(timer, ms(10)); + + task.enter(|| { + let entry = assert_ready!(queue); + assert!(entry.is_none()); + }); + }); +} + +#[test] +fn reset_entry() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let now = time.now(); + let key = queue.insert_at("foo", now + ms(5)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + turn(timer, ms(1)); + + queue.reset_at(&key, now + ms(10)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + turn(timer, ms(7)); + + assert!(!task.is_notified()); + + task.enter(|| { + assert_not_ready!(queue); + }); + + turn(timer, ms(3)); + + assert!(task.is_notified()); + + let entry = assert_ready!(queue).unwrap(); + assert_eq!(*entry.get_ref(), "foo"); + + let entry = assert_ready!(queue); + assert!(entry.is_none()) + }); +} + +#[test] +fn reset_much_later() { + // Reproduces tokio-rs/tokio#849. + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + turn(timer, ms(1)); + + let key = queue.insert_at("foo", epoch + ms(200)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + turn(timer, ms(3)); + + queue.reset_at(&key, epoch + ms(5)); + + turn(timer, ms(20)); + + assert!(task.is_notified()); + }); +} + +#[test] +fn reset_twice() { + // Reproduces tokio-rs/tokio#849. + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + turn(timer, ms(1)); + + let key = queue.insert_at("foo", epoch + ms(200)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + turn(timer, ms(3)); + + queue.reset_at(&key, epoch + ms(50)); + + turn(timer, ms(20)); + + queue.reset_at(&key, epoch + ms(40)); + + turn(timer, ms(20)); + + assert!(task.is_notified()); + }); +} + +#[test] +fn remove_expired_item() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + + let now = time.now(); + + turn(timer, 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() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + queue.insert_at("foo", epoch + ms(10_000)); + + // Delay should be set to 8.192s here. + task.enter(|| { + assert_not_ready!(queue); + }); + + // Delay should be set to the delay of the new item here + queue.insert_at("bar", epoch + ms(600)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + advance(timer, ms(600)); + + assert!(task.is_notified()); + let entry = assert_ready!(queue).unwrap().into_inner(); + assert_eq!(entry, "bar"); + }) +} + +#[test] +fn multi_reset() { + mocked(|_, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + let foo = queue.insert_at("foo", epoch + ms(200)); + let bar = queue.insert_at("bar", epoch + ms(250)); + + task.enter(|| { + assert_not_ready!(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() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + let foo = queue.insert_at("foo", epoch + ms(200)); + queue.insert_at("bar", epoch + ms(250)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + queue.reset_at(&foo, epoch + ms(100)); + + advance(timer, ms(100)); + + assert!(task.is_notified()); + let entry = assert_ready!(queue).unwrap().into_inner(); + assert_eq!(entry, "foo"); + }) +} + +#[test] +fn expire_second_key_when_reset_to_expire_earlier() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + queue.insert_at("foo", epoch + ms(200)); + let bar = queue.insert_at("bar", epoch + ms(250)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + queue.reset_at(&bar, epoch + ms(100)); + + advance(timer, ms(100)); + + assert!(task.is_notified()); + let entry = assert_ready!(queue).unwrap().into_inner(); + assert_eq!(entry, "bar"); + }) +} + +#[test] +fn reset_first_expiring_item_to_expire_later() { + mocked(|timer, time| { + let mut queue = DelayQueue::new(); + let mut task = MockTask::new(); + + let epoch = time.now(); + + let foo = queue.insert_at("foo", epoch + ms(200)); + let _bar = queue.insert_at("bar", epoch + ms(250)); + + task.enter(|| { + assert_not_ready!(queue); + }); + + queue.reset_at(&foo, epoch + ms(300)); + advance(timer, ms(250)); + + assert!(task.is_notified()); + let entry = assert_ready!(queue).unwrap().into_inner(); + assert_eq!(entry, "bar"); + }) +} diff --git a/third_party/rust/tokio-timer/tests/support/mod.rs b/third_party/rust/tokio-timer/tests/support/mod.rs new file mode 100644 index 0000000000..244d56b819 --- /dev/null +++ b/third_party/rust/tokio-timer/tests/support/mod.rs @@ -0,0 +1,261 @@ +#![allow(unused_macros, unused_imports, dead_code, deprecated)] + +use tokio_executor::park::{Park, Unpark}; +use tokio_timer::clock::Now; +use tokio_timer::timer::Timer; + +use futures::future::{lazy, Future}; + +use std::marker::PhantomData; +use std::rc::Rc; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +macro_rules! assert_ready { + ($f:expr) => {{ + use ::futures::Async::*; + + match $f.poll().unwrap() { + Ready(v) => v, + NotReady => panic!("NotReady"), + } + }}; + ($f:expr, $($msg:expr),+) => {{ + use ::futures::Async::*; + + match $f.poll().unwrap() { + Ready(v) => v, + NotReady => { + let msg = format!($($msg),+); + panic!("NotReady; {}", msg) + } + } + }} +} + +macro_rules! assert_ready_eq { + ($f:expr, $expect:expr) => { + assert_eq!($f.poll().unwrap(), ::futures::Async::Ready($expect)); + }; +} + +macro_rules! assert_not_ready { + ($f:expr) => {{ + let res = $f.poll().unwrap(); + assert!(!res.is_ready(), "actual={:?}", res) + }}; + ($f:expr, $($msg:expr),+) => {{ + let res = $f.poll().unwrap(); + if res.is_ready() { + let msg = format!($($msg),+); + panic!("actual={:?}; {}", res, msg); + } + }}; +} + +macro_rules! assert_elapsed { + ($f:expr) => { + assert!($f.poll().unwrap_err().is_elapsed()); + }; +} + +#[derive(Debug)] +pub struct MockTime { + inner: Inner, + _p: PhantomData<Rc<()>>, +} + +#[derive(Debug)] +pub struct MockNow { + inner: Inner, +} + +#[derive(Debug)] +pub struct MockPark { + inner: Inner, + _p: PhantomData<Rc<()>>, +} + +#[derive(Debug)] +pub struct MockUnpark { + inner: Inner, +} + +type Inner = Arc<Mutex<State>>; + +#[derive(Debug)] +struct State { + base: Instant, + advance: Duration, + unparked: bool, + park_for: Option<Duration>, +} + +pub fn ms(num: u64) -> Duration { + Duration::from_millis(num) +} + +pub trait IntoTimeout { + fn into_timeout(self) -> Option<Duration>; +} + +impl IntoTimeout for Option<Duration> { + fn into_timeout(self) -> Self { + self + } +} + +impl IntoTimeout for Duration { + fn into_timeout(self) -> Option<Duration> { + Some(self) + } +} + +/// Turn the timer state once +pub fn turn<T: IntoTimeout>(timer: &mut Timer<MockPark>, duration: T) { + timer.turn(duration.into_timeout()).unwrap(); +} + +/// Advance the timer the specified amount +pub fn advance(timer: &mut Timer<MockPark>, duration: Duration) { + let inner = timer.get_park().inner.clone(); + let deadline = inner.lock().unwrap().now() + duration; + + while inner.lock().unwrap().now() < deadline { + let dur = deadline - inner.lock().unwrap().now(); + turn(timer, dur); + } +} + +pub fn mocked<F, R>(f: F) -> R +where + F: FnOnce(&mut Timer<MockPark>, &mut MockTime) -> R, +{ + mocked_with_now(Instant::now(), f) +} + +pub fn mocked_with_now<F, R>(now: Instant, f: F) -> R +where + F: FnOnce(&mut Timer<MockPark>, &mut MockTime) -> R, +{ + let mut time = MockTime::new(now); + let park = time.mock_park(); + let now = ::tokio_timer::clock::Clock::new_with_now(time.mock_now()); + + let mut enter = ::tokio_executor::enter().unwrap(); + + ::tokio_timer::clock::with_default(&now, &mut enter, |enter| { + let mut timer = Timer::new(park); + let handle = timer.handle(); + + ::tokio_timer::with_default(&handle, enter, |_| { + lazy(|| Ok::<_, ()>(f(&mut timer, &mut time))) + .wait() + .unwrap() + }) + }) +} + +impl MockTime { + pub fn new(now: Instant) -> MockTime { + let state = State { + base: now, + advance: Duration::default(), + unparked: false, + park_for: None, + }; + + MockTime { + inner: Arc::new(Mutex::new(state)), + _p: PhantomData, + } + } + + pub fn mock_now(&self) -> MockNow { + let inner = self.inner.clone(); + MockNow { inner } + } + + pub fn mock_park(&self) -> MockPark { + let inner = self.inner.clone(); + MockPark { + inner, + _p: PhantomData, + } + } + + pub fn now(&self) -> Instant { + self.inner.lock().unwrap().now() + } + + /// Returns the total amount of time the time has been advanced. + pub fn advanced(&self) -> Duration { + self.inner.lock().unwrap().advance + } + + pub fn advance(&self, duration: Duration) { + let mut inner = self.inner.lock().unwrap(); + inner.advance(duration); + } + + /// The next call to park_timeout will be for this duration, regardless of + /// the timeout passed to `park_timeout`. + pub fn park_for(&self, duration: Duration) { + self.inner.lock().unwrap().park_for = Some(duration); + } +} + +impl Park for MockPark { + type Unpark = MockUnpark; + type Error = (); + + fn unpark(&self) -> Self::Unpark { + let inner = self.inner.clone(); + MockUnpark { inner } + } + + fn park(&mut self) -> Result<(), Self::Error> { + let mut inner = self.inner.lock().map_err(|_| ())?; + + let duration = inner.park_for.take().expect("call park_for first"); + + inner.advance(duration); + Ok(()) + } + + fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> { + let mut inner = self.inner.lock().unwrap(); + + if let Some(duration) = inner.park_for.take() { + inner.advance(duration); + } else { + inner.advance(duration); + } + + Ok(()) + } +} + +impl Unpark for MockUnpark { + fn unpark(&self) { + if let Ok(mut inner) = self.inner.lock() { + inner.unparked = true; + } + } +} + +impl Now for MockNow { + fn now(&self) -> Instant { + self.inner.lock().unwrap().now() + } +} + +impl State { + fn now(&self) -> Instant { + self.base + self.advance + } + + fn advance(&mut self, duration: Duration) { + self.advance += duration; + } +} diff --git a/third_party/rust/tokio-timer/tests/throttle.rs b/third_party/rust/tokio-timer/tests/throttle.rs new file mode 100644 index 0000000000..253309c988 --- /dev/null +++ b/third_party/rust/tokio-timer/tests/throttle.rs @@ -0,0 +1,51 @@ +extern crate futures; +extern crate tokio_executor; +extern crate tokio_timer; + +#[macro_use] +mod support; +use support::*; + +use futures::{prelude::*, sync::mpsc}; +use tokio_timer::throttle::Throttle; + +#[test] +fn throttle() { + mocked(|timer, _| { + let (tx, rx) = mpsc::unbounded(); + let mut stream = Throttle::new(rx, ms(1)); + + assert_not_ready!(stream); + + for i in 0..3 { + tx.unbounded_send(i).unwrap(); + } + for i in 0..3 { + assert_ready_eq!(stream, Some(i)); + assert_not_ready!(stream); + + advance(timer, ms(1)); + } + + assert_not_ready!(stream); + }); +} + +#[test] +fn throttle_dur_0() { + mocked(|_, _| { + let (tx, rx) = mpsc::unbounded(); + let mut stream = Throttle::new(rx, ms(0)); + + assert_not_ready!(stream); + + for i in 0..3 { + tx.unbounded_send(i).unwrap(); + } + for i in 0..3 { + assert_ready_eq!(stream, Some(i)); + } + + assert_not_ready!(stream); + }); +} diff --git a/third_party/rust/tokio-timer/tests/timeout.rs b/third_party/rust/tokio-timer/tests/timeout.rs new file mode 100644 index 0000000000..8e10776ae7 --- /dev/null +++ b/third_party/rust/tokio-timer/tests/timeout.rs @@ -0,0 +1,179 @@ +extern crate futures; +extern crate tokio_executor; +extern crate tokio_timer; + +#[macro_use] +mod support; +use support::*; + +use tokio_timer::*; + +use futures::sync::{mpsc, oneshot}; +use futures::{future, Future, Stream}; + +#[test] +fn simultaneous_deadline_future_completion() { + mocked(|_, time| { + // Create a future that is immediately ready + let fut = future::ok::<_, ()>(()); + + // Wrap it with a deadline + let mut fut = Timeout::new_at(fut, time.now()); + + // Ready! + assert_ready!(fut); + }); +} + +#[test] +fn completed_future_past_deadline() { + mocked(|_, time| { + // Create a future that is immediately ready + let fut = future::ok::<_, ()>(()); + + // Wrap it with a deadline + let mut fut = Timeout::new_at(fut, time.now() - ms(1000)); + + // Ready! + assert_ready!(fut); + }); +} + +#[test] +fn future_and_deadline_in_future() { + mocked(|timer, time| { + // Not yet complete + let (tx, rx) = oneshot::channel(); + + // Wrap it with a deadline + let mut fut = Timeout::new_at(rx, time.now() + ms(100)); + + // Ready! + assert_not_ready!(fut); + + // Turn the timer, it runs for the elapsed time + advance(timer, ms(90)); + + assert_not_ready!(fut); + + // Complete the future + tx.send(()).unwrap(); + + assert_ready!(fut); + }); +} + +#[test] +fn future_and_timeout_in_future() { + mocked(|timer, _time| { + // Not yet complete + let (tx, rx) = oneshot::channel(); + + // Wrap it with a deadline + let mut fut = Timeout::new(rx, ms(100)); + + // Ready! + assert_not_ready!(fut); + + // Turn the timer, it runs for the elapsed time + advance(timer, ms(90)); + + assert_not_ready!(fut); + + // Complete the future + tx.send(()).unwrap(); + + assert_ready!(fut); + }); +} + +#[test] +fn deadline_now_elapses() { + mocked(|_, time| { + let fut = future::empty::<(), ()>(); + + // Wrap it with a deadline + let mut fut = Timeout::new_at(fut, time.now()); + + assert_elapsed!(fut); + }); +} + +#[test] +fn deadline_future_elapses() { + mocked(|timer, time| { + let fut = future::empty::<(), ()>(); + + // Wrap it with a deadline + let mut fut = Timeout::new_at(fut, time.now() + ms(300)); + + assert_not_ready!(fut); + + advance(timer, ms(300)); + + assert_elapsed!(fut); + }); +} + +#[test] +fn future_errors_first() { + mocked(|_, time| { + let fut = future::err::<(), ()>(()); + + // Wrap it with a deadline + let mut fut = Timeout::new_at(fut, time.now() + ms(100)); + + // Ready! + assert!(fut.poll().unwrap_err().is_inner()); + }); +} + +#[test] +fn stream_and_timeout_in_future() { + mocked(|timer, _time| { + // Not yet complete + let (tx, rx) = mpsc::unbounded(); + + // Wrap it with a deadline + let mut stream = Timeout::new(rx, ms(100)); + + // Not ready + assert_not_ready!(stream); + + // Turn the timer, it runs for the elapsed time + advance(timer, ms(90)); + + assert_not_ready!(stream); + + // Complete the future + tx.unbounded_send(()).unwrap(); + + let item = assert_ready!(stream); + assert!(item.is_some()); + }); +} + +#[test] +fn idle_stream_timesout_periodically() { + mocked(|timer, _time| { + // Not yet complete + let (_tx, rx) = mpsc::unbounded::<()>(); + + // Wrap it with a deadline + let mut stream = Timeout::new(rx, ms(100)); + + // Not ready + assert_not_ready!(stream); + + // Turn the timer, it runs for the elapsed time + advance(timer, ms(100)); + + assert_elapsed!(stream); + // Stream's timeout should reset + assert_not_ready!(stream); + + // Turn the timer, it runs for the elapsed time + advance(timer, ms(100)); + assert_elapsed!(stream); + }); +} |