windows_ioring_sys/batch.rs
1// Copyright (c) 2026 Mike Grier
2//! `Batch`: a scoped, exclusive submission window (M3.1-M3.3, M3.5, M5).
3
4use std::ffi::c_void;
5use std::io;
6use std::mem::ManuallyDrop;
7use std::os::windows::io::AsRawHandle;
8use std::sync::Arc;
9use std::sync::atomic::{AtomicUsize, Ordering};
10
11use windows_sys::Win32::Foundation::HANDLE;
12use windows_sys::Win32::Storage::FileSystem::{
13 FILE_FLUSH_DATA, FILE_FLUSH_DEFAULT, FILE_FLUSH_MIN_METADATA, FILE_FLUSH_MODE,
14 FILE_FLUSH_NO_SYNC, FILE_WRITE_FLAGS, FILE_WRITE_FLAGS_NONE, FILE_WRITE_FLAGS_WRITE_THROUGH,
15 IORING_BUFFER_INFO, IORING_BUFFER_REF, IORING_BUFFER_REF_0, IORING_HANDLE_REF,
16 IORING_HANDLE_REF_0, IORING_REF_RAW, IORING_REF_REGISTERED, IORING_REGISTERED_BUFFER,
17 IORING_SQE_FLAGS, IOSQE_FLAGS_DRAIN_PRECEDING_OPS, IOSQE_FLAGS_NONE,
18};
19
20use crate::error::{PushRefused, check};
21use crate::ring::{AllowsRaw, Completion, Entry, Held, IoRing, Op, Open, RingId};
22use crate::token::OperationId;
23use io_buffers::{IoBuf, IoBufMut, RegisteredSpan};
24use win_shared_os_owned_handle::SharedHandle;
25
26/// Per-push options shared across every op builder (M3.2).
27#[derive(Clone, Copy, Debug, Default)]
28#[must_use]
29pub struct PushOptions {
30 drain_preceding: bool,
31}
32
33impl PushOptions {
34 /// The default options: no barrier.
35 pub fn new() -> Self {
36 Self::default()
37 }
38
39 /// Set `IOSQE_FLAGS_DRAIN_PRECEDING_OPS`: this op does not complete until
40 /// every op already queued on this batch's ring has completed. A
41 /// barrier, not a cheap tag -- but a **one-sided** one, and the direction
42 /// it does not cover is the surprising half.
43 ///
44 /// Stated in terms of completion rather than of when the op *starts*,
45 /// throughout: user mode cannot see execution begin, only completions
46 /// posted, so completion order is both what was measured and the only
47 /// thing a caller can act on ([D-47](../DESIGN-NOTES.md#d-47-detail)).
48 ///
49 /// # Ring-wide in reach, one-sided in effect
50 ///
51 /// Measured, not inferred (D-24 and D-47 in `DESIGN-NOTES.md`). The wait
52 /// reaches every operation outstanding on the *ring*, not only the ones
53 /// queued in this batch, and results were identical whether the sequence
54 /// went in one [`Batch::submit`] or three.
55 ///
56 /// **It does not hold back operations pushed after it.** Those can complete
57 /// while this one is still outstanding -- observed across about 4,500
58 /// trials, and in the worst case every one of 32 subsequent writes
59 /// completed first. An earlier version of this documentation said they were
60 /// held; that was measured over too few runs to see a fault this rare, and
61 /// it was wrong.
62 ///
63 /// The consequence to plan for: **this delays the flagged op, not the
64 /// ring.** Later work is not blocked, so it cannot be used to fence
65 /// anything that follows. A consumer that needs later work to begin only
66 /// after this op is durable must sequence that itself, by waiting for its
67 /// completion before submitting.
68 ///
69 /// # The barrier stops at the ring's edge
70 ///
71 /// It orders SQEs against SQEs, and nothing else. Any non-ring I/O -- an
72 /// overlapped `DeviceIoControl`, anything issued through
73 /// `windows-overlapped-io-sys`, a plain blocking write -- is outside it in
74 /// *both* directions: this flag can neither make a ring op wait for a
75 /// non-ring op nor make a non-ring op wait for ring ops.
76 ///
77 /// A consumer mixing both paths is the normal case rather than an exotic
78 /// one, and it must enforce that ordering in its own code. What this
79 /// crate offers toward it is [`IoRing::completion_event`], which makes
80 /// waiting on the ring *alongside* other handles expressible without
81 /// giving up the ring or blocking a thread in a drain.
82 ///
83 /// # Ordering, never durability by itself
84 ///
85 /// This flag orders operations; it does not flush anything. It is,
86 /// however, what makes a flush cover the writes before it: a flush pushed
87 /// after a batch of writes *without* this flag routinely completes while
88 /// many of those writes are still outstanding (D-23 in `DESIGN-NOTES.md`,
89 /// measured at 17 and 23 of 32 writes finishing after the flush did). So
90 /// the obvious spelling -- push the writes, then push a flush -- silently
91 /// does not make those writes durable.
92 ///
93 /// A flush does not take its barrier decision from here, for exactly that
94 /// reason: [`Batch::flush_owned`] requires a [`FlushCoverage`] instead, so the
95 /// choice cannot be inherited from a default (M12.1, D-25).
96 pub fn drain_preceding(mut self, drain: bool) -> Self {
97 self.drain_preceding = drain;
98 self
99 }
100
101 fn sqe_flags(self) -> IORING_SQE_FLAGS {
102 if self.drain_preceding {
103 IOSQE_FLAGS_DRAIN_PRECEDING_OPS
104 } else {
105 IOSQE_FLAGS_NONE
106 }
107 }
108}
109
110/// Whether a flush covers the operations queued before it (M12.1, D-23 in
111/// `DESIGN-NOTES.md`).
112///
113/// This is a **required argument** on [`Batch::flush_owned`] and
114/// [`Batch::flush_raw`] rather than a field of [`PushOptions`], because there
115/// is no defensible default. An unflagged flush does *not* cover preceding
116/// writes: it is an ordinary operation competing with them, and it frequently
117/// wins. A default would make the obvious spelling -- push the writes, then
118/// push a flush -- a silent data-loss bug rather than a missing feature, so
119/// the decision is taken at the call site instead (D-25).
120///
121/// It is an enum rather than a `bool` for the same reason: `flush(&file,
122/// true)` does not say what the `true` decides, and this is not a parameter
123/// anyone should have to look up.
124///
125/// # The barrier's scope is the ring; the flush's is the file
126///
127/// One call sets both, which makes them easy to merge, and merging them is a
128/// mistake about durability rather than about style:
129///
130/// - **This flag is ring-wide.** [`Self::CoversPrecedingOperations`] means the
131/// flush does not execute until every operation outstanding *on the ring*
132/// when it was reached has **completed** -- whatever file each one targets,
133/// and whoever queued it (D-47, measured over roughly 4,500 trials).
134/// - **The flush names one file.** [`Batch::flush_owned`] takes a [`FileTarget`], so
135/// what a syncing [`FlushMode`] pushes to stable media is that file's data
136/// and the device cache behind it.
137///
138/// **Completion is not durability.** A write completing means the kernel took
139/// the bytes, not that they reached non-volatile media. So the barrier bounds
140/// what a flush **waits for**, and the flush itself bounds what is **made
141/// durable**, and those are different sets whenever a ring carries operations
142/// against more than one file.
143///
144/// The practical consequence for a caller is cost rather than correctness:
145/// operations this flush will never make durable can still make it wait. A
146/// caller who cares about that bounds it by controlling what shares the ring;
147/// this crate does not decide that (D-8), and the flush's own target is what
148/// secures the durability of the file named.
149#[derive(Clone, Copy, Debug, PartialEq, Eq)]
150pub enum FlushCoverage {
151 /// Wait for every operation already outstanding on the ring, then flush.
152 ///
153 /// This is what makes preceding writes durable when the flush's
154 /// completion is observed, and it is what a caller closing an epoch
155 /// wants.
156 ///
157 /// It sets `IOSQE_FLAGS_DRAIN_PRECEDING_OPS`, whose reach is **ring-wide**
158 /// rather than per-file: it waits on every operation outstanding on the
159 /// ring when it is reached, not only the ones in this batch. So the flush
160 /// takes as long as the slowest of them, however unrelated. That cost is
161 /// real, and it is the only reason the other variant exists.
162 ///
163 /// # What this does not promise
164 ///
165 /// **The flag is one-sided. Operations you push *after* this flush are not
166 /// held back, and can complete before it.** Measured over about 4,500
167 /// trials at a rate between 0.03% and 0.8%, and in the worst case every one
168 /// of 32 subsequent writes completed first (D-47).
169 ///
170 /// So this flush's completion tells you about the past and not the future:
171 /// everything outstanding when the flush was reached is durable, and
172 /// nothing is implied about work submitted afterwards. If later work must
173 /// not begin until the flush is durable, submit it after you have popped
174 /// the flush's completion. The ring will not sequence it for you.
175 ///
176 /// An earlier version of this documentation said subsequent operations were
177 /// held until the flush completed. That was wrong, and it is corrected here
178 /// rather than quietly dropped because a caller may have relied on it.
179 CoversPrecedingOperations,
180 /// Queue the flush with no barrier: it may complete while writes pushed
181 /// before it are still in flight.
182 ///
183 /// **Almost never what a caller wants.** Its completion proves nothing
184 /// about any preceding write, so using it to close an epoch loses data
185 /// that the caller believes is committed -- invisibly, until power is
186 /// lost.
187 ///
188 /// Two uses are legitimate. One is *host sequencing*: the caller has
189 /// already observed the completions of every write in the epoch before
190 /// pushing this, so the ordering is established outside the ring and the
191 /// barrier would only make this flush wait on operations it has already
192 /// accounted for. The other is a flush that is not being used for
193 /// durability at all.
194 Unordered,
195}
196
197impl FlushCoverage {
198 fn sqe_flags(self) -> IORING_SQE_FLAGS {
199 match self {
200 Self::CoversPrecedingOperations => IOSQE_FLAGS_DRAIN_PRECEDING_OPS,
201 Self::Unordered => IOSQE_FLAGS_NONE,
202 }
203 }
204}
205
206/// Where a write is allowed to stop: in the system cache, or past it (M12.3,
207/// D-25 in `DESIGN-NOTES.md`).
208///
209/// This is `BuildIoRingWriteFile`'s `FILE_WRITE_FLAGS` parameter, which the
210/// crate previously hardcoded to `FILE_WRITE_FLAGS_NONE` -- so a consumer
211/// reading this API saw ordering but no way to express anything about caching
212/// at all, and could reasonably conclude the ring did not expose it. It does;
213/// the crate was narrowing the platform to what its own examples needed.
214///
215/// # What write-through is, and what it is not
216///
217/// Worth stating precisely, because conflating these produced a wrong
218/// recommendation in the exchange that prompted this work.
219///
220/// It **is** a first-level cache directive: it tells the OS not to leave the
221/// data sitting in the system cache. Its practical value is latency shaping --
222/// data already at the device makes a later flush shorter, which can matter
223/// when the flush is on a commit path.
224///
225/// It is **not** a durability guarantee, and it is **not** FUA. Whether it
226/// becomes a Force Unit Access bit on the underlying command depends on the
227/// driver, the volume, and whether the device's write cache is enabled -- none
228/// of which this API can see or promise. A write that completes with
229/// [`WriteCaching::WriteThrough`] may still be sitting in a volatile device
230/// cache, and will be lost on power failure like any other.
231///
232/// **The flush operation is the only durability primitive the ring has**, and
233/// only with [`FlushCoverage::CoversPrecedingOperations`]. Reaching for
234/// write-through instead of a covering flush is the mistake this doc exists to
235/// prevent; reaching for it *as well* is a legitimate latency optimization.
236#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
237pub enum WriteCaching {
238 /// `FILE_WRITE_FLAGS_NONE`: the write may be satisfied into the system
239 /// cache, and reaching the device is a later flush's job.
240 ///
241 /// The default, and the right choice for the streaming writes of an epoch
242 /// that a covering flush will close.
243 #[default]
244 Cached,
245 /// `FILE_WRITE_FLAGS_WRITE_THROUGH`: ask the OS not to leave the data in
246 /// the system cache.
247 ///
248 /// A latency-shaping knob, not a durability marker -- see this type's
249 /// docs. It does not remove the need for a flush.
250 WriteThrough,
251}
252
253impl WriteCaching {
254 fn raw(self) -> FILE_WRITE_FLAGS {
255 match self {
256 Self::Cached => FILE_WRITE_FLAGS_NONE,
257 Self::WriteThrough => FILE_WRITE_FLAGS_WRITE_THROUGH,
258 }
259 }
260}
261
262/// How much a flush is asked to push, and whether it syncs the device at all
263/// (M12.4, D-25 in `DESIGN-NOTES.md`).
264///
265/// This is `BuildIoRingFlushFile`'s `FILE_FLUSH_MODE` parameter, which the
266/// crate previously hardcoded to `FILE_FLUSH_DEFAULT`. Exposing it is the
267/// other half of D-25: the kernel offers a durability parameter here and a
268/// wrapper that hides it narrows the platform to what its own examples needed.
269///
270/// [`FlushMode::Default`] is what a caller closing an epoch wants, and is
271/// what the crate did before this existed.
272///
273/// # `NoSync` is not a durability operation
274///
275/// Three of these four modes issue the device sync; [`FlushMode::NoSync`] does
276/// not. That is worth stating rather than inferring, and the inference runs
277/// the other way too: **the existence of a distinct "no sync" mode is the
278/// evidence that the other three do sync.** Nothing in the Win32
279/// documentation says so directly.
280#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
281pub enum FlushMode {
282 /// `FILE_FLUSH_DEFAULT`: flush the file's data and metadata, and sync the
283 /// device.
284 ///
285 /// The default, and the mode a durability barrier wants. Pair it with
286 /// [`FlushCoverage::CoversPrecedingOperations`] or it covers nothing.
287 #[default]
288 Default,
289 /// `FILE_FLUSH_DATA`: flush the file's data, and sync the device.
290 ///
291 /// Metadata that is not required to read the data back may be left
292 /// unflushed. The analogue of `fdatasync`, and the usual choice for a log
293 /// whose records are self-describing.
294 Data,
295 /// `FILE_FLUSH_MIN_METADATA`: flush the data plus the minimum metadata
296 /// needed to retrieve it, and sync the device.
297 MinMetadata,
298 /// `FILE_FLUSH_NO_SYNC`: flush to the device **without** issuing the sync.
299 ///
300 /// **This is the one mode that makes nothing durable.** It pushes data out
301 /// of the system cache and stops there, so anything still held in a
302 /// volatile device cache is lost on power failure exactly as if no flush
303 /// had been issued at all. A completion from this mode is not a commit
304 /// point and must never be reported to a caller as one.
305 ///
306 /// It is useful for shaping when cache pressure is paid -- moving data
307 /// deviceward early so a later real flush is short -- which is the same
308 /// role [`WriteCaching::WriteThrough`] plays on the write side, and it
309 /// carries the same warning: a latency knob, not a durability primitive.
310 NoSync,
311}
312
313impl FlushMode {
314 fn raw(self) -> FILE_FLUSH_MODE {
315 match self {
316 Self::Default => FILE_FLUSH_DEFAULT,
317 Self::Data => FILE_FLUSH_DATA,
318 Self::MinMetadata => FILE_FLUSH_MIN_METADATA,
319 Self::NoSync => FILE_FLUSH_NO_SYNC,
320 }
321 }
322}
323
324/// Build the raw handle reference for `file`, refusing a [`RegisteredFile`]
325/// that did not come from `ring_id`'s own ring (PR #20 review response): the
326/// index alone names a slot in *some* ring's file table, and a slot from a
327/// different ring can legitimately hold an entirely different file, so
328/// crossing rings must be an explicit, reported error rather than silently
329/// addressing the wrong handle.
330fn handle_ref(file: FileRef, ring_id: RingId) -> io::Result<IORING_HANDLE_REF> {
331 match file {
332 FileRef::Raw(handle) => Ok(IORING_HANDLE_REF {
333 Kind: IORING_REF_RAW,
334 Handle: IORING_HANDLE_REF_0 { Handle: handle },
335 }),
336 FileRef::Registered(file) => {
337 if file.ring_id != ring_id {
338 return Err(io::Error::new(
339 io::ErrorKind::InvalidInput,
340 "this RegisteredFile was registered on a different IoRing",
341 ));
342 }
343 Ok(IORING_HANDLE_REF {
344 Kind: IORING_REF_REGISTERED,
345 Handle: IORING_HANDLE_REF_0 {
346 Index: file.index(),
347 },
348 })
349 }
350 }
351}
352
353fn raw_buffer_ref(address: *mut c_void) -> IORING_BUFFER_REF {
354 IORING_BUFFER_REF {
355 Kind: IORING_REF_RAW,
356 Buffer: IORING_BUFFER_REF_0 { Address: address },
357 }
358}
359
360fn registered_buffer_ref(index: u32, offset: u32) -> IORING_BUFFER_REF {
361 IORING_BUFFER_REF {
362 Kind: IORING_REF_REGISTERED,
363 Buffer: IORING_BUFFER_REF_0 {
364 IndexAndOffset: IORING_REGISTERED_BUFFER {
365 BufferIndex: index,
366 Offset: offset,
367 },
368 },
369 }
370}
371
372fn checked_len(len: usize) -> io::Result<u32> {
373 u32::try_from(len).map_err(|_| {
374 io::Error::new(
375 io::ErrorKind::InvalidInput,
376 format!("an IoRing buffer is limited to u32::MAX bytes; {len} does not fit"),
377 )
378 })
379}
380
381/// A file, addressed either by a raw `HANDLE` or by the index a prior
382/// [`Batch::register_files`] assigned it (M5.1).
383///
384/// This enum is the `_raw` pushes' addressing parameter. A safe push takes a
385/// [`FileTarget`] instead -- a [`SharedHandle`] or a [`RegisteredFile`] -- so
386/// reaching a registered file has not required `unsafe` since M10.4 (D-29,
387/// D-33). Constructing a `FileRef` directly is only necessary for the `_raw`
388/// pushes, whose reason to exist is a bare `HANDLE` this crate cannot keep
389/// alive for you.
390///
391/// A distinct type per addressing mode -- rather than one method accepting
392/// either a `HANDLE` or a bare `u32` -- is what makes "read against a
393/// registered file" and "read against an unregistered one" impossible to
394/// confuse by accident: a [`RegisteredFile`] cannot be mistaken for a raw
395/// `HANDLE`, or vice versa, because they are different types.
396#[derive(Clone, Copy, Debug)]
397pub enum FileRef {
398 /// `IORING_REF_RAW`: the caller's own open handle.
399 Raw(HANDLE),
400 /// `IORING_REF_REGISTERED`: an index from [`Batch::register_files`].
401 Registered(RegisteredFile),
402}
403
404impl From<HANDLE> for FileRef {
405 fn from(handle: HANDLE) -> Self {
406 FileRef::Raw(handle)
407 }
408}
409
410impl From<RegisteredFile> for FileRef {
411 fn from(file: RegisteredFile) -> Self {
412 FileRef::Registered(file)
413 }
414}
415
416/// One index a [`Batch::register_files`] registration assigned (M5.1).
417#[derive(Clone, Copy, Debug, PartialEq, Eq)]
418pub struct RegisteredFile {
419 index: u32,
420 ring_id: RingId,
421}
422
423impl RegisteredFile {
424 /// The raw registered-file index.
425 #[must_use]
426 pub fn index(self) -> u32 {
427 self.index
428 }
429}
430
431mod sealed {
432 pub trait Sealed {}
433 impl Sealed for super::SharedHandle {}
434 impl Sealed for super::RegisteredFile {}
435}
436
437/// A file a *safe* push can address: either a [`SharedHandle`] or a
438/// [`RegisteredFile`] (M10.4; D-29, D-33 and D-82 in `DESIGN-NOTES.md`).
439///
440/// This exists because the two carry different lifetime obligations, and
441/// exactly one of them has none. A `SharedHandle` is a raw `HANDLE` underneath,
442/// so something must keep it open until the kernel is finished; a
443/// `RegisteredFile` is an index into a table the ring itself owns, minted by
444/// this crate and checked against the minting ring, so there is nothing for a
445/// caller to keep alive per push. (The handles behind that table are kept
446/// open for the ring's life either by [`Batch::register_files`]' caller, under
447/// its `unsafe` contract, or by the ring itself after
448/// [`Batch::register_shared_files`].) Both are therefore safe to push, and neither needs
449/// `unsafe` -- which is why [`Batch::read_owned`] and its siblings are generic over
450/// this trait rather than hardcoding `SharedHandle`.
451///
452/// [`Guard`](FileTarget::Guard) is what the ring holds until the
453/// operation's completion is observed. For `SharedHandle` that is a clone,
454/// which shares the handle and so makes it outlive the operation; for
455/// `RegisteredFile` there is nothing to keep alive, so it is the (`Copy`)
456/// index itself, handed back for symmetry rather than out of necessity.
457///
458/// **Sealed**: this crate implements it for exactly those two types and no
459/// others. An outside implementation could return an arbitrary raw `HANDLE`
460/// from [`as_file_ref`](FileTarget::as_file_ref) with no guard keeping it
461/// alive, which is precisely the unsoundness the `_raw` pushes make a caller
462/// take responsibility for with `unsafe`.
463pub trait FileTarget: sealed::Sealed {
464 /// What the ring must hold for the operation until its completion is
465 /// observed.
466 /// What keeps the file valid while the operation runs.
467 ///
468 /// `Into<FileGuard>` is expressible only because this trait is **sealed**
469 /// (`D-73`): both implementors are this crate's, so the ring can hold a
470 /// guard in a concrete slot instead of a third generic parameter every
471 /// consumer would have to name.
472 type Guard: Send + 'static + Into<crate::ring::FileGuard>;
473
474 /// How this target addresses its file.
475 fn as_file_ref(&self) -> FileRef;
476
477 /// Produce the value the ring will hold for the operation's
478 /// duration.
479 fn guard(&self) -> Self::Guard;
480}
481
482/// A [`SharedHandle`] is how a caller gives this crate a file it does not have
483/// to prove outlives every operation pushed against it (M8; D-82). One handle is
484/// legitimately the target of many concurrent pushes, so what must survive until
485/// every one of them completes is a *reference*, not sole ownership. Every safe
486/// push (e.g. [`Batch::read_owned`], as opposed to its `_raw` sibling) clones it
487/// into the same inventory entry that tracks the operation's own payload, so the
488/// handle survives until the ring returns that entry at its pop (D-4), whatever
489/// the caller does with its own clone.
490impl FileTarget for SharedHandle {
491 type Guard = SharedHandle;
492
493 fn as_file_ref(&self) -> FileRef {
494 FileRef::Raw(self.as_raw_handle())
495 }
496
497 fn guard(&self) -> Self::Guard {
498 self.clone()
499 }
500}
501
502impl FileTarget for RegisteredFile {
503 type Guard = RegisteredFile;
504
505 fn as_file_ref(&self) -> FileRef {
506 FileRef::Registered(*self)
507 }
508
509 fn guard(&self) -> Self::Guard {
510 *self
511 }
512}
513
514/// The confirmed result of a [`Batch::register_files`] push: `count`
515/// contiguous indices starting at `base_index`.
516///
517/// `base_index` is currently always zero, because a ring accepts at most one
518/// registration that assigns an index -- do not read its presence as evidence
519/// that several registrations compose. It is kept rather than folded away
520/// because it is the correct shape if that rule is ever relaxed (M10.3, D-31).
521#[derive(Clone, Copy, Debug, PartialEq, Eq)]
522pub struct RegisteredFiles {
523 base_index: u32,
524 count: u32,
525 ring_id: RingId,
526}
527
528impl RegisteredFiles {
529 /// How many handles this registration covers.
530 #[must_use]
531 pub fn len(&self) -> u32 {
532 self.count
533 }
534
535 /// Whether this registration covers no handles.
536 #[must_use]
537 pub fn is_empty(&self) -> bool {
538 self.count == 0
539 }
540
541 /// The [`RegisteredFile`] for the `i`-th handle passed to
542 /// [`Batch::register_files`], or `None` if `i` is out of range.
543 #[must_use]
544 pub fn get(&self, i: u32) -> Option<RegisteredFile> {
545 (i < self.count).then(|| RegisteredFile {
546 index: self.base_index + i,
547 ring_id: self.ring_id,
548 })
549 }
550}
551
552/// A [`Batch::register_files`] push not yet matched to its completion.
553#[derive(Debug)]
554pub struct PendingFileRegistration {
555 user_data: usize,
556 base_index: u32,
557 count: u32,
558 ring_id: RingId,
559}
560
561impl PendingFileRegistration {
562 /// This push's `UserData` identity, to match against a popped
563 /// [`Completion`].
564 #[must_use]
565 pub fn user_data(&self) -> usize {
566 self.user_data
567 }
568
569 /// Turn this pending registration into usable [`RegisteredFiles`] once
570 /// `completion` names it, or hand it back unchanged if `completion`
571 /// names a different operation.
572 ///
573 /// # Errors
574 ///
575 /// The inner `Result` is `Err` if the registration itself failed --
576 /// nothing is lost, since no owned resource was ever handed over for
577 /// this op (M5.1, unlike an inventory entry).
578 pub fn claim_if(self, completion: &Completion) -> Result<io::Result<RegisteredFiles>, Self> {
579 if completion.user_data() != self.user_data || completion.ring_id() != self.ring_id {
580 return Err(self);
581 }
582 Ok(completion.result().map(|_| RegisteredFiles {
583 base_index: self.base_index,
584 count: self.count,
585 ring_id: self.ring_id,
586 }))
587 }
588}
589
590/// Per-buffer state a registration keeps for its whole life.
591///
592/// The registered length is captured at registration time and never
593/// recomputed. That matters because `IoBuf::bytes_len` reports the buffer's
594/// *current* length, and what the kernel holds is the length it was given --
595/// validating a span against the former would let a buffer that had grown
596/// address memory the kernel does not know about.
597struct BufferState {
598 /// Operations outstanding against this one buffer, in either direction.
599 outstanding: AtomicUsize,
600 /// Of those, the ones the kernel **writes into** this buffer -- that is,
601 /// reads. Always a subset of `outstanding`.
602 ///
603 /// Tracked separately because the two hazards are not the same. Mutating a
604 /// buffer races the kernel whichever way the kernel is touching it, so
605 /// [`RegisteredBuffers::get_mut`] refuses on `outstanding`. *Reading* a
606 /// buffer only races a kernel that is writing into it: a shared read
607 /// alongside the kernel's own read of an in-flight write is not a race at
608 /// all, and refusing it would deny a sound operation.
609 kernel_writes: AtomicUsize,
610 /// The length handed to `BuildIoRingRegisterBuffers`, in bytes.
611 registered_len: u32,
612}
613
614/// Which way the kernel touches a registered buffer for the life of an
615/// operation.
616///
617/// Named from the kernel's point of view rather than the operation's, because
618/// that is the direction the data-race question actually turns on, and
619/// "a read means the kernel writes" is exactly the inversion that makes this
620/// easy to get backwards.
621#[derive(Clone, Copy, PartialEq, Eq)]
622enum KernelAccess {
623 /// A read: the kernel writes into the buffer.
624 WritesBuffer,
625 /// A write: the kernel reads out of the buffer.
626 ReadsBuffer,
627}
628
629/// A marker the ring holds while a registered-buffer-indexed read or write
630/// is outstanding (M5.2, M5.3).
631///
632/// Decrements [`RegisteredBuffers`]'s count **for the one buffer the
633/// operation addressed** only when the ring returns the entry: an entry the
634/// ring forgets rather than returns takes this with it, which correctly
635/// leaves the registration believing that use is still outstanding, since
636/// nothing proved otherwise.
637///
638/// The count is per buffer rather than per registration (M13.2) so that
639/// [`RegisteredBuffers::get_mut`] can hand out a `&mut` to a *quiet* buffer
640/// while operations are still in flight against its neighbours -- which is
641/// what an arena a caller refills has to be able to do.
642pub struct RegisteredUse {
643 state: Arc<[BufferState]>,
644 index: usize,
645 access: KernelAccess,
646}
647
648impl Drop for RegisteredUse {
649 fn drop(&mut self) {
650 if let Some(state) = self.state.get(self.index) {
651 if self.access == KernelAccess::WritesBuffer {
652 state.kernel_writes.fetch_sub(1, Ordering::SeqCst);
653 }
654 state.outstanding.fetch_sub(1, Ordering::SeqCst);
655 }
656 }
657}
658
659/// Buffers registered with a ring so later reads and writes can address them
660/// by index instead of handing over a fresh owned buffer each call (M5.2).
661///
662/// Win32's `IoRing` has no unregister call at all: once registered, an
663/// index is valid for the ring's remaining life. This type's own job is
664/// narrower than reproducing that -- refusing to let its *backing memory*
665/// be freed while this crate still has an operation outstanding against it
666/// (M5.3) -- not proving an index will never be touched again by some
667/// later, differently-built SQE. That wider hazard belongs to
668/// [`crate::IoRing::push_raw`]'s existing `unsafe` contract, the same as any
669/// raw use of an index this crate handed out.
670pub struct RegisteredBuffers<B: IoBufMut> {
671 buffers: ManuallyDrop<Vec<B>>,
672 base_index: u32,
673 /// One entry per registered buffer, indexed by position within this
674 /// registration.
675 state: Arc<[BufferState]>,
676 ring_id: RingId,
677}
678
679impl<B: IoBufMut> RegisteredBuffers<B> {
680 /// How many buffers this registration holds.
681 #[must_use]
682 pub fn len(&self) -> u32 {
683 u32::try_from(self.buffers.len()).unwrap_or(u32::MAX)
684 }
685
686 /// Whether this registration holds no buffers.
687 ///
688 /// Always `false` in practice, and a mutation run reports the constant as
689 /// surviving for that reason rather than for want of a test. Nothing in
690 /// this crate rejects an empty vector, but the kernel refuses the
691 /// submission with `E_INVALIDARG`, so a caller never holds an empty
692 /// registration. `windows_refuses_an_empty_buffer_registration` pins that
693 /// platform behaviour; if a future Windows accepts it, that test fails and
694 /// this becomes reachable.
695 ///
696 /// Kept because it is half of the `len`/`is_empty` pair every Rust
697 /// collection surface offers, and because "cannot happen today" is a
698 /// weaker claim than "cannot happen".
699 #[must_use]
700 pub fn is_empty(&self) -> bool {
701 self.buffers.is_empty()
702 }
703
704 /// Borrow the `i`-th buffer's registered bytes, for reading a completed
705 /// read's or write's bytes without going through the ring at all.
706 ///
707 /// # Why it can refuse
708 ///
709 /// Because otherwise it is a data race with the kernel, reachable from
710 /// entirely safe code. [`Batch::read_registered_owned`] takes
711 /// `&RegisteredBuffers` and holds only a
712 /// [`RegisteredUse`] -- no borrow of this registration outlives the push.
713 /// So without the check below, this sequence compiles with no `unsafe`
714 /// anywhere and reads memory the kernel is concurrently writing:
715 ///
716 /// ```ignore
717 /// batch.read_registered_owned(&file, &arena, span, (), 0, PushOptions::new())?;
718 /// batch.submit()?; // kernel now filling that buffer
719 /// let bytes = arena.get(span.buffer_index)?;
720 /// ```
721 ///
722 /// The reverse order is the one the check alone could not stop, and is why
723 /// this takes `&mut self` -- see below:
724 ///
725 /// ```ignore
726 /// let bytes = arena.get(0)?; // quiet here, so the check passes
727 /// batch.read_registered_owned(&file, &arena, span, (), 0, PushOptions::new())?;
728 /// batch.submit()?; // kernel now filling that same buffer
729 /// let stale = bytes[0]; // ... read through the live borrow
730 /// ```
731 ///
732 /// # Why this takes `&mut self` to hand back a shared slice
733 ///
734 /// The refusal above is a check at one instant, but the slice it returns
735 /// lives as long as the borrow. Taking `&self` made those two different
736 /// lengths of time, and the gap was reachable: `read_registered` takes the
737 /// registration by *shared* reference, so a caller could take the borrow
738 /// while the buffer was quiet, then push and submit a read into that same
739 /// buffer and keep reading through the still-live slice. Measured, not
740 /// argued -- a probe watched the bytes change under a live `&[u8]` while a
741 /// fresh `get` at that same instant correctly refused.
742 ///
743 /// `&mut self` closes it by making the *borrow* conflict with starting the
744 /// operation, not merely its creation: `read_registered(&arena, ..)` cannot
745 /// be called while this slice is alive. That costs nothing real, because
746 /// **no caller needs to read a buffer during the window it is refused**.
747 /// While a read is in flight the bytes are indeterminate -- partially
748 /// written, in arbitrary order -- and only become meaningful once the
749 /// completion is observed. Earlier or later is always available.
750 ///
751 /// It does not restrict an arena. An entry holds a [`RegisteredUse`],
752 /// not a borrow of this registration, so the shared borrow ends when the
753 /// push returns; a caller may still read or fill *quiet* buffers while
754 /// operations are outstanding against their neighbours, which is what makes
755 /// a registered arena usable at all.
756 ///
757 /// # Why it refuses less than [`get_mut`] does
758 ///
759 /// `get_mut` refuses while *any* operation is outstanding, because
760 /// mutating races the kernel whichever way the kernel is touching the
761 /// buffer. Reading is narrower: it races only a kernel that is **writing
762 /// into** the buffer, so this refuses only while a *read* is outstanding --
763 /// the bytes of a write in flight are stable, and the kernel is reading
764 /// them too.
765 ///
766 /// A count returns to zero only when a completion is popped against a real
767 /// popped [`Completion`], so "not outstanding" means the operation was
768 /// *observed* to finish, never merely assumed to have.
769 ///
770 /// # Why this hands back `&[u8]` and not `&B`
771 ///
772 /// For the same reason [`get_mut`] does: the slice is exactly the length
773 /// handed to `BuildIoRingRegisterBuffers`, which is what the kernel knows
774 /// about. `B`'s own `bytes_len()` is a live value that may since have
775 /// diverged from it, so answering from the registered length keeps this in
776 /// step with the registration rather than with the buffer.
777 ///
778 /// [`get_mut`]: RegisteredBuffers::get_mut
779 ///
780 /// # Holding the borrow across a push is refused at compile time
781 ///
782 /// This is the regression guard for the hazard described above. Taking the
783 /// borrow while the buffer is quiet and *then* submitting a read into it
784 /// must not compile, and the failure it exists for is the borrow conflict,
785 /// `E0502`. That code is pinned below, with a limit worth knowing: rustdoc
786 /// checks a pinned code only on a nightly toolchain (or under
787 /// `RUSTC_BOOTSTRAP=1`, which CI's `doctest-error-codes` job sets), and on
788 /// stable a `compile_fail` example passes on any error. This one went on "passing" after `read_registered` was
789 /// removed, because calling a method that does not exist fails to compile
790 /// too; the `no_run` twin below, which must compile, is what catches a
791 /// rename that reaches both:
792 ///
793 /// ```compile_fail,E0502
794 /// # use windows_ioring_sys::{Batch, IoRing, PushOptions, RegisteredBuffers, RegisteredSpan, SharedHandle};
795 /// # fn hazard<B: windows_ioring_sys::IoBufMut>(
796 /// # ring: &mut IoRing,
797 /// # arena: &mut RegisteredBuffers<B>,
798 /// # file: &SharedHandle,
799 /// # span: RegisteredSpan,
800 /// # ) -> std::io::Result<u8> {
801 /// let bytes: &[u8] = arena.get(0)?; // quiet here, so the check passes
802 /// let mut batch = Batch::new(ring);
803 /// batch.read_registered_owned(file, arena, span, (), 0, PushOptions::new())?;
804 /// batch.submit()?; // kernel now filling that buffer
805 /// Ok(bytes[0]) // ... read through the live borrow
806 /// # }
807 /// ```
808 ///
809 /// Reading a *quiet* buffer while an operation is outstanding against a
810 /// neighbour is still allowed, because the ring's hold on a registration
811 /// is per-buffer -- that is the arena pattern, and it must keep working:
812 ///
813 /// ```no_run
814 /// # use windows_ioring_sys::{Batch, IoRing, PushOptions, RegisteredBuffers, RegisteredSpan, SharedHandle};
815 /// # fn arena_still_works<B: windows_ioring_sys::IoBufMut>(
816 /// # ring: &mut IoRing,
817 /// # arena: &mut RegisteredBuffers<B>,
818 /// # file: &SharedHandle,
819 /// # span: RegisteredSpan,
820 /// # ) -> std::io::Result<()> {
821 /// let mut batch = Batch::new(ring);
822 /// batch.read_registered_owned(file, arena, span, (), 0, PushOptions::new())?;
823 /// batch.submit()?;
824 /// let _neighbour = arena.get(1)?; // a different, quiet buffer
825 /// # Ok(())
826 /// # }
827 /// ```
828 ///
829 /// # Errors
830 ///
831 /// [`io::ErrorKind::InvalidInput`] if `i` is out of range for this
832 /// registration, or [`io::ErrorKind::WouldBlock`] if a read is still
833 /// outstanding into that buffer -- pop its completion first, which is what
834 /// returns the buffer.
835 pub fn get(&mut self, i: u32) -> io::Result<&[u8]> {
836 let index = usize::try_from(i).map_err(|_| {
837 io::Error::new(
838 io::ErrorKind::InvalidInput,
839 "buffer index does not fit usize",
840 )
841 })?;
842 let Some(state) = self.state.get(index) else {
843 return Err(io::Error::new(
844 io::ErrorKind::InvalidInput,
845 format!("buffer index {i} is out of range for this registration"),
846 ));
847 };
848 let kernel_writes = state.kernel_writes.load(Ordering::SeqCst);
849 if kernel_writes > 0 {
850 return Err(io::Error::new(
851 io::ErrorKind::WouldBlock,
852 format!(
853 "buffer {i} still has {kernel_writes} read(s) outstanding into it; the kernel \
854 may be writing through it. Pop its completion before reading this buffer."
855 ),
856 ));
857 }
858 let len = state.registered_len as usize;
859 let buffer = self
860 .buffers
861 .get(index)
862 .expect("state and buffers are built together and have equal length");
863 // SAFETY: `stable_ptr` is `IoBuf`'s promised stable address for this
864 // buffer, and `registered_len` was recorded from that same buffer at
865 // registration, so those bytes are allocated and initialized. `&mut
866 // self` excludes concurrent mutation through `get_mut` *and* -- because
867 // the returned slice borrows it for its whole life -- excludes pushing
868 // a new operation against this registration while the slice is alive,
869 // since `read_registered` needs `&RegisteredBuffers`. So the
870 // zero-`kernel_writes` check above cannot be outrun: no read into this
871 // buffer can start before the borrow ends.
872 Ok(unsafe { std::slice::from_raw_parts(buffer.stable_ptr(), len) })
873 }
874
875 /// Borrow the `i`-th buffer's registered bytes **mutably**, to fill them
876 /// before an operation reads out of them (M13.2).
877 ///
878 /// This is what makes a registered arena usable for writes a caller
879 /// composes itself -- a log record, a serialized message -- rather than
880 /// only for bytes the kernel put there. Without it, [`get`] can show a
881 /// caller what a read landed but nothing can put anything into a buffer
882 /// after it is registered.
883 ///
884 /// # Why this hands back `&mut [u8]` and not `&mut B`
885 ///
886 /// Because `&mut B` would be **unsound**, and quietly so. Win32's
887 /// `IoRing` has no unregister call: the address and length handed to
888 /// `BuildIoRingRegisterBuffers` stay live for the ring's remaining life.
889 /// A `&mut Vec<u8>` lets safe code assign a whole new vector, `reserve`,
890 /// or `resize` -- each of which frees or moves the very allocation the
891 /// kernel still holds the address of. No `unsafe` is involved, and it
892 /// happens at a moment when nothing is outstanding, so no runtime check
893 /// would catch it. [`IoBuf`]'s contract requires that address to stay put
894 /// for as long as this crate holds the value, and a registration holds it
895 /// forever.
896 ///
897 /// A byte slice of exactly the registered length grants the one power a
898 /// caller actually needs -- writing the bytes -- and none of the powers
899 /// that would break the registration.
900 ///
901 /// # Why it can refuse
902 ///
903 /// `&mut self` is not on its own enough either. It excludes concurrent
904 /// `&self` borrows, but an operation already in flight holds no borrow at
905 /// all -- [`Batch::write_registered_owned`] takes `&RegisteredBuffers` for the
906 /// length of the call, and the entry it creates keeps only a
907 /// [`RegisteredUse`]. So the compiler would happily allow mutation of a
908 /// buffer the kernel is reading through right now, which is a data race
909 /// with the kernel.
910 ///
911 /// The runtime half of the check is therefore the per-buffer outstanding
912 /// count: this refuses while any operation is still outstanding against
913 /// *this* buffer. Neighbouring buffers being in flight is fine, which is
914 /// the point of counting per buffer -- an arena a caller refills has to
915 /// be able to prepare buffer 5 while buffer 2 is still being written.
916 ///
917 /// A count returns to zero only when a completion is popped against a
918 /// real popped [`Completion`], so "not outstanding" means the operation
919 /// was *observed* to finish, never merely assumed to have.
920 ///
921 /// [`get`]: RegisteredBuffers::get
922 ///
923 /// # Errors
924 ///
925 /// [`io::ErrorKind::InvalidInput`] if `i` is out of range for this
926 /// registration, or [`io::ErrorKind::WouldBlock`] if an operation is
927 /// still outstanding against that buffer -- pop its completion first,
928 /// which is what returns the buffer.
929 pub fn get_mut(&mut self, i: u32) -> io::Result<&mut [u8]> {
930 let index = usize::try_from(i).map_err(|_| {
931 io::Error::new(
932 io::ErrorKind::InvalidInput,
933 "buffer index does not fit usize",
934 )
935 })?;
936 let Some(state) = self.state.get(index) else {
937 return Err(io::Error::new(
938 io::ErrorKind::InvalidInput,
939 format!("buffer index {i} is out of range for this registration"),
940 ));
941 };
942 let outstanding = state.outstanding.load(Ordering::SeqCst);
943 if outstanding > 0 {
944 return Err(io::Error::new(
945 io::ErrorKind::WouldBlock,
946 format!(
947 "buffer {i} still has {outstanding} operation(s) outstanding; the kernel may \
948 be reading or writing through it. Pop its completion before filling this \
949 buffer."
950 ),
951 ));
952 }
953 let len = state.registered_len as usize;
954 let buffer = self
955 .buffers
956 .get_mut(index)
957 .expect("state and buffers are built together and have equal length");
958 // SAFETY: `stable_mut_ptr` is `IoBufMut`'s promised stable, uniquely
959 // owned address for this buffer, and `registered_len` was recorded
960 // from that same buffer at registration, so those bytes are allocated
961 // and initialized. `&mut self` plus the zero-outstanding check above
962 // establish that nothing else -- kernel included -- is reading or
963 // writing them for this borrow's life.
964 Ok(unsafe { std::slice::from_raw_parts_mut(buffer.stable_mut_ptr(), len) })
965 }
966
967 /// How many operations are still outstanding against the `i`-th buffer,
968 /// or `None` if `i` is out of range.
969 ///
970 /// Lets a caller pick a quiet buffer without provoking an error from
971 /// [`RegisteredBuffers::get_mut`], which is the normal shape for an arena
972 /// cycling through its slots.
973 #[must_use]
974 pub fn outstanding(&self, i: u32) -> Option<usize> {
975 self.state
976 .get(usize::try_from(i).ok()?)
977 .map(|state| state.outstanding.load(Ordering::SeqCst))
978 }
979
980 /// Whether `span` lies within this registration -- its buffer index in
981 /// range, and `span.offset .. span.offset + span.len` within that buffer's
982 /// registered length -- asked without pushing anything. It is the check
983 /// every push against a span makes before the kernel sees it, so the
984 /// answer is the push's (M34.1).
985 ///
986 /// For a caller that accepts an operation now and pushes it later, so it
987 /// can refuse a span the push would refuse at the moment it accepts it,
988 /// rather than at a push nobody is waiting on.
989 ///
990 /// # Errors
991 ///
992 /// The [`io::ErrorKind::InvalidInput`] error a push would refuse `span`
993 /// with.
994 pub fn check_span(&self, span: RegisteredSpan) -> io::Result<()> {
995 self.checked_span(span).map(|_| ())
996 }
997
998 /// Record that one more operation is outstanding against `span`'s buffer,
999 /// and produce the marker whose drop records it finished.
1000 ///
1001 /// Call only after [`RegisteredBuffers::checked_span`] has validated
1002 /// `span`, which is what makes the index below in range.
1003 fn begin_use(&self, span: RegisteredSpan, access: KernelAccess) -> RegisteredUse {
1004 let index = span.buffer_index as usize;
1005 if let Some(state) = self.state.get(index) {
1006 // The direction-specific counter first, so a concurrent `get` can
1007 // never observe `outstanding` raised without also seeing the
1008 // reason it was raised.
1009 if access == KernelAccess::WritesBuffer {
1010 state.kernel_writes.fetch_add(1, Ordering::SeqCst);
1011 }
1012 state.outstanding.fetch_add(1, Ordering::SeqCst);
1013 }
1014 RegisteredUse {
1015 state: Arc::clone(&self.state),
1016 index,
1017 access,
1018 }
1019 }
1020
1021 /// The registered index of the `i`-th buffer, or an
1022 /// [`io::ErrorKind::InvalidInput`] error if out of range.
1023 fn checked_index(&self, i: u32) -> io::Result<u32> {
1024 if i < self.len() {
1025 Ok(self.base_index + i)
1026 } else {
1027 Err(io::Error::new(
1028 io::ErrorKind::InvalidInput,
1029 format!("buffer index {i} is out of range for this registration"),
1030 ))
1031 }
1032 }
1033
1034 /// As [`RegisteredBuffers::checked_index`], plus checking that
1035 /// `span.offset .. span.offset + span.len` actually fits inside that
1036 /// buffer -- validating only the index and handing an unchecked range to
1037 /// the kernel would let it read or write outside the registered
1038 /// allocation.
1039 ///
1040 /// Checked against the length recorded **at registration**, never against
1041 /// the buffer's current `bytes_len()`. Those are the same number today,
1042 /// because nothing this type exposes can resize a registered buffer, but
1043 /// the kernel's view is the registered one and that is what a span has to
1044 /// fit inside. Validating against a live length would mean a buffer that
1045 /// somehow grew could address memory the kernel does not know about.
1046 fn checked_span(&self, span: RegisteredSpan) -> io::Result<u32> {
1047 let index = self.checked_index(span.buffer_index)?;
1048 let registered_len = self
1049 .state
1050 .get(span.buffer_index as usize)
1051 .expect("checked_index just validated this index")
1052 .registered_len;
1053 let fits = u64::from(span.offset)
1054 .checked_add(u64::from(span.len))
1055 .is_some_and(|end| end <= u64::from(registered_len));
1056 if !fits {
1057 return Err(io::Error::new(
1058 io::ErrorKind::InvalidInput,
1059 format!(
1060 "span offset {} + len {} does not fit inside buffer {} \
1061 ({registered_len} registered bytes)",
1062 span.offset, span.len, span.buffer_index
1063 ),
1064 ));
1065 }
1066 Ok(index)
1067 }
1068}
1069
1070impl<B: IoBufMut> Drop for RegisteredBuffers<B> {
1071 fn drop(&mut self) {
1072 if self
1073 .state
1074 .iter()
1075 .any(|state| state.outstanding.load(Ordering::SeqCst) > 0)
1076 {
1077 // Refused, not silently permitted (M5.3): freeing now would
1078 // leave an outstanding `IORING_BUFFER_REF` pointing at freed
1079 // memory. Loud in debug builds; in release, leaking is the safe
1080 // failure mode -- the same choice the ring makes for its inventory
1081 // when rundown cannot prove quiescence (D-73): "leak is safe,
1082 // use-after-free is not". So `buffers` is
1083 // simply never reclaimed rather than freed out from under a
1084 // still-outstanding op. Any one buffer still in use holds the
1085 // whole registration, because they are freed together.
1086 //
1087 // Silent while already panicking (M23.4). A second panic during
1088 // unwind aborts the process, which *replaces* the failure that
1089 // started the unwind: a test asserting on a leaked slot reports
1090 // `STATUS_STACK_BUFFER_OVERRUN` instead of its own message, so
1091 // the diagnosis is destroyed by the guard meant to aid it. The
1092 // leak is still refused -- the early `return` below is what makes
1093 // leaking rather than freeing the failure mode, and it happens
1094 // either way.
1095 debug_assert!(
1096 std::thread::panicking(),
1097 "RegisteredBuffers dropped while an operation still references it"
1098 );
1099 return;
1100 }
1101 // SAFETY: every per-buffer count is zero, so no operation still
1102 // references these buffers, and this is the only place `buffers` is
1103 // ever dropped.
1104 unsafe { ManuallyDrop::drop(&mut self.buffers) };
1105 }
1106}
1107
1108/// A [`Batch::register_buffers`] push not yet matched to its completion.
1109///
1110/// `buffers` is `ManuallyDrop` and this type's own `Drop` is deliberately
1111/// empty, mirroring the inventory (PR #20 review response): the registration is
1112/// already queued via `BuildIoRingRegisterBuffers` the instant
1113/// [`Batch::register_buffers`] returns, before any completion is observed,
1114/// so a caller that drops this without ever matching a completion has no
1115/// proof the kernel is done deciding whether to retain these addresses.
1116/// Freeing them here anyway would risk handing memory the kernel still
1117/// references back to the allocator; leaking is the safe failure mode, the
1118/// same choice the ring's inventory (D-73) and [`RegisteredBuffers`] both
1119/// already make.
1120pub struct PendingBufferRegistration<B: IoBufMut> {
1121 user_data: usize,
1122 base_index: u32,
1123 buffers: ManuallyDrop<Vec<B>>,
1124 /// The lengths handed to `BuildIoRingRegisterBuffers`, captured there
1125 /// rather than recomputed here, so what a span is later validated against
1126 /// is exactly what the kernel was told.
1127 registered_lens: Vec<u32>,
1128 ring_id: RingId,
1129}
1130
1131impl<B: IoBufMut> PendingBufferRegistration<B> {
1132 /// This push's `UserData` identity, to match against a popped
1133 /// [`Completion`].
1134 #[must_use]
1135 pub fn user_data(&self) -> usize {
1136 self.user_data
1137 }
1138
1139 /// Turn this pending registration into usable [`RegisteredBuffers`] once
1140 /// `completion` names it, or hand it back unchanged if `completion`
1141 /// names a different operation.
1142 ///
1143 /// # Errors
1144 ///
1145 /// The inner `Result` is `Err` if the registration itself failed; the
1146 /// buffers are dropped normally in that case, exactly as if they had
1147 /// never been registered -- a matched completion, success or failure, is
1148 /// exactly the proof this type's `Drop` is waiting for.
1149 pub fn claim_if(
1150 mut self,
1151 completion: &Completion,
1152 ) -> Result<io::Result<RegisteredBuffers<B>>, Self> {
1153 if completion.user_data() != self.user_data || completion.ring_id() != self.ring_id {
1154 return Err(self);
1155 }
1156 // SAFETY: `completion` names this exact registration, so the kernel
1157 // has already decided whether to retain these addresses -- the
1158 // condition this type's `Drop` would otherwise wait forever for.
1159 let buffers = unsafe { ManuallyDrop::take(&mut self.buffers) };
1160 let base_index = self.base_index;
1161 let ring_id = self.ring_id;
1162 let registered_lens = std::mem::take(&mut self.registered_lens);
1163 Ok(completion.result().map(move |_| {
1164 let state: Arc<[BufferState]> = registered_lens
1165 .into_iter()
1166 .map(|registered_len| BufferState {
1167 outstanding: AtomicUsize::new(0),
1168 kernel_writes: AtomicUsize::new(0),
1169 registered_len,
1170 })
1171 .collect();
1172 debug_assert_eq!(
1173 state.len(),
1174 buffers.len(),
1175 "one state entry per registered buffer"
1176 );
1177 RegisteredBuffers {
1178 buffers: ManuallyDrop::new(buffers),
1179 base_index,
1180 state,
1181 ring_id,
1182 }
1183 }))
1184 }
1185}
1186
1187impl<B: IoBufMut> Drop for PendingBufferRegistration<B> {
1188 fn drop(&mut self) {
1189 // Deliberately empty; see this type's own doc comment.
1190 }
1191}
1192
1193impl<B: IoBufMut> std::fmt::Debug for PendingBufferRegistration<B> {
1194 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1195 // Not derived: deriving would require `B: Debug`, which a caller's
1196 // buffer type need not satisfy.
1197 f.debug_struct("PendingBufferRegistration")
1198 .field("user_data", &self.user_data)
1199 .field("base_index", &self.base_index)
1200 .finish_non_exhaustive()
1201 }
1202}
1203
1204/// A scoped, exclusive submission window over one [`IoRing`] (D-5).
1205///
1206/// Holding `&mut IoRing` for its lifetime is what turns Win32's own
1207/// "you must serialize submission" footnote into a compiler-enforced
1208/// guarantee: two batches over the same ring cannot coexist. Every push
1209/// queues its SQE the instant it succeeds -- there is no rewind -- so
1210/// `Batch` submits on [`Drop`] rather than leaving queued SQEs for some
1211/// later, unrelated submit to discover (D-5).
1212pub struct Batch<'ring, T = (), X = (), K = Open> {
1213 ring: &'ring mut IoRing<T, X, K>,
1214 submitted: bool,
1215}
1216
1217impl<'ring, T, X, K> Batch<'ring, T, X, K> {
1218 /// Open a batch over `ring`. The batch has the ring's kind: over an [`Owned`](crate::Owned) ring it has
1219 /// no call that queues an operation the ring holds nothing for ([D-91](../DESIGN-NOTES.md#d-91)).
1220 pub fn new(ring: &'ring mut IoRing<T, X, K>) -> Self {
1221 Self {
1222 ring,
1223 submitted: false,
1224 }
1225 }
1226
1227 fn require(&self, op: Op) -> io::Result<()> {
1228 if self.ring.supports(op) {
1229 Ok(())
1230 } else {
1231 Err(io::Error::new(
1232 io::ErrorKind::Unsupported,
1233 format!("this ring does not support {op:?}"),
1234 ))
1235 }
1236 }
1237
1238 /// Reserve an identity for an inventory push.
1239 ///
1240 /// Split from [`Batch::finish_owned`] because the identity has to exist
1241 /// *before* the SQE is built -- the kernel echoes it back -- while what
1242 /// the ring holds is only decided once the build has succeeded.
1243 fn begin_owned(&mut self) -> io::Result<(usize, OperationId)> {
1244 let user_data = self.ring.accounting_mut().reserve_user_data()?;
1245 Ok((user_data, OperationId::new(user_data, self.ring.ring_id())))
1246 }
1247
1248 /// The shared tail of every inventory push: stow on success; on failure,
1249 /// release the reservation, release what the ring held for the operation
1250 /// (the file guard and any registered-buffer use), and hand the payload
1251 /// and sidecar back in a [`PushRefused`] ([D-80](../DESIGN-NOTES.md#d-80)).
1252 ///
1253 /// Handing back is sound for the same reason dropping was before it: a
1254 /// `Build*` that fails queued no SQE, so the kernel never saw the address
1255 /// and nothing will ever complete against it. Leaking is correct only once
1256 /// the kernel *has* seen an address, which a failed build rules out.
1257 fn finish_owned(
1258 &mut self,
1259 hr: windows_sys::core::HRESULT,
1260 id: OperationId,
1261 payload: Option<T>,
1262 extra: X,
1263 held: Held,
1264 ) -> Result<OperationId, PushRefused<T, X>> {
1265 match check(hr) {
1266 Ok(()) => {
1267 self.ring.stow(
1268 id,
1269 Entry {
1270 payload,
1271 extra,
1272 held,
1273 },
1274 );
1275 Ok(id)
1276 }
1277 Err(error) => {
1278 self.ring.cancel_reservation(id.user_data());
1279 drop(held);
1280 Err(PushRefused::new(error, payload, extra))
1281 }
1282 }
1283 }
1284
1285 /// Refuse a [`RegisteredBuffers`] that did not come from this batch's own
1286 /// ring (PR #20 review response): its index space is only meaningful
1287 /// against the ring that registered it, and a different ring may have an
1288 /// entirely unrelated (or already-freed) buffer at the same index.
1289 fn check_registration_ring<B: IoBufMut>(
1290 &self,
1291 registration: &RegisteredBuffers<B>,
1292 ) -> io::Result<()> {
1293 if registration.ring_id == self.ring.ring_id() {
1294 Ok(())
1295 } else {
1296 Err(io::Error::new(
1297 io::ErrorKind::InvalidInput,
1298 "this RegisteredBuffers was registered on a different IoRing",
1299 ))
1300 }
1301 }
1302
1303 /// Submit everything queued so far, returning the number of entries the
1304 /// kernel accepted.
1305 ///
1306 /// That count is entries *submitted*, never entries completed (M10.2):
1307 /// draining is [`crate::IoRing::try_pop`]'s job and stays the caller's.
1308 ///
1309 /// # Errors
1310 ///
1311 /// Returns any error from `SubmitIoRing`. **An error means the entries
1312 /// were not submitted and remain in the submission queue**, which is the
1313 /// documented behaviour of that call: *"If this function returns an error
1314 /// other than IORING_E_WAIT_TIMEOUT, then all entries remain in the
1315 /// submission queue."* They are not lost and they are not rewound
1316 /// ([D-5](../DESIGN-NOTES.md#d-5)) -- a later submit on this ring is what
1317 /// runs them, so **the buffers they reference must stay alive**. Whether
1318 /// and when to submit again is the caller's policy, not this crate's.
1319 pub fn submit(self) -> io::Result<u32> {
1320 self.submit_and_wait(0, 0)
1321 }
1322
1323 /// Close the batch **without submitting** what it queued: the entries stay in the submission
1324 /// queue, held by the ring with their buffers, until a later [`Batch::submit`] -- over a batch
1325 /// that queues nothing, if need be -- issues them ([D-90](../DESIGN-NOTES.md#d-90)).
1326 ///
1327 /// A batch submits as it drops by default ([D-5](../DESIGN-NOTES.md#d-5)), so that nothing is
1328 /// left queued for a later, unrelated submit to discover. This is the caller's explicit choice to
1329 /// own the doorbell instead, to ring it once for several pushes, or to learn that it failed:
1330 /// the drop's own submit reports nothing. It is the state a failed [`Batch::submit`] leaves,
1331 /// reached on purpose. **Until the submit, the kernel has not seen the entries**, so none will
1332 /// complete.
1333 pub fn leave_queued(mut self) {
1334 self.submitted = true;
1335 }
1336
1337 /// Submit everything queued so far, then block until at least
1338 /// `wait_operations` of them have completed or `timeout_ms` elapses
1339 /// (D-3's Model B primitive: the fused submit-and-wait *is* the event
1340 /// loop for a pinned-thread consumer).
1341 ///
1342 /// Returning does not mean `wait_operations` completions are poppable:
1343 /// the timeout can expire first, and the returned count is entries
1344 /// submitted rather than completed (M10.2). Drain with
1345 /// [`crate::IoRing::try_pop`] and count for yourself.
1346 ///
1347 /// **A wait that expires is success, not failure** (M26.8). `SubmitIoRing`
1348 /// answers `IORING_E_WAIT_TIMEOUT` in that case, and its documented
1349 /// meaning is *"All operations were submitted without error and the
1350 /// subsequent wait timed out"* -- so the submission half did everything it
1351 /// was asked to. Reporting that as an `Err` is the defect `M21.6` fixed
1352 /// for [`crate::IoRing::pop_within`]; this site was missed by that sweep
1353 /// and kept it until `M26.8`. The distinction is not cosmetic: an `Err`
1354 /// here means the entries are **still queued**, and a caller who frees
1355 /// their buffers on seeing one would hand the kernel freed memory on the
1356 /// next submit.
1357 ///
1358 /// # Errors
1359 ///
1360 /// Returns any error from `SubmitIoRing` **other than an expired wait**.
1361 /// As [`Batch::submit`], an error means the entries remain in the
1362 /// submission queue and their buffers must stay alive.
1363 pub fn submit_and_wait(mut self, wait_operations: u32, timeout_ms: u32) -> io::Result<u32> {
1364 self.do_submit(wait_operations, timeout_ms)
1365 }
1366
1367 fn do_submit(&mut self, wait_operations: u32, timeout_ms: u32) -> io::Result<u32> {
1368 let mut submitted = 0_u32;
1369 // SAFETY: `self.ring`'s handle is live.
1370 let hr = unsafe {
1371 crate::sys::submit(
1372 self.ring.ring_id(),
1373 self.ring.raw_handle(),
1374 wait_operations,
1375 timeout_ms,
1376 &raw mut submitted,
1377 )
1378 };
1379 // Marked attempted *before* propagating a failure (PR #20 review
1380 // response): `submit_and_wait` takes `self` by value, so this call
1381 // failing still runs `Drop` once its caller's `self` goes out of
1382 // scope. Setting this first is what stops `Drop` from silently
1383 // retrying an explicit submit that already ran -- which could
1384 // succeed on the retry, submitting operations the caller's `Err`
1385 // never told them about.
1386 self.submitted = true;
1387 // `IORING_E_WAIT_TIMEOUT` is not a submission failure: the documented
1388 // meaning of that code is that every entry went in and only the wait
1389 // ran out (M26.8). Passing it to `check` reported a successful submit
1390 // as an error, which is `M21.6`'s defect at the one site that sweep
1391 // did not reach -- and the worse half is that it made an `Err` here
1392 // ambiguous between "still queued" and "submitted, wait expired",
1393 // which have opposite consequences for buffer ownership.
1394 if !crate::ring::every_entry_was_submitted(hr) {
1395 check(hr)?;
1396 }
1397 Ok(submitted)
1398 }
1399}
1400impl<T, X, K> Drop for Batch<'_, T, X, K> {
1401 fn drop(&mut self) {
1402 if !self.submitted {
1403 // Best-effort: Drop cannot propagate an error, and a batch that
1404 // queued nothing submits zero entries harmlessly (D-5).
1405 let _ = self.do_submit(0, 0);
1406 }
1407 }
1408}
1409
1410// The push surface lives in `push.rs`. It is a child rather than a sibling so
1411// it can see this module's private plumbing -- `require`, `begin_owned`,
1412// `finish_owned`, `check_registration_ring`, `handle_ref`, `checked_len` and
1413// the private `sqe_flags`/`raw` accessors on the option types. No public path
1414// changes: an inherent `impl` is found by the type, not by the module it was
1415// written in.
1416mod push;
1417
1418#[cfg(test)]
1419mod tests;