ntex_util/time/

mod.rs

//! Utilities for tracking time.
//!
//! Use [`sleep`], [`deadline`] or [`interval`] for asynchronous waits. To run a
//! short synchronous callback without polling a timer future, retain a
//! [`CallbackTimer`].
#![allow(
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    clippy::cast_possible_wrap
)]
use std::{cmp, future::Future, future::poll_fn, pin::Pin, task, task::Poll};

mod callbacks;
mod types;
mod wheel;

pub use self::callbacks::CallbackTimer;
pub use self::types::{Millis, Seconds};
pub use self::wheel::{TimerHandle, now, system_time};

/// Waits until `dur` has elapsed.
///
/// No work is performed while awaiting the returned [`Sleep`]. Timers have a
/// granularity of approximately 16 milliseconds and are not suitable for
/// high-resolution timing. A zero duration still waits for at least one timer
/// tick.
#[inline]
pub fn sleep<T: Into<Millis>>(dur: T) -> Sleep {
    Sleep::new(dur.into())
}

/// Waits until `dur` has elapsed.
///
/// Unlike [`sleep`], a zero-duration deadline never completes.
#[inline]
pub fn deadline<T: Into<Millis>>(dur: T) -> Deadline {
    Deadline::new(dur.into())
}

/// Creates an [`Interval`] that ticks every `period`.
///
/// The first tick completes immediately, and each later tick completes
/// `period` after the previous one was observed. An interval will tick
/// indefinitely. At any time, the [`Interval`] value can be dropped. This
/// cancels the interval.
#[inline]
pub fn interval<T: Into<Millis>>(period: T) -> Interval {
    Interval::new(period.into())
}

/// Requires a future to complete before `dur` has elapsed.
///
/// If the future completes before the duration has elapsed, then the completed
/// value is returned. Otherwise, an error is returned and the future is
/// canceled. A zero duration still represents an active timeout of at least
/// one timer tick; use [`timeout_checked`] to disable the timeout with zero. If
/// the future and timer are both ready during the same poll, the future wins.
#[inline]
pub fn timeout<T, U>(dur: U, future: T) -> Timeout<T>
where
    T: Future,
    U: Into<Millis>,
{
    Timeout::new_with_delay(future, Sleep::new(dur.into()))
}

/// Requires a future to complete before `dur` has elapsed.
///
/// If the future completes before the duration has elapsed, then the completed
/// value is returned. Otherwise, an error is returned and the future is
/// canceled. A zero duration disables the timeout.
#[inline]
pub fn timeout_checked<T, U>(dur: U, future: T) -> TimeoutChecked<T>
where
    T: Future,
    U: Into<Millis>,
{
    TimeoutChecked::new_with_delay(future, dur.into())
}

/// Future returned by [`sleep`].
///
/// # Examples
///
/// Wait 100ms and print "100 ms have elapsed".
///
/// ```
/// use ntex::time::{sleep, Millis};
///
/// #[ntex::main]
/// async fn main() {
///     sleep(Millis(100)).await;
///     println!("100 ms have elapsed");
/// }
/// ```
#[derive(Debug)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Sleep {
    // The link between the `Sleep` instance and the timer that drives it.
    hnd: TimerHandle,
}

impl Sleep {
    /// Creates a new sleep future.
    ///
    /// A zero duration is rounded up to 1 ms.
    #[inline]
    pub fn new(duration: Millis) -> Sleep {
        Sleep {
            hnd: TimerHandle::new(u64::from(cmp::max(duration.0, 1))),
        }
    }

    /// Returns `true` if `Sleep` has elapsed.
    #[inline]
    pub fn is_elapsed(&self) -> bool {
        self.hnd.is_elapsed()
    }

    /// Completes the timer immediately.
    #[inline]
    pub fn elapse(&self) {
        self.hnd.elapse();
    }

    /// Resets the `Sleep` instance to a new deadline.
    ///
    /// Calling this function allows changing the instant at which the `Sleep`
    /// future completes without having to create new associated state.
    ///
    /// This function can be called both before and after the future has
    /// completed. A zero duration is rounded up to 1 ms.
    pub fn reset<T: Into<Millis>>(&self, millis: T) {
        self.hnd.reset(u64::from(cmp::max(millis.into().0, 1)));
    }

    #[inline]
    /// Waits until this timer has elapsed.
    pub async fn wait(&self) {
        poll_fn(|cx| self.hnd.poll_elapsed(cx)).await;
    }

    #[inline]
    /// Polls until this timer has elapsed.
    pub fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<()> {
        self.hnd.poll_elapsed(cx)
    }
}

