Struct ThreadpoolIo 

pub struct ThreadpoolIo { /* private fields */ }

An owned thread-pool I/O object bound to one overlapped endpoint.

Creating a ThreadpoolIo consumes an UnassociatedEndpoint, which is the same one-time, consuming association the other backends use: CreateThreadpoolIo binds the handle to the pool’s internal completion port, and no second association is possible afterward.

Every ThreadpoolIo::submit is paired with exactly one balancing action, so ThreadpoolIo::outstanding is always the number of operations whose storage the kernel or pool still owns. Drop never frees that storage early: it cancels what is outstanding, waits for the resulting callbacks, waits for any callback still executing, and only then releases the object, the handle, and the callback context.

Examples

Read a file with one overlapped operation. Submission is unsafe because only the caller can guarantee the native call it issues matches the operation it was handed; everything after that is safe.

use std::io;
use std::os::windows::io::AsRawHandle;
use std::ptr;
use std::sync::mpsc;
use windows_overlapped_io_sys::{Issued, Operation, Submitted, UnassociatedEndpoint};
use windows_sys::Win32::Foundation::ERROR_IO_PENDING;
use windows_sys::Win32::Storage::FileSystem::ReadFile;
use windows_threadpool_sys::io::{IoCompletion, ThreadpoolIo};

let path = std::env::temp_dir().join(format!("wtps-doc-{}.tmp", std::process::id()));
std::fs::write(&path, b"overlapped hello")?;

let endpoint = UnassociatedEndpoint::open(&path, true, false, 0)?;
let (tx, rx) = mpsc::channel();
let sender = std::sync::Mutex::new(tx);

let tp = ThreadpoolIo::new(endpoint, move |completion: &IoCompletion| {
    // SAFETY: this object only ever carries `Operation<()>`, submitted
    // below, and each completion is claimed exactly once.
    let _operation = unsafe { completion.claim::<()>() };
    let _ = sender.lock().expect("send").send(completion.bytes_transferred());
}, None)?;

let mut buffer = [0_u8; 64];
let buf_ptr = buffer.as_mut_ptr();
let buf_len = buffer.len() as u32;

let mut operation = Operation::new(());
operation.set_offset(0);

// SAFETY: issues exactly one overlapped ReadFile into `buffer`, which stays
// alive until the completion is received below. The handle is not in
// skip-on-success mode, so both synchronous success and ERROR_IO_PENDING
// deliver a completion callback.
let submitted = unsafe {
    tp.submit(operation, |handle, overlapped| {
        let ok = ReadFile(handle.as_raw_handle(), buf_ptr, buf_len, ptr::null_mut(), overlapped);
        if ok != 0 {
            return Ok(Issued::Pending);
        }
        let error = io::Error::last_os_error();
        if error.raw_os_error() == Some(ERROR_IO_PENDING as i32) {
            return Ok(Issued::Pending);
        }
        Err(error)
    })
};
assert!(matches!(submitted, Submitted::Pending(_)));

let read = rx.recv().expect("a completion");
tp.run_down();
assert_eq!(&buffer[..read], b"overlapped hello");

drop(tp);
let _ = std::fs::remove_file(&path);

Implementations

impl ThreadpoolIo

pub fn new<F>( endpoint: UnassociatedEndpoint, callback: F, env: Option<&mut CallbackEnviron<'_>>, ) -> Result<Self>

where F: Fn(&IoCompletion) + Send + Sync + 'static,

Bind an overlapped endpoint to the thread pool, invoking callback for every operation completion the pool delivers.

Pass Some(env) to select a private pool or callback priority; None uses the process-default pool with default priority.

Do not point env at a cleanup group. A TP_IO object must not be closed while an overlapped operation is outstanding, because the kernel still owns that operation’s storage, and a group’s bulk release has no way to establish that – which is why CleanupGroup has no create_io. A group would also close this object while its own Drop still expects to, closing it twice. Let Drop run it down instead: it cancels, drains, and only then closes.

The callback runs on a shared, process-managed pool thread. It must restore any thread state it changes, must not terminate its thread, and must not block waiting on this object’s rundown. It must not panic: a panic unwinds to the extern "system" trampoline and aborts the process.

Errors

Returns the error from CreateThreadpoolIo, most commonly when the handle was not opened for overlapped I/O.

pub fn handle(&self) -> BorrowedHandle<'_>

Borrow the underlying handle for issuing native operations.

pub fn outstanding(&self) -> usize

The number of submitted operations whose completion callback has not yet started.

A TP_IO callback deregisters its operation on entry – before it can claim or drop the storage – so this counts operations still awaiting their callback, not live allocations: it can read zero while a final callback is still running and its operation’s storage is still alive. Use run_down to wait for callbacks to finish. It is equivalently the number of StartThreadpoolIo calls not yet balanced by a callback start or a CancelThreadpoolIo.

pub unsafe fn submit<P, F>( &self, operation: Operation<P>, issue: F, ) -> Submitted<P>

where P: Send + 'static, F: FnOnce(BorrowedHandle<'_>, *mut OVERLAPPED) -> Result<Issued>,

