tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-25 00:00:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

time: advance frozen time in park_timeout (#2059)

Browse Source 
This patch improves the behavior of frozen time (a testing utility made
available with the `test-util` feature flag). Instead of of requiring
`time::advance` to be called in order to advance the value returned by
`Instant::now`, calls to `time::Driver::park_timeout` will use the
provided duration to advance the time.

This is the desired behavior as the timeout is used to indicate when the
next scheduled delay needs to be fired.
This commit is contained in:
Carl Lerche
2020-01-06 08:47:34 -08:00
committed by GitHub
parent 84ff73e687
commit f0006006ed
10 changed files with 854 additions and 699 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/runtime/context.rs
-12
View File
@@ -81,18 +81,6 @@ impl ThreadContext {
})
}
#[cfg(all(feature = "test-util", feature = "time", test))]
pub(crate) fn with_time_handle(mut self, handle: crate::runtime::time::Handle) -> Self {
self.time_handle = handle;
self
}
#[cfg(all(feature = "test-util", feature = "time", test))]
pub(crate) fn with_clock(mut self, clock: crate::runtime::time::Clock) -> Self {
self.clock.replace(clock);
self
}
#[cfg(all(feature = "io-driver", not(loom)))]
pub(crate) fn io_handle() -> crate::runtime::io::Handle {
CONTEXT.with(|ctx| match *ctx.borrow() {
tokio/src/time/clock.rs
+13 -1
View File
@@ -5,7 +5,7 @@
//! configurable.
cfg_not_test_util! {
use crate::time::Instant;
use crate::time::{Duration, Instant};
#[derive(Debug, Clone)]
pub(crate) struct Clock {}
@@ -22,6 +22,14 @@ cfg_not_test_util! {
pub(crate) fn now(&self) -> Instant {
now()
}
pub(crate) fn is_frozen(&self) -> bool {
false
}
pub(crate) fn advance(&self, _dur: Duration) {
unreachable!();
}
}
}
@@ -137,6 +145,10 @@ cfg_test_util! {
}
}
pub(crate) fn is_frozen(&self) -> bool {
self.inner.frozen.lock().unwrap().is_some()
}
pub(crate) fn advance(&self, duration: Duration) {
let mut frozen = self.inner.frozen.lock().unwrap();
tokio/src/time/driver/entry.rs
+2 -2
View File
@@ -104,8 +104,8 @@ const ERROR: u64 = u64::MAX;
// ===== impl Entry =====
impl Entry {
pub(crate) fn new(deadline: Instant, duration: Duration) -> Arc<Entry> {
let inner = Handle::current().inner().unwrap();
pub(crate) fn new(handle: &Handle, deadline: Instant, duration: Duration) -> Arc<Entry> {
let inner = handle.inner().unwrap();
let entry: Entry;
// Increment the number of active timeouts
tokio/src/time/driver/mod.rs
+17 -3
View File
@@ -4,7 +4,7 @@ mod atomic_stack;
use self::atomic_stack::AtomicStack;
mod entry;
use self::entry::Entry;
pub(super) use self::entry::Entry;
mod handle;
pub(crate) use self::handle::Handle;
@@ -245,7 +245,14 @@ where
let deadline = self.expiration_instant(when);
if deadline > now {
self.park.park_timeout(deadline - now)?;
let dur = deadline - now;
if self.clock.is_frozen() {
self.park.park_timeout(Duration::from_secs(0))?;
self.clock.advance(dur);
} else {
self.park.park_timeout(dur)?;
}
} else {
self.park.park_timeout(Duration::from_secs(0))?;
}
@@ -269,7 +276,14 @@ where
let deadline = self.expiration_instant(when);
if deadline > now {
self.park.park_timeout(cmp::min(deadline - now, duration))?;
let duration = cmp::min(deadline - now, duration);
if self.clock.is_frozen() {
self.park.park_timeout(Duration::from_secs(0))?;
self.clock.advance(duration);
} else {
self.park.park_timeout(duration)?;
}
} else {
self.park.park_timeout(Duration::from_secs(0))?;
}
tokio/src/time/driver/registration.rs
+4 -2
View File
@@ -1,4 +1,4 @@
use crate::time::driver::Entry;
use crate::time::driver::{Entry, Handle};
use crate::time::{Duration, Error, Instant};
use std::sync::Arc;
@@ -15,8 +15,10 @@ pub(crate) struct Registration {
impl Registration {
pub(crate) fn new(deadline: Instant, duration: Duration) -> Registration {
let handle = Handle::current();
Registration {
entry: Entry::new(deadline, duration),
entry: Entry::new(&handle, deadline, duration),
}
}
tokio/src/time/tests/mod.rs
-2
View File
@@ -1,6 +1,4 @@
mod mock_clock;
mod test_delay;
mod test_queue;
use crate::time::{self, Instant};
use std::time::Duration;
tokio/src/time/tests/test_delay.rs
+269 -308
View File
@@ -1,351 +1,364 @@
#![warn(rust_2018_idioms)]
use crate::time::tests::mock_clock::mock;
use crate::time::{delay_until, Duration, Instant};
use crate::park::{Park, Unpark};
use crate::time::driver::{Driver, Entry, Handle};
use crate::time::Clock;
use crate::time::{Duration, Instant};
use tokio_test::task;
use tokio_test::{assert_pending, assert_ready};
use tokio_test::{assert_ok, assert_pending, assert_ready_ok};
use std::sync::Arc;
macro_rules! poll {
($e:expr) => {
$e.enter(|cx, e| e.poll_elapsed(cx))
};
}
#[test]
fn immediate_delay() {
mock(|clock| {
// Create `Delay` that elapsed immediately.
let mut fut = task::spawn(delay_until(clock.now()));
let (mut driver, clock, handle) = setup();
// Ready!
assert_ready!(fut.poll());
let when = clock.now();
let mut e = task::spawn(delay_until(&handle, when));
// Turn the timer, it runs for the elapsed time
clock.turn_for(ms(1000));
assert_ready_ok!(poll!(e));
// The time has not advanced. The `turn` completed immediately.
assert_eq!(clock.advanced(), ms(1000));
});
assert_ok!(driver.park_timeout(Duration::from_millis(1000)));
// The time has not advanced. The `turn` completed immediately.
assert_eq!(clock.advanced(), ms(1000));
}
#[test]
fn delayed_delay_level_0() {
let (mut driver, clock, handle) = setup();
let start = clock.now();
for &i in &[1, 10, 60] {
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(i)));
// Create a `Delay` that elapses in the future
let mut e = task::spawn(delay_until(&handle, start + ms(i)));
// The delay has not elapsed.
assert_pending!(fut.poll());
// The delay has not elapsed.
assert_pending!(poll!(e));
clock.turn();
assert_eq!(clock.advanced(), ms(i));
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(i));
assert_ready!(fut.poll());
});
assert_ready_ok!(poll!(e));
}
}
#[test]
fn sub_ms_delayed_delay() {
mock(|clock| {
for _ in 0..5 {
let deadline = clock.now() + Duration::from_millis(1) + Duration::new(0, 1);
let (mut driver, clock, handle) = setup();
let mut fut = task::spawn(delay_until(deadline));
for _ in 0..5 {
let deadline = clock.now() + ms(1) + Duration::new(0, 1);
assert_pending!(fut.poll());
let mut e = task::spawn(delay_until(&handle, deadline));
clock.turn();
assert_ready!(fut.poll());
assert_pending!(poll!(e));
assert!(clock.now() >= deadline);
assert_ok!(driver.park());
assert_ready_ok!(poll!(e));
clock.advance(Duration::new(0, 1));
}
});
assert!(clock.now() >= deadline);
clock.advance(Duration::new(0, 1));
}
}
#[test]
fn delayed_delay_wrapping_level_0() {
mock(|clock| {
clock.turn_for(ms(5));
assert_eq!(clock.advanced(), ms(5));
let (mut driver, clock, handle) = setup();
let mut fut = task::spawn(delay_until(clock.now() + ms(60)));
assert_ok!(driver.park_timeout(ms(5)));
assert_eq!(clock.advanced(), ms(5));
assert_pending!(fut.poll());
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(60)));
clock.turn();
assert_eq!(clock.advanced(), ms(64));
assert_pending!(fut.poll());
assert_pending!(poll!(e));
clock.turn();
assert_eq!(clock.advanced(), ms(65));
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(64));
assert_pending!(poll!(e));
assert_ready!(fut.poll());
});
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(65));
assert_ready_ok!(poll!(e));
}
#[test]
fn timer_wrapping_with_higher_levels() {
mock(|clock| {
// Set delay to hit level 1
let mut s1 = task::spawn(delay_until(clock.now() + ms(64)));
assert_pending!(s1.poll());
let (mut driver, clock, handle) = setup();
// Turn a bit
clock.turn_for(ms(5));
// Set delay to hit level 1
let mut e1 = task::spawn(delay_until(&handle, clock.now() + ms(64)));
assert_pending!(poll!(e1));
// Set timeout such that it will hit level 0, but wrap
let mut s2 = task::spawn(delay_until(clock.now() + ms(60)));
assert_pending!(s2.poll());
// Turn a bit
assert_ok!(driver.park_timeout(ms(5)));
// This should result in s1 firing
clock.turn();
assert_eq!(clock.advanced(), ms(64));
// Set timeout such that it will hit level 0, but wrap
let mut e2 = task::spawn(delay_until(&handle, clock.now() + ms(60)));
assert_pending!(poll!(e2));
assert_ready!(s1.poll());
assert_pending!(s2.poll());
// This should result in s1 firing
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(64));
clock.turn();
assert_eq!(clock.advanced(), ms(65));
assert_ready_ok!(poll!(e1));
assert_pending!(poll!(e2));
assert_ready!(s2.poll());
});
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(65));
assert_ready_ok!(poll!(e1));
}
#[test]
fn delay_with_deadline_in_past() {
mock(|clock| {
// Create `Delay` that elapsed immediately.
let mut fut = task::spawn(delay_until(clock.now() - ms(100)));
let (mut driver, clock, handle) = setup();
// Even though the delay expires in the past, it is not ready yet
// because the timer must observe it.
assert_ready!(fut.poll());
// Create `Delay` that elapsed immediately.
let mut e = task::spawn(delay_until(&handle, clock.now() - ms(100)));
// Turn the timer, it runs for the elapsed time
clock.turn_for(ms(1000));
// Even though the delay expires in the past, it is not ready yet
// because the timer must observe it.
assert_ready_ok!(poll!(e));
// The time has not advanced. The `turn` completed immediately.
assert_eq!(clock.advanced(), ms(1000));
});
// Turn the timer, it runs for the elapsed time
assert_ok!(driver.park_timeout(ms(1000)));
// The time has not advanced. The `turn` completed immediately.
assert_eq!(clock.advanced(), ms(1000));
}
#[test]
fn delayed_delay_level_1() {
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(234)));
let (mut driver, clock, handle) = setup();
// The delay has not elapsed.
assert_pending!(fut.poll());
// Create a `Delay` that elapses in the future
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(234)));
// Turn the timer, this will wake up to cascade the timer down.
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(192));
// The delay has not elapsed.
assert_pending!(poll!(e));
// The delay has not elapsed.
assert_pending!(fut.poll());
// Turn the timer, this will wake up to cascade the timer down.
assert_ok!(driver.park_timeout(ms(1000)));
assert_eq!(clock.advanced(), ms(192));
// Turn the timer again
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(234));
// The delay has not elapsed.
assert_pending!(poll!(e));
// The delay has elapsed.
assert_ready!(fut.poll());
});
// Turn the timer again
assert_ok!(driver.park_timeout(ms(1000)));
assert_eq!(clock.advanced(), ms(234));
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(234)));
// The delay has elapsed.
assert_ready_ok!(poll!(e));
// The delay has not elapsed.
assert_pending!(fut.poll());
let (mut driver, clock, handle) = setup();
// Turn the timer with a smaller timeout than the cascade.
clock.turn_for(ms(100));
assert_eq!(clock.advanced(), ms(100));
// Create a `Delay` that elapses in the future
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(234)));
assert_pending!(fut.poll());
// The delay has not elapsed.
assert_pending!(poll!(e));
// Turn the timer, this will wake up to cascade the timer down.
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(192));
// Turn the timer with a smaller timeout than the cascade.
assert_ok!(driver.park_timeout(ms(100)));
assert_eq!(clock.advanced(), ms(100));
// The delay has not elapsed.
assert_pending!(fut.poll());
assert_pending!(poll!(e));
// Turn the timer again
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(234));
// Turn the timer, this will wake up to cascade the timer down.
assert_ok!(driver.park_timeout(ms(1000)));
assert_eq!(clock.advanced(), ms(192));
// The delay has elapsed.
assert_ready!(fut.poll());
});
}
// The delay has not elapsed.
assert_pending!(poll!(e));
#[test]
#[should_panic]
fn creating_delay_outside_of_context() {
let now = Instant::now();
// Turn the timer again
assert_ok!(driver.park_timeout(ms(1000)));
assert_eq!(clock.advanced(), ms(234));
// This creates a delay outside of the context of a mock timer. This tests
// that it will panic.
let _fut = task::spawn(delay_until(now + ms(500)));
// The delay has elapsed.
assert_ready_ok!(poll!(e));
}
#[test]
fn concurrently_set_two_timers_second_one_shorter() {
mock(|clock| {
let mut fut1 = task::spawn(delay_until(clock.now() + ms(500)));
let mut fut2 = task::spawn(delay_until(clock.now() + ms(200)));
let (mut driver, clock, handle) = setup();
// The delay has not elapsed
assert_pending!(fut1.poll());
assert_pending!(fut2.poll());
let mut e1 = task::spawn(delay_until(&handle, clock.now() + ms(500)));
let mut e2 = task::spawn(delay_until(&handle, clock.now() + ms(200)));
// Delay until a cascade
clock.turn();
assert_eq!(clock.advanced(), ms(192));
// The delay has not elapsed
assert_pending!(poll!(e1));
assert_pending!(poll!(e2));
// Delay until the second timer.
clock.turn();
assert_eq!(clock.advanced(), ms(200));
// Delay until a cascade
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(192));
// The shorter delay fires
assert_ready!(fut2.poll());
assert_pending!(fut1.poll());
// Delay until the second timer.
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(200));
clock.turn();
assert_eq!(clock.advanced(), ms(448));
// The shorter delay fires
assert_ready_ok!(poll!(e2));
assert_pending!(poll!(e1));
assert_pending!(fut1.poll());
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(448));
// Turn again, this time the time will advance to the second delay
clock.turn();
assert_eq!(clock.advanced(), ms(500));
assert_pending!(poll!(e1));
assert_ready!(fut1.poll());
})
// Turn again, this time the time will advance to the second delay
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(500));
assert_ready_ok!(poll!(e1));
}
#[test]
fn short_delay() {
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(1)));
let (mut driver, clock, handle) = setup();
// The delay has not elapsed.
assert_pending!(fut.poll());
// Create a `Delay` that elapses in the future
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(1)));
// Turn the timer, but not enough time will go by.
clock.turn();
// The delay has not elapsed.
assert_pending!(poll!(e));
// The delay has elapsed.
assert_ready!(fut.poll());
// Turn the timer, but not enough time will go by.
assert_ok!(driver.park());
// The time has advanced to the point of the delay elapsing.
assert_eq!(clock.advanced(), ms(1));
})
// The delay has elapsed.
assert_ready_ok!(poll!(e));
// The time has advanced to the point of the delay elapsing.
assert_eq!(clock.advanced(), ms(1));
}
#[test]
fn sorta_long_delay_until() {
const MIN_5: u64 = 5 * 60 * 1000;
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(MIN_5)));
let (mut driver, clock, handle) = setup();
// The delay has not elapsed.
assert_pending!(fut.poll());
// Create a `Delay` that elapses in the future
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(MIN_5)));
let cascades = &[262_144, 262_144 + 9 * 4096, 262_144 + 9 * 4096 + 15 * 64];
// The delay has not elapsed.
assert_pending!(poll!(e));
for &elapsed in cascades {
clock.turn();
assert_eq!(clock.advanced(), ms(elapsed));
let cascades = &[262_144, 262_144 + 9 * 4096, 262_144 + 9 * 4096 + 15 * 64];
assert_pending!(fut.poll());
}
for &elapsed in cascades {
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(elapsed));
clock.turn();
assert_eq!(clock.advanced(), ms(MIN_5));
assert_pending!(poll!(e));
}
// The delay has elapsed.
assert_ready!(fut.poll());
})
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(MIN_5));
// The delay has elapsed.
assert_ready_ok!(poll!(e));
}
#[test]
fn very_long_delay() {
const MO_5: u64 = 5 * 30 * 24 * 60 * 60 * 1000;
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(MO_5)));
let (mut driver, clock, handle) = setup();
// The delay has not elapsed.
assert_pending!(fut.poll());
// Create a `Delay` that elapses in the future
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(MO_5)));
let cascades = &[
12_884_901_888,
12_952_010_752,
12_959_875_072,
12_959_997_952,
];
// The delay has not elapsed.
assert_pending!(poll!(e));
for &elapsed in cascades {
clock.turn();
assert_eq!(clock.advanced(), ms(elapsed));
let cascades = &[
12_884_901_888,
12_952_010_752,
12_959_875_072,
12_959_997_952,
];
assert_pending!(fut.poll());
}
for &elapsed in cascades {
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(elapsed));
// Turn the timer, but not enough time will go by.
clock.turn();
assert_pending!(poll!(e));
}
// The time has advanced to the point of the delay elapsing.
assert_eq!(clock.advanced(), ms(MO_5));
// Turn the timer, but not enough time will go by.
assert_ok!(driver.park());
// The delay has elapsed.
assert_ready!(fut.poll());
})
}
// The time has advanced to the point of the delay elapsing.
assert_eq!(clock.advanced(), ms(MO_5));
#[test]
#[should_panic]
fn greater_than_max() {
const YR_5: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
mock(|clock| {
// Create a `Delay` that elapses in the future
let mut fut = task::spawn(delay_until(clock.now() + ms(YR_5)));
assert_pending!(fut.poll());
clock.turn_for(ms(0));
// boom
let _ = fut.poll();
})
// The delay has elapsed.
assert_ready_ok!(poll!(e));
}
#[test]
fn unpark_is_delayed() {
mock(|clock| {
let mut fut1 = task::spawn(delay_until(clock.now() + ms(100)));
let mut fut2 = task::spawn(delay_until(clock.now() + ms(101)));
let mut fut3 = task::spawn(delay_until(clock.now() + ms(200)));
// A special park that will take much longer than the requested duration
struct MockPark(Clock);
assert_pending!(fut1.poll());
assert_pending!(fut2.poll());
assert_pending!(fut3.poll());
struct MockUnpark;
clock.park_for(ms(500));
impl Park for MockPark {
type Unpark = MockUnpark;
type Error = ();
assert_eq!(clock.advanced(), ms(500));
fn unpark(&self) -> Self::Unpark {
MockUnpark
}
assert_ready!(fut1.poll());
assert_ready!(fut2.poll());
assert_ready!(fut3.poll());
})
fn park(&mut self) -> Result<(), Self::Error> {
panic!("parking forever");
}
fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
assert_eq!(duration, ms(0));
self.0.advance(ms(436));
Ok(())
}
}
impl Unpark for MockUnpark {
fn unpark(&self) {}
}
let clock = Clock::new_frozen();
let mut driver = Driver::new(MockPark(clock.clone()), clock.clone());
let handle = driver.handle();
let mut e1 = task::spawn(delay_until(&handle, clock.now() + ms(100)));
let mut e2 = task::spawn(delay_until(&handle, clock.now() + ms(101)));
let mut e3 = task::spawn(delay_until(&handle, clock.now() + ms(200)));
assert_pending!(poll!(e1));
assert_pending!(poll!(e2));
assert_pending!(poll!(e3));
assert_ok!(driver.park());
assert_eq!(clock.advanced(), ms(500));
assert_ready_ok!(poll!(e1));
assert_ready_ok!(poll!(e2));
assert_ready_ok!(poll!(e3));
}
#[test]
@@ -353,109 +366,57 @@ fn set_timeout_at_deadline_greater_than_max_timer() {
const YR_1: u64 = 365 * 24 * 60 * 60 * 1000;
const YR_5: u64 = 5 * YR_1;
mock(|clock| {
for _ in 0..5 {
clock.turn_for(ms(YR_1));
}
let (mut driver, clock, handle) = setup();
let mut fut = task::spawn(delay_until(clock.now() + ms(1)));
assert_pending!(fut.poll());
for _ in 0..5 {
assert_ok!(driver.park_timeout(ms(YR_1)));
}
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(YR_5) + ms(1));
let mut e = task::spawn(delay_until(&handle, clock.now() + ms(1)));
assert_pending!(poll!(e));
assert_ready!(fut.poll());
});
assert_ok!(driver.park_timeout(ms(1000)));
assert_eq!(clock.advanced(), ms(YR_5) + ms(1));
assert_ready_ok!(poll!(e));
}
#[test]
fn reset_future_delay_before_fire() {
mock(|clock| {
let mut fut = task::spawn(delay_until(clock.now() + ms(100)));
fn setup() -> (Driver<MockPark>, Clock, Handle) {
let clock = Clock::new_frozen();
let driver = Driver::new(MockPark(clock.clone()), clock.clone());
let handle = driver.handle();
assert_pending!(fut.poll());
fut.reset(clock.now() + ms(200));
clock.turn();
assert_eq!(clock.advanced(), ms(192));
assert_pending!(fut.poll());
clock.turn();
assert_eq!(clock.advanced(), ms(200));
assert_ready!(fut.poll());
});
(driver, clock, handle)
}
#[test]
fn reset_past_delay_before_turn() {
mock(|clock| {
let mut fut = task::spawn(delay_until(clock.now() + ms(100)));
assert_pending!(fut.poll());
fut.reset(clock.now() + ms(80));
clock.turn();
assert_eq!(clock.advanced(), ms(64));
assert_pending!(fut.poll());
clock.turn();
assert_eq!(clock.advanced(), ms(80));
assert_ready!(fut.poll());
});
fn delay_until(handle: &Handle, when: Instant) -> Arc<Entry> {
Entry::new(&handle, when, ms(0))
}
#[test]
fn reset_past_delay_before_fire() {
mock(|clock| {
let mut fut = task::spawn(delay_until(clock.now() + ms(100)));
struct MockPark(Clock);
assert_pending!(fut.poll());
clock.turn_for(ms(10));
struct MockUnpark;
assert_pending!(fut.poll());
fut.reset(clock.now() + ms(80));
impl Park for MockPark {
type Unpark = MockUnpark;
type Error = ();
clock.turn();
assert_eq!(clock.advanced(), ms(64));
fn unpark(&self) -> Self::Unpark {
MockUnpark
}
assert_pending!(fut.poll());
fn park(&mut self) -> Result<(), Self::Error> {
panic!("parking forever");
}
clock.turn();
assert_eq!(clock.advanced(), ms(90));
assert_ready!(fut.poll());
});
fn park_timeout(&mut self, duration: Duration) -> Result<(), Self::Error> {
self.0.advance(duration);
Ok(())
}
}
#[test]
fn reset_future_delay_after_fire() {
mock(|clock| {
let mut fut = task::spawn(delay_until(clock.now() + ms(100)));
assert_pending!(fut.poll());
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(64));
clock.turn();
assert_eq!(clock.advanced(), ms(100));
assert_ready!(fut.poll());
fut.reset(clock.now() + ms(10));
assert_pending!(fut.poll());
clock.turn_for(ms(1000));
assert_eq!(clock.advanced(), ms(110));
assert_ready!(fut.poll());
});
impl Unpark for MockUnpark {
fn unpark(&self) {}
}
fn ms(n: u64) -> Duration {
tokio/src/time/tests/test_queue.rs
-369
View File
@@ -1,369 +0,0 @@
#![warn(rust_2018_idioms)]
use crate::time::tests::mock_clock::mock;
use crate::time::{DelayQueue, Duration};
use tokio_test::{assert_ok, assert_pending, assert_ready, task};
macro_rules! poll {
($queue:ident) => {
$queue.enter(|cx, mut queue| queue.poll_expired(cx))
};
}
macro_rules! assert_ready_ok {
($e:expr) => {{
assert_ok!(match assert_ready!($e) {
Some(v) => v,
None => panic!("None"),
})
}};
}
#[test]
fn single_immediate_delay() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let _key = queue.insert_at("foo", clock.now());
let entry = assert_ready_ok!(poll!(queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none())
});
}
#[test]
fn multi_immediate_delays() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let _k = queue.insert_at("1", clock.now());
let _k = queue.insert_at("2", clock.now());
let _k = queue.insert_at("3", clock.now());
let mut res = vec![];
while res.len() < 3 {
let entry = assert_ready_ok!(poll!(queue));
res.push(entry.into_inner());
}
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none());
res.sort();
assert_eq!("1", res[0]);
assert_eq!("2", res[1]);
assert_eq!("3", res[2]);
});
}
#[test]
fn single_short_delay() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let _key = queue.insert_at("foo", clock.now() + ms(5));
assert_pending!(poll!(queue));
clock.turn_for(ms(1));
assert!(!queue.is_woken());
clock.turn_for(ms(5));
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none());
});
}
#[test]
fn multi_delay_at_start() {
let long = 262_144 + 9 * 4096;
let delays = &[1000, 2, 234, long, 60, 10];
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
// Setup the delays
for &i in delays {
let _key = queue.insert_at(i, clock.now() + ms(i));
}
assert_pending!(poll!(queue));
assert!(!queue.is_woken());
for elapsed in 0..1200 {
clock.turn_for(ms(1));
let elapsed = elapsed + 1;
if delays.contains(&elapsed) {
assert!(queue.is_woken());
assert_ready!(poll!(queue));
assert_pending!(poll!(queue));
} else if queue.is_woken() {
let cascade = &[192, 960];
assert!(cascade.contains(&elapsed), "elapsed={}", elapsed);
assert_pending!(poll!(queue));
}
}
});
}
#[test]
fn insert_in_past_fires_immediately() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let now = clock.now();
clock.turn_for(ms(10));
queue.insert_at("foo", now);
assert_ready!(poll!(queue));
});
}
#[test]
fn remove_entry() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let key = queue.insert_at("foo", clock.now() + ms(5));
assert_pending!(poll!(queue));
let entry = queue.remove(&key);
assert_eq!(entry.into_inner(), "foo");
clock.turn_for(ms(10));
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none());
});
}
#[test]
fn reset_entry() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let now = clock.now();
let key = queue.insert_at("foo", now + ms(5));
assert_pending!(poll!(queue));
clock.turn_for(ms(1));
queue.reset_at(&key, now + ms(10));
assert_pending!(poll!(queue));
clock.turn_for(ms(7));
assert!(!queue.is_woken());
assert_pending!(poll!(queue));
clock.turn_for(ms(3));
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none())
});
}
#[test]
fn reset_much_later() {
// Reproduces tokio-rs/tokio#849.
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
clock.turn_for(ms(1));
let key = queue.insert_at("foo", epoch + ms(200));
assert_pending!(poll!(queue));
clock.turn_for(ms(3));
queue.reset_at(&key, epoch + ms(5));
clock.turn_for(ms(20));
assert!(queue.is_woken());
});
}
#[test]
fn reset_twice() {
// Reproduces tokio-rs/tokio#849.
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
clock.turn_for(ms(1));
let key = queue.insert_at("foo", epoch + ms(200));
assert_pending!(poll!(queue));
clock.turn_for(ms(3));
queue.reset_at(&key, epoch + ms(50));
clock.turn_for(ms(20));
queue.reset_at(&key, epoch + ms(40));
clock.turn_for(ms(20));
assert!(queue.is_woken());
});
}
#[test]
fn remove_expired_item() {
mock(|clock| {
let mut queue = DelayQueue::new();
let now = clock.now();
clock.turn_for(ms(10));
let key = queue.insert_at("foo", now);
let entry = queue.remove(&key);
assert_eq!(entry.into_inner(), "foo");
})
}
#[test]
fn expires_before_last_insert() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
queue.insert_at("foo", epoch + ms(10_000));
// Delay should be set to 8.192s here.
assert_pending!(poll!(queue));
// Delay should be set to the delay of the new item here
queue.insert_at("bar", epoch + ms(600));
assert_pending!(poll!(queue));
clock.advance(ms(600));
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "bar");
})
}
#[test]
fn multi_reset() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
let one = queue.insert_at("one", epoch + ms(200));
let two = queue.insert_at("two", epoch + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&one, epoch + ms(300));
queue.reset_at(&two, epoch + ms(350));
queue.reset_at(&one, epoch + ms(400));
})
}
#[test]
fn expire_first_key_when_reset_to_expire_earlier() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
let one = queue.insert_at("one", epoch + ms(200));
queue.insert_at("two", epoch + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&one, epoch + ms(100));
clock.advance(ms(100));
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "one");
})
}
#[test]
fn expire_second_key_when_reset_to_expire_earlier() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
queue.insert_at("one", epoch + ms(200));
let two = queue.insert_at("two", epoch + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&two, epoch + ms(100));
clock.advance(ms(100));
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "two");
})
}
#[test]
fn reset_first_expiring_item_to_expire_later() {
mock(|clock| {
let mut queue = task::spawn(DelayQueue::new());
let epoch = clock.now();
let one = queue.insert_at("one", epoch + ms(200));
let _two = queue.insert_at("two", epoch + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&one, epoch + ms(300));
clock.advance(ms(250));
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "two");
})
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
tokio/tests/time_delay.rs
+160
View File
@@ -0,0 +1,160 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
use tokio::time::{self, Duration, Instant};
use tokio_test::{assert_pending, task};
macro_rules! assert_elapsed {
($now:expr, $ms:expr) => {{
let elapsed = $now.elapsed();
let lower = ms($ms);
// Handles ms rounding
assert!(
elapsed >= lower && elapsed <= lower + ms(1),
"actual = {:?}, expected = {:?}",
elapsed,
lower
);
}};
}
#[tokio::test]
async fn immediate_delay() {
time::pause();
let now = Instant::now();
// Ready!
time::delay_until(now).await;
assert_elapsed!(now, 0);
}
#[tokio::test]
async fn delayed_delay_level_0() {
time::pause();
for &i in &[1, 10, 60] {
let now = Instant::now();
time::delay_until(now + ms(i)).await;
assert_elapsed!(now, i);
}
}
#[tokio::test]
async fn sub_ms_delayed_delay() {
time::pause();
for _ in 0..5 {
let now = Instant::now();
let deadline = now + ms(1) + Duration::new(0, 1);
time::delay_until(deadline).await;
assert_elapsed!(now, 1);
}
}
#[tokio::test]
async fn delayed_delay_wrapping_level_0() {
time::pause();
time::delay_for(ms(5)).await;
let now = Instant::now();
time::delay_until(now + ms(60)).await;
assert_elapsed!(now, 60);
}
#[tokio::test]
async fn reset_future_delay_before_fire() {
time::pause();
let now = Instant::now();
let mut delay = task::spawn(time::delay_until(now + ms(100)));
assert_pending!(delay.poll());
let mut delay = delay.into_inner();
delay.reset(Instant::now() + ms(200));
delay.await;
assert_elapsed!(now, 200);
}
#[tokio::test]
async fn reset_past_delay_before_turn() {
time::pause();
let now = Instant::now();
let mut delay = task::spawn(time::delay_until(now + ms(100)));
assert_pending!(delay.poll());
let mut delay = delay.into_inner();
delay.reset(now + ms(80));
delay.await;
assert_elapsed!(now, 80);
}
#[tokio::test]
async fn reset_past_delay_before_fire() {
time::pause();
let now = Instant::now();
let mut delay = task::spawn(time::delay_until(now + ms(100)));
assert_pending!(delay.poll());
let mut delay = delay.into_inner();
time::delay_for(ms(10)).await;
delay.reset(now + ms(80));
delay.await;
assert_elapsed!(now, 80);
}
#[tokio::test]
async fn reset_future_delay_after_fire() {
time::pause();
let now = Instant::now();
let mut delay = time::delay_until(now + ms(100));
(&mut delay).await;
assert_elapsed!(now, 100);
delay.reset(now + ms(110));
delay.await;
assert_elapsed!(now, 110);
}
#[test]
#[should_panic]
fn creating_delay_outside_of_context() {
let now = Instant::now();
// This creates a delay outside of the context of a mock timer. This tests
// that it will panic.
let _fut = time::delay_until(now + ms(500));
}
#[should_panic]
#[tokio::test]
async fn greater_than_max() {
const YR_5: u64 = 5 * 365 * 24 * 60 * 60 * 1000;
time::delay_until(Instant::now() + ms(YR_5)).await;
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}
tokio/tests/time_delay_queue.rs
+389
View File
@@ -0,0 +1,389 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
use tokio::time::{self, delay_for, DelayQueue, Duration, Instant};
use tokio_test::{assert_ok, assert_pending, assert_ready, task};
macro_rules! poll {
($queue:ident) => {
$queue.enter(|cx, mut queue| queue.poll_expired(cx))
};
}
macro_rules! assert_ready_ok {
($e:expr) => {{
assert_ok!(match assert_ready!($e) {
Some(v) => v,
None => panic!("None"),
})
}};
}
#[tokio::test]
async fn single_immediate_delay() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let _key = queue.insert_at("foo", Instant::now());
// Advance time by 1ms to handle thee rounding
delay_for(ms(1)).await;
assert_ready_ok!(poll!(queue));
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none())
}
#[tokio::test]
async fn multi_immediate_delays() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let _k = queue.insert_at("1", Instant::now());
let _k = queue.insert_at("2", Instant::now());
let _k = queue.insert_at("3", Instant::now());
delay_for(ms(1)).await;
let mut res = vec![];
while res.len() < 3 {
let entry = assert_ready_ok!(poll!(queue));
res.push(entry.into_inner());
}
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none());
res.sort();
assert_eq!("1", res[0]);
assert_eq!("2", res[1]);
assert_eq!("3", res[2]);
}
#[tokio::test]
async fn single_short_delay() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let _key = queue.insert_at("foo", Instant::now() + ms(5));
assert_pending!(poll!(queue));
delay_for(ms(1)).await;
assert!(!queue.is_woken());
delay_for(ms(5)).await;
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue));
assert_eq!(*entry.get_ref(), "foo");
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none());
}
#[tokio::test]
async fn multi_delay_at_start() {
time::pause();
let long = 262_144 + 9 * 4096;
let delays = &[1000, 2, 234, long, 60, 10];
let mut queue = task::spawn(DelayQueue::new());
// Setup the delays
for &i in delays {
let _key = queue.insert_at(i, Instant::now() + ms(i));
}
assert_pending!(poll!(queue));
assert!(!queue.is_woken());
for elapsed in 0..1200 {
delay_for(ms(1)).await;
let elapsed = elapsed + 1;
if delays.contains(&elapsed) {
assert!(queue.is_woken());
assert_ready!(poll!(queue));
assert_pending!(poll!(queue));
} else if queue.is_woken() {
let cascade = &[192, 960];
assert!(cascade.contains(&elapsed), "elapsed={}", elapsed);
assert_pending!(poll!(queue));
}
}
}
#[tokio::test]
async fn insert_in_past_fires_immediately() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
delay_for(ms(10)).await;
queue.insert_at("foo", now);
assert_ready!(poll!(queue));
}
#[tokio::test]
async fn remove_entry() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let key = queue.insert_at("foo", Instant::now() + ms(5));
assert_pending!(poll!(queue));
let entry = queue.remove(&key);
assert_eq!(entry.into_inner(), "foo");
delay_for(ms(10)).await;
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none());
}
#[tokio::test]
async fn reset_entry() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
let key = queue.insert_at("foo", now + ms(5));
assert_pending!(poll!(queue));
delay_for(ms(1)).await;
queue.reset_at(&key, now + ms(10));
assert_pending!(poll!(queue));
delay_for(ms(7)).await;
assert!(!queue.is_woken());
assert_pending!(poll!(queue));
delay_for(ms(3)).await;
assert!(queue.is_woken());
let entry = assert_ready_ok!(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() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
delay_for(ms(1)).await;
let key = queue.insert_at("foo", now + ms(200));
assert_pending!(poll!(queue));
delay_for(ms(3)).await;
queue.reset_at(&key, now + ms(5));
delay_for(ms(20)).await;
assert!(queue.is_woken());
}
// Reproduces tokio-rs/tokio#849.
#[tokio::test]
async fn reset_twice() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
delay_for(ms(1)).await;
let key = queue.insert_at("foo", now + ms(200));
assert_pending!(poll!(queue));
delay_for(ms(3)).await;
queue.reset_at(&key, now + ms(50));
delay_for(ms(20)).await;
queue.reset_at(&key, now + ms(40));
delay_for(ms(20)).await;
assert!(queue.is_woken());
}
#[tokio::test]
async fn remove_expired_item() {
time::pause();
let mut queue = DelayQueue::new();
let now = Instant::now();
delay_for(ms(10)).await;
let key = queue.insert_at("foo", now);
let entry = queue.remove(&key);
assert_eq!(entry.into_inner(), "foo");
}
#[tokio::test]
async fn expires_before_last_insert() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
queue.insert_at("foo", now + ms(10_000));
// Delay should be set to 8.192s here.
assert_pending!(poll!(queue));
// Delay should be set to the delay of the new item here
queue.insert_at("bar", now + ms(600));
assert_pending!(poll!(queue));
delay_for(ms(600)).await;
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "bar");
}
#[tokio::test]
async fn multi_reset() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
let one = queue.insert_at("one", now + ms(200));
let two = queue.insert_at("two", now + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&one, now + ms(300));
queue.reset_at(&two, now + ms(350));
queue.reset_at(&one, now + ms(400));
delay_for(ms(310)).await;
assert_pending!(poll!(queue));
delay_for(ms(50)).await;
let entry = assert_ready_ok!(poll!(queue));
assert_eq!(*entry.get_ref(), "two");
assert_pending!(poll!(queue));
delay_for(ms(50)).await;
let entry = assert_ready_ok!(poll!(queue));
assert_eq!(*entry.get_ref(), "one");
let entry = assert_ready!(poll!(queue));
assert!(entry.is_none())
}
#[tokio::test]
async fn expire_first_key_when_reset_to_expire_earlier() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
let one = queue.insert_at("one", now + ms(200));
queue.insert_at("two", now + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&one, now + ms(100));
delay_for(ms(100)).await;
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "one");
}
#[tokio::test]
async fn expire_second_key_when_reset_to_expire_earlier() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
queue.insert_at("one", now + ms(200));
let two = queue.insert_at("two", now + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&two, now + ms(100));
delay_for(ms(100)).await;
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "two");
}
#[tokio::test]
async fn reset_first_expiring_item_to_expire_later() {
time::pause();
let mut queue = task::spawn(DelayQueue::new());
let now = Instant::now();
let one = queue.insert_at("one", now + ms(200));
let _two = queue.insert_at("two", now + ms(250));
assert_pending!(poll!(queue));
queue.reset_at(&one, now + ms(300));
delay_for(ms(250)).await;
assert!(queue.is_woken());
let entry = assert_ready_ok!(poll!(queue)).into_inner();
assert_eq!(entry, "two");
}
fn ms(n: u64) -> Duration {
Duration::from_millis(n)
}