impl Future for Sleep {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
        self.hnd.poll_elapsed(cx)
    }
}

/// Future returned by [`deadline`].
///
/// # Examples
///
/// Wait 100ms and print "100 ms have elapsed".
///
/// ```
/// use ntex::time::{deadline, Millis};
///
/// #[ntex::main]
/// async fn main() {
///     deadline(Millis(100)).await;
///     println!("100 ms have elapsed");
/// }
/// ```
#[derive(Debug)]
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Deadline {
    hnd: Option<TimerHandle>,
}

impl Deadline {
    /// Creates a new deadline future.
    ///
    /// A zero duration creates a deadline that never completes.
    #[inline]
    pub fn new(duration: Millis) -> Deadline {
        if duration.0 != 0 {
            Deadline {
                hnd: Some(TimerHandle::new(u64::from(duration.0))),
            }
        } else {
            Deadline { hnd: None }
        }
    }

    #[inline]
    /// Waits until this deadline has elapsed.
    pub async fn wait(&self) {
        poll_fn(|cx| self.poll_elapsed(cx)).await;
    }

    /// Resets the `Deadline` instance to a new deadline.
    ///
    /// Calling this function allows changing the instant at which the `Deadline`
    /// future completes without having to create new associated state.
    ///
    /// This function can be called both before and after the future has
    /// completed.
    pub fn reset<T: Into<Millis>>(&mut self, millis: T) {
        let millis = millis.into();
        if millis.0 != 0 {
            if let Some(ref mut hnd) = self.hnd {
                hnd.reset(u64::from(millis.0));
            } else {
                self.hnd = Some(TimerHandle::new(u64::from(millis.0)));
            }
        } else {
            let _ = self.hnd.take();
        }
    }

    /// Returns `true` if `Deadline` has elapsed.
    ///
    /// A disabled zero-duration deadline is reported as elapsed here even
    /// though polling it remains pending. Use this method to determine whether
    /// there is an active future deadline to wait for.
    #[inline]
    pub fn is_elapsed(&self) -> bool {
        self.hnd.as_ref().is_none_or(TimerHandle::is_elapsed)
    }

    #[inline]
    /// Polls until this deadline has elapsed.
    ///
    /// A zero-duration deadline remains pending until it is reset.
    pub fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<()> {
        self.hnd
            .as_ref()
            .map_or(Poll::Pending, |t| t.poll_elapsed(cx))
    }
}

impl Future for Deadline {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
        self.poll_elapsed(cx)
    }
}

pin_project_lite::pin_project! {
    /// Future returned by [`timeout`](timeout).
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    #[derive(Debug)]
    pub struct Timeout<T> {
        #[pin]
        value: T,
        delay: Sleep,
    }
}

impl<T> Timeout<T> {
    pub(crate) fn new_with_delay(value: T, delay: Sleep) -> Timeout<T> {
        Timeout { value, delay }
    }
}

impl<T> Future for Timeout<T>
where
    T: Future,
{
    type Output = Result<T::Output, ()>;

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
        let this = self.project();

        // First, try polling the future
        if let Poll::Ready(v) = this.value.poll(cx) {
            return Poll::Ready(Ok(v));
        }

        // Now check the timer
        match this.delay.poll_elapsed(cx) {
            Poll::Ready(()) => Poll::Ready(Err(())),
            Poll::Pending => Poll::Pending,
        }
    }
}

pin_project_lite::pin_project! {
    /// Future returned by [`timeout_checked`](timeout_checked).
    #[must_use = "futures do nothing unless you `.await` or poll them"]
    pub struct TimeoutChecked<T> {
        #[pin]
        state: TimeoutCheckedState<T>,
    }
}

