dope_core/driver/schedule/app/
mod.rs

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

dope_core/driver/schedule/app/
mod.rs
