dope_runtime/process/lifecycle/
mod.rs

1use std::{
2    cell, marker, mem,
3    os::fd::{self, AsFd as _},
4    time,
5};
6
7use dope_core::{
8    Backend, Errno,
9    dope_sys::io::{
10        fd::Fd as _,
11        poll::{self, Poll as _},
12        socket::{How, Socket},
13    },
14    driver::{life::creation, settings},
15};
16use o3::base::marker::ThreadBound;
17
18use crate::{
19    executor::{self, create},
20    process::{self, failure::bus},
21    shutdown,
22};
23
24mod native;
25pub use native::{Entry, Join, Start, Task, scope};
26
27#[derive(Debug)]
28pub enum WaitError {
29    Poll(Errno),
30    Read(bus::Read),
31    Event(bus::Event),
32    Timeout {
33        timeout: time::Duration,
34        elapsed: time::Duration,
35    },
36}
37
38#[derive(Clone, Copy, Debug)]
39pub struct Control<'a> {
40    fd: fd::BorrowedFd<'a>,
41    _thread: ThreadBound,
42}
43
44pub(crate) fn pair() -> Result<(Main, fd::OwnedFd), bus::Create> {
45    let (left, right) = Backend::pair()?;
46    Ok((Main(left, marker::PhantomData), right))
47}
48
49#[derive(Debug)]
50pub struct Main(fd::OwnedFd, marker::PhantomData<cell::Cell<()>>);
51
52pub struct Worker {
53    socket: fd::OwnedFd,
54    failure: cell::Cell<Option<bus::Fault>>,
55    phase: cell::Cell<Phase>,
56}
57
58#[derive(Debug)]
59pub enum Outcome<T, E> {
60    Done(T),
61    Failed {
62        output: Result<T, E>,
63        report: bus::Report,
64    },
65}
66
67#[derive(Clone, Copy)]
68enum Phase {
69    New,
70    Installed,
71    Failed,
72}
73
74#[derive(Clone, Copy)]
75#[repr(transparent)]
76pub(in crate::process) struct Tx<'a>(&'a Worker, marker::PhantomData<fn(&'a ()) -> &'a ()>);
77
78/// Selects the worker event sequence from its `Plain` or `Gated` entry type.
79pub(in crate::process) trait Mode {
80    /// Whether the worker publishes application ready and drained events.
81    const GATED: bool;
82}
83
84/// Retains whether shutdown was requested before application readiness.
85#[derive(Clone, Copy)]
86pub(in crate::process) enum Readiness {
87    Required,
88    Cancelled,
89}
90
91impl Main {
92    pub fn try_pair() -> Result<(Self, Worker), bus::Create> {
93        let (main, right) = pair()?;
94        Ok((
95            main,
96            Worker {
97                socket: right,
98                failure: cell::Cell::new(None),
99                phase: cell::Cell::new(Phase::New),
100            },
101        ))
102    }
103
104    pub(in crate::process) fn recv(&self) -> Result<bus::Event, bus::Read> {
105        let mut byte = [0];
106        loop {
107            match Backend::recv::<0, Errno>(self.0.as_fd(), &mut byte) {
108                Ok(bytes) => {
109                    return match bytes.first() {
110                        Some(value) => bus::Event::decode(*value).map_err(bus::Read::Event),
111                        None => Err(bus::Read::Closed),
112                    };
113                }
114                Err(error) if error.is_intr() => {}
115                Err(error) => return Err(bus::Read::Syscall(error)),
116            }
117        }
118    }
119
120    pub(in crate::process) fn handoff<M: Mode>(&self, readiness: Readiness) -> Option<bus::Read> {
121        use bus::Event;
122
123        let mut failure = None;
124        let mut previous = None;
125        loop {
126            match self.recv() {
127                Err(bus::Read::Closed) => return failure,
128                Ok(event) => {
129                    if failure.is_none() {
130                        let valid = match (previous, event) {
131                            (None, _) if M::GATED => true,
132                            (None, Event::Installed | Event::Done | Event::Failed) => true,
133                            (None, Event::Ready | Event::Drain) => false,
134                            (Some(Event::Installed), Event::Ready) => M::GATED,
135                            (Some(Event::Installed), Event::Done) => {
136                                !M::GATED || matches!(readiness, Readiness::Cancelled)
137                            }
138                            (Some(Event::Installed), Event::Failed) => true,
139                            (Some(Event::Installed), Event::Installed | Event::Drain) => false,
140                            (Some(Event::Ready), Event::Drain | Event::Failed) => M::GATED,
141                            (Some(Event::Ready), Event::Installed | Event::Ready | Event::Done) => {
142                                false
143                            }
144                            (Some(Event::Drain), Event::Done | Event::Failed) => M::GATED,
145                            (
146                                Some(Event::Drain),
147                                Event::Installed | Event::Ready | Event::Drain,
148                            ) => false,
149                            (Some(Event::Done | Event::Failed), _) => false,
150                        };
151                        if !valid {
152                            failure = Some(bus::Read::Sequence(bus::Sequence {
153                                gated: M::GATED,
154                                previous,
155                                event,
156                            }));
157                        }
158                    }
159                    previous = Some(event);
160                }
161                Err(source) => {
162                    let terminal = matches!(&source, bus::Read::Syscall(_));
163                    if failure.is_none() {
164                        failure = Some(source);
165                    }
166                    if terminal {
167                        return failure;
168                    }
169                }
170            }
171        }
172    }
173
174    pub(crate) fn duplicate(&self) -> Result<Self, Errno> {
175        Backend::duplicate(self.0.as_fd()).map(|fd| Self(fd, marker::PhantomData))
176    }
177
178    pub(crate) fn request(&self) -> Result<(), Errno> {
179        shutdown(self.0.as_fd()).map_or(Ok(()), Err)
180    }
181
182    pub fn stop(self) -> Result<Self, Errno> {
183        self.request()?;
184        Ok(self)
185    }
186
187    pub fn wait_ready(&self, timeout: time::Duration) -> Result<(), WaitError> {
188        wait(self.0.as_fd(), time::Instant::now(), timeout)?;
189        match self.recv().map_err(WaitError::Read)? {
190            bus::Event::Installed => Ok(()),
191            event => Err(WaitError::Event(event)),
192        }
193    }
194
195    pub fn wait_done(&self, timeout: time::Duration) -> Result<(), WaitError> {
196        wait(self.0.as_fd(), time::Instant::now(), timeout)?;
197        match self.recv().map_err(WaitError::Read)? {
198            bus::Event::Done | bus::Event::Failed => Ok(()),
199            event => Err(WaitError::Event(event)),
200        }
201    }
202
203    pub fn run<R>(
204        self,
205        timeout: time::Duration,
206        run: impl FnOnce() -> R,
207    ) -> (Result<R, WaitError>, Result<Self, Errno>) {
208        let output = self.wait_ready(timeout).map(|()| run());
209        (output, self.stop())
210    }
211}
212
213impl fd::AsFd for Main {
214    fn as_fd(&self) -> fd::BorrowedFd<'_> {
215        self.0.as_fd()
216    }
217}
218
219impl executor::Startup for &Main {
220    fn installed(&self) -> Result<(), executor::InstallError> {
221        Ok(())
222    }
223}
224
225/// Observes endpoint readability without consuming its lifecycle event.
226impl executor::Build for &Main {
227    type Shutdown = Self;
228
229    fn try_build(
230        self,
231        attempt: &mut creation::Attempt,
232        config: settings::Config,
233    ) -> Result<executor::Executor<(), Self::Shutdown>, create::Error<'_>> {
234        executor::Executor::new(attempt, config)?
235            .shutdown(self)
236            .map_err(create::Error::from)
237    }
238}
239
240fn wait(
241    fd: fd::BorrowedFd<'_>,
242    started: time::Instant,
243    timeout: time::Duration,
244) -> Result<(), WaitError> {
245    loop {
246        let left = timeout.saturating_sub(started.elapsed());
247        let millis = left.as_millis() + u128::from(!left.subsec_nanos().is_multiple_of(1_000_000));
248        let mut entries = [poll::Entry::readable(fd)];
249        match Backend::wait::<_, Errno>(&mut entries, millis.min(i32::MAX as u128) as i32) {
250            Ok(events) if events.count() != 0 => return Ok(()),
251            Ok(_) => {}
252            Err(error) if error.is_intr() => {}
253            Err(error) => return Err(WaitError::Poll(error)),
254        }
255        let elapsed = started.elapsed();
256        if elapsed >= timeout {
257            return Err(WaitError::Timeout { timeout, elapsed });
258        }
259    }
260}
261
262impl<'a> Control<'a> {
263    pub(in crate::process) const fn new(fd: fd::BorrowedFd<'a>) -> Self {
264        Self {
265            fd,
266            _thread: ThreadBound::NEW,
267        }
268    }
269
270    pub fn stopped(&self) -> Result<bool, Errno> {
271        match Backend::recv_now::<Errno>(self.fd, &mut [0]) {
272            Ok(_) => Ok(true),
273            Err(error) if error.is_block() => Ok(false),
274            Err(error) if error.is_intr() => {
275                let mut entries = [poll::Entry::readable(self.fd)];
276                match Backend::wait::<_, Errno>(&mut entries, 0) {
277                    Ok(events) => Ok(events.count() != 0),
278                    Err(error) if error.is_intr() => Ok(false),
279                    Err(error) => Err(error),
280                }
281            }
282            Err(error) => Err(error),
283        }
284    }
285
286    pub fn wait_stop_until(&self, deadline: time::Instant) -> Result<(), WaitError> {
287        let started = time::Instant::now();
288        wait(
289            self.fd,
290            started,
291            deadline.saturating_duration_since(started),
292        )
293    }
294}
295
296impl fd::AsFd for Control<'_> {
297    fn as_fd(&self) -> fd::BorrowedFd<'_> {
298        self.fd
299    }
300}
301
302fn shutdown(socket: fd::BorrowedFd<'_>) -> Option<dope_core::Errno> {
303    loop {
304        match Backend::shutdown::<Errno>(socket, How::WRITE) {
305            Ok(()) => return None,
306            Err(errno) if errno.is_intr() => {}
307            Err(errno) => return Some(errno),
308        }
309    }
310}
311
312impl Worker {
313    pub fn run<T, E>(
314        self,
315        run: impl for<'a> FnOnce(process::Shutdown<'a>) -> Result<T, E>,
316    ) -> Outcome<T, E> {
317        let output = run(process::Shutdown::new(self.tx()));
318        let event = if output.is_ok() {
319            bus::Event::Done
320        } else {
321            bus::Event::Failed
322        };
323        let report = match self.send(event) {
324            Ok(()) => bus::Report::Clean,
325            Err(fault) => bus::Report::Failed(fault),
326        };
327        match (output, report) {
328            (Ok(value), bus::Report::Clean) if !matches!(self.phase.get(), Phase::Failed) => {
329                Outcome::Done(value)
330            }
331            (output, report) => Outcome::Failed { output, report },
332        }
333    }
334
335    pub(in crate::process) const fn tx(&self) -> Tx<'_> {
336        Tx(self, marker::PhantomData)
337    }
338
339    fn send(&self, event: bus::Event) -> Result<(), bus::Fault> {
340        if let Some(failure) = self.failure.get() {
341            return Err(failure);
342        }
343        if matches!(self.phase.get(), Phase::Failed) {
344            return Ok(());
345        }
346        if event == bus::Event::Failed {
347            self.phase.set(Phase::Failed);
348        }
349        let failure = loop {
350            match Backend::send_quiet::<Errno>(self.socket.as_fd(), &[event.encode()]) {
351                Ok([_]) => return Ok(()),
352                Ok(bytes) => {
353                    break bus::Write::Record(bus::RecordSize {
354                        expected: 1,
355                        actual: bytes.len(),
356                    });
357                }
358                Err(error) if error.is_intr() => {}
359                Err(error) => break bus::Write::Syscall(error),
360            }
361        };
362        let fault = bus::Fault {
363            event,
364            failure,
365            shutdown: shutdown(self.socket.as_fd()),
366        };
367        self.failure.set(Some(fault));
368        Err(fault)
369    }
370
371    fn report(&self) -> bus::Report {
372        match self.failure.get() {
373            Some(failure) => bus::Report::Failed(failure),
374            None => bus::Report::Clean,
375        }
376    }
377}
378
379impl<'a> Tx<'a> {
380    /// Publishes worker installation once across every session of this exact peer.
381    pub(in crate::process) fn installed(self) -> Result<(), bus::Fault> {
382        let peer = self.0;
383        if !matches!(peer.phase.get(), Phase::New) {
384            return peer.failure.get().map_or(Ok(()), Err);
385        }
386        peer.send(bus::Event::Installed)?;
387        peer.phase.set(Phase::Installed);
388        Ok(())
389    }
390
391    pub(in crate::process) fn send(self, event: bus::Event) -> Result<(), bus::Fault> {
392        self.0.send(event)
393    }
394
395    pub(in crate::process) fn report(self) -> bus::Report {
396        self.0.report()
397    }
398
399    pub(in crate::process) fn terminal(
400        self,
401        _: shutdown::Requested<process::Shutdown<'a>>,
402    ) -> Result<bus::Event, bus::Fault> {
403        match self.report() {
404            bus::Report::Failed(failure) => Err(failure),
405            bus::Report::Clean => Ok(if matches!(self.0.phase.get(), Phase::Failed) {
406                bus::Event::Failed
407            } else {
408                bus::Event::Done
409            }),
410        }
411    }
412
413    pub(in crate::process) fn as_fd(&self) -> fd::BorrowedFd<'a> {
414        self.0.socket.as_fd()
415    }
416}
417
418const _: () = {
419    assert!(mem::size_of::<Tx<'static>>() == mem::size_of::<usize>());
420    assert!(mem::align_of::<Tx<'static>>() == mem::align_of::<usize>());
421};
dope_runtime/process/lifecycle/mod.rs

dope_runtime/process/lifecycle/
mod.rs
