Skip to main content

windows_overlapped_io_sys/
fs.rs

1// Copyright (c) 2026 Mike Grier
2//! Safe file-family operation adapters, gated behind the `fs` feature.
3//!
4//! These wrappers own the I/O buffer and issue the single native `ReadFile` /
5//! `WriteFile` internally, so a caller performs file overlapped I/O without
6//! touching `OVERLAPPED`, the submission seam, or `unsafe`. They are the file
7//! family's realization of the per-family safe-adapter decision; other families
8//! follow the same shape.
9
10use std::alloc::{self, Layout};
11use std::fmt;
12use std::io;
13use std::os::windows::io::AsRawHandle;
14use std::ptr::NonNull;
15use std::slice;
16
17use windows_sys::Win32::Foundation::{ERROR_IO_PENDING, FALSE};
18use windows_sys::Win32::Storage::FileSystem::{
19    FILE_SEGMENT_ELEMENT, ReadFile, ReadFileScatter, WriteFile, WriteFileGather,
20};
21use windows_sys::Win32::System::IO::{GetOverlappedResult, OVERLAPPED};
22
23use crate::operation::{payload_ptr_from_overlapped, sync_bytes_ptr_from_overlapped};
24use crate::{
25    AssociatedEndpoint, BlockingEndpoint, Completion, IoBuf, IoBufMut, Issued, Operation,
26    OperationId, Started, Submitted,
27};
28
29impl BlockingEndpoint {
30    /// Read into `buffer` starting at `offset`, blocking until the read
31    /// completes, and return the number of bytes read.
32    ///
33    /// Takes a plain `&mut [u8]` rather than an owned buffer, and allocates
34    /// nothing: this call does not return until the operation is over, so an
35    /// ordinary borrow provably covers the whole time the kernel is writing.
36    /// That is the difference from [`AssociatedEndpoint::read`], which must take
37    /// ownership because its operation outlives the call.
38    ///
39    /// # Errors
40    ///
41    /// Returns [`io::ErrorKind::InvalidInput`] if `buffer` is longer than
42    /// `u32::MAX`, which the read's byte count cannot express, or any error from
43    /// issuing or completing the read.
44    ///
45    /// # Examples
46    ///
47    /// One owner issuing reads in sequence is the supported shape, and compiles:
48    ///
49    /// ```
50    /// use windows_overlapped_io_sys::BlockingEndpoint;
51    ///
52    /// fn read_twice(endpoint: &mut BlockingEndpoint) -> std::io::Result<()> {
53    ///     let mut buffer = [0_u8; 64];
54    ///     let _first = endpoint.read(&mut buffer, 0)?;
55    ///     let _second = endpoint.read(&mut buffer, 64)?;
56    ///     Ok(())
57    /// }
58    /// ```
59    ///
60    /// Sharing one endpoint across threads and reading from both is rejected at
61    /// compile time rather than corrupting a result at run time, because `read`
62    /// takes `&mut self` while an `Arc` can only hand out `&BlockingEndpoint`:
63    ///
64    /// ```compile_fail
65    /// use std::sync::Arc;
66    /// use windows_overlapped_io_sys::BlockingEndpoint;
67    ///
68    /// fn read_from_two_threads(endpoint: BlockingEndpoint) {
69    ///     let shared = Arc::new(endpoint);
70    ///     let other = Arc::clone(&shared);
71    ///     std::thread::spawn(move || other.read(&mut [0_u8; 64], 0));
72    ///     let _ = shared.read(&mut [0_u8; 64], 64);
73    /// }
74    /// ```
75    pub fn read(&mut self, buffer: &mut [u8], offset: u64) -> io::Result<usize> {
76        let buf_len = checked_len(buffer.len(), "read buffer")?;
77        let buf_ptr = buffer.as_mut_ptr();
78
79        let mut operation = Operation::new(());
80        operation.set_offset(offset);
81        // SAFETY: issues exactly one overlapped ReadFile into `buffer`, which
82        // outlives this blocking call; no other operation is outstanding.
83        unsafe {
84            self.run(&mut operation, |handle, overlapped| {
85                let ok = ReadFile(
86                    handle.as_raw_handle(),
87                    buf_ptr,
88                    buf_len,
89                    std::ptr::null_mut(),
90                    overlapped,
91                );
92                classify(ok)
93            })
94        }
95    }
96
97    /// Write `data` starting at `offset`, blocking until the write completes, and
98    /// return the number of bytes written.
99    ///
100    /// # Errors
101    ///
102    /// Returns [`io::ErrorKind::InvalidInput`] if `data` is longer than
103    /// `u32::MAX`, which the write's byte count cannot express, or any error
104    /// from issuing or completing the write.
105    pub fn write(&mut self, data: &[u8], offset: u64) -> io::Result<usize> {
106        let data_ptr = data.as_ptr();
107        let data_len = checked_len(data.len(), "write buffer")?;
108
109        let mut operation = Operation::new(());
110        operation.set_offset(offset);
111        // SAFETY: issues exactly one overlapped WriteFile from `data`, which
112        // outlives this blocking call; no other operation is outstanding.
113        let written = unsafe {
114            self.run(&mut operation, |handle, overlapped| {
115                let ok = WriteFile(
116                    handle.as_raw_handle(),
117                    data_ptr,
118                    data_len,
119                    std::ptr::null_mut(),
120                    overlapped,
121                );
122                classify(ok)
123            })
124        }?;
125
126        Ok(written)
127    }
128}
129
130/// Map a native `BOOL` into the submission-seam contract: native success or
131/// `ERROR_IO_PENDING` is accepted, any other error is an immediate failure.
132fn classify(ok: i32) -> io::Result<()> {
133    if ok != 0 {
134        return Ok(());
135    }
136    let error = io::Error::last_os_error();
137    if error.raw_os_error() == Some(ERROR_IO_PENDING as i32) {
138        Ok(())
139    } else {
140        Err(error)
141    }
142}
143
144/// Convert a buffer length to the `u32` byte count the Win32 calls take.
145///
146/// Rejects rather than caps, for the same reason as the device-control helper:
147/// capping would transfer a prefix of the caller's buffer and then report
148/// success for an operation that did something other than what was asked.
149fn checked_len(len: usize, which: &str) -> io::Result<u32> {
150    u32::try_from(len).map_err(|_| {
151        io::Error::new(
152            io::ErrorKind::InvalidInput,
153            format!("a {which} is limited to u32::MAX bytes; {len} does not fit"),
154        )
155    })
156}
157
158impl AssociatedEndpoint<'_> {
159    /// Submit an overlapped read into `buffer`, starting at `offset`.
160    ///
161    /// The buffer is any owned [`IoBufMut`] -- a `Vec<u8>`, a `Box<[u8]>`, a
162    /// [`PageBuffers`], or a caller's own pooled or aligned type -- handed over
163    /// for the operation's life and returned when it completes. Nothing is
164    /// copied and nothing is allocated here: a caller that wants a fresh `Vec`
165    /// writes `vec![0; n]` at the call site, where the allocation is visible.
166    ///
167    /// Returns [`Started::Pending`] with a [`FileIo`] token that recovers the
168    /// buffer and byte count from the operation's completion, or -- only on an
169    /// endpoint in `FILE_SKIP_COMPLETION_PORT_ON_SUCCESS` mode, where a
170    /// synchronous success queues no packet -- [`Started::Completed`] with the
171    /// buffer already in hand.
172    ///
173    /// # Errors
174    ///
175    /// Returns [`io::ErrorKind::InvalidInput`] if the buffer is longer than
176    /// `u32::MAX`, or any immediate failure from issuing the read.
177    #[track_caller]
178    pub fn read<B: IoBufMut>(
179        &self,
180        mut buffer: B,
181        offset: u64,
182    ) -> io::Result<Started<FileIo<B>, B>> {
183        let buf_len = checked_len(buffer.bytes_len(), "read buffer")?;
184        let skip = self.notification_modes().skip_completion_port_on_success;
185        // Captured before submission, like `buf_len` above, rather than calling
186        // this safe trait method inside the registered closure (PR #20 review
187        // response): `IoBufMut::stable_mut_ptr` is implementable by a caller's
188        // own buffer type, and `submit`'s safety contract forbids the closure
189        // unwinding -- a panic here, before the operation is registered, is
190        // merely an ordinary panic, where one from inside the closure would
191        // leave the operation permanently outstanding.
192        let buf_ptr = buffer.stable_mut_ptr();
193        let mut operation = Operation::new(buffer);
194        operation.set_offset(offset);
195        // SAFETY: issues exactly one ReadFile into the operation's own payload
196        // buffer at `buf_ptr` (captured above, and identical to what the pinned
197        // payload would report, per `IoBufMut`'s address-stability contract);
198        // `IoBufMut` promises that address is stable and exclusively owned, and
199        // the byte-count cell live until the completion is claimed.
200        let submitted = unsafe {
201            self.submit(operation, |handle, overlapped| {
202                let bytes = sync_bytes_ptr_from_overlapped(overlapped);
203                let ok = ReadFile(handle.as_raw_handle(), buf_ptr, buf_len, bytes, overlapped);
204                classify_issued(ok, skip, bytes)
205            })
206        };
207        finish(submitted)
208    }
209
210    /// Submit an overlapped write of `buffer`, starting at `offset`.
211    ///
212    /// The buffer is any owned [`IoBuf`] -- including a shared `Arc<[u8]>` or a
213    /// `&'static [u8]`, neither of which can be a read destination -- handed over
214    /// for the operation's life and returned when it completes. Nothing is
215    /// copied.
216    ///
217    /// Returns [`Started::Pending`] with a [`FileIo`] token, or
218    /// [`Started::Completed`] with the buffer already in hand when the endpoint
219    /// is in skip-on-success mode and the write completed synchronously.
220    ///
221    /// # Errors
222    ///
223    /// Returns [`io::ErrorKind::InvalidInput`] if the buffer is longer than
224    /// `u32::MAX`, or any immediate failure from issuing the write.
225    #[track_caller]
226    pub fn write<B: IoBuf>(&self, buffer: B, offset: u64) -> io::Result<Started<FileIo<B>, B>> {
227        let data_len = checked_len(buffer.bytes_len(), "write buffer")?;
228        let skip = self.notification_modes().skip_completion_port_on_success;
229        // Captured before submission; see `read`'s matching comment above for
230        // why (PR #20 review response).
231        let data_ptr = buffer.stable_ptr();
232        let mut operation = Operation::new(buffer);
233        operation.set_offset(offset);
234        // SAFETY: issues exactly one WriteFile from the operation's own payload
235        // buffer at `data_ptr` (captured above; identical to what the pinned
236        // payload would report, per `IoBuf`'s address-stability contract);
237        // `IoBuf` promises that address is stable and its bytes unmodified, and
238        // the byte-count cell live until the completion is claimed.
239        let submitted = unsafe {
240            self.submit(operation, |handle, overlapped| {
241                let bytes = sync_bytes_ptr_from_overlapped(overlapped);
242                let ok = WriteFile(
243                    handle.as_raw_handle(),
244                    data_ptr,
245                    data_len,
246                    bytes,
247                    overlapped,
248                );
249                classify_issued(ok, skip, bytes)
250            })
251        };
252        finish(submitted)
253    }
254}
255
256/// Map a native `BOOL` into the IOCP submission contract.
257///
258/// # Why an immediate `TRUE` is usually `Pending`
259///
260/// [`Issued`] does not record whether the call finished synchronously. It
261/// records whether a **completion packet will arrive**, and for an IOCP-bound
262/// overlapped handle those are different facts: the I/O Manager queues a packet
263/// for every request it completes, *including* one that succeeded immediately
264/// without returning `ERROR_IO_PENDING`. See [`Issued::Pending`].
265///
266/// The single exception is `skip_on_success`, which is why this needs to know
267/// it: on an endpoint in `FILE_SKIP_COMPLETION_PORT_ON_SUCCESS` mode no packet
268/// is queued for an immediate success, so that -- and only that -- is an
269/// [`Issued::Completed`]. Getting this backwards in either direction is a bug
270/// with teeth: claiming `Completed` when a packet is coming frees the operation
271/// under a live `OVERLAPPED`, and claiming `Pending` when none is coming leaves
272/// the operation outstanding forever and wedges rundown.
273///
274/// # Safety
275///
276/// `sync_bytes` must be the byte-count cell of the operation being submitted,
277/// which is live for the whole call.
278unsafe fn classify_issued(
279    ok: i32,
280    skip_on_success: bool,
281    sync_bytes: *mut u32,
282) -> io::Result<Issued> {
283    if ok != 0 {
284        if skip_on_success {
285            // SAFETY: the call reported immediate success, so the kernel has
286            // already written the count and will not write it again.
287            let bytes_transferred = unsafe { *sync_bytes };
288            return Ok(Issued::Completed { bytes_transferred });
289        }
290        return Ok(Issued::Pending);
291    }
292    let error = io::Error::last_os_error();
293    if error.raw_os_error() == Some(ERROR_IO_PENDING as i32) {
294        Ok(Issued::Pending)
295    } else {
296        Err(error)
297    }
298}
299
300/// As [`classify_issued`], for the scatter/gather calls.
301///
302/// `ReadFileScatter` and `WriteFileGather` take no byte-count out-parameter --
303/// the slot in that position is `lpReserved` and must be null -- so on the
304/// skip-on-success path the count comes from `GetOverlappedResult` instead.
305/// That is the sanctioned way to read it (`Internal`/`InternalHigh` are never
306/// touched directly), and it cannot block here: it is called only after the
307/// call reported immediate success, so the operation is already complete and
308/// `bWait` is `FALSE`.
309///
310/// # Safety
311///
312/// `handle` must be the endpoint's live handle and `overlapped` the identity of
313/// the operation just submitted through it.
314unsafe fn classify_scatter(
315    ok: i32,
316    skip_on_success: bool,
317    handle: std::os::windows::io::RawHandle,
318    overlapped: *mut OVERLAPPED,
319) -> io::Result<Issued> {
320    if ok != 0 {
321        if skip_on_success {
322            let mut bytes_transferred = 0_u32;
323            // SAFETY: a live handle and the completed operation's own
324            // OVERLAPPED; `bWait` is FALSE, so this only reads what is already
325            // recorded.
326            let got =
327                unsafe { GetOverlappedResult(handle, overlapped, &mut bytes_transferred, FALSE) };
328            if got == 0 {
329                return Err(io::Error::last_os_error());
330            }
331            return Ok(Issued::Completed { bytes_transferred });
332        }
333        return Ok(Issued::Pending);
334    }
335    let error = io::Error::last_os_error();
336    if error.raw_os_error() == Some(ERROR_IO_PENDING as i32) {
337        Ok(Issued::Pending)
338    } else {
339        Err(error)
340    }
341}
342
343/// Turn a submission outcome into the adapter's two-state outcome.
344fn finish<B: IoBuf>(submitted: Submitted<B>) -> io::Result<Started<FileIo<B>, B>> {
345    match submitted {
346        Submitted::Pending(id) => Ok(Started::Pending(FileIo {
347            id,
348            buffer: std::marker::PhantomData,
349        })),
350        Submitted::Completed {
351            operation,
352            bytes_transferred,
353        } => Ok(Started::Completed {
354            payload: operation.into_payload(),
355            bytes_transferred: bytes_transferred as usize,
356        }),
357        Submitted::Failed { error, .. } => Err(error),
358    }
359}
360
361/// A pending file operation submitted through [`AssociatedEndpoint::read`] or
362/// [`AssociatedEndpoint::write`].
363///
364/// The token carries the operation's identity and remembers the buffer type it
365/// was submitted with, so [`FileIo::claim`] hands back the caller's own buffer
366/// -- the same value, not a copy -- once the matching completion is dequeued.
367#[derive(Debug)]
368pub struct FileIo<B> {
369    id: OperationId,
370    /// The buffer itself is in the pinned operation, not here; this only keeps
371    /// the token's type tied to it so `claim` cannot be handed the wrong one.
372    buffer: std::marker::PhantomData<fn() -> B>,
373}
374
375impl<B: IoBuf> FileIo<B> {
376    /// The identity of the in-flight operation, for cancellation or matching.
377    #[must_use]
378    pub fn id(&self) -> OperationId {
379        self.id
380    }
381
382    /// Claim this operation's result from `completion`.
383    ///
384    /// On a match returns `Ok((buffer, result))`: `buffer` is the one the caller
385    /// handed over -- the bytes read, or the data written -- and `result` is the
386    /// byte count or the operation's error. Returns `Err(self)` when
387    /// `completion` belongs to a different operation, so the caller can try the
388    /// token against another one.
389    pub fn claim(self, completion: &Completion) -> Result<(B, io::Result<usize>), Self> {
390        if completion.id() != Some(self.id) {
391            return Err(self);
392        }
393        // SAFETY: the full identity -- address *and* generation -- matches, which
394        // an address alone would not: a recycled address can belong to a later
395        // operation of a different payload type. The match therefore proves this
396        // completion is the Operation<B> this token submitted, and the token's
397        // own type parameter names that B; claim it exactly once.
398        let operation = unsafe { completion.claim::<B>() };
399        let buffer = operation.into_payload();
400        let result = match completion.error() {
401            Some(error) => Err(io::Error::from_raw_os_error(
402                error.raw_os_error().unwrap_or_default(),
403            )),
404            None => Ok(completion.bytes_transferred() as usize),
405        };
406        Ok((buffer, result))
407    }
408}
409
410/// The memory page size assumed by the scatter/gather adapters.
411///
412/// A fixed 4 KiB, matching every Windows target this crate supports. Buffers are
413/// aligned to it and I/O lengths are multiples of it, which also satisfies the
414/// sector alignment `FILE_FLAG_NO_BUFFERING` requires.
415pub const PAGE_SIZE: usize = 4096;
416
417/// The Win32 `FILE_FLAG_NO_BUFFERING` flag.
418///
419/// The scatter/gather adapters require the endpoint be opened with this flag (in
420/// addition to `FILE_FLAG_OVERLAPPED`, which [`crate::UnassociatedEndpoint::open`]
421/// always sets); pass it as that constructor's `extra_flags`.
422pub const FILE_FLAG_NO_BUFFERING: u32 =
423    windows_sys::Win32::Storage::FileSystem::FILE_FLAG_NO_BUFFERING;
424
425/// A page-aligned set of memory pages: the buffer form the scatter/gather
426/// adapters read into and write from.
427///
428/// It owns one page-aligned allocation of `pages * PAGE_SIZE` bytes and can be
429/// viewed as a byte slice. Its page-aligned segments are what `ReadFileScatter`
430/// and `WriteFileGather` require.
431pub struct PageBuffers {
432    ptr: NonNull<u8>,
433    pages: usize,
434}
435
436// SAFETY: `PageBuffers` uniquely owns its heap allocation; moving it between
437// threads moves that ownership, and it hands out aliasing access only through
438// `&`/`&mut self`, so it is as `Send`/`Sync` as an owned `Box<[u8]>`.
439unsafe impl Send for PageBuffers {}
440unsafe impl Sync for PageBuffers {}
441
442impl PageBuffers {
443    /// Allocate `pages` zeroed, page-aligned memory pages.
444    ///
445    /// # Panics
446    ///
447    /// Panics if `pages` is zero or `pages * PAGE_SIZE` overflows.
448    #[must_use]
449    pub fn new(pages: usize) -> Self {
450        assert!(pages > 0, "PageBuffers requires at least one page");
451        let size = pages
452            .checked_mul(PAGE_SIZE)
453            .expect("page buffer size overflow");
454        let layout = Layout::from_size_align(size, PAGE_SIZE).expect("valid page layout");
455        // SAFETY: `layout` has non-zero size.
456        let raw = unsafe { alloc::alloc_zeroed(layout) };
457        let ptr = NonNull::new(raw).unwrap_or_else(|| alloc::handle_alloc_error(layout));
458        Self { ptr, pages }
459    }
460
461    /// The number of pages.
462    #[must_use]
463    pub fn pages(&self) -> usize {
464        self.pages
465    }
466
467    /// The total length in bytes (`pages * PAGE_SIZE`).
468    #[must_use]
469    pub fn len(&self) -> usize {
470        self.pages * PAGE_SIZE
471    }
472
473    /// Always `false`: a `PageBuffers` holds at least one page.
474    #[must_use]
475    pub fn is_empty(&self) -> bool {
476        false
477    }
478
479    /// View the pages as a shared byte slice.
480    #[must_use]
481    pub fn as_bytes(&self) -> &[u8] {
482        // SAFETY: `ptr` owns `len()` initialized bytes for the shared borrow.
483        unsafe { slice::from_raw_parts(self.ptr.as_ptr(), self.len()) }
484    }
485
486    /// View the pages as a mutable byte slice.
487    #[must_use]
488    pub fn as_bytes_mut(&mut self) -> &mut [u8] {
489        // SAFETY: exclusive borrow of `len()` bytes this owns.
490        unsafe { slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len()) }
491    }
492
493    /// Build the `NULL`-terminated `FILE_SEGMENT_ELEMENT` array over these pages.
494    fn segment_array(&self) -> Vec<FILE_SEGMENT_ELEMENT> {
495        let mut segments = Vec::with_capacity(self.pages + 1);
496        for i in 0..self.pages {
497            // SAFETY: `i < pages`, so the offset stays within the allocation, and
498            // each page start is page-aligned because the base is.
499            let page = unsafe { self.ptr.as_ptr().add(i * PAGE_SIZE) };
500            segments.push(FILE_SEGMENT_ELEMENT {
501                Buffer: page.cast(),
502            });
503        }
504        // A zeroed element terminates the array.
505        segments.push(FILE_SEGMENT_ELEMENT { Alignment: 0 });
506        segments
507    }
508}
509
510impl fmt::Debug for PageBuffers {
511    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
512        f.debug_struct("PageBuffers")
513            .field("pages", &self.pages)
514            .finish_non_exhaustive()
515    }
516}
517
518impl Drop for PageBuffers {
519    fn drop(&mut self) {
520        let layout = Layout::from_size_align(self.len(), PAGE_SIZE).expect("valid page layout");
521        // SAFETY: `ptr` came from `alloc_zeroed` with this exact layout.
522        unsafe { alloc::dealloc(self.ptr.as_ptr(), layout) };
523    }
524}
525
526// SAFETY: the bytes live in a page-aligned heap allocation `PageBuffers` owns
527// outright, so moving the value moves the pointer and not the bytes, and the
528// page count is fixed at construction. `alloc_zeroed` initializes all of them.
529unsafe impl crate::IoBuf for PageBuffers {
530    fn stable_ptr(&self) -> *const u8 {
531        self.ptr.as_ptr()
532    }
533
534    fn bytes_len(&self) -> usize {
535        self.len()
536    }
537}
538
539// SAFETY: as above; `PageBuffers` is a unique owner, so `&mut self` is exclusive
540// access to the same allocation `stable_ptr` reports.
541unsafe impl crate::IoBufMut for PageBuffers {
542    fn stable_mut_ptr(&mut self) -> *mut u8 {
543        self.ptr.as_ptr()
544    }
545}
546
547impl BlockingEndpoint {
548    /// Scatter-read into `buffers` starting at `offset`, blocking until the read
549    /// completes, and return the number of bytes read.
550    ///
551    /// Takes the caller's pages by `&mut` and allocates nothing, matching
552    /// [`BlockingEndpoint::write_gather`]; the endpoint must be opened with
553    /// [`FILE_FLAG_NO_BUFFERING`], or the native call fails.
554    ///
555    /// # Errors
556    ///
557    /// Returns [`io::ErrorKind::InvalidInput`] if the pages total more than
558    /// `u32::MAX` bytes, or any error from issuing or completing the
559    /// scatter-read.
560    pub fn read_scatter(&mut self, buffers: &mut PageBuffers, offset: u64) -> io::Result<usize> {
561        let total = checked_len(buffers.len(), "scatter/gather buffer set")?;
562        let segments = buffers.segment_array();
563        let seg_ptr = segments.as_ptr();
564
565        let mut operation = Operation::new(());
566        operation.set_offset(offset);
567        // SAFETY: issues exactly one ReadFileScatter into `buffers` via
568        // `segments`; both outlive this blocking call and no other operation is
569        // outstanding.
570        unsafe {
571            self.run(&mut operation, |handle, overlapped| {
572                let ok = ReadFileScatter(
573                    handle.as_raw_handle(),
574                    seg_ptr,
575                    total,
576                    std::ptr::null(),
577                    overlapped,
578                );
579                classify(ok)
580            })
581        }
582    }
583
584    /// Gather-write `buffers` starting at `offset`, blocking until the write
585    /// completes, and return the number of bytes written.
586    ///
587    /// The endpoint must be opened with [`FILE_FLAG_NO_BUFFERING`]; otherwise the
588    /// native call fails.
589    ///
590    /// # Errors
591    ///
592    /// Returns [`io::ErrorKind::InvalidInput`] if the buffers total more than
593    /// `u32::MAX` bytes, or any error from issuing or completing the
594    /// gather-write.
595    pub fn write_gather(&mut self, buffers: &PageBuffers, offset: u64) -> io::Result<usize> {
596        let segments = buffers.segment_array();
597        let total = checked_len(buffers.len(), "scatter/gather buffer set")?;
598        let seg_ptr = segments.as_ptr();
599
600        let mut operation = Operation::new(());
601        operation.set_offset(offset);
602        // SAFETY: issues exactly one WriteFileGather from `buffers` via
603        // `segments`; both outlive this blocking call and no other operation is
604        // outstanding.
605        let written = unsafe {
606            self.run(&mut operation, |handle, overlapped| {
607                let ok = WriteFileGather(
608                    handle.as_raw_handle(),
609                    seg_ptr,
610                    total,
611                    std::ptr::null(),
612                    overlapped,
613                );
614                classify(ok)
615            })
616        }?;
617
618        Ok(written)
619    }
620}
621
622/// The pinned payload for an in-flight scatter/gather operation: the buffers and
623/// the `FILE_SEGMENT_ELEMENT` array that points into them.
624struct ScatterPayload {
625    buffers: PageBuffers,
626    segments: Vec<FILE_SEGMENT_ELEMENT>,
627}
628
629// SAFETY: the raw pointers in `segments` point into `buffers`, which this payload
630// owns; moving the payload moves the whole self-referential unit together, and it
631// exposes no aliasing access, so it is `Send` like the `PageBuffers` it wraps.
632unsafe impl Send for ScatterPayload {}
633
634impl AssociatedEndpoint<'_> {
635    /// Submit an overlapped scatter-read into `buffers`, starting at `offset`.
636    ///
637    /// Takes the caller's pages rather than allocating fresh ones, so a pooled
638    /// or reused [`PageBuffers`] costs nothing to submit. The endpoint must be
639    /// opened with [`FILE_FLAG_NO_BUFFERING`].
640    ///
641    /// Returns [`Started::Pending`] with a [`ScatterGatherIo`] token, or
642    /// [`Started::Completed`] with the [`PageBuffers`] already in hand when the
643    /// endpoint is in skip-on-success mode and the read completed synchronously.
644    ///
645    /// # Errors
646    ///
647    /// Returns [`io::ErrorKind::InvalidInput`] if the pages total more than
648    /// `u32::MAX` bytes, or any immediate failure from issuing the
649    /// scatter-read.
650    #[track_caller]
651    pub fn read_scatter(
652        &self,
653        buffers: PageBuffers,
654        offset: u64,
655    ) -> io::Result<Started<ScatterGatherIo, PageBuffers>> {
656        let total = checked_len(buffers.len(), "scatter/gather buffer set")?;
657        let skip = self.notification_modes().skip_completion_port_on_success;
658        let segments = buffers.segment_array();
659        let mut operation = Operation::new(ScatterPayload { buffers, segments });
660        operation.set_offset(offset);
661        // SAFETY: issues exactly one ReadFileScatter into the payload's buffers
662        // via its segment array, both reached through the pinned OVERLAPPED; they
663        // live until the completion is claimed.
664        let submitted = unsafe {
665            self.submit(operation, |handle, overlapped| {
666                let payload = payload_ptr_from_overlapped::<ScatterPayload>(overlapped);
667                let raw = handle.as_raw_handle();
668                let ok = ReadFileScatter(
669                    raw,
670                    (*payload).segments.as_ptr(),
671                    total,
672                    std::ptr::null(),
673                    overlapped,
674                );
675                classify_scatter(ok, skip, raw, overlapped)
676            })
677        };
678        finish_scatter(submitted)
679    }
680
681    /// Submit an overlapped gather-write of `buffers` starting at `offset`.
682    ///
683    /// Returns [`Started::Pending`] with a [`ScatterGatherIo`] token, or
684    /// [`Started::Completed`] with the [`PageBuffers`] already in hand when the
685    /// endpoint is in skip-on-success mode and the write completed
686    /// synchronously. The endpoint must be opened with
687    /// [`FILE_FLAG_NO_BUFFERING`].
688    ///
689    /// # Errors
690    ///
691    /// Returns [`io::ErrorKind::InvalidInput`] if the buffers total more than
692    /// `u32::MAX` bytes, or any immediate failure from issuing the gather-write.
693    #[track_caller]
694    pub fn write_gather(
695        &self,
696        buffers: PageBuffers,
697        offset: u64,
698    ) -> io::Result<Started<ScatterGatherIo, PageBuffers>> {
699        let total = checked_len(buffers.len(), "scatter/gather buffer set")?;
700        let skip = self.notification_modes().skip_completion_port_on_success;
701        let segments = buffers.segment_array();
702        let mut operation = Operation::new(ScatterPayload { buffers, segments });
703        operation.set_offset(offset);
704        // SAFETY: issues exactly one WriteFileGather from the payload's buffers
705        // via its segment array, both reached through the pinned OVERLAPPED; they
706        // live until the completion is claimed.
707        let submitted = unsafe {
708            self.submit(operation, |handle, overlapped| {
709                let payload = payload_ptr_from_overlapped::<ScatterPayload>(overlapped);
710                let raw = handle.as_raw_handle();
711                let ok = WriteFileGather(
712                    raw,
713                    (*payload).segments.as_ptr(),
714                    total,
715                    std::ptr::null(),
716                    overlapped,
717                );
718                classify_scatter(ok, skip, raw, overlapped)
719            })
720        };
721        finish_scatter(submitted)
722    }
723}
724
725/// Turn a scatter/gather submission outcome into the adapter's two-state
726/// outcome.
727fn finish_scatter(
728    submitted: Submitted<ScatterPayload>,
729) -> io::Result<Started<ScatterGatherIo, PageBuffers>> {
730    match submitted {
731        Submitted::Pending(id) => Ok(Started::Pending(ScatterGatherIo { id })),
732        Submitted::Completed {
733            operation,
734            bytes_transferred,
735        } => Ok(Started::Completed {
736            payload: operation.into_payload().buffers,
737            bytes_transferred: bytes_transferred as usize,
738        }),
739        Submitted::Failed { error, .. } => Err(error),
740    }
741}
742
743/// A pending scatter/gather operation submitted through
744/// [`AssociatedEndpoint::read_scatter`] or [`AssociatedEndpoint::write_gather`].
745///
746/// The token carries the operation's identity and its payload type, so
747/// [`ScatterGatherIo::claim`] recovers the [`PageBuffers`] and byte count safely
748/// once the matching completion is dequeued.
749#[derive(Debug)]
750pub struct ScatterGatherIo {
751    id: OperationId,
752}
753
754impl ScatterGatherIo {
755    /// The identity of the in-flight operation, for cancellation or matching.
756    #[must_use]
757    pub fn id(&self) -> OperationId {
758        self.id
759    }
760
761    /// Claim this operation's result from `completion`.
762    ///
763    /// On a match returns `Ok((buffers, result))`: `buffers` is the payload (the
764    /// pages read, or the data written) and `result` is the byte count or the
765    /// operation's error. Returns `Err(self)` when `completion` belongs to a
766    /// different operation.
767    pub fn claim(self, completion: &Completion) -> Result<(PageBuffers, io::Result<usize>), Self> {
768        if completion.id() != Some(self.id) {
769            return Err(self);
770        }
771        // SAFETY: the full identity -- address *and* generation -- matches, which
772        // an address alone would not: a recycled address can belong to a later
773        // operation of a different payload type. The match therefore proves this
774        // completion is the
775        // Operation<ScatterPayload> this token submitted; claim it exactly once.
776        let operation = unsafe { completion.claim::<ScatterPayload>() };
777        let buffers = operation.into_payload().buffers;
778        let result = match completion.error() {
779            Some(error) => Err(io::Error::from_raw_os_error(
780                error.raw_os_error().unwrap_or_default(),
781            )),
782            None => Ok(completion.bytes_transferred() as usize),
783        };
784        Ok((buffers, result))
785    }
786}
787
788#[cfg(test)]
789mod tests;