ntex_util/time/
mod.rs

1//! Utilities for tracking time.
2#![allow(
3    clippy::cast_possible_truncation,
4    clippy::cast_sign_loss,
5    clippy::cast_possible_wrap
6)]
7use std::{cmp, future::Future, future::poll_fn, pin::Pin, task, task::Poll};
8
9mod types;
10mod wheel;
11
12pub use self::types::{Millis, Seconds};
13pub use self::wheel::{TimerHandle, now, query_system_time, system_time};
14
15/// Waits until `dur` has elapsed.
16///
17/// No work is performed while awaiting the returned [`Sleep`]. Timers have a
18/// granularity of approximately 16 milliseconds and are not suitable for
19/// high-resolution timing. A zero duration still waits for at least one timer
20/// tick.
21#[inline]
22pub fn sleep<T: Into<Millis>>(dur: T) -> Sleep {
23    Sleep::new(dur.into())
24}
25
26/// Waits until `dur` has elapsed.
27///
28/// Unlike [`sleep`], a zero-duration deadline never completes.
29#[inline]
30pub fn deadline<T: Into<Millis>>(dur: T) -> Deadline {
31    Deadline::new(dur.into())
32}
33
34/// Creates an [`Interval`] that ticks every `period`.
35///
36/// An interval will tick indefinitely. At any time, the [`Interval`] value can
37/// be dropped. This cancels the interval.
38#[inline]
39pub fn interval<T: Into<Millis>>(period: T) -> Interval {
40    Interval::new(period.into())
41}
42
43/// Requires a future to complete before `dur` has elapsed.
44///
45/// If the future completes before the duration has elapsed, then the completed
46/// value is returned. Otherwise, an error is returned and the future is
47/// canceled. A zero duration still represents an active timeout of at least
48/// one timer tick; use [`timeout_checked`] to disable the timeout with zero.
49#[inline]
50pub fn timeout<T, U>(dur: U, future: T) -> Timeout<T>
51where
52    T: Future,
53    U: Into<Millis>,
54{
55    Timeout::new_with_delay(future, Sleep::new(dur.into()))
56}
57
58/// Requires a future to complete before `dur` has elapsed.
59///
60/// If the future completes before the duration has elapsed, then the completed
61/// value is returned. Otherwise, an error is returned and the future is
62/// canceled. A zero duration disables the timeout.
63#[inline]
64pub fn timeout_checked<T, U>(dur: U, future: T) -> TimeoutChecked<T>
65where
66    T: Future,
67    U: Into<Millis>,
68{
69    TimeoutChecked::new_with_delay(future, dur.into())
70}
71
72/// Future returned by [`sleep`].
73///
74/// # Examples
75///
76/// Wait 100ms and print "100 ms have elapsed".
77///
78/// ```
79/// use ntex::time::{sleep, Millis};
80///
81/// #[ntex::main]
82/// async fn main() {
83///     sleep(Millis(100)).await;
84///     println!("100 ms have elapsed");
85/// }
86/// ```
87#[derive(Debug)]
88#[must_use = "futures do nothing unless you `.await` or poll them"]
89pub struct Sleep {
90    // The link between the `Sleep` instance and the timer that drives it.
91    hnd: TimerHandle,
92}
93
94impl Sleep {
95    /// Creates a new sleep future.
96    #[inline]
97    pub fn new(duration: Millis) -> Sleep {
98        Sleep {
99            hnd: TimerHandle::new(u64::from(cmp::max(duration.0, 1))),
100        }
101    }
102
103    /// Returns `true` if `Sleep` has elapsed.
104    #[inline]
105    pub fn is_elapsed(&self) -> bool {
106        self.hnd.is_elapsed()
107    }
108
109    /// Completes the timer immediately.
110    #[inline]
111    pub fn elapse(&self) {
112        self.hnd.elapse();
113    }
114
115    /// Resets the `Sleep` instance to a new deadline.
116    ///
117    /// Calling this function allows changing the instant at which the `Sleep`
118    /// future completes without having to create new associated state.
119    ///
120    /// This function can be called both before and after the future has
121    /// completed.
122    pub fn reset<T: Into<Millis>>(&self, millis: T) {
123        self.hnd.reset(u64::from(millis.into().0));
124    }
125
126    #[inline]
127    /// Waits until this timer has elapsed.
128    pub async fn wait(&self) {
129        poll_fn(|cx| self.hnd.poll_elapsed(cx)).await;
130    }
131
132    #[inline]
133    pub fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<()> {
134        self.hnd.poll_elapsed(cx)
135    }
136}
137
138impl Future for Sleep {
139    type Output = ();
140
141    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
142        self.hnd.poll_elapsed(cx)
143    }
144}
145
146/// Future returned by [`deadline`].
147///
148/// # Examples
149///
150/// Wait 100ms and print "100 ms have elapsed".
151///
152/// ```
153/// use ntex::time::{deadline, Millis};
154///
155/// #[ntex::main]
156/// async fn main() {
157///     deadline(Millis(100)).await;
158///     println!("100 ms have elapsed");
159/// }
160/// ```
161#[derive(Debug)]
162#[must_use = "futures do nothing unless you `.await` or poll them"]
163pub struct Deadline {
164    hnd: Option<TimerHandle>,
165}
166
167impl Deadline {
168    /// Creates a new deadline future.
169    ///
170    /// A zero duration creates a deadline that never completes.
171    #[inline]
172    pub fn new(duration: Millis) -> Deadline {
173        if duration.0 != 0 {
174            Deadline {
175                hnd: Some(TimerHandle::new(u64::from(duration.0))),
176            }
177        } else {
178            Deadline { hnd: None }
179        }
180    }
181
182    #[inline]
183    /// Waits until this deadline has elapsed.
184    pub async fn wait(&self) {
185        poll_fn(|cx| self.poll_elapsed(cx)).await;
186    }
187
188    /// Resets the `Deadline` instance to a new deadline.
189    ///
190    /// Calling this function allows changing the instant at which the `Deadline`
191    /// future completes without having to create new associated state.
192    ///
193    /// This function can be called both before and after the future has
194    /// completed.
195    pub fn reset<T: Into<Millis>>(&mut self, millis: T) {
196        let millis = millis.into();
197        if millis.0 != 0 {
198            if let Some(ref mut hnd) = self.hnd {
199                hnd.reset(u64::from(millis.0));
200            } else {
201                self.hnd = Some(TimerHandle::new(u64::from(millis.0)));
202            }
203        } else {
204            let _ = self.hnd.take();
205        }
206    }
207
208    /// Returns `true` if `Deadline` has elapsed.
209    #[inline]
210    pub fn is_elapsed(&self) -> bool {
211        self.hnd.as_ref().is_none_or(TimerHandle::is_elapsed)
212    }
213
214    #[inline]
215    pub fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<()> {
216        self.hnd
217            .as_ref()
218            .map_or(Poll::Pending, |t| t.poll_elapsed(cx))
219    }
220}
221
222impl Future for Deadline {
223    type Output = ();
224
225    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
226        self.poll_elapsed(cx)
227    }
228}
229
230pin_project_lite::pin_project! {
231    /// Future returned by [`timeout`](timeout).
232    #[must_use = "futures do nothing unless you `.await` or poll them"]
233    #[derive(Debug)]
234    pub struct Timeout<T> {
235        #[pin]
236        value: T,
237        delay: Sleep,
238    }
239}
240
241impl<T> Timeout<T> {
242    pub(crate) fn new_with_delay(value: T, delay: Sleep) -> Timeout<T> {
243        Timeout { value, delay }
244    }
245}
246
247impl<T> Future for Timeout<T>
248where
249    T: Future,
250{
251    type Output = Result<T::Output, ()>;
252
253    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
254        let this = self.project();
255
256        // First, try polling the future
257        if let Poll::Ready(v) = this.value.poll(cx) {
258            return Poll::Ready(Ok(v));
259        }
260
261        // Now check the timer
262        match this.delay.poll_elapsed(cx) {
263            Poll::Ready(()) => Poll::Ready(Err(())),
264            Poll::Pending => Poll::Pending,
265        }
266    }
267}
268
269pin_project_lite::pin_project! {
270    /// Future returned by [`timeout_checked`](timeout_checked).
271    #[must_use = "futures do nothing unless you `.await` or poll them"]
272    pub struct TimeoutChecked<T> {
273        #[pin]
274        state: TimeoutCheckedState<T>,
275    }
276}
277
278pin_project_lite::pin_project! {
279    #[project = TimeoutCheckedStateProject]
280    enum TimeoutCheckedState<T> {
281        Timeout{ #[pin] fut: Timeout<T> },
282        NoTimeout{ #[pin] fut: T },
283    }
284}
285
286impl<T> TimeoutChecked<T> {
287    pub(crate) fn new_with_delay(value: T, delay: Millis) -> TimeoutChecked<T> {
288        if delay.is_zero() {
289            TimeoutChecked {
290                state: TimeoutCheckedState::NoTimeout { fut: value },
291            }
292        } else {
293            TimeoutChecked {
294                state: TimeoutCheckedState::Timeout {
295                    fut: Timeout::new_with_delay(value, sleep(delay)),
296                },
297            }
298        }
299    }
300}
301
302impl<T> Future for TimeoutChecked<T>
303where
304    T: Future,
305{
306    type Output = Result<T::Output, ()>;
307
308    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
309        match self.project().state.as_mut().project() {
310            TimeoutCheckedStateProject::Timeout { fut } => fut.poll(cx),
311            TimeoutCheckedStateProject::NoTimeout { fut } => fut.poll(cx).map(Result::Ok),
312        }
313    }
314}
315
316/// An interval returned by [`interval`].
317///
318/// This type allows you to wait on a sequence of instants with a certain
319/// duration between each instant.
320#[must_use = "futures do nothing unless you `.await` or poll them"]
321#[derive(Debug)]
322pub struct Interval {
323    hnd: TimerHandle,
324    period: u32,
325}
326
327impl Interval {
328    /// Creates an interval with the specified period.
329    #[inline]
330    pub fn new(period: Millis) -> Interval {
331        Interval {
332            hnd: TimerHandle::new(u64::from(period.0)),
333            period: period.0,
334        }
335    }
336
337    #[inline]
338    /// Waits for the next interval tick.
339    pub async fn tick(&self) {
340        poll_fn(|cx| self.poll_tick(cx)).await;
341    }
342
343    #[inline]
344    /// Polls for the next interval tick.
345    pub fn poll_tick(&self, cx: &mut task::Context<'_>) -> Poll<()> {
346        if self.hnd.poll_elapsed(cx).is_ready() {
347            self.hnd.reset(u64::from(self.period));
348            Poll::Ready(())
349        } else {
350            Poll::Pending
351        }
352    }
353}
354
355impl crate::Stream for Interval {
356    type Item = ();
357
358    #[inline]
359    fn poll_next(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
360        self.poll_tick(cx).map(|()| Some(()))
361    }
362}
363
364#[cfg(test)]
365mod tests {
366    use futures_util::StreamExt;
367    use std::{future::poll_fn, rc::Rc, time};
368
369    use super::*;
370    use crate::future::lazy;
371
372    /// State Under Test: Two calls of `now()` return the same value if they are done within resolution interval.
373    ///
374    /// Expected Behavior: Two back-to-back calls of `now()` return the same value.
375    #[ntex::test]
376    async fn lowres_time_does_not_immediately_change() {
377        sleep(Millis(25)).await;
378
379        assert_eq!(now(), now());
380    }
381
382    /// State Under Test: `now()` updates returned value every ~1ms period.
383    ///
384    /// Expected Behavior: Two calls of `now()` made in subsequent resolution interval return different values
385    /// and second value is greater than the first one at least by a 1ms interval.
386    #[ntex::test]
387    async fn lowres_time_updates_after_resolution_interval() {
388        sleep(Millis(50)).await;
389
390        let first_time = now();
391
392        sleep(Millis(25)).await;
393
394        let second_time = now();
395        assert!(second_time - first_time >= time::Duration::from_millis(25));
396    }
397
398    /// State Under Test: Two calls of `system_time()` return the same value if they are done within 1ms interval.
399    ///
400    /// Expected Behavior: Two back-to-back calls of `now()` return the same value.
401    #[ntex::test]
402    async fn system_time_service_time_does_not_immediately_change() {
403        sleep(Seconds(1)).await;
404
405        assert_eq!(system_time(), system_time());
406        assert_eq!(system_time(), query_system_time());
407    }
408
409    /// State Under Test: `system_time()` updates returned value every 1ms period.
410    ///
411    /// Expected Behavior: Two calls of `system_time()` made in subsequent resolution interval return different values
412    /// and second value is greater than the first one at least by a resolution interval.
413    #[ntex::test]
414    async fn system_time_service_time_updates_after_resolution_interval() {
415        sleep(Millis(100)).await;
416
417        let wait_time = 300;
418
419        let first_time = system_time()
420            .duration_since(time::SystemTime::UNIX_EPOCH)
421            .unwrap();
422
423        sleep(Millis(wait_time)).await;
424
425        let second_time = system_time()
426            .duration_since(time::SystemTime::UNIX_EPOCH)
427            .unwrap();
428
429        assert!(
430            second_time.checked_sub(first_time).unwrap()
431                >= time::Duration::from_millis(u64::from(wait_time))
432        );
433    }
434
435    #[ntex::test]
436    async fn test_sleep_0() {
437        sleep(Seconds(1)).await;
438
439        let first_time = now();
440        sleep(Millis(0)).await;
441        let second_time = now();
442        assert!(second_time - first_time >= time::Duration::from_millis(1));
443
444        let first_time = now();
445        sleep(Millis(1)).await;
446        let second_time = now();
447        assert!(second_time - first_time >= time::Duration::from_millis(1));
448
449        let first_time = now();
450        let fut = sleep(Millis(10000));
451        assert!(!fut.is_elapsed());
452        fut.reset(Millis::ZERO);
453        fut.await;
454        let second_time = now();
455        assert!(second_time - first_time < time::Duration::from_millis(1));
456
457        let first_time = now();
458        let fut = Sleep {
459            hnd: TimerHandle::new(0),
460        };
461        assert!(fut.is_elapsed());
462        fut.await;
463        let second_time = now();
464        assert!(second_time - first_time < time::Duration::from_millis(1));
465
466        let first_time = now();
467        let fut = Rc::new(sleep(Millis(10_0000)));
468        let s = fut.clone();
469        ntex::rt::spawn(async move {
470            s.elapse();
471        });
472        poll_fn(|cx| fut.poll_elapsed(cx)).await;
473        assert!(fut.is_elapsed());
474        let second_time = now();
475        assert!(second_time - first_time < time::Duration::from_millis(1));
476    }
477
478    #[ntex::test]
479    async fn test_deadline() {
480        sleep(Seconds(1)).await;
481
482        let first_time = now();
483        let dl = deadline(Millis(1));
484        dl.await;
485        let second_time = now();
486        assert!(second_time - first_time >= time::Duration::from_millis(1));
487        assert!(timeout(Millis(100), deadline(Millis(0))).await.is_err());
488
489        let mut dl = deadline(Millis(1));
490        dl.reset(Millis::ZERO);
491        assert!(lazy(|cx| dl.poll_elapsed(cx)).await.is_pending());
492
493        let mut dl = deadline(Millis(1));
494        dl.reset(Millis(100));
495        let first_time = now();
496        dl.await;
497        let second_time = now();
498        assert!(second_time - first_time >= time::Duration::from_millis(100));
499
500        let mut dl = deadline(Millis(0));
501        assert!(dl.is_elapsed());
502        dl.reset(Millis(1));
503        assert!(lazy(|cx| dl.poll_elapsed(cx)).await.is_pending());
504
505        assert!(format!("{dl:?}").contains("Deadline"));
506    }
507
508    #[ntex::test]
509    async fn test_interval() {
510        let mut int = interval(Millis(250));
511
512        let time = time::Instant::now();
513        int.tick().await;
514        let elapsed = time.elapsed();
515        assert!(
516            elapsed > time::Duration::from_millis(200)
517                && elapsed < time::Duration::from_millis(450),
518            "elapsed: {elapsed:?}"
519        );
520
521        let time = time::Instant::now();
522        int.next().await;
523        let elapsed = time.elapsed();
524        assert!(
525            elapsed > time::Duration::from_millis(200)
526                && elapsed < time::Duration::from_millis(450),
527            "elapsed: {elapsed:?}"
528        );
529    }
530
531    #[ntex::test]
532    async fn test_interval_one_sec() {
533        let int = interval(Millis::ONE_SEC);
534
535        for _i in 0..3 {
536            let time = time::Instant::now();
537            int.tick().await;
538            let elapsed = time.elapsed();
539            assert!(
540                elapsed > time::Duration::from_secs(1)
541                    && elapsed < time::Duration::from_millis(1300),
542                "elapsed: {elapsed:?}"
543            );
544        }
545    }
546
547    #[ntex::test]
548    async fn test_timeout_checked() {
549        let result = timeout_checked(Millis(200), sleep(Millis(100))).await;
550        assert!(result.is_ok());
551
552        let result = timeout_checked(Millis(5), sleep(Millis(100))).await;
553        assert!(result.is_err());
554
555        let result = timeout_checked(Millis(0), sleep(Millis(100))).await;
556        assert!(result.is_ok());
557    }
558}
ntex_util/time/mod.rs

ntex_util/time/
mod.rs
