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

dope_uring/
uring.rs
