//! Linux native AIO: one thread submits up to [NR_EVENTS] `O_DIRECT` reads through one AIO
//! context and reaps completions as they arrive. Unlike a thread per read, the device sees
//! them all at once and the submitting thread pays only a few microseconds per read.
//! [read_direct] reads a single range with blocking positioned reads instead.

use super::*;
use crate::{BLOB_PAGE_SIZE, BufMut as _, IoBufMut};
use commonware_utils::NZUsize;
use futures::stream::{self, BoxStream};

/// The `nr_events` each AIO context is created with (`io_setup`): the requests it processes
/// concurrently, and so the reads one blocking task submits through one context in
/// [`read_many`].
///
/// One context's worth is one device queue's worth of reads: NVMe queues hold 256 to 1024
/// commands, and a queue depth of 256 saturates the devices this runtime targets. A larger
/// `read_many` splits into that many reads per task, so issue cost spreads across tasks while
/// each task keeps a full queue in flight.
const NR_EVENTS: usize = 256;

/// Requests (iocbs) issued per `io_submit` call before the completions that have landed are
/// reaped.
///
/// The kernel spends about 2 us issuing each direct read, so issuing a whole context's worth
/// in one call takes longer than one read takes to complete. Reaping between calls lets the
/// earliest completions reach the stream while the rest are still being issued, at the cost
/// of one non-blocking `io_getevents` per call. Much smaller calls pay more in per-call
/// overhead than they return.
const IOCBS_PER_SUBMIT: usize = 32;

/// A pending read: its index among the [`read_many`] call's ranges and the physical file range.
struct Read {
    index: usize,
    offset: u64,
    len: usize,
}

/// One completed read, sent to the stream as soon as the kernel reports it.
type Completion = Result<(usize, IoBufsMut), Error>;

/// Return the blob's `O_DIRECT` descriptor. [`crate::storage::open_direct`] opens it on first use.
pub(super) fn direct(file: &Shared) -> Option<&File> {
    crate::storage::open_direct(&file.direct, file)
}

/// Return the superset of the `len` bytes at physical `offset` aligned to a blob page, as its
/// offset, the number of bytes it starts before the range, and its length. `None` when that
/// length overflows `usize`.
fn superset(offset: u64, len: usize) -> Option<(u64, usize, usize)> {
    let block: usize = Widen::widen(BLOB_PAGE_SIZE);
    let within = offset % u64::from(BLOB_PAGE_SIZE);
    let skip = usize::try_from(within).expect("an offset within a block must fit in usize");
    let aligned_len = skip.checked_add(len)?.checked_next_multiple_of(block)?;
    Some((offset - within, skip, aligned_len))
}

/// Read the `len` bytes at physical `offset` through the blob's [direct] descriptor.
///
/// Positioned reads fill an exact-size aligned allocation with the range's superset aligned to a
/// blob page, and the allocation is returned narrowed to the range. A caller that retains the
/// bytes retains the whole allocation.
///
/// `None` when the range must be read buffered instead: its superset ends beyond the largest
/// signed file offset or no allocation holds it, the blob has no direct descriptor, or a read
/// fails, reads nothing, or ends short of the range at a length that is not a multiple of a
/// blob page. A read that ends short at such a multiple continues from there.
pub(super) fn read_direct(file: &Shared, offset: u64, len: usize) -> Option<IoBufMut> {
    let block: usize = Widen::widen(BLOB_PAGE_SIZE);
    let (aligned_offset, skip, aligned_len) = superset(offset, len)?;
    let end = skip + len;
    aligned_offset
        .checked_add(Widen::widen(aligned_len))
        .filter(|superset_end| *superset_end <= i64::MAX as u64)?;
    let direct = direct(file)?;
    let mut buf = IoBufMut::try_with_alignment(aligned_len, NZUsize!(block))?;

    // Only the kernel writes the allocation, and only the bytes it reported become readable.
    let mut filled = 0;
    while filled < end {
        let position = libc::off_t::try_from(aligned_offset + Widen::widen(filled)).ok()?;
        // SAFETY: `filled < end <= aligned_len`, so the `aligned_len - filled` bytes at
        // `filled` lie within the allocation. This call owns the allocation, and nothing else
        // accesses it until the read returns.
        let n = unsafe {
            libc::pread(
                direct.as_raw_fd(),
                buf.as_mut_ptr().add(filled).cast(),
                aligned_len - filled,
                position,
            )
        };
        if n < 0 {
            if std::io::Error::last_os_error().kind() == std::io::ErrorKind::Interrupted {
                continue;
            }
            return None;
        }
        let n = n as usize;
        filled += n;
        if filled < end && (n == 0 || !n.is_multiple_of(block)) {
            return None;
        }
    }

    // SAFETY: the kernel initialized the first `filled >= end` bytes of the allocation.
    unsafe { buf.set_len(end) };
    buf.advance(skip);
    Some(buf)
}

