nuclei 0.5.4

//! A shard's ring, as its own thread uses it: the poller, the operations in
//! flight, the source files, and the operations waiting for room.
//!
//! Every entry for a source names the source's file by its slot (see
//! `syscore::linux::iouring::files`): a fixed file of the ring's table
//! (`IOSQE_FIXED_FILE`), or past the table the source's own descriptor,
//! which the ring never copies. A source that goes ([`ShardRing::detach`])
//! has its operations cancelled and every entry naming its slot handed to
//! the kernel at once, which holds each request's file from then on, and its
//! file taken out of the table at once: so the kernel closes the file as the
//! handle's drop returns (once its requests let go of it), and the handle's
//! object, which a lease keeps open until then, closes only once no entry
//! that names its number waits to reach the kernel. The slot is handed out
//! again only once no entry names it.
//!
//! An operation gets its user data as it is submitted, and keeps it until
//! its completion: one can be cancelled by it at any time. A side has one
//! operation out at a time, so the operations in flight are bounded by the
//! sources, apart from the zero-copy sends below.
//!
//! A zero-copy send ends for its side at its first completion, and stays here
//! only to hold its write's buffer until the kernel's notification, which
//! comes once the peer acknowledged the bytes: the side goes on meanwhile, so
//! one side may leave several. Each holds at most a chunk the socket's send
//! buffer took, and none names its slot once it only waits. Only a ring a
//! keeper can take sends zero-copy: as its shard goes, what is still out goes
//! to the keeper with the ring (see `keeper`), so a buffer is kept until its
//! notification and no longer.
//!
//! A multishot receive stays in flight across its arrivals: each completion
//! that says more come (`IORING_CQE_F_MORE`) hands its buffer of the
//! receive group (see `recv`) to the source as a chunk, and only the last
//! ends the operation. Its arrivals are bounded by the group's buffers, so
//! the operations let into the kernel leave the completion queue room for
//! them.
//!
//! The group is sized to the ring: at most half its completion queue, so a
//! small ring (made so under memory pressure) is never filled by arrivals.
//! The rest of the queue is the cap on the finite operations in the kernel,
//! the files' reads and writes: past it they wait here, in order, entry and
//! memory together, and go in as completions free room. Every operation of a
//! source waits on its peer for as long as the peer pleases (a receive or an
//! accept until it sends or connects, a send or a poll for room until it
//! reads), and a child's exit watch on the child, so they go into the kernel
//! at once and never count against the cap: idle or slow peers would
//! otherwise hold the room another connection's reply, the next accept or a
//! file call needs (proofs/tla/admission). The multishot receives' arrivals
//! are bounded by the group's buffers, which the queue keeps room for. An
//! accept's connections (up to [`ACCEPT_CAP`] each), the edges of a
//! multishot poll armed by a side that reads by readiness, the ring's own
//! cancels and the completion of each of these operations have no room
//! kept: past the queue the kernel holds them (`IORING_FEAT_NODROP`, which
//! the poller requires) until the poller takes them, and none is lost. A
//! multishot poll's edge that finds the queue full ends the poll instead,
//! and the next task of its side to wait arms another, whose first look
//! sees bytes already there (proofs/tla/admission, `EndedRearmed`).

use core::cell::{Cell, RefCell};
use std::collections::VecDeque;
use std::io;
use std::os::fd::RawFd;
use std::sync::Arc;
use std::time::Duration;
#[cfg(feature = "io-stats")]
use std::time::Instant;

use super::Shard;
use super::recv::RecvGroup;
use crate::fs::ring::FileOp;

use crate::io::source::ops::Done;
use crate::io::source::requests::Lease;
use crate::io::source::ring::{RingMem, RingOp};
use crate::rt::backend::with_room;

#[cfg(feature = "io-stats")]
use crate::rt::counters::{IoKind, IoTimes};
use crate::rt::park::Notifier;
use crate::rt::park::ring::{CqeBatch, RingPoller};
use crate::syscore::linux::iouring::files::Files;
use crate::syscore::linux::iouring::ops::Ops;
use crate::syscore::linux::iouring::ring::sys::{
    IORING_CQE_F_BUF_MORE, IORING_CQE_F_BUFFER, IORING_CQE_F_MORE, IORING_CQE_F_NOTIF,
    IOSQE_FIXED_FILE, KernelTimespec, cqe_buffer_id,
};
use crate::syscore::linux::iouring::ring::{IoUringCqe, RingConfig, Sqe};
use exits::ExitOp;

/// A shard's ring.
pub(crate) struct ShardRing {
    poller: RefCell<RingPoller>,
    ops: RefCell<Ops<RingOp>>,
    /// Files' reads and writes in flight, their user data marked with
    /// [`FILE_TAG`].
    file_ops: RefCell<Ops<FileOp>>,
    /// Children's exits watched, their user data marked with [`FILE_TAG`]
    /// and [`EXIT_TAG`].
    exit_ops: RefCell<Ops<ExitOp>>,
    /// The kernel has `IORING_OP_WAITID` (Linux 6.7).
    waitid: bool,
    /// The kernel has `IORING_OP_SEND_ZC` (Linux 6.0).
    send_zc: bool,
    /// The recovery from dropped completions (see `recover`).
    recover: recover::Recover,
    /// A zero-copy send was refused for lack of lockable memory: the ring sends with copies from
    /// then on.
    zc_denied: Cell<bool>,
    /// The least a zero-copy write must have left for its next send to go zero-copy: the
    /// runtime's `zero_copy_threshold`, set as the worker's shard is made.
    zero_copy_min: Cell<usize>,
    /// Any thread may submit to the ring and its completions are posted without its thread (it
    /// is not single-issuer with deferred task work), so a keeper can take it as its shard goes.
    reapable: bool,
    files: RefCell<Files>,
    /// The fixed files the kernel's table is still to be registered with:
    /// registered at the first source's attach, so a shard that never has a
    /// source (a ring a thread makes only to run its tasks and timers)
    /// costs the kernel no table. Zero once it is, or when there is none.
    table: Cell<u32>,
    /// Slots of sources that went whose entries did not all reach the
    /// kernel yet (it held completions it had no room for), with the leases
    /// that keep the sources' objects open until they do.
    leaving: RefCell<Vec<(u32, Option<Lease>)>>,
    /// Files' calls submitted but not in the kernel yet, by user data, with
    /// their entries, in order.
    room: RefCell<VecDeque<(u64, Sqe)>>,
    /// Files' calls in the kernel, the only operations the cap counts, and
    /// the most it may hold.
    in_kernel: Cell<usize>,
    cap: usize,
    /// Write sides' operations in the kernel: no cap holds them back (a peer
    /// that never reads keeps one for good), but most end within a moment.
    writes: Cell<usize>,
    /// The buffers multishot receives draw from; `None` when the
    /// configuration asks for none, or once the ring is shut down.
    recv: RefCell<Option<RecvGroup>>,
    /// Where the latencies of the operations go, with their submissions'
    /// stamps, while the runtime records them: one word, boxed, so a ring
    /// that records nothing carries none of it in its own layout.
    #[cfg(feature = "io-stats")]
    timing: std::cell::OnceCell<Box<Timing>>,
    /// Multishot receives and accepts submitted, the receives' arrivals,
    /// and the receives that ended as the group ran dry, for a test.
    #[cfg(test)]
    counts: Cell<(usize, usize, usize)>,
    /// Multishot polls submitted, for a test.
    #[cfg(test)]
    polls_armed: Cell<usize>,
    /// Files' calls submitted, and each one's opcode, length and offset, for
    /// a test.
    #[cfg(test)]
    file_calls: Cell<usize>,
    #[cfg(test)]
    file_entries: RefCell<Vec<FileEntry>>,
    /// Zero-copy sends' first completions and notifications, for a test.
    #[cfg(test)]
    zc_seen: Cell<(usize, usize)>,
}

