diff --git a/compio-executor/Cargo.toml b/compio-executor/Cargo.toml index 2f75b6e2..262f3a35 100644 --- a/compio-executor/Cargo.toml +++ b/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)'] } diff --git a/compio-executor/benches/schedule.rs b/compio-executor/benches/schedule.rs new file mode 100644 index 00000000..dde7b7f2 --- /dev/null +++ b/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(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(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); diff --git a/compio-executor/src/lib.rs b/compio-executor/src/lib.rs index e181a27c..503273ea 100644 --- a/compio-executor/src/lib.rs +++ b/compio-executor/src/lib.rs @@ -128,9 +128,34 @@ impl Default for ExecutorConfig { pub(crate) struct Shared { waker: Option, sync: ArrayQueue, + pending: AtomicUsize, queue: SendWrapper, } +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); diff --git a/compio-executor/src/task/local.rs b/compio-executor/src/task/local.rs index d04a97ef..3cd69d72 100644 --- a/compio-executor/src/task/local.rs +++ b/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"); diff --git a/compio-executor/src/task/remote.rs b/compio-executor/src/task/remote.rs index f6e3e9b5..953cfb72 100644 --- a/compio-executor/src/task/remote.rs +++ b/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::().is_cancelled() { + // Bailing out without pushing: release the reservation. + shared.pending.fetch_sub(1, Ordering::Release); self.header().state.finish_scheduling(); return; } else {