/// Kernel ABI (`struct iocb`, little-endian layout).
#[repr(C)]
#[derive(Default, Clone, Copy)]
struct Iocb {
    aio_data: u64,
    aio_key: u32,
    aio_rw_flags: u32,
    aio_lio_opcode: u16,
    aio_reqprio: i16,
    aio_fildes: u32,
    aio_buf: u64,
    aio_nbytes: u64,
    aio_offset: i64,
    aio_reserved2: u64,
    aio_flags: u32,
    aio_resfd: u32,
}

/// Kernel ABI (`struct io_event`).
#[repr(C)]
#[derive(Default, Clone, Copy)]
struct IoEvent {
    data: u64,
    obj: u64,
    res: i64,
    res2: i64,
}

/// The `aio_lio_opcode` of a positioned read.
const IOCB_CMD_PREAD: u16 = 0;

/// A long-lived AIO context. Each submission in flight holds one of its own: `io_getevents`
/// returns any completion in a context, so sharing one would hand a submission the reads of
/// others.
///
/// Destroying a context waits for RCU grace periods (~30 ms), so one is destroyed only when
/// a submission fails with reads in flight, because the slab must outlive them. Every
/// pooled context charges [NR_EVENTS] against the host-wide `fs.aio-max-nr` budget (65,536
/// by default, so 256 contexts) until the process exits and the kernel reclaims them. A
/// submission that cannot obtain a context is served one blocking task per read, and other
/// users of Linux AIO on the same host see that budget as taken.
struct Context(libc::c_ulong);

impl Drop for Context {
    fn drop(&mut self) {
        // SAFETY: this handle belongs exclusively to this owner. Successful destruction
        // waits for every pending request before the destination buffers can be released.
        let result = unsafe { libc::syscall(libc::SYS_io_destroy, self.0) };
        if result != 0 {
            // Without a completion barrier, releasing the buffers would be unsafe.
            std::process::abort();
        }
    }
}

/// The process-wide free list of idle contexts. A submission takes one before `io_submit`
/// and returns it after reaping every completion, so the list grows to the peak number of
/// concurrent submissions. The lock guards one pop or push per submission and is never held
/// across a syscall.
static CONTEXTS: Mutex<Vec<Context>> = Mutex::new(Vec::new());

/// Take a pooled context or create one for [NR_EVENTS] concurrent requests. `None` when the kernel
/// cannot create one, for example when its outstanding-request limit (`fs.aio-max-nr`) is
/// exhausted.
fn take_context() -> Option<Context> {
    if let Some(ctx) = CONTEXTS.lock().pop() {
        return Some(ctx);
    }
    let mut ctx: libc::c_ulong = 0;
    // SAFETY: `io_setup` writes the new context handle to the valid out pointer.
    let r = unsafe { libc::syscall(libc::SYS_io_setup, NR_EVENTS as libc::c_ulong, &mut ctx) };
    (r == 0).then(|| Context(ctx))
}

/// Serve `read` with a buffered positioned read, as [`crate::Blob::read_at`] does for
/// [`ReadOptions::DONT_CACHE`] when it cannot read directly.
fn read_positioned(file: &Shared, read: &Read) -> Completion {
    let mut buf = IoBufMut::with_capacity(read.len);
    // SAFETY: read_exact_at fills all `len` bytes before the buffer is yielded.
    unsafe { buf.set_len(read.len) };
    Blob::read_exact_at(Cache::Disabled, file, buf.as_mut(), read.offset)?;
    Ok((read.index, buf.into()))
}

