ogeom_core/
progress.rs

1//! Progress reporting and cancellation for long operations.
2//!
3//! A caller who starts a tessellation, a boolean or an import may need to
4//! stop it (a user closed the dialog) or to show that it is alive. The
5//! kernel's answer is a [`Watch`]: install one around a call with
6//! [`watched`], hand its [`Canceller`] to whoever may pull the plug, and
7//! every long loop inside the kernel calls [`checkpoint`] at its own
8//! boundaries. A cancelled checkpoint returns
9//! [`OgeomError::Cancelled`], which unwinds as
10//! an ordinary error; no partial result pretends to be whole.
11//!
12//! The watch travels implicitly, by scope: operations keep their signatures,
13//! and code that never installs a watch pays one thread-local read per
14//! checkpoint. Worker threads a kernel operation spawns re-install the
15//! caller's watch through [`snapshot`]/[`with_snapshot`], so cancellation
16//! reaches into parallel stages too.
17//!
18//! Cancellation is cooperative and prompt rather than immediate: it lands at
19//! the next checkpoint, and checkpoints sit at stage and item boundaries,
20//! never inside an invariant-restoring section.
21
22use std::cell::RefCell;
23use std::sync::Arc;
24use std::sync::atomic::{AtomicBool, Ordering};
25
26use crate::{OgeomError, OgeomResult};
27
28/// A stage announcement: the name, and where the operation stands in it
29/// when the operation knows.
30#[derive(Debug, Clone, Copy)]
31pub struct Stage<'a> {
32    /// The stage's name, stable across a run: `"step: solid"`,
33    /// `"tessellate: faces"`.
34    pub name: &'a str,
35    /// `(done, total)` within this stage, when both are known: what a
36    /// determinate progress bar needs. `None` for a bare boundary.
37    pub progress: Option<(u64, u64)>,
38}
39
40/// A stage sink: hears each announcement as the operation reaches it.
41type Sink = Arc<dyn Fn(Stage<'_>) + Send + Sync>;
42
43/// What a watch carries: the flag, and an optional stage sink.
44#[derive(Clone)]
45struct State {
46    cancel: Arc<AtomicBool>,
47    sink: Option<Sink>,
48}
49
50impl core::fmt::Debug for State {
51    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
52        f.debug_struct("State")
53            .field("cancelled", &self.cancel.load(Ordering::Relaxed))
54            .field("has_sink", &self.sink.is_some())
55            .finish()
56    }
57}
58
59thread_local! {
60    static ACTIVE: RefCell<Option<State>> = const { RefCell::new(None) };
61}
62
63/// A scope's progress watch: cancellation flag plus an optional sink that
64/// receives stage names as the operation passes them.
65#[derive(Debug)]
66pub struct Watch {
67    state: State,
68}
69
70impl Watch {
71    /// A watch with no sink: cancellation only.
72    #[must_use]
73    pub fn new() -> Self {
74        Self {
75            state: State {
76                cancel: Arc::new(AtomicBool::new(false)),
77                sink: None,
78            },
79        }
80    }
81
82    /// A watch whose sink hears each stage name once as the operation
83    /// reaches it. The sink runs on whichever thread reaches the stage.
84    ///
85    /// The name-only convenience: a sink that also wants the counts behind
86    /// a determinate progress bar installs [`Watch::with_stage_sink`].
87    #[must_use]
88    pub fn with_sink(sink: impl Fn(&str) + Send + Sync + 'static) -> Self {
89        Self::with_stage_sink(move |stage: Stage<'_>| sink(stage.name))
90    }
91
92    /// A watch whose sink hears each full [`Stage`] announcement: the name,
93    /// and `(done, total)` where the operation states them. The sink runs on
94    /// whichever thread reaches the stage; a parallel stage's counts arrive
95    /// in completion order, each value once.
96    #[must_use]
97    pub fn with_stage_sink(sink: impl Fn(Stage<'_>) + Send + Sync + 'static) -> Self {
98        Self {
99            state: State {
100                cancel: Arc::new(AtomicBool::new(false)),
101                sink: Some(Arc::new(sink)),
102            },
103        }
104    }
105
106    /// The handle that cancels this watch, cloneable and sendable to
107    /// whatever owns the stop button.
108    #[must_use]
109    pub fn canceller(&self) -> Canceller {
110        Canceller {
111            cancel: Arc::clone(&self.state.cancel),
112        }
113    }
114}
115
116impl Default for Watch {
117    fn default() -> Self {
118        Self::new()
119    }
120}
121
122/// The stop button: cancel from any thread, any number of times.
123#[derive(Debug, Clone)]
124pub struct Canceller {
125    cancel: Arc<AtomicBool>,
126}
127
128impl Canceller {
129    /// Request cancellation. Takes effect at the operation's next
130    /// [`checkpoint`].
131    pub fn cancel(&self) {
132        self.cancel.store(true, Ordering::SeqCst);
133    }
134
135    /// Whether cancellation has been requested.
136    #[must_use]
137    pub fn is_cancelled(&self) -> bool {
138        self.cancel.load(Ordering::SeqCst)
139    }
140}
141
142/// Restores the previously active state when the scope ends, panics
143/// included.
144struct Scope {
145    previous: Option<State>,
146}
147
148impl Drop for Scope {
149    fn drop(&mut self) {
150        ACTIVE.with(|active| {
151            *active.borrow_mut() = self.previous.take();
152        });
153    }
154}
155
156/// Run `f` with `watch` active on this thread: every [`checkpoint`] inside
157/// answers to it. Scopes nest; the inner watch wins until it ends.
158pub fn watched<T>(watch: &Watch, f: impl FnOnce() -> T) -> T {
159    let previous = ACTIVE.with(|active| active.borrow_mut().replace(watch.state.clone()));
160    let _scope = Scope { previous };
161    f()
162}
163
164/// The point a long loop offers for cancellation. Free when no watch is
165/// installed.
166///
167/// # Errors
168///
169/// [`OgeomError::Cancelled`] if the active
170/// watch has been cancelled.
171pub fn checkpoint() -> OgeomResult<()> {
172    ACTIVE.with(|active| {
173        if let Some(state) = active.borrow().as_ref()
174            && state.cancel.load(Ordering::Relaxed)
175        {
176            return Err(OgeomError::Cancelled);
177        }
178        Ok(())
179    })
180}
181
182/// Announce a stage boundary to the active watch's sink, if there is one.
183pub fn stage(name: &str) {
184    announce(Stage {
185        name,
186        progress: None,
187    });
188}
189
190/// Announce a stage with its position: `done` of `total` items complete.
191/// What a determinate progress bar is built from; emitted by the operations
192/// that know both numbers: a reader over its solids, a tessellation over
193/// its faces.
194pub fn stage_at(name: &str, done: u64, total: u64) {
195    announce(Stage {
196        name,
197        progress: Some((done, total)),
198    });
199}
200
201fn announce(stage: Stage<'_>) {
202    ACTIVE.with(|active| {
203        if let Some(state) = active.borrow().as_ref()
204            && let Some(sink) = &state.sink
205        {
206            sink(stage);
207        }
208    });
209}
210
211/// The active watch, portable to a worker thread. `None` when unwatched.
212#[must_use]
213pub fn snapshot() -> Option<WatchSnapshot> {
214    ACTIVE.with(|active| active.borrow().clone().map(|state| WatchSnapshot { state }))
215}
216
217/// Run `f` under a snapshot taken on another thread: how a parallel stage
218/// keeps answering the caller's watch.
219pub fn with_snapshot<T>(snapshot: Option<&WatchSnapshot>, f: impl FnOnce() -> T) -> T {
220    match snapshot {
221        Some(snap) => {
222            let watch = Watch {
223                state: snap.state.clone(),
224            };
225            watched(&watch, f)
226        }
227        None => f(),
228    }
229}
230
231/// An opaque, cloneable capture of the active watch.
232#[derive(Debug, Clone)]
233pub struct WatchSnapshot {
234    state: State,
235}
236
237#[cfg(test)]
238#[allow(clippy::unwrap_used)]
239mod tests {
240    use super::*;
241
242    #[test]
243    fn counts_reach_a_stage_sink_and_names_still_reach_a_plain_one() {
244        use std::sync::Mutex;
245        type Heard = Vec<(String, Option<(u64, u64)>)>;
246        let heard: Arc<Mutex<Heard>> = Arc::new(Mutex::new(Vec::new()));
247        let record = Arc::clone(&heard);
248        let watch = Watch::with_stage_sink(move |stage: Stage<'_>| {
249            record
250                .lock()
251                .unwrap()
252                .push((stage.name.to_owned(), stage.progress));
253        });
254        watched(&watch, || {
255            stage("plain");
256            stage_at("counted", 2, 5);
257        });
258        assert_eq!(
259            *heard.lock().unwrap(),
260            vec![
261                ("plain".to_owned(), None),
262                ("counted".to_owned(), Some((2, 5))),
263            ]
264        );
265
266        // The name-only sink keeps working, counts and all announced.
267        let names: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
268        let record = Arc::clone(&names);
269        let watch =
270            Watch::with_sink(move |name: &str| record.lock().unwrap().push(name.to_owned()));
271        watched(&watch, || {
272            stage_at("counted", 1, 3);
273        });
274        assert_eq!(*names.lock().unwrap(), vec!["counted".to_owned()]);
275    }
276
277    #[test]
278    fn unwatched_checkpoints_are_free_and_fine() {
279        assert!(checkpoint().is_ok());
280        stage("nothing listens");
281    }
282
283    #[test]
284    fn a_cancelled_watch_stops_the_next_checkpoint() {
285        let watch = Watch::new();
286        let stop = watch.canceller();
287        let result: OgeomResult<()> = watched(&watch, || {
288            checkpoint()?;
289            stop.cancel();
290            checkpoint()?;
291            unreachable!("the second checkpoint must refuse");
292        });
293        assert!(matches!(result, Err(OgeomError::Cancelled)));
294        assert!(stop.is_cancelled());
295        // The scope is over: this thread is unwatched again.
296        assert!(checkpoint().is_ok());
297    }
298
299    #[test]
300    fn stages_reach_the_sink_and_scopes_nest() {
301        use std::sync::Mutex;
302        let heard: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
303        let record = Arc::clone(&heard);
304        let outer = Watch::with_sink(move |name| record.lock().unwrap().push(name.to_owned()));
305        let inner = Watch::new();
306        watched(&outer, || {
307            stage("first");
308            watched(&inner, || {
309                // The inner watch has no sink; its scope masks the outer.
310                stage("masked");
311            });
312            stage("second");
313        });
314        assert_eq!(*heard.lock().unwrap(), ["first", "second"]);
315    }
316
317    #[test]
318    fn snapshots_carry_the_watch_across_threads() {
319        let watch = Watch::new();
320        watch.canceller().cancel();
321        let snap = watched(&watch, snapshot);
322        let outcome: OgeomResult<()> =
323            std::thread::spawn(move || with_snapshot(snap.as_ref(), checkpoint))
324                .join()
325                .unwrap();
326        assert!(matches!(outcome, Err(OgeomError::Cancelled)));
327    }
328}
ogeom_core/progress.rs

ogeom_core/
progress.rs
