windows-ioring-sys 0.9.0

// Copyright (c) 2026 Mike Grier
//! `ring-copy`: engine (M7.3) -- one domain's read-then-write loop over its
//! own ring, its own registered `VirtualAllocExNuma` buffer, and its own
//! byte range, pinned to its plan's processors.

use std::io;
use std::ops::Range;
use std::ptr;
use std::time::{Duration, Instant};

use win_numa_sys::NumaNode;
use windows_ioring_sys::{
    Batch, IoRing, NumaBuffer, OperationId, PushOptions, RegisteredBuffers, RegisteredSpan,
    WriteCaching,
};
use windows_sys::Win32::Foundation::HANDLE;
use windows_sys::Win32::System::SystemInformation::GROUP_AFFINITY;
use windows_sys::Win32::System::Threading::{GetCurrentThread, SetThreadGroupAffinity};

use crate::plan::DomainPlan;

/// How long waiting for a single operation's completion may block before this
/// sample gives up on it, rather than hanging forever on a stuck device.
const OP_TIMEOUT: Duration = Duration::from_secs(30);

/// What one domain's copy pass accomplished.
pub struct DomainReport {
    pub label: String,
    pub bytes_copied: u64,
    pub elapsed: Duration,
}

/// Copy `byte_range` of `source` to the same range of `destination`, entirely
/// on the calling thread, pinned to `plan`'s processors and reading through a
/// single registered buffer placed on `numa_node` (M7.3).
///
/// # Errors
///
/// Returns any error from pinning this thread, allocating or registering the
/// buffer, or any push, submit, or completion along the way.
pub fn copy_domain(
    plan: &DomainPlan,
    source: HANDLE,
    destination: HANDLE,
    byte_range: Range<u64>,
    chunk_len: usize,
    numa_node: Option<NumaNode>,
) -> io::Result<DomainReport> {
    affinitize(plan.group, plan.mask)?;

    let mut ring = IoRing::new(8, 8)?;
    let buffer = NumaBuffer::new(chunk_len, numa_node)?;
    let registration = register_buffer(&mut ring, buffer)?;

    let start = Instant::now();
    let mut offset = byte_range.start;
    let mut bytes_copied = 0_u64;
    while offset < byte_range.end {
        let requested = u32::try_from((byte_range.end - offset).min(chunk_len as u64))
            .expect("chunk_len fits in u32");
        let read_span = RegisteredSpan {
            buffer_index: 0,
            offset: 0,
            len: requested,
        };

        // A short read is real, not an error: stop at whatever the source
        // actually had rather than writing uninitialized tail bytes or
        // spinning forever on an offset that never advances.
        let transferred = submit_one(&mut ring, |batch| {
            // SAFETY: `source` stays open for this domain's whole copy pass.
            unsafe {
                batch.read_registered_raw_owned(
                    source,
                    &registration,
                    read_span,
                    (),
                    offset,
                    PushOptions::new(),
                )
            }
        })?;
        if transferred == 0 {
            break;
        }

        let write_span = RegisteredSpan {
            buffer_index: 0,
            offset: 0,
            len: transferred,
        };
        submit_one(&mut ring, |batch| {
            // SAFETY: `destination` stays open for this domain's whole copy pass.
            unsafe {
                batch.write_registered_raw_owned(
                    destination,
                    &registration,
                    write_span,
                    (),
                    offset,
                    PushOptions::new(),
                    WriteCaching::Cached,
                )
            }
        })?;

        offset += u64::from(transferred);
        bytes_copied += u64::from(transferred);
    }

    drop(registration);
    Ok(DomainReport {
        label: plan.label.clone(),
        bytes_copied,
        elapsed: start.elapsed(),
    })
}

/// Pin the calling thread to exactly `group`/`mask` (M7.2's plan is only
/// meaningful if the thread actually runs there).
fn affinitize(group: u16, mask: usize) -> io::Result<()> {
    let affinity = GROUP_AFFINITY {
        Mask: mask,
        Group: group,
        Reserved: [0; 3],
    };
    // SAFETY: `GetCurrentThread` is a pseudo-handle needing no closing;
    // `affinity` is a fully initialized in-argument, and no previous-affinity
    // out-pointer is requested.
    let ok = unsafe { SetThreadGroupAffinity(GetCurrentThread(), &affinity, ptr::null_mut()) };
    if ok == 0 {
        Err(io::Error::last_os_error())
    } else {
        Ok(())
    }
}

/// Register `buffer` with `ring` and block until the registration is
/// confirmed, once per domain (M7.3: "registered once per ring").
fn register_buffer(
    ring: &mut IoRing,
    buffer: NumaBuffer,
) -> io::Result<RegisteredBuffers<NumaBuffer>> {
    let pending = {
        let mut batch = Batch::new(ring);
        let pending = batch.register_buffers(vec![buffer])?;
        batch.submit()?;
        pending
    };
    // One bounded wait, owned by the pop. A submit that waited would promise
    // nothing about poppability once its own timeout expired (RS-P-5), so
    // an immediate `try_pop` after it could call a slow registration missing.
    let (completion, _held) = ring.pop_within(OP_TIMEOUT)?.ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::TimedOut,
            "no completion for buffer registration",
        )
    })?;
    match pending.claim_if(&completion) {
        Ok(result) => result,
        Err(_) => Err(io::Error::other(
            "completion did not match the pending buffer registration",
        )),
    }
}

/// Push one op via `push`, submit it, then wait a bounded time for its
/// completion and claim it, returning the transferred byte count.
fn submit_one<F, E>(ring: &mut IoRing, push: F) -> io::Result<u32>
where
    F: FnOnce(&mut Batch<'_>) -> Result<OperationId, E>,
    io::Error: From<E>,
{
    {
        let mut batch = Batch::new(ring);
        push(&mut batch)?;
        batch.submit()?;
    }
    // The pop is not optional bookkeeping: this operation holds a registration
    // lease, and popping its completion is what releases that lease back to the
    // registered buffer's outstanding count. Skipping it would leave the
    // registration pinned for the rest of the run.
    let (completion, held) = ring.pop_within(OP_TIMEOUT)?.ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::TimedOut,
            "no completion within the operation timeout",
        )
    })?;
    if held.is_none() {
        return Err(io::Error::other(
            "completion did not match anything this ring was holding",
        ));
    }
    let transferred = completion.result()?;
    u32::try_from(transferred)
        .map_err(|_| io::Error::other("transferred byte count does not fit in u32"))
}