dope_runtime/
shutdown.rs

1use std::{cell::Cell, marker, mem, os::fd, time};
2
3use dope_core::{
4    driver::{life::creation, settings},
5    platform::wake::{CreateError, Ends, Notify, NotifyError, Wait, WaitError},
6};
7use o3::base::marker::ThreadBound;
8
9use crate::executor::{self, create};
10
11#[derive(Debug)]
12pub struct FireError {
13    pub trigger: Trigger,
14    pub cause: NotifyError,
15}
16
17#[must_use]
18pub struct Pair {
19    source: Source,
20    trigger: Trigger,
21}
22
23#[must_use]
24pub struct Source {
25    pub(crate) event: (fd::OwnedFd, Notify),
26}
27
28impl fd::AsFd for Source {
29    fn as_fd(&self) -> fd::BorrowedFd<'_> {
30        fd::AsFd::as_fd(&self.event.0)
31    }
32}
33
34#[must_use]
35#[derive(Debug)]
36pub struct Trigger(Notify);
37
38/// Shutdown source whose application-install edge is observable exactly once.
39#[must_use]
40pub struct ReadySource {
41    shutdown: Source,
42    ready: Cell<Option<Notify>>,
43}
44
45impl fd::AsFd for ReadySource {
46    fn as_fd(&self) -> fd::BorrowedFd<'_> {
47        fd::AsFd::as_fd(&self.shutdown)
48    }
49}
50
51/// Blocking observer for one application-install edge.
52#[must_use]
53pub struct Readiness(Wait);
54
55/// Affine shutdown and readiness authorities for a single runtime.
56#[must_use]
57pub struct ReadyPair {
58    source: ReadySource,
59    trigger: Trigger,
60    readiness: Readiness,
61}
62
63type Invariant<Q> = marker::PhantomData<(fn(Q) -> Q, ThreadBound)>;
64
65#[must_use]
66pub struct Requested<Q = Source>(Invariant<Q>);
67
68impl Pair {
69    pub fn new() -> Result<Self, CreateError> {
70        let (event, notify) = Ends::event()?.split();
71        let (guard, notify) = notify.fork()?;
72        Ok(Self {
73            source: Source {
74                event: (event, guard),
75            },
76            trigger: Trigger(notify),
77        })
78    }
79
80    pub fn split(self) -> (Source, Trigger) {
81        (self.source, self.trigger)
82    }
83
84    pub fn run<R>(self, execute: impl FnOnce(Source) -> R) -> R {
85        execute(self.source)
86    }
87}
88
89impl ReadyPair {
90    pub fn new() -> Result<Self, CreateError> {
91        let (stop, trig) = Pair::new()?.split();
92        let (ready, notify) = Ends::blocking()?.split();
93        Ok(Self {
94            source: ReadySource {
95                shutdown: stop,
96                ready: Cell::new(Some(notify)),
97            },
98            trigger: trig,
99            readiness: Readiness(ready),
100        })
101    }
102
103    pub fn split(self) -> (ReadySource, Trigger, Readiness) {
104        (self.source, self.trigger, self.readiness)
105    }
106}
107
108impl executor::Build for Source {
109    type Shutdown = Source;
110
111    fn try_build(
112        self,
113        attempt: &mut creation::Attempt,
114        config: settings::Config,
115    ) -> Result<executor::Executor<(), Self::Shutdown>, create::Error<'_>> {
116        executor::Executor::new(attempt, config)?
117            .shutdown(self)
118            .map_err(create::Error::from)
119    }
120}
121
122impl executor::Build for ReadySource {
123    type Shutdown = ReadySource;
124
125    fn try_build(
126        self,
127        attempt: &mut creation::Attempt,
128        config: settings::Config,
129    ) -> Result<executor::Executor<(), Self::Shutdown>, create::Error<'_>> {
130        executor::Executor::new(attempt, config)?
131            .shutdown(self)
132            .map_err(create::Error::from)
133    }
134}
135
136impl executor::Startup for Source {
137    fn installed(&self) -> Result<(), executor::InstallError> {
138        Ok(())
139    }
140}
141
142impl executor::Startup for ReadySource {
143    fn installed(&self) -> Result<(), executor::InstallError> {
144        let Some(notify) = self.ready.take() else {
145            return Ok(());
146        };
147        match notify.notify() {
148            Ok(()) => Ok(()),
149            Err(NotifyError::Write(error)) if error.is_pipe() => Ok(()),
150            Err(error) => {
151                self.ready.set(Some(notify));
152                Err(error.into())
153            }
154        }
155    }
156}
157
158impl Readiness {
159    pub fn wait(self) -> Result<(), WaitError> {
160        self.0.wait()
161    }
162
163    pub fn wait_timeout(self, limit: time::Duration) -> Result<(), WaitError> {
164        self.0.wait_timeout(limit)
165    }
166}
167
168impl Trigger {
169    pub fn fire(self) -> Result<(), FireError> {
170        match self.0.notify() {
171            Ok(()) => Ok(()),
172            Err(cause) => Err(FireError {
173                trigger: self,
174                cause,
175            }),
176        }
177    }
178}
179
180impl<Q> Requested<Q> {
181    pub(crate) const fn new() -> Self {
182        Self(marker::PhantomData)
183    }
184}
185
186const _: () = assert!(std::mem::size_of::<Requested>() == 0);
187const _: () = assert!(mem::size_of::<Option<Notify>>() == mem::size_of::<fd::OwnedFd>());
dope_runtime/shutdown.rs

dope_runtime/
shutdown.rs
