win_durable_ioring/
windows_backend.rs

1// Copyright (c) 2026 Mike Grier
2//! The Windows backend (DI-D-50 P10, DI-D-52): [`WindowsBackend`], an [`IoBackend`] over
3//! `windows-ioring-sys`' `IoRing`, and [`Win32FileTarget`], a target backed by a Win32 file.
4//!
5//! **Threadless.** The backend has no threads of its own. Its notification is the ring's
6//! completion event; who waits on it and drains is the driver's delivery (DI-D-52).
7//!
8//! **Queued at push, issued at the doorbell.** A push leaves its entry in the ring's submission
9//! queue (a batch left queued, the ring crate's D-90, not submitted as it drops, its D-5), and
10//! [`IoBackend::submit`] issues what is queued and reports why it could not: the engine rings it
11//! once after the pushes it makes, so a batch of pushes is one submission and a failure to submit
12//! is seen where the doorbell is rung (DI-7+.2).
13//!
14//! **Files, volumes and disks.** A [`Win32FileTarget`] is made from an [`OpenedFile`], a file this
15//! crate opened and so knows the buffering mode of, from an [`OpenedVolume`], a raw volume it opened
16//! and found the disks and sector size of, or from an [`OpenedDisk`], a whole disk it opened by
17//! number, and from nothing else (DI-D-60, DI-D-61, DI-D-62).
18//!
19//! **Targets.** A target given at construction is registered with the ring, which holds it for the
20//! ring's life (`windows-ioring-sys` D-81) and addresses it by index; one attached later is
21//! addressed through its `SharedHandle`, guarded per operation. Registered buffers are given at
22//! construction too.
23//!
24//! **Refusals** (DI-D-52), of a push or of a submission, carry the [`RefusalCause`] the ring's
25//! report maps to; and a write the ring completes short completes with a fault instead, under the
26//! `Fault` policy (DI-D-50).
27
28use std::collections::HashMap;
29use std::io;
30use std::os::windows::io::OwnedHandle;
31use std::sync::Arc;
32use std::time::Duration;
33
34use win_sync_sys::{Access, duplicate_handle_with_access};
35use windows_ioring_sys::{
36    Batch, FlushCoverage, FlushMode, IoBuf, IoBufMut, IoRing, IoRingErrorExt, OperationId, Owned,
37    PushOptions, PushRefused, RegisteredBuffers, RegisteredFile, RegisteredSpan, RingCondition,
38    SharedHandle,
39};
40
41use crate::dioring::{register_buffers, register_files};
42use crate::error_code::{WINDOWS_IORING, windows_fault};
43use durable_ioring::IoTarget;
44use durable_ioring::backend::{
45    Completion, DuplicateTarget, IoBackend, RefusalCause, Refused, RegisteredRegion, Submission,
46    SubmissionData, SubmissionOp, SubmissionToken,
47};
48use durable_ioring::{IoFault, IoTargetKey, WriteCaching};
49
50mod file_target;
51mod opened_disk;
52mod opened_file;
53mod opened_volume;
54pub(crate) mod pool;
55pub(crate) mod signalled;
56mod wait;
57
58pub use wait::CompletionEvent;
59
60pub use file_target::Win32FileTarget;
61pub use opened_disk::{DiskPath, OpenedDisk};
62pub use opened_file::{Buffering, Cached, FileOptions, OpenedFile, RemoteDevice, Unbuffered};
63pub use opened_volume::{
64    OpenedVolume, VolumeOptions, VolumePath, VolumePathError, disk_flush_domain,
65};
66
67#[cfg(test)]
68mod tests;
69
70/// A completion's count, on the Windows path. A target completes a write in full or with a fault
71/// (DI-D-50), and the I/O ring does not complete a file or volume write short, so a short count is
72/// a fault here -- the `Fault` policy -- with no code, since no layer gave one. A read may complete
73/// short.
74pub(crate) fn transfer(
75    result: io::Result<usize>,
76    write: bool,
77    requested: u32,
78) -> Result<u32, IoFault> {
79    match result {
80        // A transfer is never longer than the `u32` length it was pushed with.
81        Ok(transferred) => {
82            let transferred = u32::try_from(transferred).unwrap_or(u32::MAX);
83            if write && transferred < requested {
84                Err(IoFault {
85                    source: WINDOWS_IORING,
86                    code: None,
87                    error: Arc::new(io::Error::other(format!(
88                        "the write completed short: {transferred} of {requested} bytes"
89                    ))),
90                })
91            } else {
92                Ok(transferred)
93            }
94        }
95        Err(error) => Err(windows_fault(error)),
96    }
97}
98
99/// Why the ring refused a push or a submission, as the caller acts on it (DI-D-52). A full
100/// submission queue needs a submit; a full completion queue a drain; a submission already running,
101/// a retry; a corrupt ring is unusable. The ring's own rejections carry an `io::ErrorKind`: an
102/// operation it does not support is `Unsupported`, and a malformed one -- a buffer longer than a
103/// `u32`, a registration of another ring -- is `Invalid`. Anything else is `Unknown`, with the
104/// report.
105pub(crate) fn refusal_cause(error: io::Error) -> RefusalCause {
106    match error.ring_condition() {
107        Some(RingCondition::SubmissionQueueFull) => RefusalCause::SubmitNeeded,
108        Some(RingCondition::CompletionQueueTooFull) => RefusalCause::DrainNeeded,
109        Some(RingCondition::SubmitInProgress) => RefusalCause::Busy,
110        Some(RingCondition::Corrupt) => RefusalCause::Unusable(windows_fault(error)),
111        Some(_) => RefusalCause::Unknown(windows_fault(error)),
112        None => match error.kind() {
113            io::ErrorKind::Unsupported => RefusalCause::Unsupported,
114            io::ErrorKind::InvalidInput => RefusalCause::Invalid,
115            _ => RefusalCause::Unknown(windows_fault(error)),
116        },
117    }
118}
119
120/// What the ring carries for each operation, and hands back with its completion.
121#[derive(Clone, Copy, Debug)]
122pub(crate) struct Sidecar {
123    token: SubmissionToken,
124    target: IoTargetKey,
125    kind: Kind,
126}
127
128/// What an operation was, as its completion is read.
129#[derive(Clone, Copy, Debug)]
130enum Kind {
131    /// A write of this many bytes, which completes in full or with a fault.
132    Write(u32),
133    Read,
134    Commit,
135}
136
137/// A target and how the ring addresses it.
138struct Slot {
139    target: Win32FileTarget,
140    /// Its registration, for a target given at construction.
141    registered: Option<RegisteredFile>,
142}
143
144/// A test hook called with a ring as soon as it exists (DI-7.2).
145#[cfg(test)]
146pub(crate) type RingHook<B> = Box<dyn FnOnce(&IoRing<B, Sidecar>)>;
147
148/// What construction needs beyond [`WindowsBackendSetup`], passed down explicitly from
149/// [`Dioring::with_construction`](crate::dioring::Dioring::with_construction) (DI-7.2): never a
150/// thread-local or a global, since the ring is created and registered before its consumer can
151/// name it, and a responder installed by thread would miss a pool thread's calls (the ring
152/// crate's D-85).
153pub(crate) struct Construction<B> {
154    /// How long construction waits for each registration's completion.
155    pub(crate) registration_bound: Duration,
156    /// Called with the ring as soon as it exists, before anything is registered on it: a test
157    /// registers a responder here, which is the only moment that reaches the registration calls.
158    #[cfg(test)]
159    pub(crate) on_ring: Option<RingHook<B>>,
160    /// `B` is named only by the test hook.
161    ring: std::marker::PhantomData<fn(B)>,
162}
163
164impl<B> Default for Construction<B> {
165    /// What every instance outside a test is built with.
166    fn default() -> Self {
167        Self {
168            registration_bound: crate::dioring::REGISTRATION_BOUND,
169            #[cfg(test)]
170            on_ring: None,
171            ring: std::marker::PhantomData,
172        }
173    }
174}
175
176#[cfg(test)]
177impl<B> Construction<B> {
178    /// A construction whose ring answers through the ring crate's safe call-fault responder,
179    /// registered as the ring is created, waiting `registration_bound` for each registration. The
180    /// control arms it from any thread; nothing is armed to begin with.
181    pub(crate) fn faulted(
182        registration_bound: Duration,
183    ) -> (Self, windows_ioring_sys::sys::CallFaultControl) {
184        let (faults, control) = windows_ioring_sys::sys::call_faults();
185        let on_ring: RingHook<B> = Box::new(move |ring| {
186            let _ = windows_ioring_sys::sys::register(ring, Box::new(faults));
187        });
188        let construction = Self {
189            registration_bound,
190            on_ring: Some(on_ring),
191            ring: std::marker::PhantomData,
192        };
193        (construction, control)
194    }
195}
196
197/// What [`WindowsBackend::new`] takes.
198pub struct WindowsBackendSetup<R> {
199    /// Submission queue entries.
200    pub submission_queue_size: u32,
201    /// Completion queue entries.
202    pub completion_queue_size: u32,
203    /// Targets registered with the ring, under their keys, in order.
204    pub targets: Vec<(IoTargetKey, Win32FileTarget)>,
205    /// Buffers registered with the ring, in order; a [`RegisteredRegion`] names one by position.
206    pub buffers: Vec<R>,
207}
208
209/// Why [`WindowsBackend::new`] refused.
210#[derive(Debug)]
211pub enum WindowsBackendRefusal {
212    /// Two targets share a key.
213    DuplicateKey(IoTargetKey),
214    /// Creating the ring, registering, or attaching its completion event failed.
215    Ring(io::Error),
216}
217
218/// A refused construction, handing back what it was given: the targets always, and the buffers
219/// unless the kernel was given them and could not be shown to have given them back.
220pub struct WindowsBackendError<R> {
221    /// Why.
222    pub reason: WindowsBackendRefusal,
223    /// The targets.
224    pub targets: Vec<(IoTargetKey, Win32FileTarget)>,
225    /// The buffers, or `None`.
226    pub buffers: Option<Vec<R>>,
227}
228
229impl<R> std::fmt::Debug for WindowsBackendError<R> {
230    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
231        f.debug_struct("WindowsBackendError")
232            .field("reason", &self.reason)
233            .field("targets", &self.targets.len())
234            .field("buffers", &self.buffers.as_ref().map(Vec::len))
235            .finish()
236    }
237}
238
239/// The Windows backend: an `IoRing` and the targets it runs I/O for. `B` is the owned buffer type
240/// and `R` the registered buffers'.
241///
242/// **Write-only buffer types** (DI-D-17): an owned write needs only `B: IoBuf`, so a type such as `Arc<[u8]>` can be written. An owned read
243/// needs `B: IoBufMut`, and is pushed through a push made where that bound holds: [`new`](Self::new)
244/// makes it, [`without_owned_reads`](Self::without_owned_reads) does not, and an owned read on a
245/// backend without it is refused as [`RefusalCause::Unsupported`].
246pub struct WindowsBackend<B: IoBuf, R: IoBufMut = Vec<u8>> {
247    /// Declared before `registered`: dropping the ring runs down every operation, so it ends before
248    /// the buffers the kernel may still be using.
249    ring: IoRing<B, Sidecar, Owned>,
250    registered: Option<RegisteredBuffers<R>>,
251    targets: HashMap<IoTargetKey, Slot>,
252    /// How an owned read is pushed, made where `B: IoBufMut` holds; `None` refuses owned reads.
253    owned_read: Option<OwnedRead<B>>,
254    /// The ring's completion event, duplicated for each notification handed out.
255    event: OwnedHandle,
256    /// Something is queued that no submit has issued: pushed since the last one that succeeded.
257    queued: bool,
258    /// Failures armed by the fault seam: the next commit of the target to complete reports the
259    /// Win32 error instead of what the kernel did.
260    #[cfg(feature = "fault-injection")]
261    injected: Vec<(IoTargetKey, u32)>,
262}
263
264impl<B: IoBufMut, R: IoBufMut> WindowsBackend<B, R> {
265    /// A ring with `setup`'s queue sizes, its targets and buffers registered, and its completion
266    /// event attached, which reads into owned buffers as well as writing from them.
267    ///
268    /// # Errors
269    ///
270    /// [`WindowsBackendError`], handing back the targets and, where it can, the buffers.
271    pub fn new(setup: WindowsBackendSetup<R>) -> Result<Self, WindowsBackendError<R>> {
272        let mut backend = Self::without_owned_reads(setup)?;
273        backend.enable_owned_reads();
274        Ok(backend)
275    }
276
277    /// Read into owned buffers from now on, through a push made here, where `B: IoBufMut` holds.
278    pub(crate) fn enable_owned_reads(&mut self) {
279        self.owned_read = Some(push_owned_read::<B>);
280    }
281}
282
283impl<B: IoBuf, R: IoBufMut> WindowsBackend<B, R> {
284    /// As [`new`](WindowsBackend::new), for a buffer type that may be only `IoBuf`: it writes from
285    /// owned buffers and reads into registered ones, and refuses an owned read as
286    /// [`RefusalCause::Unsupported`] (DI-D-17).
287    ///
288    /// # Errors
289    ///
290    /// [`WindowsBackendError`], handing back the targets and, where it can, the buffers.
291    pub fn without_owned_reads(
292        setup: WindowsBackendSetup<R>,
293    ) -> Result<Self, WindowsBackendError<R>> {
294        Self::construct(setup, Construction::default())
295    }
296
297    /// [`without_owned_reads`](Self::without_owned_reads), as `construction` says (DI-7.2).
298    ///
299    /// # Errors
300    ///
301    /// As [`without_owned_reads`](Self::without_owned_reads).
302    pub(crate) fn construct(
303        setup: WindowsBackendSetup<R>,
304        construction: Construction<B>,
305    ) -> Result<Self, WindowsBackendError<R>> {
306        let Construction {
307            registration_bound,
308            #[cfg(test)]
309            on_ring,
310            ..
311        } = construction;
312        let WindowsBackendSetup {
313            submission_queue_size,
314            completion_queue_size,
315            targets,
316            buffers,
317        } = setup;
318        if let Some(key) = durable_ioring::driver::first_repeat(targets.iter().map(|(key, _)| *key))
319        {
320            return Err(WindowsBackendError {
321                reason: WindowsBackendRefusal::DuplicateKey(key),
322                targets,
323                buffers: Some(buffers),
324            });
325        }
326        let fail = |error, targets, buffers| WindowsBackendError {
327            reason: WindowsBackendRefusal::Ring(error),
328            targets,
329            buffers,
330        };
331        let mut ring = match IoRing::with_inventory(submission_queue_size, completion_queue_size) {
332            Ok(ring) => ring,
333            Err(error) => return Err(fail(error, targets, Some(buffers))),
334        };
335        #[cfg(test)]
336        if let Some(on_ring) = on_ring {
337            on_ring(&ring);
338        }
339        let handles = targets
340            .iter()
341            .map(|(_, target)| target.handle().clone())
342            .collect();
343        let files = match register_files(&mut ring, handles, registration_bound) {
344            Ok(files) => files,
345            Err(error) => return Err(fail(error, targets, Some(buffers))),
346        };
347        let registered = match register_buffers(&mut ring, buffers, registration_bound) {
348            Ok(registered) => registered,
349            Err((error, buffers)) => return Err(fail(error, targets, buffers)),
350        };
351        let event = match ring.completion_event() {
352            Ok(event) => event,
353            Err(error) => return Err(fail(error, targets, None)),
354        };
355        // Registrations were the last operations queued that the ring holds nothing for, and each
356        // was waited on, so none is in flight. If one is, this crate left it: an inconsistency, not
357        // an outcome to report (DI-D-59).
358        let ring = ring.into_owned().unwrap_or_else(|_| {
359            panic!(
360                "a registration was left in flight on the backend's ring: it cannot be sealed \
361                 against operations the ring holds nothing for"
362            )
363        });
364        let slots = targets
365            .into_iter()
366            .enumerate()
367            .map(|(i, (key, target))| {
368                let index = files
369                    .as_ref()
370                    .and_then(|files| files.get(u32::try_from(i).ok()?))
371                    .expect("every target given at construction was registered, in order");
372                (
373                    key,
374                    Slot {
375                        target,
376                        registered: Some(index),
377                    },
378                )
379            })
380            .collect();
381        Ok(Self {
382            ring,
383            registered,
384            targets: slots,
385            owned_read: None,
386            event,
387            queued: false,
388            #[cfg(feature = "fault-injection")]
389            injected: Vec::new(),
390        })
391    }
392
393    /// A thread-pool wake over this backend's ring (DI-D-53): `on_wake` is called on a pool thread
394    /// whenever the ring's completions may have gone from none to some, and once at setup, and
395    /// must drain this backend to empty each time. `env` binds the pool; `None` is the process's
396    /// default. The ring crate's `CompletionWake` is the protocol; this backend only lends it the
397    /// ring.
398    ///
399    /// # Errors
400    ///
401    /// As `CompletionWake::new`: the system does not support a ring completion event, or the
402    /// wait could not be made or armed.
403    pub(crate) fn wake(
404        &mut self,
405        on_wake: impl Fn() + Send + Sync + 'static,
406        env: Option<&mut windows_threadpool_sys::callback_env::CallbackEnviron<'_>>,
407    ) -> io::Result<windows_ioring_sys::CompletionWake> {
408        windows_ioring_sys::CompletionWake::new(&mut self.ring, on_wake, env)
409    }
410
411    /// A handle to the ring's completion event that can wait on it and do nothing else: a
412    /// duplicate with `SYNCHRONIZE` access alone, so the kernel refuses a set or a reset of it
413    /// (DI-7+.1).
414    ///
415    /// # Errors
416    ///
417    /// The duplication's: the process cannot get another handle.
418    pub(crate) fn completion_wait_handle(&self) -> io::Result<OwnedHandle> {
419        duplicate_handle_with_access(&self.event, Access::SYNCHRONIZE)
420    }
421
422    /// Fail this backend's ring's submits on demand, whichever thread makes them (DI-7.1): a
423    /// responder is registered for the ring, which its drop ends after the rundown, so the fault
424    /// is cleared before the backend drops with an operation outstanding. Once per backend.
425    #[cfg(test)]
426    pub(crate) fn submit_faults(&self) -> windows_ioring_sys::sys::SubmitFaultControl {
427        let (faults, control) = windows_ioring_sys::sys::submit_faults();
428        let _ = windows_ioring_sys::sys::register(&self.ring, Box::new(faults));
429        control
430    }
431
432    /// Fail any call of this backend's ring's the ring crate's seam carries, on demand (DI-7+.2):
433    /// a pop that fails, for one. A ring has one responder, so this is instead of
434    /// [`submit_faults`](Self::submit_faults) and not beside it.
435    #[cfg(test)]
436    pub(crate) fn call_faults(&self) -> windows_ioring_sys::sys::CallFaultControl {
437        let (faults, control) = windows_ioring_sys::sys::call_faults();
438        let _ = windows_ioring_sys::sys::register(&self.ring, Box::new(faults));
439        control
440    }
441
442    /// The target under `key`, if there is one.
443    #[must_use]
444    pub fn target(&self, key: IoTargetKey) -> Option<&Win32FileTarget> {
445        self.targets.get(&key).map(|slot| &slot.target)
446    }
447
448    /// The position of `key`'s target in the ring's file registration, for a target given at
449    /// construction; `None` for one attached later, or no target.
450    #[cfg(test)]
451    pub(crate) fn registration_index(&self, key: IoTargetKey) -> Option<u32> {
452        self.targets
453            .get(&key)
454            .and_then(|slot| slot.registered)
455            .map(RegisteredFile::index)
456    }
457
458    /// Registered buffer `index`'s bytes.
459    ///
460    /// # Errors
461    ///
462    /// For a backend built without registered buffers, or an index out of range.
463    pub fn registered_buffer(&mut self, index: u32) -> io::Result<&[u8]> {
464        self.registration()?.get(index)
465    }
466
467    /// Registered buffer `index`'s bytes, writable.
468    ///
469    /// # Errors
470    ///
471    /// As [`registered_buffer`](Self::registered_buffer).
472    pub fn registered_buffer_mut(&mut self, index: u32) -> io::Result<&mut [u8]> {
473        self.registration()?.get_mut(index)
474    }
475
476    fn registration(&mut self) -> io::Result<&mut RegisteredBuffers<R>> {
477        self.registered.as_mut().ok_or_else(|| {
478            io::Error::new(
479                io::ErrorKind::InvalidInput,
480                "this backend was built without registered buffers",
481            )
482        })
483    }
484
485    /// Why `op` on `target` cannot be pushed, before the ring is asked: an unknown target, an owned
486    /// read on a backend without owned reads (DI-D-17), a registered region without a registration,
487    /// or an operation its target's geometry does not admit. A registered region is checked for offset and length only; its buffer's address is
488    /// the registration's, and whether it lies within the registration is the ring's own check at
489    /// push, which [`IoBackend::check`] asks without pushing.
490    fn admits(&self, target: IoTargetKey, op: &SubmissionOp<B>) -> Result<(), RefusalCause> {
491        let Some(slot) = self.targets.get(&target) else {
492            return Err(RefusalCause::UnknownTarget);
493        };
494        if self.owned_read.is_none()
495            && matches!(
496                op,
497                SubmissionOp::Read {
498                    data: SubmissionData::Owned(_),
499                    ..
500                }
501            )
502        {
503            return Err(RefusalCause::Unsupported);
504        }
505        let (offset, data) = match op {
506            SubmissionOp::Write { offset, data, .. } | SubmissionOp::Read { offset, data } => {
507                (*offset, data)
508            }
509            SubmissionOp::Commit => return Ok(()),
510        };
511        let geometry = slot.target.geometry();
512        let admitted = match data {
513            SubmissionData::Owned(buffer) => {
514                let length = u64::try_from(buffer.bytes_len()).unwrap_or(u64::MAX);
515                // The ring's own limit, which a push would refuse beyond only after the operation
516                // had been held for its gate: asked here, it is refused at the call that gave it.
517                u32::try_from(length).is_ok()
518                    && geometry.admits(offset, length)
519                    && geometry.admits_buffer(buffer.stable_ptr() as usize)
520            }
521            SubmissionData::Registered(region) => {
522                if self.registered.is_none() {
523                    return Err(RefusalCause::Unsupported);
524                }
525                geometry.admits(offset, u64::from(region.length))
526            }
527        };
528        if admitted {
529            Ok(())
530        } else {
531            Err(RefusalCause::Invalid)
532        }
533    }
534}
535
536/// Push one operation against `file`, a registered index or a shared handle, which `via` also
537/// names for an owned read's push.
538macro_rules! push_to {
539    ($batch:expr, $file:expr, $via:expr, $read:expr, $registered:expr, $op:expr, $sidecar:expr) => {{
540        let options = PushOptions::new();
541        match $op {
542            Op::Write {
543                offset,
544                data: SubmissionData::Owned(buffer),
545                caching,
546            } => $batch.write_owned($file, buffer, $sidecar, offset, options, caching),
547            Op::Write {
548                offset,
549                data: SubmissionData::Registered(region),
550                caching,
551            } => $batch.write_registered_owned(
552                $file,
553                $registered.expect("refusal() checked for a registration"),
554                span(region),
555                $sidecar,
556                offset,
557                options,
558                caching,
559            ),
560            Op::Read {
561                offset,
562                data: SubmissionData::Owned(buffer),
563            } => $read.expect("admits() checked for an owned read")(
564                &mut $batch,
565                $via,
566                buffer,
567                $sidecar,
568                offset,
569            ),
570            Op::Read {
571                offset,
572                data: SubmissionData::Registered(region),
573            } => $batch.read_registered_owned(
574                $file,
575                $registered.expect("refusal() checked for a registration"),
576                span(region),
577                $sidecar,
578                offset,
579                options,
580            ),
581            Op::Commit => $batch.flush_owned(
582                $file,
583                $sidecar,
584                FlushCoverage::Unordered,
585                FlushMode::Default,
586            ),
587        }
588    }};
589}
590
591/// What a push is made against, borrowed from a target's slot.
592#[derive(Clone, Copy)]
593enum Via<'a> {
594    Registered(&'a RegisteredFile),
595    Shared(&'a SharedHandle),
596}
597
598/// An owned read's push, made where `B: IoBufMut` holds and carried by a backend whose `B` may be
599/// only `IoBuf` (DI-D-17).
600type OwnedRead<B> = for<'r, 'v> fn(
601    &mut Batch<'r, B, Sidecar, Owned>,
602    Via<'v>,
603    B,
604    Sidecar,
605    u64,
606) -> Result<OperationId, PushRefused<B, Sidecar>>;
607
608/// Push an owned read: the push [`WindowsBackend::new`] makes.
609fn push_owned_read<B: IoBufMut>(
610    batch: &mut Batch<'_, B, Sidecar, Owned>,
611    via: Via<'_>,
612    buffer: B,
613    sidecar: Sidecar,
614    offset: u64,
615) -> Result<OperationId, PushRefused<B, Sidecar>> {
616    let options = PushOptions::new();
617    match via {
618        Via::Registered(index) => batch.read_owned(index, buffer, sidecar, offset, options),
619        Via::Shared(handle) => batch.read_owned(handle, buffer, sidecar, offset, options),
620    }
621}
622
623/// An operation in the ring crate's caching terms.
624enum Op<B> {
625    Write {
626        offset: u64,
627        data: SubmissionData<B>,
628        caching: windows_ioring_sys::WriteCaching,
629    },
630    Read {
631        offset: u64,
632        data: SubmissionData<B>,
633    },
634    Commit,
635}
636
637/// A region as the ring crate names it.
638fn span(region: RegisteredRegion) -> RegisteredSpan {
639    RegisteredSpan {
640        buffer_index: region.index,
641        offset: region.offset,
642        len: region.length,
643    }
644}
645
646/// The contract's caching choice, as the ring crate spells it.
647fn ring_caching(caching: WriteCaching) -> windows_ioring_sys::WriteCaching {
648    match caching {
649        WriteCaching::Cached => windows_ioring_sys::WriteCaching::Cached,
650        WriteCaching::WriteThrough => windows_ioring_sys::WriteCaching::WriteThrough,
651    }
652}
653
654/// The submission a refused push was built from, with what the ring handed back.
655fn rebuilt<B>(
656    token: SubmissionToken,
657    target: IoTargetKey,
658    shape: Shape,
659    payload: Option<B>,
660) -> Submission<B> {
661    let data = |region| match (payload, region) {
662        (Some(buffer), _) => SubmissionData::Owned(buffer),
663        (None, Some(region)) => SubmissionData::Registered(region),
664        (None, None) => unreachable!("a refused owned push hands its buffer back"),
665    };
666    let op = match shape {
667        Shape::Write {
668            offset,
669            region,
670            caching,
671        } => SubmissionOp::Write {
672            offset,
673            data: data(region),
674            caching,
675        },
676        Shape::Read { offset, region } => SubmissionOp::Read {
677            offset,
678            data: data(region),
679        },
680        Shape::Commit => SubmissionOp::Commit,
681    };
682    Submission { token, target, op }
683}
684
685/// What a submission was, without its buffer, kept to rebuild it if the ring refuses it.
686#[derive(Clone, Copy)]
687enum Shape {
688    Write {
689        offset: u64,
690        region: Option<RegisteredRegion>,
691        caching: WriteCaching,
692    },
693    Read {
694        offset: u64,
695        region: Option<RegisteredRegion>,
696    },
697    Commit,
698}
699
700impl Shape {
701    fn of<B>(op: &SubmissionOp<B>) -> Self {
702        let region = |data: &SubmissionData<B>| match data {
703            SubmissionData::Owned(_) => None,
704            SubmissionData::Registered(region) => Some(*region),
705        };
706        match op {
707            SubmissionOp::Write {
708                offset,
709                data,
710                caching,
711            } => Shape::Write {
712                offset: *offset,
713                region: region(data),
714                caching: *caching,
715            },
716            SubmissionOp::Read { offset, data } => Shape::Read {
717                offset: *offset,
718                region: region(data),
719            },
720            SubmissionOp::Commit => Shape::Commit,
721        }
722    }
723}
724
725/// What an operation's completion is read as.
726fn kind<B: IoBuf>(op: &SubmissionOp<B>) -> Kind {
727    match op {
728        SubmissionOp::Write { data, .. } => Kind::Write(match data {
729            SubmissionData::Owned(buffer) => u32::try_from(buffer.bytes_len()).unwrap_or(u32::MAX),
730            SubmissionData::Registered(region) => region.length,
731        }),
732        SubmissionOp::Read { .. } => Kind::Read,
733        SubmissionOp::Commit => Kind::Commit,
734    }
735}
736
737impl<B: IoBuf, R: IoBufMut> IoBackend for WindowsBackend<B, R> {
738    type Target = Win32FileTarget;
739    type Buffer = B;
740    type Notification = CompletionEvent;
741
742    fn attach(
743        &mut self,
744        key: IoTargetKey,
745        target: Win32FileTarget,
746    ) -> Result<(), DuplicateTarget<Win32FileTarget>> {
747        if self.targets.contains_key(&key) {
748            return Err(DuplicateTarget { key, target });
749        }
750        self.targets.insert(
751            key,
752            Slot {
753                target,
754                registered: None,
755            },
756        );
757        Ok(())
758    }
759
760    fn detach(&mut self, key: IoTargetKey) -> Option<Win32FileTarget> {
761        self.targets.remove(&key).map(|slot| slot.target)
762    }
763
764    fn registered_buffers(&self) -> u32 {
765        self.registered.as_ref().map_or(0, RegisteredBuffers::len)
766    }
767
768    fn check(&self, target: IoTargetKey, op: &SubmissionOp<B>) -> Result<(), RefusalCause> {
769        self.admits(target, op)?;
770        // The span check the ring makes at push, asked without pushing (its M34.1), and read as a
771        // push's refusal is.
772        let region = match op {
773            SubmissionOp::Write {
774                data: SubmissionData::Registered(region),
775                ..
776            }
777            | SubmissionOp::Read {
778                data: SubmissionData::Registered(region),
779                ..
780            } => *region,
781            _ => return Ok(()),
782        };
783        match &self.registered {
784            Some(registered) => registered.check_span(span(region)).map_err(refusal_cause),
785            None => Err(RefusalCause::Unsupported),
786        }
787    }
788
789    fn push(&mut self, submission: Submission<B>) -> Result<(), Refused<B>> {
790        if let Err(cause) = self.admits(submission.target, &submission.op) {
791            return Err(Refused { cause, submission });
792        }
793        let Submission { token, target, op } = submission;
794        let shape = Shape::of(&op);
795        let sidecar = Sidecar {
796            token,
797            target,
798            kind: kind(&op),
799        };
800        let op = match op {
801            SubmissionOp::Write {
802                offset,
803                data,
804                caching,
805            } => Op::Write {
806                offset,
807                data,
808                caching: ring_caching(caching),
809            },
810            SubmissionOp::Read { offset, data } => Op::Read { offset, data },
811            SubmissionOp::Commit => Op::Commit,
812        };
813        let slot = &self.targets[&target];
814        let registered = self.registered.as_ref();
815        let read = self.owned_read;
816        let mut batch = Batch::new(&mut self.ring);
817        let pushed: Result<OperationId, PushRefused<B, Sidecar>> = match &slot.registered {
818            Some(index) => {
819                let via = Via::Registered(index);
820                push_to!(batch, index, via, read, registered, op, sidecar)
821            }
822            None => {
823                let handle = slot.target.handle();
824                let via = Via::Shared(handle);
825                push_to!(batch, handle, via, read, registered, op, sidecar)
826            }
827        };
828        match pushed {
829            Ok(_) => {
830                // Queued, not submitted: the doorbell is `submit`'s, rung by the engine, so a
831                // failure of it is seen by the caller that rings it and not lost in a drop.
832                batch.leave_queued();
833                self.queued = true;
834                Ok(())
835            }
836            Err(refused) => {
837                // Nothing of this push is queued, but an earlier one may be, which a drop would
838                // submit and not report on.
839                batch.leave_queued();
840                let PushRefused { error, payload, .. } = refused;
841                Err(Refused {
842                    cause: refusal_cause(error),
843                    submission: rebuilt(token, target, shape, payload),
844                })
845            }
846        }
847    }
848
849    fn submit(&mut self) -> Result<(), RefusalCause> {
850        if !self.queued {
851            return Ok(());
852        }
853        match Batch::new(&mut self.ring).submit() {
854            Ok(_) => {
855                self.queued = false;
856                Ok(())
857            }
858            Err(error) => Err(refusal_cause(error)),
859        }
860    }
861
862    fn pop(&mut self) -> Result<Option<Completion<B>>, IoFault> {
863        // The ring is an owned one, so it never hands back a completion it held nothing for: there
864        // is no such case to report (DI-D-59).
865        let Some((completion, buffer, sidecar)) = self.ring.try_pop().map_err(windows_fault)?
866        else {
867            return Ok(None);
868        };
869        #[cfg(feature = "fault-injection")]
870        let completion = match sidecar.kind {
871            Kind::Commit => match self
872                .injected
873                .iter()
874                .position(|(target, _)| *target == sidecar.target)
875            {
876                Some(index) => {
877                    let (_, code) = self.injected.remove(index);
878                    completion
879                        .with_injected_failure(windows_ioring_sys::InjectedFailure::Win32(code))
880                }
881                None => completion,
882            },
883            Kind::Write(_) | Kind::Read => completion,
884        };
885        let result = match sidecar.kind {
886            Kind::Write(requested) => transfer(completion.result(), true, requested),
887            Kind::Read => transfer(completion.result(), false, 0),
888            Kind::Commit => completion.result().map(|_| 0).map_err(windows_fault),
889        };
890        Ok(Some(Completion {
891            token: sidecar.token,
892            result,
893            buffer,
894        }))
895    }
896
897    fn notification(&self) -> Result<CompletionEvent, IoFault> {
898        self.event
899            .try_clone()
900            .map(CompletionEvent::new)
901            .map_err(windows_fault)
902    }
903}
904
905/// The fault seam (`DI-3.2.4`), on the backend: failures on demand, for tests. It transforms a
906/// commit's real completion through `windows-ioring-sys`' own seam, so the commit still reaches the
907/// kernel and only the answer changes.
908#[cfg(feature = "fault-injection")]
909impl<B: IoBuf, R: IoBufMut> WindowsBackend<B, R> {
910    /// Make the next commit of `target` to complete report the Win32 error `code`, whatever the
911    /// kernel answered. Armed failures for one target are used in the order armed.
912    pub fn fail_next_commit(&mut self, target: IoTargetKey, code: u32) {
913        self.injected.push((target, code));
914    }
915}
win_durable_ioring/windows_backend.rs

win_durable_ioring/
windows_backend.rs