impl ShardRing {
    /// A ring as `cfg` describes with a table of `files` fixed files (made
    /// at the first attach, see `table`), disabled until its first wait or
    /// submission.
    pub(crate) fn new(cfg: &RingConfig, files: u32) -> io::Result<Self> {
        let poller = RingPoller::new(
            &RingConfig {
                taskrun_flag: crate::syscore::linux::iouring::ring::taskrun::flag_works(),
                ..*cfg
            },
            0,
        )?;
        let recv = match group_entries(cfg.recv_buffers, poller.cq_entries()) {
            0 => None,
            entries => match RecvGroup::new(poller.ring(), entries, cfg.recv_buffer_size) {
                Ok(group) => Some(group),
                // A count or a size no group can have is the configuration's
                // mistake, said at once. Only nuclei's own checks answer this
                // way: the kernel's refusal of the feature is `Unsupported`.
                Err(err) if err.kind() == io::ErrorKind::InvalidInput => {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidInput,
                        format!("io_uring recv_buffers and recv_buffer_size: {err}"),
                    ));
                }
                // No memory for the group: the shard starts without it, said
                // and counted (`went_without`), apart from the ring's own
                // shortage, which the ring answers by shrinking (`with_room`).
                Err(err) if err.kind() == io::ErrorKind::OutOfMemory => {
                    super::recv::went_without(entries, cfg.recv_buffer_size, &err);
                    None
                }
                // Anything else, a kernel's refusal included, is
                // the ring's failure: a shard never runs without the group
                // its configuration asked for, unseen.
                Err(err) => return Err(err),
            },
        };
        // Room in the completion queue for every arrival the group's buffers
        // can hold.
        let arrivals = recv
            .as_ref()
            .map_or(0, |g: &RecvGroup| usize::from(g.entries()));
        let cap = (poller.cq_entries() as usize)
            .saturating_sub(arrivals)
            .max(1);
        use crate::syscore::linux::iouring::ring::sys::Opcode;
        let probed = poller.ring().register_probe().unwrap_or_default();
        let waitid = probed.contains(Opcode::Waitid);
        let send_zc = probed.contains(Opcode::SendZc);
        Ok(Self {
            poller: RefCell::new(poller),
            // Below the file tag's bit, so a source's user data never has it.
            ops: RefCell::new(Ops::new((1 << 31) - 1)),
            file_ops: RefCell::new(Ops::new(1 << 30)),
            // Short of the top 256 indexes, which with both tags would be a wake's (`WAKE_LOW`).
            exit_ops: RefCell::new(Ops::new((1 << 30) - 256)),
            waitid,
            send_zc,
            recover: recover::Recover::default(),
            zc_denied: Cell::new(false),
            zero_copy_min: Cell::new(crate::io::source::ring::SEND_ZC_MIN),
            reapable: cfg.multi_issuer,
            files: RefCell::new(Files::new(files)),
            table: Cell::new(files),
            leaving: RefCell::new(Vec::new()),
            room: RefCell::new(VecDeque::new()),
            in_kernel: Cell::new(0),
            cap,
            writes: Cell::new(0),
            recv: RefCell::new(recv),
            #[cfg(feature = "io-stats")]
            timing: std::cell::OnceCell::new(),
            #[cfg(test)]
            counts: Cell::new((0, 0, 0)),
            #[cfg(test)]
            polls_armed: Cell::new(0),
            #[cfg(test)]
            file_calls: Cell::new(0),
            #[cfg(test)]
            file_entries: RefCell::new(Vec::new()),
            #[cfg(test)]
            zc_seen: Cell::new((0, 0)),
        })
    }

    /// The ring of a shard of `cfg`: with fewer entries while the system has
    /// no memory for `cfg.entries`, down to [`MIN_RING_ENTRIES`], and the
    /// table of [`table_len`] fixed files. The one place a shard's ring is
    /// made, so [`check`] sees exactly what a shard gets.
    pub(crate) fn for_shard(cfg: RingConfig) -> io::Result<Self> {
        with_room(cfg.entries, MIN_RING_ENTRIES, |entries| {
            Self::new(&RingConfig { entries, ..cfg }, table_len())
        })
    }

    /// Makes the calling thread the ring's issuer, as a shard's first wait
    /// or submission does.
    pub(crate) fn enable(&self) -> io::Result<()> {
        self.poller.borrow_mut().enable()
    }

    /// Whether operations whose completions come within a moment are out: files' calls (in the
    /// kernel or waiting for room) and a write side's. A read side's waits on a peer for as long
    /// as the peer is idle, and a child's exit watch on the child, so neither is a sign of one.
    pub(crate) fn expects_completions(&self) -> bool {
        self.in_kernel.get() > 0 || self.writes.get() > 0 || !self.room.borrow().is_empty()
    }

    /// Takes in the chunk `item` of the group `back` at once, if the ring's group is that, or
    /// keeps it for the next take-in while the operations are borrowed; `false` when the group
    /// is another or borrowed (the chunk then goes the long way).
    #[inline(always)]
    pub(crate) fn take_in_now(
        &self,
        back: &Arc<super::recv::Returns>,
        item: super::recv::Back,
    ) -> bool {
        let Ok(mut recv) = self.recv.try_borrow_mut() else {
            return false;
        };
        let Some(group) = recv.as_mut() else {
            return false;
        };
        let Ok(mut ops) = self.ops.try_borrow_mut() else {
            return group.keep_local(back, item);
        };
        match group.take_in_now(back, item) {
            Some(Some((user_data, charged))) => {
                release_charge(&mut ops, user_data, charged);
                true
            }
            Some(None) => true,
            None => false,
        }
    }

    /// The group id multishot receives draw from, if the ring has a group.
    pub(crate) fn recv_group(&self) -> Option<u16> {
        self.recv.borrow().as_ref().map(RecvGroup::bgid)
    }

    /// Whether a stream send may go out zero-copy on this ring: one a keeper can take, so a
    /// buffer the kernel still holds as the shard goes comes back once it let go of it.
    pub(crate) fn zc_stream(&self) -> bool {
        self.send_zc && self.reapable && !self.zc_denied.get()
    }

    /// The least a zero-copy write must have left for its next send to go zero-copy here.
    pub(crate) fn zero_copy_min(&self) -> usize {
        self.zero_copy_min.get()
    }

    /// Sets [`zero_copy_min`](Self::zero_copy_min), once, as the worker's shard is made.
    pub(crate) fn set_zero_copy_min(&self, bytes: usize) {
        self.zero_copy_min.set(bytes);
    }

    /// Whether another thread can submit to the ring and reap its completions: a keeper can take
    /// it as its shard goes (see `keeper`).
    pub(crate) fn reapable(&self) -> bool {
        self.reapable
    }

    /// Whether the ring's group is filled in pieces, which a datagram cannot share: the kernel
    /// would cut it to the room left in a buffer.
    pub(crate) fn recv_group_incremental(&self) -> bool {
        self.recv
            .borrow()
            .as_ref()
            .is_some_and(RecvGroup::incremental)
    }

    /// The size of the group's buffers, if the ring has a group.
    pub(crate) fn recv_buf_size(&self) -> Option<u32> {
        self.recv.borrow().as_ref().map(RecvGroup::buf_size)
    }

    /// Whether the ring receives UDP datagrams ahead, with a multishot receive into its group:
    /// one of whole buffers that each hold any datagram (see `DATAGRAM_ROOM`). Elsewhere a UDP
    /// datagram is its task's to receive into its own buffer, as a Unix one always is.
    pub(crate) fn datagrams_ahead(&self) -> bool {
        !self.recv_group_incremental()
            && self
                .recv_buf_size()
                .is_some_and(|s| s >= crate::io::source::ring::DATAGRAM_ROOM)
    }

    /// The memory the ring's group takes (0 without one), charged to the shard's ledger.
    pub(crate) fn recv_reservation(&self) -> usize {
        self.recv
            .borrow()
            .as_ref()
            .map_or(0, RecvGroup::reservation)
    }

    pub(crate) fn notifier(&self) -> Notifier {
        self.poller.borrow().notifier()
    }

    /// Submits `op` with its entry `sqe` (whose pointers point into `op`'s
    /// memory, which lives on the heap and does not move with it), on the
    /// slot `op.file`, and returns its user data. It goes into the kernel at
    /// once, never counted against the cap; its slot is named from now on.
    #[inline]
    pub(crate) fn submit(&self, op: RingOp, sqe: Sqe) -> u64 {
        let sqe = match self.files.borrow_mut().name(op.file) {
            Some(own) => sqe.fd(own),
            None => sqe.flags(IOSQE_FIXED_FILE),
        };
        op.source.op_out();
        #[cfg(test)]
        if matches!(
            op.mem,
            RingMem::RecvMulti { .. } | RingMem::RecvFromMulti { .. } | RingMem::AcceptMulti { .. }
        ) {
            let (armed, arrived, dry) = self.counts.get();
            self.counts.set((armed + 1, arrived, dry));
        }
        #[cfg(test)]
        if matches!(op.mem, RingMem::ReadyMulti) {
            self.polls_armed.set(self.polls_armed.get() + 1);
        }
        let write = counts_as_write(&op);
        #[cfg(test)]
        let full = REFUSE_INSERT_FOR_TEST.with(|n| n.replace(n.get().saturating_sub(1))) > 0;
        #[cfg(not(test))]
        let full = false;
        let inserted = if full {
            Err(op)
        } else {
            self.ops.borrow_mut().insert(op)
        };
        let user_data = match inserted {
            Ok(user_data) => user_data,
            // The table holds (1 << 31) - 1 operations (two per source at most) and is full: the
            // resource ran out, which is ENOMEM. Not ENOBUFS, which a multishot receive reads as
            // its group running dry (it would arm again at once and meet the full table) and a
            // zero-copy send as the kernel refusing zero copy. A zero-copy send that ends here
            // ends for its caller with ENOMEM, as a copied send of the same full table would.
            Err(op) => {
                self.end(None, op, -libc::ENOMEM);
                return IGNORED;
            }
        };
        #[cfg(feature = "io-stats")]
        if self.timing.get().is_some() {
            self.stamp(user_data);
        }
        if write {
            self.writes.set(self.writes.get() + 1);
        }
        self.enter(user_data, sqe, false);
        user_data
    }

    /// Puts the entry of the operation of `user_data` into the kernel, or
    /// behind the ones waiting for room. One that waits on a peer or a child
    /// (`counted` false) goes in at once and is not counted.
    pub(super) fn enter(&self, user_data: u64, sqe: Sqe, counted: bool) {
        let sqe = sqe.user_data(user_data);
        if !counted {
            self.poller.borrow_mut().push(sqe);
        } else if self.room.borrow().is_empty() && self.in_kernel.get() < self.cap {
            self.in_kernel.set(self.in_kernel.get() + 1);
            self.poller.borrow_mut().push(sqe);
        } else {
            self.room.borrow_mut().push_back((user_data, sqe));
        }
    }

    /// Submits a file's read or write, on its own file (no slot of the
    /// table): the operation's lease keeps the descriptor open until it
    /// ends. With the kernel full it waits for room.
    pub(crate) fn submit_file(&self, mut op: FileOp) {
        #[cfg(test)]
        self.file_calls.set(self.file_calls.get() + 1);
        let sqe = op.entry();
        #[cfg(test)]
        self.file_entries.borrow_mut().push((
            sqe.raw().opcode,
            sqe.raw().len,
            sqe.raw().off,
            sqe.raw().op_flags,
        ));
        let inserted = self.file_ops.borrow_mut().insert(op);
        let user_data = match inserted {
            Ok(user_data) => user_data | FILE_TAG,
            // Past 2^30 file calls at once: the call goes to the pool.
            Err(op) => {
                drop(op.complete(-libc::ECANCELED));
                return;
            }
        };
        #[cfg(feature = "io-stats")]
        if self.timing.get().is_some() {
            self.stamp(user_data);
        }
        self.enter(user_data, sqe, true);
    }

    /// A file's call completed: the rest of a write goes on, else its end
    /// is left for its task.
    fn complete_file(&self, user_data: u64, res: i32) {
        let Some(op) = self.file_ops.borrow_mut().remove(user_data & !FILE_TAG) else {
            return self.swept_file_ended(user_data);
        };
        self.in_kernel.set(self.in_kernel.get().saturating_sub(1));
        self.admit_waiting();
        #[cfg(feature = "io-stats")]
        if self.timing.get().is_some() {
            self.timed(user_data, Some(IoKind::File), res);
        }
        if let Some(rest) = op.complete(res) {
            self.submit_file(rest);
        }
    }

    /// Moves waiting operations into the kernel while it has room.
    fn admit_waiting(&self) {
        while self.in_kernel.get() < self.cap {
            let Some((_, sqe)) = self.room.borrow_mut().pop_front() else {
                return;
            };
            self.in_kernel.set(self.in_kernel.get() + 1);
            self.poller.borrow_mut().push(sqe);
        }
    }

    /// Cancels the source's operation of `user_data`, which is in the kernel:
    /// its completion says what it did. Returns whether a cancel went to the
    /// kernel's queue.
    pub(crate) fn cancel(&self, user_data: u64) -> bool {
        if self.ops.borrow_mut().get_mut(user_data).is_some() {
            self.poller
                .borrow_mut()
                .push(Sqe::async_cancel(user_data).user_data(IGNORED));
            return true;
        }
        false
    }

    /// A slot holding the file of `fd`, which the caller keeps open for
    /// the call: one of the ring's table, or past a full table, one naming
    /// `fd` itself, which its source keeps open for as long as an entry
    /// waits to name it (no copy is made: a process with no descriptor to
    /// spare registers all the same).
    pub(crate) fn attach(&self, fd: RawFd) -> io::Result<u32> {
        if self.table.get() != 0 {
            self.table_or_none();
        }
        let fixed = self.files.borrow_mut().take();
        match fixed {
            Some(slot) => match self.poller.borrow().update_file(slot, fd) {
                Ok(()) => Ok(slot),
                Err(err) => {
                    self.files.borrow_mut().release(slot);
                    Err(err)
                }
            },
            None => Ok(self.files.borrow_mut().take_own(fd)),
        }
    }

    /// Registers the kernel's table of fixed files, as the first attach
    /// does.
    ///
    /// # Errors
    ///
    /// The kernel's refusal of the table.
    pub(crate) fn make_table(&self) -> io::Result<()> {
        let slots = self.table.get();
        if slots != 0 {
            self.poller.borrow().ring().register_files_sparse(slots)?;
            self.table.set(0);
        }
        Ok(())
    }

    /// [`make_table`](Self::make_table) for the first attach: a table the
    /// kernel refuses now (no memory, or a descriptor limit lowered since the
    /// ring was made) leaves the ring without one, said once, and every
    /// source's entries name its own descriptor, as past a full table.
    #[cold]
    #[inline(never)]
    fn table_or_none(&self) {
        if let Err(err) = self.make_table() {
            self.table.set(0);
            self.files.borrow_mut().without_table();
            static SAID: core::sync::atomic::AtomicBool =
                core::sync::atomic::AtomicBool::new(false);
            if !SAID.swap(true, core::sync::atomic::Ordering::Relaxed) {
                log::warn!(
                    "io_uring's table of fixed files could not be set up ({err}); sockets use their own descriptors instead."
                );
            }
        }
    }

    /// The source of `slot` went (closed, retired, evicted or lingering
    /// on): its sides' operations `out` are cancelled, and with every entry
    /// naming the slot handed to the kernel now, the file is taken out of
    /// the slot now (a fixed file out of the table; `lease`, which keeps the
    /// source's object open, let go), so the kernel closes it as soon as its
    /// requests let go of it, which the cancels see to before this returns
    /// (proofs/tla/ring/RingClose.tla). The slot stays reserved until no
    /// entry names it. Where the kernel took not every entry (it held
    /// completions it had no room for), this is finished after the shard's
    /// next wait.
    pub(crate) fn detach(&self, slot: u32, out: [Option<u64>; 2], lease: Option<Lease>) {
        let mut cancelled = false;
        for user_data in out.into_iter().flatten() {
            cancelled |= self.cancel(user_data);
        }
        // Entries naming the slot, or cancels of them, may still wait in
        // the submission queue: they go to the kernel now, and the work the
        // kernel holds for this thread runs, so the cancelled requests let
        // go of their files before this returns.
        let named = cancelled || self.files.borrow().is_named(slot);
        if named && !self.poller.borrow_mut().submit_now().unwrap_or(false) {
            // Still held, so the slot is not handed out meanwhile.
            self.leaving.borrow_mut().push((slot, lease));
            return;
        }
        self.take_out(slot, lease);
    }

    /// The source of `slot` went and no entry naming the slot waits to reach
    /// the kernel: a fixed file leaves the table, the slot is released (free
    /// once its entries completed), and the source's object may close. A
    /// take-out from the table the kernel refuses (it has no memory for the
    /// update) keeps the slot held with the file in it, since another source
    /// given the slot would have its own file taken out by the retry; the
    /// retry comes after the ring's next wait, which a timer makes come
    /// though nothing else is out (proofs/tla/ring/RingClose.tla,
    /// `MC_update_ignored`, `MC_update_frees`, `MC_update_no_wake`).
    fn take_out(&self, slot: u32, lease: Option<Lease>) {
        if self.files.borrow().is_fixed(slot)
            && let Err(err) = self.take_file_out(slot)
        {
            self.retry_take_out(slot, &err);
        } else {
            self.files.borrow_mut().release(slot);
        }
        drop(lease);
    }

    /// Takes the file of fixed slot `slot` out of the kernel's table.
    fn take_file_out(&self, slot: u32) -> io::Result<()> {
        #[cfg(test)]
        if REFUSE_TAKE_OUT_FOR_TEST.with(|n| n.replace(n.get().saturating_sub(1))) > 0 {
            return Err(io::Error::from_raw_os_error(libc::ENOMEM));
        }
        self.poller.borrow().update_file(slot, -1)
    }

    /// The kernel refused to take the file of `slot` out of its table: the
    /// slot waits on `leaving` for the next wait's end to try again, and a
    /// timer of the ring's own makes that wait come.
    #[cold]
    #[inline(never)]
    fn retry_take_out(&self, slot: u32, err: &io::Error) {
        static SAID: core::sync::atomic::AtomicBool = core::sync::atomic::AtomicBool::new(false);
        if !SAID.swap(true, core::sync::atomic::Ordering::Relaxed) {
            log::warn!(
                "io_uring could not take a closed file out of its table ({err}); trying again shortly."
            );
        }
        self.leaving.borrow_mut().push((slot, None));
        self.poller
            .borrow_mut()
            .push(Sqe::timeout(&RETRY_TAKE_OUT).user_data(IGNORED));
    }

    /// Slots of sources that went, waiting for their entries to reach the
    /// kernel: once every queued entry did, they are taken out.
    fn finish_leaving(&self) {
        if self.leaving.borrow().is_empty()
            || !self.poller.borrow_mut().submit_now().unwrap_or(false)
        {
            return;
        }
        let leaving = core::mem::take(&mut *self.leaving.borrow_mut());
        for (slot, lease) in leaving {
            self.take_out(slot, lease);
        }
    }

    /// Submits what is queued and waits up to `timeout` for completions,
    /// which are handed to their sources. Returns how many it handled and
    /// whether a wake came.
    ///
    /// `shard` is the shard whose thread calls, when it is at hand: the
    /// sources then do not look it up for each completion.
    pub(crate) fn wait(
        &self,
        shard: Option<&Shard>,
        timeout: Option<Duration>,
    ) -> io::Result<(usize, bool)> {
        self.take_back();
        let mut batch = CqeBatch::new();
        let mut woken = self.poller.borrow_mut().wait(timeout, &mut batch);
        let mut handled = 0;
        // Before every view of a source an arrival takes (see `arrived`), so it drops after them:
        // one for the whole batch, not one per arrival.
        let _unwound = crate::io::side::Unwound::enter();
        loop {
            handled += batch.len();
            for at in 0..batch.len() {
                self.complete(shard, batch.get(at));
            }
            // A full batch may have left more in the queue.
            if !batch.is_full() {
                break;
            }
            batch.clear();
            let (woke_more, _) = self.poller.borrow_mut().reap(&mut batch);
            woken = woken.map(|w| w || woke_more);
        }
        self.finish_leaving();
        Ok((handled, woken?))
    }

    /// One completion the wait reaped. One call, in the wait's loop, inlined there in every
    /// build: the cost the inliner weighs it at sits near its limit, so the code around it decides.
    #[inline(always)]
    fn complete(&self, shard: Option<&Shard>, cqe: IoUringCqe) {
        #[cfg(test)]
        if self.drops_for_test(&cqe) {
            return;
        }
        if cqe.user_data & FILE_TAG != 0 {
            if cqe.user_data & EXIT_TAG != 0 {
                self.complete_exit(shard, cqe.user_data, cqe.res);
            } else {
                self.complete_file(cqe.user_data, cqe.res);
            }
            return;
        }
        // The common end, one lookup: a plain operation's only completion.
        let plain =
            if cqe.flags & (IORING_CQE_F_MORE | IORING_CQE_F_NOTIF | IORING_CQE_F_BUFFER) == 0 {
                self.ops.borrow_mut().remove_if(cqe.user_data, ends_plain)
            } else {
                None
            };
        if let Some(op) = plain {
            if counts_as_write(&op) {
                self.writes.set(self.writes.get().saturating_sub(1));
            }
            #[cfg(feature = "io-stats")]
            if self.timing.get().is_some() {
                self.timed(cqe.user_data, io_kind(&op.mem), cqe.res);
            }
            self.end(shard, op, cqe.res);
            return;
        }
        if let Some(bid) = cqe_buffer_id(cqe.flags) {
            self.arrived(
                shard,
                cqe.user_data,
                bid,
                cqe.res,
                cqe.flags & IORING_CQE_F_BUF_MORE != 0,
                cqe.flags & IORING_CQE_F_MORE != 0,
            );
        } else if self.zero_copy(shard, &cqe) {
            return;
        } else if cqe.res >= 0 {
            self.accepted(shard, cqe.user_data, cqe.res);
        }
        // More of a multishot receive's arrivals come: it stays in flight.
        if cqe.flags & IORING_CQE_F_MORE != 0 {
            return;
        }
        let Some(op) = self.ops.borrow_mut().remove(cqe.user_data) else {
            return self.recover.late(cqe.user_data);
        };
        if let RingMem::Zombie | RingMem::RecvZombie { .. } = op.mem {
            return self.bury(cqe.user_data, op);
        }
        if counts_as_write(&op) {
            self.writes.set(self.writes.get().saturating_sub(1));
        }
        #[cfg(feature = "io-stats")]
        if self.timing.get().is_some() {
            self.timed(cqe.user_data, io_kind(&op.mem), cqe.res);
        }
        self.end(shard, op, cqe.res);
    }

    /// The calling thread runs this ring from now on, taken over from a runner the monitor took
    /// its shard from, and calls this before its first drive. A watched worker's ring is not
    /// single-issuer, so the kernel runs the completion of each request as work queued to the
    /// thread that submitted it, at its next return from the kernel: a thread in a wait no signal
    /// ends (a killable wait, a read of a slow disk) runs none of it meanwhile. A poll the old
    /// runner armed (a datagram's multishot poll, a one-shot poll for readiness or room), a
    /// multishot receive (a stream's or a datagram's), and the ring's wake read, would then bring
    /// nothing for as long. Each is cancelled from here and its side told it ended, so the side
    /// arms another, submitted by this thread; the old one stays in the slab as a zombie and is
    /// buried at its own end (`bury`), whenever its old thread runs it. A poll's zombie points at
    /// no memory and hands its side nothing. A receive's zombie still hands its side what its old
    /// thread takes before that thread sees the cancel: the kernel takes each arrival and posts
    /// it under the ring's lock, so the arrivals of both receives come in the stream's order
    /// (`RingMem::RecvZombie`). An entry still waiting to
    /// reach the kernel is this thread's, which submits it, and stays as it is; what the kernel
    /// posted already goes to its side first, so an edge in the queue is not lost with its zombie
    /// (tests/cadence.rs, each runner blocking at once). Everything else is left as it is, its
    /// completion waiting for that thread: a one-shot operation that takes or gives bytes, a
    /// file's call and a child's exit watch, whose work cannot be done again, and an accept,
    /// whose poll is an exclusive waiter of its listener, the first of which takes each
    /// connection's wake (see `RingMem::adopted`; proofs/tla/ring/RingHandoff.tla).
    #[cold]
    #[inline(never)]
    pub(crate) fn adopt(&self, shard: Option<&Shard>) {
        // With a kernel submission thread every request is that thread's.
        if self.poller.borrow().sqpoll() {
            return;
        }
        // What waits to reach the kernel this thread submits at its next enter: its own.
        let mut pending = Vec::new();
        self.poller
            .borrow()
            .pending(|user_data| pending.push(user_data));
        pending.sort_unstable();
        let old: Vec<u64> = self
            .ops
            .borrow()
            .keys()
            .filter(|user_data| pending.binary_search(user_data).is_err())
            .collect();
        self.poller.borrow_mut().adopt(IGNORED, &pending);
        // The completions the kernel posted already go to their sides first: a poll's edge in
        // the queue is delivered, not lost with its zombie. A failure is the first drive's to
        // report, as it meets it again.
        let _ = self.wait(shard, Some(Duration::ZERO));
        let mut ended = Vec::new();
        {
            let mut ops = self.ops.borrow_mut();
            for user_data in old {
                let Some(op) = ops.get_mut(user_data) else {
                    continue;
                };
                let mem = core::mem::replace(&mut op.mem, RingMem::Ready);
                let (zombie, outcome) = match mem.adopted(op.source.kind()) {
                    Ok(adopted) => adopted,
                    Err(mem) => {
                        op.mem = mem;
                        continue;
                    }
                };
                op.mem = zombie;
                self.poller
                    .borrow_mut()
                    .push(Sqe::async_cancel(user_data).user_data(IGNORED));
                if counts_as_write(op) {
                    self.writes.set(self.writes.get().saturating_sub(1));
                }
                // The zombie counts among the source's operations out until its own end.
                op.source.op_out();
                let read = core::mem::replace(&mut op.read, true);
                ended.push((Arc::clone(&op.source), op.epoch, read, outcome));
            }
        }
        for (source, epoch, read, outcome) in ended {
            source.op_done(
                Done {
                    outcome,
                    epoch,
                    read,
                },
                shard,
            );
        }
    }

    /// A zombie's end (see [`adopt`](Self::adopt)): its slot is named once less and its source
    /// counts it no more, its side told nothing.
    #[cold]
    #[inline(never)]
    fn bury(&self, user_data: u64, op: RingOp) {
        #[cfg(feature = "io-stats")]
        if self.timing.get().is_some() {
            self.timed(user_data, None, 0);
        }
        #[cfg(not(feature = "io-stats"))]
        let _ = user_data;
        self.files.borrow_mut().done(op.file);
        op.source.op_gone();
    }

    /// A completion of a zero-copy send, handled here; false for any other. The first completion,
    /// with the bytes the kernel took, ends the send for its side at once, which goes on to its
    /// next chunk or write; the send then stays in the ring only to hold the buffer the kernel may
    /// still read, until its notification (`IORING_CQE_F_NOTIF`), and no longer names its slot or
    /// counts among the write sides' operations, since it waits on the network. A first
    /// completion without `IORING_CQE_F_MORE` (a send refused before it ran) has no notification
    /// after it. The kernel posts the notification after the first completion; one that comes
    /// before it is kept for it. `proofs/tla/lent/LentZc.tla` checks the rule.
    fn zero_copy(&self, shard: Option<&Shard>, cqe: &IoUringCqe) -> bool {
        let notif = cqe.flags & IORING_CQE_F_NOTIF != 0;
        let mut ops = self.ops.borrow_mut();
        let Some(op) = ops.get_mut(cqe.user_data) else {
            return false;
        };
        let waits = match &mut op.mem {
            RingMem::SendZc { notified, .. } if notif => {
                *notified = true;
                return true;
            }
            RingMem::SendZc { notified, .. } => cqe.flags & IORING_CQE_F_MORE != 0 && !*notified,
            RingMem::ZcHeld(_) if notif => {
                // The kernel let go of what the send took: so does the send, and the last hold
                // hands the buffer back.
                let held = ops.remove(cqe.user_data);
                drop(ops);
                #[cfg(test)]
                self.zc_seen
                    .set((self.zc_seen.get().0, self.zc_seen.get().1 + 1));
                drop(held);
                return true;
            }
            _ => return false,
        };
        let (source, file, epoch) = (Arc::clone(&op.source), op.file, op.epoch);
        let ended = if waits {
            op.mem.hold_for_notification();
            None
        } else {
            ops.remove(cqe.user_data)
        };
        drop(ops);
        #[cfg(test)]
        if waits {
            self.zc_seen
                .set((self.zc_seen.get().0 + 1, self.zc_seen.get().1));
        }
        self.writes.set(self.writes.get().saturating_sub(1));
        let mut res = cqe.res;
        if zc_refused(res) {
            // No memory to pin: nothing was sent, and the chunk goes again with a copy.
            self.zc_denied.set(true);
            res = -libc::ECANCELED;
        }
        #[cfg(feature = "io-stats")]
        if self.timing.get().is_some() {
            self.timed(cqe.user_data, Some(IoKind::Write), res);
        }
        let outcome = crate::io::source::ring::zc_outcome(res);
        self.hand_over(shard, &source, file, epoch, false, outcome);
        drop(ended);
        true
    }

    #[cfg(feature = "io-stats")]
    /// Starts timing the operations submitted from now on into `times`.
    pub(super) fn time_with(&self, times: Arc<IoTimes>) {
        let _ = self.timing.set(Box::new(Timing {
            times,
            stamps: RefCell::new(std::collections::HashMap::new()),
        }));
    }

    #[cfg(feature = "io-stats")]
    /// The operation of `user_data` was submitted now, while latencies are
    /// recorded.
    #[cold]
    #[inline(never)]
    fn stamp(&self, user_data: u64) {
        if let Some(timing) = self.timing.get() {
            timing.stamps.borrow_mut().insert(user_data, Instant::now());
        }
    }

    #[cfg(feature = "io-stats")]
    /// The operation of `user_data` ended with `res`, while latencies are
    /// recorded: its stamp goes, and its latency is counted under `kind` (an
    /// operation of no timed kind, a multishot one, is only let go of).
    #[cold]
    #[inline(never)]
    fn timed(&self, user_data: u64, kind: Option<IoKind>, res: i32) {
        let stamp = self
            .timing
            .get()
            .and_then(|timing| timing.stamps.borrow_mut().remove(&user_data));
        if let (Some(at), Some(kind)) = (stamp, kind) {
            self.time(kind, res, at);
        }
    }

    #[cfg(feature = "io-stats")]
    /// An operation of `kind` stamped `issued` ended with `res`: its latency
    /// is counted, unless it was cancelled or told to try again.
    fn time(&self, kind: IoKind, res: i32, issued: Instant) {
        if let Some(times) = self.timing.get().map(|timing| &timing.times)
            && res != -libc::ECANCELED
            && res != -libc::EAGAIN
            && !(matches!(kind, IoKind::Connect) && res == -libc::EALREADY)
        {
            times.add(kind, issued.elapsed());
        }
    }

    /// `op` ended with `res`: its slot is named once less, and its source
    /// gets the completion.
    fn end(&self, shard: Option<&Shard>, op: RingOp, res: i32) {
        let RingOp {
            source,
            file,
            epoch,
            read,
            mem,
        } = op;
        #[cfg(test)]
        if res == -libc::ENOBUFS && matches!(mem, RingMem::RecvMulti { .. }) {
            let (armed, arrived, dry) = self.counts.get();
            self.counts.set((armed, arrived, dry + 1));
        }
        let outcome = mem.complete(res, source.kind());
        self.hand_over(shard, &source, file, epoch, read, outcome);
    }

    /// An operation of `source` on slot `file`, issued in `epoch` for its read side (`read`) or
    /// its write side, ended for that side with `outcome`: the slot is named once less, and the
    /// side gets the completion.
    #[inline]
    fn hand_over(
        &self,
        shard: Option<&Shard>,
        source: &Arc<crate::io::source::Source>,
        file: u32,
        epoch: u32,
        read: bool,
        outcome: crate::io::source::ops::Outcome,
    ) {
        // The slot's file left the table as its source went: a slot that
        // comes free only goes back to its list.
        self.files.borrow_mut().done(file);
        source.op_done(
            Done {
                outcome,
                epoch,
                read,
            },
            shard,
        );
    }
}

