over_there/utils/
delay.rs

1use std::sync::{
2    atomic::{AtomicBool, Ordering},
3    Arc,
4};
5use std::thread::{self, JoinHandle};
6use std::time::{Duration, Instant};
7
8/// Represents a delayed execution of a function
9pub struct Delay {
10    should_cancel: Arc<AtomicBool>,
11    thread_handle: JoinHandle<()>,
12}
13
14impl Delay {
15    /// Spawns a new thread that will invoke the provided function after the
16    /// given timeout has been exceeded. There is no guarantee that the
17    /// function will be executed exactly on the given time, only that it will
18    /// be executed no earlier than until the specified duration has elapsed
19    pub fn spawn<F, T>(timeout: Duration, f: F) -> Self
20    where
21        F: FnOnce() -> T + Send + 'static,
22    {
23        let should_cancel = Arc::new(AtomicBool::new(false));
24        let should_cancel_2 = Arc::clone(&should_cancel);
25
26        let start_time = Instant::now();
27        let thread_handle = thread::spawn(move || {
28            let mut timeout_remaining = timeout;
29            loop {
30                thread::park_timeout(timeout_remaining);
31                let elapsed = start_time.elapsed();
32                if elapsed >= timeout || should_cancel_2.load(Ordering::Acquire)
33                {
34                    break;
35                }
36                timeout_remaining = timeout - elapsed;
37            }
38
39            if !should_cancel_2.load(Ordering::Acquire) {
40                f();
41            }
42        });
43
44        Self {
45            should_cancel,
46            thread_handle,
47        }
48    }
49
50    /// Cancels the delayed execution, if it has not yet occurred
51    pub fn cancel(&self) {
52        self.should_cancel.store(true, Ordering::Release);
53        self.thread_handle.thread().unpark();
54    }
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60    use std::sync::Mutex;
61
62    #[test]
63    fn delay_should_occur_on_or_after_timeout() {
64        let delay_duration = Duration::from_millis(10);
65        let before = Instant::now();
66
67        let after = Arc::new(Mutex::new(Instant::now()));
68        let after_delay = Arc::clone(&after);
69        let _delay = Delay::spawn(delay_duration, move || {
70            *after_delay.lock().unwrap() = Instant::now();
71        });
72
73        // Wait twice as long to ensure the delay happens
74        thread::sleep(delay_duration * 2);
75
76        let elapsed = after
77            .lock()
78            .unwrap()
79            .checked_duration_since(before)
80            .unwrap();
81        assert!(
82            elapsed >= delay_duration,
83            "Delay did not occur after {:?}",
84            delay_duration
85        );
86    }
87
88    #[test]
89    fn delay_should_occur_even_if_instance_is_dropped() {
90        let delay_duration = Duration::from_millis(10);
91        let did_occur = Arc::new(AtomicBool::new(false));
92        let did_occur_2 = Arc::clone(&did_occur);
93        let delay = Delay::spawn(delay_duration, move || {
94            did_occur_2.store(true, Ordering::Release);
95        });
96
97        // Immediately drop the delay instance
98        drop(delay);
99
100        // Wait twice as long to ensure the delay happens
101        thread::sleep(delay_duration * 2);
102
103        assert!(
104            did_occur.load(Ordering::Acquire),
105            "Delayed call did not occur",
106        );
107    }
108
109    #[test]
110    fn delay_should_not_occur_if_cancelled() {
111        let delay_duration = Duration::from_millis(10);
112        let did_occur = Arc::new(AtomicBool::new(false));
113        let did_occur_2 = Arc::clone(&did_occur);
114        let delay = Delay::spawn(delay_duration, move || {
115            did_occur_2.store(true, Ordering::Release);
116        });
117
118        // Cancel immediately
119        delay.cancel();
120
121        // Wait twice as long to ensure the delay happens
122        thread::sleep(delay_duration * 2);
123
124        assert!(
125            !did_occur.load(Ordering::Acquire),
126            "Delay occurred unexpectedly",
127        );
128    }
129
130    #[test]
131    fn delay_cancel_should_do_nothing_if_delay_already_occurred() {
132        let delay_duration = Duration::from_millis(10);
133        let did_occur = Arc::new(AtomicBool::new(false));
134        let did_occur_2 = Arc::clone(&did_occur);
135        let delay = Delay::spawn(delay_duration, move || {
136            did_occur_2.store(true, Ordering::Release);
137        });
138
139        // Wait twice as long to ensure the delay happens
140        thread::sleep(delay_duration * 2);
141
142        // Cancel later, after the delay should have happened
143        delay.cancel();
144
145        assert!(
146            did_occur.load(Ordering::Acquire),
147            "Delayed call did not occur",
148        );
149    }
150}
over_there/utils/delay.rs

over_there/utils/
delay.rs
