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 { 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 route::Source for Uring {
174    type Install = Installer;
175
176    fn try_prepare(plan: route::Plan) -> Result<Self::Install, install::Failure> {
177        Installer::try_new(plan).map_err(install::Failure::Syscall)
178    }
179}
180
181impl gso::Source for Uring {
182    type Mode = Control;
183
184    const LIMITS: Option<(
185        bounded::U32<1, { MAX as u32 }>,
186        bounded::U32<2, { u16::MAX as u32 }>,
187    )> = Some((bounded::U32::exact::<65_535>(), bounded::U32::exact::<64>()));
188}
189
190impl kernel::Source for Uring {
191    type Code = op::Code;
192    type Failure<S: Copy + Eq> = error::Failure<S>;
193    type Returned = error::Returned;
194}
195
196impl Thread for Uring {
197    type Timer = Timer;
198    type Receiver = signal::raw::Receiver;
199    type Cpus = Cpus;
200
201    fn is_main() -> bool {
202        // SAFETY: both calls are pure identity queries and do not dereference
203        // any caller-provided pointer.
204        unsafe { libc::syscall(libc::SYS_gettid) == libc::getpid() as libc::c_long }
205    }
206
207    fn current() -> Result<Self::Cpus, affinity::Error> {
208        // SAFETY: zero gives the kernel initialized storage even when the CPU
209        // set representation contains padding.
210        let mut mask: libc::cpu_set_t = unsafe { mem::zeroed() };
211        // SAFETY: mask points to writable storage of the exact size supplied.
212        let result =
213            unsafe { libc::sched_getaffinity(0, mem::size_of::<libc::cpu_set_t>(), &mut mask) };
214        // SAFETY: sched_getaffinity returns zero or -1 with errno.
215        unsafe { kernel::raw::Status::new(result).try_zero::<Uring, Errno>() }
216            .map_err(affinity::Error::Current)?;
217        Ok(Cpus::new(&mask))
218    }
219
220    fn bind(cpu: affinity::raw::Cpu) -> Result<(), affinity::Error> {
221        const MAX: u32 = Cpus::LIMIT as u32 - 1;
222        let index = bounded::U32::<0, MAX>::new(u32::from(cpu.raw())).ok_or(
223            affinity::Error::BindingRange {
224                cpu,
225                maximum: MAX as u16,
226            },
227        )?;
228        // SAFETY: zero is the empty set.
229        let mut mask: libc::cpu_set_t = unsafe { mem::zeroed() };
230        // SAFETY: index retains this native mask's admitted bound and mask is initialized.
231        unsafe { libc::CPU_SET(index.into_usize(), &mut mask) };
232        // SAFETY: mask is an initialized cpu_set_t of the exact ABI size.
233        let result =
234            unsafe { libc::sched_setaffinity(0, mem::size_of::<libc::cpu_set_t>(), &mask) };
235        // SAFETY: sched_setaffinity returns zero or -1 with errno.
236        unsafe { kernel::raw::Status::new(result).try_zero::<Uring, Errno>() }
237            .map_err(|fault| affinity::Error::Bind { cpu, fault })?;
238        Ok(())
239    }
240}
241
242impl errno::Source for Uring {
243    fn clear() {
244        // SAFETY: this native TLS cell is written synchronously on its owning thread.
245        unsafe { libc::__errno_location().write(0) };
246    }
247}
248
249impl pipe::Pipe for Uring {
250    fn open() -> Result<(OwnedFd, OwnedFd), pipe::Error> {
251        Self::open_with(libc::O_CLOEXEC)
252    }
253
254    fn nonblocking() -> Result<(OwnedFd, OwnedFd), pipe::Error> {
255        Self::open_with(libc::O_CLOEXEC | libc::O_NONBLOCK)
256    }
257}
258
259impl permit::Source for Uring {
260    type Queue<K: permit::raw::Key, I: depot::raw::Compact> = Fifo<K, I>;
261}
262
263impl provided::Source for Uring {
264    const PROVIDED: u16 = 1;
265    const CANARY: u16 = 2;
266    type Entry = Entry;
267    type Index = cq::Buffer;
268    type Context<L: memory::raw::Receive> = buffers::raw::Context<L>;
269    type Canary<L: memory::raw::Canary> = Canary<L>;
270    type Cleanup<L: memory::raw::Receive> = Cleanup<L>;
271    type Producer<L: memory::raw::Receive> = Producer<L>;
272    type Pending<E: memory::raw::Extent> = Selected<E, E::Capacity>;
273    type Complete<E: memory::raw::Extent> = Selected<E, E::Length>;
274}
275
276impl file::File for Uring {
277    type Options = types::OpenHow;
278    type Metadata = crate::raw::Metadata;
279    type Request<'a> = Borrowed<'a>;
280}
281
282impl socket::Socket for Uring {
283    const MAX_IOVECS: usize = libc::UIO_MAXIOV as usize;
284    type Error = Errno;
285
286    fn socket(domain: libc::c_int, kind: libc::c_int) -> Result<OwnedFd, Self::Error> {
287        Self::try_open(domain, kind | libc::SOCK_CLOEXEC, 0)
288    }
289
290    fn pair() -> Result<(OwnedFd, OwnedFd), record::Error> {
291        Self::try_pair::<Errno>(libc::SOCK_SEQPACKET | libc::SOCK_CLOEXEC)
292            .map_err(record::Error::Open)
293    }
294}
295
296impl addr::Source for Uring {
297    const UNIX_TERMINATED: bool = false;
298    const UNIX_ABSTRACT: bool = true;
299
300    fn header(domain: socket::Domain, _: addr::Socklen) -> [u8; 2] {
301        (domain.raw() as u16).to_ne_bytes()
302    }
303}
304
305impl option::Source for Uring {
306    const KEEP_IDLE: libc::c_int = libc::TCP_KEEPIDLE;
307}
308
309impl socket::Prepare for Uring {
310    type Handle = OwnedFd;
311    type Fast = num::NonZero<i32>;
312
313    fn fast_open(depth: Option<num::NonZero<i32>>) -> Result<Option<Self::Fast>, fast::Error> {
314        Ok(depth)
315    }
316
317    fn open(domain: socket::Domain, kind: socket::Kind) -> Result<Self::Handle, socket::Create> {
318        Self::socket(domain.raw(), kind.raw()).map_err(socket::Create::socket)
319    }
320
321    fn listener(domain: socket::Domain) -> Result<Self::Handle, socket::Create> {
322        Self::open(domain, socket::Kind::Stream)
323    }
324
325    fn adopt(file: OwnedFd) -> Result<Self::Handle, socket::Setup> {
326        <Self as Fd>::update::<Cloexec>(file.as_fd()).map_err(Setup::cloexec)?;
327        <Self as Fd>::remove::<Nonblock>(file.as_fd()).map_err(Setup::nonblock)?;
328        Ok(file)
329    }
330
331    fn into_fd(handle: Self::Handle) -> OwnedFd {
332        handle
333    }
334
335    fn set_fast(file: BorrowedFd<'_>, fast: Self::Fast) -> Result<(), Errno> {
336        Self::set_option(file, libc::IPPROTO_TCP, libc::TCP_FASTOPEN, &fast.get())
337    }
338}
339impl<T: Copy + Eq, S: Copy + Eq, F: Copy + Eq> Observe<T, S, F> for Uring {
340    type Observation = crate::Observed<T>;
341}
342impl tuning::Source for Uring {
343    type State<E: Copy, const N: usize> = crate::State<E, N>;
344    type Table<E: Copy, const N: usize> = Table<E, N>;
345}
346
347impl ring::Source for Uring {
348    type Decoded<T: Copy, E> = packet::Decoded<T, E, ()>;
349    type Ready<T: complete::raw::Token, L: memory::raw::Receive> = Ready<T, L>;
350    type Packet<T: Copy, B> = packet::Packet<T, B>;
351    type Poll = enter::Poll<kernel::Errno<error::Enter>, kernel::Errno<error::Register<24>>>;
352    type Enter<'a, T: 'a> = Enter<'a, T>;
353    type Entry<T> = cq::Entry<T>;
354    type Message<'a, const NAME: usize, const CONTROL: usize> = cq::Output<'a, NAME, CONTROL>;
355    type Ring<T: complete::raw::Token> = Ring<T>;
356}
357
358impl submit::Control for Uring {
359    type Target = sq::Fixed;
360    type Owned<T: complete::raw::Token> = sq::Owned<T>;
361    type Poll<'a, T: complete::raw::Token> = Poll<'a, T>;
362
363    fn nop<T: complete::raw::Token>(complete: T) -> Self::Owned<T> {
364        sq::Entry::new(sq::Entry::NOP, sq::Descriptor::File(-1)).user_data(complete)
365    }
366
367    fn cancel<T: complete::raw::Token>(target: T, complete: T) -> Self::Owned<T> {
368        use crate::reactor::op::controls;
369
370        controls::cancel(target, false)
371            .flags(sq::Flags::SKIP_SUCCESS)
372            .user_data(complete)
373    }
374
375    fn close<T: complete::raw::Token>(slot: Self::Target, complete: T) -> Self::Owned<T> {
376        use crate::reactor::op::files;
377
378        let mut entry = sq::Entry::new(files::CLOSE, sq::Descriptor::File(0));
379        entry.fields().index(slot.encode());
380        entry.user_data(complete)
381    }
382
383    fn shutdown<T: complete::raw::Token>(slot: Self::Target, complete: T) -> Self::Owned<T> {
384        use crate::reactor::op::sockets;
385
386        sockets::shutdown(slot, socket::How::BOTH)
387            .flags(sq::Flags::HARD_LINK | sq::Flags::SKIP_SUCCESS)
388            .user_data(complete)
389    }
390}
391
392impl probe::Source for Uring {
393    type Range = Range;
394    type Release<T: token::Codec, L: memory::raw::Canary, P: Copy> = Released<T, L, P>;
395    type Cleanup<T: token::Codec, L: memory::raw::Canary, P: Copy> = Recover<T, L, P>;
396    type Payload = Payload;
397    type Accept<T: token::Codec> = Accepted<T>;
398
399    type Session<T: token::Codec, L: memory::raw::Canary> = Session<T, L>;
400
401    type Outcome<T: complete::raw::Token, P: Copy> = Outcome<T, P>;
402    type Waiting<T: token::Codec, S> = Waiting<S, T>;
403}
404
405impl create::Source for Uring {
406    type Primary = Primary;
407    type Failure<T: token::Codec> = Fault<T>;
408    type Resources<
409        T: token::Codec,
410        L: memory::raw::Canary,
411        P: Copy + From<Self::Primary> + From<memory::Error>,
412    > = Resources<T, L, P>;
413}
414
415impl token::Source for Uring {
416    type Token<C: host::Context> = Encoded<C>;
417}
418
419impl Source for Uring {
420    type Socket<H> = Operation<H>;
421}
422
423impl admit::Native for Uring {
424    type Error<F> = Infallible;
425}
426
427impl<'a, C, R> life::Finish<'a, R> for Uring
428where
429    C: Context + 'a,
430    C::Tuning: Progress<C::Token>,
431    R: life::Reclaim<Backend = Engine<C>> + 'a,
432{
433    type Quiesced = Quiesced<R>;
434
435    fn try_finish(
436        resolver: R,
437    ) -> Result<Self::Quiesced, Stopped<R, Error<<Self as ring::Source>::Poll, R::Fault>>> {
438        Engine::try_finish(resolver)
439    }
440}
441
442impl<'a, C, R> life::Quiesce<'a, R> for Uring
443where
444    C: Context + 'a,
445    C::Tuning: Progress<C::Token>,
446    R: life::Cancel<Backend = Engine<C>> + 'a,
447    R::Cancelling: 'a,
448    R::Closing: 'a,
449{
450    type Quiesced = Quiesced<R::Closing>;
451
452    fn try_quiesce(
453        resolver: R,
454    ) -> Result<Self::Quiesced, Stopped<R::Closing, Error<<Self as ring::Source>::Poll, R::Fault>>>
455    {
456        use life::{Close as _, Finish as _, Resolver as _};
457
458        let mut resolver = resolver.cancel();
459        resolver.backend().tuning().cancel_all();
460        let result = resolver.backend().ring().submit().map_err(Error::File);
461        (resolver, ()) = Stopped::attach(resolver, result)
462            .map_err(|stopped| stopped.map(life::Close::into_closing))?;
463        let result = resolver.backend().ring().cancel_all().map_err(Error::File);
464        (resolver, ()) = Stopped::attach(resolver, result)
465            .map_err(|stopped| stopped.map(life::Close::into_closing))?;
466        resolver =
467            Engine::drain(resolver).map_err(|stopped| stopped.map(life::Close::into_closing))?;
468        if !resolver.backend().tuning().is_quiescent() {
469            return Err(Stopped {
470                state: resolver.into_closing(),
471                cause: Error::Tuning,
472            });
473        }
474        Self::try_finish(resolver.into_closing())
475    }
476}
dope_uring/uring.rs

dope_uring/
uring.rs
