windows_ioring_sys/

event_delivery.rs

// Copyright (c) 2026 Mike Grier
//! Model A delivery: the ring's completion event wired to a thread-pool wait
//! (M4).

use std::io;
use std::sync::{Arc, Mutex, MutexGuard};

use windows_sys::Win32::Storage::FileSystem::IORING_OP_CODE;
use windows_threadpool_sys::callback_env::CallbackEnviron;

use crate::batch::Batch;
use crate::capability::RingVersion;
use crate::completion_wake::CompletionWake;
use crate::ring::{Completion, IoRing, Op, RingInfo};
/// What a delivery hands a caller for each completion: the completion, and
/// whatever the ring was holding for it, exactly as [`IoRing::try_pop`]
/// returns them.
///
/// The two `Option`s answer different questions, and a callback must not
/// collapse them. The **outer** `None` means the push created no entry at all
/// -- a `_raw` flush or cancel, or [`IoRing::push_raw`]. `Some((None, extra))`
/// is an `_owned` push that never had a buffer, a flush or a cancellation, and
/// its sidecar still arrives: discarding it on the inner `None` would lose
/// which group of writes the flush belonged to. `try_pop`'s "What each `None`
/// means" is the authoritative statement.
type OnCompletion<T, X> = dyn Fn(Completion, Option<(Option<T>, X)>) + Send + Sync;

/// Pop every completion currently available and hand each to `on_completion`.
///
/// Each pop is its own short lock: `on_completion` always runs with the
/// mutex released, so a slow callback does not block a submitter, and a
/// callback that calls [`EventDelivery::scope`] and submits cannot deadlock
/// against this loop. That is contract, not just how this happens to work:
/// [`EventDelivery::new`] states it, and [D-83](../DESIGN-NOTES.md#d-83)
/// records it.
fn drain<T, X>(ring: &Mutex<IoRing<T, X>>, on_completion: &OnCompletion<T, X>) {
    loop {
        let popped = {
            let mut ring = ring
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            ring.try_pop()
        };
        match popped {
            Ok(Some((completion, held))) => on_completion(completion, held),
            Ok(None) => break,
            Err(error) => {
                debug_assert!(
                    false,
                    "IoRing::try_pop failed during event-driven drain: {error}"
                );
                break;
            }
        }
    }
}

/// Delivers an [`IoRing`]'s completions on thread-pool callback threads via
/// its completion event, rather than a caller polling [`IoRing::try_pop`]
/// itself (Model A, D-3).
///
/// # Not a `CleanupGroup` member
///
/// There is no way to add an already-built `EventDelivery` to a
/// [`CleanupGroup`](windows_threadpool_sys::cleanup_group::CleanupGroup), for
/// the same reason `CleanupGroup` excludes
/// [`ThreadpoolIo`](windows_threadpool_sys::io::ThreadpoolIo): the ring must
/// run down its outstanding operations before closing, and a group's bulk
/// `CloseThreadpoolCleanupGroupMembers` has no way to run that logic for a
/// member it did not create itself. `EventDelivery` stays individually
/// owned, where its own field-drop order (below) gives the same
/// quiesce-then-close guarantee a group would otherwise provide.
pub struct EventDelivery<T = (), X = ()> {
    // Drop order matters and is why these fields are declared in this order:
    // Rust drops struct fields top-to-bottom. `wake` must go first -- its own
    // `Drop` disarms, suppresses re-arming, and drains any in-flight callback
    // before releasing its captured `Arc<Mutex<IoRing>>` clone -- so that by
    // the time `ring`'s last reference drops below and runs
    // `IoRing::run_down` then `CloseIoRing`, no callback can still be
    // touching it (M4.3).
    #[expect(
        dead_code,
        reason = "held for its Drop, which stops the wait and drains it before the ring is released"
    )]
    wake: CompletionWake,
    ring: Arc<Mutex<IoRing<T, X>>>,
}

