broker_tokio/time/
delay_queue.rs

1//! A queue of delayed elements.
2//!
3//! See [`DelayQueue`] for more details.
4//!
5//! [`DelayQueue`]: struct.DelayQueue.html
6
7use crate::time::wheel::{self, Wheel};
8use crate::time::{delay_until, Delay, Duration, Error, Instant};
9
10use slab::Slab;
11use std::cmp;
12use std::future::Future;
13use std::marker::PhantomData;
14use std::pin::Pin;
15use std::task::{self, Poll};
16
17/// A queue of delayed elements.
18///
19/// Once an element is inserted into the `DelayQueue`, it is yielded once the
20/// specified deadline has been reached.
21///
22/// # Usage
23///
24/// Elements are inserted into `DelayQueue` using the [`insert`] or
25/// [`insert_at`] methods. A deadline is provided with the item and a [`Key`] is
26/// returned. The key is used to remove the entry or to change the deadline at
27/// which it should be yielded back.
28///
29/// Once delays have been configured, the `DelayQueue` is used via its
30/// [`Stream`] implementation. [`poll`] is called. If an entry has reached its
31/// deadline, it is returned. If not, `Async::NotReady` indicating that the
32/// current task will be notified once the deadline has been reached.
33///
34/// # `Stream` implementation
35///
36/// Items are retrieved from the queue via [`Stream::poll`]. If no delays have
37/// expired, no items are returned. In this case, `NotReady` is returned and the
38/// current task is registered to be notified once the next item's delay has
39/// expired.
40///
41/// If no items are in the queue, i.e. `is_empty()` returns `true`, then `poll`
42/// returns `Ready(None)`. This indicates that the stream has reached an end.
43/// However, if a new item is inserted *after*, `poll` will once again start
44/// returning items or `NotReady.
45///
46/// Items are returned ordered by their expirations. Items that are configured
47/// to expire first will be returned first. There are no ordering guarantees
48/// for items configured to expire the same instant. Also note that delays are
49/// rounded to the closest millisecond.
50///
51/// # Implementation
52///
53/// The `DelayQueue` is backed by the same hashed timing wheel implementation as
54/// [`Timer`] as such, it offers the same performance benefits. See [`Timer`]
55/// for further implementation notes.
56///
57/// State associated with each entry is stored in a [`slab`]. This allows
58/// amortizing the cost of allocation. Space created for expired entries is
59/// reused when inserting new entries.
60///
61/// Capacity can be checked using [`capacity`] and allocated preemptively by using
62/// the [`reserve`] method.
63///
64/// # Usage
65///
66/// Using `DelayQueue` to manage cache entries.
67///
68/// ```rust,no_run
69/// use tokio::time::{delay_queue, DelayQueue, Error};
70///
71/// use futures::ready;
72/// use std::collections::HashMap;
73/// use std::task::{Context, Poll};
74/// use std::time::Duration;
75/// # type CacheKey = String;
76/// # type Value = String;
77///
78/// struct Cache {
79///     entries: HashMap<CacheKey, (Value, delay_queue::Key)>,
80///     expirations: DelayQueue<CacheKey>,
81/// }
82///
83/// const TTL_SECS: u64 = 30;
84///
85/// impl Cache {
86///     fn insert(&mut self, key: CacheKey, value: Value) {
87///         let delay = self.expirations
88///             .insert(key.clone(), Duration::from_secs(TTL_SECS));
89///
90///         self.entries.insert(key, (value, delay));
91///     }
92///
93///     fn get(&self, key: &CacheKey) -> Option<&Value> {
94///         self.entries.get(key)
95///             .map(|&(ref v, _)| v)
96///     }
97///
98///     fn remove(&mut self, key: &CacheKey) {
99///         if let Some((_, cache_key)) = self.entries.remove(key) {
100///             self.expirations.remove(&cache_key);
101///         }
102///     }
103///
104///     fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
105///         while let Some(res) = ready!(self.expirations.poll_expired(cx)) {
106///             let entry = res?;
107///             self.entries.remove(entry.get_ref());
108///         }
109///
110///         Poll::Ready(Ok(()))
111///     }
112/// }
113/// ```
114///
115/// [`insert`]: #method.insert
116/// [`insert_at`]: #method.insert_at
117/// [`Key`]: struct.Key.html
118/// [`Stream`]: https://docs.rs/futures/0.1/futures/stream/trait.Stream.html
119/// [`poll`]: #method.poll
120/// [`Stream::poll`]: #method.poll
121/// [`Timer`]: ../struct.Timer.html
122/// [`slab`]: https://docs.rs/slab
123/// [`capacity`]: #method.capacity
124/// [`reserve`]: #method.reserve
125#[derive(Debug)]
126pub struct DelayQueue<T> {
127    /// Stores data associated with entries
128    slab: Slab<Data<T>>,
129
130    /// Lookup structure tracking all delays in the queue
131    wheel: Wheel<Stack<T>>,
132
133    /// Delays that were inserted when already expired. These cannot be stored
134    /// in the wheel
135    expired: Stack<T>,
136
137    /// Delay expiring when the *first* item in the queue expires
138    delay: Option<Delay>,
139
140    /// Wheel polling state
141    poll: wheel::Poll,
142
143    /// Instant at which the timer starts
144    start: Instant,
145}
146
147/// An entry in `DelayQueue` that has expired and removed.
148///
149/// Values are returned by [`DelayQueue::poll`].
150///
151/// [`DelayQueue::poll`]: struct.DelayQueue.html#method.poll
152#[derive(Debug)]
153pub struct Expired<T> {
154    /// The data stored in the queue
155    data: T,
156
157    /// The expiration time
158    deadline: Instant,
159
160    /// The key associated with the entry
161    key: Key,
162}
163
164/// Token to a value stored in a `DelayQueue`.
165///
166/// Instances of `Key` are returned by [`DelayQueue::insert`]. See [`DelayQueue`]
167/// documentation for more details.
168///
169/// [`DelayQueue`]: struct.DelayQueue.html
170/// [`DelayQueue::insert`]: struct.DelayQueue.html#method.insert
171#[derive(Debug, Clone)]
172pub struct Key {
173    index: usize,
174}
175
176#[derive(Debug)]
177struct Stack<T> {
178    /// Head of the stack
179    head: Option<usize>,
180    _p: PhantomData<fn() -> T>,
181}
182
183#[derive(Debug)]
184struct Data<T> {
185    /// The data being stored in the queue and will be returned at the requested
186    /// instant.
187    inner: T,
188
189    /// The instant at which the item is returned.
190    when: u64,
191
192    /// Set to true when stored in the `expired` queue
193    expired: bool,
194
195    /// Next entry in the stack
196    next: Option<usize>,
197
198    /// Previous entry in the stack
199    prev: Option<usize>,
200}
201
202/// Maximum number of entries the queue can handle
203const MAX_ENTRIES: usize = (1 << 30) - 1;
204
205impl<T> DelayQueue<T> {
206    /// Create a new, empty, `DelayQueue`
207    ///
208    /// The queue will not allocate storage until items are inserted into it.
209    ///
210    /// # Examples
211    ///
212    /// ```rust
213    /// # use tokio::time::DelayQueue;
214    /// let delay_queue: DelayQueue<u32> = DelayQueue::new();
215    /// ```
216    pub fn new() -> DelayQueue<T> {
217        DelayQueue::with_capacity(0)
218    }
219
220    /// Create a new, empty, `DelayQueue` with the specified capacity.
221    ///
222    /// The queue will be able to hold at least `capacity` elements without
223    /// reallocating. If `capacity` is 0, the queue will not allocate for
224    /// storage.
225    ///
226    /// # Examples
227    ///
228    /// ```rust
229    /// # use tokio::time::DelayQueue;
230    /// # use std::time::Duration;
231    ///
232    /// # #[tokio::main]
233    /// # async fn main() {
234    ///     let mut delay_queue = DelayQueue::with_capacity(10);
235    ///
236    ///     // These insertions are done without further allocation
237    ///     for i in 0..10 {
238    ///         delay_queue.insert(i, Duration::from_secs(i));
239    ///     }
240    ///
241    ///     // This will make the queue allocate additional storage
242    ///     delay_queue.insert(11, Duration::from_secs(11));
243    /// # }
244    /// ```
245    pub fn with_capacity(capacity: usize) -> DelayQueue<T> {
246        DelayQueue {
247            wheel: Wheel::new(),
248            slab: Slab::with_capacity(capacity),
249            expired: Stack::default(),
250            delay: None,
251            poll: wheel::Poll::new(0),
252            start: Instant::now(),
253        }
254    }
255
256    /// Insert `value` into the queue set to expire at a specific instant in
257    /// time.
258    ///
259    /// This function is identical to `insert`, but takes an `Instant` instead
260    /// of a `Duration`.
261    ///
262    /// `value` is stored in the queue until `when` is reached. At which point,
263    /// `value` will be returned from [`poll`]. If `when` has already been
264    /// reached, then `value` is immediately made available to poll.
265    ///
266    /// The return value represents the insertion and is used at an argument to
267    /// [`remove`] and [`reset`]. Note that [`Key`] is token and is reused once
268    /// `value` is removed from the queue either by calling [`poll`] after
269    /// `when` is reached or by calling [`remove`]. At this point, the caller
270    /// must take care to not use the returned [`Key`] again as it may reference
271    /// a different item in the queue.
272    ///
273    /// See [type] level documentation for more details.
274    ///
275    /// # Panics
276    ///
277    /// This function panics if `when` is too far in the future.
278    ///
279    /// # Examples
280    ///
281    /// Basic usage
282    ///
283    /// ```rust
284    /// use tokio::time::{DelayQueue, Duration, Instant};
285    ///
286    /// # #[tokio::main]
287    /// # async fn main() {
288    ///     let mut delay_queue = DelayQueue::new();
289    ///     let key = delay_queue.insert_at(
290    ///         "foo", Instant::now() + Duration::from_secs(5));
291    ///
292    ///     // Remove the entry
293    ///     let item = delay_queue.remove(&key);
294    ///     assert_eq!(*item.get_ref(), "foo");
295    /// # }
296    /// ```
297    ///
298    /// [`poll`]: #method.poll
299    /// [`remove`]: #method.remove
300    /// [`reset`]: #method.reset
301    /// [`Key`]: struct.Key.html
302    /// [type]: #
303    pub fn insert_at(&mut self, value: T, when: Instant) -> Key {
304        assert!(self.slab.len() < MAX_ENTRIES, "max entries exceeded");
305
306        // Normalize the deadline. Values cannot be set to expire in the past.
307        let when = self.normalize_deadline(when);
308
309        // Insert the value in the store
310        let key = self.slab.insert(Data {
311            inner: value,
312            when,
313            expired: false,
314            next: None,
315            prev: None,
316        });
317
318        self.insert_idx(when, key);
319
320        // Set a new delay if the current's deadline is later than the one of the new item
321        let should_set_delay = if let Some(ref delay) = self.delay {
322            let current_exp = self.normalize_deadline(delay.deadline());
323            current_exp > when
324        } else {
325            true
326        };
327
328        if should_set_delay {
329            self.delay = Some(delay_until(self.start + Duration::from_millis(when)));
330        }
331
332        Key::new(key)
333    }
334
335    /// Attempt to pull out the next value of the delay queue, registering the
336    /// current task for wakeup if the value is not yet available, and returning
337    /// None if the queue is exhausted.
338    pub fn poll_expired(
339        &mut self,
340        cx: &mut task::Context<'_>,
341    ) -> Poll<Option<Result<Expired<T>, Error>>> {
342        let item = ready!(self.poll_idx(cx));
343        Poll::Ready(item.map(|result| {
344            result.map(|idx| {
345                let data = self.slab.remove(idx);
346                debug_assert!(data.next.is_none());
347                debug_assert!(data.prev.is_none());
348
349                Expired {
350                    key: Key::new(idx),
351                    data: data.inner,
352                    deadline: self.start + Duration::from_millis(data.when),
353                }
354            })
355        }))
356    }
357
358    /// Insert `value` into the queue set to expire after the requested duration
359    /// elapses.
360    ///
361    /// This function is identical to `insert_at`, but takes a `Duration`
362    /// instead of an `Instant`.
363    ///
364    /// `value` is stored in the queue until `when` is reached. At which point,
365    /// `value` will be returned from [`poll`]. If `when` has already been
366    /// reached, then `value` is immediately made available to poll.
367    ///
368    /// The return value represents the insertion and is used at an argument to
369    /// [`remove`] and [`reset`]. Note that [`Key`] is token and is reused once
370    /// `value` is removed from the queue either by calling [`poll`] after
371    /// `when` is reached or by calling [`remove`]. At this point, the caller
372    /// must take care to not use the returned [`Key`] again as it may reference
373    /// a different item in the queue.
374    ///
375    /// See [type] level documentation for more details.
376    ///
377    /// # Panics
378    ///
379    /// This function panics if `timeout` is greater than the maximum supported
380    /// duration.
381    ///
382    /// # Examples
383    ///
384    /// Basic usage
385    ///
386    /// ```rust
387    /// use tokio::time::DelayQueue;
388    /// use std::time::Duration;
389    ///
390    /// # #[tokio::main]
391    /// # async fn main() {
392    ///     let mut delay_queue = DelayQueue::new();
393    ///     let key = delay_queue.insert("foo", Duration::from_secs(5));
394    ///
395    ///     // Remove the entry
396    ///     let item = delay_queue.remove(&key);
397    ///     assert_eq!(*item.get_ref(), "foo");
398    /// # }
399    /// ```
400    ///
401    /// [`poll`]: #method.poll
402    /// [`remove`]: #method.remove
403    /// [`reset`]: #method.reset
404    /// [`Key`]: struct.Key.html
405    /// [type]: #
406    pub fn insert(&mut self, value: T, timeout: Duration) -> Key {
407        self.insert_at(value, Instant::now() + timeout)
408    }
409
410    fn insert_idx(&mut self, when: u64, key: usize) {
411        use self::wheel::{InsertError, Stack};
412
413        // Register the deadline with the timer wheel
414        match self.wheel.insert(when, key, &mut self.slab) {
415            Ok(_) => {}
416            Err((_, InsertError::Elapsed)) => {
417                self.slab[key].expired = true;
418                // The delay is already expired, store it in the expired queue
419                self.expired.push(key, &mut self.slab);
420            }
421            Err((_, err)) => panic!("invalid deadline; err={:?}", err),
422        }
423    }
424
425    /// Remove the item associated with `key` from the queue.
426    ///
427    /// There must be an item associated with `key`. The function returns the
428    /// removed item as well as the `Instant` at which it will the delay will
429    /// have expired.
430    ///
431    /// # Panics
432    ///
433    /// The function panics if `key` is not contained by the queue.
434    ///
435    /// # Examples
436    ///
437    /// Basic usage
438    ///
439    /// ```rust
440    /// use tokio::time::DelayQueue;
441    /// use std::time::Duration;
442    ///
443    /// # #[tokio::main]
444    /// # async fn main() {
445    ///     let mut delay_queue = DelayQueue::new();
446    ///     let key = delay_queue.insert("foo", Duration::from_secs(5));
447    ///
448    ///     // Remove the entry
449    ///     let item = delay_queue.remove(&key);
450    ///     assert_eq!(*item.get_ref(), "foo");
451    /// # }
452    /// ```
453    pub fn remove(&mut self, key: &Key) -> Expired<T> {
454        use crate::time::wheel::Stack;
455
456        // Special case the `expired` queue
457        if self.slab[key.index].expired {
458            self.expired.remove(&key.index, &mut self.slab);
459        } else {
460            self.wheel.remove(&key.index, &mut self.slab);
461        }
462
463        let data = self.slab.remove(key.index);
464
465        Expired {
466            key: Key::new(key.index),
467            data: data.inner,
468            deadline: self.start + Duration::from_millis(data.when),
469        }
470    }
471
472    /// Sets the delay of the item associated with `key` to expire at `when`.
473    ///
474    /// This function is identical to `reset` but takes an `Instant` instead of
475    /// a `Duration`.
476    ///
477    /// The item remains in the queue but the delay is set to expire at `when`.
478    /// If `when` is in the past, then the item is immediately made available to
479    /// the caller.
480    ///
481    /// # Panics
482    ///
483    /// This function panics if `when` is too far in the future or if `key` is
484    /// not contained by the queue.
485    ///
486    /// # Examples
487    ///
488    /// Basic usage
489    ///
490    /// ```rust
491    /// use tokio::time::{DelayQueue, Duration, Instant};
492    ///
493    /// # #[tokio::main]
494    /// # async fn main() {
495    ///     let mut delay_queue = DelayQueue::new();
496    ///     let key = delay_queue.insert("foo", Duration::from_secs(5));
497    ///
498    ///     // "foo" is scheduled to be returned in 5 seconds
499    ///
500    ///     delay_queue.reset_at(&key, Instant::now() + Duration::from_secs(10));
501    ///
502    ///     // "foo"is now scheduled to be returned in 10 seconds
503    /// # }
504    /// ```
505    pub fn reset_at(&mut self, key: &Key, when: Instant) {
506        self.wheel.remove(&key.index, &mut self.slab);
507
508        // Normalize the deadline. Values cannot be set to expire in the past.
509        let when = self.normalize_deadline(when);
510
511        self.slab[key.index].when = when;
512        self.insert_idx(when, key.index);
513
514        let next_deadline = self.next_deadline();
515        if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) {
516            delay.reset(deadline);
517        }
518    }
519
520    /// Returns the next time poll as determined by the wheel
521    fn next_deadline(&mut self) -> Option<Instant> {
522        self.wheel
523            .poll_at()
524            .map(|poll_at| self.start + Duration::from_millis(poll_at))
525    }
526
527    /// Sets the delay of the item associated with `key` to expire after
528    /// `timeout`.
529    ///
530    /// This function is identical to `reset_at` but takes a `Duration` instead
531    /// of an `Instant`.
532    ///
533    /// The item remains in the queue but the delay is set to expire after
534    /// `timeout`. If `timeout` is zero, then the item is immediately made
535    /// available to the caller.
536    ///
537    /// # Panics
538    ///
539    /// This function panics if `timeout` is greater than the maximum supported
540    /// duration or if `key` is not contained by the queue.
541    ///
542    /// # Examples
543    ///
544    /// Basic usage
545    ///
546    /// ```rust
547    /// use tokio::time::DelayQueue;
548    /// use std::time::Duration;
549    ///
550    /// # #[tokio::main]
551    /// # async fn main() {
552    ///     let mut delay_queue = DelayQueue::new();
553    ///     let key = delay_queue.insert("foo", Duration::from_secs(5));
554    ///
555    ///     // "foo" is scheduled to be returned in 5 seconds
556    ///
557    ///     delay_queue.reset(&key, Duration::from_secs(10));
558    ///
559    ///     // "foo"is now scheduled to be returned in 10 seconds
560    /// # }
561    /// ```
562    pub fn reset(&mut self, key: &Key, timeout: Duration) {
563        self.reset_at(key, Instant::now() + timeout);
564    }
565
566    /// Clears the queue, removing all items.
567    ///
568    /// After calling `clear`, [`poll`] will return `Ok(Ready(None))`.
569    ///
570    /// Note that this method has no effect on the allocated capacity.
571    ///
572    /// [`poll`]: #method.poll
573    ///
574    /// # Examples
575    ///
576    /// ```rust
577    /// use tokio::time::DelayQueue;
578    /// use std::time::Duration;
579    ///
580    /// # #[tokio::main]
581    /// # async fn main() {
582    ///     let mut delay_queue = DelayQueue::new();
583    ///
584    ///     delay_queue.insert("foo", Duration::from_secs(5));
585    ///
586    ///     assert!(!delay_queue.is_empty());
587    ///
588    ///     delay_queue.clear();
589    ///
590    ///     assert!(delay_queue.is_empty());
591    /// # }
592    /// ```
593    pub fn clear(&mut self) {
594        self.slab.clear();
595        self.expired = Stack::default();
596        self.wheel = Wheel::new();
597        self.delay = None;
598    }
599
600    /// Returns the number of elements the queue can hold without reallocating.
601    ///
602    /// # Examples
603    ///
604    /// ```rust
605    /// use tokio::time::DelayQueue;
606    ///
607    /// let delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
608    /// assert_eq!(delay_queue.capacity(), 10);
609    /// ```
610    pub fn capacity(&self) -> usize {
611        self.slab.capacity()
612    }
613
614    /// Returns the number of elements currently in the queue.
615    ///
616    /// # Examples
617    ///
618    /// ```rust
619    /// use tokio::time::DelayQueue;
620    /// use std::time::Duration;
621    ///
622    /// # #[tokio::main]
623    /// # async fn main() {
624    ///     let mut delay_queue: DelayQueue<i32> = DelayQueue::with_capacity(10);
625    ///     assert_eq!(delay_queue.len(), 0);
626    ///     delay_queue.insert(3, Duration::from_secs(5));
627    ///     assert_eq!(delay_queue.len(), 1);
628    /// # }
629    /// ```
630    pub fn len(&self) -> usize {
631        self.slab.len()
632    }
633
634    /// Reserve capacity for at least `additional` more items to be queued
635    /// without allocating.
636    ///
637    /// `reserve` does nothing if the queue already has sufficient capacity for
638    /// `additional` more values. If more capacity is required, a new segment of
639    /// memory will be allocated and all existing values will be copied into it.
640    /// As such, if the queue is already very large, a call to `reserve` can end
641    /// up being expensive.
642    ///
643    /// The queue may reserve more than `additional` extra space in order to
644    /// avoid frequent reallocations.
645    ///
646    /// # Panics
647    ///
648    /// Panics if the new capacity exceeds the maximum number of entries the
649    /// queue can contain.
650    ///
651    /// # Examples
652    ///
653    /// ```
654    /// use tokio::time::DelayQueue;
655    /// use std::time::Duration;
656    ///
657    /// # #[tokio::main]
658    /// # async fn main() {
659    ///     let mut delay_queue = DelayQueue::new();
660    ///
661    ///     delay_queue.insert("hello", Duration::from_secs(10));
662    ///     delay_queue.reserve(10);
663    ///
664    ///     assert!(delay_queue.capacity() >= 11);
665    /// # }
666    /// ```
667    pub fn reserve(&mut self, additional: usize) {
668        self.slab.reserve(additional);
669    }
670
671    /// Returns `true` if there are no items in the queue.
672    ///
673    /// Note that this function returns `false` even if all items have not yet
674    /// expired and a call to `poll` will return `NotReady`.
675    ///
676    /// # Examples
677    ///
678    /// ```
679    /// use tokio::time::DelayQueue;
680    /// use std::time::Duration;
681    ///
682    /// # #[tokio::main]
683    /// # async fn main() {
684    ///     let mut delay_queue = DelayQueue::new();
685    ///     assert!(delay_queue.is_empty());
686    ///
687    ///     delay_queue.insert("hello", Duration::from_secs(5));
688    ///     assert!(!delay_queue.is_empty());
689    /// # }
690    /// ```
691    pub fn is_empty(&self) -> bool {
692        self.slab.is_empty()
693    }
694
695    /// Polls the queue, returning the index of the next slot in the slab that
696    /// should be returned.
697    ///
698    /// A slot should be returned when the associated deadline has been reached.
699    fn poll_idx(&mut self, cx: &mut task::Context<'_>) -> Poll<Option<Result<usize, Error>>> {
700        use self::wheel::Stack;
701
702        let expired = self.expired.pop(&mut self.slab);
703
704        if expired.is_some() {
705            return Poll::Ready(expired.map(Ok));
706        }
707
708        loop {
709            if let Some(ref mut delay) = self.delay {
710                if !delay.is_elapsed() {
711                    ready!(Pin::new(&mut *delay).poll(cx));
712                }
713
714                let now = crate::time::ms(delay.deadline() - self.start, crate::time::Round::Down);
715
716                self.poll = wheel::Poll::new(now);
717            }
718
719            self.delay = None;
720
721            if let Some(idx) = self.wheel.poll(&mut self.poll, &mut self.slab) {
722                return Poll::Ready(Some(Ok(idx)));
723            }
724
725            if let Some(deadline) = self.next_deadline() {
726                self.delay = Some(delay_until(deadline));
727            } else {
728                return Poll::Ready(None);
729            }
730        }
731    }
732
733    fn normalize_deadline(&self, when: Instant) -> u64 {
734        let when = if when < self.start {
735            0
736        } else {
737            crate::time::ms(when - self.start, crate::time::Round::Up)
738        };
739
740        cmp::max(when, self.wheel.elapsed())
741    }
742}
743
744// We never put `T` in a `Pin`...
745impl<T> Unpin for DelayQueue<T> {}
746
747impl<T> Default for DelayQueue<T> {
748    fn default() -> DelayQueue<T> {
749        DelayQueue::new()
750    }
751}
752
753#[cfg(feature = "stream")]
754impl<T> futures_core::Stream for DelayQueue<T> {
755    // DelayQueue seems much more specific, where a user may care that it
756    // has reached capacity, so return those errors instead of panicking.
757    type Item = Result<Expired<T>, Error>;
758
759    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
760        DelayQueue::poll_expired(self.get_mut(), cx)
761    }
762}
763
764impl<T> wheel::Stack for Stack<T> {
765    type Owned = usize;
766    type Borrowed = usize;
767    type Store = Slab<Data<T>>;
768
769    fn is_empty(&self) -> bool {
770        self.head.is_none()
771    }
772
773    fn push(&mut self, item: Self::Owned, store: &mut Self::Store) {
774        // Ensure the entry is not already in a stack.
775        debug_assert!(store[item].next.is_none());
776        debug_assert!(store[item].prev.is_none());
777
778        // Remove the old head entry
779        let old = self.head.take();
780
781        if let Some(idx) = old {
782            store[idx].prev = Some(item);
783        }
784
785        store[item].next = old;
786        self.head = Some(item)
787    }
788
789    fn pop(&mut self, store: &mut Self::Store) -> Option<Self::Owned> {
790        if let Some(idx) = self.head {
791            self.head = store[idx].next;
792
793            if let Some(idx) = self.head {
794                store[idx].prev = None;
795            }
796
797            store[idx].next = None;
798            debug_assert!(store[idx].prev.is_none());
799
800            Some(idx)
801        } else {
802            None
803        }
804    }
805
806    fn remove(&mut self, item: &Self::Borrowed, store: &mut Self::Store) {
807        assert!(store.contains(*item));
808
809        // Ensure that the entry is in fact contained by the stack
810        debug_assert!({
811            // This walks the full linked list even if an entry is found.
812            let mut next = self.head;
813            let mut contains = false;
814
815            while let Some(idx) = next {
816                if idx == *item {
817                    debug_assert!(!contains);
818                    contains = true;
819                }
820
821                next = store[idx].next;
822            }
823
824            contains
825        });
826
827        if let Some(next) = store[*item].next {
828            store[next].prev = store[*item].prev;
829        }
830
831        if let Some(prev) = store[*item].prev {
832            store[prev].next = store[*item].next;
833        } else {
834            self.head = store[*item].next;
835        }
836
837        store[*item].next = None;
838        store[*item].prev = None;
839    }
840
841    fn when(item: &Self::Borrowed, store: &Self::Store) -> u64 {
842        store[*item].when
843    }
844}
845
846impl<T> Default for Stack<T> {
847    fn default() -> Stack<T> {
848        Stack {
849            head: None,
850            _p: PhantomData,
851        }
852    }
853}
854
855impl Key {
856    pub(crate) fn new(index: usize) -> Key {
857        Key { index }
858    }
859}
860
861impl<T> Expired<T> {
862    /// Returns a reference to the inner value.
863    pub fn get_ref(&self) -> &T {
864        &self.data
865    }
866
867    /// Returns a mutable reference to the inner value.
868    pub fn get_mut(&mut self) -> &mut T {
869        &mut self.data
870    }
871
872    /// Consumes `self` and returns the inner value.
873    pub fn into_inner(self) -> T {
874        self.data
875    }
876}