sark_core/deadline/
mod.rs

1//! Exact, protocol-neutral exchange deadline scheduling.
2
3use core::cmp;
4use std::mem;
5
6use o3::{
7    base::time::Positive,
8    depot::{
9        capacity,
10        heap::{self, resident},
11    },
12};
13
14use crate::manifold;
15
16pub mod config;
17mod exact;
18pub mod timeline;
19pub use exact::Exact;
20/// The exchange phase whose progress authority expires.
21#[repr(u8)]
22#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
23pub enum Phase {
24    Head,
25    BodyIdle,
26    Total,
27    Handler,
28    ResponseAdmission,
29}
30
31impl Phase {
32    const fn rank(self) -> u8 {
33        match self {
34            Self::Total => 0,
35            _ => self as u8 + 1,
36        }
37    }
38
39    const fn lane(self) -> usize {
40        match self {
41            Self::Total => 0,
42            Self::Head | Self::BodyIdle => 1,
43            Self::Handler | Self::ResponseAdmission => 2,
44        }
45    }
46}
47
48/// A positive, non-expiring exchange activation.
49/// Wakes carry no [`Phase`]: runnable is not expired authority.
50#[repr(u8)]
51#[derive(Clone, Copy, Eq, Ord, PartialEq, PartialOrd)]
52pub enum WakeReason {
53    DeferredOutput,
54}
55
56/// The semantic action selected when an exchange-local instant becomes due.
57#[derive(Clone, Copy, Eq, PartialEq)]
58pub enum Action<W = WakeReason> {
59    Expire(Phase),
60    Wake(W),
61}
62
63impl<W: Ord> Ord for Action<W> {
64    fn cmp(&self, other: &Self) -> cmp::Ordering {
65        match (self, other) {
66            (Self::Expire(left), Self::Expire(right)) => left.rank().cmp(&right.rank()),
67            (Self::Expire(_), Self::Wake(_)) => cmp::Ordering::Less,
68            (Self::Wake(_), Self::Expire(_)) => cmp::Ordering::Greater,
69            (Self::Wake(left), Self::Wake(right)) => left.cmp(right),
70        }
71    }
72}
73
74impl<W: Ord> PartialOrd for Action<W> {
75    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
76        Some(self.cmp(other))
77    }
78}
79
80/// One effective phase deadline for an exact exchange identity.
81#[derive(Clone, Copy, Eq, PartialEq)]
82pub struct Scheduled<P, W = WakeReason> {
83    pub at: P,
84    pub action: Action<W>,
85}
86
87impl<P, W> Scheduled<P, W> {
88    pub const fn expire(at: P, phase: Phase) -> Self {
89        Self {
90            at,
91            action: Action::Expire(phase),
92        }
93    }
94
95    pub const fn wake(at: P, reason: W) -> Self {
96        Self {
97            at,
98            action: Action::Wake(reason),
99        }
100    }
101}
102
103impl<P: Ord, W: Ord> Ord for Scheduled<P, W> {
104    fn cmp(&self, other: &Self) -> cmp::Ordering {
105        self.at
106            .cmp(&other.at)
107            .then_with(|| self.action.cmp(&other.action))
108    }
109}
110
111impl<P: Ord, W: Ord> PartialOrd for Scheduled<P, W> {
112    fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
113        Some(self.cmp(other))
114    }
115}
116
117/// Effect of one committed book mutation on its externally armed minimum.
118#[derive(Clone, Copy, Eq, PartialEq)]
119pub enum Mutation {
120    Unchanged,
121    Contents,
122    Earliest,
123}
124
125impl Mutation {
126    pub const fn is_earliest(self) -> bool {
127        matches!(self, Self::Earliest)
128    }
129
130    fn between<P: PartialEq>(before: Option<P>, after: Option<P>) -> Self {
131        if before == after {
132            Self::Contents
133        } else {
134            Self::Earliest
135        }
136    }
137}
138
139impl From<heap::Change> for Mutation {
140    fn from(change: heap::Change) -> Self {
141        match change {
142            heap::Change::Contents => Self::Contents,
143            heap::Change::Minimum => Self::Earliest,
144        }
145    }
146}
147
148/// A schedule request that cannot name a live slot in this fixed domain.
149#[derive(Clone, Copy, Debug, Eq, PartialEq)]
150pub enum Error<K> {
151    OutOfRange { key: K, capacity: usize },
152    Conflict { requested: K, current: K },
153}
154
155/// One due entry removed from the exact book.
156#[derive(Clone, Copy, Eq, PartialEq)]
157pub struct Expired<K, P, W = WakeReason> {
158    pub key: K,
159    pub scheduled: Scheduled<P, W>,
160}
161
162/// Result of visiting at most one due entry in a deadline book.
163#[derive(Clone, Copy, Debug, Eq, PartialEq)]
164#[must_use = "account for renewal work and preserve the current minimum"]
165pub enum Poll<T> {
166    Idle,
167    /// An idle deadline was renewed. Account for this work before polling again,
168    /// and preserve or rearm the book's current minimum.
169    Renewed,
170    Expired(T),
171}
172
173/// Fixed-capacity deadline storage.
174pub trait Book<K: manifold::raw::Key, P: Copy + Ord, W: Copy + Ord = WakeReason> {
175    fn schedule(&mut self, key: K, item: Scheduled<P, W>) -> Result<(), Error<K>>;
176    fn cancel(&mut self, key: K) -> Mutation;
177    fn cancel_one(&mut self, key: K, action: Action<W>) -> Mutation;
178    fn peek(&self) -> Option<(K, Scheduled<P, W>)>;
179    fn poll_due(&mut self, now: P) -> Poll<Expired<K, P, W>>;
180    /// Drains lazy renewals until an expiry or an idle book is reached.
181    /// This may visit every resident entry; event-loop turns use `poll_due` to
182    /// bound renewal work before yielding to other connections.
183    fn pop_due(&mut self, now: P) -> Option<Expired<K, P, W>> {
184        loop {
185            match self.poll_due(now) {
186                Poll::Idle => return None,
187                Poll::Renewed => {}
188                Poll::Expired(item) => return Some(item),
189            }
190        }
191    }
192    fn len(&self) -> usize;
193    fn is_empty(&self) -> bool {
194        self.len() == 0
195    }
196}
197
198/// One-allocation total, ingress, work, and wake agenda. Separate ingress and
199/// work lanes allow duplex body receipt and application work to overlap.
200pub struct Agenda<K: manifold::raw::Key, P: Copy + Ord, W: Copy + Ord = WakeReason> {
201    heap: resident::Min<K, Scheduled<P, W>, Lane<P, W>>,
202}
203
204#[derive(Clone, Copy)]
205struct Lane<P, W> {
206    phases: timeline::Timeline<P>,
207    wake: Option<Wake<P, W>>,
208}
209
210#[derive(Clone, Copy, Eq, PartialEq)]
211struct Wake<P, W = WakeReason> {
212    at: P,
213    reason: W,
214}
215
216const _: () = {
217    assert!(mem::size_of::<Wake<u64>>() <= mem::size_of::<Scheduled<u64>>());
218    assert!(mem::align_of::<Wake<u64>>() == mem::align_of::<Scheduled<u64>>());
219};
220
221enum Plan<'a, K: manifold::raw::Key, P: Copy + Ord, W: Copy + Ord> {
222    Vacant(resident::Vacant<'a, K, Scheduled<P, W>, Lane<P, W>>),
223    Occupied(resident::Occupied<'a, K, Scheduled<P, W>, Lane<P, W>>),
224}
225
226impl<P: Copy + Ord, W: Copy + Ord> Lane<P, W> {
227    const EMPTY: Self = Self {
228        phases: timeline::Timeline::EMPTY,
229        wake: None,
230    };
231
232    fn get(&self, action: Action<W>) -> Option<Scheduled<P, W>> {
233        self.resident(action).filter(|item| item.action == action)
234    }
235
236    fn resident(&self, action: Action<W>) -> Option<Scheduled<P, W>> {
237        match action {
238            Action::Expire(phase) => self
239                .phases
240                .resident(phase)
241                .map(|item| Scheduled::expire(item.at, item.phase)),
242            Action::Wake(_) => self.wake.map(|item| Scheduled::wake(item.at, item.reason)),
243        }
244    }
245
246    /// Installs `item` as this lane's resident for its action and returns the lane's
247    /// resulting minimum. The installed item is already included in that minimum.
248    fn set(&mut self, item: Scheduled<P, W>) -> Scheduled<P, W> {
249        match item.action {
250            Action::Expire(phase) => {
251                self.phases.set(item.at, phase);
252            }
253            Action::Wake(reason) => {
254                self.wake = Some(Wake {
255                    at: item.at,
256                    reason,
257                });
258            }
259        };
260        self.next().unwrap_or(item)
261    }
262
263    fn retire(&mut self, action: Action<W>) {
264        match action {
265            Action::Expire(phase) => self.phases.clear_phase(phase),
266            Action::Wake(reason) => {
267                if self.wake.is_some_and(|item| item.reason == reason) {
268                    self.wake = None;
269                }
270            }
271        }
272    }
273
274    fn next(&self) -> Option<Scheduled<P, W>> {
275        self.phases
276            .next()
277            .map(|item| Scheduled::expire(item.at, item.phase))
278            .into_iter()
279            .chain(self.wake.map(|item| Scheduled::wake(item.at, item.reason)))
280            .min()
281    }
282
283    fn clear_wake(&mut self) -> Mutation {
284        if self.wake.is_none() {
285            return Mutation::Unchanged;
286        }
287        self.wake = None;
288        Mutation::Contents
289    }
290
291    fn transition<N, F>(
292        &mut self,
293        config: config::Config,
294        change: timeline::Transition,
295        now: N,
296        add: F,
297    ) -> Mutation
298    where
299        N: Copy,
300        F: FnMut(N, Positive) -> P,
301    {
302        let before = self.next();
303        let phases = self.phases.apply(config, change, now, add);
304        let wake = if change == timeline::Transition::Retired {
305            self.clear_wake()
306        } else {
307            Mutation::Unchanged
308        };
309        if phases == timeline::Change::Unchanged && wake == Mutation::Unchanged {
310            return Mutation::Unchanged;
311        }
312        Mutation::between(before, self.next())
313    }
314}
315
316impl<K: manifold::raw::Key, P: Copy + Ord, W: Copy + Ord> Agenda<K, P, W> {
317    fn try_entry(&mut self, key: K) -> Result<Plan<'_, K, P, W>, Error<K>> {
318        match self.heap.entry(key) {
319            Err(error) => Err(Error::OutOfRange {
320                key: error.key(),
321                capacity: error.capacity(),
322            }),
323            Ok(resident::Entry::Vacant(entry)) => Ok(Plan::Vacant(entry)),
324            Ok(resident::Entry::Occupied(entry)) => Ok(Plan::Occupied(entry)),
325            Ok(resident::Entry::Conflict(entry)) => Err(Error::Conflict {
326                requested: entry.requested(),
327                current: entry.current(),
328            }),
329        }
330    }
331
332    pub fn try_new(capacity: capacity::Count) -> Result<Self, heap::Error> {
333        Ok(Self {
334            heap: resident::Min::try_repeat(capacity, Lane::EMPTY)?,
335        })
336    }
337
338    pub fn transition<N, F>(
339        &mut self,
340        key: K,
341        config: config::Config,
342        change: timeline::Transition,
343        now: N,
344        add: F,
345    ) -> Result<(), Error<K>>
346    where
347        N: Copy,
348        F: FnMut(N, Positive) -> P,
349    {
350        if change == timeline::Transition::BodyProgress {
351            let Plan::Occupied(mut entry) = self.try_entry(key)? else {
352                return Ok(());
353            };
354            entry.resident_mut().phases.progress(config, now, add);
355            return Ok(());
356        }
357        let plan = self.try_entry(key)?;
358        match plan {
359            Plan::Vacant(mut entry) => {
360                let mut lane = Lane::EMPTY;
361                if lane.transition(config, change, now, add) == Mutation::Unchanged {
362                    return Ok(());
363                }
364                let Some(next) = lane.next() else {
365                    return Ok(());
366                };
367                *entry.resident_mut() = lane;
368                entry.insert(next);
369            }
370            Plan::Occupied(mut entry) => {
371                let mutation = entry.resident_mut().transition(config, change, now, add);
372                if mutation != Mutation::Earliest {
373                    return Ok(());
374                }
375                match entry.resident().next() {
376                    Some(next) => entry.set(next),
377                    None => entry.retire(),
378                }
379            }
380        }
381        Ok(())
382    }
383}
384
385impl<K: manifold::raw::Key, P: Copy + Ord, W: Copy + Ord> Book<K, P, W> for Agenda<K, P, W> {
386    fn schedule(&mut self, key: K, item: Scheduled<P, W>) -> Result<(), Error<K>> {
387        let plan = self.try_entry(key)?;
388        match plan {
389            Plan::Vacant(mut entry) => {
390                let mut lane = Lane::EMPTY;
391                let minimum = lane.set(item);
392                *entry.resident_mut() = lane;
393                entry.insert(minimum);
394            }
395            Plan::Occupied(mut entry) => {
396                if entry.resident().resident(item.action) == Some(item) {
397                    return Ok(());
398                }
399                let minimum = entry.resident_mut().set(item);
400                entry.set(minimum);
401            }
402        }
403        Ok(())
404    }
405
406    fn cancel(&mut self, key: K) -> Mutation {
407        let heap::Probe::Occupied(entry) = self.heap.probe(key) else {
408            return Mutation::Unchanged;
409        };
410        entry.retire_min().into()
411    }
412
413    fn cancel_one(&mut self, key: K, action: Action<W>) -> Mutation {
414        let heap::Probe::Occupied(mut entry) = self.heap.probe(key) else {
415            return Mutation::Unchanged;
416        };
417        if entry.resident().get(action).is_none() {
418            return Mutation::Unchanged;
419        }
420        let next = {
421            let lane = entry.resident_mut();
422            lane.retire(action);
423            lane.next()
424        };
425        match next {
426            Some(next) => entry.set_min(next).into(),
427            None => entry.retire_min().into(),
428        }
429    }
430
431    fn peek(&self) -> Option<(K, Scheduled<P, W>)> {
432        self.heap.peek().map(|(key, item)| (key, *item))
433    }
434
435    fn poll_due(&mut self, now: P) -> Poll<Expired<K, P, W>> {
436        let Some(mut entry) = self.heap.first_entry_if(|priority| !(priority.at > now)) else {
437            return Poll::Idle;
438        };
439        let key = entry.key();
440        let item = *entry.priority();
441        let (next, renewed) = {
442            let lane = entry.resident_mut();
443            let renewed = item.action == Action::Expire(Phase::BodyIdle)
444                && lane.phases.refresh(item.at) == timeline::Change::Changed;
445            if !renewed {
446                lane.retire(item.action);
447            }
448            (lane.next(), renewed)
449        };
450        match next {
451            Some(next) => entry.set(next),
452            None => entry.retire(),
453        }
454        if renewed {
455            Poll::Renewed
456        } else {
457            Poll::Expired(Expired {
458                key,
459                scheduled: item,
460            })
461        }
462    }
463
464    fn len(&self) -> usize {
465        self.heap.len()
466    }
467}
sark_core/deadline/mod.rs

sark_core/deadline/
mod.rs