/// A ring's latencies while its runtime records them: where they go, and
/// when each operation in flight was submitted, by its user data.
#[cfg(feature = "io-stats")]
struct Timing {
    times: Arc<IoTimes>,
    stamps: RefCell<std::collections::HashMap<u64, Instant>>,
}

#[cfg(feature = "io-stats")]
/// The kind of latency an operation's end counts as, if it is timed: not a
/// readiness probe, nor a multishot operation (many arrivals for one).
fn io_kind(mem: &RingMem) -> Option<IoKind> {
    match mem {
        RingMem::Recv(_)
        | RingMem::RecvLent(_)
        | RingMem::RecvFrom(..)
        | RingMem::RecvLentFrom(..) => Some(IoKind::Read),
        RingMem::Send(_)
        | RingMem::SendLent(_)
        | RingMem::SendTo(..)
        | RingMem::SendLentTo(..)
        | RingMem::SendPinned(_) => Some(IoKind::Write),
        RingMem::Accept(_) => Some(IoKind::Accept),
        RingMem::Connect(..) => Some(IoKind::Connect),
        _ => None,
    }
}

impl ShardRing {
    /// Operations in flight: the sources', the files' and the children's.
    pub(crate) fn in_flight(&self) -> usize {
        self.ops.borrow().len() + self.file_ops.borrow().len() + self.exit_ops.borrow().len()
    }

