feat: support asyncfd and refactor uring based poller

monoio-compat/src/lib.rs

@@ -22,7 +22,7 @@ pub type UnixStreamCompat = StreamWrapper<monoio::net::UnixStream>;
 #[cfg(test)]
 mod tests {
 
-    #[monoio::test_all]
+    #[monoio::test_all(internal = true)]
     async fn test_rw() {
         let listener = monoio::net::TcpListener::bind("127.0.0.1:0").unwrap();
         let addr = listener.local_addr().unwrap();
@@ -48,7 +48,7 @@ mod tests {
         client.await;
     }
 
-    #[monoio::test_all]
+    #[monoio::test_all(internal = true)]
     async fn test_rw_unsafe() {
         let listener = monoio::net::TcpListener::bind("127.0.0.1:0").unwrap();
         let addr = listener.local_addr().unwrap();

monoio-macros/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 name = "monoio-macros"
 readme = "README.md"
 repository = "https://github.com/bytedance/monoio"
-version = "0.1.0"
+version = "0.1.1"
 
 [lib]
 proc-macro = true

monoio-macros/src/entry.rs

@@ -15,6 +15,7 @@ struct FinalConfig {
     timer_enabled: Option<bool>,
     threads: Option<u32>,
     driver: DriverType,
+    internal: Option<bool>,
 }
 
 /// Config used in case of the attribute not being able to build a valid config
@@ -23,13 +24,15 @@ const DEFAULT_ERROR_CONFIG: FinalConfig = FinalConfig {
     timer_enabled: None,
     threads: None,
     driver: DriverType::Fusion,
+    internal: None,
 };
 
 struct Configuration {
     entries: Option<(u32, Span)>,
     timer_enabled: Option<(bool, Span)>,
     threads: Option<(u32, Span)>,
     driver: Option<(DriverType, Span)>,
+    internal: Option<(bool, Span)>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -46,6 +49,7 @@ impl Configuration {
             timer_enabled: None,
             threads: None,
             driver: None,
+            internal: None,
         }
     }
 
@@ -59,6 +63,16 @@ impl Configuration {
         Ok(())
     }
 
+    fn set_internal(&mut self, enabled: syn::Lit, span: Span) -> Result<(), syn::Error> {
+        if self.internal.is_some() {
+            return Err(syn::Error::new(span, "`internal` set multiple times."));
+        }
+
+        let enabled = parse_bool(enabled, span, "internal")?;
+        self.internal = Some((enabled, span));
+        Ok(())
+    }
+
     fn set_threads(&mut self, threads: syn::Lit, span: Span) -> Result<(), syn::Error> {
         if self.threads.is_some() {
             return Err(syn::Error::new(span, "`threads` set multiple times."));
@@ -98,6 +112,7 @@ impl Configuration {
             timer_enabled: self.timer_enabled.map(|(t, _)| t),
             threads: self.threads.map(|(t, _)| t),
             driver: self.driver.map(|(d, _)| d).unwrap_or(DriverType::Fusion),
+            internal: self.internal.map(|(t, _)| t),
         })
     }
 }
@@ -199,6 +214,10 @@ fn build_config(input: syn::ItemFn, args: AttributeArgs) -> Result<FinalConfig,
                         }
                     }
                     "driver" => config.set_driver(lit.clone(), syn::spanned::Spanned::span(lit))?,
