dope_uring/
uring.rs

1use std::{
2    alloc::Layout,
3    os::fd::{AsFd, BorrowedFd, OwnedFd},
4    ptr,
5};
6
7use dope_sys::{
8    Cloexec, Nonblock,
9    engine::{life, life::Error, ring::Backend as _},
10    io::{
11        fd::{self, Fd},
12        socket::{
13            Setup, Socket as _, addr,
14            option::{Name, tuning::Quiesce as _},
15        },
16    },
17    probe::Stopped,
18    ring::ready::Ready as _,
19    thread::Thread,
20};
21use o3::depot;
22
23use crate::{
24    Accepted, Borrowed, Canary, Cleanup, Closing, Context, Control, Cpus, Encoded, Engine, Enter,
25    Entry, Errno, Fault, Fifo, Infallible, Installer, MAX, Observe, Operation, Outcome, Payload,
26    Poll, Primary, Producer, Range, Ready, Recover, Released, Resources, Ring, Selected, Session,
27    Source, Table, Timer, Turn, Waiting, accept, admit, affinity, bounded, buffer, buffers,
28    complete, cq, create, delivery,
29    driver::{Quiesced, tuning::Progress},
30    engine, enter, errno, error, fast, file, gso, host, install, kernel, mem, memory, num, op,
31    option, packet, permit, pipe, probe, provided, record, ring, route, socket, sq, submit,
32    system::signal,
33    token, tuning, types,
34};
35
36#[derive(Debug)]
37pub struct Uring;
38
39impl Uring {
40    const ALIGNMENT: num::NonZero<usize> = num::NonZero::<usize>::MIN.saturating_add(4095);
41
42    fn open_with(flags: libc::c_int) -> Result<(OwnedFd, OwnedFd), pipe::Error> {
43        let mut fds = [0; 2];
44        // SAFETY: fds names writable storage for exactly two descriptors.
45        let result = unsafe { libc::pipe2(fds.as_mut_ptr(), flags) };
46        // SAFETY: pipe2 just returned on this thread and `fds` is its exact destination.
47        unsafe {
48            fd::raw::Pair::new(fds)
49                .try_adopt::<Errno>(result, || kernel::raw::Code::last::<Self>().into())
50        }
51        .map_err(pipe::Error::Open)
52    }
53
54    pub fn layout<T>(count: usize) -> Result<num::NonZero<usize>, memory::Error> {
55        let invalid = || memory::Error::Length {
56            element: mem::size_of::<T>(),
57            count,
58        };
59        let layout = Layout::array::<T>(count).map_err(|_| invalid())?;
60        let bytes = num::NonZero::new(layout.size()).ok_or_else(invalid)?;
61        let element = layout.align();
62        let limit = Self::ALIGNMENT.get();
63        if element > limit {
64            return Err(memory::Error::Alignment {
65                element,
66                supported: limit,
67            });
68        }
69        Ok(bytes)
70    }
71
72    /// # Safety
73    /// `raw` is the unchanged return of a syscall whose success fits `0..=i32::MAX`.
74    pub(crate) unsafe fn status(raw: libc::c_long) -> kernel::raw::Status<-1, { i32::MAX }> {
75        const { assert!(mem::size_of::<libc::c_long>() == mem::size_of::<libc::ssize_t>()) };
76        // SAFETY: the caller retains this syscall's nonnegative c_int success
77        // domain or native -1; the size assertion preserves that exact scalar.
78        unsafe { kernel::raw::Status::from_count(raw as libc::ssize_t) }
79    }
80
81    /// Transfers an anonymous writable mapping, including a successful null address.
82    pub(crate) fn try_map<const FLAGS: libc::c_int>(
83        bytes: num::NonZero<usize>,
84    ) -> Result<*mut libc::c_void, Errno> {
85        const { assert!(FLAGS & !libc::MAP_NORESERVE == 0) };
86        // SAFETY: mmap validates this nonempty anonymous mapping request.
87        let address = unsafe {
88            libc::mmap(
89                ptr::null_mut(),
90                bytes.get(),
91                libc::PROT_READ | libc::PROT_WRITE,
92                libc::MAP_PRIVATE | libc::MAP_ANONYMOUS | FLAGS,
93                -1,
94                0,
95            )
96        };
97        if address == libc::MAP_FAILED {
98            // SAFETY: mmap just failed on this thread without an intervening call.
99            Err(unsafe { kernel::raw::Code::last::<Self>().into() })
100        } else {
101            Ok(address)
102        }
103    }
104
105    /// # Safety
106    /// `address` and `bytes` name one complete live mapping owned by the caller.
107    /// Success consumes the mapping; failure leaves ownership with the caller.
108    pub(crate) unsafe fn try_unmap<E: From<Errno>>(
109        address: *mut libc::c_void,
110        bytes: num::NonZero<usize>,
111    ) -> Result<(), E> {
112        // SAFETY: the caller owns this complete live mapping.
113        let result = unsafe { libc::munmap(address, bytes.get()) };
114        // SAFETY: munmap returns zero or -1; failure retains its unchanged errno.
115        unsafe {
116            kernel::raw::Status::new(result).try_zero(|| kernel::raw::Code::last::<Self>().into())
117        }
118    }
119}
120
121impl accept::Source for Uring {
122    type Failure<F: From<Errno>> = Errno;
123
124    const MAX_IN_FLIGHT: num::NonZero<usize> = num::NonZero::<usize>::MAX;
125
126    fn is_retryable(error: Errno) -> bool {
127        error.is_block()
128            || matches!(
129                error.code().get(),
130                libc::EINTR
131                    | libc::ECONNABORTED
132                    | libc::ENETDOWN
133                    | libc::EPROTO
134                    | libc::ENOPROTOOPT
135                    | libc::EHOSTDOWN
136                    | libc::ENONET
137                    | libc::EHOSTUNREACH
138                    | libc::EOPNOTSUPP
139                    | libc::ENETUNREACH
140            )
141    }
142
143    fn can_defer(error: Errno) -> bool {
144        error.code().get() == libc::ENFILE
145    }
146}
147
148impl<L: Copy + Eq> buffer::Selection<L> for Uring {
149    type Error = (<Self as provided::Source>::Index, L);
150}
151
152impl<C: Context> engine::Source<C> for Uring {
153    type Map = C::Map;
154    type Backend = Engine<C>;
155    type Closing = Closing<C>;
156}
157
158impl<C, P, R> engine::Storage<C, P, R> for Uring
159where
160    C: Context<Layout: memory::raw::Canary>,
161    P: Copy + From<Primary> + From<memory::Error>,
162    R: Copy,
163{
164    type Cleanup = <Self as create::Source>::Resources<C::Token, C::Layout, P>;
165    type Release = <Self as probe::Source>::Release<C::Token, C::Layout, R>;
166}
167
168impl engine::Removal for Uring {
169    type Slot = <Self as submit::Control>::Target;
170}
171
172impl<'a, C, H> engine::Dispatch<'a, C, H> for Uring
173where
174    C: Context + delivery::Context<H, Backend = Engine<C>>,
175    C::Tuning: Progress<C::Token>,
176    H: 'a,
177{
178    type Turn<Q: delivery::Quota<H>> = Turn<'a, H, Q>;
179}
180
181impl route::Source for Uring {
182    type Install = Installer;
183
184    fn try_prepare(plan: route::Plan) -> Result<Self::Install, install::Failure> {
185        Installer::try_new(plan).map_err(install::Failure::Syscall)
186    }
187}
188
189impl gso::Source for Uring {
190    type Mode = Control;
191
192    const LIMITS: Option<(
193        bounded::U32<1, { MAX as u32 }>,
194        bounded::U32<2, { u16::MAX as u32 }>,
195    )> = Some((bounded::U32::exact::<65_535>(), bounded::U32::exact::<64>()));
196}
197
198impl kernel::Source for Uring {
199    type Code = op::Code;
200    type Failure<S: Copy + Eq> = error::Failure<S>;
201    type Returned = error::Returned;
202}
203
204impl Thread for Uring {
205    type Timer = Timer;
206    type Receiver = signal::raw::Receiver;
207    type Cpus = Cpus;
208
209    fn is_main() -> bool {
210        // SAFETY: both calls are pure identity queries and do not dereference
211        // any caller-provided pointer.
212        unsafe { libc::syscall(libc::SYS_gettid) == libc::getpid() as libc::c_long }
213    }
214
215    fn current() -> Result<Self::Cpus, affinity::Error> {
216        // SAFETY: zero gives the kernel initialized storage even when the CPU
217        // set representation contains padding.
218        let mut mask: libc::cpu_set_t = unsafe { mem::zeroed() };
219        // SAFETY: mask points to writable storage of the exact size supplied.
220        let result =
221            unsafe { libc::sched_getaffinity(0, mem::size_of::<libc::cpu_set_t>(), &mut mask) };
222        // SAFETY: sched_getaffinity returns zero or -1 with errno.
223        unsafe {
224            kernel::raw::Status::new(result)
225                .try_zero::<Errno>(|| kernel::raw::Code::last::<Uring>().into())
226        }
227        .map_err(affinity::Error::Current)?;
228        Ok(Cpus::new(&mask))
229    }
230
231    fn bind(cpu: affinity::raw::Cpu) -> Result<(), affinity::Error> {
232        const MAX: u32 = Cpus::LIMIT as u32 - 1;
233        let index = bounded::U32::<0, MAX>::new(u32::from(cpu.raw())).ok_or(
234            affinity::Error::BindingRange {
235                cpu,
236                maximum: MAX as u16,
237            },
238        )?;
239        // SAFETY: zero is the empty set.
240        let mut mask: libc::cpu_set_t = unsafe { mem::zeroed() };
241        // SAFETY: index retains this native mask's admitted bound and mask is initialized.
242        unsafe { libc::CPU_SET(index.into_usize(), &mut mask) };
243        // SAFETY: mask is an initialized cpu_set_t of the exact ABI size.
244        let result =
245            unsafe { libc::sched_setaffinity(0, mem::size_of::<libc::cpu_set_t>(), &mask) };
246        // SAFETY: sched_setaffinity returns zero or -1 with errno.
247        unsafe {
248            kernel::raw::Status::new(result)
249                .try_zero::<Errno>(|| kernel::raw::Code::last::<Uring>().into())
250        }
251        .map_err(|fault| affinity::Error::Bind { cpu, fault })?;
252        Ok(())
253    }
254}
255
256impl errno::Source for Uring {
257    fn clear() {
258        // SAFETY: this native TLS cell is written synchronously on its owning thread.
259        unsafe { libc::__errno_location().write(0) };
260    }
261}
262
263impl pipe::Pipe for Uring {
264    fn open() -> Result<(OwnedFd, OwnedFd), pipe::Error> {
265        Self::open_with(libc::O_CLOEXEC)
266    }
267
268    fn nonblocking() -> Result<(OwnedFd, OwnedFd), pipe::Error> {
269        Self::open_with(libc::O_CLOEXEC | libc::O_NONBLOCK)
270    }
271}
272
273impl permit::Source for Uring {
274    type Queue<K: permit::raw::Key, I: depot::raw::Compact> = Fifo<K, I>;
275}
276
277impl provided::Source for Uring {
278    const PROVIDED: u16 = 1;
279    const CANARY: u16 = 2;
280    type Entry = Entry;
281    type Index = cq::Buffer;
282    type Context<L: memory::raw::Receive> = buffers::raw::Context<L>;
283    type Canary<L: memory::raw::Canary> = Canary<L>;
284    type Cleanup<L: memory::raw::Receive> = Cleanup<L>;
285    type Producer<L: memory::raw::Receive> = Producer<L>;
286    type Pending<E: memory::raw::Extent> = Selected<E, E::Capacity>;
287    type Complete<E: memory::raw::Extent> = Selected<E, E::Length>;
288}
289
290impl file::File for Uring {
291    type Options = types::OpenHow;
292    type Request<'a> = Borrowed<'a>;
293}
294
295impl socket::Socket for Uring {
296    const MAX_IOVECS: usize = libc::UIO_MAXIOV as usize;
297    type Error = Errno;
298    type Records = record::Packet;
299
300    fn socket(domain: libc::c_int, kind: libc::c_int) -> Result<OwnedFd, Self::Error> {
301        Self::try_open(domain, kind | libc::SOCK_CLOEXEC, 0)
302    }
303
304    fn pair() -> Result<(OwnedFd, OwnedFd), record::Error> {
305        Self::try_pair::<Errno>(libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC)
306            .map_err(record::Error::Open)
307    }
308}
309
310impl addr::Source for Uring {
311    const UNIX_TERMINATED: bool = false;
312    const UNIX_ABSTRACT: bool = true;
313
314    fn header(domain: socket::Domain, _: addr::Socklen) -> [u8; 2] {
315        (domain.raw() as u16).to_ne_bytes()
316    }
317}
318
319impl option::Source for Uring {
320    const KEEP_IDLE: Name = Name::new(libc::TCP_KEEPIDLE);
321}
322
323impl socket::Prepare for Uring {
324    type Handle = OwnedFd;
325    type Fast = num::NonZero<i32>;
326
327    fn fast_open(depth: Option<num::NonZero<i32>>) -> Result<Option<Self::Fast>, fast::Error> {
328        Ok(depth)
329    }
330
331    fn open(domain: socket::Domain, kind: socket::Kind) -> Result<Self::Handle, socket::Create> {
332        Self::socket(domain.raw(), kind.raw()).map_err(socket::Create::socket)
333    }
334
335    fn listener(domain: socket::Domain) -> Result<Self::Handle, socket::Create> {
336        Self::open(domain, socket::Kind::Stream)
337    }
338
339    fn adopt(file: OwnedFd) -> Result<Self::Handle, socket::Setup> {
340        <Self as Fd>::update::<Cloexec>(file.as_fd()).map_err(Setup::cloexec)?;
341        <Self as Fd>::remove::<Nonblock>(file.as_fd()).map_err(Setup::nonblock)?;
342        Ok(file)
343    }
344
345    fn into_fd(handle: Self::Handle) -> OwnedFd {
346        handle
347    }
348
349    fn set_fast(file: BorrowedFd<'_>, fast: Self::Fast) -> Result<(), Errno> {
350        Self::set_option(file, libc::IPPROTO_TCP, libc::TCP_FASTOPEN, &fast.get())
351    }
352}
353impl file::Completion for Uring {
354    type Opened = crate::Opened;
355}
356
357impl<T: Copy + Eq, S: Copy + Eq, F: Copy + Eq> Observe<T, S, F> for Uring {
358    type Observation = crate::Observed<T>;
359}
360impl tuning::Source for Uring {
361    type State<E: Copy, const N: usize> = crate::State<E, N>;
362    type Table<E: Copy, const N: usize> = Table<E, N>;
363}
364
365impl ring::Source for Uring {
366    type Decoded<T: Copy, E> = packet::Decoded<T, E, ()>;
367    type Ready<T: complete::raw::Token, L: memory::raw::Receive> = Ready<T, L>;
368    type Packet<T: Copy, B> = packet::Packet<T, B>;
369    type Poll = enter::Poll<kernel::Errno<error::Enter>, kernel::Errno<error::Register<24>>>;
370    type Enter<'a, T: 'a> = Enter<'a, T>;
371    type Entry<T> = cq::Entry<T>;
372    type Message<'a, const NAME: usize, const CONTROL: usize> = cq::Output<'a, NAME, CONTROL>;
373    type Ring<T: complete::raw::Token> = Ring<T>;
374}
375
376impl submit::Control for Uring {
377    type Target = sq::Fixed;
378    type Owned<T: complete::raw::Token> = sq::Owned<T>;
379    type Poll<'a, T: complete::raw::Token> = Poll<'a, T>;
380
381    fn nop<T: complete::raw::Token>(complete: T) -> Self::Owned<T> {
382        sq::Entry::new(sq::Entry::NOP, sq::Descriptor::File(-1)).user_data(complete)
383    }
384
385    fn cancel<T: complete::raw::Token>(target: T, complete: T) -> Self::Owned<T> {
386        use crate::reactor::op::controls;
387
388        controls::cancel(target, false)
389            .flags(sq::Flags::SKIP_SUCCESS)
390            .user_data(complete)
391    }
392
393    fn close<T: complete::raw::Token>(slot: Self::Target, complete: T) -> Self::Owned<T> {
394        use crate::reactor::op::files;
395
396        let mut entry = sq::Entry::new(files::CLOSE, sq::Descriptor::File(0));
397        entry.fields().index(slot.encode());
398        entry.user_data(complete)
399    }
400
401    fn shutdown<T: complete::raw::Token>(slot: Self::Target, complete: T) -> Self::Owned<T> {
402        use crate::reactor::op::sockets::{self, Direction};
403
404        sockets::shutdown(slot, Direction::Both)
405            .flags(sq::Flags::HARD_LINK | sq::Flags::SKIP_SUCCESS)
406            .user_data(complete)
407    }
408}
409
410impl probe::Source for Uring {
411    type Range = Range;
412    type Release<T: token::Codec, L: memory::raw::Canary, P: Copy> = Released<T, L, P>;
413    type Cleanup<T: token::Codec, L: memory::raw::Canary, P: Copy> = Recover<T, L, P>;
414    type Payload = Payload;
415    type Accept<T: token::Codec> = Accepted<T>;
416
417    type Session<T: token::Codec, L: memory::raw::Canary> = Session<T, L>;
418
419    type Outcome<T: complete::raw::Token, P: Copy> = Outcome<T, P>;
420    type Waiting<T: token::Codec, S> = Waiting<S, T>;
421}
422
423impl create::Source for Uring {
424    type Primary = Primary;
425    type Failure<T: token::Codec> = Fault<T>;
426    type Resources<
427        T: token::Codec,
428        L: memory::raw::Canary,
429        P: Copy + From<Self::Primary> + From<memory::Error>,
430    > = Resources<T, L, P>;
431}
432
433impl token::Source for Uring {
434    type Token<C: host::Context> = Encoded<C>;
435}
436
437impl Source for Uring {
438    type Socket<H> = Operation<H>;
439}
440
441impl admit::Native for Uring {
442    type Error<F> = Infallible;
443}
444
445impl<'a, C, R> life::Finish<'a, R> for Uring
446where
447    C: Context + 'a,
448    C::Tuning: Progress<C::Token>,
449    R: life::Reclaim<Backend = Engine<C>> + 'a,
450{
451    type Quiesced = Quiesced<R>;
452
453    fn try_finish(
454        resolver: R,
455    ) -> Result<Self::Quiesced, Stopped<R, Error<<Self as ring::Source>::Poll, R::Fault>>> {
456        Engine::try_finish(resolver)
457    }
458}
459
460impl<'a, C, R> life::Quiesce<'a, R> for Uring
461where
462    C: Context + 'a,
463    C::Tuning: Progress<C::Token>,
464    R: life::Cancel<Backend = Engine<C>> + 'a,
465    R::Cancelling: 'a,
466    R::Closing: 'a,
467{
468    type Quiesced = Quiesced<R::Closing>;
469
470    fn try_quiesce(
471        resolver: R,
472    ) -> Result<Self::Quiesced, Stopped<R::Closing, Error<<Self as ring::Source>::Poll, R::Fault>>>
473    {
474        use life::{Close as _, Finish as _, Resolver as _};
475
476        let mut resolver = resolver.cancel();
477        resolver.backend().tuning().cancel_all();
478        let result = resolver.backend().ring().submit().map_err(Error::File);
479        (resolver, ()) = Stopped::attach(resolver, result)
480            .map_err(|stopped| stopped.map(life::Close::into_closing))?;
481        let result = resolver.backend().ring().cancel_all().map_err(Error::File);
482        (resolver, ()) = Stopped::attach(resolver, result)
483            .map_err(|stopped| stopped.map(life::Close::into_closing))?;
484        resolver =
485            Engine::drain(resolver).map_err(|stopped| stopped.map(life::Close::into_closing))?;
486        if !resolver.backend().tuning().is_quiescent() {
487            return Err(Stopped {
488                state: resolver.into_closing(),
489                cause: Error::Tuning,
490            });
491        }
492        Self::try_finish(resolver.into_closing())
493    }
494}
dope_uring/uring.rs

dope_uring/
uring.rs
