dope_core/driver/
mod.rs

1pub mod flight;
2pub mod lifecycle;
3pub mod ops;
4pub mod retained;
5pub mod route;
6pub mod schedule;
7pub mod settings;
8pub mod storage;
9
10use std::{error, fmt, io, marker, mem, pin, time};
11
12pub(crate) use lifecycle::Source;
13use o3::{self, cell::region, collections::batch::set};
14pub(crate) use ownership::{AccountedRecvOwner, RecvOwner};
15
16use self::{lifecycle::quiesce, ops::access, route::kind, schedule::timer, storage::ownership};
17use crate::{backend, io::fd::handles, platform};
18
19#[derive(Debug, Clone, Copy)]
20pub struct SubmitError;
21
22#[doc(hidden)]
23#[derive(Clone, Copy, Debug, PartialEq, Eq)]
24#[repr(u8)]
25pub enum RecvCreditWake {
26    ResourceReturned = kind::RECV_CREDIT_RETURNED,
27    WaiterRetry = kind::RECV_CREDIT_RETRY,
28}
29
30/// One provided receive buffer made available to an exact waiting target.
31/// Dropping an unused credit transfers it to the next waiter.
32#[must_use = "a receive buffer credit must be consumed or returned"]
33#[repr(transparent)]
34pub struct RecvBufferCredit<'d>(Reference<'d>);
35
36impl<'d> RecvBufferCredit<'d> {
37    fn new(driver: Reference<'d>) -> Self {
38        Self(driver)
39    }
40
41    pub fn consume(self) {
42        let _consumed = mem::ManuallyDrop::new(self);
43    }
44}
45
46impl Drop for RecvBufferCredit<'_> {
47    fn drop(&mut self) {
48        self.0.credits().release_recv_buffer();
49    }
50}
51
52impl fmt::Display for SubmitError {
53    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
54        formatter.write_str("dope: submission was not accepted")
55    }
56}
57
58impl error::Error for SubmitError {}
59
60impl From<SubmitError> for io::Error {
61    fn from(_: SubmitError) -> Self {
62        io::Error::from(io::ErrorKind::WouldBlock)
63    }
64}
65
66type Buffer = <backend::Backend as platform::Buffer>::Token;
67
68type Invariant<'d> = marker::PhantomData<fn(&'d ()) -> &'d ()>;
69
70#[derive(Clone, Copy)]
71pub(crate) enum FixedOwner {
72    Accepted,
73    Reserved,
74    Outbound(OutboundKey),
75}
76
77#[derive(Clone, Copy, PartialEq, Eq)]
78#[repr(transparent)]
79pub(crate) struct OutboundKey(u8);
80
81// SAFETY: OutboundKey's private byte is stable across copies. Retirement
82// storage reconstructs only raw indices previously produced by this type.
83impl set::DenseIndex for OutboundKey {
84    fn into_usize(self) -> usize {
85        self.raw() as usize
86    }
87
88    fn from_usize(raw: usize) -> Self {
89        Self::from_bounded(raw as u8)
90    }
91}
92
93#[must_use = "a fixed-file close must be submitted to the backend"]
94#[repr(transparent)]
95pub(crate) struct Close<'d> {
96    slot: handles::FixedSlot,
97    _brand: Invariant<'d>,
98}
99
100const _: () =
101    assert!(std::mem::size_of::<Close<'static>>() == std::mem::size_of::<handles::FixedSlot>());
102#[must_use = "an outbound authority release must be completed"]
103pub(crate) enum CloseDisposition<'d> {
104    Submit(Close<'d>),
105    NoSubmit(Option<RetiredSlots<'d>>),
106}
107
108#[must_use = "a successful socket creation must be delivered or reclaimed"]
109pub(crate) enum CreateSuccess<'d> {
110    Deliver(OutboundKey),
111    Close(Close<'d>),
112}
113
114#[must_use = "retired fixed slots must be returned to the backend allocator"]
115#[repr(transparent)]
116pub(crate) struct RetiredSlots<'d> {
117    key: OutboundKey,
118    _brand: Invariant<'d>,
119}
120
121const _: () =
122    assert!(std::mem::size_of::<RetiredSlots<'static>>() == std::mem::size_of::<OutboundKey>());
123
124impl OutboundKey {
125    pub(crate) fn raw(self) -> u32 {
126        self.0 as u32
127    }
128
129    pub(crate) const fn from_raw(raw: u32) -> Option<Self> {
130        if raw < route::FRAMEWORK as u32 {
131            Some(Self(raw as u8))
132        } else {
133            None
134        }
135    }
136
137    const fn from_bounded(raw: u8) -> Self {
138        debug_assert!(raw < route::FRAMEWORK);
139        Self(raw)
140    }
141
142    const fn for_route<const ID: u8>() -> Option<Self> {
143        if ID == route::FRAMEWORK {
144            None
145        } else {
146            Some(Self(ID))
147        }
148    }
149}
150
151impl<'d> Close<'d> {
152    pub(crate) fn untracked(slot: handles::FixedSlot) -> Self {
153        Self {
154            slot,
155            _brand: marker::PhantomData,
156        }
157    }
158
159    fn tracked(slot: handles::FixedSlot) -> Self {
160        Self {
161            slot,
162            _brand: marker::PhantomData,
163        }
164    }
165
166    pub(crate) fn into_slot(self) -> handles::FixedSlot {
167        self.slot
168    }
169}
170
171impl<'d> RetiredSlots<'d> {
172    fn new(key: OutboundKey) -> Self {
173        Self {
174            key,
175            _brand: marker::PhantomData,
176        }
177    }
178
179    fn into_key(self) -> OutboundKey {
180        self.key
181    }
182
183    fn key(&self) -> OutboundKey {
184        self.key
185    }
186}
187
188pub struct Driver {
189    backend: backend::Backend,
190    flights: flight::Arena,
191    shared: access::Shared,
192    timer_cache_limit: settings::ScheduleCapacity,
193    _pin: marker::PhantomPinned,
194}
195
196#[derive(Clone, Copy)]
197#[repr(transparent)]
198pub struct Reference<'d> {
199    pub(in crate::driver) shared: &'d access::Shared,
200    _brand: Invariant<'d>,
201}
202
203impl<'d> Reference<'d> {
204    pub(in crate::driver) fn new(shared: &'d access::Shared) -> Self {
205        Self {
206            shared,
207            _brand: marker::PhantomData,
208        }
209    }
210
211    pub const fn ready(self) -> access::Ready<'d> {
212        access::Ready::new(self)
213    }
214
215    pub const fn targets<Tag: route::Tag>(self) -> route::Space<'d, Tag> {
216        route::Space {
217            driver: marker::PhantomData,
218        }
219    }
220
221    pub(in crate::driver) const fn credits(self) -> access::Credits<'d> {
222        access::Credits::new(self)
223    }
224
225    pub const fn scheduler(self) -> access::Scheduler<'d> {
226        access::Scheduler::new(self)
227    }
228
229    pub fn maintenance_progress(self) -> schedule::Progress<'d> {
230        if self.maintenance().has_deferred_maintenance() {
231            schedule::Progress::Runnable
232        } else {
233            schedule::Progress::Quiescent
234        }
235    }
236
237    pub(crate) const fn receive(self) -> access::Receive<'d> {
238        access::Receive::new(self)
239    }
240
241    pub(crate) const fn maintenance(self) -> access::Maintenance<'d> {
242        access::Maintenance::new(self)
243    }
244
245    pub(crate) const fn files(self) -> access::Files<'d> {
246        access::Files::new(self)
247    }
248
249    pub(crate) const fn outbound(self) -> access::Outbound<'d> {
250        access::Outbound::new(self)
251    }
252
253    pub(crate) fn same_driver(self, other: Self) -> bool {
254        (self.shared as *const access::Shared).addr()
255            == (other.shared as *const access::Shared).addr()
256    }
257}
258
259const _: () = assert!(mem::size_of::<Reference<'static>>() == mem::size_of::<usize>());
260
261pub struct Context<'a, 'd> {
262    driver: Reference<'d>,
263    backend: &'a mut backend::Backend,
264    flights: &'a mut flight::Arena,
265    region: &'a mut region::Token<'d>,
266    timer: &'d timer::Timer<'d>,
267}
268
269impl<'a, 'd> Context<'a, 'd> {
270    fn new(
271        driver: Reference<'d>,
272        backend: &'a mut backend::Backend,
273        flights: &'a mut flight::Arena,
274        region: &'a mut region::Token<'d>,
275        timer: &'d timer::Timer<'d>,
276    ) -> Self {
277        Self {
278            driver,
279            backend,
280            flights,
281            region,
282            timer,
283        }
284    }
285
286    pub fn reborrow(&mut self) -> Context<'_, 'd> {
287        Context {
288            driver: self.driver,
289            backend: &mut *self.backend,
290            flights: &mut *self.flights,
291            region: &mut *self.region,
292            timer: self.timer,
293        }
294    }
295
296    pub fn region_token(&mut self) -> &mut region::Token<'d> {
297        self.region
298    }
299
300    #[doc(hidden)]
301    pub fn region_token_ref(&self) -> &region::Token<'d> {
302        self.region
303    }
304
305    pub fn driver_ref(&self) -> Reference<'d> {
306        self.driver
307    }
308
309    #[doc(hidden)]
310    pub fn flight_slots<Tag: route::Tag>(
311        &mut self,
312        capacity: usize,
313    ) -> io::Result<flight::Slots<'d, Tag>> {
314        self.flights.try_slots(capacity, self.driver)
315    }
316
317    pub(crate) fn backend_drain(&mut self) -> (&mut backend::Backend, flight::Drain<'_, 'd>) {
318        let Self {
319            driver,
320            backend,
321            flights,
322            ..
323        } = self;
324        (backend, flight::Drain::new(flights, *driver))
325    }
326
327    pub fn timer(&self) -> &'d timer::Timer<'d> {
328        self.timer
329    }
330
331    pub fn deadline_now(&self) -> timer::Deadline<'d> {
332        self.driver.scheduler().deadline(self.turn_now())
333    }
334
335    pub fn turn_now(&self) -> time::Instant {
336        self.driver.scheduler().turn_now()
337    }
338
339    pub(crate) fn backend(&mut self) -> &mut backend::Backend {
340        self.backend
341    }
342
343    pub(crate) fn accept_slot_capacity(&self) -> usize {
344        self.driver.files().accept_capacity()
345    }
346
347    pub(crate) fn outbound_slot_capacity(&self) -> usize {
348        self.driver.files().outbound_capacity()
349    }
350
351    pub(crate) fn pop_returned_buffer(&self) -> Option<Buffer> {
352        self.driver.maintenance().pop_returned_buffer()
353    }
354}
355
356impl Driver {
357    pub fn new(config: settings::Config) -> io::Result<Self> {
358        config.validate_structure()?;
359        let file_slots = config.file_slots();
360        let receive = config.receive();
361        let state = <backend::Backend as platform::Runtime>::build(&config)?;
362        let scheduler = config.scheduler();
363        let shared = access::Shared::try_new(file_slots, scheduler.ready(), receive)?;
364        Ok(Self {
365            backend: state,
366            flights: flight::Arena::new(),
367            shared,
368            timer_cache_limit: scheduler.timer_cache_limit(),
369            _pin: marker::PhantomPinned,
370        })
371    }
372
373    pub fn scope<R>(
374        self: pin::Pin<&mut Self>,
375        owner: quiesce::Lease,
376        f: impl for<'d> FnOnce(lifecycle::Scope<'d>) -> R,
377    ) -> R {
378        let timer_cache_limit = self.as_ref().get_ref().timer_cache_limit;
379        lifecycle::Lease::take(self).run(timer_cache_limit, owner, f)
380    }
381
382    #[doc(hidden)]
383    pub fn scope_with_storage<S, R>(
384        self: pin::Pin<&mut Self>,
385        owner: quiesce::Lease,
386        factory: S,
387        f: impl for<'d> FnOnce(lifecycle::Scope<'d>, pin::Pin<&'d S::Output<'d>>) -> R,
388    ) -> Result<R, S::Error>
389    where
390        S: storage::Factory,
391    {
392        let timer_cache_limit = self.as_ref().get_ref().timer_cache_limit;
393        lifecycle::Lease::take(self).run_with_storage(timer_cache_limit, owner, factory, f)
394    }
395}
dope_core/driver/mod.rs

dope_core/driver/
mod.rs
