wta_reactor/
timers.rs

1use std::{
2    cmp::Reverse,
3    pin::Pin,
4    sync::{Mutex, MutexGuard},
5    task::{Context, Poll, Waker},
6    time::{Duration, Instant},
7};
8
9use futures::Future;
10
11use crate::context;
12
13#[derive(Default)]
14pub(crate) struct Queue(Mutex<Vec<(Instant, Waker)>>);
15impl Queue {
16    pub fn insert(&self, instant: Instant, task: Waker) {
17        let mut queue = self.0.lock().unwrap();
18        let index = match queue.binary_search_by_key(&Reverse(instant), |e| Reverse(e.0)) {
19            Ok(index) | Err(index) => index,
20        };
21        queue.insert(index, (instant, task));
22    }
23}
24
25impl<'a> IntoIterator for &'a Queue {
26    type Item = Waker;
27    type IntoIter = QueueIter<'a>;
28
29    fn into_iter(self) -> Self::IntoIter {
30        QueueIter(self.0.lock().unwrap(), Instant::now())
31    }
32}
33
34pub(crate) struct QueueIter<'a>(MutexGuard<'a, Vec<(Instant, Waker)>>, Instant);
35impl Iterator for QueueIter<'_> {
36    type Item = Waker;
37
38    fn next(&mut self) -> Option<Self::Item> {
39        let (time, task) = self.0.pop()?;
40        if time > self.1 {
41            self.0.push((time, task));
42            None
43        } else {
44            Some(task)
45        }
46    }
47}
48
49/// Future for sleeping fixed amounts of time.
50/// Does not block the thread
51pub struct Sleep {
52    instant: Instant,
53}
54
55impl Unpin for Sleep {}
56
57impl Future for Sleep {
58    type Output = ();
59
60    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
61        // if the future is not yet ready
62        if self.instant > Instant::now() {
63            context(|r| r.timers.insert(self.instant, cx.waker().clone()));
64            Poll::Pending
65        } else {
66            Poll::Ready(())
67        }
68    }
69}
70
71impl Sleep {
72    /// sleep until a specific point in time
73    #[must_use]
74    pub fn until(instant: Instant) -> Sleep {
75        Self { instant }
76    }
77    /// sleep for a specific duration of time
78    #[must_use]
79    pub fn duration(duration: Duration) -> Sleep {
80        Sleep::until(Instant::now() + duration)
81    }
82}
wta_reactor/timers.rs

wta_reactor/
timers.rs
