feat(unix): remove thread-based impl

Author: Berrysoft

Cargo.toml

@@ -4,11 +4,9 @@ resolver = "3"
 [workspace.dependencies]
 async-io = "2.6.0"
 compio = { version = "0.19.0-rc.1", default-features = false }
-flume = "0.12.0"
 futures-executor = "0.3"
 futures-util = "0.3"
 rustix = "1"
-smol = "2"
 tokio = "1"
 
 [package]
@@ -41,25 +39,20 @@ rustix = { workspace = true, features = ["event", "io_uring"] }
 [target.'cfg(unix)'.dependencies]
 tokio = { workspace = true, optional = true, features = ["net", "time"] }
 
-smol = { workspace = true, optional = true }
+async-io = { workspace = true, optional = true }
 futures-util = { workspace = true, optional = true }
 
-flume = { workspace = true, optional = true }
-rustix = { workspace = true, optional = true, features = ["pipe"] }
-
 [dev-dependencies]
 compio = { workspace = true, default-features = false, features = ["fs"] }
 
 tokio = { workspace = true, features = ["rt", "macros", "fs", "io-util"] }
 futures-executor = { workspace = true }
-smol = { workspace = true }
 
 [features]
 io-uring = ["compio/io-uring"]
 polling = ["compio/polling"]
-futures = ["dep:flume", "dep:rustix", "dep:futures-util"]
 tokio = ["dep:tokio"]
-smol = ["dep:smol", "dep:futures-util"]
+smol = ["dep:async-io", "dep:futures-util"]
 enable_log = ["compio-log/enable_log"]
 
 [patch.crates-io]

src/lib.rs

@@ -41,10 +41,7 @@ impl<A: sys::Adapter> RuntimeCompat<A> {
                 self.runtime.current_timeout()
             };
 