    /// Times the kernel began holding completions the ring's queue had no
    /// room for.
    pub(crate) fn overflows(&self) -> u64 {
        self.poller.borrow().overflows()
    }
}

/// Whether the kernel refused a zero-copy send: no memory it may pin, or a socket that has none.
fn zc_refused(res: i32) -> bool {
    [
        -libc::ENOBUFS,
        -libc::ENOMEM,
        -libc::EPERM,
        -libc::EOPNOTSUPP,
    ]
    .contains(&res)
}

/// How many buffers the receive group of a ring with `cq_entries`
/// completion slots gets when the configuration asks for `configured`: no
/// more than the largest power of two within half the queue, so arrivals
/// never fill it; none when the queue cannot hold one. A count that is not a
/// power of two is passed on unchanged, for the group to refuse as the
/// configuration's mistake.
fn group_entries(configured: u16, cq_entries: u32) -> u16 {
    if configured == 0 || !configured.is_power_of_two() {
        return configured;
    }
    let half = cq_entries / 2;
    if half == 0 {
        return 0;
    }
    let fit = 1u32 << half.ilog2();
    configured.min(u16::try_from(fit.min(1 << 15)).unwrap_or(1 << 15))
}

/// Counts `charged` bytes of a chunk given back off the multishot receive of `user_data`, if it
/// is still in flight.
#[inline(always)]
fn release_charge(ops: &mut Ops<RingOp>, user_data: u64, charged: usize) {
    if let Some(
        RingMem::RecvMulti { held, .. }
        | RingMem::RecvFromMulti { held, .. }
        | RingMem::RecvZombie { held, .. },
    ) = ops.get_mut(user_data).map(|op| &mut op.mem)
    {
        *held = held.saturating_sub(charged);
    }
}

