subetha_cxc/
progress_task.rs

1//! `ProgressTask<R>` - distributed work with live cross-process
2//! progress visibility.
3//!
4//! Composes [`SharedAtomicU64`] (the progress
5//! counter), [`SharedAtomicU64`] (the total /
6//! denominator), [`SharedAtomicBool`] (the
7//! done flag), and [`SharedCell<R>`](crate::SharedCell) (the result
8//! payload). The work closure receives a [`ProgressReporter`] handle
9//! that increments the progress counter as it proceeds; any OTHER
10//! process or thread can call `fraction_complete` / `current_progress`
11//! / `is_done` / `read_result` at O(1) atomic cost without blocking
12//! or polling a result queue.
13//!
14//! # Why this exists
15//!
16//! Long-running jobs (ETL pipelines, batch processing, big builds)
17//! benefit from out-of-band progress visibility: a separate dashboard
18//! / CLI / supervisor wants to know "47% done, ETA 3min" without
19//! drilling into the worker's log file or waiting on a result ring.
20//! Naive approaches use a separate progress channel, a counter file
21//! that the worker overwrites, or a stat-on-tempfile heuristic. With
22//! ProgressTask, the counter is one atomic load away in shared
23//! memory; updates are sub-nanosecond.
24//!
25//! # Four files per task
26//!
27//! - `<base>.progress.bin` - SharedAtomicU64, monotonically advancing
28//! - `<base>.total.bin`    - SharedAtomicU64, the denominator
29//! - `<base>.done.bin`     - SharedAtomicBool, set true on completion
30//! - `<base>.result.bin`   - `SharedCell<R>`, written once at completion
31//!
32//! Pass the BASE PATH (without extension) to `create` / `open`; the
33//! wrapper appends the extensions.
34//!
35//! # Composition with the scheduler
36//!
37//! ProgressTask is INDEPENDENT of `BackgroundScheduler`; it works
38//! standalone via `run` / `spawn`. To integrate with the scheduler,
39//! a Pass closure can `ProgressTask::open(base)` to obtain a handle
40//! and call `begin(total)` to get a reporter. The worker side only
41//! needs the base path; the observer side only needs `open` + the
42//! read APIs.
43
44use std::path::{Path, PathBuf};
45use std::sync::atomic::Ordering;
46use std::sync::Arc;
47use std::thread::{self, JoinHandle};
48
49use crate::shared_atomic::{SharedAtomicBool, SharedAtomicError, SharedAtomicU64};
50use crate::shared_cell::{SharedCell, SharedCellError};
51
52#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub enum ProgressTaskError {
54    Atomic(SharedAtomicError),
55    Cell(SharedCellError),
56}
57
58impl From<SharedAtomicError> for ProgressTaskError {
59    fn from(e: SharedAtomicError) -> Self { Self::Atomic(e) }
60}
61impl From<SharedCellError> for ProgressTaskError {
62    fn from(e: SharedCellError) -> Self { Self::Cell(e) }
63}
64
65fn progress_path(base: &Path) -> PathBuf {
66    let mut p = base.to_path_buf();
67    let stem = p.file_name().unwrap().to_string_lossy().to_string();
68    p.set_file_name(format!("{stem}.progress.bin"));
69    p
70}
71fn total_path(base: &Path) -> PathBuf {
72    let mut p = base.to_path_buf();
73    let stem = p.file_name().unwrap().to_string_lossy().to_string();
74    p.set_file_name(format!("{stem}.total.bin"));
75    p
76}
77fn done_path(base: &Path) -> PathBuf {
78    let mut p = base.to_path_buf();
79    let stem = p.file_name().unwrap().to_string_lossy().to_string();
80    p.set_file_name(format!("{stem}.done.bin"));
81    p
82}
83fn result_path(base: &Path) -> PathBuf {
84    let mut p = base.to_path_buf();
85    let stem = p.file_name().unwrap().to_string_lossy().to_string();
86    p.set_file_name(format!("{stem}.result.bin"));
87    p
88}
89
90/// Cross-process reporter handle passed to the work closure. Each
91/// call to `advance` is a single atomic fetch_add (Relaxed ordering;
92/// progress is observational, not a synchronization point).
93pub struct ProgressReporter {
94    progress: Arc<SharedAtomicU64>,
95}
96
97impl ProgressReporter {
98    /// Add `n` to the progress counter. Returns the previous value.
99    /// Use `Relaxed` because progress is observational; the `done`
100    /// flag carries the happens-before edge.
101    #[inline]
102    pub fn advance(&self, n: u64) -> u64 {
103        self.progress.fetch_add(n, Ordering::Relaxed)
104    }
105
106    /// Replace the progress counter with `v`. Useful when the work
107    /// reports absolute progress (e.g., bytes processed) rather than
108    /// per-step deltas.
109    #[inline]
110    pub fn set(&self, v: u64) {
111        self.progress.store(v, Ordering::Relaxed);
112    }
113
114    /// Read the current progress counter.
115    #[inline]
116    pub fn current(&self) -> u64 {
117        self.progress.load(Ordering::Relaxed)
118    }
119}
120
121pub struct ProgressTask<R: Copy + 'static> {
122    progress: Arc<SharedAtomicU64>,
123    total: Arc<SharedAtomicU64>,
124    done: Arc<SharedAtomicBool>,
125    result: Arc<SharedCell<R>>,
126    header_sidecar: subetha_core::HandshakeHeader,
127    ring_sidecar: Box<subetha_core::ObservationRing>,
128}
129
130impl<R: Copy + Send + Sync + 'static> subetha_sidecar::AdaptiveInstance for ProgressTask<R> {
131    fn header(&self) -> &subetha_core::HandshakeHeader { &self.header_sidecar }
132    fn ring(&self) -> &subetha_core::ObservationRing { &self.ring_sidecar }
133    fn make_policy(&self) -> Box<dyn subetha_sidecar::Policy> {
134        Box::new(subetha_sidecar::NoMigrationPolicy)
135    }
136}
137
138impl<R: Copy + 'static> ProgressTask<R> {
139    /// Create a new ProgressTask at `base_path`. Allocates four MMF
140    /// files; initialises progress=0, total=0, done=false, result=
141    /// `initial_result`.
142    pub fn create(
143        base_path: impl AsRef<Path>,
144        initial_result: R,
145    ) -> Result<Self, ProgressTaskError> {
146        let base = base_path.as_ref();
147        let progress = Arc::new(SharedAtomicU64::create(progress_path(base), 0)?);
148        let total = Arc::new(SharedAtomicU64::create(total_path(base), 0)?);
149        let done = Arc::new(SharedAtomicBool::create(done_path(base), false)?);
150        let result = Arc::new(SharedCell::<R>::create(result_path(base))?);
151        result.set(initial_result);
152        Ok(Self {
153            progress, total, done, result,
154            header_sidecar: subetha_core::HandshakeHeader::new(),
155            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
156        })
157    }
158
159    /// Open an existing ProgressTask at `base_path`. All four MMF
160    /// files must exist.
161    pub fn open(base_path: impl AsRef<Path>) -> Result<Self, ProgressTaskError> {
162        let base = base_path.as_ref();
163        let progress = Arc::new(SharedAtomicU64::open(progress_path(base))?);
164        let total = Arc::new(SharedAtomicU64::open(total_path(base))?);
165        let done = Arc::new(SharedAtomicBool::open(done_path(base))?);
166        let result = Arc::new(SharedCell::<R>::open(result_path(base))?);
167        Ok(Self {
168            progress, total, done, result,
169            header_sidecar: subetha_core::HandshakeHeader::new(),
170            ring_sidecar: Box::new(subetha_core::ObservationRing::new()),
171        })
172    }
173
174    /// Begin a new task run with the given `total` denominator.
175    /// Resets progress=0, done=false, and publishes `total`. Returns
176    /// a ProgressReporter for the work closure to advance.
177    pub fn begin(&self, total: u64) -> ProgressReporter {
178        self.progress.store(0, Ordering::Relaxed);
179        self.done.store(false, Ordering::Release);
180        self.total.store(total, Ordering::Release);
181        self.ring_sidecar
182            .push_op(crate::sidecar_ops::progress::OP_ADVANCE, 0);
183        ProgressReporter { progress: self.progress.clone() }
184    }
185
186    /// Mark the task complete and publish the final result. Other
187    /// processes observing `is_done` will see true after this call;
188    /// `read_result` returns the published value.
189    pub fn complete(&self, result: R) {
190        self.result.set(result);
191        // Release ordering: pair with Acquire load of `done` to give
192        // observers happens-before visibility of the result cell.
193        self.done.store(true, Ordering::Release);
194        self.ring_sidecar
195            .push_op(crate::sidecar_ops::progress::OP_COMPLETE, 0);
196    }
197
198    /// Current fraction in [0.0, 1.0]. Returns 0.0 when total is 0
199    /// (no task has begun); clamped at 1.0 to avoid >100% display.
200    pub fn fraction_complete(&self) -> f64 {
201        let p = self.progress.load(Ordering::Relaxed);
202        let t = self.total.load(Ordering::Acquire);
203        if t == 0 { return 0.0; }
204        (p as f64 / t as f64).min(1.0)
205    }
206
207    /// Current progress counter value.
208    #[inline]
209    pub fn current_progress(&self) -> u64 {
210        self.progress.load(Ordering::Relaxed)
211    }
212
213    /// Total denominator for the current run (0 when no run is active).
214    #[inline]
215    pub fn total(&self) -> u64 {
216        self.total.load(Ordering::Acquire)
217    }
218
219    /// True when `complete` has been called for the current run.
220    #[inline]
221    pub fn is_done(&self) -> bool {
222        self.done.load(Ordering::Acquire)
223    }
224
225    /// Read the most-recently-published result. Returns None when
226    /// `is_done` is false (a result MAY still be there from a prior
227    /// completed run, but the current run is not yet finished).
228    pub fn read_result(&self) -> Option<R> {
229        let r = if self.is_done() {
230            Some(self.result.get())
231        } else {
232            None
233        };
234        self.ring_sidecar.push_op(
235            crate::sidecar_ops::progress::OP_READ,
236            if r.is_none() { 2 } else { 0 },
237        );
238        r
239    }
240
241    /// Read the result cell unconditionally. Useful for reading the
242    /// initial value before any run, or for sampling a stale value.
243    #[inline]
244    pub fn peek_result(&self) -> R {
245        self.result.get()
246    }
247
248    /// Run the work closure synchronously on the calling thread.
249    /// Calls `begin(total)` to set up the reporter, runs the closure,
250    /// publishes the result via `complete`, and returns the result.
251    pub fn run<F>(&self, total: u64, work: F) -> R
252    where F: FnOnce(&ProgressReporter) -> R
253    {
254        let reporter = self.begin(total);
255        let r = work(&reporter);
256        self.complete(r);
257        r
258    }
259
260    /// Spawn the work closure on a background thread. Returns a
261    /// JoinHandle so the caller can wait if desired; the task's
262    /// completion is also visible via `is_done`.
263    pub fn spawn<F>(self: &Arc<Self>, total: u64, work: F) -> JoinHandle<()>
264    where
265        F: FnOnce(&ProgressReporter) -> R + Send + 'static,
266        R: Send + Sync,
267    {
268        let me = self.clone();
269        thread::spawn(move || {
270            me.run(total, work);
271        })
272    }
273
274    /// Sync all four files to disk.
275    pub fn flush(&self) -> Result<(), ProgressTaskError> {
276        self.progress.flush()?;
277        self.total.flush()?;
278        self.done.flush()?;
279        self.result.flush()?;
280        Ok(())
281    }
282
283    /// Non-blocking flush of all four files. Delegates to each inner
284    /// primitive's flush_async.
285    /// Note: Windows is only partially async (sync to page cache,
286    /// not to disk).
287    pub fn flush_async(&self) -> Result<(), ProgressTaskError> {
288        self.progress.flush_async()?;
289        self.total.flush_async()?;
290        self.done.flush_async()?;
291        self.result.flush_async()?;
292        Ok(())
293    }
294}
295
296#[cfg(test)]
297mod tests {
298    use super::*;
299    use std::time::Duration;
300
301    fn tmp_base(name: &str) -> PathBuf {
302        let mut p = std::env::temp_dir();
303        let pid = std::process::id();
304        p.push(format!("subetha-progress-{name}-{pid}"));
305        p
306    }
307
308    fn cleanup(base: &Path) {
309        std::fs::remove_file(progress_path(base)).ok();
310        std::fs::remove_file(total_path(base)).ok();
311        std::fs::remove_file(done_path(base)).ok();
312        std::fs::remove_file(result_path(base)).ok();
313    }
314
315    #[test]
316    fn create_initial_state_is_zero() {
317        let base = tmp_base("init");
318        let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
319        assert_eq!(t.current_progress(), 0);
320        assert_eq!(t.total(), 0);
321        assert!(!t.is_done());
322        assert_eq!(t.fraction_complete(), 0.0);
323        assert_eq!(t.read_result(), None);
324        cleanup(&base);
325    }
326
327    #[test]
328    fn run_advances_progress_and_publishes_result() {
329        let base = tmp_base("run");
330        let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
331        let r = t.run(100, |reporter| {
332            for _ in 0..100 {
333                reporter.advance(1);
334            }
335            42
336        });
337        assert_eq!(r, 42);
338        assert!(t.is_done());
339        assert_eq!(t.read_result(), Some(42));
340        assert_eq!(t.fraction_complete(), 1.0);
341        cleanup(&base);
342    }
343
344    #[test]
345    fn observer_sees_monotonic_progress_advance() {
346        use std::sync::atomic::{AtomicBool, Ordering};
347        let base = tmp_base("observe");
348        let t: Arc<ProgressTask<u64>>
349            = Arc::new(ProgressTask::create(&base, 0).unwrap());
350        let go = Arc::new(AtomicBool::new(false));
351
352        let t_w = t.clone();
353        let go_w = go.clone();
354        let worker = thread::spawn(move || {
355            t_w.run(50, |reporter| {
356                go_w.store(true, Ordering::Release);
357                for _ in 0..50 {
358                    reporter.advance(1);
359                    thread::sleep(Duration::from_micros(100));
360                }
361                7777
362            });
363        });
364        // Wait until worker has begun (avoid race on the initial value).
365        while !go.load(Ordering::Acquire) { thread::yield_now(); }
366
367        let mut observed = Vec::new();
368        for _ in 0..20 {
369            observed.push(t.current_progress());
370            thread::sleep(Duration::from_micros(150));
371        }
372        worker.join().unwrap();
373        // Monotonic non-decreasing.
374        for w in observed.windows(2) {
375            assert!(w[0] <= w[1], "progress went backwards: {observed:?}");
376        }
377        assert!(t.is_done());
378        assert_eq!(t.read_result(), Some(7777));
379        cleanup(&base);
380    }
381
382    #[test]
383    fn cross_handle_observer_sees_worker_state() {
384        let base = tmp_base("cross-handle");
385        let worker_h: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
386        let observer_h: ProgressTask<u64> = ProgressTask::open(&base).unwrap();
387        worker_h.run(10, |r| { r.advance(10); 999 });
388        assert!(observer_h.is_done());
389        assert_eq!(observer_h.read_result(), Some(999));
390        assert_eq!(observer_h.fraction_complete(), 1.0);
391        cleanup(&base);
392    }
393
394    #[test]
395    fn fraction_clamps_when_progress_overshoots_total() {
396        let base = tmp_base("overshoot");
397        let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
398        let r = t.begin(10);
399        r.advance(15);
400        assert_eq!(t.fraction_complete(), 1.0);
401        cleanup(&base);
402    }
403
404    #[test]
405    fn read_result_returns_none_before_done() {
406        let base = tmp_base("not-done");
407        let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
408        let _r = t.begin(100);
409        assert!(!t.is_done());
410        assert_eq!(t.read_result(), None);
411        // peek bypasses the done check.
412        assert_eq!(t.peek_result(), 0);
413        cleanup(&base);
414    }
415
416    #[test]
417    fn second_run_resets_progress_and_done() {
418        let base = tmp_base("two-runs");
419        let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
420        t.run(5, |r| { r.advance(5); 100 });
421        assert!(t.is_done());
422        assert_eq!(t.current_progress(), 5);
423
424        // Begin a new run; done resets, progress resets, total updates.
425        let _r = t.begin(10);
426        assert!(!t.is_done());
427        assert_eq!(t.current_progress(), 0);
428        assert_eq!(t.total(), 10);
429        cleanup(&base);
430    }
431
432    #[test]
433    fn spawn_background_completes_eventually() {
434        let base = tmp_base("spawn");
435        let t: Arc<ProgressTask<u64>>
436            = Arc::new(ProgressTask::create(&base, 0).unwrap());
437        let h = t.spawn(20, |r| {
438            for _ in 0..20 {
439                r.advance(1);
440                thread::sleep(Duration::from_micros(100));
441            }
442            314
443        });
444        h.join().unwrap();
445        assert!(t.is_done());
446        assert_eq!(t.read_result(), Some(314));
447        cleanup(&base);
448    }
449
450    #[test]
451    fn concurrent_observers_all_see_consistent_completion() {
452        let base = tmp_base("multi-observer");
453        let t: Arc<ProgressTask<u64>>
454            = Arc::new(ProgressTask::create(&base, 0).unwrap());
455        let n_observers = 4;
456
457        let t_w = t.clone();
458        let worker = thread::spawn(move || {
459            t_w.run(100, |r| {
460                for _ in 0..100 {
461                    r.advance(1);
462                    thread::sleep(Duration::from_micros(50));
463                }
464                5555
465            });
466        });
467
468        let mut handles = vec![];
469        for _ in 0..n_observers {
470            let t = t.clone();
471            handles.push(thread::spawn(move || {
472                let mut last = 0u64;
473                while !t.is_done() {
474                    let cur = t.current_progress();
475                    assert!(cur >= last, "observer saw progress regress");
476                    last = cur;
477                    thread::sleep(Duration::from_micros(75));
478                }
479                t.read_result()
480            }));
481        }
482        worker.join().unwrap();
483        for h in handles {
484            assert_eq!(h.join().unwrap(), Some(5555));
485        }
486        cleanup(&base);
487    }
488
489    #[test]
490    fn struct_result_round_trip() {
491        #[derive(Clone, Copy, Debug, PartialEq)]
492        #[repr(C)]
493        struct Stats { processed: u64, errors: u32, skipped: u32 }
494        let base = tmp_base("struct");
495        let t: ProgressTask<Stats> = ProgressTask::create(
496            &base, Stats { processed: 0, errors: 0, skipped: 0 },
497        ).unwrap();
498        let r = t.run(50, |reporter| {
499            for _ in 0..50 { reporter.advance(1); }
500            Stats { processed: 50, errors: 2, skipped: 1 }
501        });
502        assert_eq!(r, Stats { processed: 50, errors: 2, skipped: 1 });
503        assert_eq!(t.read_result(), Some(r));
504        cleanup(&base);
505    }
506
507    #[test]
508    fn disk_persistence_completed_task_survives_reopen() {
509        let base = tmp_base("disk");
510        {
511            let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
512            t.run(8, |r| { r.advance(8); 8888 });
513            t.flush().unwrap();
514        }
515        let t2: ProgressTask<u64> = ProgressTask::open(&base).unwrap();
516        assert!(t2.is_done());
517        assert_eq!(t2.read_result(), Some(8888));
518        assert_eq!(t2.current_progress(), 8);
519        assert_eq!(t2.total(), 8);
520        cleanup(&base);
521    }
522
523    #[test]
524    fn reporter_set_replaces_progress() {
525        let base = tmp_base("set");
526        let t: ProgressTask<u64> = ProgressTask::create(&base, 0).unwrap();
527        let r = t.begin(1000);
528        r.advance(100);
529        assert_eq!(t.current_progress(), 100);
530        r.set(500);  // jump
531        assert_eq!(t.current_progress(), 500);
532        assert_eq!(t.fraction_complete(), 0.5);
533        cleanup(&base);
534    }
535}
subetha_cxc/progress_task.rs

subetha_cxc/
progress_task.rs
