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

dope_uring/
uring.rs