+                    "internal" => {
+                        config.set_internal(lit.clone(), syn::spanned::Spanned::span(lit))?
+                    }
+
                     name => {
                         let msg = format!(
                             "Unknown attribute {name} is specified; expected one of: \
@@ -349,7 +368,7 @@ fn parse_knobs(mut input: syn::ItemFn, is_test: bool, config: FinalConfig) -> To
     } else {
         quote! {}
     };
-    let cfg_attr = if is_test {
+    let cfg_attr = if is_test && config.internal == Some(true) {
         match config.driver {
             DriverType::Legacy => quote! {
                 #[cfg(feature = "legacy")]

monoio/Cargo.toml

@@ -8,7 +8,7 @@ license = "MIT OR Apache-2.0"
 name = "monoio"
 readme = "../README.md"
 repository = "https://github.com/bytedance/monoio"
-version = "0.2.3"
+version = "0.2.4"
 
 # common dependencies
 [dependencies]
@@ -44,7 +44,7 @@ windows-sys = { version = "0.48.0", features = [
     "Win32_Networking_WinSock",
     "Win32_System_IO",
     "Win32_Storage_FileSystem",
-    "Win32_Security"
+    "Win32_Security",
 ] }
 
 # unix dependencies
@@ -89,4 +89,12 @@ poll-io = ["tokio", "mio"]
 signal = ["ctrlc", "sync"]
 signal-termination = ["signal", "ctrlc/termination"]
 # by default both iouring and legacy are enabled
-default = ["async-cancel", "bytes", "iouring", "legacy", "macros", "utils"]
+default = [
+    "async-cancel",
+    "bytes",
+    "iouring",
+    "legacy",
+    "macros",
+    "utils",
+    "poll-io",
+]

monoio/src/driver/legacy/iocp/core.rs

@@ -47,7 +47,7 @@ impl CompletionPort {
             GetQueuedCompletionStatusEx(
                 self.handle,
                 entries.as_mut_ptr(),
-                std::cmp::min(entries.len(), u32::max_value() as usize) as u32,
+                std::cmp::min(entries.len(), u32::MAX as usize) as u32,
                 &mut count,
                 duration_millis(timeout),
                 0,

monoio/src/driver/legacy/mod.rs

@@ -10,11 +10,10 @@ use std::{
 
 use super::{
     op::{CompletionMeta, Op, OpAble},
+    poll::Poll as LegacyPoll,
     ready::{self, Ready},
-    scheduled_io::ScheduledIo,
     Driver, Inner, CURRENT,
 };
-use crate::utils::slab::Slab;
 
 #[allow(missing_docs, unreachable_pub, dead_code, unused_imports)]
 #[cfg(windows)]
@@ -26,15 +25,7 @@ mod waker;
 pub(crate) use waker::UnparkHandle;
 
 pub(crate) struct LegacyInner {
-    pub(crate) io_dispatch: Slab<ScheduledIo>,
-    #[cfg(unix)]
-    events: mio::Events,
-    #[cfg(unix)]
-    poll: mio::Poll,
-    #[cfg(windows)]
-    events: iocp::Events,
-    #[cfg(windows)]
-    poll: iocp::Poller,
+    pub(crate) poller: LegacyPoll,
 
     #[cfg(feature = "sync")]
     shared_waker: std::sync::Arc<waker::EventWaker>,
@@ -66,14 +57,10 @@ impl LegacyDriver {
     }
 
     pub(crate) fn new_with_entries(entries: u32) -> io::Result<Self> {
-        #[cfg(unix)]
-        let poll = mio::Poll::new()?;
-        #[cfg(windows)]
-        let poll = iocp::Poller::new()?;
-
+        let poller = LegacyPoll::with_capacity(entries as usize)?;
         #[cfg(all(unix, feature = "sync"))]
         let shared_waker = std::sync::Arc::new(waker::EventWaker::new(mio::Waker::new(
-            poll.registry(),
+            poller.poll.registry(),
             TOKEN_WAKEUP,
         )?));
         #[cfg(all(windows, feature = "sync"))]
@@ -87,15 +74,7 @@ impl LegacyDriver {
         let thread_id = crate::builder::BUILD_THREAD_ID.with(|id| *id);
 
         let inner = LegacyInner {
-            io_dispatch: Slab::new(),
-            #[cfg(unix)]
-            events: mio::Events::with_capacity(entries as usize),
-            #[cfg(unix)]
-            poll,
-            #[cfg(windows)]
-            events: iocp::Events::with_capacity(entries as usize),
-            #[cfg(windows)]
-            poll,
+            poller,
             #[cfg(feature = "sync")]
             shared_waker,
             #[cfg(feature = "sync")]
@@ -150,13 +129,12 @@ impl LegacyDriver {
             timeout = Some(Duration::ZERO);
         }
 
-        // here we borrow 2 mut self, but its safe.
-        let events = unsafe { &mut (*self.inner.get()).events };
-        match inner.poll.poll(events, timeout) {
+        match inner.poller.poll_inside(timeout) {
             Ok(_) => {}
             Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
             Err(e) => return Err(e),
         }
+        let events = &inner.poller.events;
         #[cfg(unix)]
         let iter = events.iter();
         #[cfg(windows)]
@@ -166,103 +144,69 @@ impl LegacyDriver {
 
             #[cfg(feature = "sync")]
             if token != TOKEN_WAKEUP {
-                inner.dispatch(token, Ready::from_mio(event));
+                LegacyPoll::dispatch(
+                    &mut inner.poller.io_dispatch,
+                    token.0,
+                    Ready::from_mio(event),
+                );
             }
 
             #[cfg(not(feature = "sync"))]
-            inner.dispatch(token, Ready::from_mio(event));
+            LegacyPoll::dispatch(
+                &mut inner.poller.io_dispatch,
+                token.0,
+                Ready::from_mio(event),
+            );
         }
         Ok(())
     }
 
     #[cfg(windows)]
+    #[inline]
     pub(crate) fn register(
         this: &Rc<UnsafeCell<LegacyInner>>,
         state: &mut iocp::SocketState,
         interest: mio::Interest,
     ) -> io::Result<usize> {
         let inner = unsafe { &mut *this.get() };
-        let io = ScheduledIo::default();
-        let token = inner.io_dispatch.insert(io);
-
-        match inner.poll.register(state, mio::Token(token), interest) {
-            Ok(_) => Ok(token),
-            Err(e) => {
-                inner.io_dispatch.remove(token);
-                Err(e)
-            }
-        }
+        inner.poller.register(state, interest)
     }
 
     #[cfg(windows)]
+    #[inline]
     pub(crate) fn deregister(
         this: &Rc<UnsafeCell<LegacyInner>>,
         token: usize,
         state: &mut iocp::SocketState,
     ) -> io::Result<()> {
         let inner = unsafe { &mut *this.get() };
-
-        // try to deregister fd first, on success we will remove it from slab.
-        match inner.poll.deregister(state) {
-            Ok(_) => {
-                inner.io_dispatch.remove(token);
-                Ok(())
-            }
-            Err(e) => Err(e),
-        }
+        inner.poller.deregister(token, state)
     }
 
     #[cfg(unix)]
+    #[inline]
     pub(crate) fn register(
         this: &Rc<UnsafeCell<LegacyInner>>,
         source: &mut impl mio::event::Source,
         interest: mio::Interest,
     ) -> io::Result<usize> {
         let inner = unsafe { &mut *this.get() };
-        let token = inner.io_dispatch.insert(ScheduledIo::new());
-
-        let registry = inner.poll.registry();
-        match registry.register(source, mio::Token(token), interest) {
-            Ok(_) => Ok(token),
-            Err(e) => {
-                inner.io_dispatch.remove(token);
-                Err(e)
-            }
-        }
+        inner.poller.register(source, interest)
     }
 
     #[cfg(unix)]
+    #[inline]
     pub(crate) fn deregister(
         this: &Rc<UnsafeCell<LegacyInner>>,
         token: usize,
         source: &mut impl mio::event::Source,
     ) -> io::Result<()> {
         let inner = unsafe { &mut *this.get() };
-
-        // try to deregister fd first, on success we will remove it from slab.
-        match inner.poll.registry().deregister(source) {
-            Ok(_) => {
-                inner.io_dispatch.remove(token);
-                Ok(())
-            }
-            Err(e) => Err(e),
-        }
+        inner.poller.deregister(source, token)
     }
 }
 
 impl LegacyInner {
-    fn dispatch(&mut self, token: mio::Token, ready: Ready) {
-        let mut sio = match self.io_dispatch.get(token.0) {
-            Some(io) => io,
-            None => {
-                return;
-            }
-        };
-        let ref_mut = sio.as_mut();
-        ref_mut.set_readiness(|curr| curr | ready);
-        ref_mut.wake(ready);
-    }
-
     pub(crate) fn poll_op<T: OpAble>(
         this: &Rc<UnsafeCell<Self>>,
         data: &mut T,
@@ -282,7 +226,11 @@ impl LegacyInner {
         };
 
         // wait io ready and do syscall
-        let mut scheduled_io = inner.io_dispatch.get(index).expect("scheduled_io lost");
+        let mut scheduled_io = inner
+            .poller
+            .io_dispatch
+            .get(index)
+            .expect("scheduled_io lost");
         let ref_mut = scheduled_io.as_mut();
 
         let readiness = ready!(ref_mut.poll_readiness(cx, direction));
@@ -316,15 +264,16 @@ impl LegacyInner {
 
     pub(crate) fn cancel_op(
         this: &Rc<UnsafeCell<LegacyInner>>,
-        index: usize,
+        token: usize,
         direction: ready::Direction,
     ) {
         let inner = unsafe { &mut *this.get() };
         let ready = match direction {
             ready::Direction::Read => Ready::READ_CANCELED,
             ready::Direction::Write => Ready::WRITE_CANCELED,
+            ready::Direction::ReadOrWrite => Ready::CANCELED,
         };
-        inner.dispatch(mio::Token(index), ready);
+        LegacyPoll::dispatch(&mut inner.poller.io_dispatch, token, ready);
     }
 
     pub(crate) fn submit_with_data<T>(

monoio/src/driver/mod.rs

@@ -1,7 +1,7 @@
 /// Monoio Driver.
 #[allow(dead_code)]
 pub(crate) mod op;
-#[cfg(all(feature = "poll-io", unix))]
+#[cfg(any(feature = "legacy", feature = "poll-io"))]
 pub(crate) mod poll;
 #[cfg(any(feature = "legacy", feature = "poll-io"))]
 pub(crate) mod ready;

monoio/src/driver/op.rs

@@ -13,7 +13,7 @@ mod accept;
 mod connect;
 mod fsync;
 mod open;
-mod poll;
+pub(crate) mod poll;
 mod read;
 mod recv;
 mod send;