use std::ffi::c_void;
use std::io;
use std::mem::ManuallyDrop;
use std::os::windows::io::AsRawHandle;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use windows_sys::Win32::Foundation::HANDLE;
use windows_sys::Win32::Storage::FileSystem::{
FILE_FLUSH_DATA, FILE_FLUSH_DEFAULT, FILE_FLUSH_MIN_METADATA, FILE_FLUSH_MODE,
FILE_FLUSH_NO_SYNC, FILE_WRITE_FLAGS, FILE_WRITE_FLAGS_NONE, FILE_WRITE_FLAGS_WRITE_THROUGH,
IORING_BUFFER_INFO, IORING_BUFFER_REF, IORING_BUFFER_REF_0, IORING_HANDLE_REF,
IORING_HANDLE_REF_0, IORING_REF_RAW, IORING_REF_REGISTERED, IORING_REGISTERED_BUFFER,
IORING_SQE_FLAGS, IOSQE_FLAGS_DRAIN_PRECEDING_OPS, IOSQE_FLAGS_NONE,
};
use crate::error::{PushRefused, check};
use crate::ring::{AllowsRaw, Completion, Entry, Held, IoRing, Op, Open, RingId};
use crate::token::OperationId;
use io_buffers::{IoBuf, IoBufMut, RegisteredSpan};
use win_shared_os_owned_handle::SharedHandle;
#[derive(Clone, Copy, Debug, Default)]
#[must_use]
pub struct PushOptions {
drain_preceding: bool,
}
impl PushOptions {
pub fn new() -> Self {
Self::default()
}
pub fn drain_preceding(mut self, drain: bool) -> Self {
self.drain_preceding = drain;
self
}
fn sqe_flags(self) -> IORING_SQE_FLAGS {
if self.drain_preceding {
IOSQE_FLAGS_DRAIN_PRECEDING_OPS
} else {
IOSQE_FLAGS_NONE
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FlushCoverage {
CoversPrecedingOperations,
Unordered,
}
impl FlushCoverage {
fn sqe_flags(self) -> IORING_SQE_FLAGS {
match self {
Self::CoversPrecedingOperations => IOSQE_FLAGS_DRAIN_PRECEDING_OPS,
Self::Unordered => IOSQE_FLAGS_NONE,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum WriteCaching {
#[default]
Cached,
WriteThrough,
}
impl WriteCaching {
fn raw(self) -> FILE_WRITE_FLAGS {
match self {
Self::Cached => FILE_WRITE_FLAGS_NONE,
Self::WriteThrough => FILE_WRITE_FLAGS_WRITE_THROUGH,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum FlushMode {
#[default]
Default,
Data,
MinMetadata,
NoSync,
}
impl FlushMode {
fn raw(self) -> FILE_FLUSH_MODE {
match self {
Self::Default => FILE_FLUSH_DEFAULT,
Self::Data => FILE_FLUSH_DATA,
Self::MinMetadata => FILE_FLUSH_MIN_METADATA,
Self::NoSync => FILE_FLUSH_NO_SYNC,
}
}
}
fn handle_ref(file: FileRef, ring_id: RingId) -> io::Result<IORING_HANDLE_REF> {
match file {
FileRef::Raw(handle) => Ok(IORING_HANDLE_REF {
Kind: IORING_REF_RAW,
Handle: IORING_HANDLE_REF_0 { Handle: handle },
}),
FileRef::Registered(file) => {
if file.ring_id != ring_id {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"this RegisteredFile was registered on a different IoRing",
));
}
Ok(IORING_HANDLE_REF {
Kind: IORING_REF_REGISTERED,
Handle: IORING_HANDLE_REF_0 {
Index: file.index(),
},
})
}
}
}
fn raw_buffer_ref(address: *mut c_void) -> IORING_BUFFER_REF {
IORING_BUFFER_REF {
Kind: IORING_REF_RAW,
Buffer: IORING_BUFFER_REF_0 { Address: address },
}
}
fn registered_buffer_ref(index: u32, offset: u32) -> IORING_BUFFER_REF {
IORING_BUFFER_REF {
Kind: IORING_REF_REGISTERED,
Buffer: IORING_BUFFER_REF_0 {
IndexAndOffset: IORING_REGISTERED_BUFFER {
BufferIndex: index,
Offset: offset,
},
},
}
}
fn checked_len(len: usize) -> io::Result<u32> {
u32::try_from(len).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
format!("an IoRing buffer is limited to u32::MAX bytes; {len} does not fit"),
)
})
}
#[derive(Clone, Copy, Debug)]
pub enum FileRef {
Raw(HANDLE),
Registered(RegisteredFile),
}
impl From<HANDLE> for FileRef {
fn from(handle: HANDLE) -> Self {
FileRef::Raw(handle)
}
}
impl From<RegisteredFile> for FileRef {
fn from(file: RegisteredFile) -> Self {
FileRef::Registered(file)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RegisteredFile {
index: u32,
ring_id: RingId,
}
impl RegisteredFile {
#[must_use]
pub fn index(self) -> u32 {
self.index
}
}
mod sealed {
pub trait Sealed {}
impl Sealed for super::SharedHandle {}
impl Sealed for super::RegisteredFile {}
}
pub trait FileTarget: sealed::Sealed {
type Guard: Send + 'static + Into<crate::ring::FileGuard>;
fn as_file_ref(&self) -> FileRef;
fn guard(&self) -> Self::Guard;
}
impl FileTarget for SharedHandle {
type Guard = SharedHandle;
fn as_file_ref(&self) -> FileRef {
FileRef::Raw(self.as_raw_handle())
}
fn guard(&self) -> Self::Guard {
self.clone()
}
}
impl FileTarget for RegisteredFile {
type Guard = RegisteredFile;
fn as_file_ref(&self) -> FileRef {
FileRef::Registered(*self)
}
fn guard(&self) -> Self::Guard {
*self
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct RegisteredFiles {
base_index: u32,
count: u32,
ring_id: RingId,
}
impl RegisteredFiles {
#[must_use]
pub fn len(&self) -> u32 {
self.count
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.count == 0
}
#[must_use]
pub fn get(&self, i: u32) -> Option<RegisteredFile> {
(i < self.count).then(|| RegisteredFile {
index: self.base_index + i,
ring_id: self.ring_id,
})
}
}
#[derive(Debug)]
pub struct PendingFileRegistration {
user_data: usize,
base_index: u32,
count: u32,
ring_id: RingId,
}
impl PendingFileRegistration {
#[must_use]
pub fn user_data(&self) -> usize {
self.user_data
}
pub fn claim_if(self, completion: &Completion) -> Result<io::Result<RegisteredFiles>, Self> {
if completion.user_data() != self.user_data || completion.ring_id() != self.ring_id {
return Err(self);
}
Ok(completion.result().map(|_| RegisteredFiles {
base_index: self.base_index,
count: self.count,
ring_id: self.ring_id,
}))
}
}
struct BufferState {
outstanding: AtomicUsize,
kernel_writes: AtomicUsize,
registered_len: u32,
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum KernelAccess {
WritesBuffer,
ReadsBuffer,
}
pub struct RegisteredUse {
state: Arc<[BufferState]>,
index: usize,
access: KernelAccess,
}
impl Drop for RegisteredUse {
fn drop(&mut self) {
if let Some(state) = self.state.get(self.index) {
if self.access == KernelAccess::WritesBuffer {
state.kernel_writes.fetch_sub(1, Ordering::SeqCst);
}
state.outstanding.fetch_sub(1, Ordering::SeqCst);
}
}
}
pub struct RegisteredBuffers<B: IoBufMut> {
buffers: ManuallyDrop<Vec<B>>,
base_index: u32,
state: Arc<[BufferState]>,
ring_id: RingId,
}
impl<B: IoBufMut> RegisteredBuffers<B> {
#[must_use]
pub fn len(&self) -> u32 {
u32::try_from(self.buffers.len()).unwrap_or(u32::MAX)
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.buffers.is_empty()
}
pub fn get(&mut self, i: u32) -> io::Result<&[u8]> {
let index = usize::try_from(i).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
"buffer index does not fit usize",
)
})?;
let Some(state) = self.state.get(index) else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("buffer index {i} is out of range for this registration"),
));
};
let kernel_writes = state.kernel_writes.load(Ordering::SeqCst);
if kernel_writes > 0 {
return Err(io::Error::new(
io::ErrorKind::WouldBlock,
format!(
"buffer {i} still has {kernel_writes} read(s) outstanding into it; the kernel \
may be writing through it. Pop its completion before reading this buffer."
),
));
}
let len = state.registered_len as usize;
let buffer = self
.buffers
.get(index)
.expect("state and buffers are built together and have equal length");
Ok(unsafe { std::slice::from_raw_parts(buffer.stable_ptr(), len) })
}
pub fn get_mut(&mut self, i: u32) -> io::Result<&mut [u8]> {
let index = usize::try_from(i).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidInput,
"buffer index does not fit usize",
)
})?;
let Some(state) = self.state.get(index) else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("buffer index {i} is out of range for this registration"),
));
};
let outstanding = state.outstanding.load(Ordering::SeqCst);
if outstanding > 0 {
return Err(io::Error::new(
io::ErrorKind::WouldBlock,
format!(
"buffer {i} still has {outstanding} operation(s) outstanding; the kernel may \
be reading or writing through it. Pop its completion before filling this \
buffer."
),
));
}
let len = state.registered_len as usize;
let buffer = self
.buffers
.get_mut(index)
.expect("state and buffers are built together and have equal length");
Ok(unsafe { std::slice::from_raw_parts_mut(buffer.stable_mut_ptr(), len) })
}
#[must_use]
pub fn outstanding(&self, i: u32) -> Option<usize> {
self.state
.get(usize::try_from(i).ok()?)
.map(|state| state.outstanding.load(Ordering::SeqCst))
}
pub fn check_span(&self, span: RegisteredSpan) -> io::Result<()> {
self.checked_span(span).map(|_| ())
}
fn begin_use(&self, span: RegisteredSpan, access: KernelAccess) -> RegisteredUse {
let index = span.buffer_index as usize;
if let Some(state) = self.state.get(index) {
if access == KernelAccess::WritesBuffer {
state.kernel_writes.fetch_add(1, Ordering::SeqCst);
}
state.outstanding.fetch_add(1, Ordering::SeqCst);
}
RegisteredUse {
state: Arc::clone(&self.state),
index,
access,
}
}
fn checked_index(&self, i: u32) -> io::Result<u32> {
if i < self.len() {
Ok(self.base_index + i)
} else {
Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!("buffer index {i} is out of range for this registration"),
))
}
}
fn checked_span(&self, span: RegisteredSpan) -> io::Result<u32> {
let index = self.checked_index(span.buffer_index)?;
let registered_len = self
.state
.get(span.buffer_index as usize)
.expect("checked_index just validated this index")
.registered_len;
let fits = u64::from(span.offset)
.checked_add(u64::from(span.len))
.is_some_and(|end| end <= u64::from(registered_len));
if !fits {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
format!(
"span offset {} + len {} does not fit inside buffer {} \
({registered_len} registered bytes)",
span.offset, span.len, span.buffer_index
),
));
}
Ok(index)
}
}
impl<B: IoBufMut> Drop for RegisteredBuffers<B> {
fn drop(&mut self) {
if self
.state
.iter()
.any(|state| state.outstanding.load(Ordering::SeqCst) > 0)
{
debug_assert!(
std::thread::panicking(),
"RegisteredBuffers dropped while an operation still references it"
);
return;
}
unsafe { ManuallyDrop::drop(&mut self.buffers) };
}
}
pub struct PendingBufferRegistration<B: IoBufMut> {
user_data: usize,
base_index: u32,
buffers: ManuallyDrop<Vec<B>>,
registered_lens: Vec<u32>,
ring_id: RingId,
}
impl<B: IoBufMut> PendingBufferRegistration<B> {
#[must_use]
pub fn user_data(&self) -> usize {
self.user_data
}
pub fn claim_if(
mut self,
completion: &Completion,
) -> Result<io::Result<RegisteredBuffers<B>>, Self> {
if completion.user_data() != self.user_data || completion.ring_id() != self.ring_id {
return Err(self);
}
let buffers = unsafe { ManuallyDrop::take(&mut self.buffers) };
let base_index = self.base_index;
let ring_id = self.ring_id;
let registered_lens = std::mem::take(&mut self.registered_lens);
Ok(completion.result().map(move |_| {
let state: Arc<[BufferState]> = registered_lens
.into_iter()
.map(|registered_len| BufferState {
outstanding: AtomicUsize::new(0),
kernel_writes: AtomicUsize::new(0),
registered_len,
})
.collect();
debug_assert_eq!(
state.len(),
buffers.len(),
"one state entry per registered buffer"
);
RegisteredBuffers {
buffers: ManuallyDrop::new(buffers),
base_index,
state,
ring_id,
}
}))
}
}
impl<B: IoBufMut> Drop for PendingBufferRegistration<B> {
fn drop(&mut self) {
}
}
impl<B: IoBufMut> std::fmt::Debug for PendingBufferRegistration<B> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PendingBufferRegistration")
.field("user_data", &self.user_data)
.field("base_index", &self.base_index)
.finish_non_exhaustive()
}
}
pub struct Batch<'ring, T = (), X = (), K = Open> {
ring: &'ring mut IoRing<T, X, K>,
submitted: bool,
}
impl<'ring, T, X, K> Batch<'ring, T, X, K> {
pub fn new(ring: &'ring mut IoRing<T, X, K>) -> Self {
Self {
ring,
submitted: false,
}
}
fn require(&self, op: Op) -> io::Result<()> {
if self.ring.supports(op) {
Ok(())
} else {
Err(io::Error::new(
io::ErrorKind::Unsupported,
format!("this ring does not support {op:?}"),
))
}
}
fn begin_owned(&mut self) -> io::Result<(usize, OperationId)> {
let user_data = self.ring.accounting_mut().reserve_user_data()?;
Ok((user_data, OperationId::new(user_data, self.ring.ring_id())))
}
fn finish_owned(
&mut self,
hr: windows_sys::core::HRESULT,
id: OperationId,
payload: Option<T>,
extra: X,
held: Held,
) -> Result<OperationId, PushRefused<T, X>> {
match check(hr) {
Ok(()) => {
self.ring.stow(
id,
Entry {
payload,
extra,
held,
},
);
Ok(id)
}
Err(error) => {
self.ring.cancel_reservation(id.user_data());
drop(held);
Err(PushRefused::new(error, payload, extra))
}
}
}
fn check_registration_ring<B: IoBufMut>(
&self,
registration: &RegisteredBuffers<B>,
) -> io::Result<()> {
if registration.ring_id == self.ring.ring_id() {
Ok(())
} else {
Err(io::Error::new(
io::ErrorKind::InvalidInput,
"this RegisteredBuffers was registered on a different IoRing",
))
}
}
pub fn submit(self) -> io::Result<u32> {
self.submit_and_wait(0, 0)
}
pub fn leave_queued(mut self) {
self.submitted = true;
}
pub fn submit_and_wait(mut self, wait_operations: u32, timeout_ms: u32) -> io::Result<u32> {
self.do_submit(wait_operations, timeout_ms)
}
fn do_submit(&mut self, wait_operations: u32, timeout_ms: u32) -> io::Result<u32> {
let mut submitted = 0_u32;
let hr = unsafe {
crate::sys::submit(
self.ring.ring_id(),
self.ring.raw_handle(),
wait_operations,
timeout_ms,
&raw mut submitted,
)
};
self.submitted = true;
if !crate::ring::every_entry_was_submitted(hr) {
check(hr)?;
}
Ok(submitted)
}
}
impl<T, X, K> Drop for Batch<'_, T, X, K> {
fn drop(&mut self) {
if !self.submitted {
let _ = self.do_submit(0, 0);
}
}
}
mod push;
#[cfg(test)]
mod tests;