/// Whether `op` is one of the write sides' operations counted in `writes`: a
/// write side's, until a zero-copy send only waits for its notification. The
/// one place both the submission and the end of an operation decide it.
/// Whether `op` ends at its one completion, with no arrival, notification or follow-up to wait for.
fn ends_plain(op: &RingOp) -> bool {
    !matches!(
        op.mem,
        RingMem::SendZc { .. }
            | RingMem::ZcHeld(_)
            | RingMem::RecvMulti { .. }
            | RingMem::RecvFromMulti { .. }
            | RingMem::AcceptMulti { .. }
            | RingMem::ReadyMulti
            | RingMem::Zombie
            | RingMem::RecvZombie { .. }
    )
}

fn counts_as_write(op: &RingOp) -> bool {
    !op.read && !matches!(op.mem, RingMem::ZcHeld(_))
}

/// A file's call as a test sees it: its opcode, length, offset and flags.
#[cfg(test)]
pub(crate) type FileEntry = (u8, u32, u64, u32);

/// The bit of the user data of a file's call (in the slot index, which the
/// sources' operations never reach).
const FILE_TAG: u64 = 1 << 31;
/// With [`FILE_TAG`], the bit of the user data of a child's exit watch (in
/// the slot index, which a file's call never reaches).
const EXIT_TAG: u64 = 1 << 30;

