feat(driver): reduce eventfd IO

compio-driver/src/iour/mod.rs

@@ -1,7 +1,18 @@
 #[cfg_attr(all(doc, docsrs), doc(cfg(all())))]
 #[allow(unused_imports)]
 pub use std::os::fd::{AsRawFd, OwnedFd, RawFd};
-use std::{io, os::fd::FromRawFd, pin::Pin, ptr::NonNull, sync::Arc, task::Poll, time::Duration};
+use std::{
+    io,
+    os::fd::FromRawFd,
+    pin::Pin,
+    ptr::NonNull,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+    task::Poll,
+    time::Duration,
+};
 
 use compio_log::{instrument, trace, warn};
 use crossbeam_queue::SegQueue;
@@ -72,6 +83,7 @@ pub trait OpCode {
 pub(crate) struct Driver {
     inner: IoUring<SEntry, CEntry>,
     notifier: Notifier,
+    poll: bool,
     pool: AsyncifyPool,
     pool_completed: Arc<SegQueue<Entry>>,
 }
@@ -83,7 +95,7 @@ impl Driver {
     pub fn new(builder: &ProactorBuilder) -> io::Result<Self> {
         instrument!(compio_log::Level::TRACE, "new", ?builder);
         trace!("new iour driver");
-        let notifier = Notifier::new()?;
+        let notifier = Notifier::new(!builder.poll)?;
         let mut inner = IoUring::builder().build(builder.capacity)?;
         #[allow(clippy::useless_conversion)]
         unsafe {
@@ -101,6 +113,7 @@ impl Driver {
         Ok(Self {
             inner,
             notifier,
+            poll: builder.poll,
             pool: builder.create_or_get_thread_pool(),
             pool_completed: Arc::new(SegQueue::new()),
         })
@@ -109,6 +122,7 @@ impl Driver {
     // Auto means that it choose to wait or not automatically.
     fn submit_auto(&mut self, timeout: Option<Duration>) -> io::Result<()> {
         instrument!(compio_log::Level::TRACE, "submit_auto", ?timeout);
+        self.notifier.awake.store(false, Ordering::Release);
         let res = {
             // Last part of submission queue, wait till timeout.
             if let Some(duration) = timeout {
@@ -120,6 +134,9 @@ impl Driver {
             }
         };
         trace!("submit result: {res:?}");
+        if !self.poll {
+            self.notifier.awake.store(true, Ordering::Release);
+        }
         match res {
             Ok(_) => {
                 if self.inner.completion().is_empty() {
@@ -137,13 +154,19 @@ impl Driver {
     }
 
     fn poll_entries(&mut self, entries: &mut impl Extend<Entry>) -> bool {
-        while let Some(entry) = self.pool_completed.pop() {
-            entries.extend(Some(entry));
+        let woke = self.notifier.woke.load(Ordering::Acquire);
+        if woke {
+            self.notifier.clear().expect("cannot clear notifier");
         }
 
         let mut cqueue = self.inner.completion();
         cqueue.sync();
-        let has_entry = !cqueue.is_empty();
+        let has_entry = !cqueue.is_empty() || !self.pool_completed.is_empty() || woke;
+
+        while let Some(entry) = self.pool_completed.pop() {
+            entries.extend(Some(entry));
+        }
+
         let completed_entries = cqueue.filter_map(|entry| match entry.user_data() {
             Self::CANCEL => None,
             Self::NOTIFY => {
@@ -304,17 +327,26 @@ fn timespec(duration: std::time::Duration) -> Timespec {
 #[derive(Debug)]
 struct Notifier {
     fd: OwnedFd,
+    awake: Arc<AtomicBool>,
+    woke: Arc<AtomicBool>,
 }
 
 impl Notifier {
     /// Create a new notifier.
-    fn new() -> io::Result<Self> {
+    fn new(awake: bool) -> io::Result<Self> {
         let fd = syscall!(libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK))?;
         let fd = unsafe { OwnedFd::from_raw_fd(fd) };
-        Ok(Self { fd })
+        Ok(Self {
+            fd,
+            awake: Arc::new(AtomicBool::new(awake)),
+            woke: Arc::new(AtomicBool::new(false)),
+        })
     }
 
     pub fn clear(&self) -> io::Result<()> {
+        if self.woke.swap(false, Ordering::AcqRel) {
+            return Ok(());
+        }
         loop {
             let mut buffer = [0u64];
             let res = syscall!(libc::read(
@@ -337,7 +369,11 @@ impl Notifier {
     }
 
     pub fn handle(&self) -> io::Result<NotifyHandle> {
-        Ok(NotifyHandle::new(self.fd.try_clone()?))
+        Ok(NotifyHandle::new(
+            self.fd.try_clone()?,
+            self.awake.clone(),
+            self.woke.clone(),
+        ))
     }
 }
 
@@ -350,15 +386,21 @@ impl AsRawFd for Notifier {
 /// A notify handle to the inner driver.
 pub struct NotifyHandle {
     fd: OwnedFd,
+    awake: Arc<AtomicBool>,
+    woke: Arc<AtomicBool>,
 }
 
 impl NotifyHandle {
-    pub(crate) fn new(fd: OwnedFd) -> Self {
-        Self { fd }
+    pub(crate) fn new(fd: OwnedFd, awake: Arc<AtomicBool>, woke: Arc<AtomicBool>) -> Self {
+        Self { fd, awake, woke }
     }
 
     /// Notify the inner driver.
     pub fn notify(&self) -> io::Result<()> {
+        if self.awake.load(Ordering::Acquire) {
+            self.woke.store(true, Ordering::Release);
+            return Ok(());
+        }
         let data = 1u64;
         syscall!(libc::write(
             self.fd.as_raw_fd(),

compio-driver/src/lib.rs

@@ -391,6 +391,7 @@ impl ThreadPoolBuilder {
 #[derive(Debug, Clone)]
 pub struct ProactorBuilder {
     capacity: u32,
+    poll: bool,
     pool_builder: ThreadPoolBuilder,
 }
 
@@ -405,6 +406,7 @@ impl ProactorBuilder {
     pub fn new() -> Self {
         Self {
             capacity: 1024,
+            poll: false,
             pool_builder: ThreadPoolBuilder::new(),
         }
     }
@@ -416,6 +418,15 @@ impl ProactorBuilder {
         self
     }
 
+    /// Ensure the fd of proactor is pollable.
+    ///
+    /// ## Platform specific
+    /// * io-uring: if not enabled, some waker don't wake the ring.
+    pub fn enable_poll(&mut self, poll: bool) -> &mut Self {
+        self.poll = poll;
+        self
+    }
+
     /// Set the thread number limit of the inner thread pool, if exists. The
     /// default value is 256.
     ///

compio-runtime/tests/custom_loop.rs

@@ -201,7 +201,7 @@ fn message_queue() {
 fn glib_context() {
     use std::{future::Future, time::Duration};
 
-    use compio_driver::AsRawFd;
+    use compio_driver::{AsRawFd, ProactorBuilder};
     use compio_runtime::{event::Event, Runtime};
     use glib::{timeout_add_local_once, unix_fd_add_local, ControlFlow, IOCondition, MainContext};
 
@@ -212,7 +212,12 @@ fn glib_context() {
 
     impl GLibRuntime {
         pub fn new() -> Self {
-            let runtime = Runtime::new().unwrap();
+            let mut proactor_builder = ProactorBuilder::new();
+            proactor_builder.enable_poll(true);
+            let runtime = Runtime::builder()
+                .with_proactor(proactor_builder)
+                .build()
+                .unwrap();
             let ctx = MainContext::default();
 
             unix_fd_add_local(runtime.as_raw_fd(), IOCondition::IN, |_fd, _cond| {