how_far_along/
callback.rs

1//! Checkpoint callbacks over the same accounting state as PulseTree.
2use crate::{
3    Child, Execution, NodeId, Observer, Outcome, Phase, PhaseSpec, PlanError, Pulse, Report,
4    SharedPulse, Snapshot, Stop, StopReason, Summary, Total,
5};
6use alloc::{boxed::Box, sync::Arc, vec::Vec};
7use core::{
8    fmt,
9    sync::atomic::{AtomicU8, Ordering},
10};
11
12/// A checkpoint's lazy observation. No tree walk occurs unless requested.
13pub struct Checkpoint<'a> {
14    root: &'a Observer,
15    phase: &'a Observer,
16}
17impl Checkpoint<'_> {
18    /// Stable identity of the checking phase.
19    pub fn phase_id(&self) -> NodeId {
20        self.phase.id()
21    }
22    /// Human-readable label; IDs disambiguate duplicate names.
23    pub fn phase_name(&self) -> &str {
24        self.phase.name()
25    }
26    /// Allocation-free whole-job observation. Walks the tree.
27    pub fn summary(&self) -> Summary {
28        self.root.summary()
29    }
30    /// Whole-job observation without waiting for a metadata lock.
31    pub fn try_summary(&self) -> Option<Summary> {
32        self.root.try_summary()
33    }
34    /// Copy the whole tree, allocating names and children.
35    pub fn snapshot(&self) -> Snapshot {
36        self.root.snapshot()
37    }
38    /// Copy the tree only if metadata is immediately available.
39    pub fn try_snapshot(&self) -> Option<Snapshot> {
40        self.root.try_snapshot()
41    }
42}
43type Callback = dyn Fn(&Checkpoint<'_>) -> Result<(), StopReason> + Send + Sync;
44pub(crate) struct Dispatcher {
45    callback: Box<Callback>,
46    root: Observer,
47    stopped: AtomicU8,
48}
49impl Dispatcher {
50    pub(crate) fn check(&self, phase: &Observer) -> Result<(), StopReason> {
51        let code = self.stopped.load(Ordering::Acquire);
52        if code != 0 {
53            return Err(decode(code));
54        }
55        if self.root.is_finished() || phase.is_finished() {
56            return Ok(());
57        }
58        let result = (self.callback)(&Checkpoint {
59            root: &self.root,
60            phase,
61        });
62        if let Err(reason) = result {
63            let code = match reason {
64                StopReason::TimedOut => 2,
65                _ => 1,
66            };
67            let _ = self
68                .stopped
69                .compare_exchange(0, code, Ordering::AcqRel, Ordering::Acquire);
70        }
71        // A concurrent callback may have stopped while ours was running.
72        match self.stopped.load(Ordering::Acquire) {
73            0 => Ok(()),
74            code => Err(decode(code)),
75        }
76    }
77}
78impl crate::pulse::Policy for Arc<Dispatcher> {
79    #[track_caller]
80    fn check(&self, phase: &Observer) -> Result<(), StopReason> {
81        Dispatcher::check(self, phase)
82    }
83    fn may_stop(&self) -> bool {
84        true
85    }
86}
87
88fn decode(code: u8) -> StopReason {
89    if code == 2 {
90        StopReason::TimedOut
91    } else {
92        StopReason::Cancelled
93    }
94}
95
96/// A tracked root with a callback at each cancellation checkpoint.
97/// Reports, phase finishes, and Drop never invoke the callback. `step` invokes
98/// it once; call `Paced::finish` to flush and check a final partial batch.
99/// Callbacks may overlap on different workers; no internal lock is held.
100/// The first stop reason wins. Already-running callbacks may complete.
101/// For thread-affine UI captures use `poll::LocalPoller` with an Observer.
102#[must_use = "finish the root after its work joins; dropping it unfinished records Abandoned"]
103pub struct FnPulse {
104    tree: crate::pulse::TreePulse<Arc<Dispatcher>>,
105}
106impl FnPulse {
107    /// Start an unknown-total root with a checkpoint callback.
108    pub fn new(
109        name: &str,
110        callback: impl Fn(&Checkpoint<'_>) -> Result<(), StopReason> + Send + Sync + 'static,
111    ) -> Self {
112        Self::with_phase(Phase::new(name, Total::Unknown), callback)
113    }
114    /// Attach a callback to a configured root, including a count-only operation.
115    pub fn with_phase(
116        phase: Phase,
117        callback: impl Fn(&Checkpoint<'_>) -> Result<(), StopReason> + Send + Sync + 'static,
118    ) -> Self {
119        let root = phase.observer();
120        let dispatch = Arc::new(Dispatcher {
121            root,
122            callback: Box::new(callback),
123            stopped: AtomicU8::new(0),
124        });
125        Self {
126            tree: crate::pulse::TreePulse::new(phase, dispatch),
127        }
128    }
129    /// Observe progress without dispatching the callback.
130    pub fn observer(&self) -> Observer {
131        self.tree.observer()
132    }
133    /// Publish the final result. Read the observer for the final display update.
134    pub fn finish(self, outcome: Outcome) -> Result<(), PlanError> {
135        self.tree.finish(outcome)
136    }
137}
138impl fmt::Debug for FnPulse {
139    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
140        f.debug_struct("FnPulse")
141            .field("root", &self.tree.observer().id())
142            .finish()
143    }
144}
145
146impl how_far::Complete for FnPulse {
147    fn complete_as(self, outcome: Outcome) {
148        self.tree.complete_as(outcome);
149    }
150}
151pulse_view!([] FnPulse, tree);
how_far_along/callback.rs

how_far_along/
callback.rs
