diff --git a/tokio-util/src/time/delay_queue.rs b/tokio-util/src/time/delay_queue.rs index 3f0bc58c7..b327cc939 100644 --- a/tokio-util/src/time/delay_queue.rs +++ b/tokio-util/src/time/delay_queue.rs @@ -864,11 +864,19 @@ impl DelayQueue { self.slab[*key].expired = false; self.insert_idx(when, *key); + let inserted_expired = self.slab[*key].expired; let next_deadline = self.next_deadline(); - if let (Some(ref mut delay), Some(deadline)) = (&mut self.delay, next_deadline) { - // This should awaken us if necessary (ie, if already expired) - delay.as_mut().reset(deadline); + match (next_deadline, &mut self.delay) { + (None, _) => self.delay = None, + (Some(deadline), Some(delay)) => delay.as_mut().reset(deadline), + (Some(deadline), None) => self.delay = Some(Box::pin(sleep_until(deadline))), + } + + if inserted_expired { + if let Some(waker) = self.waker.take() { + waker.wake(); + } } } diff --git a/tokio-util/tests/time_delay_queue.rs b/tokio-util/tests/time_delay_queue.rs index fdd0844c8..74360b39e 100644 --- a/tokio-util/tests/time_delay_queue.rs +++ b/tokio-util/tests/time_delay_queue.rs @@ -202,6 +202,27 @@ async fn reset_entry() { assert!(entry.is_none()) } +#[tokio::test] +async fn reset_to_past_wakes_pending_queue() { + time::pause(); + + let mut queue = task::spawn(DelayQueue::new()); + let key = queue.insert("foo", ms(10_000)); + + assert_pending!(poll!(queue)); + assert!(!queue.is_woken()); + + queue.reset_at(&key, Instant::now() - ms(100)); + + assert!(queue.is_woken()); + + let entry = assert_ready_some!(poll!(queue)); + assert_eq!(*entry.get_ref(), "foo"); + + let entry = assert_ready!(poll!(queue)); + assert!(entry.is_none()); +} + // Reproduces tokio-rs/tokio#849. #[tokio::test] async fn reset_much_later() {