pfx_editor_shell/
worker.rs

1use std::fmt;
2use std::sync::Arc;
3use std::sync::mpsc::{self, Receiver, TryRecvError};
4
5pub type Wake = Arc<dyn Fn() + Send + Sync>;
6
7pub enum Poll<T> {
8    Idle,
9    Busy,
10    Done(T),
11    Lost,
12}
13
14struct Ring(Wake);
15
16impl Drop for Ring {
17    fn drop(&mut self) {
18        (self.0)();
19    }
20}
21
22pub struct Worker<T> {
23    wake: Wake,
24    pending: Option<Receiver<T>>,
25}
26
27impl<T> fmt::Debug for Worker<T> {
28    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
29        f.debug_struct("Worker")
30            .field("busy", &self.pending.is_some())
31            .finish_non_exhaustive()
32    }
33}
34
35impl<T: Send + 'static> Worker<T> {
36    pub fn new(wake: Wake) -> Worker<T> {
37        Worker {
38            wake,
39            pending: None,
40        }
41    }
42
43    pub fn busy(&self) -> bool {
44        self.pending.is_some()
45    }
46
47    pub fn start(&mut self, job: impl FnOnce() -> T + Send + 'static) {
48        let (send, receive) = mpsc::channel();
49        let ring = Ring(self.wake.clone());
50        std::thread::spawn(move || {
51            let ring = ring;
52            let send = send;
53            let _ = send.send(job());
54            drop(send);
55            drop(ring);
56        });
57        self.pending = Some(receive);
58    }
59
60    pub fn poll(&mut self) -> Poll<T> {
61        let Some(pending) = &self.pending else {
62            return Poll::Idle;
63        };
64        match pending.try_recv() {
65            Ok(value) => {
66                self.pending = None;
67                Poll::Done(value)
68            }
69            Err(TryRecvError::Empty) => Poll::Busy,
70            Err(TryRecvError::Disconnected) => {
71                self.pending = None;
72                Poll::Lost
73            }
74        }
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81    use std::time::Duration;
82
83    #[test]
84    fn a_job_runs_off_the_calling_thread_and_wakes_it() {
85        let (woke, wakes) = mpsc::channel();
86        let mut worker = Worker::new(Arc::new(move || {
87            let _ = woke.send(());
88        }));
89        assert!(matches!(worker.poll(), Poll::Idle));
90        let (open, gate) = mpsc::channel::<()>();
91        let caller = std::thread::current().id();
92        worker.start(move || {
93            let _ = gate.recv_timeout(Duration::from_secs(10));
94            std::thread::current().id() != caller
95        });
96        assert!(worker.busy());
97        assert!(matches!(worker.poll(), Poll::Busy));
98        open.send(()).unwrap();
99        wakes.recv_timeout(Duration::from_secs(10)).unwrap();
100        assert!(matches!(worker.poll(), Poll::Done(true)));
101        assert!(!worker.busy());
102    }
103
104    #[test]
105    fn a_job_that_panics_is_lost_and_still_wakes() {
106        let (woke, wakes) = mpsc::channel();
107        let mut worker: Worker<u32> = Worker::new(Arc::new(move || {
108            let _ = woke.send(());
109        }));
110        worker.start(|| panic!("the worker panicked on purpose"));
111        wakes.recv_timeout(Duration::from_secs(10)).unwrap();
112        assert!(matches!(worker.poll(), Poll::Lost));
113        assert!(!worker.busy());
114    }
115}
pfx_editor_shell/worker.rs

pfx_editor_shell/
worker.rs
