windows-overlapped-io-sys 0.1.0

// Copyright (c) 2026 Mike Grier
//! Raw I/O completion port backend: port ownership, association, and dequeue.
//!
//! A [`CompletionPort`] owns a completion-port handle and can service many
//! endpoints, each associated with a caller-chosen completion key. Association
//! is the consuming transition that binds an [`UnassociatedEndpoint`] to this
//! backend. The port does not create worker threads; the owner decides where
//! [`CompletionPort::get`] runs. Submission of real overlapped operations, and
//! the reclamation that follows their completion, are built on top of this
//! module.

use std::cell::Cell;
use std::collections::HashMap;
use std::fmt;
use std::io;
use std::os::windows::io::{AsHandle, AsRawHandle, BorrowedHandle, FromRawHandle, OwnedHandle};
use std::panic::Location;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, MutexGuard};

use windows_sys::Win32::Foundation::{HANDLE, INVALID_HANDLE_VALUE, WAIT_TIMEOUT};
use windows_sys::Win32::System::IO::{
    CancelIoEx, CreateIoCompletionPort, GetQueuedCompletionStatus, OVERLAPPED,
    PostQueuedCompletionStatus,
};

use crate::{Operation, OperationState, UnassociatedEndpoint};

/// Waits without timeout in `GetQueuedCompletionStatus`; used while draining.
const INFINITE: u32 = u32::MAX;

/// Optional per-operation source information, recorded only while source
/// tracking is enabled.
struct Track {
    location: &'static Location<'static>,
    #[cfg(feature = "operation-backtrace")]
    backtrace: std::backtrace::Backtrace,
}

/// State shared between a port, its completions, and the drain path.
///
/// `outstanding` is a lock-free count that always governs rundown. `tracked` is
/// consulted only when source tracking is enabled, so the mutex is never taken
/// on the submission hot path by default.
struct PortState {
    outstanding: AtomicUsize,
    tracked: Mutex<HashMap<usize, Track>>,
}

impl PortState {
    fn new() -> Self {
        Self {
            outstanding: AtomicUsize::new(0),
            tracked: Mutex::new(HashMap::new()),
        }
    }
}

/// Lock a mutex, recovering the guard even if a previous holder panicked.
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    mutex.lock().unwrap_or_else(|poison| poison.into_inner())
}

/// An owned I/O completion port.
pub struct CompletionPort {
    handle: OwnedHandle,
    state: Arc<PortState>,
}

impl CompletionPort {
    /// Create a new completion port.
    ///
    /// `concurrency` is the maximum number of threads the system lets run
    /// completions for this port concurrently; zero means one per processor.
    pub fn new(concurrency: u32) -> io::Result<Self> {
        // SAFETY: creating a fresh port with no associated file handle.
        let handle = unsafe {
            CreateIoCompletionPort(INVALID_HANDLE_VALUE, std::ptr::null_mut(), 0, concurrency)
        };
        if handle.is_null() {
            return Err(io::Error::last_os_error());
        }
        // SAFETY: the call returned a fresh, exclusively owned port handle.
        let handle = unsafe { OwnedHandle::from_raw_handle(handle) };
        Ok(Self {
            handle,
            state: Arc::new(PortState::new()),
        })
    }

