summaryrefslogtreecommitdiffstats
path: root/tokio/src/time
diff options
context:
space:
mode:
authorbdonlan <bdonlan@gmail.com>2020-10-05 14:25:04 -0700
committerGitHub <noreply@github.com>2020-10-05 14:25:04 -0700
commit9730317e94cd5bfca237376549405a6feb815223 (patch)
tree333c03937b5c804b4c79335cdb8d7d68cf665e44 /tokio/src/time
parent02311dcfa13d719c3f591c922f8a719d7c954ef0 (diff)
time: move DelayQueue to tokio-util (#2897)
This change is intended to do the minimum to unblock 0.3; as such, for now, we duplicate the internal `time::wheel` structures in tokio-util, rather than trying to refactor things at this stage. Co-authored-by: Bryan Donlan <bdonlan@amazon.com>
Diffstat (limited to 'tokio/src/time')
-rw-r--r--tokio/src/time/delay_queue.rs897
-rw-r--r--tokio/src/time/mod.rs7
2 files changed, 0 insertions, 904 deletions
diff --git a/tokio/src/time/delay_queue.rs b/tokio/src/time/delay_queue.rs
deleted file mode 100644
index 910f75fb..00000000
--- a/tokio/src/time/delay_queue.rs
+++ /dev/null
@@ -1,897 +0,0 @@
-//! A queue of delayed elements.
-//!
-//! See [`DelayQueue`] for more details.
-//!
-//! [`DelayQueue`]: struct@DelayQueue
-
-use crate::time::wheel::{self, Wheel};
-use crate::time::{sleep_until, Delay, Duration, Error, Instant};
-
-use slab::Slab;
-use std::cmp;
-use std::future::Future;
-use std::marker::PhantomData;
-use std::pin::Pin;
-use std::task::{self, Poll};
-
-/// 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_expired`] is called. If an entry has reached its
-/// deadline, it is returned. If not, `Poll::Pending` indicating that the
-/// current task will be notified once the deadline has been reached.
-///
-/// # `Stream` implementation
-///
-/// Items are retrieved from the queue via [`DelayQueue::poll_expired`]. 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 a separate instance of the same timer wheel used internally by
-/// Tokio's standalone timer utilities such as [`sleep`]. Because of this, it offers the same
-/// performance and scalability benefits.
-///
-/// State associated with each entry is stored in a [`slab`]. This amortizes the cost of allocation,
-/// and allows reuse of the memory allocated for expired entires.
-///
-/// Capacity can be checked using [`capacity`] and allocated preemptively by using
-/// the [`reserve`] method.
-///
-/// # Usage
-///
-/// Using `DelayQueue` to manage cache entries.
-///
-/// ```rust,no_run
-/// use tokio::time::{delay_queue, DelayQueue, Error};
-///
-/// use futures::ready;
-/// use std::collections::HashMap;
-/// use std::task::{Context, Poll};
-/// use std::time::Duration;
-/// # type CacheKey = String;
-/// # type Value = String;
-///
-/// struct Cache {
-/// entries: HashMap<CacheKey, (Value, delay_queue::Key)>,
-/// expirations: DelayQueue<CacheKey>,
-/// }
-///
-/// const TTL_SECS: u64 = 30;
-///
-/// impl Cache {
-/// fn insert(&mut self, key: CacheKey, value: Value) {
-/// let delay = self.expirations
-/// .insert(key.clone(), Duration::from_secs(TTL_SECS));
-///
-/// self.entries.insert(key, (value, delay));
-/// }
-///
-/// fn get(&self, key: &CacheKey) -> Option<&Value> {
-/// self.entries.get(key)
-/// .map(|&(ref v, _)| v)
-/// }
-///
-/// fn remove(&mut self, key: &CacheKey) {
-/// if let Some((_, cache_key)) = self.entries.remove(key) {
-/// self.expirations.remove(&cache_key);
-/// }
-/// }
-///
-/// fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
-/// while let Some(res) = ready!(self.expirations.poll_expired(cx)) {
-/// let entry = res?;
-/// self.entries.remove(entry.get_ref());
-/// }
-///
-/// Poll::Ready(Ok(()))
-/// }
-/// }
-/// ```
-///
-/// [`insert`]: method@Self::insert
-/// [`insert_at`]: method@Self::insert_at
-/// [`Key`]: struct@Key
-/// [`Stream`]: https://docs.rs/futures/0.1/futures/stream/trait.Stream.html
-/// [`poll_expired`]: method@Self::poll_expired
-/// [`Stream::poll_expired`]: method@Self::poll_expired
-/// [`DelayQueue`]: struct@DelayQueue
-/// [`sleep`]: fn@super::sleep
-/// [`slab`]: slab
-/// [`capacity`]: method@Self::capacity
-/// [`reserve`]: method@Self::reserve
-#[derive(Debug)]
-pub struct DelayQueue<T> {
- /// 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_expired`].
-///
-/// [`DelayQueue::poll_expired`]: method@DelayQueue::poll_expired
-#[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
-/// [`DelayQueue::insert`]: method@DelayQueue::insert
-#[derive(Debug, Clone)]
-pub struct Key {
- index: usize,
-}
-
-#[derive(Debug)]
-struct Stack<T> {
- /// Head of the stack
- head: Option<usize>,
- _p: PhantomData<fn() -> T>,
-}
-
-#[derive(Debug)]
-struct Data<T> {
- /// The data being stored in the queue and will be returned at the requested
- /// instant.
- inner: T,
-
- /// The instant at which the item is returned.
- when: u64,
-
- /// Set to true when stored in the `expired` queue
- expired: bool,
-
- /// Next entry in the stack
- next: Option<usize>,
-
- /// Previous entry in the stack
- prev: Option<usize>,
-}
-
-/// Maximum number of entries the queue can handle
-const MAX_ENTRIES: usize = (1 << 30) - 1;
-
-impl<T> DelayQueue<T> {
- /// Creates a new, empty, `DelayQueue`
- ///
- /// The queue will not allocate storage until items are inserted into it.
- ///
- /// # Examples
- ///
- /// ```rust
- /// # use tokio::time::DelayQueue;
- /// let delay_queue: DelayQueue<u32> = DelayQueue::new();
- /// ```
- pub fn new() -> DelayQueue<T> {
- DelayQueue::with_capacity(0)
- }
-
- /// Creates 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::time::DelayQueue;
- /// # use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async fn main() {
- /// 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 {
- wheel: Wheel::new(),
- slab: Slab::with_capacity(capacity),
- expired: Stack::default(),
- delay: None,
- poll: wheel::Poll::new(0),
- start: Instant::now(),
- }
- }
-
- /// Inserts `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_expired`]. 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_expired`] after
- /// `when` is reached or by calling [`remove`]. At this point, the caller
- /// must take care to not use the returned [`Key`] again as it may reference
- /// a different item in the queue.
- ///
- /// See [type] level documentation for more details.
- ///
- /// # Panics
- ///
- /// This function panics if `when` is too far in the future.
- ///
- /// # Examples
- ///
- /// Basic usage
- ///
- /// ```rust
- /// use tokio::time::{DelayQueue, Duration, Instant};
- ///
- /// # #[tokio::main]
- /// # async 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_expired`]: method@Self::poll_expired
- /// [`remove`]: method@Self::remove
- /// [`reset`]: method@Self::reset
- /// [`Key`]: struct@Key
- /// [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 {
- let delay_time = self.start + Duration::from_millis(when);
- if let Some(ref mut delay) = &mut self.delay {
- delay.reset(delay_time);
- } else {
- self.delay = Some(sleep_until(delay_time));
- }
- }
-
- Key::new(key)
- }
-
- /// Attempts to pull out the next value of the delay queue, registering the
- /// current task for wakeup if the value is not yet available, and returning
- /// None if the queue is exhausted.
- pub fn poll_expired(
- &mut self,
- cx: &mut task::Context<'_>,
- ) -> Poll<Option<Result<Expired<T>, Error>>> {
- let item = ready!(self.poll_idx(cx));
- Poll::Ready(item.map(|result| {
- result.map(|idx| {
- let data = self.slab.remove(idx);
- debug_assert!(data.next.is_none());
- debug_assert!(data.prev.is_none());
-
- Expired {
- key: Key::new(idx),
- data: data.inner,
- deadline: self.start + Duration::from_millis(data.when),
- }
- })
- }))
- }
-
- /// Inserts `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 `timeout` duration has
- /// elapsed after `insert` was called. At that point, `value` will
- /// be returned from [`poll_expired`]. If `timeout` a Duration of
- /// zero, then `value` is immediately made available to poll.
- ///
- /// The return value represents the insertion and is used as an
- /// argument to [`remove`] and [`reset`]. Note that [`Key`] is a
- /// token and is reused once `value` is removed from the queue
- /// either by calling [`poll_expired`] after `timeout` has elapsed
- /// or by calling [`remove`]. At this point, the caller must 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
- /// duration supported by the timer in the current `Runtime`.
- ///
- /// # Examples
- ///
- /// Basic usage
- ///
- /// ```rust
- /// use tokio::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async 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_expired`]: method@Self::poll_expired
- /// [`remove`]: method@Self::remove
- /// [`reset`]: method@Self::reset
- /// [`Key`]: struct@Key
- /// [type]: #
- pub fn insert(&mut self, value: T, timeout: Duration) -> Key {
- self.insert_at(value, Instant::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),
- }
- }
-
- /// Removes the key fom the expired queue or the timer wheel
- /// depending on its expiration status
- ///
- /// # Panics
- /// Panics if the key is not contained in the expired queue or the wheel
- fn remove_key(&mut self, key: &Key) {
- use crate::time::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);
- }
- }
-
- /// Removes the item associated with `key` from the queue.
- ///
- /// There must be an item associated with `key`. The function returns the
- /// removed item as well as the `Instant` at which it will the delay will
- /// have expired.
- ///
- /// # Panics
- ///
- /// The function panics if `key` is not contained by the queue.
- ///
- /// # Examples
- ///
- /// Basic usage
- ///
- /// ```rust
- /// use tokio::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async 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> {
- self.remove_key(key);
- let data = self.slab.remove(key.index);
-
- Expired {
- key: Key::new(key.index),
- data: data.inner,
- deadline: self.start + Duration::from_millis(data.when),
- }
- }
-
- /// Sets the delay of the item associated with `key` to expire at `when`.
- ///
- /// This function is identical to `reset` but takes an `Instant` instead of
- /// a `Duration`.
- ///
- /// The item remains in the queue but the delay is set to expire at `when`.
- /// If `when` is in the past, then the item is immediately made available to
- /// the caller.
- ///
- /// # Panics
- ///
- /// This function panics if `when` is too far in the future or if `key` is
- /// not contained by the queue.
- ///
- /// # Examples
- ///
- /// Basic usage
- ///
- /// ```rust
- /// use tokio::time::{DelayQueue, Duration, Instant};
- ///
- /// # #[tokio::main]
- /// # async 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.remove_key(key);
-
- // Normalize the deadline. Values cannot be set to expire in the past.
- let when = self.normalize_deadline(when);
-
- self.slab[key.index].when = when;
- self.insert_idx(when, key.index);
-
- let next_deadline = self.next_deadline();
- if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) {
- delay.reset(deadline);
- }
- }
-
- /// Returns the next time poll as determined by the wheel
- fn next_deadline(&mut self) -> Option<Instant> {
- self.wheel
- .poll_at()
- .map(|poll_at| self.start + Duration::from_millis(poll_at))
- }
-
- /// Sets the delay of the item associated with `key` to expire after
- /// `timeout`.
- ///
- /// This function is identical to `reset_at` but takes a `Duration` instead
- /// of an `Instant`.
- ///
- /// The item remains in the queue but the delay is set to expire after
- /// `timeout`. If `timeout` is zero, then the item is immediately made
- /// available to the caller.
- ///
- /// # Panics
- ///
- /// This function panics if `timeout` is greater than the maximum supported
- /// duration or if `key` is not contained by the queue.
- ///
- /// # Examples
- ///
- /// Basic usage
- ///
- /// ```rust
- /// use tokio::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async 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, Instant::now() + timeout);
- }
-
- /// Clears the queue, removing all items.
- ///
- /// After calling `clear`, [`poll_expired`] will return `Ok(Ready(None))`.
- ///
- /// Note that this method has no effect on the allocated capacity.
- ///
- /// [`poll_expired`]: method@Self::poll_expired
- ///
- /// # Examples
- ///
- /// ```rust
- /// use tokio::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async 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::time::DelayQueue;
- ///
- /// let delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
- /// assert_eq!(delay_queue.capacity(), 10);
- /// ```
- pub fn capacity(&self) -> usize {
- self.slab.capacity()
- }
-
- /// Returns the number of elements currently in the queue.
- ///
- /// # Examples
- ///
- /// ```rust
- /// use tokio::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async fn main() {
- /// let mut delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
- /// assert_eq!(delay_queue.len(), 0);
- /// delay_queue.insert(3, Duration::from_secs(5));
- /// assert_eq!(delay_queue.len(), 1);
- /// # }
- /// ```
- pub fn len(&self) -> usize {
- self.slab.len()
- }
-
- /// Reserves 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::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async fn main() {
- /// 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::time::DelayQueue;
- /// use std::time::Duration;
- ///
- /// # #[tokio::main]
- /// # async fn main() {
- /// let mut delay_queue = DelayQueue::new();
- /// assert!(delay_queue.is_empty());
- ///
- /// delay_queue.insert("hello", Duration::from_secs(5));
- /// assert!(!delay_queue.is_empty());
- /// # }
- /// ```
- pub fn is_empty(&self) -> bool {
- self.slab.is_empty()
- }
-
- /// Polls the queue, returning the index of the next slot in the slab that
- /// should be returned.
- ///
- /// A slot should be returned when the associated deadline has been reached.
- fn poll_idx(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Result<usize, Error>>> {
- use self::wheel::Stack;
-
- let expired = self.expired.pop(&mut self.slab);
-
- if expired.is_some() {
- return Poll::Ready(expired.map(Ok));
- }
-
- loop {
- if let Some(ref mut delay) = self.delay {
- if !delay.is_elapsed() {
- ready!(Pin::new(&mut *delay).poll(cx));
- }
-
- let now = crate::time::ms(delay.deadline() - self.start, crate::time::Round::Down);
-
- self.poll = wheel::Poll::new(now);
- }
-
- // We poll the wheel to get the next value out before finding the next deadline.
- let wheel_idx = self.wheel.poll(&mut self.poll, &mut self.slab);
-
- self.delay = self.next_deadline().map(sleep_until);
-
- if let Some(idx) = wheel_idx {
- return Poll::Ready(Some(Ok(idx)));
- }
-
- if self.delay.is_none() {
- return Poll::Ready(None);
- }
- }
- }
-
- fn normalize_deadline(&self, when: Instant) -> u64 {
- let when = if when < self.start {
- 0
- } else {
- crate::time::ms(when - self.start, crate::time::Round::Up)
- };
-
- cmp::max(when, self.wheel.elapsed())
- }
-}
-
-// We never put `T` in a `Pin`...
-impl<T> Unpin for DelayQueue<T> {}
-
-impl<T> Default for DelayQueue<T> {
- fn default() -> DelayQueue<T> {
- DelayQueue::new()
- }
-}
-
-#[cfg(feature = "stream")]
-impl<T> futures_core::Stream for DelayQueue<T> {
- // DelayQueue seems much more specific, where a user may care that it
- // has reached capacity, so return those errors instead of panicking.
- type Item = Result<Expired<T>, Error>;
-
- fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
- DelayQueue::poll_expired(self.get_mut(), cx)
- }
-}
-
-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
- }
-
- /// Returns the deadline that the expiration was set to.
- pub fn deadline(&self) -> Instant {
- self.deadline
- }
-}
diff --git a/tokio/src/time/mod.rs b/tokio/src/time/mod.rs
index c8c797d9..a68e11b5 100644
--- a/tokio/src/time/mod.rs
+++ b/tokio/src/time/mod.rs
@@ -14,9 +14,6 @@
//! 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 types are sufficient for handling a large number of scenarios
//! involving time.
//!
@@ -96,10 +93,6 @@ pub(crate) use self::clock::Clock;
#[cfg(feature = "test-util")]
pub use clock::{advance, pause, resume};
-pub mod delay_queue;
-#[doc(inline)]
-pub use delay_queue::DelayQueue;
-
mod delay;
pub use delay::{sleep, sleep_until, Delay};