Merge pull request #250 from Berrysoft/refactor/reduce-eventfd-io

feat(driver): reduce eventfd IO

Author: Berrysoft

compio-runtime/Cargo.toml

@@ -39,6 +39,7 @@ criterion = { workspace = true, optional = true }
 crossbeam-queue = { workspace = true }
 futures-util = { workspace = true }
 once_cell = { workspace = true }
+send_wrapper = "0.6.0"
 slab = { workspace = true, optional = true }
 smallvec = "1.11.1"
 socket2 = { workspace = true }

compio-runtime/src/runtime/mod.rs

@@ -1,6 +1,7 @@
 use std::{
     any::Any,
     cell::RefCell,
+    collections::VecDeque,
     future::{ready, Future},
     io,
     panic::AssertUnwindSafe,
@@ -18,6 +19,7 @@ use compio_driver::{
 use compio_log::{debug, instrument};
 use crossbeam_queue::SegQueue;
 use futures_util::{future::Either, FutureExt};
+use send_wrapper::SendWrapper;
 use smallvec::SmallVec;
 
 pub(crate) mod op;
@@ -46,7 +48,8 @@ impl Default for FutureState {
 
 pub(crate) struct RuntimeInner {
     driver: RefCell<Proactor>,
-    runnables: Arc<SegQueue<Runnable>>,
+    local_runnables: Arc<SendWrapper<RefCell<VecDeque<Runnable>>>>,
+    sync_runnables: Arc<SegQueue<Runnable>>,
     op_runtime: RefCell<OpRuntime>,
     #[cfg(feature = "time")]
     timer_runtime: RefCell<TimerRuntime>,
@@ -56,7 +59,8 @@ impl RuntimeInner {
     pub fn new(builder: &ProactorBuilder) -> io::Result<Self> {
         Ok(Self {
             driver: RefCell::new(builder.build()?),
-            runnables: Arc::new(SegQueue::new()),
+            local_runnables: Arc::new(SendWrapper::new(RefCell::new(VecDeque::new()))),
+            sync_runnables: Arc::new(SegQueue::new()),
             op_runtime: RefCell::default(),
             #[cfg(feature = "time")]
             timer_runtime: RefCell::new(TimerRuntime::new()),
@@ -65,27 +69,42 @@ impl RuntimeInner {
 
     // Safety: be careful about the captured lifetime.
     pub unsafe fn spawn_unchecked<F: Future>(&self, future: F) -> Task<F::Output> {
-        let runnables = self.runnables.clone();
+        let local_runnables = self.local_runnables.clone();
+        let sync_runnables = self.sync_runnables.clone();
         let handle = self
             .driver
             .borrow()
             .handle()
             .expect("cannot create notify handle of the proactor");
         let schedule = move |runnable| {
-            runnables.push(runnable);
-            handle.notify().ok();
+            if local_runnables.valid() {
+                local_runnables.borrow_mut().push_back(runnable);
+            } else {
+                sync_runnables.push(runnable);
+                handle.notify().ok();
+            }
         };
         let (runnable, task) = async_task::spawn_unchecked(future, schedule);
         runnable.schedule();
         task
     }
 
     pub fn run(&self) {
+        use std::ops::Deref;
+
+        let local_runnables = self.local_runnables.deref().deref();
         loop {
-            let next_task = self.runnables.pop();
+            let next_task = local_runnables.borrow_mut().pop_front();
+            let has_local_task = next_task.is_some();
             if let Some(task) = next_task {
                 task.run();
-            } else {
+            }
+            let next_task = self.sync_runnables.pop();
+            let has_sync_task = next_task.is_some();
+            if let Some(task) = next_task {
+                task.run();
+            }
+            if !has_local_task && !has_sync_task {
                 break;
             }
         }

compio-runtime/src/runtime/time.rs

@@ -82,8 +82,8 @@ impl TimerRuntime {
     }
 
     pub fn min_timeout(&self) -> Option<Duration> {
-        let elapsed = self.time.elapsed();
         self.wheel.peek().map(|entry| {
+            let elapsed = self.time.elapsed();
             if entry.0.delay > elapsed {
                 entry.0.delay - elapsed
             } else {

compio-runtime/tests/custom_loop.rs

@@ -14,7 +14,7 @@ fn cf_run_loop() {
     use core_foundation::{
         base::TCFType,
         filedescriptor::{kCFFileDescriptorReadCallBack, CFFileDescriptor, CFFileDescriptorRef},
-        runloop::{kCFRunLoopDefaultMode, CFRunLoop, CFRunLoopRef},
+        runloop::{kCFRunLoopDefaultMode, CFRunLoop, CFRunLoopRef, CFRunLoopStop},
         string::CFStringRef,
     };
 
@@ -52,12 +52,13 @@ fn cf_run_loop() {
             }
             .detach();
             loop {
+                self.runtime.poll_with(Some(Duration::ZERO));
+
                 self.runtime.run();
                 if let Some(result) = result.take() {
                     break result;
                 }
 
-                self.runtime.poll_with(Some(Duration::ZERO));
                 self.fd_source
                     .enable_callbacks(kCFFileDescriptorReadCallBack);
                 CFRunLoop::run_in_mode(
@@ -76,8 +77,12 @@ fn cf_run_loop() {
 
         let event = Event::new();
         let handle = Arc::new(Mutex::new(Some(event.handle())));
+        let run_loop = CFRunLoop::get_current();
         let block = StackBlock::new(move || {
             handle.lock().unwrap().take().unwrap().notify();
+            unsafe {
+                CFRunLoopStop(run_loop.as_concrete_TypeRef());
+            }
         });
         extern "C" {
             fn CFRunLoopPerformBlock(rl: CFRunLoopRef, mode: CFStringRef, block: &Block<dyn Fn()>);
@@ -133,13 +138,13 @@ fn message_queue() {
             }
             .detach();
             loop {
+                self.runtime.poll_with(Some(Duration::ZERO));
+
                 self.runtime.run();
                 if let Some(result) = result.take() {
                     break result;
                 }
 
-                self.runtime.poll_with(Some(Duration::ZERO));
-
                 let timeout = self.runtime.current_timeout();
                 let timeout = match timeout {
                     Some(timeout) => timeout.as_millis() as u32,
@@ -231,13 +236,13 @@ fn glib_context() {
             }
             .detach();
             loop {
+                self.runtime.poll_with(Some(Duration::ZERO));
+
                 self.runtime.run();
                 if let Some(result) = result.take() {
                     break result;
                 }
 
-                self.runtime.poll_with(Some(Duration::ZERO));
-
                 let timeout = self.runtime.current_timeout();
                 let source_id = timeout.map(|timeout| timeout_add_local_once(timeout, || {}));