Slight performance optimizations

CHANGELOG.md

@@ -1,5 +1,12 @@
 # `vibeio` change log
 
+## `vibeio` UNRELEASED
+
+**Not yet released**
+
+- Slightly optimized async task executor performance.
+- Optimized async timer performance.
+
 ## `vibeio` 0.2.13
 
 **Released in June 12, 2026**

vibeio/Cargo.toml

@@ -21,6 +21,7 @@ oneshot = { version = "0.2.1", features = ["async", "std"] }
 rusty_pool = { version = "0.7.0", optional = true }
 once_cell = { version = "1.21.3", optional = true }
 async-channel = { version = "2.5.0", optional = true }
+smallvec = "1.15.2"
 
 [target.'cfg(unix)'.dependencies]
 mio = { version = "1.1.1", features = ["os-poll", "os-ext"] }

vibeio/src/driver/uring.rs

@@ -10,6 +10,7 @@ use io_uring::types::{SubmitArgs, Timespec};
 use io_uring::{opcode, squeue, types, IoUring};
 use mio::{Interest, Token};
 use slab::Slab;
+use smallvec::SmallVec;
 
 use crate::driver::{CompletionIoResult, Interruptor};
 use crate::{
@@ -273,9 +274,7 @@ impl UringDriver {
         // Collect wakers in a small inline array to avoid heap allocation
         // in the common case (0-8 completions per collect_completions call).
         // Most flush/wait calls produce very few completions.
-        let mut fast_wakers: [Option<Waker>; 8] = Default::default();
-        let mut fast_count = 0;
-        let mut overflow_wakers: Vec<Waker> = Vec::new();
+        let mut wakers: SmallVec<[Waker; 8]> = SmallVec::new();
 
         {
             let mut ring = self.ring.borrow_mut();
@@ -307,12 +306,7 @@ impl UringDriver {
                         _ => None,
                     };
                     if let Some(waiter) = waiter {
-                        if fast_count < fast_wakers.len() {
-                            fast_wakers[fast_count] = Some(waiter);
-                        } else {
-                            overflow_wakers.push(waiter);
-                        }
-                        fast_count += 1;
+                        wakers.push(waiter);
                     }
                     continue;
                 }
@@ -330,12 +324,7 @@ impl UringDriver {
                     state.completions.remove(token.0);
                 }
                 if let Some(waiter) = waiter {
-                    if fast_count < fast_wakers.len() {
-                        fast_wakers[fast_count] = Some(waiter);
-                    } else {
-                        overflow_wakers.push(waiter);
-                    }
-                    fast_count += 1;
+                    wakers.push(waiter);
                 }
             }
         }
@@ -344,12 +333,7 @@ impl UringDriver {
             self.submit_interrupt();
         }
 
-        for waker in fast_wakers.iter_mut().take(fast_count) {
-            if let Some(w) = waker.take() {
-                w.wake();
-            }
-        }
-        for waker in overflow_wakers {
+        for waker in wakers {
             waker.wake();
         }
 

vibeio/src/executor.rs

@@ -26,10 +26,13 @@
 //! - Tasks are polled in batches for better performance.
 //! - The runtime supports timers, blocking pools, and file I/O offloading via features.
 
+use std::alloc::{alloc, Layout};
 use std::cell::{RefCell, UnsafeCell};
 use std::collections::VecDeque;
 use std::future::Future;
+use std::mem::MaybeUninit;
 use std::pin::Pin;
+use std::ptr;
 use std::rc::Rc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
@@ -71,6 +74,30 @@ fn update_waker_slot(waiter_slot: &mut Option<Waker>, waker: &Waker) {
     }
 }
 
