feat(driver): add AIO support for illumos & solaris (#330)

* feat(driver): AIO for illumos & solaris

* fix(poll): simplify push_blocking loop

Author: Berrysoft

Cargo.toml

@@ -48,7 +48,7 @@ crossbeam-channel = "0.5.8"
 crossbeam-queue = "0.3.8"
 futures-channel = "0.3.29"
 futures-util = "0.3.29"
-libc = "0.2.149"
+libc = "0.2.164"
 nix = "0.29.0"
 once_cell = "1.18.0"
 os_pipe = "1.1.4"

compio-driver/build.rs

@@ -12,6 +12,7 @@ fn main() {
         ) },
         gnulinux: { all(target_os = "linux", target_env = "gnu") },
         freebsd: { target_os = "freebsd" },
-        aio: { freebsd },
+        solarish: { any(target_os = "illumos", target_os = "solaris") },
+        aio: { any(freebsd, solarish) },
     }
 }

compio-driver/src/poll/mod.rs

@@ -244,57 +244,72 @@ impl Driver {
                 Poll::Pending
             }
             Decision::Completed(res) => Poll::Ready(Ok(res)),
-            Decision::Blocking => loop {
-                if self.push_blocking(user_data) {
-                    break Poll::Pending;
-                } else {
-                    self.poll_blocking();
-                }
-            },
+            Decision::Blocking => self.push_blocking(user_data),
             #[cfg(aio)]
             Decision::Aio(AioControl { mut aiocbp, submit }) => {
                 let aiocb = unsafe { aiocbp.as_mut() };
-                // sigev_notify_kqueue
-                aiocb.aio_sigevent.sigev_signo = self.poll.as_raw_fd();
-                aiocb.aio_sigevent.sigev_notify = libc::SIGEV_KEVENT;
-                aiocb.aio_sigevent.sigev_value.sival_ptr = user_data as _;
+                #[cfg(freebsd)]
+                {
+                    // sigev_notify_kqueue
+                    aiocb.aio_sigevent.sigev_signo = self.poll.as_raw_fd();
+                    aiocb.aio_sigevent.sigev_notify = libc::SIGEV_KEVENT;
+                    aiocb.aio_sigevent.sigev_value.sival_ptr = user_data as _;
+                }
+                #[cfg(solarish)]
+                let mut notify = libc::port_notify {
+                    portnfy_port: self.poll.as_raw_fd(),
+                    portnfy_user: user_data as _,
+                };
+                #[cfg(solarish)]
+                {
+                    aiocb.aio_sigevent.sigev_notify = libc::SIGEV_PORT;
+                    aiocb.aio_sigevent.sigev_value.sival_ptr = &mut notify as *mut _ as _;
+                }
                 match syscall!(submit(aiocbp.as_ptr())) {
                     Ok(_) => Poll::Pending,
+                    // FreeBSD:
+                    //   * EOPNOTSUPP: It's on a filesystem without AIO support. Just fallback to
+                    //     blocking IO.
+                    //   * EAGAIN: The process-wide queue is full. No safe way to remove the (maybe)
+                    //     dead entries.
+                    // Solarish:
+                    //   * EAGAIN: Allocation failed.
                     Err(e)
                         if matches!(
                             e.raw_os_error(),
                             Some(libc::EOPNOTSUPP) | Some(libc::EAGAIN)
                         ) =>
                     {
-                        loop {
-                            if self.push_blocking(user_data) {
-                                return Poll::Pending;
-                            } else {
-                                self.poll_blocking();
-                            }
-                        }
+                        self.push_blocking(user_data)
                     }
                     Err(e) => Poll::Ready(Err(e)),
                 }
             }
         }
     }
 
-    fn push_blocking(&mut self, user_data: usize) -> bool {
+    fn push_blocking(&mut self, user_data: usize) -> Poll<io::Result<usize>> {
         let poll = self.poll.clone();
         let completed = self.pool_completed.clone();
-        self.pool
-            .dispatch(move || {
-                let mut op = unsafe { Key::<dyn crate::sys::OpCode>::new_unchecked(user_data) };
-                let op_pin = op.as_op_pin();
-                let res = match op_pin.operate() {
-                    Poll::Pending => unreachable!("this operation is not non-blocking"),
-                    Poll::Ready(res) => res,
-                };
-                completed.push(Entry::new(user_data, res));
-                poll.notify().ok();
-            })
-            .is_ok()
+        let mut closure = move || {
+            let mut op = unsafe { Key::<dyn crate::sys::OpCode>::new_unchecked(user_data) };
+            let op_pin = op.as_op_pin();
+            let res = match op_pin.operate() {
+                Poll::Pending => unreachable!("this operation is not non-blocking"),
+                Poll::Ready(res) => res,
+            };
+            completed.push(Entry::new(user_data, res));
+            poll.notify().ok();
+        };
+        loop {
+            match self.pool.dispatch(closure) {
+                Ok(()) => return Poll::Pending,
+                Err(e) => {
+                    closure = e.0;
+                    self.poll_blocking();
+                }
+            }
+        }
     }
 
     fn poll_blocking(&mut self) {

compio-fs/build.rs

@@ -6,6 +6,7 @@ fn main() {
             target_os = "freebsd",
             target_os = "openbsd",
             target_vendor = "apple"
-        ) }
+        ) },
+        solarish: { any(target_os = "illumos", target_os = "solaris") },
     }
 }

compio-fs/src/file.rs

@@ -1,16 +1,18 @@
 use std::{future::Future, io, mem::ManuallyDrop, panic::resume_unwind, path::Path};
 
 use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut};
+#[cfg(unix)]
+use compio_driver::op::FileStat;
 use compio_driver::{
     ToSharedFd, impl_raw_fd,
     op::{BufResultExt, CloseFile, ReadAt, Sync, WriteAt},
 };
 use compio_io::{AsyncReadAt, AsyncWriteAt};
 use compio_runtime::Attacher;
-#[cfg(unix)]
+#[cfg(all(unix, not(solarish)))]
 use {
     compio_buf::{IoVectoredBuf, IoVectoredBufMut},
-    compio_driver::op::{FileStat, ReadVectoredAt, WriteVectoredAt},
+    compio_driver::op::{ReadVectoredAt, WriteVectoredAt},
 };
 
 use crate::{Metadata, OpenOptions, Permissions};
@@ -155,7 +157,7 @@ impl AsyncReadAt for File {
         compio_runtime::submit(op).await.into_inner().map_advanced()
     }
 
-    #[cfg(unix)]
+    #[cfg(all(unix, not(solarish)))]
     async fn read_vectored_at<T: IoVectoredBufMut>(
         &self,
         buffer: T,
@@ -173,7 +175,7 @@ impl AsyncWriteAt for File {
         (&*self).write_at(buf, pos).await
     }
 
-    #[cfg(unix)]
+    #[cfg(all(unix, not(solarish)))]
     #[inline]
     async fn write_vectored_at<T: IoVectoredBuf>(
         &mut self,
@@ -191,7 +193,7 @@ impl AsyncWriteAt for &File {
         compio_runtime::submit(op).await.into_inner()
     }
 
-    #[cfg(unix)]
+    #[cfg(all(unix, not(solarish)))]
     async fn write_vectored_at<T: IoVectoredBuf>(
         &mut self,
         buffer: T,