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;