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

win_durable_ioring/
windows_backend.rs
