From cf38ba627abb0d44ae7e7dc0659ddd51866e9efa Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Tue, 8 Feb 2022 18:07:33 -0800 Subject: [PATCH] util: remove error case from the infallible DelayQueue::poll_elapsed (#4241) --- tokio-util/src/time/delay_queue.rs | 43 +++++++++------------ tokio-util/tests/time_delay_queue.rs | 56 ++++++++++++++-------------- 2 files changed, 46 insertions(+), 53 deletions(-) diff --git a/tokio-util/src/time/delay_queue.rs b/tokio-util/src/time/delay_queue.rs index 697670d75..a0c5e5c5b 100644 --- a/tokio-util/src/time/delay_queue.rs +++ b/tokio-util/src/time/delay_queue.rs @@ -7,7 +7,7 @@ use crate::time::wheel::{self, Wheel}; use futures_core::ready; -use tokio::time::{error::Error, sleep_until, Duration, Instant, Sleep}; +use tokio::time::{sleep_until, Duration, Instant, Sleep}; use core::ops::{Index, IndexMut}; use slab::Slab; @@ -72,7 +72,6 @@ use std::task::{self, Poll, Waker}; /// Using `DelayQueue` to manage cache entries. /// /// ```rust,no_run -/// use tokio::time::error::Error; /// use tokio_util::time::{DelayQueue, delay_queue}; /// /// use futures::ready; @@ -108,13 +107,12 @@ use std::task::{self, Poll, Waker}; /// } /// } /// -/// fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll> { -/// while let Some(res) = ready!(self.expirations.poll_expired(cx)) { -/// let entry = res?; +/// fn poll_purge(&mut self, cx: &mut Context<'_>) -> Poll<()> { +/// while let Some(entry) = ready!(self.expirations.poll_expired(cx)) { /// self.entries.remove(entry.get_ref()); /// } /// -/// Poll::Ready(Ok(())) +/// Poll::Ready(()) /// } /// } /// ``` @@ -577,10 +575,7 @@ impl DelayQueue { /// 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, Error>>> { + pub fn poll_expired(&mut self, cx: &mut task::Context<'_>) -> Poll>> { if !self .waker .as_ref() @@ -591,18 +586,16 @@ impl DelayQueue { } let item = ready!(self.poll_idx(cx)); - Poll::Ready(item.map(|result| { - result.map(|key| { - let data = self.slab.remove(&key); - debug_assert!(data.next.is_none()); - debug_assert!(data.prev.is_none()); + Poll::Ready(item.map(|key| { + let data = self.slab.remove(&key); + debug_assert!(data.next.is_none()); + debug_assert!(data.prev.is_none()); - Expired { - key, - data: data.inner, - deadline: self.start + Duration::from_millis(data.when), - } - }) + Expired { + key, + data: data.inner, + deadline: self.start + Duration::from_millis(data.when), + } })) } @@ -1017,13 +1010,13 @@ impl DelayQueue { /// 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>> { + fn poll_idx(&mut self, cx: &mut task::Context<'_>) -> Poll> { use self::wheel::Stack; let expired = self.expired.pop(&mut self.slab); if expired.is_some() { - return Poll::Ready(expired.map(Ok)); + return Poll::Ready(expired); } loop { @@ -1043,7 +1036,7 @@ impl DelayQueue { self.delay = self.next_deadline().map(|when| Box::pin(sleep_until(when))); if let Some(idx) = wheel_idx { - return Poll::Ready(Some(Ok(idx))); + return Poll::Ready(Some(idx)); } if self.delay.is_none() { @@ -1075,7 +1068,7 @@ impl Default for DelayQueue { impl futures_core::Stream for DelayQueue { // 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, Error>; + type Item = Expired; fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll> { DelayQueue::poll_expired(self.get_mut(), cx) diff --git a/tokio-util/tests/time_delay_queue.rs b/tokio-util/tests/time_delay_queue.rs index 1c30446af..cb163adf3 100644 --- a/tokio-util/tests/time_delay_queue.rs +++ b/tokio-util/tests/time_delay_queue.rs @@ -3,7 +3,7 @@ #![cfg(feature = "full")] use tokio::time::{self, sleep, sleep_until, Duration, Instant}; -use tokio_test::{assert_ok, assert_pending, assert_ready, task}; +use tokio_test::{assert_pending, assert_ready, task}; use tokio_util::time::DelayQueue; macro_rules! poll { @@ -12,12 +12,12 @@ macro_rules! poll { }; } -macro_rules! assert_ready_ok { +macro_rules! assert_ready_some { ($e:expr) => {{ - assert_ok!(match assert_ready!($e) { + match assert_ready!($e) { Some(v) => v, None => panic!("None"), - }) + } }}; } @@ -31,7 +31,7 @@ async fn single_immediate_delay() { // Advance time by 1ms to handle thee rounding sleep(ms(1)).await; - assert_ready_ok!(poll!(queue)); + assert_ready_some!(poll!(queue)); let entry = assert_ready!(poll!(queue)); assert!(entry.is_none()) @@ -52,7 +52,7 @@ async fn multi_immediate_delays() { let mut res = vec![]; while res.len() < 3 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); } @@ -83,7 +83,7 @@ async fn single_short_delay() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); assert_eq!(*entry.get_ref(), "foo"); let entry = assert_ready!(poll!(queue)); @@ -193,7 +193,7 @@ async fn reset_entry() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); assert_eq!(*entry.get_ref(), "foo"); let entry = assert_ready!(poll!(queue)); @@ -271,7 +271,7 @@ async fn repeatedly_reset_entry_inserted_as_expired() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "foo"); let entry = assert_ready!(poll!(queue)); @@ -311,7 +311,7 @@ async fn remove_at_timer_wheel_threshold() { sleep(ms(80)).await; - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); match entry { "foo" => { @@ -348,7 +348,7 @@ async fn expires_before_last_insert() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "bar"); } @@ -375,14 +375,14 @@ async fn multi_reset() { sleep(ms(50)).await; - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); assert_eq!(*entry.get_ref(), "two"); assert_pending!(poll!(queue)); sleep(ms(50)).await; - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); assert_eq!(*entry.get_ref(), "one"); let entry = assert_ready!(poll!(queue)); @@ -408,7 +408,7 @@ async fn expire_first_key_when_reset_to_expire_earlier() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "one"); } @@ -431,7 +431,7 @@ async fn expire_second_key_when_reset_to_expire_earlier() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "two"); } @@ -453,7 +453,7 @@ async fn reset_first_expiring_item_to_expire_later() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "two"); } @@ -479,7 +479,7 @@ async fn insert_before_first_after_poll() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "two"); } @@ -504,7 +504,7 @@ async fn insert_after_ready_poll() { let mut res = vec![]; while res.len() < 3 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); queue.insert_at("foo", now + ms(500)); } @@ -549,7 +549,7 @@ async fn reset_later_after_slot_starts() { sleep(ms(1)).await; assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "foo"); } @@ -568,7 +568,7 @@ async fn reset_inserted_expired() { sleep(ms(200)).await; - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "foo"); assert_eq!(queue.len(), 0); @@ -607,7 +607,7 @@ async fn reset_earlier_after_slot_starts() { sleep(ms(1)).await; assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "foo"); } @@ -630,7 +630,7 @@ async fn insert_in_past_after_poll_fires_immediately() { assert!(queue.is_woken()); - let entry = assert_ready_ok!(poll!(queue)).into_inner(); + let entry = assert_ready_some!(poll!(queue)).into_inner(); assert_eq!(entry, "bar"); } @@ -657,7 +657,7 @@ async fn compact_expire_empty() { let mut res = vec![]; while res.len() < 2 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); } @@ -704,7 +704,7 @@ async fn compact_remove_remapped_keys() { let mut res = vec![]; while res.len() < 2 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); } @@ -743,7 +743,7 @@ async fn compact_change_deadline() { let mut res = vec![]; while res.len() < 2 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); } @@ -763,14 +763,14 @@ async fn compact_change_deadline() { sleep(ms(10)).await; while res.len() < 4 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); } sleep(ms(10)).await; while res.len() < 6 { - let entry = assert_ready_ok!(poll!(queue)); + let entry = assert_ready_some!(poll!(queue)); res.push(entry.into_inner()); } @@ -803,7 +803,7 @@ async fn remove_after_compact_poll() { queue.insert_at("bar", now + ms(20)); sleep(ms(10)).await; - assert_eq!(assert_ready_ok!(poll!(queue)).key(), foo_key); + assert_eq!(assert_ready_some!(poll!(queue)).key(), foo_key); queue.compact();