    /// Associate an overlapped endpoint with this port under `key`.
    ///
    /// Completions for operations issued on the endpoint are delivered to this
    /// port and tagged with `key`. The association is permanent for the life of
    /// the handle, so the returned endpoint borrows the port.
    pub fn associate(
        &self,
        endpoint: UnassociatedEndpoint,
        key: usize,
    ) -> io::Result<AssociatedEndpoint<'_>> {
        let handle = endpoint.into_handle();
        // SAFETY: associating a valid handle with a valid port; the concurrency
        // argument is ignored when an existing port is supplied.
        let result = unsafe { CreateIoCompletionPort(handle.as_raw_handle(), self.raw(), key, 0) };
        if result.is_null() {
            return Err(io::Error::last_os_error());
        }
        Ok(AssociatedEndpoint {
            port: self,
            handle,
            key,
        })
    }

    /// Post a user-defined wakeup packet to this port.
    ///
    /// The packet carries `key` and `bytes_transferred` with a null `OVERLAPPED`,
    /// which keeps it distinguishable from operation completions; identify it by
    /// its `key`.
    pub fn post(&self, key: usize, bytes_transferred: u32) -> io::Result<()> {
        // SAFETY: the port handle is valid; a null overlapped marks a user packet.
        let ok = unsafe {
            PostQueuedCompletionStatus(self.raw(), bytes_transferred, key, std::ptr::null())
        };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn post_raw(
        &self,
        key: usize,
        bytes_transferred: u32,
        overlapped: *mut OVERLAPPED,
    ) -> io::Result<()> {
        // SAFETY: tests use this to simulate an operation completion for a live
        // operation's OVERLAPPED pointer.
        let ok = unsafe {
            PostQueuedCompletionStatus(self.raw(), bytes_transferred, key, overlapped.cast_const())
        };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(())
    }

    /// Dequeue one completion packet, waiting up to `timeout_ms` milliseconds.
    ///
    /// Returns `Ok(None)` when the wait times out with no packet. A packet is
    /// returned even when its operation failed; the failure is reported through
    /// [`Completion::error`].
    pub fn get(&self, timeout_ms: u32) -> io::Result<Option<Completion>> {
        let mut bytes_transferred: u32 = 0;
        let mut key: usize = 0;
        let mut overlapped: *mut OVERLAPPED = std::ptr::null_mut();
        // SAFETY: all out-parameters are valid for the duration of the call.
        let ok = unsafe {
            GetQueuedCompletionStatus(
                self.raw(),
                &mut bytes_transferred,
                &mut key,
                &mut overlapped,
                timeout_ms,
            )
        };
        if ok != 0 {
            return Ok(Some(Completion {
                key,
                bytes_transferred,
                overlapped,
                error: None,
                state: Arc::clone(&self.state),
                claimed: Cell::new(false),
            }));
        }

        let error = io::Error::last_os_error();
        if overlapped.is_null() {
            if error.raw_os_error() == Some(WAIT_TIMEOUT as i32) {
                return Ok(None);
            }
            return Err(error);
        }
        // A packet for a failed operation was dequeued.
        Ok(Some(Completion {
            key,
            bytes_transferred,
            overlapped,
            error: Some(error),
            state: Arc::clone(&self.state),
            claimed: Cell::new(false),
        }))
    }

    fn raw(&self) -> HANDLE {
        self.handle.as_raw_handle()
    }

    /// The number of operations submitted through this port that have not yet
    /// been claimed, reclaimed, or drained.
    #[must_use]
    pub fn outstanding(&self) -> usize {
        self.state.outstanding.load(Ordering::SeqCst)
    }

    /// Block until every outstanding operation has completed and been reclaimed.
    ///
    /// Every outstanding operation must already be cancelled or otherwise
    /// destined to complete -- which closing or cancelling the endpoints
    /// guarantees -- or this waits indefinitely. Each dequeued completion
    /// reclaims its own operation when dropped, so draining is just dequeuing
    /// until the count reaches zero.
    pub fn run_down(&self) -> io::Result<()> {
        while self.outstanding() > 0 {
            self.get(INFINITE)?;
        }
        Ok(())
    }

    fn report_outstanding_at_drop(&self, count: usize) {
        let tracked = lock(&self.state.tracked);
        let mut message = format!(
            "windows-overlapped-io-sys: CompletionPort dropped with {count} operation(s) still \
             outstanding; call run_down() before dropping to control when this blocks."
        );
        if tracked.is_empty() {
            message.push_str(
                " Enable source tracking (WINDOWS_OVERLAPPED_IO_SYS_TRACK=1, or \
                 set_source_tracking) to identify the submit sites.",
            );
        } else {
            message.push_str(" Sources:");
            for track in tracked.values() {
                message.push_str("\n  - ");
                message.push_str(&track.location.to_string());
                #[cfg(feature = "operation-backtrace")]
                {
                    message.push_str("\n    backtrace:\n");
                    message.push_str(&track.backtrace.to_string());
                }
            }
        }
        eprintln!("{message}");
    }
}

impl fmt::Debug for CompletionPort {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CompletionPort")
            .field("outstanding", &self.outstanding())
            .finish_non_exhaustive()
    }
}