/// Submit every read in `reads` through `ctx`, delivering each one as it completes. A read
/// the kernel rejects, fails, or completes short is served by [`read_positioned`] instead.
/// Returns `ctx` once every submitted read has completed. On error, dropping it waits for the
/// reads still in flight.
fn submit(
    file: &Shared,
    direct: &File,
    ctx: Context,
    reads: Vec<Read>,
    tx: &tokio::sync::mpsc::UnboundedSender<Completion>,
) -> Result<Context, Error> {
    let fd = u32::try_from(direct.as_raw_fd()).expect("an open descriptor is non-negative");
    let n = reads.len();

    // Every read covers the superset of its range aligned to a blob page, which is the largest
    // logical block size of supported devices and so satisfies direct I/O's alignment of
    // offsets, lengths, and buffers. The supersets lie back to back in one aligned slab, and
    // the requested bytes are copied into exact-size buffers on completion.
    let block: usize = Widen::widen(BLOB_PAGE_SIZE);
    let mut spans = Vec::with_capacity(n);
    let mut slab_len = 0usize;
    for read in &reads {
        let (aligned_offset, skip, aligned_len) =
            superset(read.offset, read.len).ok_or(Error::OffsetOverflow)?;
        spans.push((aligned_offset, skip, slab_len, aligned_len));
        slab_len = slab_len
            .checked_add(aligned_len)
            .ok_or(Error::OffsetOverflow)?;
    }

    // Like an overflowing span, a slab no allocation layout can hold fails the submission.
    let mut slab =
        IoBufMut::try_with_alignment(slab_len, NZUsize!(block)).ok_or(Error::OffsetOverflow)?;
    let slab_ptr = slab.chunk_mut().as_mut_ptr();

    // This local drops before the slab. Context destruction waits for pending kernel
    // writes, including when submission or completion processing unwinds.
    let active = ctx;
    let mut iocbs: Vec<Iocb> = spans
        .iter()
        .enumerate()
        .map(|(i, &(aligned_offset, _, start, aligned_len))| Iocb {
            aio_data: Widen::widen(i),
            aio_lio_opcode: IOCB_CMD_PREAD,
            aio_fildes: fd,
            // SAFETY: `start + aligned_len <= slab_len`. The kernel writes through this
            // integer address, so it exposes the slab's provenance.
            aio_buf: Widen::widen(unsafe { slab_ptr.add(start) }.expose_provenance()),
            aio_nbytes: Widen::widen(aligned_len),
            // An offset past the largest signed file offset wraps negative. The kernel
            // rejects it at submission and the read is served positioned.
            aio_offset: aligned_offset as i64,
            ..Iocb::default()
        })
        .collect();

    // Each completion is delivered as it is reaped: a request the kernel completed in full
    // is copied out of the slab, anything else is re-served by a positioned read.
    let ptrs: Vec<*mut Iocb> = iocbs.iter_mut().map(|iocb| iocb as *mut Iocb).collect();
    let deliver = |event: &IoEvent| -> Result<(), Error> {
        let i = event.data as usize;
        let read = &reads[i];

        // A superset ending past the file completes short but still covers its range. A
        // read that failed or stopped before the end of its range (for example, it was
        // interrupted or capped at the most one read returns) is served by a positioned
        // read, which fails only where read_at would.
        let (_, skip, start, _) = spans[i];
        let item = if event.res < 0 || (event.res as usize) < skip + read.len {
            read_positioned(file, read)?
        } else {
            // SAFETY: this completed request initialized the requested range, which
            // is disjoint from every other request's destination in the slab.
            let bytes = unsafe { std::slice::from_raw_parts(slab_ptr.add(start + skip), read.len) };
            let mut buf = IoBufMut::with_capacity(read.len);
            buf.put_slice(bytes);
            #[cfg(test)]
            file.test.direct_reads.fetch_add(1, Ordering::Relaxed);
            (read.index, buf.into())
        };
        let _ = tx.send(Ok(item));
        Ok(())
    };

    // A failed submission accepted none of the requests passed to it. The first of them may
    // be one the kernel rejects (for example, a superset ending beyond the largest signed
    // file offset), so it is served here and the rest are submitted again.
    let mut events = vec![IoEvent::default(); n];
    let mut next = 0;
    let mut accepted = 0;
    let mut completed = 0;
    while next < n {
        let count = (n - next).min(IOCBS_PER_SUBMIT);
        // SAFETY: `ptrs[next..next + count]` are valid iocbs whose buffers lie in the slab,
        // which outlives every accepted request. The context accepts at least `n` requests
        // because `n <= NR_EVENTS`.
        let r = unsafe {
            libc::syscall(
                libc::SYS_io_submit,
                active.0,
                count as libc::c_long,
                ptrs.as_ptr().add(next),
            )
        };
        if r < 0 {
            let _ = tx.send(Ok(read_positioned(file, &reads[next])?));
            next += 1;
            continue;
        }
        next += r as usize;
        accepted += r as usize;
        if next < n {
            let got = reap(&active, &mut events[..accepted - completed], false)?;
            events[..got].iter().try_for_each(&deliver)?;
            completed += got;
        }
    }
    while completed < accepted {
        let got = reap(&active, &mut events[..accepted - completed], true)?;
        events[..got].iter().try_for_each(&deliver)?;
        completed += got;
    }
    Ok(active)
}

