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, channel, 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: channel::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        terminal: Option<bus::Fault>,
62        handoff: Option<bus::Read>,
63    },
64    Read(bus::Read),
65}
66
67/// Exact terminal failure produced by one returned worker.
68#[derive(Debug)]
69pub enum Worker<W> {
70    Affinity {
71        cpu: affinity::Cpu,
72        source: affinity::Error,
73        bus: bus::Report,
74        terminal: Option<bus::Fault>,
75        handoff: Option<bus::Read>,
76    },
77    Entry {
78        cpu: affinity::Cpu,
79        source: W,
80        bus: bus::Report,
81        terminal: Option<bus::Fault>,
82        handoff: Option<bus::Read>,
83    },
84    Bus {
85        cpu: affinity::Cpu,
86        source: Bus,
87    },
88}
89
90/// Every exact failure observed during one process run.
91#[derive(Debug)]
92pub enum Fault<'r, S, W> {
93    Control(Cause<S>),
94    Stop(dope_core::Errno),
95    Release(signal::Release<'r>),
96    Worker(Worker<W>),
97}
98
99/// Affine owner of all failures from one completed process run.
100#[must_use = "every process failure must be consumed"]
101#[derive(Debug)]
102pub struct Error<'r, S, W> {
103    control: Option<Cause<S>>,
104    missing: Option<affinity::Cpu>,
105    stop: Option<dope_core::Errno>,
106    release: Option<signal::Release<'r>>,
107    workers: Vec<Worker<W>>,
108}
109
110impl<'r, S, W> Error<'r, S, W> {
111    pub(in crate::process) fn failed(
112        control: Cause<S>,
113        missing: Option<affinity::Cpu>,
114        stop: Option<dope_core::Errno>,
115        workers: Vec<Worker<W>>,
116    ) -> Self {
117        Self {
118            control: Some(control),
119            missing,
120            stop,
121            release: None,
122            workers,
123        }
124    }
125
126    pub(in crate::process) fn finish(
127        missing: Option<affinity::Cpu>,
128        stop: Option<dope_core::Errno>,
129        workers: Vec<Worker<W>>,
130    ) -> Result<(), Self> {
131        if missing.is_none() && stop.is_none() && workers.is_empty() {
132            return Ok(());
133        }
134        Err(Self {
135            control: None,
136            missing,
137            stop,
138            release: None,
139            workers,
140        })
141    }
142
143    pub fn fold<R>(self, mut state: R, mut apply: impl FnMut(R, Fault<'r, S, W>) -> R) -> R {
144        if let Some(cause) = self.control {
145            state = apply(state, Fault::Control(cause));
146        }
147        if let Some(cpu) = self.missing {
148            state = apply(state, Fault::Control(Cause::WorkerMissing { cpu }));
149        }
150        if let Some(source) = self.stop {
151            state = apply(state, Fault::Stop(source));
152        }
153        let state = self
154            .workers
155            .into_iter()
156            .fold(state, |state, failure| apply(state, Fault::Worker(failure)));
157        match self.release {
158            Some(cause) => apply(state, Fault::Release(cause)),
159            None => state,
160        }
161    }
162}
163
164impl<'r, S, W> From<Cause<S>> for Error<'r, S, W> {
165    fn from(cause: Cause<S>) -> Self {
166        Self::failed(cause, None, None, Vec::new())
167    }
168}
169
170impl<S> From<alloc::Error> for Cause<S> {
171    fn from(error: alloc::Error) -> Self {
172        Self::Allocation(error)
173    }
174}
175
176impl<'r, S, W> From<signal::Error<'r, Error<'r, S, W>>> for Error<'r, S, W> {
177    fn from(error: signal::Error<'r, Error<'r, S, W>>) -> Self {
178        match error {
179            signal::Error::Acquire(error) => {
180                Self::failed(Cause::Signal(error), None, None, Vec::new())
181            }
182            signal::Error::Run(error) => error,
183            signal::Error::Release { run, cause } => {
184                let mut error = run.unwrap_or(Self {
185                    control: None,
186                    missing: None,
187                    stop: None,
188                    release: None,
189                    workers: Vec::new(),
190                });
191                error.release = Some(cause);
192                error
193            }
194        }
195    }
196}
dope_runtime/process/failure/execute.rs

dope_runtime/process/failure/
execute.rs
