dope_core/driver/ops/
mod.rs

1pub mod access;
2pub mod poll;
3pub(crate) mod reactors;
4pub(in crate::driver) mod retirements;
5
6use std::{io, marker, mem, net};
7
8use o3::permit;
9
10use crate::{
11    backend::{self, bound, fixed},
12    driver::{
13        self, flight,
14        route::{self, kind},
15        settings,
16    },
17    io::{
18        fd::handles,
19        socket::{self, establishment, option},
20    },
21    platform::reactor,
22};
23
24mod supported;
25#[doc(hidden)]
26pub use supported::Supported;
27
28pub struct OutboundReservation<'d, const ID: u8> {
29    lease: permit::Lease<OutboundReturn<'d>>,
30    _brand: marker::PhantomData<fn(&'d ()) -> &'d ()>,
31    _route: marker::PhantomData<route::KeyTag<ID>>,
32    _thread: o3::ThreadBound,
33}
34
35struct OutboundReturn<'d> {
36    driver: driver::Reference<'d>,
37}
38
39impl<'d> OutboundReturn<'d> {
40    fn release(&self, key: driver::OutboundKey) -> Option<driver::RetiredSlots<'d>> {
41        self.driver.outbound().release_outbound_owner(key)
42    }
43}
44
45impl permit::Return for OutboundReturn<'_> {
46    type Item = driver::OutboundKey;
47
48    fn return_item(&self, key: Self::Item) {
49        if let Some(slots) = self.release(key) {
50            self.driver.maintenance().defer_outbound_slots(slots);
51        }
52    }
53}
54
55const _: () = {
56    assert!(
57        mem::size_of::<OutboundReservation<'static, 0>>()
58            == mem::size_of::<(driver::Reference<'static>, driver::OutboundKey)>()
59    );
60    assert!(
61        mem::align_of::<OutboundReservation<'static, 0>>()
62            == mem::align_of::<(driver::Reference<'static>, driver::OutboundKey)>()
63    );
64};
65
66impl<'d, const ID: u8> OutboundReservation<'d, ID> {
67    pub(crate) fn new(driver: driver::Reference<'d>, key: driver::OutboundKey) -> Self {
68        use o3::ThreadBound;
69        Self {
70            lease: permit::Lease::new(OutboundReturn { driver }, key),
71            _brand: marker::PhantomData,
72            _route: marker::PhantomData,
73            _thread: ThreadBound::NEW,
74        }
75    }
76
77    /// Binds one local pool target to its physical descriptor slot.
78    ///
79    /// A target from another route cannot acquire this reservation:
80    ///
81    /// ```compile_fail
82    /// use dope_core::driver::{ops::OutboundReservation, route::{KeyTag, Target}};
83    ///
84    /// fn cross_route<'d>(
85    ///     reservation: &OutboundReservation<'d, 1>,
86    ///     target: Target<'d, KeyTag<2>>,
87    /// ) {
88    ///     let _ = reservation.bind(target);
89    /// }
90    /// ```
91    pub fn bind(
92        &self,
93        target: route::Target<'d, route::KeyTag<ID>>,
94    ) -> Option<route::Bound<'d, route::KeyTag<ID>, handles::SocketSlot<'d>>> {
95        use handles::SocketSlot;
96        let return_outbound = self.lease.sink();
97        let key = *self.lease.item();
98        let slot = return_outbound
99            .driver
100            .files()
101            .acquire_outbound_descriptor(key, target.slot())?;
102        let slot = SocketSlot::from_outbound_slot(slot, return_outbound.driver)?;
103        Some(target.bind(slot))
104    }
105
106    #[doc(hidden)]
107    pub fn physical_index(&self, local: route::SlotIndex) -> Option<u32> {
108        self.lease
109            .sink()
110            .driver
111            .files()
112            .outbound_physical_index(*self.lease.item(), local)
113    }
114
115    pub(crate) fn retire(self) -> Option<driver::RetiredSlots<'d>> {
116        let (return_outbound, key) = self.lease.into_parts();
117        return_outbound.release(key)
118    }
119}
120
121pub trait Bootstrap<'d>: Buffers + Supported {
122    fn bind_listener_slot(
123        &mut self,
124        addr: net::SocketAddr,
125        backlog: i32,
126        config: &socket::ListenerConfig,
127    ) -> io::Result<(handles::Descriptor<'d>, net::SocketAddr)>;
128    fn bind_datagram_slot(
129        &mut self,
130        addr: net::SocketAddr,
131    ) -> io::Result<(handles::DatagramDescriptor<'d>, net::SocketAddr)> {
132        if self.buffer_len() < settings::Receive::MIN_DATAGRAM_BUFFER_LEN as usize {
133            return Err(io::Error::new(
134                io::ErrorKind::InvalidInput,
135                "dope: receive buffer slot cannot hold a datagram envelope",
136            ));
137        }
138        self.bind_datagram_slot_raw(addr)
139            .map(|(descriptor, addr)| (handles::DatagramDescriptor::validated(descriptor), addr))
140    }
141
142    #[doc(hidden)]
143    fn bind_datagram_slot_raw(
144        &mut self,
145        addr: net::SocketAddr,
146    ) -> io::Result<(handles::Descriptor<'d>, net::SocketAddr)>;
147}
148
149pub trait Buffers {
150    fn buffer_len(&self) -> usize;
151    fn buffer_count(&self) -> usize;
152    fn accept_capacity(&self) -> usize;
153    fn outbound_capacity(&self) -> usize;
154}
155
156/// Driver control operations whose routed resources retain their typed owner.
157///
158/// Raw tokens are not accepted by the safe submission boundary:
159///
160/// ```compile_fail
161/// use dope_core::{
162///     driver::{Context, flight, ops::Control, route::{KeyTag, Token}},
163///     io::{fd::handles::SocketSlot, socket::StreamSpec},
164/// };
165///
166/// fn submit<'d>(
167///     driver: &mut Context<'_, 'd>,
168///     flights: &flight::Slots<'d, KeyTag<1>>,
169///     slot: SocketSlot<'d>,
170///     socket: StreamSpec,
171///     token: Token,
172/// ) {
173///     let _ = Control::submit_socket(driver, flights, slot, socket, token);
174/// }
175/// ```
176pub trait Control<'d> {
177    fn reserve_route(&mut self, id: u8) -> bool;
178    fn release_route(&mut self, id: u8);
179    fn submit_socket<Tag: route::Tag>(
180        &mut self,
181        flights: &flight::Slots<'d, Tag>,
182        request: route::Bound<'d, Tag, handles::SocketSlot<'d>>,
183        socket: socket::StreamSpec,
184    ) -> Result<handles::CreatingSocket<'d>, driver::SubmitError>;
185    fn submit_tuning<Tag: route::Tag>(
186        &mut self,
187        request: route::Bound<'d, Tag, handles::Descriptor<'d>>,
188        options: option::StreamOptions,
189    ) -> Result<option::Tuning<'d>, handles::Descriptor<'d>>;
190    fn cancel_tuning(
191        &mut self,
192        pending: establishment::TuningPending<'d>,
193    ) -> Result<
194        establishment::TuningPending<'d, establishment::Cancelled>,
195        (establishment::TuningPending<'d>, driver::SubmitError),
196    >;
197    fn cancel_connection(
198        &mut self,
199        pending: establishment::ConnectionPending<'d>,
200    ) -> Result<
201        establishment::ConnectionPending<'d, establishment::Cancelled>,
202        (establishment::ConnectionPending<'d>, driver::SubmitError),
203    >;
204}
205
206pub trait Files<'d> {
207    fn close(&mut self, fd: impl Into<handles::Descriptor<'d>>);
208    fn reserve_outbound<const ID: u8>(
209        &mut self,
210        count: u32,
211    ) -> io::Result<OutboundReservation<'d, ID>>;
212
213    #[doc(hidden)]
214    fn retire_outbound<const ID: u8>(&mut self, reservation: OutboundReservation<'d, ID>);
215}
216
217/// Submits driver-branded operations.
218///
219/// Cancellation cannot be redirected with an unbranded raw token:
220///
221/// ```compile_fail
222/// use dope_core::driver::{Context, ops::Submit, route::Token};
223///
224/// fn cancel_raw<'d>(driver: &mut Context<'_, 'd>, target: Token) {
225///     let _ = Submit::cancel(driver, target);
226/// }
227/// ```
228pub trait Submit<'d> {
229    /// Submits a provided-buffer stream receive. The backend copies the fixed
230    /// descriptor slot and retains no borrow from this call.
231    #[doc(hidden)]
232    fn submit_recv<Tag: route::Tag>(
233        &mut self,
234        slots: &flight::Slots<'d, Tag>,
235        fd: &handles::Descriptor<'d>,
236        target: route::Target<'d, Tag>,
237    ) -> Result<flight::Flight<'d>, driver::SubmitError>;
238
239    /// Submits a provided-buffer datagram receive. The backend copies the fixed
240    /// descriptor slot and retains no borrow from this call.
241    #[doc(hidden)]
242    fn submit_recv_datagram<Tag: route::Tag>(
243        &mut self,
244        slots: &flight::Slots<'d, Tag>,
245        fd: &handles::DatagramDescriptor<'d>,
246        identity: route::Operation<'d, Tag>,
247    ) -> Result<flight::Flight<'d>, driver::SubmitError>;
248
249    /// Submits a multishot accept. The backend copies the fixed listener slot
250    /// and retains no borrow from this call.
251    #[doc(hidden)]
252    fn submit_accept_multishot<Tag: route::Tag>(
253        &mut self,
254        slots: &flight::Slots<'d, Tag>,
255        listener: &handles::Descriptor<'d>,
256        identity: route::Operation<'d, Tag>,
257    ) -> Result<flight::Flight<'d>, driver::SubmitError>;
258
259    fn cancel<Tag: route::Tag>(
260        &mut self,
261        flight: &mut flight::Flight<'d>,
262        target: route::Operation<'d, Tag>,
263    ) -> Result<(), driver::SubmitError>;
264}
265
266impl Buffers for driver::Context<'_, '_> {
267    fn buffer_len(&self) -> usize {
268        self.driver_ref().receive().layout().buffer_len()
269    }
270
271    fn buffer_count(&self) -> usize {
272        self.driver_ref().receive().layout().entries() as usize
273    }
274
275    fn accept_capacity(&self) -> usize {
276        self.accept_slot_capacity()
277    }
278
279    fn outbound_capacity(&self) -> usize {
280        self.outbound_slot_capacity()
281    }
282}
283
284impl<'d> Control<'d> for driver::Context<'_, 'd>
285where
286    backend::Backend: backend::Socket,
287{
288    fn reserve_route(&mut self, id: u8) -> bool {
289        self.backend().routes.reserve(id)
290    }
291
292    fn release_route(&mut self, id: u8) {
293        self.backend().routes.release(id);
294    }
295
296    fn submit_socket<Tag: route::Tag>(
297        &mut self,
298        flights: &flight::Slots<'d, Tag>,
299        request: route::Bound<'d, Tag, handles::SocketSlot<'d>>,
300        socket: socket::StreamSpec,
301    ) -> Result<handles::CreatingSocket<'d>, driver::SubmitError> {
302        let (target, slot) = request.into_parts();
303        backend::Socket::submit_socket(self.backend(), flights, target, slot, socket)
304    }
305
306    fn submit_tuning<Tag: route::Tag>(
307        &mut self,
308        request: route::Bound<'d, Tag, handles::Descriptor<'d>>,
309        options: option::StreamOptions,
310    ) -> Result<option::Tuning<'d>, handles::Descriptor<'d>> {
311        let (target, fd) = request.into_parts();
312        backend::Socket::submit_tuning(self.backend(), target, fd, options)
313    }
314
315    fn cancel_tuning(
316        &mut self,
317        mut pending: establishment::TuningPending<'d>,
318    ) -> Result<
319        establishment::TuningPending<'d, establishment::Cancelled>,
320        (establishment::TuningPending<'d>, driver::SubmitError),
321    > {
322        let target = establishment::CancelTarget::Tuning(&mut pending.fd);
323        match backend::Socket::cancel_establishment(self.backend(), target) {
324            Ok(()) => Ok(establishment::TuningPending {
325                fd: pending.fd,
326                target: pending.target,
327                state: marker::PhantomData,
328            }),
329            Err(error) => Err((pending, error)),
330        }
331    }
332
333    fn cancel_connection(
334        &mut self,
335        mut pending: establishment::ConnectionPending<'d>,
336    ) -> Result<
337        establishment::ConnectionPending<'d, establishment::Cancelled>,
338        (establishment::ConnectionPending<'d>, driver::SubmitError),
339    > {
340        let target = match pending.flight.as_mut() {
341            Some(flight) => establishment::CancelTarget::Connect(flight),
342            None => establishment::CancelTarget::Tuning(&mut pending.fd),
343        };
344        match backend::Socket::cancel_establishment(self.backend(), target) {
345            Ok(()) => Ok(establishment::ConnectionPending {
346                fd: pending.fd,
347                target: pending.target,
348                flight: pending.flight,
349                state: marker::PhantomData,
350            }),
351            Err(error) => Err((pending, error)),
352        }
353    }
354}
355
356impl<'d> Files<'d> for driver::Context<'_, 'd> {
357    fn close(&mut self, fd: impl Into<handles::Descriptor<'d>>) {
358        fd.into().close(self.backend());
359    }
360
361    fn reserve_outbound<const ID: u8>(
362        &mut self,
363        count: u32,
364    ) -> io::Result<OutboundReservation<'d, ID>> {
365        if count == 0 {
366            return Err(io::Error::new(
367                io::ErrorKind::InvalidInput,
368                "dope: outbound reservation must be non-empty",
369            ));
370        }
371        let driver = self.driver_ref();
372        let files = driver.files();
373        if ID == route::FRAMEWORK {
374            return Err(io::Error::new(
375                io::ErrorKind::InvalidInput,
376                "dope: framework route cannot own outbound slots",
377            ));
378        }
379        if files.has_outbound_route::<ID>() {
380            return Err(io::Error::new(
381                io::ErrorKind::AlreadyExists,
382                "dope: route already owns outbound slots",
383            ));
384        }
385        let slots = fixed::Lifecycle::alloc_slots(self.backend(), count, driver)
386            .map_err(|error| files.map_fixed_allocation_error(error))?;
387        let key = match files.track_outbound_slots::<ID>(slots) {
388            Ok(key) => key,
389            Err(slots) => {
390                fixed::Lifecycle::release_slots(self.backend(), slots);
391                return Err(io::Error::new(
392                    io::ErrorKind::InvalidData,
393                    "dope: outbound slots could not be tracked",
394                ));
395            }
396        };
397        Ok(OutboundReservation::new(driver, key))
398    }
399
400    fn retire_outbound<const ID: u8>(&mut self, reservation: OutboundReservation<'d, ID>) {
401        if let Some(retired) = reservation.retire() {
402            let slots = self.driver_ref().outbound().take_retired_slots(retired);
403            fixed::Lifecycle::release_slots(self.backend(), slots);
404        }
405    }
406}
407
408impl<'d> Submit<'d> for driver::Context<'_, 'd>
409where
410    backend::Backend: reactor::Source,
411{
412    fn submit_recv<Tag: route::Tag>(
413        &mut self,
414        slots: &flight::Slots<'d, Tag>,
415        fd: &handles::Descriptor<'d>,
416        target: route::Target<'d, Tag>,
417    ) -> Result<flight::Flight<'d>, driver::SubmitError> {
418        let raw = backend::Copied::recv(fd);
419        submit_nonretaining(self, slots, raw, target.operation(kind::RECV))
420    }
421
422    fn submit_recv_datagram<Tag: route::Tag>(
423        &mut self,
424        slots: &flight::Slots<'d, Tag>,
425        fd: &handles::DatagramDescriptor<'d>,
426        identity: route::Operation<'d, Tag>,
427    ) -> Result<flight::Flight<'d>, driver::SubmitError> {
428        let raw = backend::Copied::recv_datagram(fd);
429        submit_nonretaining(self, slots, raw, identity.with_kind(kind::RECV))
430    }
431
432    fn submit_accept_multishot<Tag: route::Tag>(
433        &mut self,
434        slots: &flight::Slots<'d, Tag>,
435        listener: &handles::Descriptor<'d>,
436        identity: route::Operation<'d, Tag>,
437    ) -> Result<flight::Flight<'d>, driver::SubmitError> {
438        let raw = backend::Copied::accept_multishot(listener);
439        submit_nonretaining(self, slots, raw, identity.with_kind(kind::ACCEPT))
440    }
441
442    fn cancel<Tag: route::Tag>(
443        &mut self,
444        flight: &mut flight::Flight<'d>,
445        target: route::Operation<'d, Tag>,
446    ) -> Result<(), driver::SubmitError> {
447        if !flight.matches(target.into_token()) {
448            return Err(driver::SubmitError);
449        }
450        let mut queue = reactor::Source::queue(self.backend());
451        reactor::Queue::cancel(&mut queue, flight)
452    }
453}
454
455fn submit_nonretaining<'d, Tag: route::Tag>(
456    context: &mut driver::Context<'_, 'd>,
457    slots: &flight::Slots<'d, Tag>,
458    copied: backend::Copied,
459    target: route::Operation<'d, Tag>,
460) -> Result<flight::Flight<'d>, driver::SubmitError>
461where
462    backend::Backend: reactor::Source,
463{
464    let submission =
465        bound::Bound::reserve(copied.into_raw(), target, slots).ok_or(driver::SubmitError)?;
466    let mut queue = reactor::Source::queue(context.backend());
467    reactor::Queue::submit(&mut queue, submission)
468}
dope_core/driver/ops/mod.rs

dope_core/driver/ops/
mod.rs