Submit an owned operation on this endpoint.

StartThreadpoolIo is issued before issue runs, as the SDK requires, and is balanced exactly once: by the I/O callback on the Issued::Pending path, or by CancelThreadpoolIo here on the synchronous-completion and immediate-failure paths.

issue performs the single native overlapped call using the endpoint’s handle and the operation’s stable OVERLAPPED pointer, and classifies the outcome as an Issued: Issued::Pending when the pool will deliver a completion callback, or Issued::Completed when the call finished synchronously and no callback will arrive – the outcome a handle in FILE_SKIP_COMPLETION_PORT_ON_SUCCESS mode reports on synchronous success. It returns Err for an immediate failure that yields no callback.

On the pending path the operation’s storage is transferred out and is recovered later inside the callback with IoCompletion::claim. On the synchronous and failure paths the operation is returned intact through Submitted so its storage can be reused or inspected.

Panics

Panics if this object already has a live operation registered at the new operation’s storage address. That cannot happen through ordinary use – operation owns freshly boxed storage – and indicates a defect in this crate’s own bookkeeping rather than in the calling code. See OperationRegistry::insert in windows-overlapped-io-sys for the invariant involved.

Safety

issue must start exactly one overlapped operation using the provided OVERLAPPED pointer and no other storage, and must classify the outcome correctly: Issued::Pending only when a completion callback will be delivered for this object, Issued::Completed only when the operation is already complete and no callback will arrive, and Err only when the submission failed and no callback will arrive. Misclassifying either unbalances the pool’s accounting or frees storage the kernel still owns.

issue must not unwind. StartThreadpoolIo is issued before it runs and is balanced only on the paths below; a panic out of issue skips that balancing while leaving the start pending, and – because a panic before starting the I/O is indistinguishable from one after – rundown could then wait forever for a callback that will never arrive. A closure that might panic must catch it and return Err instead.

P: 'static because submitting leaks the operation’s storage, to be freed later through a thunk carrying no lifetime – see Operation::into_overlapped.

pub fn cancel(&self, id: OperationId) -> Result<()>

Request cancellation of a single outstanding operation.

Cancellation is only a request: the operation still completes, typically with ERROR_OPERATION_ABORTED, and that completion callback remains the point at which its storage is reclaimed.

The identity is checked against this object’s live operations first. An identity whose operation has already completed is rejected with io::ErrorKind::NotFound and no native call is made, even if another operation has since been given the same storage address. Cancelling therefore races safely against completion: the worst outcome of a late cancel is this error, never the cancellation of an unrelated operation.

Errors

Returns io::ErrorKind::NotFound if id no longer names a live operation, or the error from CancelIoEx if the native request fails.

pub fn cancel_all(&self) -> Result<()>

Request cancellation of every outstanding operation on this endpoint.

Errors

Returns the error from CancelIoEx, which reports ERROR_NOT_FOUND when nothing was outstanding.

pub fn run_down(&self)

Block until every outstanding operation has completed and every completion callback has finished running.

Every outstanding operation must already be cancelled or otherwise destined to complete – which ThreadpoolIo::cancel_all guarantees – or this waits indefinitely.

Two things are waited for, because an operation is deregistered when its callback is entered rather than when that callback returns: first that no operation is outstanding, then that no callback is still executing. Together they mean a caller can read whatever its callbacks recorded as soon as this returns.

Must not be called from inside this object’s own callback, which would wait on the callback’s own completion.

pub fn wait(&self)

Block until no I/O callback for this object is executing.

This waits for callbacks that have already started; it does not cancel pending ones. There is deliberately no cancelling variant: cancelling a pending I/O callback would neither cancel the underlying operation nor make its OVERLAPPED storage safe to free, so the only sound way to stop outstanding I/O is ThreadpoolIo::cancel_all followed by ThreadpoolIo::run_down.

Trait Implementations

impl Debug for ThreadpoolIo

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for ThreadpoolIo

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.

impl Send for ThreadpoolIo

impl Sync for ThreadpoolIo