dope_runtime/process/runtime/
mod.rs

1use std::convert;
2
3use dope_core::{platform::signal, random};
4use o3::mem::slot;
5
6use crate::{
7    process::{self, failure::execute, gate, layout, watch},
8    shutdown,
9};
10
11pub(in crate::process) mod bus;
12mod control;
13pub(in crate::process) mod worker;
14
15/// Owns process workers through cooperative shutdown and joining.
16///
17/// Enter on the original thread during single-threaded startup. Worker entries
18/// must observe shutdown and return; every started worker is joined before these
19/// methods return. A stalled worker can keep the caller waiting indefinitely.
20/// Heartbeat callbacks do not interrupt worker execution.
21pub struct Runtime<I> {
22    pub(in crate::process) workers: layout::Set<I>,
23}
24
25impl<I> Runtime<I> {
26    /// Runs workers whose application readiness is their executor-install edge.
27    pub fn run<'r, E, X>(
28        self,
29        storage: slot::Vacant<'r, signal::Receiver>,
30        entry: E,
31    ) -> Result<(), execute::Error<'r, convert::Infallible, X>>
32    where
33        I: Send,
34        E: for<'run> Fn(
35                I,
36                process::Context<'run>,
37            ) -> Result<shutdown::Requested<process::Shutdown<'run>>, X>
38            + Clone
39            + Send,
40        X: Send,
41    {
42        let entry = worker::Plain::new(entry);
43        let Self { workers } = self;
44        signal::Source::scope(storage, move |signal| {
45            run_waiting(workers, entry, control::Control::signal(signal))
46        })
47        .map_err(execute::Error::from)
48    }
49
50    /// Runs workers and supervises an explicit steady-state heartbeat.
51    pub fn run_watched<'r, N, E, X>(
52        self,
53        storage: slot::Vacant<'r, signal::Receiver>,
54        entry: E,
55        notify: N,
56    ) -> Result<(), execute::Error<'r, N::Error, X>>
57    where
58        I: Send,
59        N: watch::Notify,
60        E: for<'run> Fn(
61                I,
62                process::Context<'run>,
63            ) -> Result<shutdown::Requested<process::Shutdown<'run>>, X>
64            + Clone
65            + Send,
66        X: Send,
67    {
68        self.watch_inner(storage, worker::Plain::new(entry), notify)
69    }
70
71    /// Runs gated workers and supervises an explicit steady-state heartbeat.
72    pub fn watch_gated<'r, N, E, X>(
73        self,
74        storage: slot::Vacant<'r, signal::Receiver>,
75        entry: E,
76        notify: N,
77    ) -> Result<(), execute::Error<'r, N::Error, X>>
78    where
79        I: Send,
80        N: watch::Notify,
81        E: for<'run> Fn(
82                I,
83                process::Context<'run>,
84                gate::Closed<'run>,
85            ) -> Result<shutdown::Requested<process::Shutdown<'run>>, X>
86            + Clone
87            + Send,
88        X: Send,
89    {
90        self.watch_inner(storage, worker::Gated::new(entry), notify)
91    }
92
93    fn watch_inner<'r, N, E, X>(
94        self,
95        storage: slot::Vacant<'r, signal::Receiver>,
96        entry: E,
97        notify: N,
98    ) -> Result<(), execute::Error<'r, N::Error, X>>
99    where
100        I: Send,
101        N: watch::Notify,
102        E: worker::Entry<I, X>,
103        X: Send,
104    {
105        let Self { workers } = self;
106        signal::Source::scope(storage, move |signal| {
107            run_waiting_watched(workers, entry, notify, control::Control::signal(signal))
108        })
109        .map_err(execute::Error::from)
110    }
111}
112
113fn run_waiting<'r, I, E, X>(
114    workers: layout::Set<I>,
115    entry: worker::Plain<E>,
116    control: control::Control<'_>,
117) -> Result<(), execute::Error<'r, convert::Infallible, X>>
118where
119    I: Send,
120    worker::Plain<E>: worker::Entry<I, X>,
121    X: Send,
122{
123    let count = workers.len();
124    let seed = random::Seed::acquire().map_err(execute::Cause::Entropy)?;
125    worker::scope(workers, seed, entry, |running| {
126        let outcome = control.ready(count, &running);
127        let stopped = running.stop();
128        match outcome {
129            Ok(()) => stopped.clean(bus::Readiness::Required),
130            Err(error) => Err(stopped.fail(error)),
131        }
132    })
133}
134
135fn run_waiting_watched<'r, I, N, E, X>(
136    workers: layout::Set<I>,
137    entry: E,
138    notify: N,
139    control: control::Control<'_>,
140) -> Result<(), execute::Error<'r, N::Error, X>>
141where
142    I: Send,
143    N: watch::Notify,
144    E: worker::Entry<I, X>,
145    X: Send,
146{
147    let count = workers.len();
148    let seed = random::Seed::acquire().map_err(execute::Cause::Entropy)?;
149    worker::scope(workers, seed, entry, |running| {
150        let outcome = control.watch(count, notify, &running);
151        let stopped = running.stop();
152        let outcome = outcome.and_then(|clock| match clock {
153            Some(clock) => control::Finish::new(&stopped)
154                .map_or(Ok(()), |finish| finish.drain(count, clock))
155                .map(|()| bus::Readiness::Required),
156            None => Ok(bus::Readiness::Cancelled),
157        });
158        match outcome {
159            Ok(readiness) => stopped.clean(readiness),
160            Err(error) => Err(stopped.fail(error)),
161        }
162    })
163}
dope_runtime/process/runtime/mod.rs

dope_runtime/process/runtime/
mod.rs
