dope_runtime/executor/
mod.rs

1use std::{convert, num::NonZero, os::fd};
2
3use dope_core::{
4    driver::{
5        life,
6        life::{creation, domain, recovery},
7        settings, storage,
8    },
9    random,
10};
11use o3::buffer;
12
13use crate::process::failure::bus;
14
15pub mod boot;
16pub mod create;
17pub mod drain;
18pub mod drive;
19pub mod raw;
20pub mod root;
21pub mod session;
22#[doc(hidden)]
23pub mod work;
24/// Runs once inside an executor generation with its exact storage family.
25pub trait Op<S: storage::Plan, Q>: Sized {
26    type Output;
27
28    fn run<'scope, 'life, 'p, 'd>(
29        self,
30        scope: session::Scope<'scope, 'life, 'p, 'd, S, Q>,
31    ) -> Self::Output
32    where
33        'd: 'p,
34        'p: 'life,
35        'd: 'life,
36        'life: 'scope,
37        Self: 'd,
38        S: 'd,
39        Q: 'd;
40}
41
42/// Adapts one closure to the executor's full storage and loan lifetime contract.
43#[repr(transparent)]
44pub struct Scoped<F>(F);
45
46impl<F> Scoped<F> {
47    pub const fn new<S: storage::Plan, Q, R>(run: F) -> Self
48    where
49        F: for<'scope, 'life, 'p, 'd> FnOnce(session::Scope<'scope, 'life, 'p, 'd, S, Q>) -> R,
50    {
51        Self(run)
52    }
53
54    pub fn into_inner(self) -> F {
55        self.0
56    }
57}
58
59impl<S, Q, F, R> Op<S, Q> for Scoped<F>
60where
61    S: storage::Plan,
62    F: for<'scope, 'life, 'p, 'd> FnOnce(session::Scope<'scope, 'life, 'p, 'd, S, Q>) -> R,
63{
64    type Output = R;
65
66    fn run<'scope, 'life, 'p, 'd>(
67        self,
68        scope: session::Scope<'scope, 'life, 'p, 'd, S, Q>,
69    ) -> Self::Output
70    where
71        'd: 'p,
72        'p: 'life,
73        'd: 'life,
74        'life: 'scope,
75        Self: 'd,
76        S: 'd,
77        Q: 'd,
78    {
79        (self.0)(scope)
80    }
81}
82
83struct Entry<O>(O);
84
85impl<S, Q, O> life::Operation<S, Q> for Entry<O>
86where
87    S: storage::Plan,
88    Q: Startup,
89    O: Op<S, Q>,
90{
91    type Output = O::Output;
92
93    fn rejected(&self, source: &Q) {
94        source.failed();
95    }
96
97    fn run<'life, 'p, 'd>(
98        self,
99        scope: life::Scope<'life, 'd>,
100        store: storage::Loan<'life, 'p, 'd, S>,
101        source: &'life Q,
102    ) -> Self::Output
103    where
104        'd: 'p,
105        'p: 'life,
106        'd: 'life,
107        Self: 'd,
108        S: 'd,
109        Q: 'd,
110    {
111        let mut core = session::Core::new(scope, source);
112        let scope = session::Scope::new(store, &mut core);
113        Op::run(self.0, scope)
114    }
115}
116
117/// Lifecycle behavior selected by an executor's shutdown source.
118pub trait Startup: Sized {
119    const INTEREST: dope_core::dope_sys::engine::register::Interest =
120        dope_core::dope_sys::engine::register::Interest::Readable;
121
122    /// Publishes installation readiness or preserves its exact notification failure.
123    fn installed(&self) -> Result<(), InstallError>;
124
125    /// Observes the registered shutdown source becoming ready.
126    fn observed(&self, _request: crate::shutdown::Requested<Self>) {}
127
128    /// Records installation failure before native cleanup.
129    fn failed(&self) {}
130}
131
132impl Startup for () {
133    fn installed(&self) -> Result<(), InstallError> {
134        Ok(())
135    }
136}
137
138#[derive(Clone, Copy, Debug, Eq, PartialEq)]
139pub enum InstallError {
140    Claim(settings::errors::Claim),
141    Coordinators { count: NonZero<usize>, limit: usize },
142    CoordinatorCountOverflow,
143    Process(bus::Fault),
144}
145
146impl From<settings::errors::Claim> for InstallError {
147    fn from(error: settings::errors::Claim) -> Self {
148        Self::Claim(error)
149    }
150}
151
152impl From<bus::Fault> for InstallError {
153    fn from(error: bus::Fault) -> Self {
154        Self::Process(error)
155    }
156}
157
158pub struct Executor<S = (), Q = ()> {
159    storage: S,
160    generation: domain::Generation<Q>,
161    teardown: domain::Teardown,
162}
163
164impl Executor<(), ()> {
165    pub fn new(
166        attempt: &mut creation::Attempt,
167        cfg: settings::Config,
168    ) -> Result<Self, create::Error<'_>> {
169        let driver = attempt.try_driver(cfg)?;
170        Ok(Self {
171            storage: (),
172            generation: domain::Generation::new(driver),
173            teardown: domain::Teardown::default(),
174        })
175    }
176
177    pub(crate) fn with_seed(
178        attempt: &mut creation::Attempt,
179        cfg: settings::Config,
180        seed: random::Seed,
181    ) -> Result<Self, create::Error<'_>> {
182        let driver = attempt.try_seed(cfg, seed)?;
183        Ok(Self {
184            storage: (),
185            generation: domain::Generation::new(driver),
186            teardown: domain::Teardown::default(),
187        })
188    }
189}
190
191impl<S, Q> Executor<S, Q> {
192    pub fn with_storage<T>(self, value: T) -> Executor<storage::Value<T>, Q> {
193        self.with_plan(storage::Value::new(value))
194    }
195
196    pub fn with_plan<T: storage::Plan>(self, plan: T) -> Executor<T, Q> {
197        Executor {
198            storage: plan,
199            generation: self.generation,
200            teardown: self.teardown,
201        }
202    }
203}
204
205pub trait Build: Sized {
206    type Shutdown: Startup;
207
208    fn try_build(
209        self,
210        attempt: &mut creation::Attempt,
211        config: settings::Config,
212    ) -> Result<Executor<(), Self::Shutdown>, create::Error<'_>>;
213}
214
215impl Build for () {
216    type Shutdown = ();
217
218    fn try_build(
219        self,
220        attempt: &mut creation::Attempt,
221        config: settings::Config,
222    ) -> Result<Executor<(), Self::Shutdown>, create::Error<'_>> {
223        Executor::new(attempt, config)
224    }
225}
226
227impl<S> Executor<S, ()> {
228    /// Installs and owns a process-local shutdown source for this driver domain.
229    pub fn shutdown<Q: fd::AsFd + Startup>(
230        self,
231        source: Q,
232    ) -> Result<Executor<S, Q>, domain::Register> {
233        let Self {
234            storage: store,
235            generation: epoch,
236            teardown,
237        } = self;
238        Ok(Executor {
239            storage: store,
240            generation: epoch.poll(source, Q::INTEREST)?,
241            teardown,
242        })
243    }
244}
245
246type ExecResult<'owner, S, O, Q> = Result<
247    <O as Op<S, Q>>::Output,
248    domain::Error<
249        'owner,
250        domain::Rejected<<S as storage::Build>::Error, O>,
251        <O as Op<S, Q>>::Output,
252        Q,
253    >,
254>;
255type CreateResult<'owner, S, F, R, Q> =
256    Result<R, domain::Error<'owner, domain::Rejected<<S as storage::Create>::Error, F>, R, Q>>;
257
258/// One execution borrows its exact persistent native retirement owner.
259pub struct Execution<'owner, S, Q> {
260    storage: S,
261    domain: domain::Session<'owner, Q>,
262}
263
264impl<S, Q> Executor<S, Q> {
265    pub fn run<R>(self, run: impl for<'owner> FnOnce(Execution<'owner, S, Q>) -> R) -> R {
266        let Self {
267            storage,
268            generation: epoch,
269            mut teardown,
270        } = self;
271        let session = epoch.start(&mut teardown);
272        run(Execution {
273            storage,
274            domain: session,
275        })
276    }
277}
278
279impl<'owner, S: storage::Build, Q: Startup> Execution<'owner, S, Q> {
280    pub fn try_enter<O>(self, op: O, report: impl recovery::Observe) -> ExecResult<'owner, S, O, Q>
281    where
282        O: Op<S, Q>,
283    {
284        let Self {
285            storage: store,
286            domain,
287        } = self;
288        domain
289            .try_enter(store, Entry(op), report)
290            .map_err(|error| error.map_storage(|rejected| rejected.map_op(|entry| entry.0)))
291    }
292}
293
294impl<'owner, S: storage::Create, Q: Startup> Execution<'owner, S, Q> {
295    pub fn try_create<F, R>(
296        self,
297        op: F,
298        report: impl recovery::Observe,
299    ) -> CreateResult<'owner, S, F, R, Q>
300    where
301        F: for<'scope, 'life, 'p, 'd> FnOnce(session::Scope<'scope, 'life, 'p, 'd, S, Q>) -> R,
302    {
303        self.try_scope(Scoped(op), report)
304            .map_err(|error| error.map_storage(|rejected| rejected.map_op(|entry| entry.0)))
305    }
306
307    pub(crate) fn try_scope<O>(
308        self,
309        op: O,
310        report: impl recovery::Observe,
311    ) -> CreateResult<'owner, S, O, O::Output, Q>
312    where
313        O: Op<S, Q>,
314    {
315        let Self {
316            storage: store,
317            domain,
318        } = self;
319        domain
320            .try_create(store, Entry(op), report)
321            .map_err(|error| error.map_storage(|rejected| rejected.map_op(|entry| entry.0)))
322    }
323}
324
325impl<'owner, S, Q> Execution<'owner, S, Q>
326where
327    S: storage::Direct,
328    Q: Startup,
329{
330    pub fn enter<O>(
331        self,
332        op: O,
333        report: impl recovery::Observe,
334    ) -> Result<O::Output, domain::Error<'owner, convert::Infallible, O::Output, Q>>
335    where
336        O: Op<S, Q>,
337    {
338        let Self {
339            storage: store,
340            domain,
341        } = self;
342        domain.enter(store, Entry(op), report)
343    }
344
345    pub fn enter_with<R>(
346        self,
347        op: impl for<'scope, 'life, 'p, 'd> FnOnce(session::Scope<'scope, 'life, 'p, 'd, S, Q>) -> R,
348        report: impl recovery::Observe,
349    ) -> Result<R, domain::Error<'owner, convert::Infallible, R, Q>> {
350        self.enter(Scoped(op), report)
351    }
352}
353
354impl<'owner, A: buffer::raw::Arenas, Q: Startup> Execution<'owner, storage::Scope<A>, Q> {
355    /// Lends exact arena roots from the same generation as the driver.
356    pub fn scope<R>(
357        self,
358        op: impl for<'scope, 'life, 'p, 'd> FnOnce(
359            session::Scope<'scope, 'life, 'p, 'd, storage::Scope<A>, Q>,
360        ) -> R,
361        report: impl recovery::Observe,
362    ) -> Result<R, domain::Error<'owner, convert::Infallible, R, Q>> {
363        let Self {
364            storage: store,
365            domain,
366        } = self;
367        domain.scope(store, Entry(Scoped(op)), report)
368    }
369}
dope_runtime/executor/mod.rs

dope_runtime/executor/
mod.rs
