use std::collections::HashMap;
use std::marker::PhantomData;
use std::mem::ManuallyDrop;
use crate::token::OperationId;
use std::ffi::c_void;
use std::io;
use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle};
use std::time::{Duration, Instant};
use windows_sys::Win32::Storage::FileSystem::{
CloseIoRing, CreateIoRing, GetIoRingInfo, IORING_BUFFER_INFO, IORING_CQE,
IORING_CREATE_ADVISORY_FLAGS_NONE, IORING_CREATE_FLAGS, IORING_CREATE_REQUIRED_FLAGS_NONE,
IORING_INFO, IORING_OP_CANCEL, IORING_OP_CODE, IORING_OP_FLUSH, IORING_OP_NOP, IORING_OP_READ,
IORING_OP_REGISTER_BUFFERS, IORING_OP_REGISTER_FILES, IORING_OP_WRITE, IsIoRingOpSupported,
};
use windows_sys::Win32::System::Threading::{CreateEventW, SetEvent};
use crate::accounting::Accounting;
use crate::capability::{RingVersion, capabilities};
use crate::error::check;
pub(crate) use crate::accounting::RingId;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum Op {
Nop,
Read,
Write,
Flush,
RegisterFiles,
RegisterBuffers,
Cancel,
}
impl Op {
const ALL: [Op; 7] = [
Op::Nop,
Op::Read,
Op::Write,
Op::Flush,
Op::RegisterFiles,
Op::RegisterBuffers,
Op::Cancel,
];
#[must_use]
pub fn code(self) -> IORING_OP_CODE {
match self {
Op::Nop => IORING_OP_NOP,
Op::Read => IORING_OP_READ,
Op::Write => IORING_OP_WRITE,
Op::Flush => IORING_OP_FLUSH,
Op::RegisterFiles => IORING_OP_REGISTER_FILES,
Op::RegisterBuffers => IORING_OP_REGISTER_BUFFERS,
Op::Cancel => IORING_OP_CANCEL,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
struct OpSupport(u8);
impl OpSupport {
unsafe fn probe(handle: *mut c_void) -> Self {
let mut mask = 0_u8;
for (index, op) in Op::ALL.iter().enumerate() {
let supported = unsafe { IsIoRingOpSupported(handle, op.code()) };
if supported != 0 {
mask |= 1 << index;
}
}
Self(mask)
}
fn contains(self, op: Op) -> bool {
let index = Op::ALL
.iter()
.position(|&candidate| candidate == op)
.expect("Op::ALL is exhaustive");
self.0 & (1 << index) != 0
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RingInfo {
pub version: RingVersion,
pub submission_queue_size: u32,
pub completion_queue_size: u32,
}
#[cfg(any(test, feature = "fault-injection"))]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum InjectedFailure {
Ring(crate::RingCondition),
Win32(u32),
Hresult(windows_sys::core::HRESULT),
}
#[cfg(any(test, feature = "fault-injection"))]
impl InjectedFailure {
pub(crate) fn as_hresult(self) -> windows_sys::core::HRESULT {
let code = match self {
Self::Ring(condition) => condition.code(),
Self::Win32(code) => (0x8007_0000_u32 | (code & 0xFFFF)).cast_signed(),
Self::Hresult(code) => code,
};
assert!(
code < 0,
"an injected failure must be a failing HRESULT, but {code:#010x} is not: \
this seam injects failure only, never success"
);
code
}
}
#[derive(Default)]
pub(crate) struct Held {
#[allow(
dead_code,
reason = "held so the file outlives the operation; dropped at reclaim"
)]
pub(crate) guard: Option<FileGuard>,
#[allow(dead_code, reason = "held so the registration outlives the operation")]
pub(crate) registration: Option<crate::batch::RegisteredUse>,
}
pub enum FileGuard {
Shared(win_shared_os_owned_handle::SharedHandle),
Registered(crate::batch::RegisteredFile),
}
impl From<win_shared_os_owned_handle::SharedHandle> for FileGuard {
fn from(guard: win_shared_os_owned_handle::SharedHandle) -> Self {
Self::Shared(guard)
}
}
impl From<crate::batch::RegisteredFile> for FileGuard {
fn from(guard: crate::batch::RegisteredFile) -> Self {
Self::Registered(guard)
}
}
pub type HeldCompletion<T, X> = (Completion, Option<(Option<T>, X)>);
pub(crate) struct Entry<T, X> {
pub(crate) payload: Option<T>,
pub(crate) extra: X,
#[expect(
dead_code,
reason = "read when the guarded pushes migrate in M28.4.1; see M28.3+M28.4 in CHECKLIST.md"
)]
pub(crate) held: Held,
}
#[derive(Clone, Copy, Debug)]
pub struct Completion {
user_data: usize,
result_code: windows_sys::core::HRESULT,
information: usize,
ring_id: RingId,
}
impl Completion {
#[must_use]
pub fn user_data(&self) -> usize {
self.user_data
}
#[must_use]
pub(crate) fn ring_id(&self) -> RingId {
self.ring_id
}
pub fn result(&self) -> io::Result<usize> {
check(self.result_code)?;
Ok(self.information)
}
#[cfg(any(test, feature = "fault-injection"))]
#[must_use]
pub fn with_injected_failure(self, failure: InjectedFailure) -> Self {
Self {
result_code: failure.as_hresult(),
information: 0,
..self
}
}
#[cfg(any(test, feature = "fault-injection"))]
#[must_use]
pub fn with_injected_transfer(self, transferred: usize) -> Self {
Self {
information: transferred,
..self
}
}
#[cfg(test)]
pub(crate) fn synthetic(
user_data: usize,
result_code: windows_sys::core::HRESULT,
ring_id: RingId,
) -> Self {
Self {
user_data,
result_code,
information: 0,
ring_id,
}
}
}
const S_FALSE: windows_sys::core::HRESULT = 1;
const RUN_DOWN_POLL_MS: u32 = 50;
pub(crate) const IORING_E_WAIT_TIMEOUT: windows_sys::core::HRESULT =
(0x8007_0000_u32 | windows_sys::Win32::Foundation::ERROR_TIMEOUT) as windows_sys::core::HRESULT;
const MAX_WAIT_MS: u32 = u32::MAX - 1;
pub(crate) fn every_entry_was_submitted(hr: windows_sys::core::HRESULT) -> bool {
hr == IORING_E_WAIT_TIMEOUT || check(hr).is_ok()
}
fn wait_outcome(hr: windows_sys::core::HRESULT) -> io::Result<()> {
if hr == IORING_E_WAIT_TIMEOUT {
return Ok(());
}
check(hr)
}
#[repr(C)]
pub struct IoRing<T = (), X = (), K = Open> {
handle: *mut c_void,
version: RingVersion,
supported_ops: OpSupport,
accounting: Accounting,
late_read: ManuallyDrop<LateReadArrays>,
registered_files: ManuallyDrop<Vec<win_shared_os_owned_handle::SharedHandle>>,
completion_event: Option<OwnedHandle>,
inventory: ManuallyDrop<HashMap<usize, Entry<T, X>>>,
kind: PhantomData<fn() -> K>,
}
#[derive(Default)]
pub(crate) struct LateReadArrays {
pub(crate) buffer_infos: Vec<IORING_BUFFER_INFO>,
pub(crate) file_handles: Vec<*mut c_void>,
}
impl<T, X, K> std::fmt::Debug for IoRing<T, X, K> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("IoRing")
.field("handle", &self.handle)
.field("version", &self.version)
.field("supported_ops", &self.supported_ops)
.field("accounting", &self.accounting)
.field("late_read_buffer_infos", &self.late_read.buffer_infos.len())
.field("late_read_file_handles", &self.late_read.file_handles.len())
.field("registered_files", &self.registered_files.len())
.field("completion_event", &self.completion_event)
.finish()
}
}
unsafe impl<T: Send, X: Send, K> Send for IoRing<T, X, K> {}
impl IoRing<()> {
pub fn new(submission_queue_size: u32, completion_queue_size: u32) -> io::Result<Self> {
let caps = capabilities()?;
let version = RingVersion::HIGHEST_KNOWN.min(caps.max_version);
Self::with_version(version, submission_queue_size, completion_queue_size)
}
pub fn with_version(
version: RingVersion,
submission_queue_size: u32,
completion_queue_size: u32,
) -> io::Result<Self> {
Self::with_version_and_inventory(version, submission_queue_size, completion_queue_size)
}
}
impl<T, X> IoRing<T, X, Open> {
pub fn with_version_and_inventory(
version: RingVersion,
submission_queue_size: u32,
completion_queue_size: u32,
) -> io::Result<Self> {
let flags = IORING_CREATE_FLAGS {
Required: IORING_CREATE_REQUIRED_FLAGS_NONE,
Advisory: IORING_CREATE_ADVISORY_FLAGS_NONE,
};
let mut handle: *mut c_void = std::ptr::null_mut();
let hr = unsafe {
CreateIoRing(
version.raw(),
flags,
submission_queue_size,
completion_queue_size,
&raw mut handle,
)
};
check(hr)?;
let supported_ops = unsafe { OpSupport::probe(handle) };
Ok(Self {
handle,
version,
supported_ops,
accounting: Accounting::new(),
late_read: ManuallyDrop::new(LateReadArrays::default()),
registered_files: ManuallyDrop::new(Vec::new()),
completion_event: None,
inventory: ManuallyDrop::new(HashMap::new()),
kind: PhantomData,
})
}
}
impl<T, X, K> IoRing<T, X, K> {
fn is_quiescent(&self) -> bool {
let quiescent = self.accounting.is_quiescent();
debug_assert!(
!quiescent || self.inventory.is_empty() || std::thread::panicking(),
"an owned entry outlived its identity's retirement"
);
quiescent
}
fn drain_for_rundown(&mut self) -> io::Result<()> {
while let Some(completion) = self.pop_raw()? {
drop(self.reclaim(completion.user_data()));
}
Ok(())
}
#[cold]
fn unminted_completion(&self, user_data: usize) {
#[cfg(feature = "threadpool")]
windows_threadpool_sys::trace_record!(
"ring",
"unminted-completion",
user_data,
self.accounting.outstanding()
);
if !std::thread::panicking() {
panic!(
"IoRing popped a completion for user_data {user_data:#x}, which is not in flight \
on this ring: it was never minted here, has already completed, or its build \
failed. Every queued SQE completes exactly once with the identity it was built \
with, so this is a defect in a push_raw closure, a kernel-seam responder, or \
the kernel (D-79)"
);
}
}
fn pop_raw(&mut self) -> io::Result<Option<Completion>> {
let mut cqe = IORING_CQE {
UserData: 0,
ResultCode: 0,
Information: 0,
};
let hr = unsafe { crate::sys::pop(self.ring_id(), self.handle, &raw mut cqe) };
if hr == S_FALSE {
return Ok(None);
}
check(hr)?;
if !self.record_completion(cqe.UserData) {
self.unminted_completion(cqe.UserData);
}
Ok(Some(Completion {
user_data: cqe.UserData,
result_code: cqe.ResultCode,
information: cqe.Information,
ring_id: self.accounting.ring_id(),
}))
}
}
impl<T, X, K: RingKind> IoRing<T, X, K> {
pub fn pop_within(&mut self, timeout: Duration) -> io::Result<Option<K::Popped<T, X>>> {
self.pop_within_with(&mut SubmitWait, timeout)
}
pub fn pop_within_with<W: CompletionWait + ?Sized>(
&mut self,
wait: &mut W,
timeout: Duration,
) -> io::Result<Option<K::Popped<T, X>>> {
let Some(completion) = self.pop_within_raw_with(wait, timeout)? else {
return Ok(None);
};
let held = self
.reclaim(completion.user_data())
.map(|entry| (entry.payload, entry.extra));
K::shape(completion, held).map(Some)
}
}
impl<T, X, K> IoRing<T, X, K> {
fn pop_within_raw_with<W: CompletionWait + ?Sized>(
&mut self,
wait: &mut W,
timeout: Duration,
) -> io::Result<Option<Completion>> {
let deadline = Instant::now().checked_add(timeout);
loop {
if let Some(completion) = self.pop_raw()? {
return Ok(Some(completion));
}
if self.is_quiescent() {
return Ok(None);
}
let remaining = match deadline {
Some(deadline) => deadline.saturating_duration_since(Instant::now()),
None => Duration::MAX,
};
if remaining.is_zero() {
return Ok(None);
}
let ms = u32::try_from(remaining.as_millis())
.unwrap_or(MAX_WAIT_MS)
.clamp(1, MAX_WAIT_MS);
wait.wait(
&mut RingWait {
handle: self.handle,
accounting: &self.accounting,
},
ms,
)?;
}
}
}
pub trait CompletionWait {
fn wait(&mut self, ring: &mut RingWait<'_>, timeout_ms: u32) -> io::Result<()>;
}
#[cfg_attr(
feature = "threadpool",
doc = "[`RingScope`](crate::RingScope) under [D-43](../DESIGN-NOTES.md#d-43)."
)]
#[cfg_attr(
not(feature = "threadpool"),
doc = "`RingScope` (the default `threadpool` feature) under [D-43](../DESIGN-NOTES.md#d-43)."
)]
pub struct RingWait<'ring> {
handle: *mut c_void,
accounting: &'ring Accounting,
}
impl RingWait<'_> {
pub fn block(&mut self, timeout_ms: u32) -> io::Result<()> {
let mut submitted = 0_u32;
let hr = unsafe {
crate::sys::submit(
self.accounting.ring_id(),
self.handle,
1,
timeout_ms,
&raw mut submitted,
)
};
wait_outcome(hr)
}
#[must_use]
pub fn outstanding(&self) -> usize {
self.accounting.outstanding()
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct SubmitWait;
impl CompletionWait for SubmitWait {
fn wait(&mut self, ring: &mut RingWait<'_>, timeout_ms: u32) -> io::Result<()> {
ring.block(timeout_ms)
}
}
#[cfg(test)]
impl IoRing {
fn refused_by_the_kernel() -> Self {
Self {
handle: std::ptr::null_mut(),
version: RingVersion::V1,
supported_ops: OpSupport::default(),
accounting: Accounting::new(),
late_read: ManuallyDrop::new(LateReadArrays::default()),
registered_files: ManuallyDrop::new(Vec::new()),
completion_event: None,
inventory: ManuallyDrop::new(HashMap::new()),
kind: PhantomData,
}
}
}
impl<T, X, K> Drop for IoRing<T, X, K> {
fn drop(&mut self) {
#[cfg(test)]
DROP_RUNS.with(|runs| runs.set(runs.get() + 1));
let quiesced = match self.run_down() {
Ok(()) => true,
Err(error) => {
debug_assert!(
std::thread::panicking(),
"IoRing rundown failed before close: {error}"
);
false
}
};
#[cfg(feature = "kernel-seam")]
crate::sys::ring_dropped(self.accounting.ring_id());
let mut released_files = None;
if quiesced && self.is_quiescent() {
unsafe { ManuallyDrop::drop(&mut self.inventory) };
let arrays = unsafe { ManuallyDrop::take(&mut self.late_read) };
#[cfg(test)]
{
let non_empty = usize::from(!arrays.buffer_infos.is_empty())
+ usize::from(!arrays.file_handles.is_empty());
LATE_READ_ARRAYS_RELEASED.with(|released| released.set(released.get() + non_empty));
}
drop(arrays);
released_files = Some(unsafe { ManuallyDrop::take(&mut self.registered_files) });
}
let hr = unsafe { CloseIoRing(self.handle) };
if let Some(files) = released_files {
#[cfg(test)]
REGISTERED_FILES_RELEASED.with(|released| released.set(released.get() + files.len()));
drop(files);
}
debug_assert!(
hr >= 0 || std::thread::panicking(),
"CloseIoRing failed: 0x{:08X}",
hr as u32
);
}
}
#[cfg(test)]
thread_local! {
pub(crate) static DROP_RUNS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[cfg(test)]
thread_local! {
pub(crate) static LATE_READ_ARRAYS_RELEASED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[cfg(test)]
thread_local! {
pub(crate) static REGISTERED_FILES_RELEASED: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[cfg(test)]
pub(crate) fn pop_within<T, X>(ring: &mut IoRing<T, X>, what: &str) -> Completion {
const BOUND: std::time::Duration = std::time::Duration::from_secs(30);
ring.pop_within(BOUND)
.expect("pop")
.unwrap_or_else(|| panic!("timed out after {BOUND:?} waiting for {what}"))
.0
}
mod bookkeeping;
mod inventory;
mod kind;
pub use kind::{AllowsRaw, Open, Owned, RingKind};
#[cfg(test)]
mod tests;