how_far/
adapters.rs

1//! Erased cancellation adapters; no copy of their machinery per inner type.
2use crate::{
3    Child, ChildPulse, Execution, Outcome, PhaseSpec, PlanError, Pulse, Report, SharedPulse, Stop,
4    StopReason, Total, stop_only::StopSource,
5};
6use alloc::{boxed::Box, sync::Arc, vec::Vec};
7
8/// Adds or replaces a pulse's checkpoint policy, including through descendants.
9///
10/// Borrowed policies need no allocation and accept a local `&dyn Stop`. They
11/// cannot outlive that borrow: [`Pulse::share`] returns [`PlanError::NotShareable`].
12/// Use an owned policy with [`Self::new`] or [`Self::replacing`] for spawned workers.
13/// Scoped workers can borrow this adapter directly.
14///
15/// Replacement bypasses the inner `check` entirely, including its callbacks.
16/// Reporting, planning, ownership and completion still reach the inner pulse.
17/// Combining is usually appropriate when adding a library-local cancellation source.
18pub struct WithStop<'a> {
19    inner: Inner<'a>,
20    stop: StopSource<'a>,
21    mode: Mode,
22}
23enum Inner<'a> {
24    Borrowed(&'a dyn Pulse),
25    Shared(SharedPulse),
26    Child(Child<'a>),
27}
28#[derive(Clone, Copy)]
29enum Mode {
30    Combine,
31    Replace,
32}
33impl<'a> WithStop<'a> {
34    /// Check the inner policy, then `stop`. Both remain attached to descendants.
35    pub fn new(inner: &'a dyn Pulse, stop: impl Stop + 'static) -> Self {
36        Self {
37            inner: Inner::Borrowed(inner),
38            stop: StopSource::Owned(Arc::new(stop)),
39            mode: Mode::Combine,
40        }
41    }
42    /// Combine policies without owning or allocating the additional stop source.
43    pub fn borrowed(inner: &'a dyn Pulse, stop: &'a dyn Stop) -> Self {
44        Self {
45            inner: Inner::Borrowed(inner),
46            stop: StopSource::Borrowed(stop),
47            mode: Mode::Combine,
48        }
49    }
50    /// Use only `stop` for checkpoints, bypassing the inner stop and callbacks.
51    pub fn replacing(inner: &'a dyn Pulse, stop: impl Stop + 'static) -> Self {
52        Self {
53            inner: Inner::Borrowed(inner),
54            stop: StopSource::Owned(Arc::new(stop)),
55            mode: Mode::Replace,
56        }
57    }
58    /// Replace checkpoints with a borrowed policy; see [`Self::replacing`].
59    pub fn replacing_borrowed(inner: &'a dyn Pulse, stop: &'a dyn Stop) -> Self {
60        Self {
61            inner: Inner::Borrowed(inner),
62            stop: StopSource::Borrowed(stop),
63            mode: Mode::Replace,
64        }
65    }
66    fn inner(&self) -> &dyn Pulse {
67        match &self.inner {
68            Inner::Borrowed(p) => *p,
69            Inner::Shared(p) => p.as_pulse(),
70            Inner::Child(p) => p.pulse(),
71        }
72    }
73}
74impl Stop for WithStop<'_> {
75    #[track_caller]
76    fn check(&self) -> Result<(), StopReason> {
77        if matches!(self.mode, Mode::Combine) {
78            self.inner().check()?;
79        }
80        self.stop.get().check()
81    }
82    fn may_stop(&self) -> bool {
83        (matches!(self.mode, Mode::Combine) && self.inner().may_stop())
84            || self.stop.get().may_stop()
85    }
86}
87impl Report for WithStop<'_> {
88    #[track_caller]
89    fn advance(&self, n: u64) {
90        self.inner().advance(n)
91    }
92    fn may_report(&self) -> bool {
93        self.inner().may_report()
94    }
95}
96impl Pulse for WithStop<'_> {
97    #[track_caller]
98    fn record_issue(&self, error: PlanError) {
99        self.inner().record_issue(error);
100    }
101    fn split(&self, e: Execution, parts: &[PhaseSpec<'_>]) -> Result<Vec<Child<'_>>, PlanError> {
102        let children = self.inner().split(e, parts)?;
103        Ok(children
104            .into_iter()
105            .map(|child| {
106                Child::new(WithStop {
107                    inner: Inner::Child(child),
108                    stop: self.stop.clone(),
109                    mode: self.mode,
110                })
111            })
112            .collect())
113    }
114    fn share(&self) -> Result<SharedPulse, PlanError> {
115        let stop = self.stop.owned()?;
116        Ok(SharedPulse::new(WithStop {
117            inner: Inner::Shared(self.inner().share()?),
118            stop: StopSource::Owned(stop),
119            mode: self.mode,
120        }))
121    }
122    fn start(&self) -> Result<(), PlanError> {
123        self.inner().start()
124    }
125    fn set_total(&self, total: Total) -> Result<(), PlanError> {
126        self.inner().set_total(total)
127    }
128}
129impl ChildPulse for WithStop<'_> {
130    fn complete_inferred(self: Box<Self>, outcome: Outcome) {
131        if let Inner::Child(child) = self.inner {
132            child.complete_inferred(outcome);
133        }
134    }
135    fn complete_as(self: Box<Self>, outcome: Outcome) {
136        if let Inner::Child(child) = self.inner {
137            crate::Complete::complete_as(child, outcome);
138        }
139    }
140    fn finish(self: Box<Self>, outcome: Outcome) -> Result<(), PlanError> {
141        match self.inner {
142            Inner::Child(child) => child.finish(outcome),
143            _ => Err(PlanError::Unsupported),
144        }
145    }
146}
how_far/adapters.rs

how_far/
adapters.rs