/// The connections one multishot accept hands over before the ring stops
/// it: its source asks for another once its tasks took them.
const ACCEPT_CAP: usize = 256;

/// The user data of the ring's own cancels, whose completions say nothing
/// to anyone: no operation of the slab has it (its index is the one the
/// slab never hands out), and it is not the wake's.
const IGNORED: u64 = u32::MAX as u64;

/// The user data of the poller's request for a recovery from dropped completions, and of the
/// timeout that runs its sweep: [`IGNORED`]'s index with a generation of their own, which no
/// slab hands out (see `recover`).
pub(crate) const RECOVER: u64 = IGNORED | (1 << 32);
const SWEEP: u64 = IGNORED | (2 << 32);

impl Drop for ShardRing {
    fn drop(&mut self) {
        self.recover.keep_leftovers();
    }
}

/// How long a ring waits before it tries again to take a file out of its
/// table, once the kernel refused: static, as the kernel reads it after the
/// entry is queued.
static RETRY_TAKE_OUT: KernelTimespec = KernelTimespec {
    tv_sec: 0,
    tv_nsec: 10_000_000,
};

#[cfg(test)]
thread_local! {
    /// How many of the next take-outs from a table the kernel refuses on
    /// this thread, for a test of a kernel with no memory for the update.
    pub(crate) static REFUSE_TAKE_OUT_FOR_TEST: Cell<u32> = const { Cell::new(0) };
}

