strop_core/worker/
effect.rs

1//! Admitted effects finish with their observed result, not an eager cancel fiction.
2use super::{CancelHandle, CancelToken, Cancellation, Failure, FailureKind, Outcome};
3use parking_lot::Mutex;
4use std::sync::Arc;
5
6type Emit<T> = Box<dyn FnOnce(Outcome<T>) + Send>;
7struct State<T> {
8    emit: Option<Emit<T>>,
9    outcome: Option<Outcome<T>>,
10    cancelling: bool,
11    cleanup_error: Option<Failure>,
12}
13impl<T> State<T> {
14    fn ready(&mut self) -> Option<(Emit<T>, Outcome<T>)> {
15        use super::recovery_policy::ready_to_deliver;
16        if !ready_to_deliver(self.outcome.is_some(), self.cancelling) {
17            return None;
18        }
19        let emit = self.emit.take()?;
20        let outcome = self.outcome.take()?;
21        let outcome = match (outcome, self.cleanup_error.take()) {
22            (Outcome::Success(value), Some(failure)) => Outcome::Failed {
23                failure,
24                partial: Some(value),
25            },
26            (
27                Outcome::Failed {
28                    mut failure,
29                    partial,
30                },
31                Some(cleanup),
32            ) => {
33                failure
34                    .message
35                    .push_str(&format!("; cancellation cleanup: {}", cleanup.message));
36                Outcome::Failed { failure, partial }
37            }
38            (Outcome::Cancelled(_), Some(failure)) => Outcome::Failed {
39                failure,
40                partial: None,
41            },
42            (outcome, None) => outcome,
43        };
44        Some((emit, outcome))
45    }
46}
47struct Shared<T> {
48    token: CancelToken,
49    state: Mutex<State<T>>,
50}
51impl<T> Shared<T> {
52    fn complete(&self, outcome: Outcome<T>) {
53        let delivery = {
54            let mut state = self.state.lock();
55            state.outcome = Some(outcome);
56            state.ready()
57        };
58        if let Some((emit, outcome)) = delivery {
59            emit(outcome);
60        }
61    }
62    fn request_cancel(&self) {
63        {
64            let mut state = self.state.lock();
65            if state.emit.is_none() || state.cancelling || self.token.is_cancelled() {
66                return;
67            }
68            state.cancelling = true;
69        }
70        let failure = self.token.cancel_resource().err();
71        let delivery = {
72            let mut state = self.state.lock();
73            state.cancelling = false;
74            state.cleanup_error = failure;
75            state.ready()
76        };
77        if let Some((emit, outcome)) = delivery {
78            emit(outcome);
79        }
80    }
81}
82
83/// Cancellation requests stop resources promptly but do not replace a mutation
84/// receipt. The work closure always runs and must check its token before effects,
85/// allowing it to return exact per-item cancellations even before native launch.
86/// Cleanup failure preserves a successful observed value in `Failed.partial`.
87pub fn spawn_effect<T: Send + 'static>(
88    name: &'static str,
89    emit: impl FnOnce(Outcome<T>) + Send + 'static,
90    work: impl FnOnce(CancelToken) -> Outcome<T> + Send + 'static,
91) -> CancelHandle {
92    let shared = Arc::new(Shared {
93        token: CancelToken(Arc::new(Cancellation::default())),
94        state: Mutex::new(State {
95            emit: Some(Box::new(emit)),
96            outcome: None,
97            cancelling: false,
98            cleanup_error: None,
99        }),
100    });
101    let cancellation = shared.clone();
102    let handle = CancelHandle {
103        cancel: Some(Box::new(move |_| cancellation.request_cancel())),
104    };
105    let worker = shared.clone();
106    let started = std::thread::Builder::new()
107        .name(name.into())
108        .spawn(move || {
109            let outcome = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
110                work(worker.token.clone())
111            })) {
112                Ok(outcome) => outcome,
113                Err(_) => match worker.token.cancel_resource() {
114                    Ok(()) => Outcome::failed(FailureKind::Panic, "effect worker panicked"),
115                    Err(failure) => Outcome::Failed {
116                        failure,
117                        partial: None,
118                    },
119                },
120            };
121            worker.token.clear_cancel_resource();
122            worker.complete(outcome);
123        });
124    if let Err(error) = started {
125        shared.complete(Outcome::failed(FailureKind::ThreadStart, error.to_string()));
126    }
127    handle
128}
129
130#[cfg(test)]
131mod tests;
strop_core/worker/effect.rs

strop_core/worker/
effect.rs
