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<C = ()> {
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        context: &mut C,
31        take: impl FnOnce() -> Option<P>,
32    ) -> Selection<Self::Item<'_>, P, Self::Candidate<'_>>;
33}
34
35#[derive(Clone, Copy)]
36#[repr(transparent)]
37pub struct Work<'turn, 'd> {
38    pub(in crate::driver::schedule) state: &'turn state::Core<'d>,
39}
40
41#[must_use = "an application permit represents one admitted transition"]
42pub struct Permit<'turn, 'd> {
43    proof: marker::PhantomData<(&'turn mut (), crate::driver::Invariant<'d>, ThreadBound)>,
44}
45
46#[must_use = "dropping an uncommitted application lease returns its turn"]
47pub struct Lease<'turn, 'd> {
48    inner: mem::Lease<quota::Return<'turn, state::AppLane>, lease::Open>,
49    driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
50}
51
52#[must_use = "next reports whether work was ready or the application lane was exhausted"]
53pub enum Next<'turn, 'd, T, C = T> {
54    Item(T, Permit<'turn, 'd>),
55    Empty,
56    Exhausted(C),
57}
58
59#[must_use = "an application share scopes one participant's scheduler turn"]
60#[repr(transparent)]
61pub struct Share<'turn, 'd> {
62    turn: schedule::Turn<'turn, 'd>,
63}
64
65impl<'turn, 'd> Work<'turn, 'd> {
66    pub fn remaining(self) -> usize {
67        self.state.lanes.app.remaining()
68    }
69
70    pub fn clock(self) -> clock::Clock<'turn, 'd> {
71        self.state.driver.scheduler().clock()
72    }
73
74    /// Requests another driver turn for known runnable work.
75    pub fn defer(self) {
76        self.state.deferred.set(true);
77    }
78
79    pub fn run<T>(self, run: impl FnOnce() -> Option<T>) -> quota::Admission<T> {
80        self.state.lanes.app.admit_with(run)
81    }
82
83    pub fn apply(self, run: impl FnOnce() -> quota::Acquire) {
84        self.state.lanes.app.apply(run);
85    }
86
87    pub fn permit(self) -> Option<Permit<'turn, 'd>> {
88        match self.state.lanes.app.take() {
89            quota::Take::Taken => Some(Permit::new()),
90            quota::Take::Exhausted => None,
91        }
92    }
93
94    pub fn lease(self) -> Option<Lease<'turn, 'd>> {
95        Some(Lease {
96            inner: self.state.lanes.app.open()?,
97            driver: marker::PhantomData,
98        })
99    }
100
101    pub fn claim<T>(
102        self,
103        claim: impl FnOnce() -> Option<T>,
104    ) -> quota::Admission<(T, Permit<'turn, 'd>)> {
105        self.run(|| claim().map(|value| (value, Permit::new())))
106    }
107
108    pub fn next<'a, C, R: Drain<C>>(
109        self,
110        ready: &'a mut R,
111        context: &mut C,
112    ) -> Next<'turn, 'd, R::Item<'a>, R::Candidate<'a>> {
113        match Drain::next_when(ready, context, || self.permit()) {
114            None => Next::Empty,
115            Some(Ok((value, permit))) => Next::Item(value, permit),
116            Some(Err(value)) => Next::Exhausted(value),
117        }
118    }
119
120    pub fn share<const N: usize, T>(
121        self,
122        run: impl for<'scope> FnOnce(Share<'scope, 'd>) -> T,
123    ) -> T {
124        let _reserve = lanes::share::<N, _>(&self.state.lanes.app, 0);
125        run(Share {
126            turn: schedule::Turn::new(self.state),
127        })
128    }
129}
130
131impl<'turn, 'd> Share<'turn, 'd> {
132    pub fn turn(&self) -> schedule::Turn<'turn, 'd> {
133        self.turn
134    }
135
136    pub fn work(&self) -> Work<'turn, 'd> {
137        self.turn.app()
138    }
139}
140
141impl<'turn, 'd> Permit<'turn, 'd> {
142    fn new() -> Self {
143        Self {
144            proof: marker::PhantomData,
145        }
146    }
147
148    pub fn run<T>(self, run: impl FnOnce() -> T) -> T {
149        run()
150    }
151}
152
153impl<'turn, 'd> Lease<'turn, 'd> {
154    pub fn commit(self) -> Permit<'turn, 'd> {
155        let Self {
156            inner,
157            driver: marker::PhantomData,
158        } = self;
159        let () = inner.into_parts().1;
160        Permit::new()
161    }
162}
163
164const _: () = {
165    let word = size_of::<usize>();
166    assert!(size_of::<Work<'static, 'static>>() == word);
167    assert!(size_of::<Permit<'static, 'static>>() == 0);
168    assert!(size_of::<Lease<'static, 'static>>() == word);
169    assert!(size_of::<Next<'static, 'static, usize>>() == 2 * word);
170    assert!(size_of::<Share<'static, 'static>>() == word);
171};
dope_core/driver/schedule/app/mod.rs

dope_core/driver/schedule/app/
mod.rs
