ripsync_core/
control.rs

1//! Cooperative pause and cancellation shared by planning, apply, and verification.
2
3use std::sync::atomic::{AtomicBool, Ordering};
4use std::sync::{Arc, Condvar, Mutex};
5
6use crate::{Error, Result};
7
8#[derive(Default)]
9struct State {
10    paused: AtomicBool,
11    cancelled: AtomicBool,
12    lock: Mutex<()>,
13    wake: Condvar,
14}
15
16/// Cloneable cooperative run control.
17#[derive(Clone, Default)]
18pub struct RunControl {
19    state: Arc<State>,
20}
21
22impl RunControl {
23    /// Pause before the next cooperative checkpoint.
24    pub fn pause(&self) {
25        self.state.paused.store(true, Ordering::Release);
26    }
27
28    /// Resume a paused run.
29    pub fn resume(&self) {
30        self.state.paused.store(false, Ordering::Release);
31        self.state.wake.notify_all();
32    }
33
34    /// Request graceful cancellation and wake paused workers.
35    pub fn cancel(&self) {
36        self.state.cancelled.store(true, Ordering::Release);
37        self.state.wake.notify_all();
38    }
39
40    /// Whether pause is currently requested.
41    #[must_use]
42    pub fn is_paused(&self) -> bool {
43        self.state.paused.load(Ordering::Acquire)
44    }
45
46    /// Whether cancellation is currently requested.
47    #[must_use]
48    pub fn is_cancelled(&self) -> bool {
49        self.state.cancelled.load(Ordering::Acquire)
50    }
51
52    /// Block while paused and return [`Error::Cancelled`] when cancelled.
53    ///
54    /// # Errors
55    ///
56    /// Returns [`Error::Cancelled`] after cancellation is requested.
57    pub fn checkpoint(&self) -> Result<()> {
58        if self.is_cancelled() {
59            return Err(Error::Cancelled);
60        }
61        if self.is_paused() {
62            let mut guard = self
63                .state
64                .lock
65                .lock()
66                .unwrap_or_else(std::sync::PoisonError::into_inner);
67            while self.is_paused() && !self.is_cancelled() {
68                guard = self
69                    .state
70                    .wake
71                    .wait(guard)
72                    .unwrap_or_else(std::sync::PoisonError::into_inner);
73            }
74        }
75        if self.is_cancelled() {
76            Err(Error::Cancelled)
77        } else {
78            Ok(())
79        }
80    }
81}
82
83#[cfg(test)]
84mod tests {
85    use super::RunControl;
86
87    #[test]
88    fn pause_resume_and_cancel() {
89        let control = RunControl::default();
90        control.pause();
91        assert!(control.is_paused());
92        control.resume();
93        assert!(control.checkpoint().is_ok());
94        control.cancel();
95        assert!(control.checkpoint().is_err());
96    }
97}
ripsync_core/control.rs

ripsync_core/
control.rs
