sark_core/manifold/schedule/
mod.rs

1//! Allocation-free scheduling over carrier-owned stream records.
2
3pub mod data;
4pub mod typed;
5
6mod drive;
7mod occupied;
8mod policy;
9mod records;
10mod runtime;
11mod vacant;
12
13use std::{convert::Infallible, marker};
14
15use o3::{
16    base::cell::region,
17    depot::{array::Index, map::direct},
18};
19pub use occupied::Occupied;
20pub use records::Records;
21pub use runtime::Runtime;
22pub use vacant::Vacant;
23
24use crate::manifold;
25use drive::TurnOutcome;
26
27const LANES: usize = 7;
28const CONTROL_READY: Index<LANES> = Index::at::<0>();
29const DATA_READY: Index<LANES> = Index::at::<1>();
30const CONNECTION_WAIT: Index<LANES> = Index::at::<2>();
31const STREAM_WAIT: Index<LANES> = Index::at::<3>();
32const CONTROL_EGRESS_WAIT: Index<LANES> = Index::at::<4>();
33const CONTROL_CONNECTION_WAIT: Index<LANES> = Index::at::<5>();
34const CONTROL_STREAM_WAIT: Index<LANES> = Index::at::<6>();
35const ALL: [Index<LANES>; LANES] = [
36    CONTROL_READY,
37    DATA_READY,
38    CONNECTION_WAIT,
39    STREAM_WAIT,
40    CONTROL_EGRESS_WAIT,
41    CONTROL_CONNECTION_WAIT,
42    CONTROL_STREAM_WAIT,
43];
44
45/// Physical resource that owns a blocked stream's next wakeup.
46#[derive(Clone, Copy, PartialEq, Eq)]
47pub enum Wait {
48    Connection,
49    Stream,
50    Egress,
51}
52
53/// Static classification of one carrier-native blocked result.
54pub trait Block: Copy + Eq {
55    fn wait(self) -> Wait;
56}
57
58/// A carrier wakeup delivered to its resident scheduling domain.
59#[derive(Clone, Copy, PartialEq, Eq)]
60pub enum Wake<K> {
61    Connection,
62    Stream(K),
63}
64
65/// Whether owned work still prevents retirement.
66#[derive(Clone, Copy, PartialEq, Eq)]
67#[repr(u8)]
68pub enum Quiescence {
69    Busy,
70    Quiet,
71}
72
73const _: () = {
74    assert!(core::mem::size_of::<Quiescence>() == core::mem::size_of::<bool>());
75    assert!(core::mem::align_of::<Quiescence>() == core::mem::align_of::<bool>());
76};
77
78/// Whether this turn released carrier egress capacity.
79#[derive(Clone, Copy, PartialEq, Eq)]
80pub enum Egress {
81    Unchanged,
82    Released,
83}
84
85const _: () = {
86    assert!(core::mem::size_of::<Egress>() == core::mem::size_of::<bool>());
87    assert!(core::mem::align_of::<Egress>() == core::mem::align_of::<bool>());
88};
89
90/// Result of one bounded egress turn.
91#[derive(Clone, Copy, PartialEq, Eq)]
92#[must_use]
93pub enum Turn<B> {
94    Complete,
95    Progress,
96    Blocked(B),
97    Idle,
98}
99
100/// The statically selected kind of a stream quantum.
101#[derive(Clone, Copy, PartialEq, Eq)]
102pub enum WorkClass {
103    Control,
104    Data,
105}
106
107impl WorkClass {
108    const fn ready_lane(self) -> Index<LANES> {
109        match self {
110            Self::Control => CONTROL_READY,
111            Self::Data => DATA_READY,
112        }
113    }
114
115    const fn blocked_lane(self, blocked: Option<Wait>) -> Index<LANES> {
116        match (blocked, self) {
117            (None, class) => class.ready_lane(),
118            (Some(Wait::Egress), Self::Control) => CONTROL_EGRESS_WAIT,
119            (Some(Wait::Egress), Self::Data) => DATA_READY,
120            (Some(Wait::Connection), Self::Control) => CONTROL_CONNECTION_WAIT,
121            (Some(Wait::Connection), Self::Data) => CONNECTION_WAIT,
122            (Some(Wait::Stream), Self::Control) => CONTROL_STREAM_WAIT,
123            (Some(Wait::Stream), Self::Data) => STREAM_WAIT,
124        }
125    }
126}
127
128/// One control quantum and its complete scheduling continuation.
129#[derive(Clone, Copy, PartialEq, Eq)]
130pub struct ControlTurn<B> {
131    step: Step<B>,
132    activity: WorkClass,
133}
134
135impl<B> ControlTurn<B> {
136    pub const fn progress(next: WorkClass) -> Self {
137        Self {
138            step: Step::Progress(next),
139            activity: WorkClass::Control,
140        }
141    }
142
143    pub const fn blocked(block: B) -> Self {
144        Self {
145            step: Step::Blocked(block),
146            activity: WorkClass::Control,
147        }
148    }
149
150    pub const fn complete() -> Self {
151        Self {
152            step: Step::Complete,
153            activity: WorkClass::Control,
154        }
155    }
156
157    pub const fn idle() -> Self {
158        Self {
159            step: Step::Idle,
160            activity: WorkClass::Control,
161        }
162    }
163
164    fn into_parts(self) -> (Step<B>, WorkClass) {
165        (self.step, self.activity)
166    }
167}
168
169impl<B: Copy> ControlTurn<B> {
170    pub const fn new(egress: Turn<B>, next: WorkClass) -> Self {
171        Self::from_turn(egress, next, WorkClass::Control)
172    }
173
174    pub const fn fused(egress: Turn<B>, next: WorkClass) -> Self {
175        Self::from_turn(egress, next, WorkClass::Data)
176    }
177
178    const fn from_turn(egress: Turn<B>, next: WorkClass, activity: WorkClass) -> Self {
179        Self {
180            step: DataTurn::new(egress, next),
181            activity,
182        }
183    }
184}
185
186/// One data quantum and its complete scheduling continuation.
187#[derive(Clone, Copy, PartialEq, Eq)]
188pub enum DataTurn<B> {
189    Complete,
190    Progress(WorkClass),
191    Blocked(B),
192    Idle,
193}
194
195impl<B: Copy> DataTurn<B> {
196    pub const fn new(egress: Turn<B>, next: WorkClass) -> Self {
197        match egress {
198            Turn::Complete => Self::Complete,
199            Turn::Progress => Self::Progress(next),
200            Turn::Blocked(block) => Self::Blocked(block),
201            Turn::Idle => Self::Idle,
202        }
203    }
204}
205
206/// The same scheduling steps as [`DataTurn`], named for the control quantum.
207type Step<B> = DataTurn<B>;
208
209/// Result of a control quantum that consumed its resident value.
210pub enum Control<T, B, E = Infallible> {
211    Retain { value: T, turn: ControlTurn<B> },
212    Failed { value: T, error: E },
213    Complete,
214    Remove,
215}
216
217/// Exact storage destination after a terminal event.
218pub enum Settlement<T> {
219    Keep(T),
220    Remove,
221}
222
223/// The terminal event delivered to a record owner.
224pub enum Terminal<E> {
225    Complete,
226    Idle,
227    Failed(E),
228}
229
230/// Whether a successful record transaction also makes it runnable.
231#[derive(Clone, Copy, Default, PartialEq, Eq)]
232pub enum Activation<B> {
233    #[default]
234    Keep,
235    Activate {
236        blocked: Option<B>,
237        class: WorkClass,
238    },
239}
240
241impl<B> Activation<B> {
242    pub const fn data(blocked: Option<B>) -> Self {
243        Self::Activate {
244            blocked,
245            class: WorkClass::Data,
246        }
247    }
248
249    pub const fn control(blocked: Option<B>) -> Self {
250        Self::Activate {
251            blocked,
252            class: WorkClass::Control,
253        }
254    }
255}
256
257/// Failure to update one exact resident stream.
258#[derive(Clone, Copy, PartialEq, Eq, Debug)]
259pub enum Error<E> {
260    Missing,
261    Rejected(E),
262}
263
264/// Terminal policy for one payload-specific resident state.
265pub trait Settle<T> {
266    type Key: manifold::raw::Key;
267    type Error;
268
269    fn settle(
270        &mut self,
271        stream: Self::Key,
272        value: T,
273        terminal: Terminal<Self::Error>,
274    ) -> Settlement<T>;
275}
276
277#[repr(transparent)]
278pub struct Decision<'id, K> {
279    outcome: TurnOutcome<K>,
280    _scope: marker::PhantomData<region::Identity<'id>>,
281}
282
283impl<'id, K> Decision<'id, K> {
284    fn bind(outcome: TurnOutcome<K>, _scope: region::Identity<'id>) -> Self {
285        Self {
286            outcome,
287            _scope: marker::PhantomData,
288        }
289    }
290
291    const fn running(stream: K, activity: WorkClass) -> TurnOutcome<K> {
292        TurnOutcome::new(stream, activity, drive::Effect::None)
293    }
294
295    const fn blocked(stream: K, activity: WorkClass) -> TurnOutcome<K> {
296        TurnOutcome::new(stream, activity, drive::Effect::BlockData)
297    }
298}
299
300/// Payload-specific state machine driven inside one resident domain.
301pub trait Driver<T>: Settle<T> {
302    type Block: Block;
303    type Permit;
304
305    fn permit(&mut self) -> Option<Self::Permit>;
306
307    fn control<'id>(
308        &mut self,
309        permit: Self::Permit,
310        entry: Occupied<'_, Self::Key, T, Self::Block, region::Identity<'id>>,
311    ) -> Decision<'id, Self::Key>;
312
313    fn data(
314        &mut self,
315        permit: Self::Permit,
316        stream: Self::Key,
317        value: &mut T,
318    ) -> Result<DataTurn<Self::Block>, Self::Error>;
319}
320
321/// Exact amount of scheduler work performed by one drive.
322#[derive(Clone, Copy, Default, PartialEq, Eq)]
323pub struct Report {
324    pub turns: usize,
325    pub completed: usize,
326}
327
328/// Admission commits the resident owner or returns its exact rejection.
329pub type InsertResult<K, T> = Result<(), Insert<K, T>>;
330
331/// Exact rejection from one bounded resident-domain insertion.
332#[must_use = "a rejected resident owner must be handled"]
333#[derive(Debug)]
334pub enum Insert<K, T> {
335    Full {
336        stream: K,
337        value: T,
338        used: usize,
339        limit: usize,
340    },
341    Occupied {
342        stream: K,
343        stored: K,
344        value: T,
345    },
346    Excluded {
347        stream: K,
348        value: T,
349        index: usize,
350        capacity: usize,
351    },
352}
353
354impl<K: manifold::raw::Key, T> Insert<K, T> {
355    pub(super) fn from_direct(error: direct::Error<K>, value: T) -> Self {
356        match error {
357            direct::Error::Outside {
358                key,
359                index,
360                capacity,
361            } => Self::Excluded {
362                stream: key,
363                value,
364                index,
365                capacity,
366            },
367            direct::Error::Occupied { key, stored } => Self::Occupied {
368                stream: key,
369                stored,
370                value,
371            },
372        }
373    }
374}
sark_core/manifold/schedule/mod.rs

sark_core/manifold/schedule/
mod.rs
