radiate-core 1.3.1

use std::sync::{
    Arc, Condvar, Mutex,
    atomic::{AtomicBool, Ordering},
};

#[derive(Debug, Default)]
struct State {
    paused: bool,
    permits: usize,
}

#[derive(Clone, Default)]
pub struct ThreadSync {
    stop_flag: Arc<AtomicBool>,
    inner: Arc<(Mutex<State>, Condvar)>,
}

impl ThreadSync {
    pub fn new() -> Self {
        Self {
            stop_flag: Arc::new(AtomicBool::new(false)),
            inner: Arc::new((
                Mutex::new(State {
                    paused: false,
                    permits: 0,
                }),
                Condvar::new(),
            )),
        }
    }

    pub fn pair() -> (Self, Self) {
        let ctl = Self::new();
        (ctl.clone(), ctl)
    }

    #[inline]
    pub fn stop(&self) {
        self.stop_flag.store(true, Ordering::SeqCst);
        // wake anything blocked
        self.set_paused(true);
    }

    #[inline]
    pub fn is_stopped(&self) -> bool {
        self.stop_flag.load(Ordering::Relaxed)
    }

    #[inline]
    pub fn stop_flag(&self) -> Arc<AtomicBool> {
        self.stop_flag.clone()
    }

    #[inline]
    pub fn set_paused(&self, paused: bool) {
        let (lock, cv) = &*self.inner;
        let mut st = lock.lock().unwrap();
        st.paused = paused;
        if !paused {
            st.permits = 0; // permits irrelevant when running
        }
        cv.notify_all();
    }

    #[inline]
    pub fn toggle_pause(&self) -> bool {
        let (lock, cv) = &*self.inner;
        let mut st = lock.lock().unwrap();
        st.paused = !st.paused;
        if !st.paused {
            st.permits = 0;
        }
        let now = st.paused;
        cv.notify_all();
        now
    }

    #[inline]
    pub fn step_once(&self) {
        self.step_n(1);
    }

    #[inline]
    pub fn step_n(&self, n: usize) {
        let (lock, cv) = &*self.inner;
        let mut st = lock.lock().unwrap();
        st.paused = true;
        st.permits += n;
        cv.notify_all();
    }

    #[inline]
    pub fn wait(&self) {
        let (lock, cv) = &*self.inner;
        let mut st = lock.lock().unwrap();

        while !self.stop_flag.load(Ordering::Relaxed) {
            if !st.paused {
                return;
            }

            if st.permits > 0 {
                st.permits -= 1;
                return;
            }

            st = cv.wait(st).unwrap();
        }
    }

    #[inline]
    pub fn is_paused(&self) -> bool {
        let (lock, _) = &*self.inner;
        lock.lock().unwrap().paused
    }
}

#[cfg(test)]
mod diag_tests {
    use super::*;
    use std::sync::atomic::AtomicUsize;
    use std::time::Duration;

    #[test]
    fn step_n_blocks_after_permits_exhausted() {
        let control = ThreadSync::new();
        control.step_n(10);

        let count = Arc::new(AtomicUsize::new(0));
        let count2 = Arc::clone(&count);
        let control2 = control.clone();

        let handle = std::thread::spawn(move || {
            for _ in 0..15 {
                control2.wait();
                count2.fetch_add(1, Ordering::SeqCst);
            }
        });

        std::thread::sleep(Duration::from_millis(300));
        let progressed = count.load(Ordering::SeqCst);
        println!("progressed before stop: {progressed}");
        control.stop();
        handle.join().unwrap();
        let after_stop = count.load(Ordering::SeqCst);
        println!("progressed after stop: {after_stop}");

        assert_eq!(
            progressed, 10,
            "expected exactly 10 waits to return before blocking"
        );
    }
}