perf(executor): skip empty sync-queue drain with a pending counter

compio-executor/Cargo.toml

@@ -24,6 +24,7 @@ slotmap = { workspace = true }
 loom = { version = "0.7", features = ["checkpoint"] }
 
 [dev-dependencies]
+criterion = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 
 [target.'cfg(unix)'.dev-dependencies]
@@ -32,5 +33,9 @@ nix = { workspace = true, features = ["resource", "signal"] }
 [features]
 enable_log = ["compio-log/enable_log"]
 
+[[bench]]
+name = "schedule"
+harness = false
+
 [lints.rust]
 unexpected_cfgs = { level = "warn", check-cfg = ['cfg(loom)'] }

compio-executor/benches/schedule.rs

@@ -0,0 +1,144 @@
+//! Benchmarks for the executor scheduling hot paths.
+//!
+//! The key path exercised here is the *local* wake: a task that wakes itself
+//! on the same thread as the executor. Every such wake goes through
+//! `Local::schedule`, which piggyback-drains the cross-thread "sync" queue. On
+//! a single-threaded workload that queue is always empty, so this benchmark
+//! measures the cost of the empty-drain fast path (the subject of issue #852).
+
+use std::{
+    future::Future,
+    hint::black_box,
+    pin::{Pin, pin},
+    task::{Context, Poll, Waker},
+    thread,
+};
+
+use compio_executor::Executor;
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+
+/// A future that re-wakes itself `n` times before completing.
+///
+/// Each self-wake schedules the task again through the local path, so driving
+/// it to completion performs exactly `n` local schedules.
+struct SelfWake {
+    remaining: usize,
+}
+
+impl Future for SelfWake {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        if self.remaining == 0 {
+            Poll::Ready(())
+        } else {
+            self.remaining -= 1;
+            cx.waker().wake_by_ref();
+            Poll::Pending
+        }
+    }
+}
+
+/// A future that is woken `n` times from another thread, exercising the remote
+/// (cross-thread) schedule path and the piggyback drain on the consumer side.
+struct RemoteWake {
+    remaining: usize,
+}
+
+impl Future for RemoteWake {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        if self.remaining == 0 {
+            Poll::Ready(())
+        } else {
+            self.remaining -= 1;
+            let waker = cx.waker().clone();
+            thread::spawn(move || waker.wake());
+            Poll::Pending
+        }
+    }
+}
+
+/// Drive a same-thread future to completion, ticking as fast as possible.
+fn drive_local<F: Future + 'static>(exe: &Executor, fut: F) {
+    let handle = exe.spawn(fut);
+    let mut handle = pin!(handle);
+    let cx = &mut Context::from_waker(Waker::noop());
+    while handle.as_mut().poll(cx).is_pending() {
+        exe.tick();
+    }
+}
+
+/// Drive a future woken from other threads to completion, yielding while we
+/// wait on the waking thread.
+fn drive_remote<F: Future + 'static>(exe: &Executor, fut: F) {
+    let handle = exe.spawn(fut);
+    let mut handle = pin!(handle);
+    let cx = &mut Context::from_waker(Waker::noop());
+    while handle.as_mut().poll(cx).is_pending() {
+        exe.tick();
+        thread::yield_now();
+    }
+}
+
+fn bench_local(c: &mut Criterion) {
+    let exe = Executor::new();
+    let mut group = c.benchmark_group("local_wake");
+    for n in [1usize, 100, 1_000, 10_000] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter(|| {
+                drive_local(
+                    &exe,
+                    SelfWake {
+                        remaining: black_box(n),
+                    },
+                )
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_spawn(c: &mut Criterion) {
+    let exe = Executor::new();
+    let mut group = c.benchmark_group("spawn_ready");
+    // Baseline: spawn a batch of tasks that complete on first poll, no wakes.
+    // Isolates the per-tick sync-queue drain overhead from the wake overhead.
+    for n in [1usize, 100, 1_000] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter(|| {
+                for _ in 0..n {
+                    exe.spawn(async { black_box(()) }).detach();
+                }
+                while exe.tick() {}
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_remote(c: &mut Criterion) {
+    let exe = Executor::new();
+    let mut group = c.benchmark_group("remote_wake");
+    // Fewer iterations: cross-thread wakes are inherently noisier.
+    for n in [1usize, 10, 100] {
+        group.throughput(Throughput::Elements(n as u64));
+        group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| {
+            b.iter(|| {
+                drive_remote(
+                    &exe,
+                    RemoteWake {
+                        remaining: black_box(n),
+                    },
+                )
+            });
+        });
+    }
+    group.finish();
+}
+
+criterion_group!(schedule, bench_local, bench_spawn, bench_remote);
+criterion_main!(schedule);

compio-executor/src/lib.rs

@@ -128,9 +128,34 @@ impl Default for ExecutorConfig {
 pub(crate) struct Shared {
     waker: Option<Waker>,
     sync: ArrayQueue<TaskId>,
+    pending: AtomicUsize,
     queue: SendWrapper<TaskQueue>,
 }
 
+impl Shared {
+    /// Drain all pending cross-thread wakes into the local hot `queue`.
+    ///
+    /// Skips the expensive [`ArrayQueue::pop`] entirely when nothing has been
+    /// pushed, using a single relaxed-ish load of [`Shared::pending`] instead
+    /// of crossbeam's `SeqCst` empty check.
+    #[inline]
+    pub(crate) fn drain_sync(&self, queue: &TaskQueue) {
+        if self.pending.load(Ordering::Acquire) == 0 {
+            return;
+        }
+
+        let mut drained: usize = 0;
+        while let Some(id) = self.sync.pop() {
+            queue.make_hot(id);
+            drained += 1;
+        }
+
+        if drained != 0 {
+            self.pending.fetch_sub(drained, Ordering::Release);
+        }
+    }
+}
+
 impl Executor {
     /// Create a new executor.
     pub fn new() -> Self {
@@ -142,6 +167,7 @@ impl Executor {
         let ptr = Box::into_raw(Box::new(Shared {
             waker: config.waker.take(),
             sync: ArrayQueue::new(config.sync_queue_size),
+            pending: AtomicUsize::new(0),
             queue: SendWrapper::new(TaskQueue::new(config.local_queue_size)),
         }));
 
@@ -174,9 +200,7 @@ impl Executor {
     pub fn tick(&self) -> bool {
         let queue = self.queue();
 
-        while let Some(id) = self.shared().sync.pop() {
-            queue.make_hot(id);
-        }
+        self.shared().drain_sync(queue);
 
         for id in queue.iter_hot().take(self.config.max_interval as _) {
             queue.make_cold(id);

compio-executor/src/task/local.rs

@@ -51,10 +51,10 @@ impl<'a> Local<'a> {
 
         // SAFETY: Type invariant
         let queue = unsafe { shared.queue.get_unchecked() };
-        while let Some(id) = shared.sync.pop() {
-            trace!(?id, "Scheduling");
-            queue.make_hot(id);
-        }
+        // Piggyback any pending cross-thread wakes. The fast-path guard inside
+        // avoids touching the sync queue at all when none are pending, which
+        // is the common case for single-threaded workloads.
+        shared.drain_sync(queue);
 
         trace!(id = ?self.header().id, "Scheduling self");
 

compio-executor/src/task/remote.rs

@@ -54,12 +54,19 @@ impl<'a> Remote<'a> {
 
         crate::panic_guard!();
 
+        // Reserve a pending slot *before* pushing so the consumer's fast-path
+        // counter is always an upper bound on the queued items and its
+        // `fetch_sub` can never underflow.
+        shared.pending.fetch_add(1, Ordering::Release);
+
         let mut notified = false;
         while shared.sync.push(self.header().id).is_err() {
             if !notified && let Some(ref waker) = shared.waker {
                 waker.wake_by_ref();
                 notified = true;
             } else if self.header().state.load::<Strong>().is_cancelled() {
+                // Bailing out without pushing: release the reservation.
+                shared.pending.fetch_sub(1, Ordering::Release);
                 self.header().state.finish_scheduling();
                 return;
             } else {