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

dope_core/driver/
mod.rs
