nuclei 0.5.1

Proactive IO & runtime system
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
//! 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 ring's own copy of the
//! descriptor. Either way the ring holds its own reference to the file, so
//! a handle that closes its descriptor on any thread never changes what an
//! entry reaches, queued or submitted. A slot, and the file it holds, is let
//! go 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. At most as many
//! are in the kernel as the completion queue holds; the rest wait here, in
//! order, entry and memory together, and go in as completions free room.
//! One that is cancelled or whose source goes while it waits ends here,
//! never submitted. A side has one operation out at a time, so the
//! operations here are bounded by the sources.
//!
//! 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.

use core::cell::{Cell, RefCell};
use std::collections::VecDeque;
use std::io;
use std::os::fd::RawFd;
use std::time::Duration;

use super::recv::RecvGroup;
use crate::fs::ring::FileOp;
use crate::io::source::ops::Done;
use crate::io::source::ring::{Arrival, RingMem, RingOp};
use crate::io::sys;
use crate::rt::backend::with_room;
use crate::rt::park::Notifier;
use crate::rt::park::ring::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_MORE, IOSQE_FIXED_FILE, 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,
    files: RefCell<Files>,
    /// Operations submitted but not in the kernel yet, by user data, with
    /// their entries, in order.
    room: RefCell<VecDeque<(u64, Sqe)>>,
    /// Operations in the kernel, and the most it may hold.
    in_kernel: Cell<usize>,
    cap: usize,
    /// Completions of the last wait, reused.
    cqes: RefCell<Vec<IoUringCqe>>,
    /// The buffers multishot receives draw from; `None` when the
    /// configuration asks for none or the kernel could not make them.
    recv: RefCell<Option<RecvGroup>>,
    /// 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)>,
    /// Files' calls submitted, for a test.
    #[cfg(test)]
    file_calls: Cell<usize>,
}