impl Drop for CompletionPort {
    fn drop(&mut self) {
        let count = self.outstanding();
        if count == 0 {
            return;
        }
        // A blocking Drop signals that run_down() was skipped; name the sources.
        self.report_outstanding_at_drop(count);
        // Block until the kernel is done with every operation's storage.
        let _ = self.run_down();
    }
}

/// An overlapped endpoint bound to exactly one [`CompletionPort`].
///
/// The endpoint owns its handle and borrows the port it is associated with, so
/// the port cannot be dropped while any endpoint still routes completions to it.
/// It is intentionally not `Clone`.
#[derive(Debug)]
pub struct AssociatedEndpoint<'port> {
    port: &'port CompletionPort,
    handle: OwnedHandle,
    key: usize,
}

impl<'port> AssociatedEndpoint<'port> {
    /// Borrow the underlying handle for issuing native operations.
    #[must_use]
    pub fn handle(&self) -> BorrowedHandle<'_> {
        self.handle.as_handle()
    }

    /// The completion key packets from this endpoint are tagged with.
    #[must_use]
    pub fn key(&self) -> usize {
        self.key
    }

    /// The completion port this endpoint is associated with.
    #[must_use]
    pub fn port(&self) -> &'port CompletionPort {
        self.port
    }

    /// Submit an owned operation on this endpoint.
    ///
    /// `issue` performs the single native overlapped call using the endpoint's
    /// handle and the operation's stable `OVERLAPPED` pointer. It returns `Ok`
    /// when a completion will arrive (native success or `ERROR_IO_PENDING`) and
    /// `Err` for an immediate failure that yields no completion.
    ///
    /// On the completion path the operation's storage is transferred to the
    /// kernel and recovered later with [`Completion::claim`]. On the failure
    /// path the operation is returned intact so its storage can be reused.
    ///
    /// # Safety
    ///
    /// `issue` must start exactly one overlapped operation using the provided
    /// `OVERLAPPED` pointer and no other storage, and must classify the outcome
    /// correctly: `Ok` only when a completion packet will be delivered to this
    /// endpoint's port, `Err` only when none will.
    #[track_caller]
    pub unsafe fn submit<P, F>(&self, operation: Operation<P>, issue: F) -> Submitted<P>
    where
        P: Send,
        F: FnOnce(BorrowedHandle<'_>, *mut OVERLAPPED) -> io::Result<()>,
    {
        // Transfer the operation's storage out; the caller (kernel) owns it until
        // it is reclaimed. `into_overlapped` arms the type-erased reclaim thunk.
        let overlapped = operation.into_overlapped();
        let identity = overlapped as usize;

        // Count before issuing so a completion cannot race ahead of the count.
        let state = &self.port.state;
        state.outstanding.fetch_add(1, Ordering::SeqCst);
        let tracking = crate::source_tracking_enabled();
        if tracking {
            lock(&state.tracked).insert(
                identity,
                Track {
                    location: Location::caller(),
                    #[cfg(feature = "operation-backtrace")]
                    backtrace: std::backtrace::Backtrace::capture(),
                },
            );
        }

        match issue(self.handle(), overlapped) {
            Ok(()) => Submitted::Pending(OperationId(overlapped)),
            Err(error) => {
                state.outstanding.fetch_sub(1, Ordering::SeqCst);
                if tracking {
                    lock(&state.tracked).remove(&identity);
                }
                // SAFETY: no completion will arrive, so reclaim the operation we
                // just leaked, exactly once.
                let mut operation = unsafe { Operation::<P>::from_overlapped(overlapped) };
                operation.set_state(OperationState::Idle);
                Submitted::Failed { operation, error }
            }
        }
    }

    /// Request cancellation of a single outstanding operation.
    ///
    /// Cancellation is only a request: the operation still completes, typically
    /// with `ERROR_OPERATION_ABORTED`, and that completion remains the point at
    /// which its storage is reclaimed with [`Completion::claim`].
    pub fn cancel(&self, id: OperationId) -> io::Result<()> {
        // SAFETY: cancelling by a valid handle and an OVERLAPPED identity.
        let ok = unsafe { CancelIoEx(self.raw_handle(), id.as_ptr()) };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(())
    }

    /// Request cancellation of every outstanding operation on this endpoint.
    pub fn cancel_all(&self) -> io::Result<()> {
        // SAFETY: a null OVERLAPPED cancels all operations on the handle.
        let ok = unsafe { CancelIoEx(self.raw_handle(), std::ptr::null()) };
        if ok == 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(())
    }

    fn raw_handle(&self) -> HANDLE {
        self.handle.as_raw_handle()
    }
}

