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    /// Requests another driver turn for known runnable work.
74    pub fn defer(self) {
75        self.state.deferred.set(true);
76    }
77
78    pub fn run<T>(self, run: impl FnOnce() -> Option<T>) -> quota::Admission<T> {
79        self.state.lanes.app.admit_with(run)
80    }
81
82    pub fn apply(self, run: impl FnOnce() -> quota::Acquire) {
83        self.state.lanes.app.apply(run);
84    }
85
86    pub fn permit(self) -> Option<Permit<'turn, 'd>> {
87        match self.state.lanes.app.take() {
88            quota::Take::Taken => Some(Permit::new()),
89            quota::Take::Exhausted => None,
90        }
91    }
92
93    pub fn lease(self) -> Option<Lease<'turn, 'd>> {
94        Some(Lease {
95            inner: self.state.lanes.app.open()?,
96            driver: marker::PhantomData,
97        })
98    }
99
100    pub fn claim<T>(
101        self,
102        claim: impl FnOnce() -> Option<T>,
103    ) -> quota::Admission<(T, Permit<'turn, 'd>)> {
104        self.run(|| claim().map(|value| (value, Permit::new())))
105    }
106
107    pub fn next<'a, R: Drain>(
108        self,
109        ready: &'a mut R,
110    ) -> Next<'turn, 'd, R::Item<'a>, R::Candidate<'a>> {
111        match Drain::next_when(ready, || self.permit()) {
112            None => Next::Empty,
113            Some(Ok((value, permit))) => Next::Item(value, permit),
114            Some(Err(value)) => Next::Exhausted(value),
115        }
116    }
117
118    pub fn share<const N: usize, T>(
119        self,
120        run: impl for<'scope> FnOnce(Share<'scope, 'd>) -> T,
121    ) -> T {
122        let _reserve = lanes::share::<N, _>(&self.state.lanes.app, 0);
123        run(Share {
124            turn: schedule::Turn::new(self.state),
125        })
126    }
127}
128
129impl<'turn, 'd> Share<'turn, 'd> {
130    pub fn turn(&self) -> schedule::Turn<'turn, 'd> {
131        self.turn
132    }
133
134    pub fn work(&self) -> Work<'turn, 'd> {
135        self.turn.app()
136    }
137}
138
139impl<'turn, 'd> Permit<'turn, 'd> {
140    fn new() -> Self {
141        Self {
142            proof: marker::PhantomData,
143        }
144    }
145
146    pub fn run<T>(self, run: impl FnOnce() -> T) -> T {
147        run()
148    }
149}
150
151impl<'turn, 'd> Lease<'turn, 'd> {
152    pub fn commit(self) -> Permit<'turn, 'd> {
153        let Self {
154            inner,
155            driver: marker::PhantomData,
156        } = self;
157        let () = inner.into_parts().1;
158        Permit::new()
159    }
160}
161
162const _: () = {
163    let word = size_of::<usize>();
164    assert!(size_of::<Work<'static, 'static>>() == word);
165    assert!(size_of::<Permit<'static, 'static>>() == 0);
166    assert!(size_of::<Lease<'static, 'static>>() == word);
167    assert!(size_of::<Next<'static, 'static, usize>>() == 2 * word);
168    assert!(size_of::<Share<'static, 'static>>() == word);
169};
dope_core/driver/schedule/app/mod.rs

dope_core/driver/schedule/app/
mod.rs
