1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230
//! Definitions for [`SleepProviderExt`] and related types.
use crate::traits::SleepProvider;
use futures::{Future, FutureExt};
use pin_project::pin_project;
use std::{
pin::Pin,
task::{Context, Poll},
time::{Duration, SystemTime},
};
/// An error value given when a function times out.
///
/// This value is generated when the timeout from
/// [`SleepProviderExt::timeout`] expires before the provided future
/// is ready.
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
#[allow(clippy::exhaustive_structs)]
pub struct TimeoutError;
impl std::error::Error for TimeoutError {}
impl std::fmt::Display for TimeoutError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Timeout expired")
}
}
impl From<TimeoutError> for std::io::Error {
fn from(err: TimeoutError) -> std::io::Error {
std::io::Error::new(std::io::ErrorKind::TimedOut, err)
}
}
/// An extension trait on [`SleepProvider`] for timeouts and clock delays.
pub trait SleepProviderExt: SleepProvider {
/// Wrap a [`Future`] with a timeout.
///
/// The output of the new future will be the returned value of
/// `future` if it completes within `duration`. Otherwise, it
/// will be `Err(TimeoutError)`.
///
/// # Limitations
///
/// This uses [`SleepProvider::sleep`] for its timer, and is
/// subject to the same limitations.
#[must_use = "timeout() returns a future, which does nothing unless used"]
fn timeout<F: Future>(&self, duration: Duration, future: F) -> Timeout<F, Self::SleepFuture> {
let sleep_future = self.sleep(duration);
Timeout {
future,
sleep_future,
}
}
/// Pause until the wall-clock is at `when` or later, trying to
/// recover from clock jumps.
///
/// Unlike [`SleepProvider::sleep()`], the future returned by this function will
/// wake up periodically to check the current time, and see if
/// it is at or past the target.
///
/// # Limitations
///
/// The ability of this function to detect clock jumps is limited
/// to its granularity; it may finish a while after the declared
/// wallclock time if the system clock jumps forward.
///
/// This function does not detect backward clock jumps; arguably,
/// we should have another function to do that.
///
/// This uses [`SleepProvider::sleep`] for its timer, and is
/// subject to the same limitations.
#[must_use = "sleep_until_wallclock() returns a future, which does nothing unless used"]
fn sleep_until_wallclock(&self, when: SystemTime) -> SleepUntilWallclock<'_, Self> {
SleepUntilWallclock {
provider: self,
target: when,
sleep_future: None,
}
}
}
impl<T: SleepProvider> SleepProviderExt for T {}
/// A timeout returned by [`SleepProviderExt::timeout`].
#[pin_project]
pub struct Timeout<T, S> {
/// The future we want to execute.
#[pin]
future: T,
/// The future implementing the timeout.
#[pin]
sleep_future: S,
}
impl<T, S> Future for Timeout<T, S>
where
T: Future,
S: Future<Output = ()>,
{
type Output = Result<T::Output, TimeoutError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
if let Poll::Ready(x) = this.future.poll(cx) {
return Poll::Ready(Ok(x));
}
match this.sleep_future.poll(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(()) => Poll::Ready(Err(TimeoutError)),
}
}
}
/// A future implementing [`SleepProviderExt::sleep_until_wallclock`].
pub struct SleepUntilWallclock<'a, SP: SleepProvider + ?Sized> {
/// Reference to the provider that we use to make new SleepFutures.
provider: &'a SP,
/// The time that we are waiting for.
target: SystemTime,
/// The future representing our current delay.
sleep_future: Option<Pin<Box<SP::SleepFuture>>>,
}
impl<'a, SP> Future for SleepUntilWallclock<'a, SP>
where
SP: SleepProvider + ?Sized,
{
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
// Strategy: we implement sleep_until_wallclock by
// waiting in increments of up to MAX_SLEEP, checking the
// wall clock before and after each increment. This makes
// us wake up a bit more frequently, but enables us to detect it
// if the system clock jumps forward.
let target = self.target;
loop {
let now = self.provider.wallclock();
if now >= target {
return Poll::Ready(());
}
let (last_delay, delay) = calc_next_delay(now, target);
// Note that we store this future to keep it from being
// cancelled, even though we don't ever poll it more than
// once.
//
// TODO: I'm not sure that it's actually necessary to keep
// this future around.
self.sleep_future.take();
let mut sleep_future = Box::pin(self.provider.sleep(delay));
match sleep_future.poll_unpin(cx) {
Poll::Pending => {
self.sleep_future = Some(sleep_future);
return Poll::Pending;
}
Poll::Ready(()) => {
if last_delay {
return Poll::Ready(());
}
}
}
}
}
}
/// We never sleep more than this much, in case our system clock jumps.
///
/// Note that there's a tradeoff here: Making this duration
/// shorter helps our accuracy, but makes us wake up more
/// frequently and consume more CPU.
const MAX_SLEEP: Duration = Duration::from_secs(600);
/// Return the amount of time we should wait next, when running
/// sleep_until_wallclock(). Also return a boolean indicating whether we
/// expect this to be the final delay.
///
/// (This is a separate function for testing.)
pub(crate) fn calc_next_delay(now: SystemTime, when: SystemTime) -> (bool, Duration) {
let remainder = when
.duration_since(now)
.unwrap_or_else(|_| Duration::from_secs(0));
if remainder > MAX_SLEEP {
(false, MAX_SLEEP)
} else {
(true, remainder)
}
}
#[cfg(test)]
mod test {
// @@ begin test lint list maintained by maint/add_warning @@
#![allow(clippy::bool_assert_comparison)]
#![allow(clippy::clone_on_copy)]
#![allow(clippy::dbg_macro)]
#![allow(clippy::mixed_attributes_style)]
#![allow(clippy::print_stderr)]
#![allow(clippy::print_stdout)]
#![allow(clippy::single_char_pattern)]
#![allow(clippy::unwrap_used)]
#![allow(clippy::unchecked_duration_subtraction)]
#![allow(clippy::useless_vec)]
#![allow(clippy::needless_pass_by_value)]
//! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
#![allow(clippy::erasing_op)]
use super::*;
#[test]
fn sleep_delay() {
fn calc(now: SystemTime, when: SystemTime) -> Duration {
calc_next_delay(now, when).1
}
let minute = Duration::from_secs(60);
let second = Duration::from_secs(1);
let start = SystemTime::now();
let target = start + 30 * minute;
assert_eq!(calc(start, target), minute * 10);
assert_eq!(calc(target + minute, target), minute * 0);
assert_eq!(calc(target, target), minute * 0);
assert_eq!(calc(target - second, target), second);
assert_eq!(calc(target - minute * 9, target), minute * 9);
assert_eq!(calc(target - minute * 11, target), minute * 10);
}
}