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 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    const INTEREST: dope_core::dope_sys::engine::register::Interest =
221        dope_core::dope_sys::engine::register::Interest::Closed;
222
223    fn installed(&self) -> Result<(), executor::InstallError> {
224        Ok(())
225    }
226}
227
228/// Observes terminal shutdown without consuming lifecycle records.
229impl executor::Build for &Main {
230    type Shutdown = Self;
231
232    fn try_build(
233        self,
234        attempt: &mut creation::Attempt,
235        config: settings::Config,
236    ) -> Result<executor::Executor<(), Self::Shutdown>, create::Error<'_>> {
237        executor::Executor::new(attempt, config)?
238            .shutdown(self)
239            .map_err(create::Error::from)
240    }
241}
242
243fn wait(
244    fd: fd::BorrowedFd<'_>,
245    started: time::Instant,
246    timeout: time::Duration,
247) -> Result<(), WaitError> {
248    loop {
249        let left = timeout.saturating_sub(started.elapsed());
250        let millis = left.as_millis() + u128::from(!left.subsec_nanos().is_multiple_of(1_000_000));
251        let mut entries = [poll::Entry::readable(fd)];
252        match Backend::wait::<_, Errno>(&mut entries, millis.min(i32::MAX as u128) as i32) {
253            Ok(events) if events.count() != 0 => return Ok(()),
254            Ok(_) => {}
255            Err(error) if error.is_intr() => {}
256            Err(error) => return Err(WaitError::Poll(error)),
257        }
258        let elapsed = started.elapsed();
259        if elapsed >= timeout {
260            return Err(WaitError::Timeout { timeout, elapsed });
261        }
262    }
263}
264
265impl<'a> Control<'a> {
266    pub(in crate::process) const fn new(fd: fd::BorrowedFd<'a>) -> Self {
267        Self {
268            fd,
269            _thread: ThreadBound::NEW,
270        }
271    }
272
273    pub fn stopped(&self) -> Result<bool, Errno> {
274        match Backend::recv_now::<Errno>(self.fd, &mut [0]) {
275            Ok(_) => Ok(true),
276            Err(error) if error.is_block() => Ok(false),
277            Err(error) if error.is_intr() => {
278                let mut entries = [poll::Entry::readable(self.fd)];
279                match Backend::wait::<_, Errno>(&mut entries, 0) {
280                    Ok(events) => Ok(events.count() != 0),
281                    Err(error) if error.is_intr() => Ok(false),
282                    Err(error) => Err(error),
283                }
284            }
285            Err(error) => Err(error),
286        }
287    }
288
289    pub fn wait_stop_until(&self, deadline: time::Instant) -> Result<(), WaitError> {
290        let started = time::Instant::now();
291        wait(
292            self.fd,
293            started,
294            deadline.saturating_duration_since(started),
295        )
296    }
297}
298
299impl fd::AsFd for Control<'_> {
300    fn as_fd(&self) -> fd::BorrowedFd<'_> {
301        self.fd
302    }
303}
304
305fn shutdown(socket: fd::BorrowedFd<'_>) -> Option<dope_core::Errno> {
306    loop {
307        match Backend::shutdown::<Errno>(socket, How::WRITE) {
308            Ok(()) => return None,
309            Err(errno) if errno.is_intr() => {}
310            Err(errno) => return Some(errno),
311        }
312    }
313}
314
315impl Worker {
316    pub fn run<T, E>(
317        self,
318        run: impl for<'a> FnOnce(process::Shutdown<'a>) -> Result<T, E>,
319    ) -> Outcome<T, E> {
320        let output = run(process::Shutdown::new(self.tx()));
321        let event = if output.is_ok() {
322            bus::Event::Done
323        } else {
324            bus::Event::Failed
325        };
326        let report = match self.send(event) {
327            Ok(()) => bus::Report::Clean,
328            Err(fault) => bus::Report::Failed(fault),
329        };
330        match (output, report) {
331            (Ok(value), bus::Report::Clean) if !matches!(self.phase.get(), Phase::Failed) => {
332                Outcome::Done(value)
333            }
334            (output, report) => Outcome::Failed { output, report },
335        }
336    }
337
338    pub(in crate::process) const fn tx(&self) -> Tx<'_> {
339        Tx(self, marker::PhantomData)
340    }
341
342    fn send(&self, event: bus::Event) -> Result<(), bus::Fault> {
343        if let Some(failure) = self.failure.get() {
344            return Err(failure);
345        }
346        if matches!(self.phase.get(), Phase::Failed) {
347            return Ok(());
348        }
349        if event == bus::Event::Failed {
350            self.phase.set(Phase::Failed);
351        }
352        let result = loop {
353            match Backend::send_quiet::<Errno>(self.socket.as_fd(), &[event.encode()]) {
354                Ok([_]) => break Ok(()),
355                Ok(bytes) => {
356                    break Err(bus::Write::Record(bus::RecordSize {
357                        expected: 1,
358                        actual: bytes.len(),
359                    }));
360                }
361                Err(error) if error.is_intr() => {}
362                Err(error) => break Err(bus::Write::Syscall(error)),
363            }
364        };
365        let shutdown = if result.is_err() || event == bus::Event::Failed {
366            shutdown(self.socket.as_fd())
367        } else {
368            None
369        };
370        let (failure, shutdown) = match (result, shutdown) {
371            (Ok(()), None) => return Ok(()),
372            (Ok(()), Some(error)) => (bus::Write::Shutdown(error), None),
373            (Err(failure), shutdown) => (failure, shutdown),
374        };
375        let fault = bus::Fault {
376            event,
377            failure,
378            shutdown,
379        };
380        self.failure.set(Some(fault));
381        Err(fault)
382    }
383
384    fn report(&self) -> bus::Report {
385        match self.failure.get() {
386            Some(failure) => bus::Report::Failed(failure),
387            None => bus::Report::Clean,
388        }
389    }
390}
391
392impl<'a> Tx<'a> {
393    /// Publishes worker installation once across every session of this exact peer.
394    pub(in crate::process) fn installed(self) -> Result<(), bus::Fault> {
395        let peer = self.0;
396        if !matches!(peer.phase.get(), Phase::New) {
397            return peer.failure.get().map_or(Ok(()), Err);
398        }
399        peer.send(bus::Event::Installed)?;
400        peer.phase.set(Phase::Installed);
401        Ok(())
402    }
403
404    pub(in crate::process) fn send(self, event: bus::Event) -> Result<(), bus::Fault> {
405        self.0.send(event)
406    }
407
408    pub(in crate::process) fn report(self) -> bus::Report {
409        self.0.report()
410    }
411
412    pub(in crate::process) fn terminal(
413        self,
414        _: shutdown::Requested<process::Shutdown<'a>>,
415    ) -> Result<bus::Event, bus::Fault> {
416        match self.report() {
417            bus::Report::Failed(failure) => Err(failure),
418            bus::Report::Clean => Ok(if matches!(self.0.phase.get(), Phase::Failed) {
419                bus::Event::Failed
420            } else {
421                bus::Event::Done
422            }),
423        }
424    }
425
426    pub(in crate::process) fn as_fd(&self) -> fd::BorrowedFd<'a> {
427        self.0.socket.as_fd()
428    }
429}
430
431const _: () = {
432    assert!(mem::size_of::<Tx<'static>>() == mem::size_of::<usize>());
433    assert!(mem::align_of::<Tx<'static>>() == mem::align_of::<usize>());
434};
dope_runtime/process/lifecycle/mod.rs

dope_runtime/process/lifecycle/
mod.rs