impl ShardRing {
    /// A ring as `cfg` describes with a table of `files` fixed files,
    /// disabled until its first wait or submission.
    pub(crate) fn new(cfg: &RingConfig, files: u32) -> io::Result<Self> {
        let poller = RingPoller::new(cfg, files)?;
        let recv = match cfg.recv_buffers {
            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}"),
                    ));
                }
                // Short of memory: each read takes a buffer of the pool.
                Err(err) if err.raw_os_error() == Some(libc::ENOMEM) => 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);
        let waitid = poller.ring().register_probe().is_ok_and(|ops| {
            ops.contains(crate::syscore::linux::iouring::ring::sys::Opcode::Waitid)
        });
        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)),
            exit_ops: RefCell::new(Ops::new(1 << 30)),
            waitid,
            files: RefCell::new(Files::new(files)),
            room: RefCell::new(VecDeque::new()),
            in_kernel: Cell::new(0),
            cap,
            cqes: RefCell::new(Vec::with_capacity(256)),
            recv: RefCell::new(recv),
            #[cfg(test)]
            counts: Cell::new((0, 0, 0)),
            #[cfg(test)]
            file_calls: Cell::new(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()
    }

    /// 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)
    }

    /// 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)
    }

    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. With the kernel full it
    /// waits for room; its slot is named from now on either way.
    pub(crate) fn submit(&self, op: RingOp, sqe: Sqe) -> u64 {
        self.files.borrow_mut().named(op.file);
        let sqe = match self.files.borrow().copy_of(op.file) {
            Some(copy) => sqe.fd(copy),
            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));
        }
        let inserted = self.ops.borrow_mut().insert(op);
        let user_data = match inserted {
            Ok(user_data) => user_data,
            // Past 2^32 - 2 operations at once (two per source at most).
            Err(op) => {
                self.end(op, -libc::ENOBUFS);
                return IGNORED;
            }
        };
        self.enter(user_data, sqe);
        user_data
    }

    /// Puts the entry of the operation of `user_data` into the kernel, or
    /// behind the ones waiting for room.
    fn enter(&self, user_data: u64, sqe: Sqe) {
        let sqe = sqe.user_data(user_data);
        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 holds a copy of the descriptor 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();
        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;
            }
        };
        self.enter(user_data, sqe);
    }

    /// 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.in_kernel.set(self.in_kernel.get().saturating_sub(1));
        self.admit_waiting();
        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 operation of `user_data`: at once if it waits for room,
    /// else by the kernel (its completion says what it did).
    pub(crate) fn cancel(&self, user_data: u64) {
        let waiting = {
            let mut room = self.room.borrow_mut();
            room.iter()
                .position(|(u, _)| *u == user_data)
                .and_then(|at| room.remove(at))
        };
        if waiting.is_some() {
            if let Some(op) = self.ops.borrow_mut().remove(user_data) {
                self.end(op, -libc::ECANCELED);
            }
            return;
        }
        if self.ops.borrow_mut().get_mut(user_data).is_some() {
            self.poller
                .borrow_mut()
                .push(Sqe::async_cancel(user_data).user_data(IGNORED));
        }
    }

    /// 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, the ring's
    /// own copy of `fd`.
    pub(crate) fn attach(&self, fd: RawFd) -> io::Result<u32> {
        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 => {
                let copy = sys::duplicate(fd)?;
                Ok(self.files.borrow_mut().take_copy(copy))
            }
        }
    }

    /// The source of `slot` went: its sides' operations `out` are
    /// cancelled, and the slot, with its file, is let go once no entry
    /// names it (at once if none is out). A cancel by user data needs no
    /// file, so nothing else keeps the slot.
    pub(crate) fn detach(&self, slot: u32, out: [Option<u64>; 2]) {
        for user_data in out.into_iter().flatten() {
            self.cancel(user_data);
        }
        let freed = self.files.borrow_mut().release(slot);
        self.freed(slot, freed);
    }

    /// A slot that came free: its fixed file is let go (a copy closed as
    /// it was freed).
    fn freed(&self, slot: u32, freed: bool) {
        if freed && self.files.borrow().is_fixed(slot) {
            // A failure leaves the file in the free slot until the slot's
            // next source replaces it.
            let _ = self.poller.borrow().update_file(slot, -1);
        }
    }

    /// 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.
    pub(crate) fn wait(&self, timeout: Option<Duration>) -> io::Result<(usize, bool)> {
        self.take_back();
        let mut cqes = core::mem::take(&mut *self.cqes.borrow_mut());
        let woken = self.poller.borrow_mut().wait(timeout, &mut cqes);
        let handled = cqes.len();
        for cqe in cqes.drain(..) {
            self.complete(cqe);
        }
        *self.cqes.borrow_mut() = cqes;
        Ok((handled, woken?))
    }

    fn complete(&self, cqe: IoUringCqe) {
        if cqe.user_data & FILE_TAG != 0 {
            if cqe.user_data & EXIT_TAG != 0 {
                self.complete_exit(cqe.user_data, cqe.res);
            } else {
                self.complete_file(cqe.user_data, cqe.res);
            }
            return;
        }
        if let Some(bid) = cqe_buffer_id(cqe.flags) {
            self.arrived(cqe.user_data, bid, cqe.res);
        } else if cqe.res >= 0 {
            self.accepted(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.in_kernel.set(self.in_kernel.get().saturating_sub(1));
        self.admit_waiting();
        self.end(op, cqe.res);
    }

    /// Buffer `bid` of the receive group came with `res` bytes for the
    /// multishot receive of `user_data`: its source gets them as a chunk.
    /// Past the cap of bytes it holds, the receive is stopped; its source
    /// asks for another once its reader took them.
    fn arrived(&self, user_data: u64, bid: u16, res: i32) {
        let mut stop = false;
        let delivery = {
            let mut recv = self.recv.borrow_mut();
            let Some(group) = recv.as_mut() else {
                return;
            };
            let mut ops = self.ops.borrow_mut();
            let held = ops.get_mut(user_data).and_then(|op| match &mut op.mem {
                RingMem::RecvMulti { held, stopping } => Some((op.epoch, held, stopping, false)),
                RingMem::RecvFromMulti { held, stopping, .. } => {
                    Some((op.epoch, held, stopping, true))
                }
                _ => None,
            });
            match (u32::try_from(res), held) {
                (Ok(len @ 1..), Some((epoch, held, stopping, datagram))) => {
                    match group.hand_out(user_data, bid, len) {
                        Some(chunk) => {
                            *held += len as usize;
                            if *held > group.cap() && !*stopping {
                                *stopping = true;
                                stop = true;
                            }
                            ops.get_mut(user_data).map(|op| {
                                (std::sync::Arc::clone(&op.source), chunk, epoch, datagram)
                            })
                        }
                        None => None,
                    }
                }
                // No bytes, or no receive to give them to: the buffer goes
                // back at once.
                _ => {
                    group.give_back(bid);
                    None
                }
            }
        };
        if stop {
            self.cancel(user_data);
        }
        // Handed over with nothing borrowed: the source may submit at once.
        if let Some((source, chunk, epoch, datagram)) = delivery {
            #[cfg(test)]
            {
                let (armed, arrived, dry) = self.counts.get();
                self.counts.set((armed, arrived + 1, dry));
            }
            let arrival = if datagram {
                // A buffer not laid out as the kernel says is only given back.
                match crate::io::source::ring::datagram_of(chunk, source.kind()) {
                    Some((payload, from)) => Arrival::Datagram(payload, from),
                    None => return,
                }
            } else {
                Arrival::Chunk(chunk)
            };
            source.ring_arrival(arrival, epoch);
        }
    }

    /// A completion `res` of the operation of `user_data` that carries no
    /// buffer: for a multishot accept, a connection (its descriptor), handed
    /// to the source as it comes. Past the cap of connections one accept
    /// hands over, the accept is stopped; its source asks for another once
    /// its tasks took them.
    fn accepted(&self, user_data: u64, res: i32) {
        let delivery = {
            let mut ops = self.ops.borrow_mut();
            let Some(op) = ops.get_mut(user_data) else {
                return;
            };
            let RingMem::AcceptMulti { handed, stopping } = &mut op.mem else {
                return;
            };
            *handed += 1;
            let stop = *handed > ACCEPT_CAP && !*stopping;
            if stop {
                *stopping = true;
            }
            // SAFETY: the kernel made this descriptor for us.
            let conn =
                unsafe { <std::os::fd::OwnedFd as std::os::fd::FromRawFd>::from_raw_fd(res) };
            (std::sync::Arc::clone(&op.source), conn, op.epoch, stop)
        };
        let (source, conn, epoch, stop) = delivery;
        if stop {
            self.cancel(user_data);
        }
        source.ring_arrival(Arrival::Conn(conn), epoch);
    }

    /// Takes in the buffers chunks gave back and gives them to the kernel
    /// again, counting each off the receive it went to.
    fn take_back(&self) {
        let mut recv = self.recv.borrow_mut();
        let Some(group) = recv.as_mut() else {
            return;
        };
        let mut ops = self.ops.borrow_mut();
        group.take_back(|user_data, len| {
            if let Some(RingMem::RecvMulti { held, .. } | RingMem::RecvFromMulti { held, .. }) =
                ops.get_mut(user_data).map(|op| &mut op.mem)
            {
                *held = held.saturating_sub(len as usize);
            }
        });
    }

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

    /// The shard is going, its sources detached: every operation still out
    /// is cancelled and its completion waited for, for at most
    /// [`SHUTDOWN_WAIT`]. One that never completes keeps its memory for
    /// good (the kernel may still write it), and is counted (`leaked`).
    pub(crate) fn shut_down(&self) {
        let waiting: Vec<u64> = self.room.borrow_mut().drain(..).map(|(u, _)| u).collect();
        for user_data in waiting {
            // Never submitted: nothing points at its memory. A file's call
            // goes to the pool; a child goes to be watched elsewhere.
            if user_data & (FILE_TAG | EXIT_TAG) == FILE_TAG | EXIT_TAG {
                self.drop_waiting_exit(user_data);
            } else if user_data & FILE_TAG != 0 {
                let op = self.file_ops.borrow_mut().remove(user_data & !FILE_TAG);
                if let Some(op) = op {
                    drop(op.complete(-libc::ECANCELED));
                }
            } else {
                let op = self.ops.borrow_mut().remove(user_data);
                if let Some(op) = op {
                    self.end(op, -libc::ECANCELED);
                }
            }
        }
        if self.in_flight() > 0 {
            self.poller
                .borrow_mut()
                .push(Sqe::async_cancel_any(true).user_data(IGNORED));
        }
        self.settle(|| {});
        let stuck: Vec<RingOp> = self.ops.borrow_mut().drain().collect();
        let stuck_files: Vec<FileOp> = self.file_ops.borrow_mut().drain().collect();
        let stuck_exits: Vec<ExitOp> = self.exit_ops.borrow_mut().drain().collect();
        LEAKED.fetch_add(
            stuck.len() + stuck_files.len() + stuck_exits.len(),
            core::sync::atomic::Ordering::Relaxed,
        );
        // A file's call still in the kernel keeps its buffer and its
        // descriptor for good, a child's wait its info and its pidfd.
        core::mem::forget(stuck_files);
        core::mem::forget(stuck_exits);
        if !stuck.is_empty() {
            // A receive still in the kernel may write the group's buffers:
            // they stay registered and mapped for good.
            core::mem::forget(self.recv.borrow_mut().take());
        }
        core::mem::forget(stuck);
    }
}

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()
    }

    /// Submits what is queued and waits until no operation is out, for at
    /// most [`SHUTDOWN_WAIT`]: what its sources cancelled as they went.
    /// `between` runs before each wait (the shard's mailbox, where a
    /// source's holder on another thread leaves its cancel).
    pub(crate) fn settle(&self, mut between: impl FnMut()) {
        let deadline = std::time::Instant::now() + SHUTDOWN_WAIT;
        loop {
            between();
            if self.in_flight() == 0 {
                return;
            }
            let left = deadline.saturating_duration_since(std::time::Instant::now());
            if left.is_zero()
                || self
                    .wait(Some(left.min(Duration::from_millis(10))))
                    .is_err()
            {
                return;
            }
        }
    }
}

/// 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;

/// How long a shard that goes waits for its ring's cancelled operations.
const SHUTDOWN_WAIT: Duration = Duration::from_secs(1);

static LEAKED: core::sync::atomic::AtomicUsize = core::sync::atomic::AtomicUsize::new(0);

/// Operations whose completion never came as their shard went: their memory
/// is kept for good.
#[cfg(test)]
pub(crate) fn leaked() -> usize {
    LEAKED.load(core::sync::atomic::Ordering::Relaxed)
}

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

    /// 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()
    }

    /// 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)
    }

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

    /// 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 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), at most [`MAX_TABLE`], whose slots
/// cost kernel memory per ring. Sources past it run on copies; 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)
        .clamp(64, 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 = 64;

/// Whether the kernel gives a shard of `cfg` everything its ring needs: the
/// ring, its disabled start, the table of fixed files, the group of buffers
/// for multishot receives, and the enabling by the thread that runs it. It
/// makes the ring through [`ShardRing::for_shard`] and enables it here, 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.
pub(crate) fn check(cfg: RingConfig) -> io::Result<()> {
    ShardRing::for_shard(cfg)?.enable()
}

pub(crate) mod exits;
#[cfg(test)]
mod tests;
#[cfg(test)]
mod tests_recv;