From c41a0f7d4a33a90bee16518252b0d23fe3e28c3e Mon Sep 17 00:00:00 2001 From: Orthur Date: Sun, 30 Aug 2026 05:08:18 -0400 Subject: [PATCH 1/5] feat(event): add a manual-reset event --- CHANGELOG.md | 1 + README.md | 43 +- asyncband/Cargo.toml | 1 + asyncband/src/event/mod.rs | 455 +++++++++++++++++++++ asyncband/src/internal/mod.rs | 10 +- asyncband/src/lib.rs | 45 ++- benchmarks/Cargo.toml | 1 + benchmarks/asyncband/event/mod.rs | 18 + benchmarks/asyncband/event/wait.rs | 130 ++++++ benchmarks/asyncband/main.rs | 1 + tests-integration/Cargo.toml | 1 + tests-integration/tests/event_test.rs | 537 +++++++++++++++++++++++++ tests-integration/tests/traits_test.rs | 9 + 13 files changed, 1209 insertions(+), 43 deletions(-) create mode 100644 asyncband/src/event/mod.rs create mode 100644 benchmarks/asyncband/event/mod.rs create mode 100644 benchmarks/asyncband/event/wait.rs create mode 100644 tests-integration/tests/event_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 847c4a8..d161cb6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ All notable changes to this project will be documented in this file. * Implement `broadcast::mpmc::unbounded`, an unbounded broadcast channel that retains messages until all active receivers consume them or are dropped. * Add an opt-in latest-state channel under `asyncband::watch`. +* Add opt-in `asyncband::event::ManualResetEvent`, a reusable level-triggered signal that releases every waiter registered before a set, stays ready until an explicit reset, and commits released waiters even when a reset races their next poll. * Add opt-in `asyncband::once::LazyCell` for values that own one asynchronous initializer and preserve its in-flight future across caller cancellation. * Add opt-in bounded and unbounded runtime-agnostic object pools under `asyncband::pool`. * Add an opt-in `asyncband::blocking::FutureExt` bridge with `block_on` and `wait_timeout` methods for waiting on runtime-agnostic futures from synchronous code. diff --git a/README.md b/README.md index 2cb481e..4c6443d 100644 --- a/README.md +++ b/README.md @@ -62,27 +62,28 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon ## API map -| Area | API | Feature | Use | -|-----------------------|--------------------------------------------------------------------------------------|----------------|--------------------------------------------------------------------------------------------------------| -| Shared state | [`Mutex`](https://docs.rs/asyncband/*/asyncband/mutex/struct.Mutex.html) | `mutex` | Protect shared data with asynchronous mutual exclusion. | -| | [`RwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.RwLock.html) | `rwlock` | Allow multiple readers or one writer. | -| | [`Condvar`](https://docs.rs/asyncband/*/asyncband/condvar/struct.Condvar.html) | `condvar` | Wait for notifications while releasing a mutex. | -| Initialization | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Run asynchronous initialization exactly once. | -| | [`OnceCell`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceCell.html) | `once-cell` | Initialize and store one asynchronous value. | -| | [`LazyCell`](https://docs.rs/asyncband/*/asyncband/once/struct.LazyCell.html) | `lazy-cell` | Lazily initialize a value with a stored asynchronous function. | -| | [`OnceMap`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceMap.html) | `once-map` | Initialize and store one value per key. | -| Task coordination | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Wait until all participants reach a synchronization point. | -| | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a one-way countdown completes. | -| | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Wait for a dynamic group of tasks to finish. | -| | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Coordinate shutdown signals and completion. | -| Channels | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. | -| | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver through a bounded or unbounded queue. | -| | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Broadcast values from one or more producers and retain them until every active receiver consumes them. | -| | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Publish the latest state to independently tracked receivers and coalesce intermediate updates. | -| Resource reuse | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/) | `pool` | Reuse objects through bounded or unbounded pool variants. | -| Workload coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Control concurrent access with permits. | -| | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce concurrent calls for the same key. | -| Sync interop | [`FutureExt`](https://docs.rs/asyncband/*/asyncband/blocking/trait.FutureExt.html) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | +| Area | API | Feature | Use | +|-----------------------|------------------------------------------------------------------------------------------------|----------------|--------------------------------------------------------------------------------------------------------| +| Shared state | [`Mutex`](https://docs.rs/asyncband/*/asyncband/mutex/struct.Mutex.html) | `mutex` | Protect shared data with asynchronous mutual exclusion. | +| | [`RwLock`](https://docs.rs/asyncband/*/asyncband/rwlock/struct.RwLock.html) | `rwlock` | Allow multiple readers or one writer. | +| | [`Condvar`](https://docs.rs/asyncband/*/asyncband/condvar/struct.Condvar.html) | `condvar` | Wait for notifications while releasing a mutex. | +| Initialization | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Run asynchronous initialization exactly once. | +| | [`OnceCell`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceCell.html) | `once-cell` | Initialize and store one asynchronous value. | +| | [`LazyCell`](https://docs.rs/asyncband/*/asyncband/once/struct.LazyCell.html) | `lazy-cell` | Lazily initialize a value with a stored asynchronous function. | +| | [`OnceMap`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceMap.html) | `once-map` | Initialize and store one value per key. | +| Task coordination | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Wait until all participants reach a synchronization point. | +| | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Reuse a level-triggered signal that releases all current waiters. | +| | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a one-way countdown completes. | +| | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Wait for a dynamic group of tasks to finish. | +| | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Coordinate shutdown signals and completion. | +| Channels | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. | +| | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver through a bounded or unbounded queue. | +| | [`broadcast`](https://docs.rs/asyncband/*/asyncband/broadcast/) | `broadcast` | Broadcast values from one or more producers and retain them until every active receiver consumes them. | +| | [`watch`](https://docs.rs/asyncband/*/asyncband/watch/) | `watch` | Publish the latest state to independently tracked receivers and coalesce intermediate updates. | +| Resource reuse | [`pool`](https://docs.rs/asyncband/*/asyncband/pool/) | `pool` | Reuse objects through bounded or unbounded pool variants. | +| Workload coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Control concurrent access with permits. | +| | [`Group`](https://docs.rs/asyncband/*/asyncband/singleflight/struct.Group.html) | `singleflight` | Coalesce concurrent calls for the same key. | +| Sync interop | [`FutureExt`](https://docs.rs/asyncband/*/asyncband/blocking/trait.FutureExt.html) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | ## Synchronous interoperability diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index 9b831d7..61f2aab 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -48,6 +48,7 @@ barrier = [] blocking = [] broadcast = [] condvar = ["mutex"] +event = [] latch = [] lazy-cell = ["mutex"] mpsc = [] diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs new file mode 100644 index 0000000..647e0d3 --- /dev/null +++ b/asyncband/src/event/mod.rs @@ -0,0 +1,455 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A reusable, level-triggered signal for coordinating tasks. +//! +//! A [`ManualResetEvent`] is either set or unset. Calling [`set`](ManualResetEvent::set) makes the +//! event ready, releases every waiter registered during the current unset period, and keeps later +//! waits ready. Calling [`reset`](ManualResetEvent::reset) makes subsequent waits block again. +//! +//! A waiter registered before `set` is committed to completion even if another task calls `reset` +//! before the waiter is polled again. This differs from a condition variable: the event retains its +//! set state and does not require an external predicate or mutex. +//! +//! Registration happens on a wait's first poll, so `set` followed immediately by `reset` is not a +//! reliable way to release everyone waiting at that moment: a future constructed before the `set` +//! but not yet polled was never a waiter of it. Keep the event set for as long as the condition it +//! reports holds. +//! +//! # Examples +//! +//! ``` +//! # #[tokio::main] +//! # async fn main() { +//! use std::sync::Arc; +//! +//! use asyncband::event::ManualResetEvent; +//! +//! let ready = Arc::new(ManualResetEvent::new()); +//! let waiter = tokio::spawn(ready.clone().wait_owned()); +//! +//! ready.set(); +//! waiter.await.unwrap(); +//! +//! // The signal remains set until it is reset explicitly. +//! ready.wait().await; +//! ready.reset(); +//! # } +//! ``` + +use std::fmt; +use std::future::Future; +use std::mem; +use std::pin::Pin; +use std::sync::Arc; +use std::task::Context; +use std::task::Poll; +use std::task::Waker; + +use crate::internal::mutex::Mutex; +use crate::internal::waitlist::WaitList; +use crate::internal::waitlist::WaiterId; +use crate::internal::waitset::wake_all; + +/// A reusable event that releases all waiters when set and stays ready until reset. +/// +/// `set` and `reset` are idempotent. A wait that has returned `Pending` and is still registered +/// when a successful unset to set transition occurs is committed to completion by that transition, +/// even if `reset` happens before the future is polled again. +/// +/// Dropping a pending wait and a concurrent `set` linearize on the same internal lock, and +/// whichever acquires it first decides the outcome. If the drop wins, the waiter is already gone +/// and that `set` never commits it. If the `set` wins, the waiter is committed and the drop then +/// removes a node whose completion no one will observe; because `set` invokes wakers after +/// releasing the lock, that waker may still run after the drop has returned. +/// +/// Neither order withholds the signal from another waiter: a commitment is not a permit that a +/// cancelled wait could consume. +/// +/// Registration happens on the first poll, not when [`wait`](Self::wait) constructs the future, so +/// a wait first polled after a reset belongs to the new unset period even when it was constructed +/// before the preceding `set`. +/// +/// # Synchronization +/// +/// Memory operations sequenced before a [`set`](Self::set) call that performs an unset to set +/// transition have a happens-before relationship with code that runs after any wait completed by +/// that transition, including a wait first polled while the resulting set state is still current. A +/// producer may therefore publish shared state and then call `set`, and every waiter that call +/// releases observes it. +/// +/// A `set` call that finds the event already set performs no transition and does not establish the +/// guarantee above. +/// +/// The API makes no publication guarantee for state observed through [`is_set`](Self::is_set), so +/// that query can neither stand in for a wait nor support check-then-act: the state it reports may +/// change before the caller acts on it. +pub struct ManualResetEvent { + state: Mutex, +} + +impl ManualResetEvent { + /// Creates an unset event. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::new(); + /// assert!(!event.is_set()); + /// ``` + pub const fn new() -> Self { + Self::with_state(false) + } + + /// Creates an event with the specified initial state. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::ManualResetEvent; + /// + /// let ready = ManualResetEvent::with_state(true); + /// assert!(ready.is_set()); + /// ``` + pub const fn with_state(is_set: bool) -> Self { + Self { + state: Mutex::new(State { + is_set, + waiters: WaitList::new(), + }), + } + } + + /// Returns whether the event is currently set. + /// + /// This is a snapshot only; it does not reserve or consume the set state. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::new(); + /// assert!(!event.is_set()); + /// + /// event.set(); + /// assert!(event.is_set()); + /// + /// event.reset(); + /// assert!(!event.is_set()); + /// ``` + pub fn is_set(&self) -> bool { + self.state.lock().is_set + } + + /// Sets the event and releases every waiter registered during the current unset period. + /// + /// The event remains set until [`reset`](Self::reset) is called. Calling `set` while it is + /// already set has no effect. Wakers are invoked after the internal lock is released; if one + /// panics, the remaining waiters are still woken and the first panic reaches the caller. + /// + /// A waker may therefore re-enter the event. The whole registered cohort is detached before any + /// waker runs, so a wait registered from a wake callback belongs to the period current when it + /// registers rather than to this call. No ordering is guaranteed among the released waits. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::new(); + /// event.set(); + /// + /// // The event stays ready, so every later wait completes without blocking. + /// event.wait().await; + /// event.wait().await; + /// # } + /// ``` + pub fn set(&self) { + let wakers = { + let mut state = self.state.lock(); + if state.is_set { + return; + } + + state.is_set = true; + let mut wakers = Vec::new(); + while let Some((_id, waiter)) = state.waiters.unlink_first_waiter(|waiter| { + waiter.notified = true; + true + }) { + if let Some(waker) = waiter.waker.take() { + wakers.push(waker); + } + } + wakers + }; + + wake_all(wakers.into_iter()); + } + + /// Resets the event so subsequent waits block until another [`set`](Self::set). + /// + /// Waiters already committed by a preceding `set` remain ready. Calling `reset` while the + /// event is already unset has no effect. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::with_state(true); + /// event.wait().await; + /// + /// // Subsequent waits block again until the next `set`. + /// event.reset(); + /// assert!(!event.is_set()); + /// + /// event.set(); + /// event.wait().await; + /// # } + /// ``` + pub fn reset(&self) { + self.state.lock().is_set = false; + } + + /// Returns a future that waits until the event is set. + /// + /// A poll that observes the event set returns `Ready` immediately. A wait that has returned + /// `Pending` and is still registered is committed to completion by the next + /// [`set`](Self::set), even if a [`reset`](Self::reset) happens before it is polled again. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::new(); + /// let waiter = async { + /// event.wait().await; + /// "released" + /// }; + /// let setter = async { event.set() }; + /// + /// let (released, ()) = tokio::join!(waiter, setter); + /// assert_eq!(released, "released"); + /// # } + /// ``` + pub fn wait(&self) -> ManualResetEventWait<'_> { + ManualResetEventWait { + waiter: None, + event: self, + } + } + + /// Returns an owned future that waits until the event is set. + /// + /// The event must be held in an [`Arc`]. The returned future owns that `Arc` and therefore has + /// no borrowing lifetime, which makes it suitable for spawned tasks. + /// + /// # Examples + /// + /// ``` + /// # #[tokio::main] + /// # async fn main() { + /// use std::sync::Arc; + /// + /// use asyncband::event::ManualResetEvent; + /// + /// let event = Arc::new(ManualResetEvent::new()); + /// let waiter = tokio::spawn(event.clone().wait_owned()); + /// + /// event.set(); + /// waiter.await.unwrap(); + /// # } + /// ``` + pub fn wait_owned(self: Arc) -> OwnedManualResetEventWait { + OwnedManualResetEventWait { + waiter: None, + event: self, + } + } + + /// Polls a wait, registering `waiter_id` on the first poll that observes an unset event. + /// + /// `set` unlinks every queued waiter and marks it notified, and a wait that starts while the + /// 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, replaced_waker) = { + let mut state = self.state.lock(); + match *waiter_id { + Some(id) if state.waiters.waiter_mut(id).notified => { + let waiter = state.remove_waiter(id); + *waiter_id = None; + (Poll::Ready(()), waiter.waker) + } + Some(id) => { + debug_assert!( + !state.is_set, + "a linked waiter must belong to an unset event" + ); + let waiter = state.waiters.waiter_mut(id); + let replaced = waiter.update_waker(cx); + (Poll::Pending, replaced) + } + None if state.is_set => (Poll::Ready(()), None), + None => { + *waiter_id = Some(state.waiters.push_back(Waiter { + notified: false, + waker: Some(cx.waker().clone()), + })); + (Poll::Pending, None) + } + } + }; + + drop(replaced_waker); + poll + } + + fn unregister_waiter(&self, waiter_id: &mut Option) { + let Some(id) = waiter_id.take() else { + return; + }; + let waiter = { + let mut state = self.state.lock(); + state.remove_waiter(id) + }; + drop(waiter); + } +} + +impl Default for ManualResetEvent { + fn default() -> Self { + Self::new() + } +} + +impl fmt::Debug for ManualResetEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ManualResetEvent") + .field("is_set", &self.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, + waker: Option, +} + +impl Waiter { + fn update_waker(&mut self, cx: &mut Context<'_>) -> Option { + let current = self + .waker + .as_mut() + .expect("an unnotified waiter must retain its waker"); + if current.will_wake(cx.waker()) { + return None; + } + Some(mem::replace(current, cx.waker().clone())) + } +} + +/// A borrowed future returned by [`ManualResetEvent::wait`]. +/// +/// Dropping a pending wait unregisters only that waiter; it leaves the event state and every other +/// waiter untouched. +#[must_use = "futures do nothing unless you `.await` or poll them"] +pub struct ManualResetEventWait<'a> { + waiter: Option, + event: &'a ManualResetEvent, +} + +impl fmt::Debug for ManualResetEventWait<'_> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ManualResetEventWait") + .finish_non_exhaustive() + } +} + +impl Future for ManualResetEventWait<'_> { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let Self { waiter, event } = self.get_mut(); + event.poll_wait(waiter, cx) + } +} + +impl Drop for ManualResetEventWait<'_> { + fn drop(&mut self) { + self.event.unregister_waiter(&mut self.waiter); + } +} + +/// An owned future returned by [`ManualResetEvent::wait_owned`]. +/// +/// This behaves like [`ManualResetEventWait`] and keeps the event alive through its [`Arc`]. +#[must_use = "futures do nothing unless you `.await` or poll them"] +pub struct OwnedManualResetEventWait { + waiter: Option, + event: Arc, +} + +impl fmt::Debug for OwnedManualResetEventWait { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("OwnedManualResetEventWait") + .finish_non_exhaustive() + } +} + +impl Future for OwnedManualResetEventWait { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let Self { waiter, event } = self.get_mut(); + event.poll_wait(waiter, cx) + } +} + +impl Drop for OwnedManualResetEventWait { + fn drop(&mut self) { + self.event.unregister_waiter(&mut self.waiter); + } +} diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 9cabd55..706e520 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -21,6 +21,7 @@ pub(crate) mod atomic_waker; #[cfg(any( feature = "barrier", feature = "broadcast", + feature = "event", feature = "latch", feature = "mpsc", feature = "mutex", @@ -49,6 +50,7 @@ pub(crate) mod value_cell; #[cfg(any( feature = "barrier", feature = "broadcast", + feature = "event", feature = "latch", feature = "mpsc", feature = "mutex", @@ -72,22 +74,28 @@ pub(crate) mod mutex; pub(crate) mod semaphore; #[cfg(any( + feature = "event", feature = "mpsc", feature = "mutex", feature = "rwlock", feature = "semaphore", ))] +// The event and semaphore-backed primitives use different queue operations. A single-feature +// build therefore leaves part of this shared API unused, while the all-feature build uses it. +#[allow(dead_code)] pub(crate) mod waitlist; #[cfg(any( feature = "barrier", feature = "broadcast", + feature = "event", feature = "latch", feature = "once", feature = "waitgroup", feature = "watch", ))] // `barrier` constructs a wait set with `with_capacity`, while countdown-based primitives use -// `new`. One constructor is therefore unused in every single-primitive build. +// `new`. One constructor is therefore unused in every single-primitive build, and `event` uses +// only the free `wake_all` helper. #[allow(dead_code)] pub(crate) mod waitset; diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index a29f659..bef0aca 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -51,27 +51,28 @@ //! //! # API map //! -//! | Area | API | Feature | Use | -//! |-----------------------|-------------------------------------|----------------|--------------------------------------------------------------------------------------------------------| -//! | Shared state | [`Mutex`](mutex::Mutex) | `mutex` | Protect shared data with asynchronous mutual exclusion. | -//! | | [`RwLock`](rwlock::RwLock) | `rwlock` | Allow multiple readers or one writer. | -//! | | [`Condvar`](condvar::Condvar) | `condvar` | Wait for notifications while releasing a mutex. | -//! | Initialization | [`Once`](once::Once) | `once` | Run asynchronous initialization exactly once. | -//! | | [`OnceCell`](once::OnceCell) | `once-cell` | Initialize and store one asynchronous value. | -//! | | [`LazyCell`](once::LazyCell) | `lazy-cell` | Lazily initialize a value with a stored asynchronous function. | -//! | | [`OnceMap`](once::OnceMap) | `once-map` | Initialize and store one value per key. | -//! | Task coordination | [`Barrier`](barrier::Barrier) | `barrier` | Wait until all participants reach a synchronization point. | -//! | | [`Latch`](latch::Latch) | `latch` | Wait until a one-way countdown completes. | -//! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Wait for a dynamic group of tasks to finish. | -//! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Coordinate shutdown signals and completion. | -//! | Channels | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. | -//! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver through a bounded or unbounded queue. | -//! | | [`broadcast`] | `broadcast` | Broadcast values from one or more producers and retain them until every active receiver consumes them. | -//! | | [`watch`] | `watch` | Publish the latest state to independently tracked receivers and coalesce intermediate updates. | -//! | Resource reuse | [`pool`] | `pool` | Reuse objects through bounded or unbounded pool variants. | -//! | Workload coordination | [`Semaphore`](semaphore::Semaphore) | `semaphore` | Control concurrent access with permits. | -//! | | [`Group`](singleflight::Group) | `singleflight` | Coalesce concurrent calls for the same key. | -//! | Sync interop | [`FutureExt`](blocking::FutureExt) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | +//! | Area | API | Feature | Use | +//! |-----------------------|-----------------------------------------------|----------------|--------------------------------------------------------------------------------------------------------| +//! | Shared state | [`Mutex`](mutex::Mutex) | `mutex` | Protect shared data with asynchronous mutual exclusion. | +//! | | [`RwLock`](rwlock::RwLock) | `rwlock` | Allow multiple readers or one writer. | +//! | | [`Condvar`](condvar::Condvar) | `condvar` | Wait for notifications while releasing a mutex. | +//! | Initialization | [`Once`](once::Once) | `once` | Run asynchronous initialization exactly once. | +//! | | [`OnceCell`](once::OnceCell) | `once-cell` | Initialize and store one asynchronous value. | +//! | | [`LazyCell`](once::LazyCell) | `lazy-cell` | Lazily initialize a value with a stored asynchronous function. | +//! | | [`OnceMap`](once::OnceMap) | `once-map` | Initialize and store one value per key. | +//! | Task coordination | [`Barrier`](barrier::Barrier) | `barrier` | Wait until all participants reach a synchronization point. | +//! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Reuse a level-triggered signal that releases all current waiters. | +//! | | [`Latch`](latch::Latch) | `latch` | Wait until a one-way countdown completes. | +//! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Wait for a dynamic group of tasks to finish. | +//! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Coordinate shutdown signals and completion. | +//! | Channels | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. | +//! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver through a bounded or unbounded queue. | +//! | | [`broadcast`] | `broadcast` | Broadcast values from one or more producers and retain them until every active receiver consumes them. | +//! | | [`watch`] | `watch` | Publish the latest state to independently tracked receivers and coalesce intermediate updates. | +//! | Resource reuse | [`pool`] | `pool` | Reuse objects through bounded or unbounded pool variants. | +//! | Workload coordination | [`Semaphore`](semaphore::Semaphore) | `semaphore` | Control concurrent access with permits. | +//! | | [`Group`](singleflight::Group) | `singleflight` | Coalesce concurrent calls for the same key. | +//! | Sync interop | [`FutureExt`](blocking::FutureExt) | `blocking` | Drive one runtime-agnostic future from a blocking thread. | //! //! # Scope and runtime model //! @@ -122,6 +123,8 @@ pub mod blocking; pub mod broadcast; #[cfg(feature = "condvar")] pub mod condvar; +#[cfg(feature = "event")] +pub mod event; #[cfg(feature = "latch")] pub mod latch; #[cfg(feature = "mpsc")] diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 44accc9..b4ea55f 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -30,6 +30,7 @@ asyncband = { workspace = true, features = [ "blocking", "broadcast", "condvar", + "event", "latch", "mpsc", "mutex", diff --git a/benchmarks/asyncband/event/mod.rs b/benchmarks/asyncband/event/mod.rs new file mode 100644 index 0000000..3554069 --- /dev/null +++ b/benchmarks/asyncband/event/mod.rs @@ -0,0 +1,18 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod wait; diff --git a/benchmarks/asyncband/event/wait.rs b/benchmarks/asyncband/event/wait.rs new file mode 100644 index 0000000..966b6d1 --- /dev/null +++ b/benchmarks/asyncband/event/wait.rs @@ -0,0 +1,130 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::pin::Pin; +use std::pin::pin; + +use asyncband::event::ManualResetEvent; +use divan::Bencher; +use divan::black_box; +use divan::counter::ItemsCount; + +use crate::support::bench_context; +use crate::support::poll_pending; +use crate::support::poll_pinned_ready; + +const WAITER_COUNTS: &[usize] = &[1, 8, 32]; +const THREAD_COUNTS: &[usize] = &[1, 2, 8, 32]; +const CONTENDED_SAMPLE_SIZE: u32 = 256; + +#[divan::bench(threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] +fn wait_already_set(bencher: Bencher) { + let event = ManualResetEvent::with_state(true); + + bencher.bench(|| { + let mut context = bench_context(); + let mut wait = pin!(event.wait()); + poll_pinned_ready(wait.as_mut(), &mut context); + black_box(&event) + }); +} + +#[divan::bench(threads = THREAD_COUNTS, sample_size = CONTENDED_SAMPLE_SIZE)] +fn is_set_contended(bencher: Bencher) { + let event = ManualResetEvent::with_state(true); + + bencher.bench(|| black_box(event.is_set())); +} + +#[divan::bench] +fn set_reset_cycle(bencher: Bencher) { + let event = ManualResetEvent::new(); + + bencher.bench_local(|| { + event.set(); + event.reset(); + black_box(&event) + }); +} + +#[divan::bench] +fn cancel_pending(bencher: Bencher) { + let mut context = bench_context(); + + bencher.bench_local(|| { + let event = ManualResetEvent::new(); + { + let mut wait = pin!(event.wait()); + poll_pending(wait.as_mut(), &mut context); + } + black_box(event) + }); +} + +#[divan::bench] +fn wake_waiter(bencher: Bencher) { + let mut context = bench_context(); + + bencher.bench_local(|| { + let event = ManualResetEvent::new(); + { + let mut wait = pin!(event.wait()); + poll_pending(wait.as_mut(), &mut context); + + event.set(); + poll_pinned_ready(wait.as_mut(), &mut context); + } + black_box(event) + }); +} + +#[divan::bench] +fn wake_waiter_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); + + event.set(); + poll_pinned_ready(wait.as_mut(), &mut context); + event.reset(); + black_box(&event) + }); +} + +#[divan::bench(args = WAITER_COUNTS)] +fn waiter_fan_out(bencher: Bencher, waiter_count: usize) { + let mut context = bench_context(); + + bencher + .counter(ItemsCount::new(waiter_count)) + .bench_local(|| { + let event = ManualResetEvent::new(); + let mut waiters = (0..waiter_count).map(|_| event.wait()).collect::>(); + for waiter in &mut waiters { + poll_pending(Pin::new(waiter), &mut context); + } + + event.set(); + for mut waiter in waiters { + poll_pinned_ready(Pin::new(&mut waiter), &mut context); + } + black_box(event) + }); +} diff --git a/benchmarks/asyncband/main.rs b/benchmarks/asyncband/main.rs index 980c875..9eb693c 100644 --- a/benchmarks/asyncband/main.rs +++ b/benchmarks/asyncband/main.rs @@ -19,6 +19,7 @@ mod barrier; mod blocking; mod broadcast; mod condvar; +mod event; mod latch; mod mpsc; mod mutex; diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index a43060d..620c5e5 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -31,6 +31,7 @@ asyncband = { workspace = true, features = [ "blocking", "broadcast", "condvar", + "event", "latch", "lazy-cell", "mpsc", diff --git a/tests-integration/tests/event_test.rs b/tests-integration/tests/event_test.rs new file mode 100644 index 0000000..cd75f64 --- /dev/null +++ b/tests-integration/tests/event_test.rs @@ -0,0 +1,537 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::future::Future; +use std::panic; +use std::panic::AssertUnwindSafe; +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::sync::mpsc; +use std::task::Context; +use std::task::Wake; +use std::task::Waker; +use std::thread; +use std::time::Duration; + +use asyncband::event::ManualResetEvent; +use asyncband::event::OwnedManualResetEventWait; +use tests_integration::poll_once; + +struct TrackWake(AtomicUsize); + +impl Wake for TrackWake { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } +} + +#[test] +fn constructors_and_state_queries() { + let unset = ManualResetEvent::new(); + let set = ManualResetEvent::with_state(true); + + assert!(!unset.is_set()); + assert!(set.is_set()); + assert!(!ManualResetEvent::default().is_set()); +} + +#[test] +fn set_is_sticky_and_reset_blocks_new_waiters() { + let event = ManualResetEvent::new(); + event.set(); + + let mut ready = pin!(event.wait()); + assert!(poll_once(ready.as_mut()).is_ready()); + + event.reset(); + let mut pending = pin!(event.wait()); + assert!(poll_once(pending.as_mut()).is_pending()); +} + +// Registration happens on the first poll, not when `wait` builds the future. +#[test] +fn an_unpolled_wait_is_not_a_waiter_of_a_preceding_set() { + let event = ManualResetEvent::new(); + let mut unpolled = pin!(event.wait()); + + event.set(); + event.reset(); + + assert!(poll_once(unpolled.as_mut()).is_pending()); +} + +#[test] +fn polling_with_a_new_waker_replaces_the_registration() { + let event = ManualResetEvent::new(); + let first = Arc::new(TrackWake(AtomicUsize::new(0))); + let second = Arc::new(TrackWake(AtomicUsize::new(0))); + let first_waker = Waker::from(first.clone()); + let second_waker = Waker::from(second.clone()); + let baseline = Arc::strong_count(&first); + let mut wait = pin!(event.wait()); + + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&first_waker)) + .is_pending() + ); + assert_eq!(Arc::strong_count(&first), baseline + 1); + + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&second_waker)) + .is_pending() + ); + assert_eq!(Arc::strong_count(&first), baseline); + + event.set(); + assert_eq!(first.0.load(Ordering::Relaxed), 0); + assert_eq!(second.0.load(Ordering::Relaxed), 1); +} + +#[test] +fn cancelling_a_committed_waiter_leaves_the_others_committed() { + let event = ManualResetEvent::new(); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let mut context = Context::from_waker(&waker); + let mut cancelled = Box::pin(event.wait()); + let mut survivor = Box::pin(event.wait()); + + assert!(cancelled.as_mut().poll(&mut context).is_pending()); + assert!(survivor.as_mut().poll(&mut context).is_pending()); + + event.set(); + event.reset(); + assert_eq!(tracker.0.load(Ordering::Relaxed), 2); + + // A committed waiter holds no consumable permit, so dropping it hands nothing on. + drop(cancelled); + assert!(survivor.as_mut().poll(&mut context).is_ready()); + assert!(!event.is_set()); +} + +struct PanicOnWake; + +impl Wake for PanicOnWake { + fn wake(self: Arc) { + panic!("waker panicked"); + } +} + +// `set` releases every waiter registered for the current unset period, so one broken waker must +// not strand the waiters queued behind it. +#[test] +fn a_panicking_waker_still_releases_the_remaining_waiters() { + let event = ManualResetEvent::new(); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let tracking_waker = Waker::from(tracker.clone()); + let panicking_waker = Waker::from(Arc::new(PanicOnWake)); + + let mut first = Box::pin(event.wait()); + let mut exploding = Box::pin(event.wait()); + let mut last = Box::pin(event.wait()); + assert!( + first + .as_mut() + .poll(&mut Context::from_waker(&tracking_waker)) + .is_pending() + ); + assert!( + exploding + .as_mut() + .poll(&mut Context::from_waker(&panicking_waker)) + .is_pending() + ); + assert!( + last.as_mut() + .poll(&mut Context::from_waker(&tracking_waker)) + .is_pending() + ); + + let panicked = panic::catch_unwind(AssertUnwindSafe(|| event.set())); + + assert!(panicked.is_err(), "the waker panic must reach the caller"); + assert_eq!( + tracker.0.load(Ordering::Relaxed), + 2, + "the waiter queued behind the panicking waker was never woken" + ); + assert!(event.is_set()); + assert!( + last.as_mut() + .poll(&mut Context::from_waker(&tracking_waker)) + .is_ready() + ); +} + +/// A waker that re-enters the event it belongs to, both when woken and when dropped. +/// +/// The internal lock is a non-reentrant `std::sync::Mutex`, so waking or dropping this waker inside +/// the critical section blocks forever instead of returning. +struct ReentrantWaker(Arc); + +impl Wake for ReentrantWaker { + fn wake(self: Arc) { + self.0.reset(); + } +} + +impl Drop for ReentrantWaker { + fn drop(&mut self) { + self.0.is_set(); + } +} + +// Wakers must be invoked and dropped after the internal lock is released. +// +// The scenario runs on a worker thread so that a regression surfaces as a bounded failure here +// rather than hanging until the harness times out. +#[test] +fn wakers_are_woken_and_dropped_outside_the_internal_lock() { + let (done, finished) = mpsc::channel(); + + thread::spawn(move || { + let event = Arc::new(ManualResetEvent::new()); + + // `set` wakes the waiter, and the waker re-enters `reset`. + let mut woken = Box::pin(event.clone().wait_owned()); + { + let waker = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + woken + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + } + event.set(); + assert!(!event.is_set(), "the waker's reset did not land"); + assert!( + woken + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + .is_ready() + ); + + // Polling again with a different waker drops the one it replaces. The registration holds + // the last reference to the first waker, so its `Drop` runs during that poll. + let mut replaced = Box::pin(event.clone().wait_owned()); + { + let first = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + replaced + .as_mut() + .poll(&mut Context::from_waker(&first)) + .is_pending() + ); + } + { + let second = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + replaced + .as_mut() + .poll(&mut Context::from_waker(&second)) + .is_pending() + ); + } + + // Cancelling a pending wait drops the registered waker, which re-enters `is_set`. The + // registration holds the last reference, so the drop runs here. + drop(replaced); + + done.send(()).unwrap(); + }); + + finished + .recv_timeout(Duration::from_secs(10)) + .expect("a waker was invoked or dropped while the internal lock was held"); +} + +// A waker that resets the event and registers a fresh waiter from inside the wake callback. +struct ResetAndRegister { + event: Arc, + fresh: Mutex>, + fresh_was_pending: AtomicBool, +} + +impl Wake for ResetAndRegister { + fn wake(self: Arc) { + self.event.reset(); + let mut wait = self.event.clone().wait_owned(); + let pending = Pin::new(&mut wait) + .poll(&mut Context::from_waker(Waker::noop())) + .is_pending(); + self.fresh_was_pending.store(pending, Ordering::Relaxed); + *self.fresh.lock().unwrap() = Some(wait); + } +} + +// `set` detaches the whole registered cohort before any waker runs, so a waiter registered from a +// wake callback belongs to the period that callback's `reset` opened, not to the `set` in progress. +// This guards the cohort boundary across future drain or atomic slow-path refactors; a naive +// incremental wake-as-you-drain would lose it. +#[test] +fn a_wait_registered_from_a_wake_callback_belongs_to_the_next_period() { + let event = Arc::new(ManualResetEvent::new()); + let hook = Arc::new(ResetAndRegister { + event: event.clone(), + fresh: Mutex::new(None), + fresh_was_pending: AtomicBool::new(false), + }); + let waker = Waker::from(hook.clone()); + let mut released = Box::pin(event.clone().wait_owned()); + + assert!( + released + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + + event.set(); + + assert!(!event.is_set(), "the callback's reset did not land"); + assert!( + hook.fresh_was_pending.load(Ordering::Relaxed), + "a wait registered after the callback's reset was committed by the outer set" + ); + assert!( + released + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + .is_ready(), + "the cohort registered before the set stays committed" + ); + + let mut fresh = hook + .fresh + .lock() + .unwrap() + .take() + .expect("the callback registered a waiter"); + assert!( + Pin::new(&mut fresh) + .poll(&mut Context::from_waker(Waker::noop())) + .is_pending() + ); + + event.set(); + assert!( + Pin::new(&mut fresh) + .poll(&mut Context::from_waker(Waker::noop())) + .is_ready() + ); +} + +#[test] +fn set_then_reset_commits_registered_waiters() { + let event = ManualResetEvent::new(); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let mut context = Context::from_waker(&waker); + let mut wait = pin!(event.wait()); + + assert!(wait.as_mut().poll(&mut context).is_pending()); + event.set(); + event.reset(); + + assert_eq!(tracker.0.load(Ordering::Relaxed), 1); + assert!(wait.as_mut().poll(&mut context).is_ready()); + + let mut next_generation = pin!(event.wait()); + assert!(next_generation.as_mut().poll(&mut context).is_pending()); +} + +#[test] +fn repeated_set_is_coalesced() { + let event = ManualResetEvent::new(); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let mut context = Context::from_waker(&waker); + let mut wait = pin!(event.wait()); + + assert!(wait.as_mut().poll(&mut context).is_pending()); + event.set(); + event.set(); + + assert_eq!(tracker.0.load(Ordering::Relaxed), 1); + assert!(wait.as_mut().poll(&mut context).is_ready()); +} + +#[test] +fn cancelling_a_waiter_releases_its_waker() { + let event = ManualResetEvent::new(); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let baseline = Arc::strong_count(&tracker); + let mut context = Context::from_waker(&waker); + let mut wait = Box::pin(event.wait()); + + assert!(wait.as_mut().poll(&mut context).is_pending()); + assert_eq!(Arc::strong_count(&tracker), baseline + 1); + + drop(wait); + assert_eq!(Arc::strong_count(&tracker), baseline); + event.set(); + assert_eq!(tracker.0.load(Ordering::Relaxed), 0); +} + +#[test] +fn cancelling_an_owned_waiter_releases_its_waker_and_event_handle() { + let event = Arc::new(ManualResetEvent::new()); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let baseline = Arc::strong_count(&tracker); + let mut context = Context::from_waker(&waker); + let mut wait = Box::pin(event.clone().wait_owned()); + + assert!(wait.as_mut().poll(&mut context).is_pending()); + assert_eq!(Arc::strong_count(&tracker), baseline + 1); + + drop(wait); + assert_eq!(Arc::strong_count(&tracker), baseline); + assert_eq!(Arc::strong_count(&event), 1); + + event.set(); + assert_eq!(tracker.0.load(Ordering::Relaxed), 0); +} + +// Registrations must not be lost when they race a concurrent `set`. +// +// Each round only resets after every waiter has completed, so a waiter that registers late +// observes the set state instead of blocking. A lost wake-up therefore hangs the join below. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn set_racing_registration_wakes_every_waiter() { + const ROUNDS: usize = 128; + const WAITERS: usize = 8; + + let event = Arc::new(ManualResetEvent::new()); + for _ in 0..ROUNDS { + let waiters = (0..WAITERS) + .map(|_| tokio::spawn(event.clone().wait_owned())) + .collect::>(); + + event.set(); + for waiter in waiters { + waiter.await.unwrap(); + } + + event.reset(); + } + + assert!(!event.is_set()); + assert!(poll_once(pin!(event.wait())).is_pending()); +} + +// A `set` immediately followed by a `reset` on another thread still commits every waiter that +// registered before the transition. +#[test] +fn cross_thread_set_then_reset_commits_registered_waiters() { + const ROUNDS: usize = 128; + const WAITERS: usize = 8; + + for _ in 0..ROUNDS { + let event = Arc::new(ManualResetEvent::new()); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let mut context = Context::from_waker(&waker); + let mut waits = (0..WAITERS) + .map(|_| Box::pin(event.clone().wait_owned())) + .collect::>(); + for wait in &mut waits { + assert!(wait.as_mut().poll(&mut context).is_pending()); + } + + let signaller = thread::spawn({ + let event = event.clone(); + move || { + event.set(); + event.reset(); + } + }); + signaller.join().unwrap(); + + assert!(!event.is_set()); + assert_eq!(tracker.0.load(Ordering::Relaxed), WAITERS); + for wait in &mut waits { + assert!(wait.as_mut().poll(&mut context).is_ready()); + } + } +} + +// Cancelling one waiter while another thread sets the event must reclaim that waiter's waker +// under either outcome of the race, and must not disturb the remaining waiter. +#[test] +fn cancellation_racing_set_reclaims_the_waker() { + const ROUNDS: usize = 256; + + for _ in 0..ROUNDS { + let event = Arc::new(ManualResetEvent::new()); + let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); + let waker = Waker::from(tracker.clone()); + let baseline = Arc::strong_count(&tracker); + let mut context = Context::from_waker(&waker); + let mut cancelled = Box::pin(event.clone().wait_owned()); + let mut survivor = Box::pin(event.clone().wait_owned()); + + assert!(cancelled.as_mut().poll(&mut context).is_pending()); + assert!(survivor.as_mut().poll(&mut context).is_pending()); + + let start = Arc::new(Barrier::new(2)); + let signaller = thread::spawn({ + let event = event.clone(); + let start = start.clone(); + move || { + start.wait(); + event.set(); + } + }); + + start.wait(); + drop(cancelled); + signaller.join().unwrap(); + + assert!(survivor.as_mut().poll(&mut context).is_ready()); + drop(survivor); + assert_eq!(Arc::strong_count(&tracker), baseline); + assert_eq!(Arc::strong_count(&event), 1); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn set_releases_all_current_and_future_waiters() { + let event = Arc::new(ManualResetEvent::new()); + let mut tasks = Vec::new(); + for _ in 0..16 { + tasks.push(tokio::spawn(event.clone().wait_owned())); + } + + tokio::task::yield_now().await; + event.set(); + + for task in tasks { + task.await.unwrap(); + } + event.clone().wait_owned().await; +} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 2a62838..203e9a3 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -20,6 +20,9 @@ use std::cell::Cell; use asyncband::barrier::Barrier; use asyncband::broadcast; use asyncband::condvar::Condvar; +use asyncband::event::ManualResetEvent; +use asyncband::event::ManualResetEventWait; +use asyncband::event::OwnedManualResetEventWait; use asyncband::latch::Latch; use asyncband::mpsc; use asyncband::mutex::Mutex; @@ -70,6 +73,9 @@ fn public_types_are_send_and_sync() { assert_send_and_sync::(); assert_send_and_sync::(); + assert_send_and_sync::(); + assert_send_and_sync::>(); + assert_send_and_sync::(); assert_send_and_sync::>>(); assert_send_and_sync::(); assert_send_and_sync::>(); @@ -129,6 +135,9 @@ fn public_types_are_unpin() { assert_unpin::(); assert_unpin::(); + assert_unpin::(); + assert_unpin::>(); + assert_unpin::(); assert_unpin::(); assert_unpin::>>(); assert_unpin::(); From d3bfede726a9d5fdcfb2970fdd389e61eb8636c0 Mon Sep 17 00:00:00 2001 From: tison Date: Sun, 30 Aug 2026 23:49:37 +0800 Subject: [PATCH 2/5] fix(event): clone wakers outside state lock --- asyncband/src/event/mod.rs | 90 ++++++++++++------- .../tests/waitset_reentrancy_test.rs | 30 +++++++ 2 files changed, 87 insertions(+), 33 deletions(-) diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index 647e0d3..8789478 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -298,36 +298,56 @@ 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, replaced_waker) = { - let mut state = self.state.lock(); - match *waiter_id { - Some(id) if state.waiters.waiter_mut(id).notified => { - let waiter = state.remove_waiter(id); - *waiter_id = None; - (Poll::Ready(()), waiter.waker) - } - Some(id) => { - debug_assert!( - !state.is_set, - "a linked waiter must belong to an unset event" - ); - let waiter = state.waiters.waiter_mut(id); - let replaced = waiter.update_waker(cx); - (Poll::Pending, replaced) - } - None if state.is_set => (Poll::Ready(()), None), - None => { - *waiter_id = Some(state.waiters.push_back(Waiter { - notified: false, - waker: Some(cx.waker().clone()), - })); - (Poll::Pending, None) + // Ready waits require no waker, and a pending wait normally keeps the same one. Inspect the + // state before cloning to preserve those fast paths. If registration needs a new waker, + // release the lock, clone, and repeat the full state check because the clone callback may + // re-enter and set or reset the event. + let mut prepared_waker = None; + loop { + let (poll, retired_waker) = { + let mut state = self.state.lock(); + match *waiter_id { + Some(id) if state.waiters.waiter_mut(id).notified => { + let waiter = state.remove_waiter(id); + *waiter_id = None; + (Poll::Ready(()), waiter.waker) + } + Some(id) => { + debug_assert!( + !state.is_set, + "a linked waiter must belong to an unset event" + ); + let waiter = state.waiters.waiter_mut(id); + if prepared_waker.is_none() && waiter.will_wake(cx.waker()) { + return Poll::Pending; + } + let Some(waker) = prepared_waker.take() else { + drop(state); + prepared_waker = Some(cx.waker().clone()); + continue; + }; + (Poll::Pending, Some(waiter.replace_waker(waker))) + } + None if state.is_set => (Poll::Ready(()), None), + None => { + let Some(waker) = prepared_waker.take() else { + drop(state); + prepared_waker = Some(cx.waker().clone()); + continue; + }; + *waiter_id = Some(state.waiters.push_back(Waiter { + notified: false, + waker: Some(waker), + })); + (Poll::Pending, None) + } } - } - }; + }; - drop(replaced_waker); - poll + drop(retired_waker); + drop(prepared_waker); + return poll; + } } fn unregister_waiter(&self, waiter_id: &mut Option) { @@ -379,15 +399,19 @@ struct Waiter { } impl Waiter { - fn update_waker(&mut self, cx: &mut Context<'_>) -> Option { + fn will_wake(&self, waker: &Waker) -> bool { + self.waker + .as_ref() + .expect("an unnotified waiter must retain its waker") + .will_wake(waker) + } + + fn replace_waker(&mut self, waker: Waker) -> Waker { let current = self .waker .as_mut() .expect("an unnotified waiter must retain its waker"); - if current.will_wake(cx.waker()) { - return None; - } - Some(mem::replace(current, cx.waker().clone())) + mem::replace(current, waker) } } diff --git a/tests-integration/tests/waitset_reentrancy_test.rs b/tests-integration/tests/waitset_reentrancy_test.rs index 2df903a..76b2062 100644 --- a/tests-integration/tests/waitset_reentrancy_test.rs +++ b/tests-integration/tests/waitset_reentrancy_test.rs @@ -31,6 +31,7 @@ use std::time::Duration; use asyncband::barrier::Barrier; use asyncband::broadcast::mpmc; use asyncband::completion; +use asyncband::event::ManualResetEvent; use asyncband::latch::Latch; use asyncband::watch; @@ -112,6 +113,35 @@ fn completion_clones_wakers_outside_its_state_lock() { ); } +#[test] +fn event_clones_wakers_outside_its_state_lock() { + assert_completes_without_deadlock( + "waker clone callback deadlocked against the event lock", + || { + let event = Arc::new(ManualResetEvent::new()); + let callback_event = event.clone(); + let waker = waker_with_clone_callback(move || callback_event.set()); + let mut first_wait = Box::pin(event.wait()); + + assert_eq!(poll_with(first_wait.as_mut(), &waker), Poll::Ready(())); + + event.reset(); + let mut repolled_wait = Box::pin(event.wait()); + assert_eq!( + poll_with(repolled_wait.as_mut(), Waker::noop()), + Poll::Pending + ); + + let callback_event = event.clone(); + let replacement = waker_with_clone_callback(move || callback_event.set()); + assert_eq!( + poll_with(repolled_wait.as_mut(), &replacement), + Poll::Ready(()) + ); + }, + ); +} + #[test] fn latch_clones_wakers_outside_its_waiter_lock() { assert_completes_without_deadlock( From a198e9604358cc804073cb83859455f5379197b3 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 31 Aug 2026 01:10:51 +0800 Subject: [PATCH 3/5] refactor(event): hide wait future types --- asyncband/src/event/mod.rs | 8 ++++---- tests-integration/tests/event_test.rs | 14 ++++++++------ tests-integration/tests/traits_test.rs | 15 +++++++++------ 3 files changed, 21 insertions(+), 16 deletions(-) diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index 8789478..e9f0784 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -257,7 +257,7 @@ impl ManualResetEvent { /// assert_eq!(released, "released"); /// # } /// ``` - pub fn wait(&self) -> ManualResetEventWait<'_> { + pub fn wait(&self) -> impl Future + Send + Sync + Unpin + '_ { ManualResetEventWait { waiter: None, event: self, @@ -285,7 +285,7 @@ impl ManualResetEvent { /// waiter.await.unwrap(); /// # } /// ``` - pub fn wait_owned(self: Arc) -> OwnedManualResetEventWait { + pub fn wait_owned(self: Arc) -> impl Future + Send + Sync + Unpin + 'static { OwnedManualResetEventWait { waiter: None, event: self, @@ -420,7 +420,7 @@ impl Waiter { /// Dropping a pending wait unregisters only that waiter; it leaves the event state and every other /// waiter untouched. #[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct ManualResetEventWait<'a> { +struct ManualResetEventWait<'a> { waiter: Option, event: &'a ManualResetEvent, } @@ -451,7 +451,7 @@ impl Drop for ManualResetEventWait<'_> { /// /// This behaves like [`ManualResetEventWait`] and keeps the event alive through its [`Arc`]. #[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct OwnedManualResetEventWait { +struct OwnedManualResetEventWait { waiter: Option, event: Arc, } diff --git a/tests-integration/tests/event_test.rs b/tests-integration/tests/event_test.rs index cd75f64..0bb2b18 100644 --- a/tests-integration/tests/event_test.rs +++ b/tests-integration/tests/event_test.rs @@ -34,7 +34,6 @@ use std::thread; use std::time::Duration; use asyncband::event::ManualResetEvent; -use asyncband::event::OwnedManualResetEventWait; use tests_integration::poll_once; struct TrackWake(AtomicUsize); @@ -271,15 +270,16 @@ fn wakers_are_woken_and_dropped_outside_the_internal_lock() { // A waker that resets the event and registers a fresh waiter from inside the wake callback. struct ResetAndRegister { event: Arc, - fresh: Mutex>, + fresh: Mutex + Send>>>>, fresh_was_pending: AtomicBool, } impl Wake for ResetAndRegister { fn wake(self: Arc) { self.event.reset(); - let mut wait = self.event.clone().wait_owned(); - let pending = Pin::new(&mut wait) + let mut wait = Box::pin(self.event.clone().wait_owned()); + let pending = wait + .as_mut() .poll(&mut Context::from_waker(Waker::noop())) .is_pending(); self.fresh_was_pending.store(pending, Ordering::Relaxed); @@ -331,14 +331,16 @@ fn a_wait_registered_from_a_wake_callback_belongs_to_the_next_period() { .take() .expect("the callback registered a waiter"); assert!( - Pin::new(&mut fresh) + fresh + .as_mut() .poll(&mut Context::from_waker(Waker::noop())) .is_pending() ); event.set(); assert!( - Pin::new(&mut fresh) + fresh + .as_mut() .poll(&mut Context::from_waker(Waker::noop())) .is_ready() ); diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index dca8bbf..39be30e 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -16,14 +16,13 @@ // under the License. use std::cell::Cell; +use std::sync::Arc; use asyncband::barrier::Barrier; use asyncband::broadcast; use asyncband::completion; use asyncband::condvar::Condvar; use asyncband::event::ManualResetEvent; -use asyncband::event::ManualResetEventWait; -use asyncband::event::OwnedManualResetEventWait; use asyncband::latch::Latch; use asyncband::mpsc; use asyncband::mutex::Mutex; @@ -71,12 +70,14 @@ impl ManageObject for PoolManager { #[test] fn public_types_are_send_and_sync() { fn assert_send_and_sync() {} + fn assert_send_and_sync_value(_: T) {} assert_send_and_sync::(); assert_send_and_sync::(); assert_send_and_sync::(); - assert_send_and_sync::>(); - assert_send_and_sync::(); + let event = ManualResetEvent::new(); + assert_send_and_sync_value(event.wait()); + assert_send_and_sync_value(Arc::new(ManualResetEvent::new()).wait_owned()); assert_send_and_sync::>>(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -140,12 +141,14 @@ fn movable_public_types_are_send() { #[test] fn public_types_are_unpin() { fn assert_unpin() {} + fn assert_unpin_value(_: T) {} assert_unpin::(); assert_unpin::(); assert_unpin::(); - assert_unpin::>(); - assert_unpin::(); + let event = ManualResetEvent::new(); + assert_unpin_value(event.wait()); + assert_unpin_value(Arc::new(ManualResetEvent::new()).wait_owned()); assert_unpin::>(); assert_unpin::>(); assert_unpin::(); From 05f8c0526d2acc8f1c065c78dfb6445b82269401 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 31 Aug 2026 01:25:15 +0800 Subject: [PATCH 4/5] refactor(event): use async wait methods --- asyncband/src/event/mod.rs | 14 ++++++++------ benchmarks/asyncband/event/wait.rs | 9 +++++---- tests-integration/tests/traits_test.rs | 4 ---- 3 files changed, 13 insertions(+), 14 deletions(-) diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index e9f0784..27ee5e8 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -257,11 +257,12 @@ impl ManualResetEvent { /// assert_eq!(released, "released"); /// # } /// ``` - pub fn wait(&self) -> impl Future + Send + Sync + Unpin + '_ { - ManualResetEventWait { + pub async fn wait(&self) { + let fut = ManualResetEventWait { waiter: None, event: self, - } + }; + fut.await } /// Returns an owned future that waits until the event is set. @@ -285,11 +286,12 @@ impl ManualResetEvent { /// waiter.await.unwrap(); /// # } /// ``` - pub fn wait_owned(self: Arc) -> impl Future + Send + Sync + Unpin + 'static { - OwnedManualResetEventWait { + pub async fn wait_owned(self: Arc) { + let fut = OwnedManualResetEventWait { waiter: None, event: self, - } + }; + fut.await } /// Polls a wait, registering `waiter_id` on the first poll that observes an unset event. diff --git a/benchmarks/asyncband/event/wait.rs b/benchmarks/asyncband/event/wait.rs index 966b6d1..18ada48 100644 --- a/benchmarks/asyncband/event/wait.rs +++ b/benchmarks/asyncband/event/wait.rs @@ -15,7 +15,6 @@ // specific language governing permissions and limitations // under the License. -use std::pin::Pin; use std::pin::pin; use asyncband::event::ManualResetEvent; @@ -116,14 +115,16 @@ fn waiter_fan_out(bencher: Bencher, waiter_count: usize) { .counter(ItemsCount::new(waiter_count)) .bench_local(|| { let event = ManualResetEvent::new(); - let mut waiters = (0..waiter_count).map(|_| event.wait()).collect::>(); + let mut waiters = (0..waiter_count) + .map(|_| Box::pin(event.wait())) + .collect::>(); for waiter in &mut waiters { - poll_pending(Pin::new(waiter), &mut context); + poll_pending(waiter.as_mut(), &mut context); } event.set(); for mut waiter in waiters { - poll_pinned_ready(Pin::new(&mut waiter), &mut context); + poll_pinned_ready(waiter.as_mut(), &mut context); } black_box(event) }); diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 39be30e..4cdd7a4 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -141,14 +141,10 @@ fn movable_public_types_are_send() { #[test] fn public_types_are_unpin() { fn assert_unpin() {} - fn assert_unpin_value(_: T) {} assert_unpin::(); assert_unpin::(); assert_unpin::(); - let event = ManualResetEvent::new(); - assert_unpin_value(event.wait()); - assert_unpin_value(Arc::new(ManualResetEvent::new()).wait_owned()); assert_unpin::>(); assert_unpin::>(); assert_unpin::(); From 9bf14dda4cc52b36b8112463f8d2d27797ce7cf5 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 31 Aug 2026 01:33:24 +0800 Subject: [PATCH 5/5] docs(event): tighten contracts and tests --- CHANGELOG.md | 2 +- README.md | 2 +- asyncband/src/event/mod.rs | 113 ++++++++------------ asyncband/src/lib.rs | 2 +- tests-integration/tests/event_test.rs | 136 +------------------------ tests-integration/tests/traits_test.rs | 5 - 6 files changed, 45 insertions(+), 215 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33e17ee..2949716 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,7 +24,7 @@ All notable changes to this project will be documented in this file. * Implement `broadcast::mpmc::unbounded`, an unbounded broadcast channel that retains messages until all active receivers consume them or are dropped. * Add an opt-in latest-state channel under `asyncband::watch`. -* Add opt-in `asyncband::event::ManualResetEvent`, a reusable level-triggered signal that releases every waiter registered before a set, stays ready until an explicit reset, and commits released waiters even when a reset races their next poll. +* Add opt-in `asyncband::event::ManualResetEvent`, a reusable level-triggered signal that releases registered waits and remains ready for future waits until explicitly reset. * Add an opt-in shared one-shot completion primitive under `asyncband::completion` with a single-use completer, cloneable observers, a retained borrowed result, and observable abandonment. * Add opt-in `asyncband::once::LazyCell` for values that own one asynchronous initializer and preserve its in-flight future across caller cancellation. * Add opt-in bounded and unbounded runtime-agnostic object pools under `asyncband::pool`. diff --git a/README.md b/README.md index 4621f55..34e1e14 100644 --- a/README.md +++ b/README.md @@ -73,7 +73,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon | | [`OnceMap`](https://docs.rs/asyncband/*/asyncband/once/struct.OnceMap.html) | `once-map` | Initialize and store one value per key. | | Task coordination | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Wait until all participants reach a synchronization point. | | | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html) | `completion` | Publish one shared result to any number of current and future observers. | -| | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Reuse a level-triggered signal that releases all current waiters. | +| | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Signal current and future waits until explicitly reset. | | | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a one-way countdown completes. | | | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Wait for a dynamic group of tasks to finish. | | | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Coordinate shutdown signals and completion. | diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index 27ee5e8..96b3ebf 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -17,18 +17,17 @@ //! A reusable, level-triggered signal for coordinating tasks. //! -//! A [`ManualResetEvent`] is either set or unset. Calling [`set`](ManualResetEvent::set) makes the -//! event ready, releases every waiter registered during the current unset period, and keeps later -//! waits ready. Calling [`reset`](ManualResetEvent::reset) makes subsequent waits block again. +//! A [`ManualResetEvent`] is either set or unset. Calling [`set`](ManualResetEvent::set) releases +//! every registered wait and makes future waits ready. The signal remains set until +//! [`reset`](ManualResetEvent::reset) makes new waits block again. //! -//! A waiter registered before `set` is committed to completion even if another task calls `reset` -//! before the waiter is polled again. This differs from a condition variable: the event retains its -//! set state and does not require an external predicate or mutex. +//! The retained set state distinguishes this primitive from a condition variable, whose +//! notifications are not buffered. Unlike a latch, a manual-reset event can be reset and reused. //! -//! Registration happens on a wait's first poll, so `set` followed immediately by `reset` is not a -//! reliable way to release everyone waiting at that moment: a future constructed before the `set` -//! but not yet polled was never a waiter of it. Keep the event set for as long as the condition it -//! reports holds. +//! A wait registered before `set` is committed to completion even if another task calls `reset` +//! before that wait is polled again. Registration happens on the first poll, not when the future is +//! constructed, so `set` followed immediately by `reset` is not a pulse for unpolled futures. Keep +//! the event set for as long as the condition it represents holds. //! //! # Examples //! @@ -65,39 +64,18 @@ use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; use crate::internal::waitset::wake_all; -/// A reusable event that releases all waiters when set and stays ready until reset. +/// A reusable event that remains set until explicitly reset. /// -/// `set` and `reset` are idempotent. A wait that has returned `Pending` and is still registered -/// when a successful unset to set transition occurs is committed to completion by that transition, -/// even if `reset` happens before the future is polled again. -/// -/// Dropping a pending wait and a concurrent `set` linearize on the same internal lock, and -/// whichever acquires it first decides the outcome. If the drop wins, the waiter is already gone -/// and that `set` never commits it. If the `set` wins, the waiter is committed and the drop then -/// removes a node whose completion no one will observe; because `set` invokes wakers after -/// releasing the lock, that waker may still run after the drop has returned. -/// -/// Neither order withholds the signal from another waiter: a commitment is not a permit that a -/// cancelled wait could consume. -/// -/// Registration happens on the first poll, not when [`wait`](Self::wait) constructs the future, so -/// a wait first polled after a reset belongs to the new unset period even when it was constructed -/// before the preceding `set`. +/// See the [module-level documentation](self) for its waiting semantics. /// /// # Synchronization /// -/// Memory operations sequenced before a [`set`](Self::set) call that performs an unset to set -/// transition have a happens-before relationship with code that runs after any wait completed by -/// that transition, including a wait first polled while the resulting set state is still current. A -/// producer may therefore publish shared state and then call `set`, and every waiter that call -/// releases observes it. +/// An unset-to-set transition synchronizes with the waits it releases and with waits first polled +/// while the event remains set. Memory operations sequenced before [`set`](Self::set) are therefore +/// visible after those waits complete. /// -/// A `set` call that finds the event already set performs no transition and does not establish the -/// guarantee above. -/// -/// The API makes no publication guarantee for state observed through [`is_set`](Self::is_set), so -/// that query can neither stand in for a wait nor support check-then-act: the state it reports may -/// change before the caller acts on it. +/// A `set` call that finds the event already set does not establish this guarantee. +/// [`is_set`](Self::is_set) is only a snapshot and cannot replace a wait or support check-then-act. pub struct ManualResetEvent { state: Mutex, } @@ -158,15 +136,15 @@ impl ManualResetEvent { self.state.lock().is_set } - /// Sets the event and releases every waiter registered during the current unset period. + /// Sets the event and releases every currently registered wait. /// /// The event remains set until [`reset`](Self::reset) is called. Calling `set` while it is - /// already set has no effect. Wakers are invoked after the internal lock is released; if one - /// panics, the remaining waiters are still woken and the first panic reaches the caller. + /// already set has no effect. No ordering is guaranteed among the released waits. + /// + /// # Panics /// - /// A waker may therefore re-enter the event. The whole registered cohort is detached before any - /// waker runs, so a wait registered from a wake callback belongs to the period current when it - /// registers rather than to this call. No ordering is guaranteed among the released waits. + /// Panics if a registered waker panics. The state transition remains committed, and every + /// remaining registered waker is still notified before the first panic resumes. /// /// # Examples /// @@ -191,6 +169,8 @@ impl ManualResetEvent { } state.is_set = true; + // 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. let mut wakers = Vec::new(); while let Some((_id, waiter)) = state.waiters.unlink_first_waiter(|waiter| { waiter.notified = true; @@ -206,7 +186,7 @@ impl ManualResetEvent { wake_all(wakers.into_iter()); } - /// Resets the event so subsequent waits block until another [`set`](Self::set). + /// Resets the event so new waits block until another [`set`](Self::set). /// /// Waiters already committed by a preceding `set` remain ready. Calling `reset` while the /// event is already unset has no effect. @@ -233,11 +213,18 @@ impl ManualResetEvent { self.state.lock().is_set = false; } - /// Returns a future that waits until the event is set. + /// Waits until the event is set. /// - /// A poll that observes the event set returns `Ready` immediately. A wait that has returned - /// `Pending` and is still registered is committed to completion by the next - /// [`set`](Self::set), even if a [`reset`](Self::reset) happens before it is polled again. + /// If the event is already set, the wait completes immediately. Once a [`set`](Self::set) + /// commits a registered wait, a later [`reset`](Self::reset) cannot make that wait pending + /// again. + /// + /// # Cancel safety + /// + /// Dropping a pending wait unregisters only that call; it does not change the event or affect + /// other waiters. If cancellation races with `set`, either cancellation unregisters first or + /// `set` commits the wait first. A waker already detached by `set` may still run after the wait + /// is dropped. /// /// # Examples /// @@ -265,10 +252,11 @@ impl ManualResetEvent { fut.await } - /// Returns an owned future that waits until the event is set. + /// Waits until the event is set without borrowing it. /// - /// The event must be held in an [`Arc`]. The returned future owns that `Arc` and therefore has - /// no borrowing lifetime, which makes it suitable for spawned tasks. + /// The event must be held in an [`Arc`]. The returned future owns that `Arc`, which makes it + /// suitable for spawned tasks. Its waiting and cancellation semantics match + /// [`wait`](Self::wait). /// /// # Examples /// @@ -417,23 +405,12 @@ impl Waiter { } } -/// A borrowed future returned by [`ManualResetEvent::wait`]. -/// -/// Dropping a pending wait unregisters only that waiter; it leaves the event state and every other -/// waiter untouched. #[must_use = "futures do nothing unless you `.await` or poll them"] struct ManualResetEventWait<'a> { waiter: Option, event: &'a ManualResetEvent, } -impl fmt::Debug for ManualResetEventWait<'_> { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("ManualResetEventWait") - .finish_non_exhaustive() - } -} - impl Future for ManualResetEventWait<'_> { type Output = (); @@ -449,22 +426,12 @@ impl Drop for ManualResetEventWait<'_> { } } -/// An owned future returned by [`ManualResetEvent::wait_owned`]. -/// -/// This behaves like [`ManualResetEventWait`] and keeps the event alive through its [`Arc`]. #[must_use = "futures do nothing unless you `.await` or poll them"] struct OwnedManualResetEventWait { waiter: Option, event: Arc, } -impl fmt::Debug for OwnedManualResetEventWait { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("OwnedManualResetEventWait") - .finish_non_exhaustive() - } -} - impl Future for OwnedManualResetEventWait { type Output = (); diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 4a71a25..31039bc 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -65,7 +65,7 @@ //! | | [`OnceMap`](once::OnceMap) | `once-map` | Initialize and store one value per key. | //! | Task coordination | [`Barrier`](barrier::Barrier) | `barrier` | Wait until all participants reach a synchronization point. | //! | | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. | -//! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Reuse a level-triggered signal that releases all current waiters. | +//! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Signal current and future waits until explicitly reset. | //! | | [`Latch`](latch::Latch) | `latch` | Wait until a one-way countdown completes. | //! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Wait for a dynamic group of tasks to finish. | //! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Coordinate shutdown signals and completion. | diff --git a/tests-integration/tests/event_test.rs b/tests-integration/tests/event_test.rs index 0bb2b18..87ed2bd 100644 --- a/tests-integration/tests/event_test.rs +++ b/tests-integration/tests/event_test.rs @@ -21,7 +21,6 @@ use std::panic::AssertUnwindSafe; 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; @@ -44,16 +43,6 @@ impl Wake for TrackWake { } } -#[test] -fn constructors_and_state_queries() { - let unset = ManualResetEvent::new(); - let set = ManualResetEvent::with_state(true); - - assert!(!unset.is_set()); - assert!(set.is_set()); - assert!(!ManualResetEvent::default().is_set()); -} - #[test] fn set_is_sticky_and_reset_blocks_new_waiters() { let event = ManualResetEvent::new(); @@ -184,10 +173,8 @@ fn a_panicking_waker_still_releases_the_remaining_waiters() { ); } -/// A waker that re-enters the event it belongs to, both when woken and when dropped. -/// -/// The internal lock is a non-reentrant `std::sync::Mutex`, so waking or dropping this waker inside -/// the critical section blocks forever instead of returning. +// A waker that re-enters the event it belongs to, both when woken and when dropped. The internal +// lock is non-reentrant, so waking or dropping this waker inside the critical section deadlocks. struct ReentrantWaker(Arc); impl Wake for ReentrantWaker { @@ -418,122 +405,3 @@ fn cancelling_an_owned_waiter_releases_its_waker_and_event_handle() { event.set(); assert_eq!(tracker.0.load(Ordering::Relaxed), 0); } - -// Registrations must not be lost when they race a concurrent `set`. -// -// Each round only resets after every waiter has completed, so a waiter that registers late -// observes the set state instead of blocking. A lost wake-up therefore hangs the join below. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn set_racing_registration_wakes_every_waiter() { - const ROUNDS: usize = 128; - const WAITERS: usize = 8; - - let event = Arc::new(ManualResetEvent::new()); - for _ in 0..ROUNDS { - let waiters = (0..WAITERS) - .map(|_| tokio::spawn(event.clone().wait_owned())) - .collect::>(); - - event.set(); - for waiter in waiters { - waiter.await.unwrap(); - } - - event.reset(); - } - - assert!(!event.is_set()); - assert!(poll_once(pin!(event.wait())).is_pending()); -} - -// A `set` immediately followed by a `reset` on another thread still commits every waiter that -// registered before the transition. -#[test] -fn cross_thread_set_then_reset_commits_registered_waiters() { - const ROUNDS: usize = 128; - const WAITERS: usize = 8; - - for _ in 0..ROUNDS { - let event = Arc::new(ManualResetEvent::new()); - let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); - let waker = Waker::from(tracker.clone()); - let mut context = Context::from_waker(&waker); - let mut waits = (0..WAITERS) - .map(|_| Box::pin(event.clone().wait_owned())) - .collect::>(); - for wait in &mut waits { - assert!(wait.as_mut().poll(&mut context).is_pending()); - } - - let signaller = thread::spawn({ - let event = event.clone(); - move || { - event.set(); - event.reset(); - } - }); - signaller.join().unwrap(); - - assert!(!event.is_set()); - assert_eq!(tracker.0.load(Ordering::Relaxed), WAITERS); - for wait in &mut waits { - assert!(wait.as_mut().poll(&mut context).is_ready()); - } - } -} - -// Cancelling one waiter while another thread sets the event must reclaim that waiter's waker -// under either outcome of the race, and must not disturb the remaining waiter. -#[test] -fn cancellation_racing_set_reclaims_the_waker() { - const ROUNDS: usize = 256; - - for _ in 0..ROUNDS { - let event = Arc::new(ManualResetEvent::new()); - let tracker = Arc::new(TrackWake(AtomicUsize::new(0))); - let waker = Waker::from(tracker.clone()); - let baseline = Arc::strong_count(&tracker); - let mut context = Context::from_waker(&waker); - let mut cancelled = Box::pin(event.clone().wait_owned()); - let mut survivor = Box::pin(event.clone().wait_owned()); - - assert!(cancelled.as_mut().poll(&mut context).is_pending()); - assert!(survivor.as_mut().poll(&mut context).is_pending()); - - let start = Arc::new(Barrier::new(2)); - let signaller = thread::spawn({ - let event = event.clone(); - let start = start.clone(); - move || { - start.wait(); - event.set(); - } - }); - - start.wait(); - drop(cancelled); - signaller.join().unwrap(); - - assert!(survivor.as_mut().poll(&mut context).is_ready()); - drop(survivor); - assert_eq!(Arc::strong_count(&tracker), baseline); - assert_eq!(Arc::strong_count(&event), 1); - } -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn set_releases_all_current_and_future_waiters() { - let event = Arc::new(ManualResetEvent::new()); - let mut tasks = Vec::new(); - for _ in 0..16 { - tasks.push(tokio::spawn(event.clone().wait_owned())); - } - - tokio::task::yield_now().await; - event.set(); - - for task in tasks { - task.await.unwrap(); - } - event.clone().wait_owned().await; -} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 4cdd7a4..d868368 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -16,7 +16,6 @@ // under the License. use std::cell::Cell; -use std::sync::Arc; use asyncband::barrier::Barrier; use asyncband::broadcast; @@ -70,14 +69,10 @@ impl ManageObject for PoolManager { #[test] fn public_types_are_send_and_sync() { fn assert_send_and_sync() {} - fn assert_send_and_sync_value(_: T) {} assert_send_and_sync::(); assert_send_and_sync::(); assert_send_and_sync::(); - let event = ManualResetEvent::new(); - assert_send_and_sync_value(event.wait()); - assert_send_and_sync_value(Arc::new(ManualResetEvent::new()).wait_owned()); assert_send_and_sync::>>(); assert_send_and_sync::>(); assert_send_and_sync::>();