diff --git a/CHANGELOG.md b/CHANGELOG.md index 0930a418..4f67e3e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,7 @@ All notable changes to this project will be documented in this file. ### Improvements +* Reduce `ManualResetEvent` state-query and already-set wait overhead, especially with concurrent readers. * Add a lock-free fast path for releasing `Semaphore` permits while permits remain available, which speeds up dropping `RwLock` read guards, releasing multi-permit `Semaphore` permits, and returning objects to a bounded pool that is not at capacity. `Mutex` releases are unchanged. ### Notable changes diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs index 45cfb33c..f272fc44 100644 --- a/asyncband/src/event/manual_reset.rs +++ b/asyncband/src/event/manual_reset.rs @@ -19,6 +19,8 @@ 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; @@ -45,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. /// @@ -71,7 +73,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 +89,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 +101,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 +127,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 +147,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,30 +216,32 @@ 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<()> { + // Only an unregistered wait can skip the lock: a released waiter still owns a node. + if waiter_id.is_none() && self.is_set() { + return Poll::Ready(()); + } + let (poll, retired_waker) = { - let mut state = self.state.lock(); + 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) } 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); + (Poll::Pending, register_waker(&mut waiter.waker, cx.waker())) } - 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) => (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()), })); @@ -249,8 +256,10 @@ impl ManualResetEvent { 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 +273,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..95b79420 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,15 @@ 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); + + 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); @@ -68,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(); @@ -91,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(); @@ -104,3 +120,71 @@ fn cancel_pending(bencher: Bencher) { black_box(event) }); } + +// Reusing the event retains waiter storage capacity across registration/cancellation cycles. +#[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); + }); +} + +// 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(); + 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) {} + } + + // 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()); + 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; + }); +} + +// 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 + .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/tests-integration/tests/event_test/manual_reset.rs b/tests-integration/tests/event_test/manual_reset.rs index db46cb7f..32681a5c 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,104 @@ 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 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 || { + 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); + + 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), 2); +} + +#[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(); + } + }); + }); +}