/// Reap up to `events.len()` completions from `ctx` into `events`, waiting for at least one
/// when `wait` is set and returning whatever has landed otherwise.
fn reap(ctx: &Context, events: &mut [IoEvent], wait: bool) -> Result<usize, Error> {
    if events.is_empty() {
        return Ok(0);
    }
    let zero = libc::timespec {
        tv_sec: 0,
        tv_nsec: 0,
    };
    let timeout: *const libc::timespec = if wait { std::ptr::null() } else { &zero };
    loop {
        // SAFETY: `events` has room for `events.len()` events and `timeout` is null or a
        // valid timespec.
        let got = unsafe {
            libc::syscall(
                libc::SYS_io_getevents,
                ctx.0,
                wait as libc::c_long,
                events.len() as libc::c_long,
                events.as_mut_ptr(),
                timeout,
            )
        };
        if got >= 0 {
            return Ok(got as usize);
        }

        // A signal interrupts the wait before it reaps anything, and the kernel never
        // restarts it.
        let err = std::io::Error::last_os_error();
        if err.kind() != std::io::ErrorKind::Interrupted {
            return Err(err.into());
        }
    }
}

/// Serve `reads` without native AIO: one blocking task per read, as
/// [`crate::Blob::read_at`] does.
fn read_positioned_each(
    file: &Arc<Shared>,
    reads: Vec<Read>,
    tx: &tokio::sync::mpsc::UnboundedSender<Completion>,
) {
    for read in reads {
        let (file, tx) = (file.clone(), tx.clone());
        task::spawn_blocking(move || {
            let _ = tx.send(read_positioned(&file, &read));
        });
    }
}

/// Run one submission on the calling (blocking) thread.
fn run(file: &Arc<Shared>, reads: Vec<Read>, tx: &tokio::sync::mpsc::UnboundedSender<Completion>) {
    // Without a direct descriptor (the filesystem rejects direct I/O, or a transient open
    // failure) or a context (for example, the kernel's request limit is exhausted), the
    // submission is served by one buffered positioned read per range, each on its own
    // blocking task.
    let Some(direct) = direct(file) else {
        return read_positioned_each(file, reads, tx);
    };
    let Some(ctx) = take_context() else {
        return read_positioned_each(file, reads, tx);
    };
    match submit(file, direct, ctx, reads, tx) {
        Ok(ctx) => CONTEXTS.lock().push(ctx),
        Err(err) => {
            let _ = tx.send(Err(err));
        }
    }
}

