dope_runtime/process/failure/
execute.rs

1//! Worker execution and process-watch failures.
2use dope_core::{
3    dope_sys::thread::signal::Signal,
4    platform::{affinity, signal, thread},
5};
6use o3::alloc;
7
8use crate::process::failure::{bus, config};
9
10/// One process-control failure observed before worker return completes.
11#[derive(Debug)]
12pub enum Cause<S> {
13    Config(config::Error),
14    Allocation(alloc::Error),
15    Signal(signal::Acquire),
16    Entropy(dope_core::Errno),
17    WorkerBus {
18        cpu: affinity::Cpu,
19        source: bus::Create,
20    },
21    BusRead {
22        cpu: affinity::Cpu,
23        source: bus::Read,
24    },
25    BusState(bus::State),
26    WorkerFailed {
27        cpu: affinity::Cpu,
28    },
29    WorkerMissing {
30        cpu: affinity::Cpu,
31    },
32    WorkerSpawn {
33        cpu: affinity::Cpu,
34        source: thread::Error,
35    },
36    Poll(dope_core::Errno),
37    TerminationWait(<signal::Source<'static> as Signal>::Error),
38    WatchReady(S),
39    WatchPulse(S),
40    WatchDrain(S),
41}
42
43impl<S> Cause<S> {
44    pub(in crate::process) fn received(
45        result: Result<bus::Event, bus::Read>,
46        cpu: affinity::Cpu,
47    ) -> Result<bus::Event, Self> {
48        match result {
49            Ok(bus::Event::Failed) => Err(Self::WorkerFailed { cpu }),
50            Ok(event) => Ok(event),
51            Err(source) => Err(Self::BusRead { cpu, source }),
52        }
53    }
54}
55
56/// Exact bus failure observed while terminating one worker.
57#[derive(Debug)]
58pub enum Bus {
59    Write {
60        source: bus::Fault,
61        handoff: Option<bus::Read>,
62    },
63    Read(bus::Read),
64}
65
66/// Exact terminal failure produced by one returned worker.
67#[derive(Debug)]
68pub enum Worker<W> {
69    Failed {
70        cpu: affinity::Cpu,
71        handoff: Option<bus::Read>,
72    },
73    Affinity {
74        cpu: affinity::Cpu,
75        source: affinity::Error,
76        bus: bus::Report,
77        handoff: Option<bus::Read>,
78    },
79    Entry {
80        cpu: affinity::Cpu,
81        source: W,
82        bus: bus::Report,
83        handoff: Option<bus::Read>,
84    },
85    Bus {
86        cpu: affinity::Cpu,
87        source: Bus,
88    },
89}
90
91/// Every exact failure observed during one process run.
92#[derive(Debug)]
93pub enum Fault<'r, S, W> {
94    Control(Cause<S>),
95    Stop(dope_core::Errno),
96    Release(signal::Release<'r>),
97    Worker(Worker<W>),
98}
99
100/// Affine owner of all failures from one completed process run.
101#[must_use = "every process failure must be consumed"]
102#[derive(Debug)]
103pub struct Error<'r, S, W> {
104    control: Option<Cause<S>>,
105    missing: Option<affinity::Cpu>,
106    stop: Option<dope_core::Errno>,
107    release: Option<signal::Release<'r>>,
108    workers: Vec<Worker<W>>,
109}
110
111impl<'r, S, W> Error<'r, S, W> {
112    pub(in crate::process) fn failed(
113        control: Cause<S>,
114        missing: Option<affinity::Cpu>,
115        stop: Option<dope_core::Errno>,
116        workers: Vec<Worker<W>>,
117    ) -> Self {
118        Self {
119            control: Some(control),
120            missing,
121            stop,
122            release: None,
123            workers,
124        }
125    }
126
127    pub(in crate::process) fn finish(
128        missing: Option<affinity::Cpu>,
129        stop: Option<dope_core::Errno>,
130        workers: Vec<Worker<W>>,
131    ) -> Result<(), Self> {
132        if missing.is_none() && stop.is_none() && workers.is_empty() {
133            return Ok(());
134        }
135        Err(Self {
136            control: None,
137            missing,
138            stop,
139            release: None,
140            workers,
141        })
142    }
143
144    pub fn fold<R>(self, mut state: R, mut apply: impl FnMut(R, Fault<'r, S, W>) -> R) -> R {
145        if let Some(cause) = self.control {
146            state = apply(state, Fault::Control(cause));
147        }
148        if let Some(cpu) = self.missing {
149            state = apply(state, Fault::Control(Cause::WorkerMissing { cpu }));
150        }
151        if let Some(source) = self.stop {
152            state = apply(state, Fault::Stop(source));
153        }
154        let state = self
155            .workers
156            .into_iter()
157            .fold(state, |state, failure| apply(state, Fault::Worker(failure)));
158        match self.release {
159            Some(cause) => apply(state, Fault::Release(cause)),
160            None => state,
161        }
162    }
163}
164
165impl<'r, S, W> From<Cause<S>> for Error<'r, S, W> {
166    fn from(cause: Cause<S>) -> Self {
167        Self::failed(cause, None, None, Vec::new())
168    }
169}
170
171impl<S> From<alloc::Error> for Cause<S> {
172    fn from(error: alloc::Error) -> Self {
173        Self::Allocation(error)
174    }
175}
176
177impl<'r, S, W> From<signal::Error<'r, Error<'r, S, W>>> for Error<'r, S, W> {
178    fn from(error: signal::Error<'r, Error<'r, S, W>>) -> Self {
179        match error {
180            signal::Error::Acquire(error) => {
181                Self::failed(Cause::Signal(error), None, None, Vec::new())
182            }
183            signal::Error::Run(error) => error,
184            signal::Error::Release { run, cause } => {
185                let mut error = run.unwrap_or(Self {
186                    control: None,
187                    missing: None,
188                    stop: None,
189                    release: None,
190                    workers: Vec::new(),
191                });
192                error.release = Some(cause);
193                error
194            }
195        }
196    }
197}
dope_runtime/process/failure/execute.rs

dope_runtime/process/failure/
execute.rs
