dope_core/driver/
mod.rs

1pub mod bootstrap;
2pub mod buffers;
3pub mod completion;
4pub mod control;
5pub mod datagram;
6pub mod ext;
7pub mod profile;
8pub mod ready;
9pub mod route;
10pub mod submission;
11pub mod token;
12
13use std::cell::Cell;
14use std::error::Error;
15use std::fmt::{self, Display, Formatter};
16use std::io::{self, ErrorKind, Result};
17use std::marker::{PhantomData, PhantomPinned};
18use std::pin::Pin;
19use std::time::Instant;
20
21use o3::cell::{BrandToken, RegionToken};
22use o3::collections::CellQueue;
23use o3::marker::ThreadBound;
24
25use crate::backend::Backend;
26use crate::backend::ops::buffers::BufferBackend;
27use crate::io::fd::{Fd, FdGuard, FdSlot};
28use crate::io::provided::raw::buffer::BufferId;
29use crate::platform::Platform;
30use crate::platform::raw::file::{FileLimit, lock_memory_best_effort};
31use profile::DriverProfile;
32use ready::{Arena, ReadyHandle, ReadyKey, ReadySlot};
33use token::{SlotIndex, Token, TokenCapacity};
34
35type Invariant<'d> = PhantomData<fn(&'d ()) -> &'d ()>;
36
37struct Shared {
38    arena: Box<Arena>,
39    returned_buffers: CellQueue<BufferId>,
40    turn_clock: Cell<Instant>,
41}
42
43pub struct Driver {
44    shared: Shared,
45    backend: Backend,
46    _pin: PhantomPinned,
47}
48
49pub struct DriverRef<'d> {
50    shared: &'d Shared,
51    _brand: Invariant<'d>,
52}
53
54pub struct Scope<'d> {
55    driver: DriverRef<'d>,
56    backend: &'d mut Backend,
57    region: RegionToken<'d>,
58    token: BrandToken<'d>,
59}
60
61pub struct DriverContext<'a, 'd> {
62    driver: DriverRef<'d>,
63    backend: &'a mut Backend,
64    region: &'a mut RegionToken<'d>,
65}
66
67impl Driver {
68    pub fn init_process() -> Result<()> {
69        lock_memory_best_effort();
70        FileLimit::get()?.raise()
71    }
72
73    pub(crate) fn from_state(
74        state: Backend,
75        fixed_slots: usize,
76        dynamic_slots: usize,
77        provided_buffers: usize,
78    ) -> Result<Self> {
79        Ok(Self {
80            shared: Shared {
81                arena: Arena::new(fixed_slots, dynamic_slots)?,
82                returned_buffers: CellQueue::with_capacity(provided_buffers),
83                turn_clock: Cell::new(Instant::now()),
84            },
85            backend: state,
86            _pin: PhantomPinned,
87        })
88    }
89
90    pub fn scope<R>(self: Pin<&mut Self>, f: impl for<'d> FnOnce(Scope<'d>) -> R) -> R {
91        // SAFETY: the pointer is used only within the synchronous branded scope,
92        // while the pinned exclusive borrow remains active.
93        let this = unsafe { self.get_unchecked_mut() as *mut Self };
94        BrandToken::scope_with_region(move |token, region| {
95            // SAFETY: the higher-ranked closure prevents the generated lifetime
96            // and every reference derived from this pointer from escaping.
97            let this = unsafe { &mut *this };
98            f(Scope {
99                driver: DriverRef::new(&this.shared),
100                backend: &mut this.backend,
101                region,
102                token,
103            })
104        })
105    }
106}
107
108impl<'d> DriverRef<'d> {
109    fn new(shared: &'d Shared) -> Self {
110        Self {
111            shared,
112            _brand: PhantomData,
113        }
114    }
115
116    fn arena(self) -> &'d Arena {
117        &self.shared.arena
118    }
119
120    pub(crate) fn fixed_ready(self, slot: FdSlot) -> ReadyHandle<'d> {
121        self.arena().fixed_slot(slot)
122    }
123
124    pub(crate) fn fixed_fd_slot(self, raw: u32) -> Option<FdSlot> {
125        self.arena().fd_slot(raw)
126    }
127
128    pub fn make_ready_slot(self, target: Token) -> Result<ReadySlot<'d>> {
129        self.arena().make_slot(target)
130    }
131
132    pub fn make_ready_slot_reserving(self, target: Token, reserve: usize) -> Result<ReadySlot<'d>> {
133        self.arena().make_slot_reserving(target, reserve)
134    }
135
136    pub fn make_ready_slots<I>(self, targets: I) -> Result<Box<[ReadySlot<'d>]>>
137    where
138        I: IntoIterator<Item = Token>,
139        I::IntoIter: ExactSizeIterator,
140    {
141        self.arena().make_slots(targets)
142    }
143
144    pub fn activate_ready(self, key: ReadyKey<'d>) {
145        self.arena().activate(key);
146    }
147
148    #[doc(hidden)]
149    pub fn arm_recv_credit(self, key: ReadyKey<'d>, target: Token) -> bool {
150        self.arena().arm_recv_credit(key, target)
151    }
152
153    #[doc(hidden)]
154    pub fn release_recv_credit(self, key: ReadyKey<'d>, target: Token) {
155        self.arena().release_recv_credit(key, target);
156    }
157
158    #[doc(hidden)]
159    pub fn take_recv_credit(self, key: ReadyKey<'d>, target: Token) -> bool {
160        self.arena().take_recv_credit(key, target)
161    }
162
163    pub fn drain_ready(self, activate: impl FnMut(Token)) {
164        self.arena().drain(activate);
165    }
166
167    pub fn has_ready(self) -> bool {
168        self.arena().has_ready()
169    }
170
171    /// Returns the monotonic-clock snapshot for the current driver turn.
172    ///
173    /// Unlike [`Instant::now`], this is a cached read. The runtime refreshes it
174    /// at completion-batch entry and immediately before preparing to park, so
175    /// callbacks in one turn share a coherent time base without performing a
176    /// clock read for every event.
177    pub fn turn_now(self) -> Instant {
178        self.shared.turn_clock.get()
179    }
180
181    pub(crate) fn return_buffer(self, id: BufferId) {
182        assert!(
183            self.shared.returned_buffers.push_back(id).is_ok(),
184            "dope: provided-buffer return queue overflow"
185        );
186    }
187}
188
189impl Copy for DriverRef<'_> {}
190
191impl Clone for DriverRef<'_> {
192    fn clone(&self) -> Self {
193        *self
194    }
195}
196
197impl PartialEq for DriverRef<'_> {
198    fn eq(&self, other: &Self) -> bool {
199        core::ptr::eq(self.shared, other.shared)
200    }
201}
202
203impl Eq for DriverRef<'_> {}
204
205impl<'d> Scope<'d> {
206    pub fn context(&mut self) -> DriverContext<'_, 'd> {
207        DriverContext {
208            driver: self.driver,
209            backend: &mut *self.backend,
210            region: &mut self.region,
211        }
212    }
213
214    pub fn token(&mut self) -> &mut BrandToken<'d> {
215        &mut self.token
216    }
217
218    pub fn token_and_context(&mut self) -> (&mut BrandToken<'d>, DriverContext<'_, 'd>) {
219        (
220            &mut self.token,
221            DriverContext {
222                driver: self.driver,
223                backend: &mut *self.backend,
224                region: &mut self.region,
225            },
226        )
227    }
228
229    pub fn driver_ref(&self) -> DriverRef<'d> {
230        self.driver
231    }
232}
233
234impl<'a, 'd> DriverContext<'a, 'd> {
235    pub fn reborrow(&mut self) -> DriverContext<'_, 'd> {
236        DriverContext {
237            driver: self.driver,
238            backend: &mut *self.backend,
239            region: &mut *self.region,
240        }
241    }
242
243    pub fn region_token(&mut self) -> &mut RegionToken<'d> {
244        self.region
245    }
246
247    #[doc(hidden)]
248    pub fn region_token_ref(&self) -> &RegionToken<'d> {
249        self.region
250    }
251
252    pub fn driver_ref(&self) -> DriverRef<'d> {
253        self.driver
254    }
255
256    /// Returns the monotonic-clock snapshot for the current driver turn.
257    pub fn turn_now(&self) -> Instant {
258        self.driver.turn_now()
259    }
260
261    /// Starts a new driver-clock epoch and returns its snapshot.
262    ///
263    /// Runtimes should refresh once before dispatching a completion/ready batch
264    /// and once after application callbacks, immediately before timeout
265    /// expiration and park-duration calculations.
266    #[doc(hidden)]
267    pub fn refresh_turn_clock(&mut self) -> Instant {
268        let now = Instant::now();
269        self.driver.shared.turn_clock.set(now);
270        now
271    }
272
273    pub(crate) fn backend(&mut self) -> &mut Backend {
274        self.backend
275    }
276
277    pub(crate) fn backend_ref(&self) -> &Backend {
278        self.backend
279    }
280
281    pub(crate) fn release_buffer(&mut self, id: BufferId) {
282        <Backend as BufferBackend>::release_buffer(self.backend(), id);
283    }
284
285    pub(crate) fn flush_returned_buffers(&mut self) {
286        while let Some(id) = self.driver.shared.returned_buffers.pop_front() {
287            self.release_buffer(id);
288        }
289    }
290    pub fn guard(&mut self, fd: Fd<'d>) -> FdGuard<'_, 'd> {
291        let (slot, driver, retire_slot) = fd.into_parts();
292        FdGuard::new(self.backend(), slot, driver, retire_slot)
293    }
294}
295
296#[derive(Clone, Copy, Debug, PartialEq, Eq)]
297pub struct ProvidedBufferConfig {
298    pub len: usize,
299    pub entries: u16,
300}
301
302#[derive(Clone, Copy, Debug, PartialEq, Eq)]
303pub struct Config {
304    pub ring_entries: u32,
305    pub cq_entries: u32,
306    pub fixed_file_slots: u32,
307    pub accept_slots: u32,
308    pub provided: ProvidedBufferConfig,
309    pub defer_taskrun: bool,
310    pub ready_slots: usize,
311}
312
313impl Default for Config {
314    fn default() -> Self {
315        Self {
316            ring_entries: 1024,
317            cq_entries: 2048,
318            fixed_file_slots: 65536,
319            accept_slots: 65536,
320            provided: ProvidedBufferConfig {
321                len: 4096,
322                entries: 128,
323            },
324            defer_taskrun: false,
325            ready_slots: 65536,
326        }
327    }
328}
329
330impl Config {
331    const MAX_ENTRIES: u32 = 32768;
332
333    pub fn fixed_file_slots(&self) -> u32 {
334        self.fixed_file_slots
335    }
336
337    fn sized_sq(max_connections: u32, outbound_reserve: u32) -> u32 {
338        max_connections
339            .saturating_add(outbound_reserve)
340            .next_power_of_two()
341            .clamp(64, Self::MAX_ENTRIES)
342    }
343
344    pub fn for_profile<P: DriverProfile>() -> Self {
345        Self {
346            ring_entries: P::RING_ENTRIES,
347            cq_entries: P::CQ_ENTRIES,
348            fixed_file_slots: P::FIXED_FILE_SLOTS,
349            accept_slots: P::FIXED_FILE_SLOTS.saturating_sub(P::OUTBOUND_RESERVE),
350            provided: ProvidedBufferConfig {
351                len: P::PROVIDED_BUF_LEN,
352                entries: P::PROVIDED_BUF_ENTRIES,
353            },
354            defer_taskrun: P::DEFER_TASKRUN,
355            ready_slots: P::READY_SLOTS,
356        }
357    }
358
359    pub fn for_tcp_profile<P: DriverProfile>(max_connections: usize) -> Self {
360        let max_connections = u32::try_from(max_connections).unwrap_or(u32::MAX);
361        let accept_slots =
362            max_connections.min(P::FIXED_FILE_SLOTS.saturating_sub(P::OUTBOUND_RESERVE));
363        Self {
364            ring_entries: Self::sized_sq(accept_slots, P::OUTBOUND_RESERVE).min(P::RING_ENTRIES),
365            cq_entries: P::CQ_ENTRIES,
366            fixed_file_slots: accept_slots.saturating_add(P::OUTBOUND_RESERVE),
367            accept_slots,
368            provided: ProvidedBufferConfig::for_accept(
369                accept_slots,
370                P::PROVIDED_BUF_LEN,
371                Self::MAX_ENTRIES,
372            ),
373            defer_taskrun: P::DEFER_TASKRUN,
374            ready_slots: P::READY_SLOTS,
375        }
376    }
377
378    pub fn for_quic_udp(provided_buf_entries: u32, provided_buf_len: u32) -> Self {
379        Self {
380            ring_entries: 256,
381            cq_entries: 1024,
382            fixed_file_slots: 16,
383            accept_slots: 0,
384            provided: ProvidedBufferConfig {
385                len: provided_buf_len as usize,
386                entries: provided_buf_entries.min(u16::MAX as u32) as u16,
387            },
388            defer_taskrun: false,
389            ready_slots: 1024,
390        }
391    }
392
393    pub fn with_provided(mut self, len: usize, entries: u16) -> Self {
394        self.provided.apply_overrides(len, entries);
395        self
396    }
397
398    pub(crate) fn validate(&self) -> Result<()> {
399        if self.accept_slots > self.fixed_file_slots {
400            return Err(io::Error::new(
401                ErrorKind::InvalidInput,
402                "dope: accept slots exceed fixed-file capacity",
403            ));
404        }
405        if TokenCapacity::new(self.fixed_file_slots as usize).is_none() {
406            return Err(io::Error::new(
407                ErrorKind::InvalidInput,
408                "dope: fixed-file capacity exceeds token slots",
409            ));
410        }
411        Backend::snapshot()?
412            .check_slots(self.fixed_file_slots)
413            .map_err(io::Error::from)
414    }
415}
416
417impl ProvidedBufferConfig {
418    pub fn for_accept(accept_slots: u32, buf_len: usize, max_entries: u32) -> Self {
419        const PROVIDED_FLOOR: u32 = 1024;
420        const K_BATCH: u32 = 4;
421        const DRAIN_BATCH: u32 = 256;
422
423        let buf_len_ratio = (buf_len / 4096).max(1) as u32;
424        let hwm_entries = K_BATCH
425            .saturating_mul(DRAIN_BATCH)
426            .min(accept_slots.max(PROVIDED_FLOOR));
427        let target = hwm_entries.min(max_entries) / buf_len_ratio;
428        Self {
429            len: buf_len,
430            entries: target.max(PROVIDED_FLOOR).min(u16::MAX as u32) as u16,
431        }
432    }
433
434    pub fn apply_overrides(&mut self, len: usize, entries: u16) {
435        if len != 0 {
436            self.len = len;
437        }
438        if entries != 0 {
439            self.entries = entries;
440        }
441    }
442}
443
444#[derive(Debug, Clone, Copy)]
445pub struct PushError;
446
447impl Display for PushError {
448    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
449        formatter.write_str("dope: SQE push failed")
450    }
451}
452
453impl Error for PushError {}
454
455impl From<PushError> for io::Error {
456    fn from(_: PushError) -> Self {
457        io::Error::from(ErrorKind::WouldBlock)
458    }
459}
460
461pub struct OutboundReservation<'d> {
462    base: u32,
463    capacity: u32,
464    _brand: Invariant<'d>,
465    _thread: ThreadBound,
466}
467
468#[derive(Clone, Copy)]
469pub struct OutboundSlot<'d> {
470    fd: FdSlot,
471    _brand: Invariant<'d>,
472}
473
474impl<'d> OutboundSlot<'d> {
475    pub fn bind(self, driver: DriverRef<'d>) -> Fd<'d> {
476        Fd::from_range_slot(self.fd, driver)
477    }
478}
479
480impl<'d> OutboundReservation<'d> {
481    pub(crate) fn new(base: u32, capacity: u32) -> Self {
482        Self {
483            base,
484            capacity,
485            _brand: PhantomData,
486            _thread: ThreadBound::NEW,
487        }
488    }
489
490    pub fn empty() -> Self {
491        Self {
492            base: 0,
493            capacity: 0,
494            _brand: PhantomData,
495            _thread: ThreadBound::NEW,
496        }
497    }
498
499    pub fn slot(&self, local: SlotIndex) -> Option<OutboundSlot<'d>> {
500        if local.raw() >= self.capacity {
501            return None;
502        }
503        let index = SlotIndex::try_new(self.base.checked_add(local.raw())?)?;
504        Some(OutboundSlot {
505            fd: FdSlot::from_index(index),
506            _brand: PhantomData,
507        })
508    }
509
510    pub(crate) fn into_range(self) -> (u32, u32) {
511        (self.base, self.capacity)
512    }
513}
dope_core/driver/mod.rs

dope_core/driver/
mod.rs
