dope_core/driver/receive/
mod.rs

1//! Provided-buffer exhaustion contracts.
2
3use std::{marker, task::Poll};
4
5pub use attempt::Attempt;
6use dope_sys::{
7    engine,
8    memory::buffer,
9    table::{permit, permit::Queue},
10};
11use o3::base::num::bounded;
12
13use crate::{
14    Backend, Engine, driver,
15    driver::{schedule::ready, settings},
16};
17
18pub(crate) mod access;
19mod attempt;
20pub mod credit;
21mod raw;
22pub(crate) mod tokens;
23
24/// Receive-message frame capacity retained by a failed driver probe.
25#[derive(Clone, Copy, PartialEq, Eq, Debug)]
26#[repr(transparent)]
27pub struct Capacity(bounded::U64<0, { u64::MAX }>);
28
29/// Exact receive-message frame length retained by a failed driver probe.
30#[derive(Clone, Copy, PartialEq, Eq, Debug)]
31#[repr(transparent)]
32pub struct Length(bounded::U64<0, { u64::MAX }>);
33
34impl From<bounded::U64<0, { u64::MAX }>> for Capacity {
35    fn from(value: bounded::U64<0, { u64::MAX }>) -> Self {
36        Self(value)
37    }
38}
39
40impl From<bounded::U64<0, { u64::MAX }>> for Length {
41    fn from(value: bounded::U64<0, { u64::MAX }>) -> Self {
42        Self(value)
43    }
44}
45
46/// Exact out-of-range provided-buffer selection observed from a backend.
47#[derive(Clone, Copy, PartialEq, Eq, Debug)]
48#[repr(transparent)]
49pub struct Selection(<Backend as buffer::Selection<settings::Buffers>>::Error);
50
51impl Selection {
52    pub(crate) fn from_native<E>(value: E) -> Self
53    where
54        Backend: buffer::Selection<settings::Buffers, Error = E>,
55    {
56        Self(value)
57    }
58}
59
60/// Exact ready key owned by provided-buffer receive authority.
61#[repr(transparent)]
62pub(crate) struct Proof(ready::Stored);
63
64impl Proof {
65    pub(in crate::driver) const fn new(key: ready::Stored) -> Self {
66        Self(key)
67    }
68
69    pub(in crate::driver) const fn key(&self) -> ready::Stored {
70        self.0
71    }
72
73    pub(crate) fn release(
74        permit: &<crate::Permits as Queue>::Capability,
75        store: &crate::Permits,
76        driver: driver::Reference<'_>,
77    ) {
78        while let Some((index, key)) = store.grant_one(permit) {
79            let stale = match driver.ready().arena().stored().try_receive(Self::new(key)) {
80                Ok(()) => continue,
81                Err(stale) => stale.key(),
82            };
83            match store.take(permit, index, stale) {
84                permit::State::Granted(returned) => {
85                    if returned != stale {
86                        return;
87                    }
88                }
89                permit::State::Gone | permit::State::Waiting | permit::State::Stale => return,
90            }
91        }
92    }
93}
94
95/// Driver-branded proof of one terminal provided-buffer exhaustion.
96/// Retains the driver's invariant lifetime and thread confinement without storage.
97#[repr(transparent)]
98pub struct Exhausted<'d> {
99    driver: marker::PhantomData<driver::Reference<'d>>,
100}
101
102/// Exact ready-target registration for provided-buffer exhaustion.
103#[must_use = "a receive registration must be polled, cancelled, or retired"]
104#[repr(transparent)]
105pub struct Registration<'d> {
106    target: ready::Target<'d>,
107    permit: <crate::Permits as Queue>::Capability,
108}
109
110/// A registration no longer names the live receive target that created it.
111#[derive(Clone, Copy, Debug, PartialEq, Eq)]
112pub struct Stale;
113
114/// The selected backend does not provide receive registrations.
115#[derive(Clone, Copy, Debug, PartialEq, Eq)]
116pub struct Absent;
117
118impl<'d> Exhausted<'d> {
119    pub(crate) const fn from_driver(_driver: driver::Reference<'d>) -> Self {
120        Self {
121            driver: marker::PhantomData,
122        }
123    }
124
125    /// Consumes terminal exhaustion for this generative driver scope.
126    pub(in crate::driver) fn retire(self) -> dope_sys::Errno {
127        dope_sys::Errno::NO_BUFFER_SPACE
128    }
129}
130
131impl<'d, P> Registration<'d>
132where
133    crate::Permits: Queue<Capability = P>,
134{
135    /// Acquires provided-buffer receive authority for this target.
136    pub fn try_acquire(target: ready::Target<'d>) -> Result<Self, Absent> {
137        let permit = crate::Permits::try_acquire().map_err(|permit::Absent| Absent)?;
138        Ok(Self { target, permit })
139    }
140
141    /// Retires a registration when the selected backend provides one.
142    pub fn clear(target: ready::Target<'d>) {
143        let reg = match Self::try_acquire(target) {
144            Ok(reg) => reg,
145            Err(Absent) => return,
146        };
147        reg.retire();
148    }
149
150    /// Registers exhaustion before permitting one optimistic retry.
151    pub fn exhausted(
152        self,
153        proof: Exhausted<'d>,
154    ) -> Result<Poll<Attempt<'d>>, (Exhausted<'d>, Stale)> {
155        let Self { target, permit } = self;
156        let (driver, key) = target.receive();
157        if !Self::is_live(driver, key) {
158            return Err((proof, Stale));
159        }
160        let store = driver.receive().exhausted();
161        match store.try_arm(&permit, Index::from_key(key), key) {
162            Ok(()) => Ok(Poll::Ready(Attempt::new(permit, driver))),
163            Err(key) => Self::take(permit, store, driver, key).map_err(|stale| (proof, stale)),
164        }
165    }
166
167    /// Observes the exact central registration after its target was woken.
168    /// Stale targets discard any leftover grant before reporting staleness.
169    pub fn resume(self) -> Result<Poll<Attempt<'d>>, Stale> {
170        let Self { target, permit } = self;
171        let (driver, key) = target.receive();
172        let store = driver.receive().exhausted();
173        if !Self::is_live(driver, key) {
174            return match Self::cancel_inner(&permit, store, key) {
175                Ok(()) | Err(Stale) => Err(Stale),
176            };
177        }
178        Self::take(permit, store, driver, key)
179    }
180
181    /// Cancels this exact registration and discards an unconsumed grant.
182    pub fn cancel(self) -> Result<(), Stale> {
183        let Self { target, permit } = self;
184        let (driver, key) = target.receive();
185        Self::cancel_inner(&permit, driver.receive().exhausted(), key)
186    }
187
188    /// Retires this registration, discarding a matching unconsumed grant.
189    /// A stale registration has no matching grant to retire.
190    pub fn retire(self) {
191        match self.cancel() {
192            Ok(()) | Err(Stale) => {}
193        }
194    }
195
196    fn is_live(driver: driver::Reference<'_>, key: ready::Stored) -> bool {
197        driver.ready().arena().stored().is_dispatch(Proof::new(key))
198    }
199
200    fn take(
201        permit: P,
202        store: &'d crate::Permits,
203        driver: driver::Reference<'d>,
204        key: ready::Stored,
205    ) -> Result<Poll<Attempt<'d>>, Stale> {
206        match store.take(&permit, Index::from_key(key), key) {
207            permit::State::Waiting => Ok(Poll::Pending),
208            permit::State::Granted(_) => Ok(Poll::Ready(Attempt::new(permit, driver))),
209            permit::State::Gone | permit::State::Stale => Err(Stale),
210        }
211    }
212
213    fn cancel_inner(permit: &P, store: &crate::Permits, key: ready::Stored) -> Result<(), Stale> {
214        match store.cancel(permit, Index::from_key(key), key) {
215            permit::State::Gone | permit::State::Waiting | permit::State::Granted(_) => Ok(()),
216            permit::State::Stale => Err(Stale),
217        }
218    }
219}
220
221const _: () = {
222    assert!(
223        std::mem::size_of::<Selection>()
224            == std::mem::size_of::<<Engine as engine::Buffer>::Selection>()
225    );
226    assert!(
227        std::mem::align_of::<Selection>()
228            == std::mem::align_of::<<Engine as engine::Buffer>::Selection>()
229    );
230    assert!(std::mem::size_of::<Proof>() == std::mem::size_of::<ready::Stored>());
231    assert!(std::mem::align_of::<Proof>() == std::mem::align_of::<ready::Stored>());
232    assert!(std::mem::size_of::<Exhausted<'static>>() == 0);
233    assert!(std::mem::align_of::<Exhausted<'static>>() == 1);
234    assert!(std::mem::size_of::<<crate::Permits as Queue>::Capability>() == 0);
235    assert!(std::mem::size_of::<permit::Absent>() == 0);
236    assert!(std::mem::size_of::<Absent>() == 0);
237    assert!(std::mem::size_of::<Registration<'static>>() == 2 * std::mem::size_of::<usize>());
238    assert!(
239        std::mem::size_of::<Registration<'static>>()
240            == std::mem::size_of::<ready::Target<'static>>()
241    );
242    assert!(
243        std::mem::align_of::<Registration<'static>>()
244            == std::mem::align_of::<ready::Target<'static>>()
245    );
246};
247
248/// Compact receive-permit position retaining its ready-arena domain.
249#[derive(Clone, Copy)]
250#[repr(transparent)]
251pub struct Index(ready::DynamicIndex);
252
253impl Index {
254    const fn from_key(key: ready::Stored) -> Self {
255        Self(key.dense())
256    }
257}
dope_core/driver/receive/mod.rs

dope_core/driver/receive/
mod.rs
