dope_core/driver/schedule/ready/wake/
queue.rs

1use std::{mem, num};
2
3use o3::depot::queue::resident::{
4    self, fifo,
5    fifo::{Index, Scope},
6};
7
8use crate::{
9    driver,
10    driver::schedule::{
11        app,
12        ready::{
13            self, wake,
14            wake::{
15                Drain, Once, Waker,
16                fanout::storage::Step,
17                resident::{drain, one},
18            },
19        },
20    },
21};
22
23/// Bounded FIFO wakes with a constructor-proved lane extent.
24pub struct Queue<'d, L = num::NonZero<usize>> {
25    driver: driver::Reference<'d>,
26    inner: fifo::Fifo<ready::Retained<'d>, L>,
27}
28
29/// Linear ownership of one resident queue slot.
30#[repr(transparent)]
31pub struct Registration<'target, 'd: 'target, L = num::NonZero<usize>> {
32    pub(super) inner: fifo::Registration<'target, ready::Retained<'d>, L>,
33}
34
35#[derive(Clone, Copy, Debug, Eq, PartialEq)]
36#[must_use = "a rejected wake registration must be handled"]
37pub enum Error {
38    Lane(resident::Bounds<num::NonZero<u32>>),
39    Active { position: wake::Position },
40    Draining { position: wake::Position },
41    Full { position: wake::Position },
42}
43
44impl From<fifo::Reject> for Error {
45    fn from(error: fifo::Reject) -> Self {
46        match error {
47            fifo::Reject::Active { index } => Self::Active {
48                position: wake::Position::new(index),
49            },
50            fifo::Reject::Draining { index } => Self::Draining {
51                position: wake::Position::new(index),
52            },
53            fifo::Reject::Full { index } => Self::Full {
54                position: wake::Position::new(index),
55            },
56        }
57    }
58}
59
60impl<'d, L> Queue<'d, L> {
61    pub fn try_new(
62        driver: driver::Reference<'d>,
63        capacity: usize,
64        lanes: L,
65    ) -> Result<Self, resident::Error>
66    where
67        L: resident::Extent,
68    {
69        Ok(Self {
70            driver,
71            inner: fifo::Fifo::try_new(capacity, lanes)?,
72        })
73    }
74}
75
76impl<'d> Queue<'d> {
77    /// Creates one queue and borrows its source under a fresh invariant scope.
78    pub fn try_with<S, R>(
79        driver: driver::Reference<'d>,
80        capacity: usize,
81        lanes: num::NonZero<usize>,
82        source: S,
83        run: impl for<'scope> FnOnce(&'scope S, Queue<'d, Scope<'scope>>) -> R,
84    ) -> Result<R, resident::Error> {
85        fifo::Fifo::try_with(capacity, lanes, source, |source, inner| {
86            run(source, Queue { driver, inner })
87        })
88    }
89
90    pub fn try_register<'target>(
91        &'target self,
92        position: usize,
93        slot: &mut Registration<'target, 'd>,
94        wake: &Waker<'d>,
95    ) -> Result<(), Error> {
96        let inner = self.inner.try_lane(position).map_err(Error::Lane)?;
97        inner
98            .try_keep(&mut slot.inner, *wake.key())
99            .map_err(Error::from)
100    }
101
102    pub fn try_wake_one(
103        &self,
104        position: usize,
105    ) -> Result<Once, resident::Bounds<num::NonZero<u32>>> {
106        let lane = self.inner.try_lane(position)?;
107        Ok(one(self.driver, lane.take_one()))
108    }
109}
110
111impl<'scope, 'd> Queue<'d, Scope<'scope>> {
112    pub fn try_index(
113        &self,
114        position: usize,
115    ) -> Result<Index<'scope>, resident::Bounds<num::NonZero<u32>>> {
116        self.inner.try_index(position)
117    }
118
119    pub fn try_register<'target>(
120        &'target self,
121        index: Index<'scope>,
122        slot: &mut Registration<'target, 'd, Scope<'scope>>,
123        wake: &Waker<'d>,
124    ) -> Result<(), fifo::Reject> {
125        self.inner
126            .lane(index)
127            .try_keep(&mut slot.inner, *wake.key())
128    }
129
130    pub fn wake_one(&self, index: Index<'scope>) -> Once {
131        one(self.driver, self.inner.lane(index).take_one())
132    }
133}
134
135impl<'d, const N: usize> Queue<'d, resident::Fixed<N>> {
136    pub fn try_register<'target, const I: usize>(
137        &'target self,
138        slot: &mut Registration<'target, 'd, resident::Fixed<N>>,
139        wake: &Waker<'d>,
140    ) -> Result<(), fifo::Reject> {
141        self.inner
142            .lane::<I>()
143            .try_keep(&mut slot.inner, *wake.key())
144    }
145
146    pub fn wake_one<const I: usize>(&self) -> Once {
147        one(self.driver, self.inner.lane::<I>().take_one())
148    }
149}
150
151impl<'d, L> Queue<'d, L> {
152    /// Cancels the registration, forwarding a taken wake within its original lane.
153    /// Active drains retain their work; successful admission uses `clear`.
154    pub fn cancel(&self, slot: &mut Registration<'_, 'd, L>) -> Once {
155        one(self.driver, slot.inner.cancel())
156    }
157
158    /// Registers one wake in the constructor-proved first lane.
159    pub fn try_first<'target>(
160        &'target self,
161        slot: &mut Registration<'target, 'd, L>,
162        wake: &Waker<'d>,
163    ) -> Result<(), fifo::Reject> {
164        self.inner.first().try_keep(&mut slot.inner, *wake.key())
165    }
166
167    /// Wakes one registration from lane zero.
168    pub fn wake_first(&self) -> Once {
169        one(self.driver, self.inner.first().take_one())
170    }
171
172    pub fn wake_next(&self) -> Once {
173        one(self.driver, self.inner.take_next())
174    }
175
176    pub fn has_waiters(&self) -> bool {
177        self.inner.has_waiters()
178    }
179
180    pub fn begin_drain(&self) {
181        self.inner.begin_drain();
182    }
183
184    pub fn is_draining(&self) -> bool {
185        self.inner.is_draining()
186    }
187
188    pub fn drain_one(&self, work: app::Work<'_, 'd>) -> Drain {
189        drain::one(
190            self.driver,
191            work,
192            || self.inner.is_draining() && self.inner.has_waiters(),
193            || self.inner.drain_next().map(Step::Item),
194        )
195    }
196}
197
198impl<L> Default for Registration<'_, '_, L> {
199    fn default() -> Self {
200        Self::new()
201    }
202}
203
204impl<L> Registration<'_, '_, L> {
205    pub const fn new() -> Self {
206        Self {
207            inner: fifo::Registration::new(),
208        }
209    }
210
211    pub fn is_active(&self) -> bool {
212        self.inner.is_active()
213    }
214
215    pub fn clear(&mut self) {
216        *self = Self::new();
217    }
218}
219
220const _: () = {
221    assert!(mem::size_of::<Registration<'static, 'static>>() == 2 * mem::size_of::<usize>());
222    assert!(mem::align_of::<Registration<'static, 'static>>() == mem::align_of::<usize>());
223    assert!(
224        size_of::<Registration<'static, 'static, resident::Fixed<1>>>() == 2 * size_of::<usize>()
225    );
226    assert!(size_of::<Queue<'static, resident::Fixed<1>>>() == size_of::<Queue<'static>>());
227    assert!(size_of::<Queue<'static, Scope<'static>>>() == size_of::<Queue<'static>>());
228    assert!(size_of::<Index<'static>>() == size_of::<u32>());
229    assert!(size_of::<Registration<'static, 'static, Scope<'static>>>() == 2 * size_of::<usize>());
230};
dope_core/driver/schedule/ready/wake/queue.rs

dope_core/driver/schedule/ready/wake/
queue.rs
