gam_runtime/
loop_progress.rs

1//! Low-overhead progress ticker for long parallel loops.
2//!
3//! `LoopProgress::tick` advances a shared counter and lets exactly one
4//! worker emit after each wall-clock interval. Callers own the log message
5//! so units and totals stay local to the loop.
6
7use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
8use std::time::Instant;
9
10pub const DEFAULT_LOOP_PROGRESS_INTERVAL_SECS: u64 = 25;
11
12fn elapsed_nanos(elapsed: std::time::Duration) -> u64 {
13    u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX)
14}
15
16pub struct LoopProgress {
17    started: Instant,
18    last_emit_nanos: AtomicU64,
19    interval_nanos: u64,
20    progress: AtomicUsize,
21}
22
23impl LoopProgress {
24    pub fn new(interval_secs: u64) -> Self {
25        Self {
26            started: Instant::now(),
27            last_emit_nanos: AtomicU64::new(0),
28            interval_nanos: interval_secs.saturating_mul(1_000_000_000),
29            progress: AtomicUsize::new(0),
30        }
31    }
32
33    pub fn default_interval() -> Self {
34        Self::new(DEFAULT_LOOP_PROGRESS_INTERVAL_SECS)
35    }
36
37    /// Advance the progress counter by `delta` and, if at least
38    /// `interval` of wall time has passed since the last claimed print,
39    /// invoke `emit(progress, elapsed_secs)` exactly once across all
40    /// threads. The closure typically issues a `log::info!`.
41    pub fn tick(&self, delta: usize, emit: impl FnOnce(usize, f64)) {
42        let progress = self
43            .progress
44            .fetch_add(delta, Ordering::Relaxed)
45            .saturating_add(delta);
46        let elapsed = elapsed_nanos(self.started.elapsed());
47        let last = self.last_emit_nanos.load(Ordering::Relaxed);
48        if elapsed < last.saturating_add(self.interval_nanos) {
49            return;
50        }
51        if self
52            .last_emit_nanos
53            .compare_exchange(last, elapsed, Ordering::Relaxed, Ordering::Relaxed)
54            .is_ok()
55        {
56            emit(progress, elapsed as f64 / 1.0e9);
57        }
58    }
59}
60
61#[cfg(test)]
62mod tests {
63    use super::*;
64    use std::sync::atomic::{AtomicBool, AtomicUsize};
65
66    #[test]
67    fn default_interval_constant_matches_expectation() {
68        assert_eq!(DEFAULT_LOOP_PROGRESS_INTERVAL_SECS, 25);
69    }
70
71    #[test]
72    fn new_with_zero_interval_emits_on_first_tick() {
73        let lp = LoopProgress::new(0);
74        let called = AtomicBool::new(false);
75        lp.tick(1, |_, _| {
76            called.store(true, Ordering::Relaxed);
77        });
78        assert!(
79            called.load(Ordering::Relaxed),
80            "emit should be called with zero interval"
81        );
82    }
83
84    #[test]
85    fn tick_accumulates_progress_across_calls() {
86        let lp = LoopProgress::new(0);
87        let last_seen = AtomicUsize::new(0);
88        lp.tick(5, |progress, _| {
89            last_seen.store(progress, Ordering::Relaxed);
90        });
91        assert_eq!(last_seen.load(Ordering::Relaxed), 5);
92    }
93
94    #[test]
95    fn tick_with_large_interval_does_not_emit_on_first_call() {
96        // With a 1-hour interval the first tick will have elapsed < interval,
97        // so emit should NOT be called (elapsed ≥ 0 but < 3600 seconds).
98        // Use an intermediate small value: 3600 seconds is definitely not elapsed
99        // in a unit test.
100        let lp = LoopProgress::new(3600);
101        let called = AtomicBool::new(false);
102        lp.tick(1, |_, _| {
103            called.store(true, Ordering::Relaxed);
104        });
105        // The first tick starts with last=0; elapsed is a small positive number;
106        // 3_600_000_000_000 ns >> any realistic elapsed, so emit is skipped.
107        assert!(
108            !called.load(Ordering::Relaxed),
109            "emit should not fire with 1-hour interval"
110        );
111    }
112
113    #[test]
114    fn tick_delta_zero_still_works() {
115        let lp = LoopProgress::new(0);
116        let seen = AtomicUsize::new(usize::MAX);
117        lp.tick(0, |progress, _| {
118            seen.store(progress, Ordering::Relaxed);
119        });
120        // A zero-delta tick must not panic and leaves the counter at 0; the
121        // zero interval still lets the single emit fire with progress 0.
122        assert_eq!(
123            seen.load(Ordering::Relaxed),
124            0,
125            "zero-delta tick must emit progress 0"
126        );
127    }
128
129    #[test]
130    fn elapsed_nanoseconds_saturate_instead_of_truncating() {
131        assert_eq!(elapsed_nanos(std::time::Duration::MAX), u64::MAX);
132    }
133}
gam_runtime/loop_progress.rs

gam_runtime/
loop_progress.rs