pin_project_lite::pin_project! {
    #[project = TimeoutCheckedStateProject]
    enum TimeoutCheckedState<T> {
        Timeout{ #[pin] fut: Timeout<T> },
        NoTimeout{ #[pin] fut: T },
    }
}

impl<T> TimeoutChecked<T> {
    pub(crate) fn new_with_delay(value: T, delay: Millis) -> TimeoutChecked<T> {
        if delay.is_zero() {
            TimeoutChecked {
                state: TimeoutCheckedState::NoTimeout { fut: value },
            }
        } else {
            TimeoutChecked {
                state: TimeoutCheckedState::Timeout {
                    fut: Timeout::new_with_delay(value, sleep(delay)),
                },
            }
        }
    }
}

impl<T> Future for TimeoutChecked<T>
where
    T: Future,
{
    type Output = Result<T::Output, ()>;

    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
        match self.project().state.as_mut().project() {
            TimeoutCheckedStateProject::Timeout { fut } => fut.poll(cx),
            TimeoutCheckedStateProject::NoTimeout { fut } => fut.poll(cx).map(Result::Ok),
        }
    }
}

/// An interval returned by [`interval`].
///
/// This type allows you to wait on a sequence of instants with a certain
/// duration between each instant.
#[must_use = "futures do nothing unless you `.await` or poll them"]
#[derive(Debug)]
pub struct Interval {
    hnd: TimerHandle,
    period: u32,
}

impl Interval {
    /// Creates an interval with the specified period.
    ///
    /// The first tick completes immediately. A zero period is rounded up to
    /// 1 ms.
    #[inline]
    pub fn new(period: Millis) -> Interval {
        Interval {
            hnd: TimerHandle::new(0),
            period: cmp::max(period.0, 1),
        }
    }

    #[inline]
    /// Waits for the next interval tick.
    pub async fn tick(&self) {
        poll_fn(|cx| self.poll_tick(cx)).await;
    }

    #[inline]
    /// Polls for the next interval tick.
    pub fn poll_tick(&self, cx: &mut task::Context<'_>) -> Poll<()> {
        if self.hnd.poll_elapsed(cx).is_ready() {
            self.hnd.reset(u64::from(self.period));
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }
}

impl crate::Stream for Interval {
    type Item = ();

    #[inline]
    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
        self.poll_tick(cx).map(|()| Some(()))
    }
}

#[cfg(test)]
mod tests {
    use futures_util::StreamExt;
    use std::{future::poll_fn, rc::Rc, time};

    use super::*;
    use crate::future::lazy;

    /// State Under Test: `now()` returns the start of the current runtime turn.
    ///
    /// Expected Behavior: `now()` is not later than the current instant and is close to it.
    #[ntex::test]
    async fn now_is_current() {
        sleep(Millis(25)).await;

        let now = now();
        let cur = time::Instant::now();
        assert!(cur >= now);
        assert!(cur - now < time::Duration::from_millis(50));
    }

    /// State Under Test: `now()` advances between runtime turns.
    ///
    /// Expected Behavior: Two calls of `now()` made in subsequent resolution interval return different values
    /// and second value is greater than the first one at least by a 1ms interval.
    #[ntex::test]
    async fn now_updates_after_sleep() {
        sleep(Millis(50)).await;

        let first_time = now();

        sleep(Millis(25)).await;

        let second_time = now();
        assert!(second_time - first_time >= time::Duration::from_millis(25));
    }

    /// State Under Test: Two calls of `system_time()` return the same value if they are done within the
    /// resolution interval.
    ///
    /// Expected Behavior: Two back-to-back calls of `now()` return the same value.
    #[ntex::test]
    async fn system_time_service_time_does_not_immediately_change() {
        sleep(Seconds(1)).await;

        assert_eq!(system_time(), system_time());
    }

    /// State Under Test: `system_time()` updates returned value every resolution interval (150ms).
    ///
    /// Expected Behavior: Two calls of `system_time()` made in subsequent resolution interval return different values
    /// and second value is greater than the first one at least by a resolution interval.
    #[ntex::test]
    async fn system_time_service_time_updates_after_resolution_interval() {
        sleep(Millis(100)).await;

        let wait_time = 600;

        let first_time = system_time()
            .duration_since(time::SystemTime::UNIX_EPOCH)
            .unwrap();

        sleep(Millis(wait_time)).await;

        let second_time = system_time()
            .duration_since(time::SystemTime::UNIX_EPOCH)
            .unwrap();

        assert!(
            second_time.checked_sub(first_time).unwrap()
                >= time::Duration::from_millis(u64::from(wait_time))
        );
    }

    #[ntex::test]
    async fn test_sleep_0() {
        sleep(Seconds(1)).await;

        let first_time = time::Instant::now();
        sleep(Millis(0)).await;
        let second_time = time::Instant::now();
        assert!(second_time - first_time >= time::Duration::from_millis(1));

        let first_time = time::Instant::now();
        sleep(Millis(1)).await;
        let second_time = time::Instant::now();
        assert!(second_time - first_time >= time::Duration::from_millis(1));

        let first_time = time::Instant::now();
        let fut = sleep(Millis(10000));
        assert!(!fut.is_elapsed());
        // a zero delay is rounded up to 1 ms
        fut.reset(Millis::ZERO);
        assert!(!fut.is_elapsed());
        fut.await;
        let second_time = time::Instant::now();
        assert!(second_time - first_time >= time::Duration::from_millis(1));

        let fut = sleep(Millis::ZERO);
        assert!(!fut.is_elapsed());
        fut.await;

        // `now()` may advance between two calls, measure
        // with the clock. An elapsed timer completes without waiting for the
        // timer driver, the bound leaves room for scheduling latency
        let first_time = time::Instant::now();
        let fut = Sleep {
            hnd: TimerHandle::new(0),
        };
        assert!(fut.is_elapsed());
        fut.await;
        let second_time = time::Instant::now();
        assert!(second_time - first_time < time::Duration::from_millis(50));

        let first_time = time::Instant::now();
        let fut = Rc::new(sleep(Millis(10_0000)));
        let s = fut.clone();
        ntex::rt::spawn(async move {
            s.elapse();
        });
        poll_fn(|cx| fut.poll_elapsed(cx)).await;
        assert!(fut.is_elapsed());
        let second_time = time::Instant::now();
        assert!(second_time - first_time < time::Duration::from_millis(50));
    }

    #[ntex::test]
    async fn test_deadline() {
        sleep(Seconds(1)).await;

        let first_time = now();
        let dl = deadline(Millis(1));
        dl.await;
        let second_time = now();
        assert!(second_time - first_time >= time::Duration::from_millis(1));
        assert!(timeout(Millis(100), deadline(Millis(0))).await.is_err());

        let mut dl = deadline(Millis(1));
        dl.reset(Millis::ZERO);
        assert!(lazy(|cx| dl.poll_elapsed(cx)).await.is_pending());

        let mut dl = deadline(Millis(1));
        dl.reset(Millis(100));
        let first_time = now();
        dl.await;
        let second_time = now();
        assert!(second_time - first_time >= time::Duration::from_millis(100));

        let mut dl = deadline(Millis(0));
        assert!(dl.is_elapsed());
        dl.reset(Millis(1));
        assert!(lazy(|cx| dl.poll_elapsed(cx)).await.is_pending());

        assert!(format!("{dl:?}").contains("Deadline"));
    }

    #[ntex::test]
    async fn test_interval_zero_period() {
        let int = interval(Millis::ZERO);
        int.tick().await;

        // a zero period is rounded up to 1 ms instead of ticking on every poll
        assert!(lazy(|cx| int.poll_tick(cx)).await.is_pending());
        int.tick().await;
    }

    #[ntex::test]
    async fn test_interval() {
        let mut int = interval(Millis(250));

        // the first tick completes immediately
        let time = time::Instant::now();
        int.tick().await;
        assert!(time.elapsed() < time::Duration::from_millis(50));

        let time = time::Instant::now();
        int.tick().await;
        let elapsed = time.elapsed();
        assert!(
            elapsed > time::Duration::from_millis(200)
                && elapsed < time::Duration::from_millis(450),
            "elapsed: {elapsed:?}"
        );

        let time = time::Instant::now();
        int.next().await;
        let elapsed = time.elapsed();
        assert!(
            elapsed > time::Duration::from_millis(200)
                && elapsed < time::Duration::from_millis(450),
            "elapsed: {elapsed:?}"
        );
    }

    #[ntex::test]
    async fn test_interval_one_sec() {
        let int = interval(Millis::ONE_SEC);
        int.tick().await;

        for _i in 0..3 {
            let time = time::Instant::now();
            int.tick().await;
            let elapsed = time.elapsed();
            assert!(
                elapsed > time::Duration::from_secs(1)
                    && elapsed < time::Duration::from_millis(1300),
                "elapsed: {elapsed:?}"
            );
        }
    }

    #[ntex::test]
    async fn test_timeout_checked() {
        let result = timeout_checked(Millis(200), sleep(Millis(100))).await;
        assert!(result.is_ok());

        // a future that never completes, a late timer driver would elapse
        // a competing sleep in the same pass
        let result = timeout_checked(Millis(5), std::future::pending::<()>()).await;
        assert!(result.is_err());

        let result = timeout_checked(Millis(0), sleep(Millis(100))).await;
        assert!(result.is_ok());
    }

    #[ntex::test]
    async fn sleep_and_deadline_wait() {
        let start = now();
        sleep(Millis(5)).wait().await;
        deadline(Millis(5)).wait().await;
        assert!(now() >= start);

        let hnd = Deadline::new(Millis::ZERO);
        assert!(
            crate::time::timeout(Millis(20), hnd.wait()).await.is_err(),
            "zero deadline never completes"
        );
    }
}