radiate_core/domain/sync/
control.rs

1use std::sync::{
2    Arc, Condvar, Mutex,
3    atomic::{AtomicBool, Ordering},
4};
5
6#[derive(Debug, Default)]
7struct State {
8    paused: bool,
9    permits: usize,
10}
11
12#[derive(Clone, Default)]
13pub struct ThreadSync {
14    stop_flag: Arc<AtomicBool>,
15    inner: Arc<(Mutex<State>, Condvar)>,
16}
17
18impl ThreadSync {
19    pub fn new() -> Self {
20        Self {
21            stop_flag: Arc::new(AtomicBool::new(false)),
22            inner: Arc::new((
23                Mutex::new(State {
24                    paused: false,
25                    permits: 0,
26                }),
27                Condvar::new(),
28            )),
29        }
30    }
31
32    pub fn pair() -> (Self, Self) {
33        let ctl = Self::new();
34        (ctl.clone(), ctl)
35    }
36
37    #[inline]
38    pub fn stop(&self) {
39        self.stop_flag.store(true, Ordering::SeqCst);
40        // wake anything blocked
41        self.set_paused(true);
42    }
43
44    #[inline]
45    pub fn is_stopped(&self) -> bool {
46        self.stop_flag.load(Ordering::Relaxed)
47    }
48
49    #[inline]
50    pub fn stop_flag(&self) -> Arc<AtomicBool> {
51        self.stop_flag.clone()
52    }
53
54    #[inline]
55    pub fn set_paused(&self, paused: bool) {
56        let (lock, cv) = &*self.inner;
57        let mut st = lock.lock().unwrap();
58        st.paused = paused;
59        if !paused {
60            st.permits = 0; // permits irrelevant when running
61        }
62        cv.notify_all();
63    }
64
65    #[inline]
66    pub fn toggle_pause(&self) -> bool {
67        let (lock, cv) = &*self.inner;
68        let mut st = lock.lock().unwrap();
69        st.paused = !st.paused;
70        if !st.paused {
71            st.permits = 0;
72        }
73        let now = st.paused;
74        cv.notify_all();
75        now
76    }
77
78    #[inline]
79    pub fn step_once(&self) {
80        self.step_n(1);
81    }
82
83    #[inline]
84    pub fn step_n(&self, n: usize) {
85        let (lock, cv) = &*self.inner;
86        let mut st = lock.lock().unwrap();
87        st.paused = true;
88        st.permits += n;
89        cv.notify_all();
90    }
91
92    #[inline]
93    pub fn wait(&self) {
94        let (lock, cv) = &*self.inner;
95        let mut st = lock.lock().unwrap();
96
97        while !self.stop_flag.load(Ordering::Relaxed) {
98            if !st.paused {
99                return;
100            }
101
102            if st.permits > 0 {
103                st.permits -= 1;
104                return;
105            }
106
107            st = cv.wait(st).unwrap();
108        }
109    }
110
111    #[inline]
112    pub fn is_paused(&self) -> bool {
113        let (lock, _) = &*self.inner;
114        lock.lock().unwrap().paused
115    }
116}
117
118#[cfg(test)]
119mod diag_tests {
120    use super::*;
121    use std::sync::atomic::AtomicUsize;
122    use std::time::Duration;
123
124    #[test]
125    fn step_n_blocks_after_permits_exhausted() {
126        let control = ThreadSync::new();
127        control.step_n(10);
128
129        let count = Arc::new(AtomicUsize::new(0));
130        let count2 = Arc::clone(&count);
131        let control2 = control.clone();
132
133        let handle = std::thread::spawn(move || {
134            for _ in 0..15 {
135                control2.wait();
136                count2.fetch_add(1, Ordering::SeqCst);
137            }
138        });
139
140        std::thread::sleep(Duration::from_millis(300));
141        let progressed = count.load(Ordering::SeqCst);
142        println!("progressed before stop: {progressed}");
143        control.stop();
144        handle.join().unwrap();
145        let after_stop = count.load(Ordering::SeqCst);
146        println!("progressed after stop: {after_stop}");
147
148        assert_eq!(
149            progressed, 10,
150            "expected exactly 10 waits to return before blocking"
151        );
152    }
153}
radiate_core/domain/sync/control.rs

radiate_core/domain/sync/
control.rs
