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

time: allow users to specify Interval behavior when delayed (#3721)

Browse Source
This commit is contained in:
sb64
2021-06-16 13:51:27 +02:00
committed by GitHub
parent 5ad3dd3378
commit e979ad7f2d
3 changed files with 446 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/time/interval.rs
+296 -33
View File
@@ -1,17 +1,20 @@
use crate::future::poll_fn;
use crate::time::{sleep_until, Duration, Instant, Sleep};
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::{convert::TryInto, future::Future};
/// Creates new `Interval` that yields with interval of `duration`. The first
/// tick completes immediately.
/// Creates new [`Interval`] that yields with interval of `period`. The first
/// tick completes immediately. The default [`MissedTickBehavior`] is
/// [`Burst`](MissedTickBehavior::Burst), but this can be configured
/// by calling [`set_missed_tick_behavior`](Interval::set_missed_tick_behavior).
///
/// An interval will tick indefinitely. At any time, the `Interval` value can be
/// dropped. This cancels the interval.
/// An interval will tick indefinitely. At any time, the [`Interval`] value can
/// be dropped. This cancels the interval.
///
/// This function is equivalent to `interval_at(Instant::now(), period)`.
/// This function is equivalent to
/// [`interval_at(Instant::now(), period)`](interval_at).
///
/// # Panics
///
@@ -26,9 +29,9 @@ use std::task::{Context, Poll};
/// async fn main() {
/// let mut interval = time::interval(Duration::from_millis(10));
///
/// interval.tick().await;
/// interval.tick().await;
/// interval.tick().await;
/// interval.tick().await; // ticks immediately
/// interval.tick().await; // ticks after 10ms
/// interval.tick().await; // ticks after 10ms
///
/// // approximately 20ms have elapsed.
/// }
@@ -36,10 +39,10 @@ use std::task::{Context, Poll};
///
/// A simple example using `interval` to execute a task every two seconds.
///
/// The difference between `interval` and [`sleep`] is that an `interval`
/// measures the time since the last tick, which means that `.tick().await`
/// The difference between `interval` and [`sleep`] is that an [`Interval`]
/// measures the time since the last tick, which means that [`.tick().await`]
/// may wait for a shorter time than the duration specified for the interval
/// if some time has passed between calls to `.tick().await`.
/// if some time has passed between calls to [`.tick().await`].
///
/// If the tick in the example below was replaced with [`sleep`], the task
/// would only be executed once every three seconds, and not every two
@@ -64,17 +67,20 @@ use std::task::{Context, Poll};
/// ```
///
/// [`sleep`]: crate::time::sleep()
/// [`.tick().await`]: Interval::tick
pub fn interval(period: Duration) -> Interval {
assert!(period > Duration::new(0, 0), "`period` must be non-zero.");
interval_at(Instant::now(), period)
}
/// Creates new `Interval` that yields with interval of `period` with the
/// first tick completing at `start`.
/// Creates new [`Interval`] that yields with interval of `period` with the
/// first tick completing at `start`. The default [`MissedTickBehavior`] is
/// [`Burst`](MissedTickBehavior::Burst), but this can be configured
/// by calling [`set_missed_tick_behavior`](Interval::set_missed_tick_behavior).
///
/// An interval will tick indefinitely. At any time, the `Interval` value can be
/// dropped. This cancels the interval.
/// An interval will tick indefinitely. At any time, the [`Interval`] value can
/// be dropped. This cancels the interval.
///
/// # Panics
///
@@ -90,9 +96,9 @@ pub fn interval(period: Duration) -> Interval {
/// let start = Instant::now() + Duration::from_millis(50);
/// let mut interval = interval_at(start, Duration::from_millis(10));
///
/// interval.tick().await;
/// interval.tick().await;
/// interval.tick().await;
/// interval.tick().await; // ticks after 50ms
/// interval.tick().await; // ticks after 10ms
/// interval.tick().await; // ticks after 10ms
///
/// // approximately 70ms have elapsed.
/// }
@@ -103,19 +109,249 @@ pub fn interval_at(start: Instant, period: Duration) -> Interval {
Interval {
delay: Box::pin(sleep_until(start)),
period,
missed_tick_behavior: Default::default(),
}
}
/// Interval returned by [`interval`](interval) and [`interval_at`](interval_at).
/// Defines the behavior of an [`Interval`] when it misses a tick.
///
/// Sometimes, an [`Interval`]'s tick is missed. For example, consider the
/// following:
///
/// ```
/// use tokio::time::{self, Duration};
/// # async fn task_that_takes_one_to_three_millis() {}
///
/// #[tokio::main]
/// async fn main() {
/// // ticks every 2 seconds
/// let mut interval = time::interval(Duration::from_millis(2));
/// for _ in 0..5 {
/// interval.tick().await;
/// // if this takes more than 2 milliseconds, a tick will be delayed
/// task_that_takes_one_to_three_millis().await;
/// }
/// }
/// ```
///
/// Generally, a tick is missed if too much time is spent without calling
/// [`Interval::tick()`].
///
/// By default, when a tick is missed, [`Interval`] fires ticks as quickly as it
/// can until it is "caught up" in time to where it should be.
/// `MissedTickBehavior` can be used to specify a different behavior for
/// [`Interval`] to exhibit. Each variant represents a different strategy.
///
/// Note that because the executor cannot guarantee exact precision with timers,
/// these strategies will only apply when the delay is greater than 5
/// milliseconds.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MissedTickBehavior {
/// Tick as fast as possible until caught up.
///
/// When this strategy is used, [`Interval`] schedules ticks "normally" (the
/// same as it would have if the ticks hadn't been delayed), which results
/// in it firing ticks as fast as possible until it is caught up in time to
/// where it should be. Unlike [`Delay`] and [`Skip`], the ticks yielded
/// when `Burst` is used (the [`Instant`]s that [`tick`](Interval::tick)
/// yields) aren't different than they would have been if a tick had not
/// been missed. Like [`Skip`], and unlike [`Delay`], the ticks may be
/// shortened.
///
/// This looks something like this:
/// ```text
/// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 |
/// Actual ticks: | work -----| delay | work | work | work -| work -----|
/// ```
///
/// In code:
///
/// ```
/// use tokio::time::{interval, Duration};
/// # async fn task_that_takes_200_millis() {}
///
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let mut interval = interval(Duration::from_millis(50));
///
/// task_that_takes_200_millis().await;
/// // The `Interval` has missed a tick
///
/// // Since we have exceeded our timeout, this will resolve immediately
/// interval.tick().await;
///
/// // Since we are more than 100ms after the start of `interval`, this will
/// // also resolve immediately.
/// interval.tick().await;
///
/// // Also resolves immediately, because it was supposed to resolve at
/// // 150ms after the start of `interval`
/// interval.tick().await;
///
/// // Resolves immediately
/// interval.tick().await;
///
/// // Since we have gotten to 200ms after the start of `interval`, this
/// // will resolve after 50ms
/// interval.tick().await;
/// # }
/// ```
///
/// This is the default behavior when [`Interval`] is created with
/// [`interval`] and [`interval_at`].
///
/// [`Delay`]: MissedTickBehavior::Delay
/// [`Skip`]: MissedTickBehavior::Skip
Burst,
/// Tick at multiples of `period` from when [`tick`] was called, rather than
/// from `start`.
///
/// When this strategy is used and [`Interval`] has missed a tick, instead
/// of scheduling ticks to fire at multiples of `period` from `start` (the
/// time when the first tick was fired), it schedules all future ticks to
/// happen at a regular `period` from the point when [`tick`] was called.
/// Unlike [`Burst`] and [`Skip`], ticks are not shortened, and they aren't
/// guaranteed to happen at a multiple of `period` from `start` any longer.
///
/// This looks something like this:
/// ```text
/// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 |
/// Actual ticks: | work -----| delay | work -----| work -----| work -----|
/// ```
///
/// In code:
///
/// ```
/// use tokio::time::{interval, Duration, MissedTickBehavior};
/// # async fn task_that_takes_more_than_50_millis() {}
///
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let mut interval = interval(Duration::from_millis(50));
/// interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
///
/// task_that_takes_more_than_50_millis().await;
/// // The `Interval` has missed a tick
///
/// // Since we have exceeded our timeout, this will resolve immediately
/// interval.tick().await;
///
/// // But this one, rather than also resolving immediately, as might happen
/// // with the `Burst` or `Skip` behaviors, will not resolve until
/// // 50ms after the call to `tick` up above. That is, in `tick`, when we
/// // recognize that we missed a tick, we schedule the next tick to happen
/// // 50ms (or whatever the `period` is) from right then, not from when
/// // were were *supposed* to tick
/// interval.tick().await;
/// # }
/// ```
///
/// [`Burst`]: MissedTickBehavior::Burst
/// [`Skip`]: MissedTickBehavior::Skip
/// [`tick`]: Interval::tick
Delay,
/// Skip missed ticks and tick on the next multiple of `period` from
/// `start`.
///
/// When this strategy is used, [`Interval`] schedules the next tick to fire
/// at the next-closest tick that is a multiple of `period` away from
/// `start` (the point where [`Interval`] first ticked). Like [`Burst`], all
/// ticks remain multiples of `period` away from `start`, but unlike
/// [`Burst`], the ticks may not be *one* multiple of `period` away from the
/// last tick. Like [`Delay`], the ticks are no longer the same as they
/// would have been if ticks had not been missed, but unlike [`Delay`], and
/// like [`Burst`], the ticks may be shortened to be less than one `period`
/// away from each other.
///
/// This looks something like this:
/// ```text
/// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 |
/// Actual ticks: | work -----| delay | work ---| work -----| work -----|
/// ```
///
/// In code:
///
/// ```
/// use tokio::time::{interval, Duration, MissedTickBehavior};
/// # async fn task_that_takes_75_millis() {}
///
/// # #[tokio::main(flavor = "current_thread")]
/// # async fn main() {
/// let mut interval = interval(Duration::from_millis(50));
/// interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
///
/// task_that_takes_75_millis().await;
/// // The `Interval` has missed a tick
///
/// // Since we have exceeded our timeout, this will resolve immediately
/// interval.tick().await;
///
/// // This one will resolve after 25ms, 100ms after the start of
/// // `interval`, which is the closest multiple of `period` from the start
/// // of `interval` after the call to `tick` up above.
/// interval.tick().await;
/// # }
/// ```
///
/// [`Burst`]: MissedTickBehavior::Burst
/// [`Delay`]: MissedTickBehavior::Delay
Skip,
}
impl MissedTickBehavior {
/// If a tick is missed, this method is called to determine when the next tick should happen.
fn next_timeout(&self, timeout: Instant, now: Instant, period: Duration) -> Instant {
match self {
Self::Burst => timeout + period,
Self::Delay => now + period,
Self::Skip => {
now + period
- Duration::from_nanos(
((now - timeout).as_nanos() % period.as_nanos())
.try_into()
// This operation is practically guaranteed not to
// fail, as in order for it to fail, `period` would
// have to be longer than `now - timeout`, and both
// would have to be longer than 584 years.
//
// If it did fail, there's not a good way to pass
// the error along to the user, so we just panic.
.expect(
"too much time has elapsed since the interval was supposed to tick",
),
)
}
}
}
}
impl Default for MissedTickBehavior {
/// Returns [`MissedTickBehavior::Burst`].
///
/// For most usecases, the [`Burst`] strategy is what is desired.
/// Additionally, to preserve backwards compatibility, the [`Burst`]
/// strategy must be the default. For these reasons,
/// [`MissedTickBehavior::Burst`] is the default for [`MissedTickBehavior`].
/// See [`Burst`] for more details.
///
/// [`Burst`]: MissedTickBehavior::Burst
fn default() -> Self {
Self::Burst
}
}
/// Interval returned by [`interval`] and [`interval_at`]
///
/// This type allows you to wait on a sequence of instants with a certain
/// duration between each instant. Unlike calling [`sleep`](crate::time::sleep)
/// in a loop, this lets you count the time spent between the calls to `sleep`
/// as well.
/// duration between each instant. Unlike calling [`sleep`] in a loop, this lets
/// you count the time spent between the calls to [`sleep`] as well.
///
/// An `Interval` can be turned into a `Stream` with [`IntervalStream`].
///
/// [`IntervalStream`]: https://docs.rs/tokio-stream/0.1/tokio_stream/wrappers/struct.IntervalStream.html
/// [`IntervalStream`]: https://docs.rs/tokio-stream/latest/tokio_stream/wrappers/struct.IntervalStream.html
/// [`sleep`]: crate::time::sleep
#[derive(Debug)]
pub struct Interval {
/// Future that completes the next time the `Interval` yields a value.
@@ -123,6 +359,9 @@ pub struct Interval {
/// The duration between values yielded by `Interval`.
period: Duration,
/// The strategy `Interval` should use when a tick is missed.
missed_tick_behavior: MissedTickBehavior,
}
impl Interval {
@@ -159,22 +398,46 @@ impl Interval {
///
/// When this method returns `Poll::Pending`, the current task is scheduled
/// to receive a wakeup when the instant has elapsed. Note that on multiple
/// calls to `poll_tick`, only the `Waker` from the `Context` passed to the
/// most recent call is scheduled to receive a wakeup.
/// calls to `poll_tick`, only the [`Waker`](std::task::Waker) from the
/// [`Context`] passed to the most recent call is scheduled to receive a
/// wakeup.
pub fn poll_tick(&mut self, cx: &mut Context<'_>) -> Poll<Instant> {
// Wait for the delay to be done
ready!(Pin::new(&mut self.delay).poll(cx));
// Get the `now` by looking at the `delay` deadline
let now = self.delay.deadline();
// Get the time when we were scheduled to tick
let timeout = self.delay.deadline();
let now = Instant::now();
// If a tick was not missed, and thus we are being called before the
// next tick is due, just schedule the next tick normally, one `period`
// after `timeout`
//
// However, if a tick took excessively long and we are now behind,
// schedule the next tick according to how the user specified with
// `MissedTickBehavior`
let next = if now > timeout + Duration::from_millis(5) {
self.missed_tick_behavior
.next_timeout(timeout, now, self.period)
} else {
timeout + self.period
};
// The next interval value is `duration` after the one that just
// yielded.
let next = now + self.period;
self.delay.as_mut().reset(next);
// Return the current instant
Poll::Ready(now)
// Return the time when we were scheduled to tick
Poll::Ready(timeout)
}
/// Returns the [`MissedTickBehavior`] strategy currently being used.
pub fn missed_tick_behavior(&self) -> MissedTickBehavior {
self.missed_tick_behavior
}
/// Sets the [`MissedTickBehavior`] strategy that should be used.
pub fn set_missed_tick_behavior(&mut self, behavior: MissedTickBehavior) {
self.missed_tick_behavior = behavior;
}
/// Returns the period of the interval.
tokio/src/time/mod.rs
+8 -9
View File
@@ -3,21 +3,21 @@
//! This module provides a number of types for executing code after a set period
//! of time.
//!
//! * `Sleep` is a future that does no work and completes at a specific `Instant`
//! * [`Sleep`] is a future that does no work and completes at a specific [`Instant`]
//! in time.
//!
//! * `Interval` is a stream yielding a value at a fixed period. It is
//! initialized with a `Duration` and repeatedly yields each time the duration
//! * [`Interval`] is a stream yielding a value at a fixed period. It is
//! initialized with a [`Duration`] and repeatedly yields each time the duration
//! elapses.
//!
//! * `Timeout`: Wraps a future or stream, setting an upper bound to the amount
//! * [`Timeout`]: Wraps a future or stream, setting an upper bound to the amount
//! of time it is allowed to execute. If the future or stream does not
//! complete in time, then it is canceled and an error is returned.
//!
//! These types are sufficient for handling a large number of scenarios
//! involving time.
//!
//! These types must be used from within the context of the `Runtime`.
//! These types must be used from within the context of the [`Runtime`](crate::runtime::Runtime).
//!
//! # Examples
//!
@@ -55,8 +55,8 @@
//! A simple example using [`interval`] to execute a task every two seconds.
//!
//! The difference between [`interval`] and [`sleep`] is that an [`interval`]
//! measures the time since the last tick, which means that `.tick().await`
//! may wait for a shorter time than the duration specified for the interval
//! measures the time since the last tick, which means that `.tick().await` may
//! wait for a shorter time than the duration specified for the interval
//! if some time has passed between calls to `.tick().await`.
//!
//! If the tick in the example below was replaced with [`sleep`], the task
@@ -81,7 +81,6 @@
//! }
//! ```
//!
//! [`sleep`]: crate::time::sleep()
//! [`interval`]: crate::time::interval()
mod clock;
@@ -100,7 +99,7 @@ mod instant;
pub use self::instant::Instant;
mod interval;
pub use interval::{interval, interval_at, Interval};
pub use interval::{interval, interval_at, Interval, MissedTickBehavior};
mod timeout;
#[doc(inline)]
tokio/tests/time_interval.rs
+142 -25
View File
@@ -1,56 +1,173 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
use tokio::time::{self, Duration, Instant};
use tokio::time::{self, Duration, Instant, MissedTickBehavior};
use tokio_test::{assert_pending, assert_ready_eq, task};
use std::future::Future;
use std::task::Poll;
// Takes the `Interval` task, `start` variable, and optional time deltas
// For each time delta, it polls the `Interval` and asserts that the result is
// equal to `start` + the specific time delta. Then it asserts that the
// `Interval` is pending.
macro_rules! check_interval_poll {
($i:ident, $start:ident, $($delta:expr),*$(,)?) => {
$(
assert_ready_eq!(poll_next(&mut $i), $start + ms($delta));
)*
assert_pending!(poll_next(&mut $i));
};
($i:ident, $start:ident) => {
check_interval_poll!($i, $start,);
};
}
#[tokio::test]
#[should_panic]
async fn interval_zero_duration() {
let _ = time::interval_at(Instant::now(), ms(0));
}
#[tokio::test]
async fn usage() {
time::pause();
// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 |
// Actual ticks: | work -----| delay | work | work | work -| work -----|
// Poll behavior: | | | | | | | |
// | | | | | | | |
// Ready(s) | | Ready(s + 2p) | | | |
// Pending | Ready(s + 3p) | | |
// Ready(s + p) Ready(s + 4p) | |
// Ready(s + 5p) |
// Ready(s + 6p)
#[tokio::test(start_paused = true)]
async fn burst() {
let start = Instant::now();
// TODO: Skip this
// This is necessary because the timer is only so granular, and in order for
// all our ticks to resolve, the time needs to be 1ms ahead of what we
// expect, so that the runtime will see that it is time to resolve the timer
time::advance(ms(1)).await;
let mut i = task::spawn(time::interval_at(start, ms(300)));
assert_ready_eq!(poll_next(&mut i), start);
assert_pending!(poll_next(&mut i));
check_interval_poll!(i, start, 0);
time::advance(ms(100)).await;
assert_pending!(poll_next(&mut i));
check_interval_poll!(i, start);
time::advance(ms(200)).await;
assert_ready_eq!(poll_next(&mut i), start + ms(300));
assert_pending!(poll_next(&mut i));
check_interval_poll!(i, start, 300);
time::advance(ms(400)).await;
assert_ready_eq!(poll_next(&mut i), start + ms(600));
assert_pending!(poll_next(&mut i));
time::advance(ms(650)).await;
check_interval_poll!(i, start, 600, 900);
time::advance(ms(500)).await;
assert_ready_eq!(poll_next(&mut i), start + ms(900));
assert_ready_eq!(poll_next(&mut i), start + ms(1200));
assert_pending!(poll_next(&mut i));
time::advance(ms(200)).await;
check_interval_poll!(i, start);
time::advance(ms(100)).await;
check_interval_poll!(i, start, 1200);
time::advance(ms(250)).await;
check_interval_poll!(i, start, 1500);
time::advance(ms(300)).await;
check_interval_poll!(i, start, 1800);
}
// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 |
// Actual ticks: | work -----| delay | work -----| work -----| work -----|
// Poll behavior: | | | | | | | |
// | | | | | | | |
// Ready(s) | | Ready(s + 2p) | | | |
// Pending | Pending | | |
// Ready(s + p) Ready(s + 2p + d) | |
// Ready(s + 3p + d) |
// Ready(s + 4p + d)
#[tokio::test(start_paused = true)]
async fn delay() {
let start = Instant::now();
// This is necessary because the timer is only so granular, and in order for
// all our ticks to resolve, the time needs to be 1ms ahead of what we
// expect, so that the runtime will see that it is time to resolve the timer
time::advance(ms(1)).await;
let mut i = task::spawn(time::interval_at(start, ms(300)));
i.set_missed_tick_behavior(MissedTickBehavior::Delay);
check_interval_poll!(i, start, 0);
time::advance(ms(100)).await;
check_interval_poll!(i, start);
time::advance(ms(200)).await;
check_interval_poll!(i, start, 300);
time::advance(ms(650)).await;
check_interval_poll!(i, start, 600);
time::advance(ms(100)).await;
check_interval_poll!(i, start);
// We have to add one here for the same reason as is above.
// Because `Interval` has reset its timer according to `Instant::now()`,
// we have to go forward 1 more millisecond than is expected so that the
// runtime realizes that it's time to resolve the timer.
time::advance(ms(201)).await;
// We add one because when using the `Delay` behavior, `Interval`
// adds the `period` from `Instant::now()`, which will always be off by one
// because we have to advance time by 1 (see above).
check_interval_poll!(i, start, 1251);
time::advance(ms(300)).await;
// Again, we add one.
check_interval_poll!(i, start, 1551);
time::advance(ms(300)).await;
check_interval_poll!(i, start, 1851);
}
// Expected ticks: | 1 | 2 | 3 | 4 | 5 | 6 |
// Actual ticks: | work -----| delay | work ---| work -----| work -----|
// Poll behavior: | | | | | | |
// | | | | | | |
// Ready(s) | | Ready(s + 2p) | | |
// Pending | Ready(s + 4p) | |
// Ready(s + p) Ready(s + 5p) |
// Ready(s + 6p)
#[tokio::test(start_paused = true)]
async fn skip() {
let start = Instant::now();
// This is necessary because the timer is only so granular, and in order for
// all our ticks to resolve, the time needs to be 1ms ahead of what we
// expect, so that the runtime will see that it is time to resolve the timer
time::advance(ms(1)).await;
let mut i = task::spawn(time::interval_at(start, ms(300)));
i.set_missed_tick_behavior(MissedTickBehavior::Skip);
check_interval_poll!(i, start, 0);
time::advance(ms(100)).await;
check_interval_poll!(i, start);
time::advance(ms(200)).await;
check_interval_poll!(i, start, 300);
time::advance(ms(650)).await;
check_interval_poll!(i, start, 600);
time::advance(ms(250)).await;
check_interval_poll!(i, start, 1200);
time::advance(ms(300)).await;
check_interval_poll!(i, start, 1500);
time::advance(ms(300)).await;
check_interval_poll!(i, start, 1800);
}
fn poll_next(interval: &mut task::Spawn<time::Interval>) -> Poll<Instant> {
interval.enter(|cx, mut interval| {
tokio::pin! {
let fut = interval.tick();
}
fut.poll(cx)
})
interval.enter(|cx, mut interval| interval.poll_tick(cx))
}
fn ms(n: u64) -> Duration {