mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-27 00:00:12 +02:00
time: fix DelayQueue rewriting delay on insert after Poll::Ready (#2285)
When the queue was polled and yielded an index from the wheel, the delay until the next item was never updated. As a result, when one item was yielded from `poll_idx` the following insert erronously updated the delay to the instant of the inserted item. Fixes: #1700
This commit is contained in:
@@ -721,15 +721,16 @@ impl<T> DelayQueue<T> {
|
|||||||
self.poll = wheel::Poll::new(now);
|
self.poll = wheel::Poll::new(now);
|
||||||
}
|
}
|
||||||
|
|
||||||
self.delay = None;
|
// 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);
|
||||||
|
|
||||||
if let Some(idx) = self.wheel.poll(&mut self.poll, &mut self.slab) {
|
self.delay = self.next_deadline().map(delay_until);
|
||||||
|
|
||||||
|
if let Some(idx) = wheel_idx {
|
||||||
return Poll::Ready(Some(Ok(idx)));
|
return Poll::Ready(Some(Ok(idx)));
|
||||||
}
|
}
|
||||||
|
|
||||||
if let Some(deadline) = self.next_deadline() {
|
if self.delay.is_none() {
|
||||||
self.delay = Some(delay_until(deadline));
|
|
||||||
} else {
|
|
||||||
return Poll::Ready(None);
|
return Poll::Ready(None);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -410,6 +410,39 @@ async fn insert_before_first_after_poll() {
|
|||||||
assert_eq!(entry, "two");
|
assert_eq!(entry, "two");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn insert_after_ready_poll() {
|
||||||
|
time::pause();
|
||||||
|
|
||||||
|
let mut queue = task::spawn(DelayQueue::new());
|
||||||
|
|
||||||
|
let now = Instant::now();
|
||||||
|
|
||||||
|
queue.insert_at("1", now + ms(100));
|
||||||
|
queue.insert_at("2", now + ms(100));
|
||||||
|
queue.insert_at("3", now + ms(100));
|
||||||
|
|
||||||
|
assert_pending!(poll!(queue));
|
||||||
|
|
||||||
|
delay_for(ms(100)).await;
|
||||||
|
|
||||||
|
assert!(queue.is_woken());
|
||||||
|
|
||||||
|
let mut res = vec![];
|
||||||
|
|
||||||
|
while res.len() < 3 {
|
||||||
|
let entry = assert_ready_ok!(poll!(queue));
|
||||||
|
res.push(entry.into_inner());
|
||||||
|
queue.insert_at("foo", now + ms(500));
|
||||||
|
}
|
||||||
|
|
||||||
|
res.sort();
|
||||||
|
|
||||||
|
assert_eq!("1", res[0]);
|
||||||
|
assert_eq!("2", res[1]);
|
||||||
|
assert_eq!("3", res[2]);
|
||||||
|
}
|
||||||
|
|
||||||
fn ms(n: u64) -> Duration {
|
fn ms(n: u64) -> Duration {
|
||||||
Duration::from_millis(n)
|
Duration::from_millis(n)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user