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

dope_core/driver/
mod.rs