impl<T: Send + 'static, X: Send + 'static> EventDelivery<T, X> {
    /// Wire `ring`'s completion event to a thread-pool wait, delivering every
    /// popped [`Completion`] to `on_completion` on a pool thread (M4.2).
    ///
    /// The wait is armed before this returns, so the calling thread never
    /// waits for a completion itself (M4.4).
    ///
    /// # Completions already queued when `ring` is handed over
    ///
    /// They are delivered too -- but that is a guarantee this method has to
    /// buy, not one it inherits, and saying so is the point of this section.
    /// The ring's event is edge-triggered on the completion queue going empty
    /// to non-empty ([`IoRing::completion_event`], D-19), so attaching to a
    /// ring whose queue is *already* non-empty signals nothing, and no later
    /// completion signals either, because the queue never returns to empty to
    /// re-arm the edge. Such a backlog is stranded permanently.
    ///
    /// What closes that gap is a deliberate signal raised on the event once
    /// it has been attached: the first callback then drains the backlog
    /// exactly as it would any other wakeup.
    ///
    /// That signal is raised *after* the wait has been armed, which is the
    /// order `SetThreadpoolWait` documents -- "you must re-register the event
    /// with the wait object before signaling it each time to trigger the wait
    /// callback". The ordering binds to that documented rule, not to any
    /// observed tolerance for the other order. This method therefore attaches
    /// the event unsignalled and raises the signal itself, rather than going
    /// through [`IoRing::completion_event`], which raises the signal before it
    /// returns and so leaves a caller no way to arm in between.
    ///
    /// This was false in the implementation, and asserted anyway in this
    /// rustdoc, before M11.3 -- every test until then handed over a fresh
    /// ring, so nothing contradicted it. A caller on an earlier version
    /// cannot rely on the guarantee; `tests/event_delivery.rs` keeps the
    /// repro that now holds it. The ordering above was wrong until M26.9, in
    /// a way that stranded the backlog in roughly one run in a hundred. It
    /// was wrong again until `M26.12`, which raised the signal only when this
    /// method had itself attached the event: a caller that attached earlier
    /// and consumed that signal lost its whole backlog, every time rather
    /// than rarely.
    ///
    /// # What `on_completion` may rely on
    ///
    /// Two properties of the callback are part of this method's contract
    /// ([D-83](../DESIGN-NOTES.md#d-83)), so a consumer may build on them:
    ///
    /// - **It runs with the ring unlocked.** Each completion is popped under
    ///   the ring's lock and handed to `on_completion` after the lock is
    ///   released, so the callback may call [`EventDelivery::scope`] and
    ///   submit without deadlocking. This method hands the callback nothing
    ///   to reach its `EventDelivery` by; the caller arranges that, with a
    ///   `Weak` set after construction, for instance.
    /// - **Invocations may overlap.** Two calls may run at once on different
    ///   pool threads, and completions handed to concurrent calls carry no
    ///   order relative to each other. A consumer that needs an order imposes
    ///   it. The overlap comes from the delivery loop's shape -- drain, re-arm
    ///   the wait, drain again -- since a completion arriving after the
    ///   re-arm can start a second call while the first is still draining.
    ///
    /// Neither makes the ring a synchronization provider: its lock is its
    /// own, and a callback holds it only for as long as a scope it opens.
    ///
    /// # Errors
    ///
    /// Returns [`io::ErrorKind::Unsupported`] if the running system does not
    /// report `IORING_FEATURE_SET_COMPLETION_EVENT` (M4.1), which
    /// [`IoRing::completion_event`] is what decides. This crate refuses to
    /// silently substitute a thread-based polling loop instead -- a caller
    /// who asked for event-driven delivery and got a spun-up thread has been
    /// told something false. Also returns any other error from attaching the
    /// ring's completion event, from `ThreadpoolWait::new`, or from raising
    /// the setup signal.
    pub fn new<F>(
        ring: IoRing<T, X>,
        on_completion: F,
        env: Option<&mut CallbackEnviron<'_>>,
    ) -> io::Result<Self>
    where
        F: Fn(Completion, Option<(Option<T>, X)>) + Send + Sync + 'static,
    {
        // The protocol -- attach the event unsignalled, arm the wait, raise the
        // setup signal; drain, re-arm, drain -- is `CompletionWake`'s, written
        // once there (D-84). This adds only the drain it calls, over a ring this
        // delivery owns.
        let on_completion: Arc<OnCompletion<T, X>> = Arc::new(on_completion);
        let ring = Arc::new(Mutex::new(ring));
        let ring_for_wake = Arc::clone(&ring);
        // Built under the ring's lock, which the setup signal needs; a callback
        // the signal starts waits for the lock until this returns.
        let wake = CompletionWake::new(
            &mut *ring
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner),
            move || drain(&ring_for_wake, on_completion.as_ref()),
            env,
        )?;
        Ok(Self { wake, ring })
    }

    /// Lock the wrapped ring and return a scope for submitting work to it.
    ///
    /// A caller submits by opening a [`RingScope::batch`], the same ring the
    /// wait callback locks to pop completions. Mutex poisoning is absorbed
    /// here rather than surfaced: a panic elsewhere does not invalidate a ring
    /// handle, and the callback's own drain already takes the same view, so
    /// making every caller write `unwrap_or_else(PoisonError::into_inner)` only
    /// invited the inconsistency this method removes.
    ///
    /// # What the scope deliberately withholds
    ///
    /// The rule is: **every read-only part of [`IoRing`], plus batch
    /// construction -- and nothing that can retarget the ring or steal the
    /// pool's completions.** So there is no `try_pop` (D-21 makes the pool the
    /// single drainer; a second one breaks the drain-to-empty that re-arms the
    /// edge), no `completion_event` (it hands back a duplicate of the event
    /// this `EventDelivery` already waits on, giving two waiters on one ring),
    /// no `run_down`, and above all **no `&mut IoRing`**.
    ///
    /// That last one is the point, and it is why this returns a scope rather
    /// than the `&Mutex<IoRing>` it used to. Any `&mut IoRing` permits
    /// whole-value assignment, so safe code could replace the ring while the
    /// pool's wait stayed armed on the *original* ring's event -- measured, and
    /// delivery stopped silently ([D-43](../DESIGN-NOTES.md#d-43)). Note a
    /// `Deref`/`DerefMut` newtype would not have closed that, since
    /// `*scope = ...` works through `DerefMut` just as well; nor would handing
    /// a `&mut IoRing` to a closure.
    ///
    /// Replacing the ring is therefore refused at compile time. The expected
    /// error is pinned, though rustdoc enforces a pinned code only on a
    /// nightly toolchain or under `RUSTC_BOOTSTRAP=1`, which CI's
    /// `doctest-error-codes` job sets; on plain stable, the compiling twin
    /// below is what keeps this example honest:
    ///
    /// ```compile_fail,E0614
    /// # use windows_ioring_sys::{EventDelivery, IoRing};
    /// let delivery =
    ///     EventDelivery::new(IoRing::new(8, 8).unwrap(), |_, _| {}, None).unwrap();
    /// let mut scope = delivery.scope();
    /// // No `DerefMut`, so there is no `&mut IoRing` to assign through.
    /// *scope = IoRing::new(8, 8).unwrap();
    /// ```
    ///
    /// The same setup doing the legitimate thing must still compile. A
    /// `compile_fail` example passes on *any* error, including a typo in its
    /// own setup, so this pair is what keeps the one above honest:
    ///
    /// ```
    /// # use windows_ioring_sys::{EventDelivery, IoRing};
    /// let delivery =
    ///     EventDelivery::new(IoRing::new(8, 8).unwrap(), |_, _| {}, None).unwrap();
    /// let scope = delivery.scope();
    /// assert_eq!(scope.outstanding(), 0);
    /// ```
    #[must_use]
    pub fn scope(&self) -> RingScope<'_, T, X> {
        RingScope {
            ring: self
                .ring
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner),
        }
    }
}

