subetha_cxc/kernel_async_ring.rs
1//! `kernel_async_ring`: the kernel async-I/O ring (io_uring on Linux,
2//! IoRing on Windows, POSIX aio on FreeBSD / macOS) exposed as a
3//! substrate ring primitive.
4//!
5//! Both OSes ship the same architecture - a user->kernel SUBMISSION ring
6//! and a kernel->user COMPLETION ring - which is exactly the substrate's
7//! SharedRing shape applied to the user/kernel boundary. This wraps that
8//! kernel object behind one cross-platform surface; only the ring
9//! syscalls are gated, the verb shape (prepare / submit / reap) and the
10//! normalized [`Completion`] are shared:
11//!
12//! - Linux (`#[cfg(target_os = "linux")]`): `io_uring` via the
13//! mainline `io-uring` crate (`IoUring::new` / `opcode::Read` /
14//! `submit_and_wait` / completion iterator).
15//! - Windows (`#[cfg(windows)]`): `IoRing` via `windows-sys`
16//! (`CreateIoRing` / `BuildIoRingReadFile` / `SubmitIoRing` /
17//! `PopIoRingCompletion`), gated on `QueryIoRingCapabilities` +
18//! `IsIoRingOpSupported` so an unsupported build degrades to `Err`
19//! rather than UB.
20//! - FreeBSD (`#[cfg(target_os = "freebsd")]`): POSIX `aio` with kqueue
21//! completion. Each `aio_read` carries an `aio_sigevent` set to
22//! `SIGEV_KEVENT` against the ring's kqueue, so completion posts an
23//! `EVFILT_AIO` kevent (`ident` = the aiocb pointer, `udata` = the
24//! caller's tag); `submit_and_wait` is a `kevent` wait, `reap` calls
25//! `aio_return`. There is no separate batched submit - `aio_read`
26//! issues each op immediately - but the prepare / wait / reap verb
27//! shape is identical.
28//! - macOS (`#[cfg(target_os = "macos")]`): POSIX `aio` with
29//! `aio_suspend` completion. Darwin has no `SIGEV_KEVENT` and its
30//! `EVFILT_AIO` kqueue filter rejects registration, so completion is
31//! driven by `aio_suspend` over the in-flight set rather than a
32//! `kevent` wait; `aio_read` still does the real kernel async I/O and
33//! `reap` calls `aio_return`. Same prepare / wait / reap verb shape.
34//!
35//! The completion encodings differ - io_uring packs bytes-or-`-errno`
36//! into one `i32`; IoRing splits `ResultCode` (HRESULT) and
37//! `Information` (bytes); aio reports via `aio_error` + `aio_return` - so
38//! all three are normalized to a single [`Completion`] with an
39//! `io::Result<usize>` byte count.
40
41#![cfg(any(target_os = "linux", windows, target_os = "freebsd", target_os = "macos"))]
42
43use std::io;
44
45/// One reaped completion: which submission it answers (`user_data`, the
46/// tag the caller passed to `prepare_read`) and its result - the number
47/// of bytes transferred, or the error the kernel reported.
48pub struct Completion {
49 pub user_data: u64,
50 pub bytes: io::Result<usize>,
51}
52
53/// Open a file for reading through the kernel async ring. On Windows the
54/// handle must carry `FILE_FLAG_OVERLAPPED` for `IoRing`; on Linux and
55/// FreeBSD a plain read handle is fine. Keep the returned `File` alive
56/// while its reads are in flight.
57#[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
58pub fn open_for_async_read(path: impl AsRef<std::path::Path>) -> io::Result<std::fs::File> {
59 std::fs::File::open(path)
60}
61
62/// Open a file for reading through the kernel async ring. The handle is
63/// opened with `FILE_FLAG_OVERLAPPED`, required for `IoRing` operations.
64/// Keep the returned `File` alive while its reads are in flight.
65#[cfg(windows)]
66pub fn open_for_async_read(path: impl AsRef<std::path::Path>) -> io::Result<std::fs::File> {
67 use std::os::windows::fs::OpenOptionsExt;
68 use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OVERLAPPED;
69 std::fs::OpenOptions::new()
70 .read(true)
71 .custom_flags(FILE_FLAG_OVERLAPPED)
72 .open(path)
73}
74
75// ---------------------------------------------------------------------
76// Linux: io_uring via the mainline `io-uring` crate.
77// ---------------------------------------------------------------------
78
79#[cfg(target_os = "linux")]
80mod linux_impl {
81 use super::Completion;
82 use io_uring::{opcode, types, IoUring};
83 use std::io;
84
85 /// A kernel async-I/O ring backed by an `io_uring` instance.
86 pub struct KernelAsyncRing {
87 ring: IoUring,
88 }
89
90 impl KernelAsyncRing {
91 /// Create a ring with `entries` submission slots. Fails on kernels
92 /// without io_uring (e.g. some containers) so callers can fall
93 /// back to synchronous I/O.
94 pub fn new(entries: u32) -> io::Result<Self> {
95 Ok(Self { ring: IoUring::new(entries)? })
96 }
97
98 /// Queue a read of `len` bytes from `file` at `offset` into `buf`,
99 /// tagged with `user_data`. Open `file` with
100 /// [`open_for_async_read`](super::open_for_async_read).
101 ///
102 /// # Safety
103 /// `buf` must remain valid and not move until the matching
104 /// completion is reaped - the kernel writes into it asynchronously.
105 pub unsafe fn prepare_read(
106 &mut self,
107 file: &std::fs::File,
108 buf: *mut u8,
109 len: u32,
110 offset: u64,
111 user_data: u64,
112 ) -> io::Result<()> {
113 use std::os::unix::io::AsRawFd;
114 let entry = opcode::Read::new(types::Fd(file.as_raw_fd()), buf, len)
115 .offset(offset)
116 .build()
117 .user_data(user_data);
118 // SAFETY: `entry` references `buf`, which the caller pledges to
119 // keep valid until the completion is reaped (this fn's contract).
120 unsafe {
121 self.ring
122 .submission()
123 .push(&entry)
124 .map_err(|_| io::Error::other("io_uring submission queue full"))?;
125 }
126 Ok(())
127 }
128
129 /// Submit all queued ops and block until at least `want` complete.
130 pub fn submit_and_wait(&mut self, want: u32) -> io::Result<u32> {
131 self.ring.submit_and_wait(want as usize).map(|n| n as u32)
132 }
133
134 /// Reap one completion if available.
135 pub fn reap(&mut self) -> Option<Completion> {
136 self.ring.completion().next().map(|cqe| {
137 let r = cqe.result();
138 let bytes = if r >= 0 {
139 Ok(r as usize)
140 } else {
141 Err(io::Error::from_raw_os_error(-r))
142 };
143 Completion { user_data: cqe.user_data(), bytes }
144 })
145 }
146 }
147}
148
149#[cfg(target_os = "linux")]
150pub use linux_impl::KernelAsyncRing;
151
152// ---------------------------------------------------------------------
153// FreeBSD: POSIX aio with kqueue (EVFILT_AIO) completion.
154// ---------------------------------------------------------------------
155
156#[cfg(target_os = "freebsd")]
157mod freebsd_impl {
158 use super::Completion;
159 use std::collections::{HashMap, VecDeque};
160 use std::io;
161 use std::os::unix::io::AsRawFd;
162
163 /// A kernel async-I/O ring backed by POSIX `aio` with kqueue
164 /// completion. `aio_read` issues each op immediately with its
165 /// `aio_sigevent` set to `SIGEV_KEVENT` against this ring's kqueue;
166 /// `submit_and_wait` is a `kevent` wait and `reap` calls `aio_return`.
167 pub struct KernelAsyncRing {
168 kq: i32,
169 // aiocbs in flight, keyed by their stable (boxed) address - which
170 // is exactly the kevent `ident` the kernel reports on completion,
171 // so the completion demux is a HashMap lookup.
172 inflight: HashMap<usize, Box<libc::aiocb>>,
173 // completions gathered by submit_and_wait, not yet reaped:
174 // (aiocb address = kevent ident, user_data = kevent udata).
175 ready: VecDeque<(usize, u64)>,
176 }
177
178 impl KernelAsyncRing {
179 /// Create a ring (a kqueue). `entries` is a capacity hint for the
180 /// in-flight + completion tracking maps. Fails if `kqueue(2)` does.
181 pub fn new(entries: u32) -> io::Result<Self> {
182 let kq = unsafe { libc::kqueue() };
183 if kq < 0 {
184 return Err(io::Error::last_os_error());
185 }
186 Ok(Self {
187 kq,
188 inflight: HashMap::with_capacity(entries as usize),
189 ready: VecDeque::with_capacity(entries as usize),
190 })
191 }
192
193 /// Queue a read of `len` bytes from `file` at `offset` into `buf`,
194 /// tagged with `user_data`. Open `file` with
195 /// [`open_for_async_read`](super::open_for_async_read).
196 ///
197 /// # Safety
198 /// `buf` must remain valid and not move until the matching
199 /// completion is reaped - the kernel writes into it asynchronously.
200 pub unsafe fn prepare_read(
201 &mut self,
202 file: &std::fs::File,
203 buf: *mut u8,
204 len: u32,
205 offset: u64,
206 user_data: u64,
207 ) -> io::Result<()> {
208 let mut cb: Box<libc::aiocb> = Box::new(unsafe { std::mem::zeroed() });
209 cb.aio_fildes = file.as_raw_fd();
210 cb.aio_buf = buf as *mut libc::c_void;
211 cb.aio_nbytes = len as libc::size_t;
212 cb.aio_offset = offset as libc::off_t;
213 cb.aio_sigevent.sigev_notify = libc::SIGEV_KEVENT;
214 // By FreeBSD convention sigev_notify_kqueue *is* sigev_signo
215 // (see <sys/signal.h>); the completion's udata carries our tag.
216 cb.aio_sigevent.sigev_signo = self.kq;
217 cb.aio_sigevent.sigev_value.sival_ptr = user_data as usize as *mut libc::c_void;
218 // Stable boxed address == the kevent `ident` reported back.
219 let key = &*cb as *const libc::aiocb as usize;
220 // SAFETY: `cb` is boxed (fixed address `key`) and held in
221 // `inflight` until reaped; `buf` validity is the caller's
222 // contract above.
223 let rc = unsafe { libc::aio_read(&mut *cb) };
224 if rc != 0 {
225 return Err(io::Error::last_os_error());
226 }
227 self.inflight.insert(key, cb);
228 Ok(())
229 }
230
231 /// Block until at least `want` of the in-flight reads complete,
232 /// buffering their completions for [`reap`](Self::reap). `aio_read`
233 /// already submitted them, so this only waits on the kqueue.
234 pub fn submit_and_wait(&mut self, want: u32) -> io::Result<u32> {
235 let target = (want as usize).min(self.inflight.len());
236 let mut gathered = 0u32;
237 while self.ready.len() < target {
238 let cap = self.inflight.len().max(1);
239 let mut evs: Vec<libc::kevent> =
240 (0..cap).map(|_| unsafe { std::mem::zeroed() }).collect();
241 // NULL changelist: aio_read self-registered the knotes, we
242 // only retrieve. NULL timeout: block until >= 1 posts.
243 let n = unsafe {
244 libc::kevent(
245 self.kq,
246 std::ptr::null(),
247 0,
248 evs.as_mut_ptr(),
249 cap as libc::c_int,
250 std::ptr::null(),
251 )
252 };
253 if n < 0 {
254 let e = io::Error::last_os_error();
255 if e.kind() == io::ErrorKind::Interrupted {
256 continue;
257 }
258 return Err(e);
259 }
260 for ev in &evs[..n as usize] {
261 self.ready.push_back((ev.ident, ev.udata as usize as u64));
262 gathered += 1;
263 }
264 }
265 Ok(gathered)
266 }
267
268 /// Reap one gathered completion if available, retrieving its byte
269 /// count via `aio_return` (or the error via `aio_error`).
270 pub fn reap(&mut self) -> Option<Completion> {
271 let (ident, user_data) = self.ready.pop_front()?;
272 let mut cb = self.inflight.remove(&ident)?;
273 let err = unsafe { libc::aio_error(&*cb) };
274 let ret = unsafe { libc::aio_return(&mut *cb) };
275 let bytes = if err == 0 {
276 Ok(ret.max(0) as usize)
277 } else {
278 Err(io::Error::from_raw_os_error(err))
279 };
280 Some(Completion { user_data, bytes })
281 }
282 }
283
284 impl Drop for KernelAsyncRing {
285 fn drop(&mut self) {
286 // Cancel + reclaim any still-in-flight aios so the kernel stops
287 // referencing their about-to-be-freed aiocbs, then close kq.
288 for cb in self.inflight.values_mut() {
289 unsafe {
290 libc::aio_cancel(cb.aio_fildes, &mut **cb);
291 libc::aio_return(&mut **cb);
292 }
293 }
294 unsafe { libc::close(self.kq) };
295 }
296 }
297}
298
299#[cfg(target_os = "freebsd")]
300pub use freebsd_impl::KernelAsyncRing;
301
302// ---------------------------------------------------------------------
303// macOS: POSIX aio with aio_suspend completion.
304//
305// Darwin has no `SIGEV_KEVENT`, and its `EVFILT_AIO` kqueue filter rejects
306// registration (`kevent` returns ENOTSUP), so the FreeBSD aio+kqueue path
307// does not port. `aio_read` still performs the real kernel async I/O; the
308// event-driven wait is `aio_suspend` over the in-flight set instead of a
309// `kevent` wait. The prepare / submit / reap verb shape is identical.
310// ---------------------------------------------------------------------
311
312#[cfg(target_os = "macos")]
313mod macos_impl {
314 use super::Completion;
315 use std::collections::{HashMap, VecDeque};
316 use std::io;
317 use std::os::unix::io::AsRawFd;
318
319 /// A kernel async-I/O ring backed by POSIX `aio` with `aio_suspend`
320 /// completion. `aio_read` issues each op immediately; `submit_and_wait`
321 /// blocks in `aio_suspend` until in-flight ops finish, and `reap` calls
322 /// `aio_return`. The caller's `user_data` tag rides in each aiocb's
323 /// `aio_sigevent.sigev_value` (with `SIGEV_NONE`), so completion demux
324 /// needs no side table.
325 pub struct KernelAsyncRing {
326 // aiocbs in flight, keyed by their stable (boxed) address.
327 inflight: HashMap<usize, Box<libc::aiocb>>,
328 // addresses of completed-but-not-yet-reaped aiocbs.
329 ready: VecDeque<usize>,
330 }
331
332 impl KernelAsyncRing {
333 /// Create a ring. `entries` is a capacity hint for the in-flight +
334 /// completion tracking maps. Infallible on macOS (no kernel object
335 /// is created until the first `aio_read`).
336 pub fn new(entries: u32) -> io::Result<Self> {
337 Ok(Self {
338 inflight: HashMap::with_capacity(entries as usize),
339 ready: VecDeque::with_capacity(entries as usize),
340 })
341 }
342
343 /// Queue a read of `len` bytes from `file` at `offset` into `buf`,
344 /// tagged with `user_data`. Open `file` with
345 /// [`open_for_async_read`](super::open_for_async_read).
346 ///
347 /// # Safety
348 /// `buf` must remain valid and not move until the matching
349 /// completion is reaped - the kernel writes into it asynchronously.
350 pub unsafe fn prepare_read(
351 &mut self,
352 file: &std::fs::File,
353 buf: *mut u8,
354 len: u32,
355 offset: u64,
356 user_data: u64,
357 ) -> io::Result<()> {
358 let mut cb: Box<libc::aiocb> = Box::new(unsafe { std::mem::zeroed() });
359 cb.aio_fildes = file.as_raw_fd();
360 cb.aio_buf = buf as *mut libc::c_void;
361 cb.aio_nbytes = len as libc::size_t;
362 cb.aio_offset = offset as libc::off_t;
363 // No completion event: aio_suspend polls the in-flight set. The
364 // tag rides in sigev_value so reap recovers it without a side map.
365 cb.aio_sigevent.sigev_notify = libc::SIGEV_NONE;
366 cb.aio_sigevent.sigev_value.sival_ptr = user_data as usize as *mut libc::c_void;
367 let key = &*cb as *const libc::aiocb as usize;
368 // SAFETY: `cb` is boxed (fixed address `key`) and held in
369 // `inflight` until reaped; `buf` validity is the caller's contract.
370 let rc = unsafe { libc::aio_read(&mut *cb) };
371 if rc != 0 {
372 return Err(io::Error::last_os_error());
373 }
374 self.inflight.insert(key, cb);
375 Ok(())
376 }
377
378 /// Block until at least `want` of the in-flight reads complete,
379 /// buffering their completions for [`reap`](Self::reap).
380 pub fn submit_and_wait(&mut self, want: u32) -> io::Result<u32> {
381 let target = (want as usize).min(self.inflight.len());
382 // Sweep any already-finished ops first.
383 let mut gathered = self.harvest();
384 while self.ready.len() < target {
385 // Suspend on the in-flight ops not already harvested.
386 let pending: Vec<*const libc::aiocb> = self
387 .inflight
388 .iter()
389 .filter_map(|(&k, cb)| {
390 if self.ready.contains(&k) {
391 None
392 } else {
393 Some(&**cb as *const libc::aiocb)
394 }
395 })
396 .collect();
397 if pending.is_empty() {
398 break;
399 }
400 // SAFETY: every pointer references a live boxed aiocb still
401 // owned by `inflight`. NULL timeout: block until >= 1 posts.
402 let rc = unsafe {
403 libc::aio_suspend(pending.as_ptr(), pending.len() as libc::c_int, std::ptr::null())
404 };
405 if rc != 0 {
406 let e = io::Error::last_os_error();
407 if e.kind() == io::ErrorKind::Interrupted {
408 continue;
409 }
410 return Err(e);
411 }
412 gathered += self.harvest();
413 }
414 Ok(gathered)
415 }
416
417 /// Move every in-flight aiocb whose `aio_error` is no longer
418 /// `EINPROGRESS` into `ready`; returns how many were newly added.
419 fn harvest(&mut self) -> u32 {
420 let mut newly: Vec<usize> = Vec::new();
421 for (&k, cb) in self.inflight.iter() {
422 if self.ready.contains(&k) {
423 continue;
424 }
425 // SAFETY: `cb` is a live boxed aiocb owned by `inflight`.
426 if unsafe { libc::aio_error(&**cb) } != libc::EINPROGRESS {
427 newly.push(k);
428 }
429 }
430 let added = newly.len() as u32;
431 for k in newly {
432 self.ready.push_back(k);
433 }
434 added
435 }
436
437 /// Reap one gathered completion if available, retrieving its byte
438 /// count via `aio_return` (or the error via `aio_error`).
439 pub fn reap(&mut self) -> Option<Completion> {
440 let key = self.ready.pop_front()?;
441 let mut cb = self.inflight.remove(&key)?;
442 let user_data = cb.aio_sigevent.sigev_value.sival_ptr as usize as u64;
443 // SAFETY: `cb` is still a valid aiocb whose op has completed.
444 let err = unsafe { libc::aio_error(&*cb) };
445 let ret = unsafe { libc::aio_return(&mut *cb) };
446 let bytes = if err == 0 {
447 Ok(ret.max(0) as usize)
448 } else {
449 Err(io::Error::from_raw_os_error(err))
450 };
451 Some(Completion { user_data, bytes })
452 }
453 }
454
455 impl Drop for KernelAsyncRing {
456 fn drop(&mut self) {
457 // Cancel + reclaim any still-in-flight aios so the kernel stops
458 // referencing their about-to-be-freed aiocbs.
459 for cb in self.inflight.values_mut() {
460 unsafe {
461 libc::aio_cancel(cb.aio_fildes, &mut **cb);
462 libc::aio_return(&mut **cb);
463 }
464 }
465 }
466 }
467}
468
469#[cfg(target_os = "macos")]
470pub use macos_impl::KernelAsyncRing;
471
472// ---------------------------------------------------------------------
473// Windows: IoRing via windows-sys.
474// ---------------------------------------------------------------------
475
476#[cfg(windows)]
477mod windows_impl {
478 use super::Completion;
479 use std::io;
480 use windows_sys::Win32::Storage::FileSystem::{
481 BuildIoRingReadFile, CloseIoRing, CreateIoRing, IsIoRingOpSupported,
482 PopIoRingCompletion, QueryIoRingCapabilities, SubmitIoRing, HIORING,
483 IORING_BUFFER_REF, IORING_BUFFER_REF_0, IORING_CAPABILITIES, IORING_CQE,
484 IORING_CREATE_ADVISORY_FLAGS_NONE, IORING_CREATE_FLAGS,
485 IORING_CREATE_REQUIRED_FLAGS_NONE, IORING_HANDLE_REF, IORING_HANDLE_REF_0,
486 IORING_OP_READ, IORING_REF_RAW,
487 };
488
489 const INFINITE: u32 = 0xFFFF_FFFF;
490
491 /// A kernel async-I/O ring backed by a Windows `IoRing`.
492 pub struct KernelAsyncRing {
493 ring: HIORING,
494 }
495
496 impl KernelAsyncRing {
497 /// Create a ring with `entries` submission slots. Queries the
498 /// runtime IoRing capabilities to pick a supported version + clamp
499 /// the queue sizes, and verifies READ is supported. Fails (so the
500 /// caller can fall back) on Windows builds without IoRing.
501 pub fn new(entries: u32) -> io::Result<Self> {
502 let mut caps: IORING_CAPABILITIES = unsafe { std::mem::zeroed() };
503 let hr = unsafe { QueryIoRingCapabilities(&mut caps) };
504 if hr < 0 {
505 return Err(io::Error::other(
506 "IoRing not supported on this Windows build",
507 ));
508 }
509 let sq = entries.min(caps.MaxSubmissionQueueSize.max(1));
510 let cq = entries
511 .saturating_mul(2)
512 .min(caps.MaxCompletionQueueSize.max(1));
513 let flags = IORING_CREATE_FLAGS {
514 Required: IORING_CREATE_REQUIRED_FLAGS_NONE,
515 Advisory: IORING_CREATE_ADVISORY_FLAGS_NONE,
516 };
517 let mut ring: HIORING = std::ptr::null_mut();
518 let hr = unsafe { CreateIoRing(caps.MaxVersion, flags, sq, cq, &mut ring) };
519 if hr < 0 {
520 return Err(io::Error::other(format!("CreateIoRing failed: {hr:#x}")));
521 }
522 if unsafe { IsIoRingOpSupported(ring, IORING_OP_READ) } == 0 {
523 unsafe { CloseIoRing(ring) };
524 return Err(io::Error::other("IoRing READ op not supported"));
525 }
526 Ok(Self { ring })
527 }
528
529 /// Queue a read of `len` bytes from `file` at `offset` into `buf`,
530 /// tagged with `user_data`. Open `file` with
531 /// [`open_for_async_read`](super::open_for_async_read) so its
532 /// handle carries `FILE_FLAG_OVERLAPPED`.
533 ///
534 /// # Safety
535 /// `buf` must remain valid and not move until the matching
536 /// completion is reaped - the kernel writes into it asynchronously.
537 pub unsafe fn prepare_read(
538 &mut self,
539 file: &std::fs::File,
540 buf: *mut u8,
541 len: u32,
542 offset: u64,
543 user_data: u64,
544 ) -> io::Result<()> {
545 use std::os::windows::io::AsRawHandle;
546 let fileref = IORING_HANDLE_REF {
547 Kind: IORING_REF_RAW,
548 Handle: IORING_HANDLE_REF_0 {
549 Handle: file.as_raw_handle(),
550 },
551 };
552 let dataref = IORING_BUFFER_REF {
553 Kind: IORING_REF_RAW,
554 Buffer: IORING_BUFFER_REF_0 {
555 Address: buf as *mut core::ffi::c_void,
556 },
557 };
558 // SAFETY: `dataref` points at `buf`, which the caller pledges to
559 // keep valid until the completion is reaped (this fn's contract);
560 // `fileref` wraps a live file handle.
561 let hr = unsafe {
562 BuildIoRingReadFile(
563 self.ring,
564 fileref,
565 dataref,
566 len,
567 offset,
568 user_data as usize,
569 0, // IORING_SQE_FLAGS_NONE
570 )
571 };
572 if hr < 0 {
573 Err(io::Error::other(format!("BuildIoRingReadFile failed: {hr:#x}")))
574 } else {
575 Ok(())
576 }
577 }
578
579 /// Submit all queued ops and block until at least `want` complete.
580 pub fn submit_and_wait(&mut self, want: u32) -> io::Result<u32> {
581 let mut submitted: u32 = 0;
582 let hr = unsafe { SubmitIoRing(self.ring, want, INFINITE, &mut submitted) };
583 if hr < 0 {
584 Err(io::Error::other(format!("SubmitIoRing failed: {hr:#x}")))
585 } else {
586 Ok(submitted)
587 }
588 }
589
590 /// Reap one completion if available. `PopIoRingCompletion` returns
591 /// `S_OK` (0) when it popped one and `S_FALSE` (1) when the queue
592 /// is empty.
593 pub fn reap(&mut self) -> Option<Completion> {
594 let mut cqe: IORING_CQE = unsafe { std::mem::zeroed() };
595 let hr = unsafe { PopIoRingCompletion(self.ring, &mut cqe) };
596 if hr != 0 {
597 return None; // S_FALSE (empty) or an error
598 }
599 let bytes = if cqe.ResultCode >= 0 {
600 Ok(cqe.Information)
601 } else {
602 Err(io::Error::other(format!(
603 "IoRing op failed: {:#x}",
604 cqe.ResultCode
605 )))
606 };
607 Some(Completion { user_data: cqe.UserData as u64, bytes })
608 }
609 }
610
611 impl Drop for KernelAsyncRing {
612 fn drop(&mut self) {
613 unsafe { CloseIoRing(self.ring) };
614 }
615 }
616}
617
618#[cfg(windows)]
619pub use windows_impl::KernelAsyncRing;
620
621#[cfg(test)]
622mod tests {
623 use super::*;
624 use std::io::Write;
625
626 /// Submit a single async read of a known file through the kernel ring
627 /// and verify the reaped completion returns the exact bytes. Skips
628 /// cleanly when the kernel ring is unavailable (old kernel / Windows
629 /// build, or a sandbox without io_uring).
630 #[test]
631 fn single_async_read_round_trips() {
632 let path = std::env::temp_dir().join(format!(
633 "karing_ut_{}_{:?}",
634 std::process::id(),
635 std::thread::current().id()
636 ));
637 let content = b"kernel-async-ring round trip payload";
638 {
639 let mut f = std::fs::File::create(&path).expect("create");
640 f.write_all(content).expect("write");
641 f.flush().expect("flush");
642 }
643
644 let mut ring = match KernelAsyncRing::new(8) {
645 Ok(r) => r,
646 Err(e) => {
647 eprintln!("skipping: kernel async ring unavailable ({e})");
648 std::fs::remove_file(&path).ok();
649 return;
650 }
651 };
652 let file = open_for_async_read(&path).expect("open_for_async_read");
653
654 let mut buf = vec![0u8; content.len()];
655 unsafe {
656 ring.prepare_read(&file, buf.as_mut_ptr(), buf.len() as u32, 0, 0xAB)
657 .expect("prepare_read");
658 }
659 let submitted = ring.submit_and_wait(1).expect("submit_and_wait");
660 assert!(submitted >= 1, "at least one entry submitted");
661
662 let c = ring.reap().expect("a completion");
663 assert_eq!(c.user_data, 0xAB, "completion carries the submission tag");
664 let n = c.bytes.expect("read succeeded");
665 assert_eq!(n, content.len(), "read the whole payload");
666 assert_eq!(&buf[..n], content, "bytes match the file content");
667
668 drop(file);
669 std::fs::remove_file(&path).ok();
670 }
671}