dope_core/driver/schedule/
mod.rs

1use std::{cell, marker, time};
2
3use o3::{cell::region, collections::batch::set, mem::quota};
4
5mod credits;
6mod progress;
7pub mod ready;
8pub mod reservation;
9pub mod timer;
10mod wait;
11
12pub(crate) use credits::Budget;
13pub use progress::Progress;
14pub use wait::Wait;
15
16#[must_use = "admission reports whether work was acquired or the budget was exhausted"]
17pub enum Admission<T> {
18    Item(T),
19    Empty,
20    Exhausted,
21}
22
23use crate::driver::{self, route};
24
25/// Hard ceiling for every scheduler work class in one driver turn.
26pub const MAX_TURN_WORK_BUDGET: usize = 256;
27pub(crate) const REACTOR_LANES: u8 = 2;
28
29enum ReactorLane {}
30enum CoordinationLane {}
31enum ApplicationLane {}
32enum MaintenanceLane {}
33
34pub(crate) struct Work {
35    reactor: quota::Ledger<ReactorLane>,
36    reactor_cursor: cell::Cell<u8>,
37    ready: quota::Ledger<ready::Lane>,
38    coordination: quota::Ledger<CoordinationLane>,
39    application: quota::Ledger<ApplicationLane>,
40    timers: quota::Ledger<timer::Lane>,
41    maintenance: quota::Ledger<MaintenanceLane>,
42    maintenance_half: cell::Cell<Option<usize>>,
43}
44
45pub struct Controller<'scope, 'd> {
46    driver: driver::Reference<'d>,
47    work: &'scope mut Work,
48}
49
50#[must_use = "an active scheduler turn must remain live for the complete driver cycle"]
51pub struct ActiveTurn<'turn, 'd> {
52    driver: driver::Reference<'d>,
53    work: &'turn mut Work,
54}
55
56const _: () = {
57    assert!(
58        std::mem::size_of::<Controller<'static, 'static>>() == 2 * std::mem::size_of::<usize>()
59    );
60    assert!(
61        std::mem::size_of::<ActiveTurn<'static, 'static>>() == 2 * std::mem::size_of::<usize>()
62    );
63};
64
65#[derive(Clone, Copy)]
66#[repr(transparent)]
67#[doc = include_str!("turn.md")]
68pub struct Turn<'turn, 'd> {
69    work: &'turn Work,
70    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
71}
72
73#[doc = include_str!("share.md")]
74#[must_use = "maintenance share must remain live while its participant runs"]
75pub struct Share<'turn, 'd, const PARTICIPANTS: usize> {
76    _reserve: credits::Quota<'turn, 'd>,
77    _participants: marker::PhantomData<ParticipantCount<PARTICIPANTS>>,
78}
79
80struct ParticipantCount<const PARTICIPANTS: usize>;
81
82const _: () =
83    assert!(std::mem::size_of::<Share<'static, 'static, 1>>() == 2 * std::mem::size_of::<usize>());
84
85#[derive(Clone, Copy)]
86#[repr(transparent)]
87pub struct Half<'turn, 'd> {
88    work: &'turn Work,
89    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
90}
91
92#[doc(hidden)]
93#[must_use = "reactor authority must be consumed by Poll or Source"]
94pub struct Reactor<'turn, 'd> {
95    quota: credits::Quota<'turn, 'd>,
96    cursor: u8,
97    _turn: marker::PhantomData<&'turn mut ()>,
98    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
99}
100
101#[derive(Clone, Copy)]
102#[repr(transparent)]
103pub struct Application<'turn, 'd> {
104    remaining: &'turn quota::Ledger<ApplicationLane>,
105    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
106}
107
108/// One non-duplicable application transition admitted by a driver turn.
109///
110/// The permit is created only by spending one unit from the turn ledger. Its
111/// invariant driver brand lets a driver-owned consumer require the same
112/// branded lifetime instead of relying on a runtime owner comparison.
113///
114/// A permit cannot be duplicated:
115///
116/// ```compile_fail,E0382
117/// use dope_core::driver::schedule::ApplicationPermit;
118///
119/// fn duplicate<'turn, 'd>(permit: ApplicationPermit<'turn, 'd>) {
120///     let moved = permit;
121///     drop(permit);
122///     drop(moved);
123/// }
124/// ```
125#[must_use = "an application permit represents one already-spent transition"]
126pub struct ApplicationPermit<'turn, 'd> {
127    _turn: marker::PhantomData<&'turn mut ()>,
128    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
129}
130
131#[must_use = "admission reports whether work was acquired or the application budget was exhausted"]
132pub enum ApplicationAdmission<'turn, 'd, T> {
133    Item(T, ApplicationPermit<'turn, 'd>),
134    Empty,
135    Exhausted(T),
136}
137
138const _: () = {
139    use std::mem;
140
141    assert!(mem::size_of::<ApplicationPermit<'static, 'static>>() == 0);
142    assert!(
143        mem::size_of::<ApplicationAdmission<'static, 'static, usize>>()
144            == mem::size_of::<quota::Admission<usize>>()
145    );
146};
147
148/// Linear coordination work class for one driver turn.
149#[doc(hidden)]
150pub struct Coordination<'turn, 'd> {
151    remaining: &'turn quota::Ledger<CoordinationLane>,
152    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
153}
154
155#[derive(Clone, Copy)]
156#[repr(transparent)]
157#[doc(hidden)]
158pub struct Timers<'turn, 'd> {
159    remaining: &'turn quota::Ledger<timer::Lane>,
160    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
161}
162
163#[derive(Clone, Copy)]
164#[repr(transparent)]
165pub struct Maintenance<'turn, 'd> {
166    work: &'turn Work,
167    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
168}
169
170#[doc = include_str!("maintenance_permit.md")]
171#[must_use = "dropping the permit consumes its admitted maintenance transition"]
172pub struct MaintenancePermit<'turn, 'd> {
173    _turn: marker::PhantomData<&'turn mut ()>,
174    _driver: marker::PhantomData<fn(&'d ()) -> &'d ()>,
175}
176
177impl<'turn, 'd> Turn<'turn, 'd> {
178    pub fn reborrow(&self) -> Turn<'_, 'd> {
179        Turn {
180            work: self.work,
181            _driver: marker::PhantomData,
182        }
183    }
184
185    pub fn application(self) -> Application<'turn, 'd> {
186        Application {
187            remaining: &self.work.application,
188            _driver: marker::PhantomData,
189        }
190    }
191
192    #[doc(hidden)]
193    pub fn coordination(self) -> Coordination<'turn, 'd> {
194        Coordination {
195            remaining: &self.work.coordination,
196            _driver: marker::PhantomData,
197        }
198    }
199
200    pub fn maintenance(self) -> Maintenance<'turn, 'd> {
201        Maintenance {
202            work: self.work,
203            _driver: marker::PhantomData,
204        }
205    }
206
207    #[doc(hidden)]
208    pub fn timers(self) -> Timers<'turn, 'd> {
209        Timers {
210            remaining: &self.work.timers,
211            _driver: marker::PhantomData,
212        }
213    }
214
215    pub fn drain_ready(
216        self,
217        driver: driver::Reference<'d>,
218        limit: usize,
219        activate: impl FnMut(route::Token),
220    ) -> usize {
221        let mut budget = Budget::from_ready(self, limit);
222        ready::Access::with(&driver, |access| {
223            driver.ready().arena().drain(access, &mut budget, activate)
224        })
225    }
226}
227
228impl<'turn, 'd> Reactor<'turn, 'd> {
229    pub(crate) fn remaining(&self) -> usize {
230        self.quota.remaining()
231    }
232
233    pub(crate) const fn cursor(&self) -> usize {
234        self.cursor as usize
235    }
236
237    pub(crate) fn budget<Lane>(&self, limit: usize) -> Budget<'_, 'd, Lane> {
238        self.quota.reserve_budget(limit)
239    }
240}
241
242impl Application<'_, '_> {
243    pub fn remaining(self) -> usize {
244        self.remaining.remaining()
245    }
246
247    /// Spends one credit for work completed directly in this call.
248    /// Use [`Application::permit`] when the admitted transition is delegated
249    /// to another API so the credit cannot be duplicated before it is used.
250    pub fn take(self) -> bool {
251        self.remaining.take()
252    }
253
254    pub fn admit_with<T>(self, acquire: impl FnOnce() -> Option<T>) -> Admission<T> {
255        match self.remaining.admit_with(acquire) {
256            quota::Admission::Item(value) => Admission::Item(value),
257            quota::Admission::Empty => Admission::Empty,
258            quota::Admission::Exhausted => Admission::Exhausted,
259        }
260    }
261}
262
263impl<'turn, 'd> Application<'turn, 'd> {
264    /// Spends one application credit and returns its linear proof.
265    pub fn permit(self) -> Option<ApplicationPermit<'turn, 'd>> {
266        self.take().then_some(ApplicationPermit {
267            _turn: marker::PhantomData,
268            _driver: marker::PhantomData,
269        })
270    }
271
272    pub fn admit_next<I: set::DenseIndex>(
273        self,
274        ready: &mut set::Drain<'_, I>,
275    ) -> ApplicationAdmission<'turn, 'd, I> {
276        match ready.next_with_quota(self.remaining) {
277            set::Next::Item(value) => ApplicationAdmission::Item(
278                value,
279                ApplicationPermit {
280                    _turn: marker::PhantomData,
281                    _driver: marker::PhantomData,
282                },
283            ),
284            set::Next::Empty => ApplicationAdmission::Empty,
285            set::Next::Exhausted(value) => ApplicationAdmission::Exhausted(value),
286        }
287    }
288}
289
290impl Coordination<'_, '_> {
291    #[doc(hidden)]
292    pub fn take(&mut self) -> bool {
293        self.remaining.take()
294    }
295}
296
297impl<'turn, 'd> Timers<'turn, 'd> {
298    pub fn remaining(self) -> usize {
299        self.remaining.remaining()
300    }
301
302    pub fn take(self) -> bool {
303        self.remaining.take()
304    }
305}
306
307impl<'turn, 'd> Maintenance<'turn, 'd> {
308    pub fn remaining(self) -> usize {
309        self.work.maintenance.remaining()
310    }
311
312    pub fn take(self) -> bool {
313        self.work.maintenance.take()
314    }
315
316    pub fn share<const PARTICIPANTS: usize>(self) -> Share<'turn, 'd, PARTICIPANTS> {
317        const { assert!(PARTICIPANTS != 0) };
318        if PARTICIPANTS == 1 {
319            return Share {
320                _reserve: credits::Quota::from_maintenance(self, 0),
321                _participants: marker::PhantomData,
322            };
323        }
324        let total = self.work.maintenance.remaining();
325        let quota = total.div_ceil(PARTICIPANTS);
326        let reserve = total - quota;
327        Share {
328            _reserve: credits::Quota::from_maintenance(self, reserve),
329            _participants: marker::PhantomData,
330        }
331    }
332
333    /// Returns the persistent half-turn quota for one maintenance producer.
334    /// Re-entering in the same turn resumes the original quota rather than
335    /// taking half of the remaining global budget again.
336    pub fn half(self) -> Half<'turn, 'd> {
337        if self.work.maintenance_half.get().is_none() {
338            self.work
339                .maintenance_half
340                .set(Some(self.work.maintenance.remaining().div_ceil(2)));
341        }
342        Half {
343            work: self.work,
344            _driver: marker::PhantomData,
345        }
346    }
347}
348
349impl Half<'_, '_> {
350    pub fn remaining(self) -> usize {
351        match self.work.maintenance_half.get() {
352            Some(quota) => self.work.maintenance.remaining().min(quota),
353            None => self.work.maintenance.remaining(),
354        }
355    }
356
357    pub fn take(self) -> bool {
358        let Some(quota) = self.work.maintenance_half.get() else {
359            return false;
360        };
361        if quota == 0 || self.work.maintenance.remaining() == 0 {
362            return false;
363        }
364        self.work.maintenance_half.set(Some(quota - 1));
365        let admitted = self.work.maintenance.take();
366        debug_assert!(admitted);
367        admitted
368    }
369}
370
371impl<'turn, 'd> MaintenancePermit<'turn, 'd> {
372    pub fn try_take(work: Maintenance<'turn, 'd>) -> Option<Self> {
373        if !work.take() {
374            return None;
375        }
376        Some(Self {
377            _turn: marker::PhantomData,
378            _driver: marker::PhantomData,
379        })
380    }
381
382    /// Admits one transition and its matching region under one turn borrow.
383    #[doc(hidden)]
384    pub fn try_take_with_region<'context, 'access>(
385        work: Maintenance<'turn, 'd>,
386        context: &'access mut driver::Context<'context, 'd>,
387    ) -> Option<(Self, &'access mut region::Token<'d>)> {
388        if !work.take() {
389            return None;
390        }
391        Some((
392            Self {
393                _turn: marker::PhantomData,
394                _driver: marker::PhantomData,
395            },
396            context.region,
397        ))
398    }
399}
400
401impl Work {
402    pub(crate) fn new() -> Self {
403        Self {
404            reactor: quota::Ledger::new(MAX_TURN_WORK_BUDGET),
405            reactor_cursor: cell::Cell::new(0),
406            ready: quota::Ledger::new(MAX_TURN_WORK_BUDGET),
407            coordination: quota::Ledger::new(MAX_TURN_WORK_BUDGET),
408            application: quota::Ledger::new(MAX_TURN_WORK_BUDGET),
409            timers: quota::Ledger::new(MAX_TURN_WORK_BUDGET),
410            maintenance: quota::Ledger::new(MAX_TURN_WORK_BUDGET),
411            maintenance_half: cell::Cell::new(None),
412        }
413    }
414
415    fn set(&mut self, value: usize) {
416        self.reactor.reset(value);
417        self.ready.reset(value);
418        self.coordination.reset(value);
419        self.application.reset(value);
420        self.timers.reset(value);
421        self.maintenance.reset(value);
422        self.maintenance_half.set(None);
423    }
424
425    fn exhausted(&self) -> bool {
426        self.reactor.remaining() == 0
427            || self.ready.remaining() == 0
428            || self.coordination.remaining() == 0
429            || self.application.remaining() == 0
430            || self.timers.remaining() == 0
431            || self.maintenance.remaining() == 0
432    }
433}
434
435impl<'scope, 'd> Controller<'scope, 'd> {
436    pub(in crate::driver) fn new(driver: driver::Reference<'d>, work: &'scope mut Work) -> Self {
437        Self { driver, work }
438    }
439
440    pub fn begin(&mut self, limit: usize) -> ActiveTurn<'_, 'd> {
441        self.work.set(limit.min(MAX_TURN_WORK_BUDGET));
442        ActiveTurn {
443            driver: self.driver,
444            work: &mut *self.work,
445        }
446    }
447}
448
449impl<'turn, 'd> ActiveTurn<'turn, 'd> {
450    pub fn turn(&self) -> Turn<'_, 'd> {
451        Turn {
452            work: &*self.work,
453            _driver: marker::PhantomData,
454        }
455    }
456
457    #[doc(hidden)]
458    pub fn refresh_clock(&self) -> time::Instant {
459        let now = time::Instant::now();
460        self.driver.scheduler().set_turn_now(now);
461        now
462    }
463
464    #[doc(hidden)]
465    pub fn reactor(&mut self) -> Reactor<'_, 'd> {
466        self.reactor_with_turn().0
467    }
468
469    /// Splits the reactor's linear quota from a copyable view of the other
470    /// turn lanes so source-owned completions can be dispatched immediately.
471    #[doc(hidden)]
472    pub fn reactor_with_turn(&mut self) -> (Reactor<'_, 'd>, Turn<'_, 'd>) {
473        let work = &*self.work;
474        let cursor = work.reactor_cursor.get();
475        work.reactor_cursor.set((cursor + 1) % REACTOR_LANES);
476        (
477            Reactor {
478                quota: credits::Quota::from_reactor(&work.reactor),
479                cursor,
480                _turn: marker::PhantomData,
481                _driver: marker::PhantomData,
482            },
483            Turn {
484                work,
485                _driver: marker::PhantomData,
486            },
487        )
488    }
489
490    pub fn exhausted(&self) -> bool {
491        self.work.exhausted()
492    }
493
494    pub fn drain_ready(&mut self, limit: usize, activate: impl FnMut(route::Token)) -> usize {
495        self.turn().drain_ready(self.driver, limit, activate)
496    }
497}
dope_core/driver/schedule/mod.rs

dope_core/driver/schedule/
mod.rs