/// Read `ranges` of a blob's data with direct I/O, yielding each range's `(index, bytes)` as
/// it completes. Offsets are relative to `data_offset`. Without an error every range is
/// yielded exactly once, and an error ends the stream.
///
/// Every [NR_EVENTS] reads run on one blocking task through one AIO context, so the device
/// sees them all at once and no thread blocks per read. The reads use direct I/O on every
/// Linux kernel, so they do not read data the OS page cache holds. Their concurrency assumes
/// filesystem-native direct I/O (such as ext4 or XFS): a filesystem that serves direct I/O
/// through a buffered fallback stays correct but completes each submission's reads one at a
/// time.
pub(super) fn read_many(
    file: &Arc<Shared>,
    data_offset: u64,
    ranges: &[(u64, usize)],
) -> BoxStream<'static, Completion> {
    // Translate each range to its physical file range, keeping its index so the stream can
    // name it. An offset beyond the file's address space or a length no buffer can hold fails
    // the call before anything is read, whether the reads would be direct or buffered.
    let mut reads = Vec::with_capacity(ranges.len());
    for (index, &(offset, len)) in ranges.iter().enumerate() {
        let Some(offset) = physical_offset(data_offset, offset, len) else {
            return stream::iter(vec![Err(Error::OffsetOverflow)]).boxed();
        };
        reads.push(Read { index, offset, len });
    }

    // Run every [NR_EVENTS] reads as one submission on its own blocking task. The tasks share
    // the channel the stream drains, so completions arrive in whatever order they land.
    let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
    let mut reads = reads.into_iter().peekable();
    while reads.peek().is_some() {
        let chunk: Vec<_> = reads.by_ref().take(NR_EVENTS).collect();
        let (file, tx) = (file.clone(), tx.clone());
        task::spawn_blocking(move || run(&file, chunk, &tx));
    }

    // The stream ends once every range is yielded or at the first error. A submitting task
    // that ends without reporting its reads (it panicked, or the runtime shut down) closes
    // the channel early, which is an error.
    stream::unfold((rx, ranges.len()), |(mut rx, remaining)| async move {
        if remaining == 0 {
            return None;
        }
        let item = rx.recv().await.unwrap_or(Err(Error::ReadFailed));
        let remaining = if item.is_ok() { remaining - 1 } else { 0 };
        Some((item, (rx, remaining)))
    })
    .boxed()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        Blob as _, Storage as _,
        storage::{
            Layout,
            tokio::{Config, Storage},
        },
    };
    use futures::TryStreamExt as _;
    use std::{
        path::PathBuf,
        sync::atomic::{AtomicBool, Ordering},
        time::Duration,
    };

    /// Open a blob holding `data` in a fresh directory that supports direct I/O.
    async fn direct_blob(label: &str, data: Vec<u8>) -> (Storage, Blob, PathBuf) {
        let directory =
            std::env::temp_dir().join(format!("storage_tokio_aio_{label}_{}", std::process::id()));
        let storage = Storage::new(Config::new(directory.clone(), Layout::ALL));
        let (blob, _) = storage.open("partition", b"blob").await.unwrap();
        blob.write_at(0, data, WriteOptions::SYNC).await.unwrap();
        assert!(
            direct(&blob.shared).is_some(),
            "temporary directory must support O_DIRECT"
        );
        (storage, blob, directory)
    }

    async fn remove(storage: Storage, blob: Blob, directory: PathBuf) {
        drop(blob);
        storage.remove("partition", None).await.unwrap();
        drop(storage);
        std::fs::remove_dir_all(directory).unwrap();
    }

    #[tokio::test]
    async fn test_failed_submission_keeps_slab_until_reads_complete() {
        const LEN: usize = 64 << 10;
        let block: usize = Widen::widen(BLOB_PAGE_SIZE);
        let (storage, blob, directory) = direct_blob("failed_submission", vec![0xAB; LEN]).await;
        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();

        // The kernel accepts the first read and rejects the second (its offset exceeds the
        // largest signed file offset), whose positioned read then fails, so the submission
        // returns an error while the first read may still be in flight. A slab released
        // before that read completes receives the file's bytes after its memory is reused.
        for _ in 0..20 {
            let reads = vec![
                Read {
                    index: 0,
                    offset: 0,
                    len: LEN,
                },
                Read {
                    index: 1,
                    offset: 1 << 63,
                    len: block,
                },
            ];
            let ctx = take_context().unwrap();
            let direct = direct(&blob.shared).unwrap();
            assert!(submit(&blob.shared, direct, ctx, reads, &tx).is_err());

            let canary = IoBufMut::zeroed_with_alignment(LEN + block, NZUsize!(block));
            std::thread::sleep(Duration::from_millis(10));
            assert!(!canary.as_ref().contains(&0xAB));
        }
        remove(storage, blob, directory).await;
    }

    /// Ranges that each fit an allocation but whose aligned supersets no slab can hold together
    /// fail the submission before anything is submitted or delivered.
    #[tokio::test]
    async fn test_submission_beyond_one_slab_fails() {
        let (storage, blob, directory) = direct_blob("slab_overflow", vec![0; 4096]).await;
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        // Each range passes read_many's length check, but together their supersets exceed the
        // largest allocation layout.
        let len = 1usize << (usize::BITS - 2);
        let reads = (0..2)
            .map(|index| Read {
                index,
                offset: 0,
                len,
            })
            .collect();
        let ctx = take_context().unwrap();
        let direct = direct(&blob.shared).unwrap();
        let result = submit(&blob.shared, direct, ctx, reads, &tx);

        // The submission fails without delivering a completion.
        assert!(matches!(result, Err(Error::OffsetOverflow)));
        drop(tx);
        assert!(rx.recv().await.is_none());

        remove(storage, blob, directory).await;
    }

    #[tokio::test]
    async fn test_submission_survives_signals() {
        extern "C" fn ignore(_: libc::c_int) {}

        // A handled signal interrupts a blocked io_getevents, which the kernel never restarts.
        // SAFETY: a zeroed sigaction is valid, `ignore` has the handler ABI, and both
        // pointers are valid.
        let previous = unsafe {
            let mut action: libc::sigaction = std::mem::zeroed();
            action.sa_sigaction = ignore as extern "C" fn(libc::c_int) as usize;
            let mut previous: libc::sigaction = std::mem::zeroed();
            assert_eq!(libc::sigaction(libc::SIGUSR1, &action, &mut previous), 0);
            previous
        };

        // One large read keeps io_getevents waiting long enough to be interrupted.
        const LEN: usize = 64 << 20;
        let (storage, blob, directory) = direct_blob("signals", vec![0xAB; LEN]).await;
        let reads = vec![Read {
            index: 0,
            offset: blob.data_offset,
            len: LEN,
        }];
        let shared = blob.shared.clone();
        let done = Arc::new(AtomicBool::new(false));
        let (thread_tx, thread_rx) = std::sync::mpsc::channel();
        let submitter = {
            let done = done.clone();
            std::thread::spawn(move || {
                // SAFETY: pthread_self has no preconditions.
                thread_tx.send(unsafe { libc::pthread_self() }).unwrap();
                let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
                let ctx = take_context().unwrap();
                let direct = direct(&shared).unwrap();
                let result = submit(&shared, direct, ctx, reads, &tx);
                done.store(true, Ordering::Release);
                (result.is_ok(), rx.try_recv().ok(), rx.try_recv().is_err())
            })
        };
        let thread = thread_rx.recv().unwrap();
        while !done.load(Ordering::Acquire) {
            // SAFETY: the submitter has not been joined, so its thread id is still valid.
            unsafe { libc::pthread_kill(thread, libc::SIGUSR1) };
            std::thread::sleep(Duration::from_micros(10));
        }
        let (submitted, first, drained) = submitter.join().unwrap();

        // SAFETY: restores the action saved above.
        unsafe { libc::sigaction(libc::SIGUSR1, &previous, std::ptr::null_mut()) };
        assert!(submitted);
        assert!(drained);
        let Some(Ok((0, bufs))) = first else {
            panic!("expected the read");
        };
        assert_eq!(bufs.coalesce().as_ref(), vec![0xAB; LEN].as_slice());
        remove(storage, blob, directory).await;
    }

    /// An open that fails for a reason that may not recur leaves the blob free to try again,
    /// while a filesystem's rejection is final for the blob.
    #[tokio::test]
    async fn test_direct_open_retries_after_descriptor_limit() {
        let directory = std::env::temp_dir().join(format!(
            "storage_tokio_aio_retry_open_{}",
            std::process::id()
        ));
        let storage = Storage::new(Config::new(directory.clone(), Layout::ALL));
        let (blob, _) = storage.open("partition", b"blob").await.unwrap();
        blob.write_at(0, vec![0; Widen::widen(BLOB_PAGE_SIZE)], WriteOptions::SYNC)
            .await
            .unwrap();

        // Forbid further descriptors for the duration of one open attempt, which must leave
        // the blob free to try again.
        // SAFETY: getrlimit and setrlimit take valid pointers to an rlimit.
        let denied = unsafe {
            let mut previous: libc::rlimit = std::mem::zeroed();
            assert_eq!(libc::getrlimit(libc::RLIMIT_NOFILE, &mut previous), 0);
            let none = libc::rlimit {
                rlim_cur: 0,
                rlim_max: previous.rlim_max,
            };
            assert_eq!(libc::setrlimit(libc::RLIMIT_NOFILE, &none), 0);
            let denied = direct(&blob.shared);
            assert_eq!(libc::setrlimit(libc::RLIMIT_NOFILE, &previous), 0);
            denied.is_none()
        };
        assert!(denied);
        assert!(blob.shared.direct.get().is_none());

        // The next direct read settles the cell. It opens the descriptor or records the
        // filesystem's rejection of direct I/O.
        let opened = direct(&blob.shared).is_some();
        assert_eq!(opened, matches!(blob.shared.direct.get(), Some(Some(_))));
        assert!(blob.shared.direct.get().is_some());
        remove(storage, blob, directory).await;
    }

    /// More ranges than one context holds are split across blocking tasks and still yield
    /// every range exactly once, each served by its direct read.
    #[tokio::test]
    async fn test_read_many_serves_more_ranges_than_one_submission() {
        let block: usize = Widen::widen(BLOB_PAGE_SIZE);
        let data: Vec<u8> = (0..8 * block).map(|i| (i % 251) as u8).collect();
        let (storage, blob, directory) = direct_blob("many_submissions", data.clone()).await;

        let ranges: Vec<(u64, usize)> = (0..2 * NR_EVENTS + 1)
            .map(|i| (Widen::widen((i % 16) * 2048), 2048))
            .collect();
        let before = blob.shared.test.direct_reads.load(Ordering::Relaxed);
        let mut served: Vec<(usize, IoBufsMut)> =
            read_many(&blob.shared, blob.data_offset, &ranges)
                .try_collect()
                .await
                .unwrap();
        assert_eq!(
            blob.shared.test.direct_reads.load(Ordering::Relaxed) - before,
            ranges.len()
        );
        served.sort_by_key(|(index, _)| *index);
        assert!(served.iter().map(|(index, _)| *index).eq(0..ranges.len()));
        for (index, bufs) in served {
            let (offset, len) = ranges[index];
            let offset = usize::try_from(offset).unwrap();
            assert_eq!(bufs.coalesce().as_ref(), &data[offset..offset + len]);
        }
        remove(storage, blob, directory).await;
    }

    #[tokio::test]
    async fn test_reads_without_context_are_served_individually() {
        let block: usize = Widen::widen(BLOB_PAGE_SIZE);
        let data: Vec<u8> = (0..3 * block).map(|i| (i % 251) as u8).collect();
        let (storage, blob, directory) = direct_blob("without_context", data.clone()).await;

        // Each read is served on its own blocking task, and one past the end fails alone.
        let ranges = [
            (10u64, 5000usize),
            (u64::from(BLOB_PAGE_SIZE) - 1, 2),
            (data.len() as u64, 1),
        ];
        let reads = ranges
            .iter()
            .enumerate()
            .map(|(index, &(offset, len))| Read {
                index,
                offset: offset + blob.data_offset,
                len,
            })
            .collect();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        read_positioned_each(&blob.shared, reads, &tx);
        drop(tx);
        let mut served = Vec::new();
        let mut failed = 0;
        while let Some(item) = rx.recv().await {
            match item {
                Ok(item) => served.push(item),
                Err(_) => failed += 1,
            }
        }
        assert_eq!(failed, 1);
        served.sort_by_key(|(index, _)| *index);
        assert_eq!(served.len(), 2);
        for ((offset, len), (index, bufs)) in ranges.iter().zip(served) {
            let offset = *offset as usize;
            assert_eq!(
                bufs.coalesce().as_ref(),
                &data[offset..offset + len],
                "{index}"
            );
        }
        remove(storage, blob, directory).await;
    }

    /// A single direct read covers its range's superset aligned to a blob page and hands out
    /// only the range, including a superset that ends past an unaligned end of file.
    ///
    /// A read that reads nothing or ends short of the range at an unaligned length, a read that
    /// ends at a blob page at the end of file and reads nothing when it continues, and a superset
    /// that is not representable leave the range to a buffered read. Regular descriptors stand in
    /// for direct ones. A continued read that succeeds needs a read beyond the kernel's
    /// single-read limit, which only the ignored test_read_many_beyond_single_read_limit makes.
    #[commonware_macros::test_group("miri")]
    #[test]
    fn test_read_direct_covers_the_aligned_superset() {
        let block: usize = Widen::widen(BLOB_PAGE_SIZE);
        let directory = std::env::temp_dir().join(format!(
            "storage_tokio_aio_read_direct_{}",
            std::process::id()
        ));
        let storage = Storage::new(Config::new(directory.clone(), Layout::ALL));
        let data: Vec<u8> = (0..3 * block).map(|i| (i % 251) as u8).collect();

        // Each open's direct descriptor reads the first `len` bytes of `data`.
        let open = |name: &[u8], len: usize| {
            let path = directory.join(hex(name));
            std::fs::write(&path, &data[..len]).unwrap();
            let (generation, _, _) = storage.pending.admit("partition", name, false).unwrap();
            let blob = Blob::new(
                File::open(&path).unwrap(),
                0,
                storage.hold.clone(),
                generation,
            );
            assert!(
                blob.shared
                    .direct
                    .set(Some(File::open(&path).unwrap()))
                    .is_ok()
            );
            blob
        };

        // Miri ends file reads short at random, and a short read leaves the range to a buffered
        // read. Under Miri, a read the file can serve in full is retried until it succeeds.
        let served = |blob: &Blob, offset: u64, len: usize| {
            let attempts = if cfg!(miri) { 64 } else { 1 };
            (0..attempts)
                .find_map(|_| read_direct(&blob.shared, offset, len))
                .expect("the file must serve the range")
        };

        // A file ending inside its last block serves ranges up to its end, each from its
        // superset's exact-size aligned allocation.
        let unaligned = open(b"unaligned", 3 * block - 50);
        for (offset, len) in [(0, block), (100, 5000), (2 * block + 10, block - 60)] {
            let buf = served(&unaligned, offset as u64, len);
            let skip = offset % block;
            assert_eq!(buf.as_ref(), &data[offset..offset + len]);
            assert_eq!(buf.as_ref().as_ptr().addr() % block, skip);
            assert_eq!(buf.capacity(), (skip + len).next_multiple_of(block) - skip);
        }

        // Past its end a read ends short at an unaligned length or reads nothing, and a
        // superset whose length overflows or that ends beyond the largest signed file offset
        // is never read.
        for (offset, len) in [
            (2 * block as u64 + 10, block),
            (3 * block as u64, 1),
            (0, usize::MAX - 10),
            (i64::MAX as u64 - 10, 1),
        ] {
            assert!(read_direct(&unaligned.shared, offset, len).is_none());
        }

        // A file ending at a block boundary ends a longer read short there, and continuing reads
        // nothing, so that range is left to a buffered read. A range whose superset ends at the
        // file's end is served.
        let aligned = open(b"aligned", 2 * block);
        assert!(read_direct(&aligned.shared, 100, 2 * block).is_none());
        let buf = served(&aligned, 100, block);
        assert_eq!(buf.as_ref(), &data[100..100 + block]);

        drop(unaligned);
        drop(aligned);
        drop(storage);
        std::fs::remove_dir_all(directory).unwrap();
    }
}