dope_core/driver/schedule/app/
mod.rs

1use std::{marker, mem::size_of};
2
3use mem::{lease, quota};
4use o3::{base::marker::ThreadBound, mem};
5use state::lanes;
6
7use crate::driver::{
8    schedule,
9    schedule::{clock, state},
10};
11
12pub mod batch;
13pub mod step;
14
15pub type Selection<I, P, C> = Option<Result<(I, P), C>>;
16
17/// A stable batch whose front is consumed only after application admission.
18/// A selected item carries the exact receipt returned by `take`; implementations
19/// cannot create an admission independently of that callback.
20pub trait Drain {
21    type Candidate<'a>
22    where
23        Self: 'a;
24    type Item<'a>
25    where
26        Self: 'a;
27
28    fn next_when<P>(
29        &mut self,
30        take: impl FnOnce() -> Option<P>,
31    ) -> Selection<Self::Item<'_>, P, Self::Candidate<'_>>;
32}
33
34#[derive(Clone, Copy)]
35#[repr(transparent)]
36pub struct Work<'turn, 'd> {
37    pub(in crate::driver::schedule) state: &'turn state::Core<'d>,
38}
39
40#[must_use = "an application permit represents one admitted transition"]
41pub struct Permit<'turn, 'd> {
42    proof: marker::PhantomData<(&'turn mut (), crate::driver::Invariant<'d>, ThreadBound)>,
43}
44
45#[must_use = "dropping an uncommitted application lease returns its turn"]
46pub struct Lease<'turn, 'd> {
47    inner: mem::Lease<quota::Return<'turn, state::AppLane>, lease::Open>,
48    driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
49}
50
51#[must_use = "next reports whether work was ready or the application lane was exhausted"]
52pub enum Next<'turn, 'd, T, C = T> {
53    Item(T, Permit<'turn, 'd>),
54    Empty,
55    Exhausted(C),
56}
57
58#[must_use = "an application share scopes one participant's scheduler turn"]
59#[repr(transparent)]
60pub struct Share<'turn, 'd> {
61    turn: schedule::Turn<'turn, 'd>,
62}
63
64impl<'turn, 'd> Work<'turn, 'd> {
65    pub fn remaining(self) -> usize {
66        self.state.lanes.app.remaining()
67    }
68
69    pub fn clock(self) -> clock::Clock<'turn, 'd> {
70        self.state.driver.scheduler().clock()
71    }
72
73    pub fn run<T>(self, run: impl FnOnce() -> Option<T>) -> quota::Admission<T> {
74        self.state.lanes.app.admit_with(run)
75    }
76
77    pub fn apply(self, run: impl FnOnce() -> quota::Acquire) {
78        self.state.lanes.app.apply(run);
79    }
80
81    pub fn permit(self) -> Option<Permit<'turn, 'd>> {
82        match self.state.lanes.app.take() {
83            quota::Take::Taken => Some(Permit::new()),
84            quota::Take::Exhausted => None,
85        }
86    }
87
88    pub fn lease(self) -> Option<Lease<'turn, 'd>> {
89        Some(Lease {
90            inner: self.state.lanes.app.open()?,
91            driver: marker::PhantomData,
92        })
93    }
94
95    pub fn claim<T>(
96        self,
97        claim: impl FnOnce() -> Option<T>,
98    ) -> quota::Admission<(T, Permit<'turn, 'd>)> {
99        self.run(|| claim().map(|value| (value, Permit::new())))
100    }
101
102    pub fn next<'a, R: Drain>(
103        self,
104        ready: &'a mut R,
105    ) -> Next<'turn, 'd, R::Item<'a>, R::Candidate<'a>> {
106        match Drain::next_when(ready, || self.permit()) {
107            None => Next::Empty,
108            Some(Ok((value, permit))) => Next::Item(value, permit),
109            Some(Err(value)) => Next::Exhausted(value),
110        }
111    }
112
113    pub fn share<const N: usize, T>(
114        self,
115        run: impl for<'scope> FnOnce(Share<'scope, 'd>) -> T,
116    ) -> T {
117        let _reserve = lanes::Quota::share::<N, _>(&self.state.lanes.app, 0);
118        run(Share {
119            turn: schedule::Turn::new(self.state),
120        })
121    }
122}
123
124impl<'turn, 'd> Share<'turn, 'd> {
125    pub fn turn(&self) -> schedule::Turn<'turn, 'd> {
126        self.turn
127    }
128
129    pub fn work(&self) -> Work<'turn, 'd> {
130        self.turn.app()
131    }
132}
133
134impl<'turn, 'd> Permit<'turn, 'd> {
135    fn new() -> Self {
136        Self {
137            proof: marker::PhantomData,
138        }
139    }
140
141    pub fn run<T>(self, run: impl FnOnce() -> T) -> T {
142        run()
143    }
144}
145
146impl<'turn, 'd> Lease<'turn, 'd> {
147    pub fn commit(self) -> Permit<'turn, 'd> {
148        let Self {
149            inner,
150            driver: marker::PhantomData,
151        } = self;
152        let () = inner.into_parts().1;
153        Permit::new()
154    }
155}
156
157const _: () = {
158    let word = size_of::<usize>();
159    assert!(size_of::<Work<'static, 'static>>() == word);
160    assert!(size_of::<Permit<'static, 'static>>() == 0);
161    assert!(size_of::<Lease<'static, 'static>>() == word);
162    assert!(size_of::<Next<'static, 'static, usize>>() == 2 * word);
163    assert!(size_of::<Share<'static, 'static>>() == word);
164};
dope_core/driver/schedule/app/mod.rs

dope_core/driver/schedule/app/
mod.rs
