dope_core/driver/schedule/ready/
mod.rs

1#![doc = include_str!("compile_fail.md")]
2
3use std::{io, marker, mem};
4
5use o3::{
6    self, cell,
7    collections::{self, batch::set},
8};
9
10use crate::{
11    driver::{
12        self,
13        route::{self, kind},
14        schedule::{self, credits},
15        settings,
16    },
17    io::fd::handles,
18};
19
20pub(in crate::driver) mod credit;
21mod fixed;
22mod handle;
23#[doc(hidden)]
24pub mod raw;
25mod sealed;
26mod waiters;
27pub use fixed::identity::FixedIdentity;
28pub use handle::Handle;
29pub(in crate::driver::schedule::ready) use sealed::{
30    Dispatch, Dynamic, Entry, FreeIndex, FreeLink, Kind, Layout, Pool, Reservation, Resolved,
31    Slots, Table,
32};
33pub(super) enum Lane {}
34
35#[derive(Clone, Copy)]
36pub(in crate::driver) struct Access<'a, 'd> {
37    _access: marker::PhantomData<fn(&'a ()) -> &'a ()>,
38    _driver: marker::PhantomData<fn(driver::Reference<'d>) -> driver::Reference<'d>>,
39}
40
41impl<'d> Access<'_, 'd> {
42    pub(in crate::driver) fn with<'a, R>(
43        _driver: &'a driver::Reference<'d>,
44        f: impl FnOnce(Access<'a, 'd>) -> R,
45    ) -> R
46    where
47        'd: 'a,
48    {
49        f(Access {
50            _access: marker::PhantomData,
51            _driver: marker::PhantomData,
52        })
53    }
54}
55
56const _: () = assert!(mem::size_of::<Access<'static, 'static>>() == 0);
57
58pub(in crate::driver::schedule::ready) enum TaskParent<'d> {
59    Root,
60    Task(completion::Wake<'d>),
61    Stale,
62}
63
64pub mod completion;
65#[doc(hidden)]
66pub mod task;
67
68/// A generation-checked address in a driver's local ready queue.
69#[derive(Clone, Copy)]
70pub struct Key<'d> {
71    index: u32,
72    epoch: u32,
73    _arena: marker::PhantomData<fn(&'d Arena) -> &'d Arena>,
74    _thread: o3::ThreadBound,
75}
76
77#[derive(Clone, Copy, PartialEq, Eq)]
78pub(crate) struct FixedKey<'d>(Key<'d>);
79
80#[repr(transparent)]
81pub(crate) struct FixedRelease<'d> {
82    slot: handles::FixedSlot,
83    _arena: marker::PhantomData<fn(&'d Arena) -> &'d Arena>,
84}
85
86#[derive(Clone, Copy)]
87pub(in crate::driver::schedule::ready) struct DynamicKey<'d> {
88    index: FreeIndex,
89    epoch: u32,
90    _arena: marker::PhantomData<fn(&'d Arena) -> &'d Arena>,
91    _thread: o3::ThreadBound,
92}
93
94impl Key<'static> {
95    pub const NONE: Self = Self {
96        index: u32::MAX,
97        epoch: 0,
98        _arena: marker::PhantomData,
99        _thread: o3::ThreadBound::NEW,
100    };
101}
102
103impl PartialEq for Key<'_> {
104    fn eq(&self, other: &Self) -> bool {
105        self.index == other.index && self.epoch == other.epoch
106    }
107}
108
109impl Eq for Key<'_> {}
110
111const _: () = {
112    assert!(mem::size_of::<FixedKey<'static>>() == mem::size_of::<Key<'static>>());
113    assert!(mem::size_of::<FixedRelease<'static>>() == mem::size_of::<handles::FixedSlot>());
114    assert!(mem::size_of::<DynamicKey<'static>>() == mem::size_of::<Key<'static>>());
115    assert!(mem::align_of::<DynamicKey<'static>>() == mem::align_of::<Key<'static>>());
116};
117
118impl<'d> DynamicKey<'d> {
119    fn new(index: FreeIndex, epoch: u32) -> Self {
120        Self {
121            index,
122            epoch,
123            _arena: marker::PhantomData,
124            _thread: o3::ThreadBound::NEW,
125        }
126    }
127
128    fn key(self) -> Key<'d> {
129        Key {
130            index: self.index.raw(),
131            epoch: self.epoch,
132            _arena: marker::PhantomData,
133            _thread: o3::ThreadBound::NEW,
134        }
135    }
136
137    fn index(self) -> FreeIndex {
138        self.index
139    }
140
141    fn epoch(self) -> u32 {
142        self.epoch
143    }
144}
145
146impl<'d> FixedKey<'d> {
147    pub(crate) fn new(index: u32, epoch: u32) -> Self {
148        Self(Key {
149            index,
150            epoch,
151            _arena: marker::PhantomData,
152            _thread: o3::ThreadBound::NEW,
153        })
154    }
155
156    pub(crate) fn key(self) -> Key<'d> {
157        self.0
158    }
159
160    pub(crate) fn index(self) -> u32 {
161        self.0.index
162    }
163
164    pub(crate) fn epoch(self) -> u32 {
165        self.0.epoch
166    }
167}
168
169impl FixedRelease<'_> {
170    pub(crate) fn slot(&self) -> handles::FixedSlot {
171        self.slot
172    }
173
174    pub(crate) fn into_slot(self) -> handles::FixedSlot {
175        self.slot
176    }
177}
178
179/// A generation-checked driver readiness target that cannot name a task node.
180#[derive(Clone, Copy)]
181pub struct Target<'d> {
182    driver: driver::Reference<'d>,
183    key: Key<'d>,
184}
185
186impl PartialEq for Target<'_> {
187    fn eq(&self, other: &Self) -> bool {
188        self.driver.same_driver(other.driver) && self.key == other.key
189    }
190}
191
192impl Eq for Target<'_> {}
193
194impl<'d> Target<'d> {
195    pub(crate) fn new(driver: driver::Reference<'d>, key: Key<'d>) -> Self {
196        Self { driver, key }
197    }
198
199    pub fn wake(self) {
200        self.driver.ready().activate_ready(self.key);
201    }
202
203    #[doc(hidden)]
204    pub fn into_key(self) -> Key<'d> {
205        self.key
206    }
207
208    pub(super) fn into_parts(self) -> (driver::Reference<'d>, Key<'d>) {
209        (self.driver, self.key)
210    }
211}
212
213const _: () = {
214    assert!(mem::size_of::<Target<'static>>() == 2 * mem::size_of::<usize>());
215};
216
217pub struct Slot<'d, Tag: route::Tag> {
218    driver: driver::Reference<'d>,
219    key: DynamicKey<'d>,
220    target: marker::PhantomData<*mut Tag>,
221}
222
223impl<'d, Tag: route::Tag> Slot<'d, Tag> {
224    fn new(driver: driver::Reference<'d>, key: DynamicKey<'d>) -> Self {
225        Self {
226            driver,
227            key,
228            target: marker::PhantomData,
229        }
230    }
231
232    pub fn set_target(&self, target: route::Operation<'d, Tag>) {
233        self.driver
234            .ready()
235            .arena()
236            .set_target(self.key.key(), target.into_token());
237    }
238
239    pub fn activate(&self) {
240        Access::with(&self.driver, |access| {
241            self.driver.ready().arena().activate(access, self.key.key());
242        });
243    }
244
245    pub fn key(&self) -> Key<'d> {
246        self.key.key()
247    }
248
249    /// Returns the driver- and generation-branded root owned by this slot.
250    pub fn target(&self) -> Target<'d> {
251        Target::new(self.driver, self.key.key())
252    }
253}
254
255impl<Tag: route::Tag> Drop for Slot<'_, Tag> {
256    fn drop(&mut self) {
257        self.driver.ready().arena().release(self.key);
258    }
259}
260
261pub(in crate::driver) struct Arena {
262    ready: set::Set,
263    entries: Table,
264    recv_credit_refs: Box<[cell::LocalRefCount]>,
265    buffer_waiters: waiters::Waiters,
266}
267
268impl Arena {
269    pub(in crate::driver) fn new(
270        fixed: usize,
271        dynamic: settings::ScheduleCapacity,
272    ) -> io::Result<Box<Self>> {
273        use collections::{BoxExt, BoxSliceExt};
274
275        use crate::driver::route::{Epoch, FRAMEWORK, SlotIndex, Token};
276        let layout = Layout::new(fixed, dynamic)?;
277        let recv_credit_refs = BoxSliceExt::try_box_with(fixed, |_| cell::LocalRefCount::empty())
278            .map_err(|error| io::Error::other(error.to_string()))?;
279
280        let dummy = Token::new(FRAMEWORK, SlotIndex::ZERO, Epoch::INITIAL);
281        let arena = Self {
282            ready: set::Set::try_with_capacity(layout.capacity())?,
283            entries: Table::try_new(layout, dummy)?,
284            recv_credit_refs,
285            buffer_waiters: waiters::Waiters::try_with_capacity(fixed)?,
286        };
287        Ok(BoxExt::try_box(arena)?)
288    }
289
290    pub(crate) fn claim_fixed(&self, slot: handles::FixedSlot) -> Option<FixedKey<'_>> {
291        use crate::driver::route::{Epoch, FRAMEWORK, SlotIndex, Token};
292
293        let index = slot.raw();
294        debug_assert!(self.entries.slots.is_fixed(index as usize));
295        let dummy = Token::new(FRAMEWORK, SlotIndex::ZERO, Epoch::INITIAL);
296        let epoch = self.entries.slots.claim_fixed(index as usize, dummy)?;
297        Some(FixedKey::new(index, epoch))
298    }
299
300    pub(crate) fn fixed_handle<'d>(
301        &'d self,
302        driver: driver::Reference<'d>,
303        key: FixedKey<'d>,
304    ) -> Handle<'d> {
305        Handle::new(driver, key)
306    }
307
308    pub(crate) fn release_fixed<'d>(&self, key: FixedKey<'d>) -> Option<FixedRelease<'d>> {
309        let transfer = self.buffer_waiters.retire(self, key);
310        let released = self
311            .entries
312            .slots
313            .release_fixed(key, |index| self.ready.remove(index));
314        if transfer {
315            self.buffer_waiters.wake(self);
316        }
317        if !released {
318            return None;
319        }
320        Some(FixedRelease {
321            slot: handles::FixedSlot::from_index(route::SlotIndex::from_bounded(key.index())),
322            _arena: marker::PhantomData,
323        })
324    }
325
326    pub(crate) fn make_slot<'d, Tag: route::Tag>(
327        &'d self,
328        driver: driver::Reference<'d>,
329        target: route::Operation<'d, Tag>,
330    ) -> io::Result<Slot<'d, Tag>> {
331        self.make_slot_reserving(driver, target, 0)
332    }
333
334    pub(crate) fn make_slots<'d, Tag, I>(
335        &'d self,
336        driver: driver::Reference<'d>,
337        targets: I,
338    ) -> io::Result<Box<[Slot<'d, Tag>]>>
339    where
340        Tag: route::Tag,
341        I: IntoIterator<Item = route::Operation<'d, Tag>>,
342        I::IntoIter: ExactSizeIterator,
343    {
344        let targets = targets.into_iter();
345        let requested = targets.len();
346        let available = self.entries.pool.available();
347        if requested > available {
348            return Err(Pool::capacity_error(requested, available));
349        }
350        let mut slots: Vec<Slot<'d, Tag>> = collections::VecExt::try_vec_with_capacity(requested)?;
351        for (index, target) in targets.enumerate() {
352            let reserve = index
353                .checked_add(1)
354                .and_then(|used| requested.checked_sub(used))
355                .ok_or_else(|| {
356                    io::Error::other("dope: ready iterator exceeded its exact length")
357                })?;
358            slots.push(self.make_slot_reserving(driver, target, reserve)?);
359        }
360        if slots.len() != requested {
361            return Err(io::Error::other(
362                "dope: ready iterator ended before its exact length",
363            ));
364        }
365        Ok(slots.into_boxed_slice())
366    }
367
368    pub(crate) fn make_slot_reserving<'d, Tag: route::Tag>(
369        &'d self,
370        driver: driver::Reference<'d>,
371        target: route::Operation<'d, Tag>,
372        reserve: usize,
373    ) -> io::Result<Slot<'d, Tag>> {
374        let key = self.entries.pool.reserve_dispatch(
375            &self.entries.slots,
376            reserve,
377            target.into_token(),
378        )?;
379        Ok(Slot::new(driver, key))
380    }
381
382    fn set_target(&self, key: Key<'_>, target: route::Token) {
383        let Some(resolved) = self.entries.slots.resolve(key) else {
384            return;
385        };
386        let Some(dispatch): Option<Dispatch<'_>> = resolved.dispatch() else {
387            return;
388        };
389        let current = dispatch.get();
390        let transfer = if current.kind() == kind::RECV_BUFFER_WAITING {
391            self.buffer_waiters.unlink_index(resolved.index());
392            false
393        } else {
394            current.kind() == kind::RECV_BUFFER_GRANTED
395        };
396        if transfer {
397            self.ready.remove(resolved.index());
398        }
399        dispatch.set(target);
400        if transfer {
401            self.buffer_waiters.wake(self);
402        }
403    }
404
405    pub(crate) fn activate<'a, 'd>(&'a self, access: Access<'a, 'd>, key: Key<'d>)
406    where
407        'd: 'a,
408    {
409        let Some(resolved) = self.entries.slots.resolve(key) else {
410            return;
411        };
412        match resolved.entry(access) {
413            Entry::Vacant => {}
414            Entry::Dispatch(_) => {
415                self.ready.insert(resolved.index());
416            }
417            Entry::Task(node) if !task::raw::Binding::is_ready(node) => {
418                self.ready.insert(resolved.index());
419            }
420            Entry::Task(_) => {}
421        }
422    }
423
424    fn activate_dispatch(&self, key: Key<'_>) {
425        let Some(resolved) = self.entries.slots.resolve(key) else {
426            return;
427        };
428        if resolved.dispatch().is_some() {
429            self.ready.insert(resolved.index());
430        }
431    }
432
433    fn release<'d>(&'d self, key: DynamicKey<'d>) -> bool {
434        self.entries
435            .pool
436            .release(&self.entries.slots, key, |index| self.ready.remove(index))
437    }
438
439    fn reclaim_task<'d>(&'d self, key: DynamicKey<'d>) -> (Option<Reservation<'d>>, bool) {
440        self.entries
441            .pool
442            .reclaim_task(&self.entries.slots, key, |index| self.ready.remove(index))
443    }
444
445    pub(in crate::driver::schedule) fn drain<'a, 'd>(
446        &'a self,
447        access: Access<'a, 'd>,
448        budget: &mut credits::Budget<'_, 'd, Lane>,
449        mut activate: impl FnMut(route::Token),
450    ) -> usize
451    where
452        'd: 'a,
453    {
454        let Some(mut ready) = self.ready.drain_batch() else {
455            return 0;
456        };
457        let mut drained = 0;
458        loop {
459            let index = match budget.admit_with(|| ready.next()) {
460                schedule::Admission::Item(index) => index,
461                schedule::Admission::Empty => break,
462                schedule::Admission::Exhausted => {
463                    ready.pause();
464                    break;
465                }
466            };
467            drained += 1;
468            match self.entries.slots.entry(index, access) {
469                Entry::Vacant => {}
470                Entry::Dispatch(target) => activate(target),
471                Entry::Task(node) => {
472                    if let Some(parent) = task::raw::Binding::activate(node) {
473                        parent.wake();
474                    }
475                }
476            }
477        }
478        drained
479    }
480
481    pub(crate) fn has_ready(&self) -> bool {
482        !self.ready.is_empty()
483    }
484
485    fn task_parent<'a, 'd>(&'a self, access: Access<'a, 'd>, key: Key<'d>) -> TaskParent<'d>
486    where
487        'd: 'a,
488    {
489        let Some(resolved) = self.entries.slots.resolve(key) else {
490            return TaskParent::Stale;
491        };
492        match resolved.entry(access) {
493            Entry::Dispatch(_) => TaskParent::Root,
494            Entry::Task(node) => task::raw::Binding::parent(node)
495                .map(TaskParent::Task)
496                .unwrap_or(TaskParent::Stale),
497            Entry::Vacant => TaskParent::Stale,
498        }
499    }
500}
dope_core/driver/schedule/ready/mod.rs

dope_core/driver/schedule/ready/
mod.rs
