From f4b95c20d724df65c6ca61ad3600c944b6d0f933 Mon Sep 17 00:00:00 2001 From: Kaileshwar16 Date: Sat, 10 Oct 2026 22:17:28 +0530 Subject: [PATCH 1/3] perf(event): avoid locking manual-reset state reads and ready waits --- CHANGELOG.md | 4 + asyncband/src/event/manual_reset.rs | 107 +++++++------ benchmarks/benches/primitives/event/wait.rs | 74 +++++++++ benchmarks/manual-reset-event.md | 84 ++++++++++ .../tests/event_test/manual_reset.rs | 151 +++++++++++++++++- 5 files changed, 370 insertions(+), 50 deletions(-) create mode 100644 benchmarks/manual-reset-event.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 6f226004..a9e2c236 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,10 @@ All notable changes to this project will be documented in this file. * Prevent deadlocks in `singleflight::Group::work` and `try_work` when a duplicate key's destructor calls back into the same group. +### Improvements + +* Reduce `ManualResetEvent` state-query and already-set wait overhead, especially with concurrent readers. + ### Notable changes * Executor waker operations, including cloning, waking, and dropping, are expected not to panic; recovery from panicking waker callbacks is no longer supported. diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs index 45cfb33c..a5b8a6db 100644 --- a/asyncband/src/event/manual_reset.rs +++ b/asyncband/src/event/manual_reset.rs @@ -19,12 +19,13 @@ use std::fmt; use std::future::Future; use std::pin::Pin; use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; use std::task::Context; use std::task::Poll; use std::task::Waker; use crate::internal::mutex::Mutex; -use crate::internal::register_waker; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; use crate::internal::waker_batch::WakerBatch; @@ -71,7 +72,9 @@ use crate::internal::waker_batch::WakerBatch; /// # } /// ``` pub struct ManualResetEvent { - state: Mutex, + // Flag writes and waiter registration are serialized by the same lock. + is_set: AtomicBool, + waiters: Mutex>, } impl ManualResetEvent { @@ -85,10 +88,8 @@ impl ManualResetEvent { /// If `is_set` is `true`, waits complete immediately until the event is reset. pub const fn with_state(is_set: bool) -> Self { Self { - state: Mutex::new(State { - is_set, - waiters: WaitList::new(), - }), + is_set: AtomicBool::new(is_set), + waiters: Mutex::new(WaitList::new()), } } @@ -99,15 +100,16 @@ impl ManualResetEvent { pub fn set(&self) { let mut wakers = WakerBatch::new(); { - let mut state = self.state.lock(); - if state.is_set { + let mut waiters = self.waiters.lock(); + if self.is_set.load(Ordering::Relaxed) { return; } - state.is_set = true; + // Publish to successful observations that do not acquire the waiter lock. + self.is_set.store(true, Ordering::Release); // Detach the complete cohort before invoking any waker. A wake callback may reset the // event and register a new wait, which must belong to the state current at that point. - while let Some((_id, waiter)) = state.waiters.unlink_first_waiter(|waiter| { + while let Some((_id, waiter)) = waiters.unlink_first_waiter(|waiter| { waiter.notified = true; true }) { @@ -124,7 +126,9 @@ impl ManualResetEvent { /// Waits already released by a preceding [`set`](Self::set) remain ready. If the event is /// already unset, this has no effect. pub fn reset(&self) { - self.state.lock().is_set = false; + let _waiters = self.waiters.lock(); + // Reset cannot race ahead of a set that is still detaching its waiter cohort. + self.is_set.store(false, Ordering::Relaxed); } /// Returns whether the event is currently set. @@ -142,7 +146,7 @@ impl ManualResetEvent { /// assert!(event.is_set()); /// ``` pub fn is_set(&self) -> bool { - self.state.lock().is_set + self.is_set.load(Ordering::Acquire) } /// Attempts to wait without registering a waiter. @@ -211,46 +215,69 @@ impl ManualResetEvent { /// event is set never enqueues. A linked waiter therefore always belongs to an unset event, so /// `notified` alone decides whether a registered waiter is already committed. fn poll_wait(&self, waiter_id: &mut Option, cx: &mut Context<'_>) -> Poll<()> { - let (poll, retired_waker) = { - let mut state = self.state.lock(); + // Only an unregistered wait can skip the lock: a released waiter still owns a node. + let mut prepared_waker = if waiter_id.is_none() { + if self.is_set() { + return Poll::Ready(()); + } + Some(cx.waker().clone()) + } else { + None + }; + + let (poll, retired_waker) = loop { + let mut waiters = self.waiters.lock(); match *waiter_id { - Some(id) if state.waiters.waiter_mut(id).notified => { - let waiter = state.remove_waiter(id); + Some(id) if waiters.waiter_mut(id).notified => { + let waiter = waiters.remove_unlinked_waiter(id); *waiter_id = None; - (Poll::Ready(()), waiter.waker) + break (Poll::Ready(()), waiter.waker); } Some(id) => { debug_assert!( - !state.is_set, + !self.is_set.load(Ordering::Relaxed), "a linked waiter must belong to an unset event" ); - let waiter = state.waiters.waiter_mut(id); - assert!( - waiter.waker.is_some(), - "an unnotified waiter must retain its waker" - ); - let retired = register_waker(&mut waiter.waker, cx.waker()); - (Poll::Pending, retired) + let waiter = waiters.waiter_mut(id); + let current = waiter + .waker + .as_ref() + .expect("an unnotified waiter must retain its waker"); + if current.will_wake(cx.waker()) { + break (Poll::Pending, None); + } + if let Some(waker) = prepared_waker.take() { + break (Poll::Pending, waiter.waker.replace(waker)); + } + // Clone only a changed waker, outside the lock, then recheck commitment: + // set/reset may run while preparing the replacement. + drop(waiters); + prepared_waker = Some(cx.waker().clone()); } - None if state.is_set => (Poll::Ready(()), None), + // A set after the false fast probe must either be observed here or release + // the node that we register while holding this same lock. + None if self.is_set.load(Ordering::Relaxed) => break (Poll::Ready(()), None), None => { - *waiter_id = Some(state.waiters.push_back(Waiter { + *waiter_id = Some(waiters.push_back(Waiter { notified: false, - waker: Some(cx.waker().clone()), + waker: prepared_waker.take(), })); - (Poll::Pending, None) + break (Poll::Pending, None); } } }; + drop(prepared_waker); drop(retired_waker); poll } fn unregister_waiter(&self, id: WaiterId) { let waiter = { - let mut state = self.state.lock(); - state.remove_waiter(id) + let mut waiters = self.waiters.lock(); + // Released waits retain their detached nodes until repolled or cancelled. + waiters.unlink_waiter(id, |_| true); + waiters.remove_unlinked_waiter(id) }; drop(waiter); } @@ -264,29 +291,13 @@ impl Default for ManualResetEvent { impl fmt::Debug for ManualResetEvent { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let is_set = self.state.lock().is_set; + let is_set = self.is_set(); f.debug_struct("ManualResetEvent") .field("is_set", &is_set) .finish_non_exhaustive() } } -#[derive(Debug)] -struct State { - is_set: bool, - waiters: WaitList, -} - -impl State { - /// Removes a waiter whether or not [`ManualResetEvent::set`] already unlinked it. - fn remove_waiter(&mut self, id: WaiterId) -> Waiter { - // Unlinking is idempotent: a waiter that `set` detached keeps its node until it is removed - // here, and an unconditional predicate never declines. - self.waiters.unlink_waiter(id, |_| true); - self.waiters.remove_unlinked_waiter(id) - } -} - #[derive(Debug)] struct Waiter { notified: bool, diff --git a/benchmarks/benches/primitives/event/wait.rs b/benchmarks/benches/primitives/event/wait.rs index 5c9fc209..d9dca730 100644 --- a/benchmarks/benches/primitives/event/wait.rs +++ b/benchmarks/benches/primitives/event/wait.rs @@ -16,9 +16,14 @@ // under the License. use std::pin::pin; +use std::sync::Arc; +use std::task::Context; +use std::task::Wake; +use std::task::Waker; use asyncband::event::ManualResetEvent; use benchmarks::support::bench_context; +use benchmarks::support::defer_input_drop; use benchmarks::support::poll_pending; use benchmarks::support::poll_pinned_ready; use divan::Bencher; @@ -29,6 +34,13 @@ const WAITER_COUNTS: &[usize] = &[1, 4, 16]; const THREAD_COUNTS: &[usize] = &[1, 2, 4, 8]; const CONTENDED_SAMPLE_SIZE: u32 = 256; +#[divan::bench(args = [false, true], threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] +fn is_set(bencher: Bencher, is_set: bool) { + let event = ManualResetEvent::with_state(is_set); + + bencher.bench(|| black_box(black_box(&event).is_set())); +} + #[divan::bench(threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] fn wait_already_set(bencher: Bencher) { let event = ManualResetEvent::with_state(true); @@ -104,3 +116,65 @@ fn cancel_pending(bencher: Bencher) { black_box(event) }); } + +#[divan::bench] +fn cancel_pending_reused(bencher: Bencher) { + let mut context = bench_context(); + let event = ManualResetEvent::new(); + + bencher.bench_local(|| { + let mut wait = pin!(event.wait()); + poll_pending(wait.as_mut(), &mut context); + }); +} + +#[divan::bench] +fn repoll_pending(bencher: Bencher) { + let mut context = bench_context(); + let event = ManualResetEvent::new(); + let mut wait = pin!(event.wait()); + poll_pending(wait.as_mut(), &mut context); + + bencher.bench_local(|| poll_pending(wait.as_mut(), &mut context)); +} + +#[divan::bench] +fn replace_pending_waker(bencher: Bencher) { + struct Task; + #[allow(clippy::manual_noop_waker)] + impl Wake for Task { + fn wake(self: Arc) {} + } + + let wakers = [Waker::from(Arc::new(Task)), Waker::from(Arc::new(Task))]; + let event = ManualResetEvent::new(); + let mut wait = pin!(event.wait()); + poll_pending(wait.as_mut(), &mut Context::from_waker(&wakers[0])); + let mut next = 1; + + bencher.bench_local(|| { + poll_pending(wait.as_mut(), &mut Context::from_waker(&wakers[next])); + next ^= 1; + }); +} + +#[divan::bench(args = [1, 16, 256], sample_size = 64)] +fn set_registered(bencher: Bencher, waiter_count: usize) { + bencher + .with_inputs(|| { + let event = Arc::new(ManualResetEvent::new()); + let mut context = bench_context(); + let mut waiters = (0..waiter_count) + .map(|_| Box::pin(event.clone().wait_owned())) + .collect::>(); + for waiter in &mut waiters { + poll_pending(waiter.as_mut(), &mut context); + } + (event, waiters) + }) + .bench_local_values(|(event, waiters)| { + // Time cohort detachment and wake callbacks, excluding registration and cleanup. + event.set(); + defer_input_drop((event, waiters), ()) + }); +} diff --git a/benchmarks/manual-reset-event.md b/benchmarks/manual-reset-event.md new file mode 100644 index 00000000..e39e5e17 --- /dev/null +++ b/benchmarks/manual-reset-event.md @@ -0,0 +1,84 @@ + + +# ManualResetEvent atomic-read experiment + +This is the first experiment from [#252](https://github.com/apache/asyncband/issues/252), using the final manual-reset design in the closed, unmerged [#315](https://github.com/apache/asyncband/pull/315) as a reference. It retains `WaitList` and optimizes state observations; waiter-storage replacement remains a separate experiment. + +## Design and correctness + +`is_set` and `try_wait` use an Acquire load. An unregistered wait can complete from the same load when the event is set. `set` publishes true with Release; `set`, `reset`, and registration still hold the same waiter mutex. Reset clears the flag with Relaxed ordering because it does not publish data to successful waits. The locked recheck after a false fast probe prevents registering behind a set that has already finished detaching its cohort. + +Registered waits still own their nodes. `set` marks and unlinks the entire cohort before releasing the lock and waking tasks. Polling an old node observes its committed state even after reset; cancellation removes that node without affecting another wait. A wait created but not registered before a set/reset pulse is still pending in the next period. + +For the issue's callback requirement, initial waker cloning happens after the false probe but before locking. A repoll with an unchanged waker needs no clone. Replacing a waker checks its identity, unlocks to clone, and then rechecks commitment under the lock. Both the replaced waker and an unused prepared clone are dropped outside the lock. This adds a second lock acquisition only when preparing a changed waker; there is no panic-recovery machinery. + +Focused schedules exercise set and set/reset during initial waker preparation, set/reset during replacement, cancellation before and after commitment, and callbacks that reset and register a new wait. The destruction callback test calls `reset`, which still locks; calling the now-lock-free `is_set` would no longer detect destruction under a lock. A repeated cross-thread publication test covers `wait`, `try_wait`, and `is_set` without using its round barrier to publish the payload. + +## Measurements + +Baseline: `44fc2d8` with only this change's benchmark additions copied into it. Candidate: the implementation accompanying this report. Both use identical benchmark sources, dependencies, compiler, and release settings, with separate build directories. The latest published tag, `v0.7.3` (`46972ddd371370d831c18301940d16848511816f`), also uses the mutex-protected boolean and `WaitList`. + +Measured on an Intel Core i5-1135G7 (4 cores, 8 hardware threads), Linux x86_64, rustc 1.98.0 (`88d9e12ae`), Divan 0.1.21. Values below are medians of three run medians, alternating baseline/candidate order. CPU frequency was not fixed and these are local microbenchmarks, not application throughput measurements. The benchmark waker performs reference-counted task-waker operations but does not schedule an executor task. + +| Operation | Baseline ns | Candidate ns | Baseline / candidate | +| ----------------------------- | ----------- | ------------ | -------------------- | +| is_set (unset, 1 thread) | 11.71 | 1.04 | 11.27x | +| is_set (set, 1 thread) | 13.09 | 1.04 | 12.54x | +| is_set (set, 4 threads) | 221.20 | 2.40 | 91.98x | +| is_set (set, 8 threads) | 535.00 | 2.44 | 218.90x | +| Already-set wait (1 thread) | 15.03 | 5.83 | 2.58x | +| Already-set wait (4 threads) | 240.00 | 9.51 | 25.23x | +| Already-set wait (8 threads) | 715.20 | 13.23 | 54.06x | +| Register/cancel, fresh event | 93.36 | 92.43 | 1.01x | +| Register/cancel, reused event | 45.01 | 44.40 | 1.01x | +| Repoll with same waker | 16.25 | 16.08 | 1.01x | +| Repoll with changed waker | 34.73 | 44.56 | 0.78x | +| Register/set/complete/reset | 84.41 | 76.49 | 1.10x | +| Set/reset, no waiters | 29.52 | 29.01 | 1.02x | +| Full fan-out, 1 waiter | 158.50 | 133.90 | 1.18x | +| Full fan-out, 16 waiters | 1301.00 | 1135.00 | 1.15x | + +Values above 1 in the last column favor the candidate. Single-thread reads improve about 12.5x and already-set waits about 2.6x. Concurrent readers avoid mutex contention. Replacing a pending waker is about 28% slower in these runs, consistent with the extra lock/recheck needed to clone outside the critical section. Unchanged-waker repolls and reused registration/cancellation are essentially unchanged. The smaller lifecycle and full fan-out differences are noisy and are not a claimed improvement. + +A separate fan-out pass measures only set/detach/wake, excluding registration and future destruction. Using 200 samples of 64 iterations avoids the low sample count caused by setup costs under the general run's time limit. The median of run medians was 25.36 → 26.20 ns for one waiter, 154.70 → 153.10 ns for 16, and 2639 → 2561 ns for 256. The three candidate medians for 256 ranged from 2548 to 3122 ns, overlapping the baseline's 2611–2662 ns; this does not establish a fan-out improvement. + +On this target, `size_of` measurements are unchanged: the event is 72 bytes, the borrowed wait future 32 bytes, and the owned wait future 56 bytes. The waiter node representation is unchanged. The atomic signal occupies space previously used by the locked flag and padding; this experiment does not claim a waiter-memory reduction. + +## Reproduction + +Use a separate checkout of `44fc2d8` with the candidate's `benchmarks/benches/primitives/event/wait.rs` copied into it, leaving its library implementation untouched. Build both checkouts before timing and keep their Cargo target directories separate. Run the following in each checkout, alternating their order for three trials: + +```sh +cargo x --help +cargo x bench --help +cargo x bench --bench primitives --no-run +cargo x bench --bench primitives -- event::wait --sample-count 200 --sample-size 4096 --min-time 0.05 --max-time 0.15 --color never +cargo x bench --bench primitives -- event::wait::set_registered --sample-count 200 --sample-size 64 --color never +``` + +`cancel_pending` includes event allocation; `cancel_pending_reused` retains the event and reuses arena capacity. `waiter_fan_out` includes allocation, registration, set, and completion; `set_registered` isolates notification. The replacement benchmark alternates two distinct reference-counted wakers so that it cannot measure the unchanged-waker path by accident. + +## Waiter-storage decision + +The current shared type is `WakerSet`, whose one-word `WakerToken` is invalid after draining and has no epoch. It cannot recognize a committed wait after reset. Directly replacing `WaitList` would allow an old token to address a reused slot or would lose the old wait's commitment. Keeping `WaitList` avoids adding a generation protocol to this experiment. + +A future storage prototype could pair a token with an event-owned generation identity. An integer generation needs an explicit overflow/aliasing policy for indefinitely suspended futures; simply wrapping or panicking does not preserve unlimited reuse. An `Arc` identity can stay unique while old waits retain it, at the cost of allocation and reference counting. Either alternative needs its own future-size, queued-memory, cancellation, and fan-out measurements. No storage alternative was benchmarked here, so this report makes no performance claim about replacing `WaitList`. + +The measured state-read benefit is sufficient to retain the small atomic-read experiment for readiness-heavy use. Applications that frequently migrate pending waits between different wakers face the measured replacement cost. Lock-free state transitions and a new storage generation protocol are not necessary to obtain the read-path benefit. diff --git a/tests-integration/tests/event_test/manual_reset.rs b/tests-integration/tests/event_test/manual_reset.rs index db46cb7f..fff8f5e4 100644 --- a/tests-integration/tests/event_test/manual_reset.rs +++ b/tests-integration/tests/event_test/manual_reset.rs @@ -19,13 +19,19 @@ use std::future::Future; use std::pin::Pin; use std::pin::pin; use std::sync::Arc; +use std::sync::Barrier; use std::sync::Mutex; use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::task::Context; +use std::task::RawWaker; +use std::task::RawWakerVTable; use std::task::Wake; use std::task::Waker; +use std::thread; +use asyncband::blocking::FutureExt; use asyncband::event::ManualResetEvent; use tests_integration::WakeCounter; use tests_integration::assert_completes_without_deadlock; @@ -124,7 +130,7 @@ impl Wake for ReentrantWaker { impl Drop for ReentrantWaker { fn drop(&mut self) { - self.0.is_set(); + self.0.reset(); } } @@ -176,7 +182,7 @@ fn wakers_are_woken_and_dropped_outside_the_internal_lock() { ); } - // Cancelling a pending wait drops the registered waker, which re-enters `is_set`. The + // Cancelling a pending wait drops the registered waker, which re-enters `reset`. The // registration holds the last reference, so the drop runs here. drop(replaced); }); @@ -335,3 +341,144 @@ fn cancelling_an_owned_waiter_releases_its_waker_and_event_handle() { event.set(); assert_eq!(tracker.count(), 0); } + +// Unlike Wake's Arc clone, this exercises the callback used to clone a raw executor waker. +fn waker_on_clone(callback: impl Fn() + Send + Sync + 'static) -> Waker { + struct Hook(Box); + + unsafe fn clone(data: *const ()) -> RawWaker { + // SAFETY: Every raw waker owns an Arc; the source remains alive during clone. + let hook = unsafe { &*data.cast::() }; + (hook.0)(); + // SAFETY: The source waker keeps the allocation alive, and the returned waker owns + // exactly the additional strong reference created here. + unsafe { Arc::increment_strong_count(data.cast::()) }; + RawWaker::new(data, &VTABLE) + } + + unsafe fn drop_ref(data: *const ()) { + // SAFETY: A consuming wake or drop releases exactly its own raw Arc reference. + drop(unsafe { Arc::from_raw(data.cast::()) }); + } + + unsafe fn wake_by_ref(_: *const ()) {} + + static VTABLE: RawWakerVTable = RawWakerVTable::new(clone, drop_ref, wake_by_ref, drop_ref); + let data = Arc::into_raw(Arc::new(Hook(Box::new(callback)))).cast(); + // SAFETY: VTABLE maintains the Arc ownership rules, and Hook is Send + Sync. + unsafe { Waker::from_raw(RawWaker::new(data, &VTABLE)) } +} + +#[test] +fn registration_rechecks_a_set_during_waker_preparation() { + assert_completes_without_deadlock(|| { + for reset in [false, true] { + let event = Arc::new(ManualResetEvent::new()); + let on_clone = event.clone(); + let waker = waker_on_clone(move || { + on_clone.set(); + if reset { + on_clone.reset(); + } + }); + let mut wait = Box::pin(event.wait()); + let result = wait.as_mut().poll(&mut Context::from_waker(&waker)); + if reset { + // This wait was not registered during the pulse. It belongs to the new period. + assert!(result.is_pending()); + event.set(); + assert!(poll_once(wait.as_mut()).is_ready()); + } else { + // A false fast probe cannot cause registration after an unretracted set. + assert!(result.is_ready()); + } + } + }); +} + +#[test] +fn waker_replacement_rechecks_commitment_after_set_and_reset() { + assert_completes_without_deadlock(|| { + let event = Arc::new(ManualResetEvent::new()); + let mut wait = Box::pin(event.wait()); + assert!(poll_once(wait.as_mut()).is_pending()); + + let on_clone = event.clone(); + let replacement = waker_on_clone(move || { + on_clone.set(); + on_clone.reset(); + }); + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&replacement)) + .is_ready() + ); + assert!(!event.is_set()); + + let mut next = Box::pin(event.wait()); + assert!(poll_once(next.as_mut()).is_pending()); + event.set(); + assert!(poll_once(next.as_mut()).is_ready()); + }); +} + +#[test] +fn ready_and_unchanged_pending_waits_do_not_clone_wakers() { + let clones = Arc::new(AtomicUsize::new(0)); + let on_clone = clones.clone(); + let waker = waker_on_clone(move || { + on_clone.fetch_add(1, Ordering::Relaxed); + }); + let mut context = Context::from_waker(&waker); + let event = ManualResetEvent::with_state(true); + assert!(pin!(event.wait()).as_mut().poll(&mut context).is_ready()); + assert_eq!(clones.load(Ordering::Relaxed), 0); + + event.reset(); + let mut wait = pin!(event.wait()); + assert!(wait.as_mut().poll(&mut context).is_pending()); + assert!(wait.as_mut().poll(&mut context).is_pending()); + assert_eq!(clones.load(Ordering::Relaxed), 1); + event.set(); + assert!(wait.as_mut().poll(&mut context).is_ready()); + assert_eq!(clones.load(Ordering::Relaxed), 1); +} + +#[test] +fn concurrent_sets_and_waits_publish_state_across_reset_cycles() { + assert_completes_without_deadlock(|| { + let event = ManualResetEvent::new(); + let round = Barrier::new(2); + let value = AtomicUsize::new(0); + thread::scope(|scope| { + scope.spawn(|| { + for expected in 1..=300 { + round.wait(); + value.store(expected, Ordering::Relaxed); + event.set(); + round.wait(); + } + }); + for expected in 1..=300 { + // The opening barrier precedes publication; only the event publishes value. + round.wait(); + match expected % 3 { + 0 => event.wait().block_on(), + 1 => { + while !event.try_wait() { + thread::yield_now(); + } + } + _ => { + while !event.is_set() { + thread::yield_now(); + } + } + } + assert_eq!(value.load(Ordering::Relaxed), expected); + round.wait(); + event.reset(); + } + }); + }); +} From c4ea62aa14eff2466e0f46397510c3d429c890e9 Mon Sep 17 00:00:00 2001 From: Kaileshwar16 Date: Sun, 11 Oct 2026 11:11:00 +0530 Subject: [PATCH 2/3] docs(benchmarks): explain event measurements inline --- benchmarks/benches/primitives/event/wait.rs | 10 +++ benchmarks/manual-reset-event.md | 84 --------------------- 2 files changed, 10 insertions(+), 84 deletions(-) delete mode 100644 benchmarks/manual-reset-event.md diff --git a/benchmarks/benches/primitives/event/wait.rs b/benchmarks/benches/primitives/event/wait.rs index d9dca730..95b79420 100644 --- a/benchmarks/benches/primitives/event/wait.rs +++ b/benchmarks/benches/primitives/event/wait.rs @@ -34,6 +34,7 @@ const WAITER_COUNTS: &[usize] = &[1, 4, 16]; const THREAD_COUNTS: &[usize] = &[1, 2, 4, 8]; const CONTENDED_SAMPLE_SIZE: u32 = 256; +// All benchmark threads read the same event without concurrent state transitions. #[divan::bench(args = [false, true], threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] fn is_set(bencher: Bencher, is_set: bool) { let event = ManualResetEvent::with_state(is_set); @@ -41,6 +42,7 @@ fn is_set(bencher: Bencher, is_set: bool) { bencher.bench(|| black_box(black_box(&event).is_set())); } +// Each iteration creates and polls a fresh wait against the same already-set event. #[divan::bench(threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] fn wait_already_set(bencher: Bencher) { let event = ManualResetEvent::with_state(true); @@ -80,6 +82,7 @@ fn reused_waiter_handoff(bencher: Bencher) { }); } +// Measures the full fan-out lifecycle, including allocation, registration, set, and completion. #[divan::bench(args = WAITER_COUNTS)] fn waiter_fan_out(bencher: Bencher, waiter_count: usize) { let mut context = bench_context(); @@ -103,6 +106,7 @@ fn waiter_fan_out(bencher: Bencher, waiter_count: usize) { }); } +// A fresh event per iteration includes allocating its waiter storage. #[divan::bench] fn cancel_pending(bencher: Bencher) { let mut context = bench_context(); @@ -117,6 +121,7 @@ fn cancel_pending(bencher: Bencher) { }); } +// Reusing the event retains waiter storage capacity across registration/cancellation cycles. #[divan::bench] fn cancel_pending_reused(bencher: Bencher) { let mut context = bench_context(); @@ -128,6 +133,7 @@ fn cancel_pending_reused(bencher: Bencher) { }); } +// Initial registration is outside the timed loop; every repoll uses the same waker. #[divan::bench] fn repoll_pending(bencher: Bencher) { let mut context = bench_context(); @@ -146,6 +152,8 @@ fn replace_pending_waker(bencher: Bencher) { fn wake(self: Arc) {} } + // Alternate distinct reference-counted wakers so every timed poll replaces the registration. + // These wakers exercise clone/drop costs but do not schedule executor tasks. let wakers = [Waker::from(Arc::new(Task)), Waker::from(Arc::new(Task))]; let event = ManualResetEvent::new(); let mut wait = pin!(event.wait()); @@ -158,6 +166,8 @@ fn replace_pending_waker(bencher: Bencher) { }); } +// Isolates set/detach/wake from registration and cleanup. The smaller sample size limits setup +// costs for large cohorts; use --sample-count 200 without a time limit for repeated measurements. #[divan::bench(args = [1, 16, 256], sample_size = 64)] fn set_registered(bencher: Bencher, waiter_count: usize) { bencher diff --git a/benchmarks/manual-reset-event.md b/benchmarks/manual-reset-event.md deleted file mode 100644 index e39e5e17..00000000 --- a/benchmarks/manual-reset-event.md +++ /dev/null @@ -1,84 +0,0 @@ - - -# ManualResetEvent atomic-read experiment - -This is the first experiment from [#252](https://github.com/apache/asyncband/issues/252), using the final manual-reset design in the closed, unmerged [#315](https://github.com/apache/asyncband/pull/315) as a reference. It retains `WaitList` and optimizes state observations; waiter-storage replacement remains a separate experiment. - -## Design and correctness - -`is_set` and `try_wait` use an Acquire load. An unregistered wait can complete from the same load when the event is set. `set` publishes true with Release; `set`, `reset`, and registration still hold the same waiter mutex. Reset clears the flag with Relaxed ordering because it does not publish data to successful waits. The locked recheck after a false fast probe prevents registering behind a set that has already finished detaching its cohort. - -Registered waits still own their nodes. `set` marks and unlinks the entire cohort before releasing the lock and waking tasks. Polling an old node observes its committed state even after reset; cancellation removes that node without affecting another wait. A wait created but not registered before a set/reset pulse is still pending in the next period. - -For the issue's callback requirement, initial waker cloning happens after the false probe but before locking. A repoll with an unchanged waker needs no clone. Replacing a waker checks its identity, unlocks to clone, and then rechecks commitment under the lock. Both the replaced waker and an unused prepared clone are dropped outside the lock. This adds a second lock acquisition only when preparing a changed waker; there is no panic-recovery machinery. - -Focused schedules exercise set and set/reset during initial waker preparation, set/reset during replacement, cancellation before and after commitment, and callbacks that reset and register a new wait. The destruction callback test calls `reset`, which still locks; calling the now-lock-free `is_set` would no longer detect destruction under a lock. A repeated cross-thread publication test covers `wait`, `try_wait`, and `is_set` without using its round barrier to publish the payload. - -## Measurements - -Baseline: `44fc2d8` with only this change's benchmark additions copied into it. Candidate: the implementation accompanying this report. Both use identical benchmark sources, dependencies, compiler, and release settings, with separate build directories. The latest published tag, `v0.7.3` (`46972ddd371370d831c18301940d16848511816f`), also uses the mutex-protected boolean and `WaitList`. - -Measured on an Intel Core i5-1135G7 (4 cores, 8 hardware threads), Linux x86_64, rustc 1.98.0 (`88d9e12ae`), Divan 0.1.21. Values below are medians of three run medians, alternating baseline/candidate order. CPU frequency was not fixed and these are local microbenchmarks, not application throughput measurements. The benchmark waker performs reference-counted task-waker operations but does not schedule an executor task. - -| Operation | Baseline ns | Candidate ns | Baseline / candidate | -| ----------------------------- | ----------- | ------------ | -------------------- | -| is_set (unset, 1 thread) | 11.71 | 1.04 | 11.27x | -| is_set (set, 1 thread) | 13.09 | 1.04 | 12.54x | -| is_set (set, 4 threads) | 221.20 | 2.40 | 91.98x | -| is_set (set, 8 threads) | 535.00 | 2.44 | 218.90x | -| Already-set wait (1 thread) | 15.03 | 5.83 | 2.58x | -| Already-set wait (4 threads) | 240.00 | 9.51 | 25.23x | -| Already-set wait (8 threads) | 715.20 | 13.23 | 54.06x | -| Register/cancel, fresh event | 93.36 | 92.43 | 1.01x | -| Register/cancel, reused event | 45.01 | 44.40 | 1.01x | -| Repoll with same waker | 16.25 | 16.08 | 1.01x | -| Repoll with changed waker | 34.73 | 44.56 | 0.78x | -| Register/set/complete/reset | 84.41 | 76.49 | 1.10x | -| Set/reset, no waiters | 29.52 | 29.01 | 1.02x | -| Full fan-out, 1 waiter | 158.50 | 133.90 | 1.18x | -| Full fan-out, 16 waiters | 1301.00 | 1135.00 | 1.15x | - -Values above 1 in the last column favor the candidate. Single-thread reads improve about 12.5x and already-set waits about 2.6x. Concurrent readers avoid mutex contention. Replacing a pending waker is about 28% slower in these runs, consistent with the extra lock/recheck needed to clone outside the critical section. Unchanged-waker repolls and reused registration/cancellation are essentially unchanged. The smaller lifecycle and full fan-out differences are noisy and are not a claimed improvement. - -A separate fan-out pass measures only set/detach/wake, excluding registration and future destruction. Using 200 samples of 64 iterations avoids the low sample count caused by setup costs under the general run's time limit. The median of run medians was 25.36 → 26.20 ns for one waiter, 154.70 → 153.10 ns for 16, and 2639 → 2561 ns for 256. The three candidate medians for 256 ranged from 2548 to 3122 ns, overlapping the baseline's 2611–2662 ns; this does not establish a fan-out improvement. - -On this target, `size_of` measurements are unchanged: the event is 72 bytes, the borrowed wait future 32 bytes, and the owned wait future 56 bytes. The waiter node representation is unchanged. The atomic signal occupies space previously used by the locked flag and padding; this experiment does not claim a waiter-memory reduction. - -## Reproduction - -Use a separate checkout of `44fc2d8` with the candidate's `benchmarks/benches/primitives/event/wait.rs` copied into it, leaving its library implementation untouched. Build both checkouts before timing and keep their Cargo target directories separate. Run the following in each checkout, alternating their order for three trials: - -```sh -cargo x --help -cargo x bench --help -cargo x bench --bench primitives --no-run -cargo x bench --bench primitives -- event::wait --sample-count 200 --sample-size 4096 --min-time 0.05 --max-time 0.15 --color never -cargo x bench --bench primitives -- event::wait::set_registered --sample-count 200 --sample-size 64 --color never -``` - -`cancel_pending` includes event allocation; `cancel_pending_reused` retains the event and reuses arena capacity. `waiter_fan_out` includes allocation, registration, set, and completion; `set_registered` isolates notification. The replacement benchmark alternates two distinct reference-counted wakers so that it cannot measure the unchanged-waker path by accident. - -## Waiter-storage decision - -The current shared type is `WakerSet`, whose one-word `WakerToken` is invalid after draining and has no epoch. It cannot recognize a committed wait after reset. Directly replacing `WaitList` would allow an old token to address a reused slot or would lose the old wait's commitment. Keeping `WaitList` avoids adding a generation protocol to this experiment. - -A future storage prototype could pair a token with an event-owned generation identity. An integer generation needs an explicit overflow/aliasing policy for indefinitely suspended futures; simply wrapping or panicking does not preserve unlimited reuse. An `Arc` identity can stay unique while old waits retain it, at the cost of allocation and reference counting. Either alternative needs its own future-size, queued-memory, cancellation, and fan-out measurements. No storage alternative was benchmarked here, so this report makes no performance claim about replacing `WaitList`. - -The measured state-read benefit is sufficient to retain the small atomic-read experiment for readiness-heavy use. Applications that frequently migrate pending waits between different wakers face the measured replacement cost. Lock-free state transitions and a new storage generation protocol are not necessary to obtain the read-path benefit. From 673a13c3a0c7709619f72ab64f56b9ba94404ed9 Mon Sep 17 00:00:00 2001 From: Kaileshwar16 Date: Mon, 12 Oct 2026 17:05:25 +0530 Subject: [PATCH 3/3] perf(event): register manual-reset wakers under the lock --- asyncband/src/event/manual_reset.rs | 44 ++++-------- .../tests/event_test/manual_reset.rs | 70 ++++--------------- 2 files changed, 28 insertions(+), 86 deletions(-) diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs index a5b8a6db..f272fc44 100644 --- a/asyncband/src/event/manual_reset.rs +++ b/asyncband/src/event/manual_reset.rs @@ -26,6 +26,7 @@ use std::task::Poll; use std::task::Waker; use crate::internal::mutex::Mutex; +use crate::internal::register_waker; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; use crate::internal::waker_batch::WakerBatch; @@ -46,9 +47,9 @@ use crate::internal::waker_batch::WakerBatch; /// # Synchronization /// /// An unset-to-set transition synchronizes with the waits it releases, with waits first polled -/// while the event remains set, and with successful [`try_wait`](Self::try_wait) calls that observe -/// that set state. Memory operations sequenced before [`set`](Self::set) are therefore visible -/// after those waits complete. +/// while the event remains set, and with [`is_set`](Self::is_set) and [`try_wait`](Self::try_wait) +/// calls that observe that set state and return `true`. Memory operations sequenced before +/// [`set`](Self::set) are therefore visible after those waits complete or those calls return. /// /// A `set` call that finds the event already set does not establish this guarantee. /// @@ -216,22 +217,17 @@ impl ManualResetEvent { /// `notified` alone decides whether a registered waiter is already committed. fn poll_wait(&self, waiter_id: &mut Option, cx: &mut Context<'_>) -> Poll<()> { // Only an unregistered wait can skip the lock: a released waiter still owns a node. - let mut prepared_waker = if waiter_id.is_none() { - if self.is_set() { - return Poll::Ready(()); - } - Some(cx.waker().clone()) - } else { - None - }; + if waiter_id.is_none() && self.is_set() { + return Poll::Ready(()); + } - let (poll, retired_waker) = loop { + let (poll, retired_waker) = { let mut waiters = self.waiters.lock(); match *waiter_id { Some(id) if waiters.waiter_mut(id).notified => { let waiter = waiters.remove_unlinked_waiter(id); *waiter_id = None; - break (Poll::Ready(()), waiter.waker); + (Poll::Ready(()), waiter.waker) } Some(id) => { debug_assert!( @@ -239,35 +235,21 @@ impl ManualResetEvent { "a linked waiter must belong to an unset event" ); let waiter = waiters.waiter_mut(id); - let current = waiter - .waker - .as_ref() - .expect("an unnotified waiter must retain its waker"); - if current.will_wake(cx.waker()) { - break (Poll::Pending, None); - } - if let Some(waker) = prepared_waker.take() { - break (Poll::Pending, waiter.waker.replace(waker)); - } - // Clone only a changed waker, outside the lock, then recheck commitment: - // set/reset may run while preparing the replacement. - drop(waiters); - prepared_waker = Some(cx.waker().clone()); + (Poll::Pending, register_waker(&mut waiter.waker, cx.waker())) } // A set after the false fast probe must either be observed here or release // the node that we register while holding this same lock. - None if self.is_set.load(Ordering::Relaxed) => break (Poll::Ready(()), None), + None if self.is_set.load(Ordering::Relaxed) => (Poll::Ready(()), None), None => { *waiter_id = Some(waiters.push_back(Waiter { notified: false, - waker: prepared_waker.take(), + waker: Some(cx.waker().clone()), })); - break (Poll::Pending, None); + (Poll::Pending, None) } } }; - drop(prepared_waker); drop(retired_waker); poll } diff --git a/tests-integration/tests/event_test/manual_reset.rs b/tests-integration/tests/event_test/manual_reset.rs index fff8f5e4..32681a5c 100644 --- a/tests-integration/tests/event_test/manual_reset.rs +++ b/tests-integration/tests/event_test/manual_reset.rs @@ -370,60 +370,7 @@ fn waker_on_clone(callback: impl Fn() + Send + Sync + 'static) -> Waker { } #[test] -fn registration_rechecks_a_set_during_waker_preparation() { - assert_completes_without_deadlock(|| { - for reset in [false, true] { - let event = Arc::new(ManualResetEvent::new()); - let on_clone = event.clone(); - let waker = waker_on_clone(move || { - on_clone.set(); - if reset { - on_clone.reset(); - } - }); - let mut wait = Box::pin(event.wait()); - let result = wait.as_mut().poll(&mut Context::from_waker(&waker)); - if reset { - // This wait was not registered during the pulse. It belongs to the new period. - assert!(result.is_pending()); - event.set(); - assert!(poll_once(wait.as_mut()).is_ready()); - } else { - // A false fast probe cannot cause registration after an unretracted set. - assert!(result.is_ready()); - } - } - }); -} - -#[test] -fn waker_replacement_rechecks_commitment_after_set_and_reset() { - assert_completes_without_deadlock(|| { - let event = Arc::new(ManualResetEvent::new()); - let mut wait = Box::pin(event.wait()); - assert!(poll_once(wait.as_mut()).is_pending()); - - let on_clone = event.clone(); - let replacement = waker_on_clone(move || { - on_clone.set(); - on_clone.reset(); - }); - assert!( - wait.as_mut() - .poll(&mut Context::from_waker(&replacement)) - .is_ready() - ); - assert!(!event.is_set()); - - let mut next = Box::pin(event.wait()); - assert!(poll_once(next.as_mut()).is_pending()); - event.set(); - assert!(poll_once(next.as_mut()).is_ready()); - }); -} - -#[test] -fn ready_and_unchanged_pending_waits_do_not_clone_wakers() { +fn waits_clone_wakers_only_for_new_or_changed_registrations() { let clones = Arc::new(AtomicUsize::new(0)); let on_clone = clones.clone(); let waker = waker_on_clone(move || { @@ -439,9 +386,22 @@ fn ready_and_unchanged_pending_waits_do_not_clone_wakers() { assert!(wait.as_mut().poll(&mut context).is_pending()); assert!(wait.as_mut().poll(&mut context).is_pending()); assert_eq!(clones.load(Ordering::Relaxed), 1); + + let on_clone = clones.clone(); + let replacement = waker_on_clone(move || { + on_clone.fetch_add(1, Ordering::Relaxed); + }); + let mut replacement_context = Context::from_waker(&replacement); + assert!(wait.as_mut().poll(&mut replacement_context).is_pending()); + assert_eq!(clones.load(Ordering::Relaxed), 2); + assert!(wait.as_mut().poll(&mut replacement_context).is_pending()); + assert_eq!(clones.load(Ordering::Relaxed), 2); + event.set(); + event.reset(); + // A released waiter remains ready with a different waker, without cloning it. assert!(wait.as_mut().poll(&mut context).is_ready()); - assert_eq!(clones.load(Ordering::Relaxed), 1); + assert_eq!(clones.load(Ordering::Relaxed), 2); } #[test]