-            // io-uring needs to be submitted before waiting.
-            if self.runtime.driver_type().is_iouring() {
-                self.runtime.poll_with(Some(Duration::ZERO));
-            }
+            self.runtime.submit();
 
             match self.runtime.wait(timeout).await {
                 Ok(_) => {}

src/sys/mod.rs

@@ -17,6 +17,13 @@ cfg_if::cfg_if! {
 pub trait Adapter: Sized + Deref<Target = Runtime> {
     fn new(runtime: Runtime) -> io::Result<Self>;
 
+    fn submit(&self) {
+        // io-uring needs to be submitted before waiting.
+        if self.driver_type().is_iouring() {
+            self.poll_with(Some(Duration::ZERO));
+        }
+    }
+
     async fn wait(&self, timeout: Option<Duration>) -> io::Result<()>;
 
     fn clear(&self) -> io::Result<()>;

src/sys/unix/futures.rs

@@ -2,6 +2,7 @@
 use std::os::fd::OwnedFd;
 use std::{
     io,
+    ops::Deref,
     os::fd::{AsFd, AsRawFd, BorrowedFd, RawFd},
 };
 
@@ -11,22 +12,18 @@ use mod_use::mod_use;
 #[cfg(feature = "tokio")]
 mod_use![tokio];
 
-#[cfg(feature = "futures")]
-mod_use![futures];
-
 #[cfg(feature = "smol")]
 mod_use![smol];
 
 struct UnixAdapter {
-    driver: RawFd,
+    runtime: Runtime,
     #[cfg(target_os = "linux")]
     efd: Option<OwnedFd>,
 }
 
 #[cfg(target_os = "linux")]
 impl UnixAdapter {
-    fn new(runtime: &Runtime) -> io::Result<Self> {
-        let driver = runtime.as_raw_fd();
+    fn new(runtime: Runtime) -> io::Result<Self> {
         if runtime.driver_type().is_iouring() {
             use rustix::{
                 event::{EventfdFlags, eventfd},
@@ -37,18 +34,18 @@ impl UnixAdapter {
             let efd_raw = efd.as_raw_fd();
             unsafe {
                 io_uring_register(
-                    BorrowedFd::borrow_raw(driver),
+                    BorrowedFd::borrow_raw(runtime.as_raw_fd()),
                     IoringRegisterOp::RegisterEventfd,
                     (&raw const efd_raw).cast(),
                     1,
                 )?;
             }
             Ok(Self {
-                driver,
+                runtime,
                 efd: Some(efd),
             })
         } else {
-            Ok(Self { driver, efd: None })
+            Ok(Self { runtime, efd: None })
         }
     }
 
@@ -80,11 +77,11 @@ impl AsRawFd for UnixAdapter {
             self.efd
                 .as_ref()
                 .map(|f| f.as_raw_fd())
-                .unwrap_or(self.driver)
+                .unwrap_or_else(|| self.runtime.as_raw_fd())
         }
         #[cfg(not(target_os = "linux"))]
         {
-            self.driver
+            self.runtime.as_raw_fd()
         }
     }
 }
@@ -94,3 +91,11 @@ impl AsFd for UnixAdapter {
         unsafe { BorrowedFd::borrow_raw(self.as_raw_fd()) }
     }
 }
+
+impl Deref for UnixAdapter {
+    type Target = Runtime;
+
+    fn deref(&self) -> &Self::Target {
+        &self.runtime
+    }
+}

src/sys/unix/mod.rs

@@ -1,22 +1,22 @@
-use std::{io, time::Duration};
+use std::{io, ops::Deref, time::Duration};
 
+use async_io::Async;
 use compio::runtime::Runtime;
 use futures_util::FutureExt;
-use smol::Async;
 
 use crate::{Adapter, sys::unix::UnixAdapter};
 
 pub struct SmolAdapter(Async<UnixAdapter>);
 
 impl Adapter for SmolAdapter {
-    fn new(runtime: &Runtime) -> io::Result<Self> {
+    fn new(runtime: Runtime) -> io::Result<Self> {
         Ok(Self(Async::new_nonblocking(UnixAdapter::new(runtime)?)?))
     }
 
     async fn wait(&self, timeout: Option<Duration>) -> io::Result<()> {
         let fut = self.0.readable();
         if let Some(timeout) = timeout {
-            let timer = smol::Timer::after(timeout);
+            let timer = async_io::Timer::after(timeout);
             futures_util::select! {
                 res = fut.fuse() => res,
                 _ = timer.fuse() => Err(io::ErrorKind::TimedOut.into()),
@@ -30,3 +30,11 @@ impl Adapter for SmolAdapter {
         self.0.get_ref().clear()
     }
 }
+
+impl Deref for SmolAdapter {
+    type Target = Runtime;
+
+    fn deref(&self) -> &Self::Target {
+        self.0.get_ref()
+    }
+}

src/sys/unix/smol.rs

@@ -1,4 +1,4 @@
-use std::{io, time::Duration};
+use std::{io, ops::Deref, time::Duration};
 
 use compio::runtime::Runtime;
 use tokio::io::{Interest, unix::AsyncFd};
@@ -8,7 +8,7 @@ use crate::{Adapter, sys::unix::UnixAdapter};
 pub struct TokioAdapter(AsyncFd<UnixAdapter>);
 
 impl Adapter for TokioAdapter {
-    fn new(runtime: &Runtime) -> io::Result<Self> {
+    fn new(runtime: Runtime) -> io::Result<Self> {
         Ok(Self(AsyncFd::with_interest(
             UnixAdapter::new(runtime)?,
             Interest::READABLE,
@@ -30,3 +30,11 @@ impl Adapter for TokioAdapter {
         self.0.get_ref().clear()
     }
 }
+
+impl Deref for TokioAdapter {
+    type Target = Runtime;
+
+    fn deref(&self) -> &Self::Target {
+        self.0.get_ref()
+    }
+}

src/sys/unix/tokio.rs

@@ -90,9 +90,6 @@ macro_rules! impl_adapter {
     };
 }
 
-#[cfg(feature = "futures")]
-impl_adapter!(FuturesAdapter);
-
 #[cfg(feature = "tokio")]
 impl_adapter!(TokioAdapter);
 

src/sys/windows.rs

@@ -27,18 +27,10 @@ async fn tokio() {
     test_impl::<compio_compat::TokioAdapter>().await;
 }
 
-#[cfg(feature = "futures")]
-#[test]
-fn futures() {
-    futures_executor::block_on(async {
-        test_impl::<compio_compat::FuturesAdapter>().await;
-    })
-}
-
 #[cfg(feature = "smol")]
 #[test]
 fn smol() {
-    smol::block_on(async {
+    futures_executor::block_on(async {
         test_impl::<compio_compat::SmolAdapter>().await;
     })
 }