/// Exclusive access to an [`EventDelivery`]'s ring, narrowed to what a
/// submitting caller needs (M18.6).
///
/// Held for as long as the value lives, so keep it to the shortest scope that
/// covers a submission: the wait callback needs the same lock to deliver
/// completions.
///
/// See [`EventDelivery::scope`] for what this deliberately does not expose,
/// and why handing out anything that yields a `&mut IoRing` would reopen
/// [D-43](../DESIGN-NOTES.md#d-43).
pub struct RingScope<'delivery, T = (), X = ()> {
    ring: MutexGuard<'delivery, IoRing<T, X>>,
}

impl<T, X> RingScope<'_, T, X> {
    /// Open a [`Batch`] against the ring.
    ///
    /// The borrow is confined to the returned batch, so no `&mut IoRing`
    /// escapes to the caller.
    pub fn batch(&mut self) -> Batch<'_, T, X> {
        Batch::new(&mut self.ring)
    }

    /// Operations submitted but not yet popped, as [`IoRing::outstanding`].
    #[must_use]
    pub fn outstanding(&self) -> usize {
        self.ring.outstanding()
    }

    /// This ring's negotiated version, as [`IoRing::version`].
    #[must_use]
    pub fn version(&self) -> RingVersion {
        self.ring.version()
    }

    /// Query the ring, as [`IoRing::info`].
    ///
    /// # Errors
    ///
    /// As [`IoRing::info`].
    pub fn info(&self) -> io::Result<RingInfo> {
        self.ring.info()
    }

    /// Whether this ring supports `op`, as [`IoRing::supports`].
    #[must_use]
    pub fn supports(&self, op: Op) -> bool {
        self.ring.supports(op)
    }

    /// Whether this ring supports `op_code`, as [`IoRing::supports_raw`].
    #[must_use]
    pub fn supports_raw(&self, op_code: IORING_OP_CODE) -> bool {
        self.ring.supports_raw(op_code)
    }

    /// Registered file count, as [`IoRing::registered_file_count`].
    #[must_use]
    pub fn registered_file_count(&self) -> u32 {
        self.ring.registered_file_count()
    }

    /// Registered buffer count, as [`IoRing::registered_buffer_count`].
    #[must_use]
    pub fn registered_buffer_count(&self) -> u32 {
        self.ring.registered_buffer_count()
    }
}

#[cfg(test)]
mod tests;