Skip to main content

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}