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

dope_core/driver/
mod.rs