#[cfg(test)]
thread_local! {
    /// How many of the next inserts into the operation table `submit` makes on this
    /// thread act as a full table, for a test.
    pub(crate) static REFUSE_INSERT_FOR_TEST: Cell<u32> = const { Cell::new(0) };
}

#[cfg(test)]
impl ShardRing {
    /// Slots held or still named, and of them the ones past the table (which
    /// name their sources' own descriptors).
    pub(crate) fn files_in_use(&self) -> (usize, usize) {
        let files = self.files.borrow();
        (files.in_use(), files.own_in_use())
    }

    /// The fixed files the kernel's table is still to be registered with.
    pub(crate) fn table_to_make(&self) -> u32 {
        self.table.get()
    }

    /// Multishot receives submitted so far, the arrivals they handed out,
    /// and those that ended as the group ran dry.
    pub(crate) fn multishot_counts(&self) -> (usize, usize, usize) {
        self.counts.get()
    }

    /// Multishot polls submitted so far.
    pub(crate) fn polls_armed(&self) -> usize {
        self.polls_armed.get()
    }

    /// Bytes of the group's buffers handed out and not given back, whether
    /// or not the receive they went to is still in flight.
    pub(crate) fn held(&self) -> usize {
        self.take_back();
        self.recv.borrow().as_ref().map_or(0, RecvGroup::out)
    }

