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

strop_core/worker/
effect.rs