+/// Pin-heap-allocate a `SpawnFuture<F, T>` by writing each field directly to
+/// its final heap location via `ptr::write`. This avoids constructing the full
+/// `SpawnFuture` struct on the stack, preventing large stack-to-stack copies
+/// when `F` is a large future (e.g. contains a big I/O buffer).
+///
+/// # Safety
+/// The caller must ensure the layout is non-zero (guaranteed for `SpawnFuture`
+/// since it contains at least an `Rc` and a future).
+#[inline]
+unsafe fn pin_spawn_future<F, T>(
+    future: F,
+    state: Rc<RefCell<JoinState<T>>>,
+) -> Pin<Box<SpawnFuture<F, T>>> {
+    let layout = Layout::new::<MaybeUninit<SpawnFuture<F, T>>>();
+    // SAFETY: layout is non-zero because SpawnFuture contains Rc and F (which is
+    // 'static and therefore non-zero-sized in practice).
+    let raw = unsafe { alloc(layout) as *mut SpawnFuture<F, T> };
+    unsafe {
+        ptr::write(&mut (*raw).future, future);
+        ptr::write(&mut (*raw).state, state);
+        Pin::new_unchecked(Box::from_raw(raw))
+    }
+}
+
 struct SpawnFuture<F, T> {
     future: F,
     state: Rc<RefCell<JoinState<T>>>,
@@ -154,15 +181,15 @@ impl<T> Future for JoinHandle<T> {
 struct BlockOnNotify {
     ready: AtomicBool,
     thread_id: std::thread::ThreadId,
-    interruptor: AnyInterruptor,
+    interruptor: Arc<AnyInterruptor>,
     waiting: Arc<AtomicBool>,
     interrupt_pending: Arc<AtomicBool>,
 }
 
 impl BlockOnNotify {
     #[inline]
     fn new(
-        interruptor: AnyInterruptor,
+        interruptor: Arc<AnyInterruptor>,
         waiting: Arc<AtomicBool>,
         interrupt_pending: Arc<AtomicBool>,
     ) -> Arc<Self> {
@@ -405,14 +432,24 @@ pub(crate) fn offload_fs() -> bool {
     })
 }
 
+/// Shared runtime state for cross-thread coordination.
+///
+/// Groups the remote queue, waiting flag, interrupt flag, and interruptor
+/// into a single allocation. Tasks hold a `Weak<RuntimeShared>` instead of
+/// three separate `Arc::Weak` pointers, reducing atomic operations on wake.
+pub(crate) struct RuntimeShared {
+    pub remote_queue: SegQueue<usize>,
+    pub waiting: Arc<AtomicBool>,
+    pub interrupt_pending: Arc<AtomicBool>,
+    pub interruptor: Arc<AnyInterruptor>,
+}
+
 pub(crate) struct RuntimeInner {
     queue: Rc<UnsafeCell<VecDeque<Arc<Task>>>>,
     next_task: Rc<RefCell<Option<Arc<Task>>>>,
-    remote_queue: Arc<SegQueue<usize>>,
+    runtime_shared: Arc<RuntimeShared>,
     token_to_task: RefCell<Slab<Arc<Task>>>,
     driver: Rc<AnyDriver>,
-    waiting: Arc<AtomicBool>,
-    interrupt_pending: Arc<AtomicBool>,
     blocking_pool: Option<Box<dyn BlockingThreadPool>>,
     #[cfg(feature = "fs")]
     fs_offload: bool,
@@ -447,15 +484,21 @@ impl RuntimeInner {
     where
         T: 'static,
     {
+        self.spawn_inner(future)
+    }
+
+    #[inline]
+    fn spawn_inner<F: Future<Output = T> + 'static, T: 'static>(&self, future: F) -> JoinHandle<T> {
         let state = Rc::new(RefCell::new(JoinState {
             output: None,
             waker: None,
             canceled: false,
         }));
-        let future = Box::pin(SpawnFuture {
-            future,
-            state: state.clone(),
-        });
+
+        // Use raw allocation + ptr::write to move the future directly into its
+        // final heap location, avoiding an intermediate stack construction of
+        // SpawnFuture<F, T> which could be expensive for large futures.
+        let future = unsafe { pin_spawn_future(future, state.clone()) };
 
         let mut slab = self.token_to_task.borrow_mut();
         let vacant_slab_entry = slab.vacant_entry();
@@ -464,12 +507,9 @@ impl RuntimeInner {
             future: RefCell::new(Some(future)),
             queue: Rc::downgrade(&self.queue),
             next_task: Rc::downgrade(&self.next_task),
-            remote_queue: Arc::downgrade(&self.remote_queue),
+            runtime: Arc::downgrade(&self.runtime_shared),
             queued: AtomicBool::new(true),
             thread_id: std::thread::current().id(),
-            interruptor: self.driver.get_interruptor(),
-            waiting: Arc::downgrade(&self.waiting),
-            interrupt_pending: Arc::downgrade(&self.interrupt_pending),
             token: vacant_slab_entry.key(),
         });
         vacant_slab_entry.insert(task.clone());
@@ -508,7 +548,7 @@ impl RuntimeInner {
         if budget != 0 {
             let slab = self.token_to_task.borrow();
             while budget != 0 {
-                let Some(token) = self.remote_queue.pop() else {
+                let Some(token) = self.runtime_shared.remote_queue.pop() else {
                     break;
                 };
                 if let Some(task) = slab.get(token) {
@@ -534,13 +574,15 @@ impl RuntimeInner {
 
     #[inline]
     fn stop_waiting(&self) {
-        self.waiting.store(false, Ordering::Release);
-        self.interrupt_pending.store(false, Ordering::Release);
+        self.runtime_shared.waiting.store(false, Ordering::Release);
+        self.runtime_shared
+            .interrupt_pending
+            .store(false, Ordering::Release);
     }
 
     #[inline]
     fn should_skip_wait(&self) -> bool {
-        if self.next_task.borrow().is_some() || !self.remote_queue.is_empty() {
+        if self.next_task.borrow().is_some() || !self.runtime_shared.remote_queue.is_empty() {
             return true;
         }
 
@@ -586,15 +628,20 @@ impl Runtime {
         let _ = fs_offload;
 
         let ready_queue = Rc::new(UnsafeCell::new(VecDeque::with_capacity(4096)));
+        let interruptor = driver.get_interruptor();
+        let runtime_shared = Arc::new(RuntimeShared {
+            remote_queue: SegQueue::new(),
+            waiting: Arc::new(AtomicBool::new(false)),
+            interrupt_pending: Arc::new(AtomicBool::new(false)),
+            interruptor: Arc::new(interruptor),
+        });
         Runtime {
             inner: Some(Rc::new(RuntimeInner {
                 queue: ready_queue,
                 next_task: Rc::new(RefCell::new(None)),
-                remote_queue: Arc::new(SegQueue::new()),
+                runtime_shared,
                 token_to_task: RefCell::new(Slab::with_capacity(4096)),
                 driver: Rc::new(driver),
-                waiting: Arc::new(AtomicBool::new(false)),
-                interrupt_pending: Arc::new(AtomicBool::new(false)),
                 blocking_pool,
                 #[cfg(feature = "fs")]
                 fs_offload,
@@ -649,9 +696,9 @@ impl Runtime {
 
         let mut future = std::pin::pin!(future);
         let root_notify = BlockOnNotify::new(
-            inner.driver.get_interruptor(),
-            inner.waiting.clone(),
-            inner.interrupt_pending.clone(),
+            inner.runtime_shared.interruptor.clone(),
+            Arc::clone(&inner.runtime_shared.waiting),
+            Arc::clone(&inner.runtime_shared.interrupt_pending),
         );
         let root_waker = root_notify.waker();
         let mut batch = Vec::with_capacity(256);
@@ -684,8 +731,11 @@ impl Runtime {
                     continue;
                 }
 
-                inner.interrupt_pending.store(false, Ordering::Release);
-                inner.waiting.store(true, Ordering::Release);
+                inner
+                    .runtime_shared
+                    .interrupt_pending
+                    .store(false, Ordering::Release);
+                inner.runtime_shared.waiting.store(true, Ordering::Release);
 
                 if root_notify.is_ready() || inner.should_skip_wait() {
                     inner.stop_waiting();

vibeio/src/task.rs

@@ -5,22 +5,18 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::task::{RawWaker, RawWakerVTable, Waker};
 
-use crossbeam_queue::SegQueue;
 use futures_util::future::LocalBoxFuture;
 
-use crate::driver::AnyInterruptor;
+use crate::executor::RuntimeShared;
 
 pub struct Task {
     pub future: RefCell<Option<LocalBoxFuture<'static, ()>>>,
     pub queue: Weak<UnsafeCell<VecDeque<Arc<Task>>>>,
     pub next_task: Weak<RefCell<Option<Arc<Task>>>>,
-    pub remote_queue: std::sync::Weak<SegQueue<usize>>,
-    pub interruptor: AnyInterruptor,
+    pub runtime: std::sync::Weak<RuntimeShared>,
     pub queued: AtomicBool,
     pub thread_id: std::thread::ThreadId,
     pub token: usize,
-    pub waiting: std::sync::Weak<AtomicBool>,
-    pub interrupt_pending: std::sync::Weak<AtomicBool>,
 }
 
 impl Task {
@@ -94,25 +90,17 @@ impl Task {
             return;
         }
 
-        if !task.queued.swap(true, Ordering::Relaxed) {
-            if let Some(remote_queue) = task.remote_queue.upgrade() {
-                remote_queue.push(task.token);
+        // Cross-thread wake path
+        if let Some(shared) = task.runtime.upgrade() {
+            if !task.queued.swap(true, Ordering::Relaxed) {
+                shared.remote_queue.push(task.token);
             }
-        }
 
-        // Interrupt the driver if it's waiting
-        if task
-            .waiting
-            .upgrade()
-            .is_some_and(|waiting| waiting.load(Ordering::Acquire))
-        {
-            let should_interrupt = task
-                .interrupt_pending
-                .upgrade()
-                .is_none_or(|pending| !pending.swap(true, Ordering::AcqRel));
-            if should_interrupt {
-                // Interrupt the driver if the waker is not on the same thread as the runtime
-                task.interruptor.interrupt();
+            // Interrupt the driver if it's waiting
+            if shared.waiting.load(Ordering::Acquire)
+                && !shared.interrupt_pending.swap(true, Ordering::AcqRel)
+            {
+                shared.interruptor.interrupt();
             }
         }
     }