    /// Chunks of the group handed out and not given back yet, and whether the group is filled in
    /// pieces.
    pub(crate) fn live_chunks(&self) -> (u32, bool) {
        self.take_back();
        self.recv
            .borrow()
            .as_ref()
            .map_or((0, false), |g| (g.live(), g.incremental()))
    }

    /// Zero-copy sends' first completions and notifications seen so far.
    pub(crate) fn zc_seen(&self) -> (usize, usize) {
        self.zc_seen.get()
    }

    /// Files' calls submitted so far.
    pub(crate) fn file_calls(&self) -> usize {
        self.file_calls.get()
    }

    /// The opcode, length, offset and flags of each file's call submitted
    /// since the last look.
    pub(crate) fn take_file_entries(&self) -> Vec<FileEntry> {
        core::mem::take(&mut *self.file_entries.borrow_mut())
    }

    /// Multishot receives and accepts in flight.
    pub(crate) fn receives_in_flight(&self) -> usize {
        self.ops
            .borrow()
            .iter()
            .filter(|op| {
                matches!(
                    op.mem,
                    RingMem::RecvMulti { .. }
                        | RingMem::RecvFromMulti { .. }
                        | RingMem::AcceptMulti { .. }
                )
            })
            .count()
    }

    /// The buffers the receive group has.
    pub(crate) fn group_buffers(&self) -> Option<u16> {
        self.recv.borrow().as_ref().map(RecvGroup::entries)
    }

    /// The bytes one receive may hold before the ring stops it.
    pub(crate) fn recv_cap(&self) -> Option<usize> {
        self.recv.borrow().as_ref().map(RecvGroup::cap)
    }
}

#[cfg(test)]
thread_local! {
    /// The table a ring made on this thread gets, for a test.
    pub(crate) static TABLE_FOR_TEST: Cell<Option<u32>> = const { Cell::new(None) };
}

/// The fixed-file table of a ring: the process's descriptor limit (no more
/// sources can be open while it holds, and the kernel refuses a larger table
/// with `EMFILE`), at most [`MAX_TABLE`], whose slots cost kernel memory per
/// ring. Sources past it run on their own descriptors; every source does
/// where the kernel lets go of a file out of the table only late (see
/// `files::release_at_once`).
pub(crate) fn table_len() -> u32 {
    #[cfg(test)]
    if let Some(len) = TABLE_FOR_TEST.with(Cell::get) {
        return len;
    }
    if !crate::syscore::linux::iouring::files::release_at_once() {
        return 0;
    }
    // SAFETY: plain data; the call fills it.
    let mut limit: libc::rlimit = unsafe { core::mem::zeroed() };
    // SAFETY: a plain syscall writing `limit`.
    let soft = if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) } == 0 {
        limit.rlim_cur
    } else {
        1024
    };
    u32::try_from(soft).unwrap_or(MAX_TABLE).min(MAX_TABLE)
}

/// The most fixed files a ring's table holds.
const MAX_TABLE: u32 = 1 << 14;

/// The fewest submission entries a ring short of memory is made with.
const MIN_RING_ENTRIES: u32 = crate::config::MIN_QUEUE_LEN;

/// The fixed files, and buffers of a receive group, [`check`] makes: the
/// features are the kernel's to give, whatever the sizes.
const CHECK_TABLE: u32 = 64;
const CHECK_BUFFERS: u16 = 2;

/// Whether the kernel gives a shard of `cfg` everything its ring needs: the
/// ring with its setup flags and disabled start, a table of fixed files, a
/// group of buffers for multishot receives, and the enabling by the thread
/// that runs it. It makes the ring as a shard does ([`ShardRing::new`]) but
/// at the smallest sizes that prove each feature, then drops it (nothing is
/// in flight in a ring that never waited), so the backend choice sees every
/// feature a shard needs and never disagrees with the shard, at the cost of
/// a small ring instead of a shard's.
pub(crate) fn check(cfg: RingConfig) -> io::Result<()> {
    let group = cfg.recv_buffers != 0;
    let small = RingConfig {
        entries: MIN_RING_ENTRIES,
        cq_entries: None,
        recv_buffers: cfg.recv_buffers.min(CHECK_BUFFERS),
        recv_buffer_size: if group {
            crate::config::MIN_RECV_BUFFER_SIZE
        } else {
            cfg.recv_buffer_size
        },
        ..cfg
    };
    let files = table_len().min(CHECK_TABLE);
    let ring = ShardRing::new(&small, files)?;
    ring.make_table()?;
    ring.enable()
}

pub(crate) mod exits;
pub(crate) mod keeper;
mod recover;
#[cfg(test)]
pub(crate) use keeper::leaked;
mod arrivals;
#[cfg(test)]
mod tests;
#[cfg(test)]
mod tests_armed;
#[cfg(test)]
mod tests_cap;
#[cfg(test)]
mod tests_end;
#[cfg(test)]
mod tests_inc;
#[cfg(test)]
mod tests_recv;
#[cfg(test)]
mod tests_reserve;
#[cfg(test)]
mod tests_slab;
#[cfg(test)]
mod tests_zc;