diff --git a/CHANGELOG.md b/CHANGELOG.md index 9543c02..2949716 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +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 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 8e832ba..34e1e14 100644 --- a/README.md +++ b/README.md @@ -62,28 +62,29 @@ 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. | -| | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html) | `completion` | Publish one shared result to any number of current and future observers. | -| | [`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. | +| | [`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` | 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. | +| 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 0370aca..d76cad3 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -49,6 +49,7 @@ blocking = [] broadcast = [] completion = [] 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..96b3ebf --- /dev/null +++ b/asyncband/src/event/mod.rs @@ -0,0 +1,448 @@ +// 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) releases +//! every registered wait and makes future waits ready. The signal remains set until +//! [`reset`](ManualResetEvent::reset) makes new waits block again. +//! +//! 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. +//! +//! 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 +//! +//! ``` +//! # #[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 remains set until explicitly reset. +/// +/// See the [module-level documentation](self) for its waiting semantics. +/// +/// # Synchronization +/// +/// 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 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, +} + +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 currently registered wait. + /// + /// The event remains set until [`reset`](Self::reset) is called. Calling `set` while it is + /// already set has no effect. No ordering is guaranteed among the released waits. + /// + /// # Panics + /// + /// 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 + /// + /// ``` + /// # #[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; + // 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; + true + }) { + if let Some(waker) = waiter.waker.take() { + wakers.push(waker); + } + } + wakers + }; + + wake_all(wakers.into_iter()); + } + + /// 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. + /// + /// # 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; + } + + /// Waits until the event is set. + /// + /// 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 + /// + /// ``` + /// # #[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 async fn wait(&self) { + let fut = ManualResetEventWait { + waiter: None, + event: self, + }; + fut.await + } + + /// Waits until the event is set without borrowing it. + /// + /// 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 + /// + /// ``` + /// # #[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 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. + /// + /// `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<()> { + // 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(retired_waker); + drop(prepared_waker); + return 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 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"); + mem::replace(current, waker) + } +} + +#[must_use = "futures do nothing unless you `.await` or poll them"] +struct ManualResetEventWait<'a> { + waiter: Option, + event: &'a ManualResetEvent, +} + +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); + } +} + +#[must_use = "futures do nothing unless you `.await` or poll them"] +struct OwnedManualResetEventWait { + waiter: Option, + event: Arc, +} + +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 6bec43f..143c60b 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 = "completion", feature = "latch", feature = "mpsc", @@ -50,6 +51,7 @@ pub(crate) mod value_cell; #[cfg(any( feature = "barrier", feature = "broadcast", + feature = "event", feature = "completion", feature = "latch", feature = "mpsc", @@ -74,16 +76,21 @@ 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 = "completion", feature = "latch", feature = "once", @@ -91,6 +98,7 @@ pub(crate) mod waitlist; feature = "watch", ))] // `barrier` constructs a wait set with `with_capacity`, while completion and countdown-based -// primitives use `new`. One constructor is therefore unused in every single-primitive build. +// primitives use `new`; `event` uses only the free `wake_all` helper. One constructor is therefore +// unused in every single-primitive build. #[allow(dead_code)] pub(crate) mod waitset; diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 6beb257..31039bc 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -54,28 +54,29 @@ //! //! # 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. | -//! | | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. | -//! | | [`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. | +//! | | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. | +//! | | [`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. | +//! | 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 //! @@ -128,6 +129,8 @@ pub mod broadcast; pub mod completion; #[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 ba177a2..334072a 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -31,6 +31,7 @@ asyncband = { workspace = true, features = [ "broadcast", "completion", "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..18ada48 --- /dev/null +++ b/benchmarks/asyncband/event/wait.rs @@ -0,0 +1,131 @@ +// 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 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(|_| Box::pin(event.wait())) + .collect::>(); + for waiter in &mut waiters { + poll_pending(waiter.as_mut(), &mut context); + } + + event.set(); + for mut waiter in waiters { + poll_pinned_ready(waiter.as_mut(), &mut context); + } + black_box(event) + }); +} diff --git a/benchmarks/asyncband/main.rs b/benchmarks/asyncband/main.rs index 0388f7e..53ae706 100644 --- a/benchmarks/asyncband/main.rs +++ b/benchmarks/asyncband/main.rs @@ -20,6 +20,7 @@ mod blocking; mod broadcast; mod completion; mod condvar; +mod event; mod latch; mod mpsc; mod mutex; diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 1762c5d..04df589 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -32,6 +32,7 @@ asyncband = { workspace = true, features = [ "broadcast", "completion", "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..87ed2bd --- /dev/null +++ b/tests-integration/tests/event_test.rs @@ -0,0 +1,407 @@ +// 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::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 tests_integration::poll_once; + +struct TrackWake(AtomicUsize); + +impl Wake for TrackWake { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } +} + +#[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 non-reentrant, so waking or dropping this waker inside the critical section deadlocks. +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 + Send>>>>, + fresh_was_pending: AtomicBool, +} + +impl Wake for ResetAndRegister { + fn wake(self: Arc) { + self.event.reset(); + 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); + *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!( + fresh + .as_mut() + .poll(&mut Context::from_waker(Waker::noop())) + .is_pending() + ); + + event.set(); + assert!( + fresh + .as_mut() + .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); +} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 34ff506..d868368 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -21,6 +21,7 @@ use asyncband::barrier::Barrier; use asyncband::broadcast; use asyncband::completion; use asyncband::condvar::Condvar; +use asyncband::event::ManualResetEvent; use asyncband::latch::Latch; use asyncband::mpsc; use asyncband::mutex::Mutex; @@ -71,6 +72,7 @@ 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::>(); @@ -137,6 +139,7 @@ fn public_types_are_unpin() { assert_unpin::(); assert_unpin::(); + assert_unpin::(); assert_unpin::>(); assert_unpin::>(); assert_unpin::(); 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(