io-uring: add readable/writable

src/utils/buffd.rs

@@ -1,4 +1,7 @@
-use {crate::async_engine::AsyncError, thiserror::Error};
+use {
+    crate::{io_uring::IoUringError, utils::oserror::OsError},
+    thiserror::Error,
+};
 pub use {
     buf_in::BufFdIn,
     buf_out::{BufFdOut, OutBuffer, OutBufferSwapchain},
@@ -14,9 +17,9 @@ mod parser;
 #[derive(Debug, Error)]
 pub enum BufFdError {
     #[error("An IO error occurred")]
-    Io(#[source] crate::utils::oserror::OsError),
-    #[error("An async error occurred")]
-    Async(#[from] AsyncError),
+    Io(#[source] OsError),
+    #[error("An io-uring error occurred")]
+    Ring(#[from] IoUringError),
     #[error("The peer did not send a file descriptor")]
     NoFd,
     #[error("The peer sent too many file descriptors")]

src/utils/buffd/buf_in.rs

@@ -1,14 +1,15 @@
 use {
     crate::{
-        async_engine::AsyncFd,
+        io_uring::IoUring,
         utils::buffd::{BufFdError, BUF_SIZE, CMSG_BUF_SIZE, MAX_IN_FD},
     },
-    std::{collections::VecDeque, mem::MaybeUninit},
+    std::{collections::VecDeque, mem::MaybeUninit, rc::Rc},
     uapi::{c, Errno, OwnedFd, Pod},
 };
 
 pub struct BufFdIn {
-    fd: AsyncFd,
+    fd: Rc<OwnedFd>,
+    ring: Rc<IoUring>,
 
     in_fd: VecDeque<OwnedFd>,
 
@@ -19,9 +20,10 @@ pub struct BufFdIn {
 }
 
 impl BufFdIn {
-    pub fn new(fd: AsyncFd) -> Self {
+    pub fn new(fd: &Rc<OwnedFd>, ring: &Rc<IoUring>) -> Self {
         Self {
-            fd,
+            fd: fd.clone(),
+            ring: ring.clone(),
             in_fd: Default::default(),
             in_buf: Box::new([MaybeUninit::uninit(); BUF_SIZE]),
             in_cmsg_buf: Box::new([MaybeUninit::uninit(); CMSG_BUF_SIZE]),
@@ -35,7 +37,7 @@ impl BufFdIn {
         let mut offset = 0;
         while offset < bytes.len() {
             if self.read_full_(bytes, &mut offset)? {
-                self.fd.readable().await?;
+                self.ring.readable(&self.fd).await?;
             }
         }
         Ok(())

src/utils/buffd/buf_out.rs

@@ -1,6 +1,6 @@
 use {
     crate::{
-        async_engine::AsyncFd,
+        io_uring::IoUring,
         utils::buffd::{BufFdError, BUF_SIZE, CMSG_BUF_SIZE},
         wheel::{Wheel, WheelTimeoutFuture},
     },
@@ -79,16 +79,18 @@ impl OutBufferSwapchain {
 }
 
 pub struct BufFdOut {
-    fd: AsyncFd,
+    fd: Rc<OwnedFd>,
+    ring: Rc<IoUring>,
     wheel: Rc<Wheel>,
     cmsg_buf: Box<[MaybeUninit<u8>; CMSG_BUF_SIZE]>,
     fd_ids: Vec<i32>,
 }
 
 impl BufFdOut {
-    pub fn new(fd: AsyncFd, wheel: &Rc<Wheel>) -> Self {
+    pub fn new(fd: &Rc<OwnedFd>, ring: &Rc<IoUring>, wheel: &Rc<Wheel>) -> Self {
         Self {
-            fd,
+            fd: fd.clone(),
+            ring: ring.clone(),
             wheel: wheel.clone(),
             cmsg_buf: Box::new([MaybeUninit::uninit(); CMSG_BUF_SIZE]),
             fd_ids: vec![],
@@ -109,7 +111,7 @@ impl BufFdOut {
                     _ = timeout.as_mut().unwrap() => {
                         return Err(BufFdError::Timeout);
                     },
-                    res = self.fd.writable().fuse() => {
+                    res = self.ring.writable(&self.fd).fuse() => {
                         res?;
                     },
                 }
@@ -123,7 +125,7 @@ impl BufFdOut {
     pub async fn flush_no_timeout(&mut self, buf: &mut OutBuffer) -> Result<(), BufFdError> {
         while buf.read_pos < buf.write_pos {
             if self.flush_sync(buf)? {
-                self.fd.writable().await?;
+                let _ = self.ring.writable(&self.fd).await?;
             }
         }
         buf.read_pos = 0;
@@ -186,7 +188,7 @@ impl BufFdOut {
         let mut read_pos = 0;
         while read_pos < buf.len() {
             if self.flush_sync2(&mut read_pos, buf, fds)? {
-                self.fd.writable().await?;
+                self.ring.writable(&self.fd).await?;
             }
         }
         Ok(())

src/utils/bufio.rs

@@ -1,6 +1,6 @@
 use {
     crate::{
-        async_engine::{AsyncError, AsyncFd},
+        io_uring::{IoUring, IoUringError},
         utils::{
             oserror::OsError,
             queue::AsyncQueue,
@@ -26,9 +26,9 @@ pub enum BufIoError {
     #[error("Could not read from the socket")]
     ReadError(#[source] OsError),
     #[error("Cannot wait for fd to become writable")]
-    Writable(#[source] AsyncError),
+    Writable(#[source] IoUringError),
     #[error("Cannot wait for fd to become readable")]
-    Readable(#[source] AsyncError),
+    Readable(#[source] IoUringError),
     #[error("The socket is closed")]
     Closed,
 }
@@ -44,7 +44,8 @@ struct MessageOffset {
 }
 
 pub struct BufIo {
-    fd: AsyncFd,
+    fd: Rc<OwnedFd>,
+    ring: Rc<IoUring>,
     bufs: Stack<Vec<u8>>,
     outgoing: AsyncQueue<BufIoMessage>,
 }
@@ -69,9 +70,10 @@ struct Outgoing {
 }
 
 impl BufIo {
-    pub fn new(fd: AsyncFd) -> Self {
+    pub fn new(fd: &Rc<OwnedFd>, ring: &Rc<IoUring>) -> Self {
         Self {
-            fd,
+            fd: fd.clone(),
+            ring: ring.clone(),
             bufs: Default::default(),
             outgoing: Default::default(),
         }
@@ -130,7 +132,7 @@ impl BufIoIncoming {
                     if e.0 != c::EAGAIN {
                         return Err(BufIoError::ReadError(e.into()));
                     }
-                    if let Err(e) = self.bufio.fd.readable().await {
+                    if let Err(e) = self.bufio.ring.readable(&self.bufio.fd).await {
                         return Err(BufIoError::Readable(e));
                     }
                 }
@@ -184,7 +186,7 @@ impl Outgoing {
                 if e != Errno(c::EAGAIN) {
                     return Err(BufIoError::FlushError(e.into()));
                 }
-                if let Err(e) = self.bufio.fd.writable().await {
+                if let Err(e) = self.bufio.ring.writable(&self.bufio.fd).await {
                     return Err(BufIoError::Writable(e));
                 }
             }