/// An identity for an in-flight operation: the address of its `OVERLAPPED`.
///
/// It is used to cancel a specific operation and to match its later completion.
/// The pointer must not be dereferenced or freed; the kernel owns the storage
/// until the completion is claimed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OperationId(*mut OVERLAPPED);

impl OperationId {
    /// The `OVERLAPPED` pointer this identity refers to.
    #[must_use]
    pub fn as_ptr(self) -> *mut OVERLAPPED {
        self.0
    }
}

/// The outcome of [`AssociatedEndpoint::submit`].
#[derive(Debug)]
pub enum Submitted<P> {
    /// A completion will arrive; the storage was transferred to the kernel and
    /// is recovered later with [`Completion::claim`]. The [`OperationId`]
    /// identifies the in-flight operation for cancellation and matching.
    Pending(OperationId),
    /// Submission failed immediately with no completion; the operation is
    /// returned so its storage can be reused or dropped.
    Failed {
        /// The operation whose submission failed.
        operation: Operation<P>,
        /// The immediate failure reported by the native call.
        error: io::Error,
    },
}

/// A completion packet dequeued from a [`CompletionPort`].
pub struct Completion {
    key: usize,
    bytes_transferred: u32,
    overlapped: *mut OVERLAPPED,
    error: Option<io::Error>,
    state: Arc<PortState>,
    claimed: Cell<bool>,
}

impl fmt::Debug for Completion {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Completion")
            .field("key", &self.key)
            .field("bytes_transferred", &self.bytes_transferred)
            .field("overlapped", &self.overlapped)
            .field("error", &self.error)
            .finish_non_exhaustive()
    }
}

impl Drop for Completion {
    fn drop(&mut self) {
        // Claimed completions handed ownership to the caller; user packets carry
        // a null overlapped and own nothing.
        if self.claimed.get() || self.overlapped.is_null() {
            return;
        }
        // An operation completion observed but never claimed: reclaim it so the
        // port's rundown can finish.
        self.state.outstanding.fetch_sub(1, Ordering::SeqCst);
        if crate::source_tracking_enabled() {
            lock(&self.state.tracked).remove(&(self.overlapped as usize));
        }
        // SAFETY: the completion arrived, so the kernel is done with the storage;
        // the operation's armed reclaim thunk frees the box exactly once.
        unsafe { crate::operation::reclaim_from_overlapped(self.overlapped) };
    }
}

impl Completion {
    /// The completion key the packet was tagged with.
    #[must_use]
    pub fn key(&self) -> usize {
        self.key
    }

    /// The number of bytes transferred by the operation.
    #[must_use]
    pub fn bytes_transferred(&self) -> u32 {
        self.bytes_transferred
    }

    /// The `OVERLAPPED` pointer identifying the completed operation.
    ///
    /// For a user packet this is whatever value was passed to
    /// [`CompletionPort::post`].
    #[must_use]
    pub fn overlapped_ptr(&self) -> *mut OVERLAPPED {
        self.overlapped
    }

    /// The failure of the completed operation, if it did not succeed.
    #[must_use]
    pub fn error(&self) -> Option<&io::Error> {
        self.error.as_ref()
    }

    /// Recover the owned operation whose completion this is.
    ///
    /// # Safety
    ///
    /// This completion must have been produced by submitting an `Operation<P>`
    /// of this exact type through [`AssociatedEndpoint::submit`], and it must be
    /// claimed exactly once.
    pub unsafe fn claim<P>(&self) -> Operation<P> {
        // Mark claimed so this completion's own drop will not also reclaim it.
        self.claimed.set(true);
        self.state.outstanding.fetch_sub(1, Ordering::SeqCst);
        if crate::source_tracking_enabled() {
            lock(&self.state.tracked).remove(&(self.overlapped as usize));
        }
        // SAFETY: by this function's contract, the identity is a matching leaked
        // Operation<P>, reclaimed exactly once here.
        let mut operation = unsafe { Operation::<P>::from_overlapped(self.overlapped) };
        operation.set_state(OperationState::Completed);
        operation
    }
}

#[cfg(test)]
mod tests;