diff --git a/CHANGELOG.md b/CHANGELOG.md index 087014c..9b4aa46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,8 @@ All notable changes to this project will be documented in this file. ### New features +* Add `event::AutoResetEvent`, a reusable signal that releases one waiter, retains at most one unassigned signal that can be cleared with `reset`, and transfers assigned signals when waits are cancelled. +* Add `ManualResetEvent::try_wait` to check readiness without registering a waiter or consuming the set state. * Implement `broadcast::mpmc::bounded`, a lossless bounded broadcast channel that retains at most the requested capacity and makes producers wait for the slowest active receiver. * Add opt-in bounded and unbounded `asyncband::mpmc` queues with cloneable producers and competing consumers, delivering each accepted value to exactly one receiver while a receiver remains. * Add an opt-in runtime-agnostic `Phaser` with shared observer handles, dynamic RAII participants registered individually or in batches through an owning iterator, `u64` phase numbers, split arrival/wait with cancellation-resilient retries, and a `close` operation that releases unfinished waits with `Closed`. diff --git a/README.md b/README.md index ed3dae3..2a1bcad 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon | Coordination | [`Semaphore`](https://docs.rs/asyncband/*/asyncband/semaphore/struct.Semaphore.html) | `semaphore` | Limit concurrent work by acquiring permits. | | | [`Barrier`](https://docs.rs/asyncband/*/asyncband/barrier/struct.Barrier.html) | `barrier` | Synchronize a fixed number of participants at a reusable rendezvous. | | | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Signal current and future waits until explicitly reset. | +| | [`AutoResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.AutoResetEvent.html) | `event` | Retain one signal and release one waiter per consumed signal. | | | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a fixed one-way countdown reaches zero. | | | [`Phaser`](https://docs.rs/asyncband/*/asyncband/phaser/struct.Phaser.html) | `phaser` | Coordinate repeated phases with a dynamic participant set. | | | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Dynamically register participants and wait until all have completed. | diff --git a/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs new file mode 100644 index 0000000..af72eae --- /dev/null +++ b/asyncband/src/event/auto_reset.rs @@ -0,0 +1,320 @@ +// 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::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; + +/// A reusable signal that releases one waiter and resets automatically. +/// +/// Each [`set`](Self::set) assigns a signal to one registered wait, or stores one signal +/// if no wait is queued. Repeated sets coalesce only while an unassigned signal is stored. A +/// signal assigned to a wait belongs to that wait until it completes or is cancelled; subsequent +/// sets can release other waits even before previously selected waits are polled again. +/// An unassigned signal can be cleared with [`reset`](Self::reset). +/// +/// Waiting consumes a signal without returning it on completion. Unlike a +/// [`ManualResetEvent`](super::ManualResetEvent), this event does not release all observers of a +/// condition. Unlike a semaphore, it does not count unused signals or return a permit guard. +/// +/// # Usage +/// +/// Use this event for a single worker that rechecks external state after a signal. Publish the +/// state before calling `set`, and check the predicate in a loop. A signal arriving between the +/// predicate check and the first poll is retained, so the worker does not miss it. A leftover +/// signal can cause an extra predicate check without implying new work. +/// +/// Multiple waits compete for signals. The simple check-then-wait loop is not a general +/// multi-consumer queue protocol: several changes can coalesce before those consumers register +/// their waits. +/// +/// # Synchronization +/// +/// Memory operations sequenced before a `set` are visible after a wait or +/// [`try_wait`](Self::try_wait) consumes its signal. This includes sets coalesced into a stored +/// signal and signals passed on after cancellation. +/// +/// The event carries no application state: callers must synchronize access to external predicates +/// separately. +/// +/// # Examples +/// +/// ``` +/// # #[tokio::main] +/// # async fn main() { +/// use asyncband::event::AutoResetEvent; +/// +/// let event = AutoResetEvent::new(); +/// event.set(); +/// event.set(); +/// event.wait().await; +/// assert!(!event.try_wait()); // The two sets coalesced into one signal. +/// +/// # } +/// ``` +pub struct AutoResetEvent { + state: Mutex, +} + +impl AutoResetEvent { + /// Creates an unset event. + pub const fn new() -> Self { + Self::with_state(false) + } + + /// Creates an event with the specified initial state. + /// + /// If `is_set` is `true`, the event stores one signal for a future wait. + pub const fn with_state(is_set: bool) -> Self { + Self { + state: Mutex::new(State { + is_set, + waiters: WaitList::new(), + }), + } + } + + /// Signals one registered wait, or stores one signal if no wait is queued. + /// + /// A stored signal is available to a future wait. Further sets coalesce while it remains + /// unassigned. Creating a wait future does not register it; registration happens when it is + /// first polled without a stored signal. + /// + /// # Panics + /// + /// Panics if waking a selected task panics. Its signal remains assigned and can still be + /// consumed by polling that wait or passed on by dropping it. + pub fn set(&self) { + let waker = self.state.lock().signal(); + if let Some(waker) = waker { + waker.wake(); + } + } + + /// Clears any stored, unassigned signal. + /// + /// Signals already assigned to waits remain theirs. Cancelling such a wait can still transfer + /// or restore its signal after this call. If no signal is stored, this has no effect. + pub fn reset(&self) { + self.state.lock().is_set = false; + } + + /// Returns whether the event is currently set. + /// + /// The event is set while it stores an unassigned signal. Signals already assigned to waits + /// are not reflected in this state. + /// + /// This is a snapshot only; it does not change the event or reserve a signal for a later wait. + /// The state may change immediately after this call. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::AutoResetEvent; + /// + /// let event = AutoResetEvent::with_state(true); + /// assert!(event.is_set()); + /// assert!(event.is_set()); + /// ``` + pub fn is_set(&self) -> bool { + self.state.lock().is_set + } + + /// Attempts to wait without registering a waiter. + /// + /// Returns `true` if a stored signal was consumed. This never takes a signal assigned to + /// another wait. A `false` result is only a snapshot; use [`wait`](Self::wait) to wait for a + /// future signal. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::AutoResetEvent; + /// + /// let event = AutoResetEvent::with_state(true); + /// assert!(event.try_wait()); + /// assert!(!event.try_wait()); // A successful wait consumes the signal. + /// ``` + pub fn try_wait(&self) -> bool { + mem::take(&mut self.state.lock().is_set) + } + + /// Waits for and consumes one signal. + /// + /// The first poll consumes a stored signal immediately, or registers the wait. + /// Merely creating this future neither reserves a signal nor registers the wait. + /// Signals assigned to other waits cannot be consumed by this wait. + /// + /// # Cancel safety + /// + /// Dropping this future before it returns `Ready` removes its registration. If a signal was + /// assigned to it, that signal is passed to another registered wait or stored for a future + /// wait, coalescing with any signal already stored. Dropping a completed wait does not return + /// its consumed signal. + pub async fn wait(&self) { + Wait { + event: self, + waiter: None, + } + .await + } + + /// Waits without borrowing the event. + /// + /// The future owns the [`Arc`], making 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::AutoResetEvent; + /// + /// let event = Arc::new(AutoResetEvent::new()); + /// let waiter = tokio::spawn(event.clone().wait_owned()); + /// event.set(); + /// waiter.await.unwrap(); + /// # } + /// ``` + pub async fn wait_owned(self: Arc) { + self.wait().await; + } + + fn poll_wait(&self, waiter_id: &mut Option, cx: &mut Context<'_>) -> Poll<()> { + let (poll, retired_waker) = { + let mut state = self.state.lock(); + match *waiter_id { + Some(id) => match state.waiters.waiter_mut(id) { + Waiter::Notified => { + state.waiters.remove_unlinked_waiter(id); + *waiter_id = None; + (Poll::Ready(()), None) + } + Waiter::Waiting(waker) => { + let retired = (!waker.will_wake(cx.waker())) + .then(|| mem::replace(waker, cx.waker().clone())); + (Poll::Pending, retired) + } + }, + None if state.is_set => { + state.is_set = false; + (Poll::Ready(()), None) + } + None => { + *waiter_id = Some(state.waiters.push_back(Waiter::Waiting(cx.waker().clone()))); + (Poll::Pending, None) + } + } + }; + drop(retired_waker); + poll + } + + fn unregister_waiter(&self, id: WaiterId) { + let (waiter, waker) = { + let mut state = self.state.lock(); + // A selected waiter is already detached, but still owns its signal until removal. + state.waiters.unlink_waiter(id, |_| true); + let waiter = state.waiters.remove_unlinked_waiter(id); + let waker = match &waiter { + Waiter::Notified => state.signal(), + Waiter::Waiting(_) => None, + }; + (waiter, waker) + }; + drop(waiter); + if let Some(waker) = waker { + waker.wake(); + } + } +} + +impl Default for AutoResetEvent { + fn default() -> Self { + Self::new() + } +} + +impl fmt::Debug for AutoResetEvent { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let is_set = self.state.lock().is_set; + f.debug_struct("AutoResetEvent") + .field("is_set", &is_set) + .finish_non_exhaustive() + } +} + +struct State { + // A stored signal and queued (unselected) waits never coexist. Detached, selected waits can + // coexist with either: their signals are reserved until consumption or cancellation. + is_set: bool, + waiters: WaitList, +} + +impl State { + fn signal(&mut self) -> Option { + if let Some((_, waiter)) = self.waiters.unlink_first_waiter(|_| true) { + let Waiter::Waiting(waker) = mem::replace(waiter, Waiter::Notified) else { + unreachable!("only unselected waits remain queued") + }; + Some(waker) + } else { + self.is_set = true; + None + } + } +} + +enum Waiter { + Waiting(Waker), + Notified, +} + +#[must_use = "futures do nothing unless you `.await` or poll them"] +struct Wait<'a> { + event: &'a AutoResetEvent, + waiter: Option, +} + +impl Future for Wait<'_> { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + this.event.poll_wait(&mut this.waiter, cx) + } +} + +impl Drop for Wait<'_> { + fn drop(&mut self) { + if let Some(id) = self.waiter.take() { + self.event.unregister_waiter(id); + } + } +} diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs new file mode 100644 index 0000000..d51b16a --- /dev/null +++ b/asyncband/src/event/manual_reset.rs @@ -0,0 +1,339 @@ +// 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::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::wake_all; +use crate::internal::waker_batch::WakerBatch; + +/// A reusable signal that releases all waiters and remains set until explicitly reset. +/// +/// Each [`set`](Self::set) releases all registered waits and makes future waits ready. The event +/// remains set until [`reset`](Self::reset) is called. A released wait remains ready even if the +/// event is reset before that wait is polled again. +/// +/// # Usage +/// +/// Use this event as a readiness gate, keeping it set for as long as the condition holds. Unlike +/// a latch, it can be reset and reused. Creating a wait future does not register it; registration +/// happens on the first poll that needs to wait. A `set` followed immediately by `reset` therefore +/// does not release an unpolled wait. +/// +/// # Synchronization +/// +/// An unset-to-set transition synchronizes with the waits it releases, with waits first polled +/// while the event remains set, and with successful [`try_wait`](Self::try_wait) calls that observe +/// that set state. Memory operations sequenced before [`set`](Self::set) are therefore visible +/// after those waits complete. +/// +/// A `set` call that finds the event already set does not establish this guarantee. +/// +/// The event carries no application state: callers must synchronize access to external predicates +/// separately. +/// +/// # Examples +/// +/// ``` +/// # #[tokio::main] +/// # async fn main() { +/// use asyncband::event::ManualResetEvent; +/// +/// let event = ManualResetEvent::new(); +/// event.set(); +/// event.wait().await; +/// event.wait().await; // Waiting leaves the event set. +/// event.reset(); +/// assert!(!event.is_set()); +/// +/// # } +/// ``` +pub struct ManualResetEvent { + state: Mutex, +} + +impl ManualResetEvent { + /// Creates an unset event. + pub const fn new() -> Self { + Self::with_state(false) + } + + /// Creates an event with the specified initial state. + /// + /// If `is_set` is `true`, waits complete immediately until the event is reset. + pub const fn with_state(is_set: bool) -> Self { + Self { + state: Mutex::new(State { + is_set, + waiters: WaitList::new(), + }), + } + } + + /// Signals all registered waits and keeps the event set. + /// + /// The event remains set until [`reset`](Self::reset) is called. Calling `set` while it is + /// already set has no effect. + /// + /// # Panics + /// + /// Panics if waking a selected task panics. The event remains set, and waking is still + /// attempted for every other selected task before the panic resumes. + 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 = WakerBatch::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()); + } + + /// Clears the set state. + /// + /// Waits already released by a preceding [`set`](Self::set) remain ready. If the event is + /// already unset, this has no effect. + pub fn reset(&self) { + self.state.lock().is_set = false; + } + + /// Returns whether the event is currently set. + /// + /// This is a snapshot only; it does not change the event or reserve a signal for a later wait. + /// The state may change immediately after this call. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::with_state(true); + /// assert!(event.is_set()); + /// assert!(event.is_set()); + /// ``` + pub fn is_set(&self) -> bool { + self.state.lock().is_set + } + + /// Attempts to wait without registering a waiter. + /// + /// Returns `true` if the event is set, leaving it set. A `false` result is only a snapshot; + /// use [`wait`](Self::wait) to wait for a future signal. + /// + /// # Examples + /// + /// ``` + /// use asyncband::event::ManualResetEvent; + /// + /// let event = ManualResetEvent::with_state(true); + /// assert!(event.try_wait()); + /// assert!(event.try_wait()); // A successful wait leaves the event set. + /// ``` + pub fn try_wait(&self) -> bool { + self.is_set() + } + + /// Waits until the event is set. + /// + /// The first poll completes immediately if the event is set, or registers the wait. + /// Merely creating this future does not register the wait. Once a [`set`](Self::set) releases + /// a registered wait, a later [`reset`](Self::reset) cannot make that wait pending again. + /// + /// # Cancel safety + /// + /// Dropping this future before it returns `Ready` removes its registration. This does not + /// change the event or affect other waits. + pub async fn wait(&self) { + Wait { + event: self, + waiter: None, + } + .await + } + + /// Waits without borrowing the event. + /// + /// The future owns the [`Arc`], making 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) { + self.wait().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<()> { + 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); + let retired = (!waiter.will_wake(cx.waker())) + .then(|| waiter.replace_waker(cx.waker().clone())); + (Poll::Pending, retired) + } + 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(retired_waker); + poll + } + + fn unregister_waiter(&self, id: WaiterId) { + 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 { + let is_set = self.state.lock().is_set; + f.debug_struct("ManualResetEvent") + .field("is_set", &is_set) + .finish_non_exhaustive() + } +} + +#[derive(Debug)] +struct State { + is_set: bool, + waiters: WaitList, +} + +impl State { + /// Removes a waiter whether or not [`ManualResetEvent::set`] already unlinked it. + fn remove_waiter(&mut self, id: WaiterId) -> Waiter { + // Unlinking is idempotent: a waiter that `set` detached keeps its node until it is removed + // here, and an unconditional predicate never declines. + self.waiters.unlink_waiter(id, |_| true); + self.waiters.remove_unlinked_waiter(id) + } +} + +#[derive(Debug)] +struct Waiter { + notified: bool, + 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 Wait<'a> { + event: &'a ManualResetEvent, + waiter: Option, +} + +impl Future for Wait<'_> { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + this.event.poll_wait(&mut this.waiter, cx) + } +} + +impl Drop for Wait<'_> { + fn drop(&mut self) { + if let Some(id) = self.waiter.take() { + self.event.unregister_waiter(id); + } + } +} diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index 1afa051..8f94320 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -15,415 +15,21 @@ // specific language governing permissions and limitations // under the License. -//! A reusable, level-triggered signal for coordinating tasks. +//! Reusable signals for coordinating tasks without carrying a value. //! -//! 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. +//! An [`AutoResetEvent`] releases one registered wait per assigned signal. With no queued waits, +//! it retains at most one signal, coalescing further sets. A [`ManualResetEvent`] releases all +//! registered waits and remains set until explicitly reset. //! -//! 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. +//! Both types offer `is_set` to inspect the state without changing it and `try_wait` to attempt +//! an immediate wait. A successful wait consumes the signal only for an auto-reset event. //! -//! 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::wake_all; -use crate::internal::waker_batch::WakerBatch; - -/// 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 notifying a waiting task panics. The event remains set, and notification is still - /// attempted for every other current waiter before the 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 = WakerBatch::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`](Self::set), the wait either unregisters - /// first or has already been released by that call. - /// - /// # 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<()> { - 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); - let retired = (!waiter.will_wake(cx.waker())) - .then(|| waiter.replace_waker(cx.waker().clone())); - (Poll::Pending, retired) - } - 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(retired_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 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 = (); +//! Both types retain state, unlike a condition variable's unbuffered notifications. Use a +//! semaphore when unused permits must accumulate, or a watch channel when each receiver needs to +//! observe state changes independently. - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Self { waiter, event } = self.get_mut(); - event.poll_wait(waiter, cx) - } -} +mod auto_reset; +mod manual_reset; -impl Drop for OwnedManualResetEventWait { - fn drop(&mut self) { - self.event.unregister_waiter(&mut self.waiter); - } -} +pub use self::auto_reset::AutoResetEvent; +pub use self::manual_reset::ManualResetEvent; diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index afd6faf..a68e63d 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -62,6 +62,7 @@ //! | Coordination | [`Semaphore`](semaphore::Semaphore) | `semaphore` | Limit concurrent work by acquiring permits. | //! | | [`Barrier`](barrier::Barrier) | `barrier` | Synchronize a fixed number of participants at a reusable rendezvous. | //! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Signal current and future waits until explicitly reset. | +//! | | [`AutoResetEvent`](event::AutoResetEvent) | `event` | Retain one signal and release one waiter per consumed signal. | //! | | [`Latch`](latch::Latch) | `latch` | Wait until a fixed one-way countdown reaches zero. | //! | | [`Phaser`](phaser::Phaser) | `phaser` | Coordinate repeated phases with a dynamic participant set. | //! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Dynamically register participants and wait until all have completed. | diff --git a/examples/AGENTS.md b/examples/AGENTS.md new file mode 100644 index 0000000..fc6cb48 --- /dev/null +++ b/examples/AGENTS.md @@ -0,0 +1,26 @@ + + +# Example Style + +- Start each example with concise module documentation describing the scenario and the behavior it demonstrates. +- Omit run commands, basic Cargo instructions, and comments that merely restate the code. Assume readers know how to run a Rust example. +- Keep each example focused on a small, coherent scenario. Explain relevant semantic differences beside the code; avoid catalogs of unrelated primitives. +- For migration examples, show how the same application scenario works before and after the change, including any protocol changes or limits. Do not substitute a tour of primitive APIs for a migration. +- Keep example explanations in the example source. The repository README only needs a concise entry point. diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 47a82de..61f060e 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -28,10 +28,12 @@ release = false [dependencies] asyncband = { workspace = true, features = [ "completion", + "event", "lazy-cell", "once-cell", "phaser", "shutdown", + "watch", ] } tokio = { workspace = true, features = [ "macros", @@ -44,6 +46,14 @@ tokio = { workspace = true, features = [ [lints] workspace = true +[[example]] +name = "coalesced_worker" +path = "src/coalesced_worker.rs" + +[[example]] +name = "notify_vs_event" +path = "src/notify_vs_event.rs" + [[example]] name = "once_cell_vs_lazy_cell" path = "src/once_cell_vs_lazy_cell.rs" diff --git a/examples/src/coalesced_worker.rs b/examples/src/coalesced_worker.rs new file mode 100644 index 0000000..52a1cd3 --- /dev/null +++ b/examples/src/coalesced_worker.rs @@ -0,0 +1,79 @@ +// 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. + +//! One worker rebuilds a snapshot of the latest requested revision. Intermediate revisions may +//! coalesce: this is not a queue of jobs that must each run, or a broadcast to multiple observers. + +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +use asyncband::event::AutoResetEvent; + +#[derive(Default)] +struct Rebuilder { + requested: AtomicUsize, + stopped: AtomicBool, + changed: AutoResetEvent, +} + +impl Rebuilder { + fn request(&self, revision: usize) { + self.requested.fetch_max(revision, Ordering::Release); + self.changed.set(); + } + + fn stop(&self) { + self.stopped.store(true, Ordering::Release); + self.changed.set(); + } + + async fn run(&self) { + let mut rebuilt = 0; + loop { + // Observe stop before the revision so all requests preceding stop are included. + let stopped = self.stopped.load(Ordering::Acquire); + let requested = self.requested.load(Ordering::Acquire); + if requested != rebuilt { + println!("Rebuild snapshot at revision {requested}"); + rebuilt = requested; + } + if stopped { + break; + } + // A request between the check and this wait leaves a consumable signal. There is + // only one worker, so no other observer can consume that signal instead. + self.changed.wait().await; + } + } +} + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let rebuilder = Rebuilder::default(); + tokio::join!(biased; rebuilder.run(), async { + // The worker is already waiting. This burst still requires only the latest snapshot. + rebuilder.request(1); + rebuilder.request(2); + rebuilder.request(3); + tokio::task::yield_now().await; + + // After rebuilding revision 3, the worker waits for another update. + rebuilder.request(4); + rebuilder.stop(); + }); +} diff --git a/examples/src/graceful_shutdown.rs b/examples/src/graceful_shutdown.rs index abbfd46..36ec43e 100644 --- a/examples/src/graceful_shutdown.rs +++ b/examples/src/graceful_shutdown.rs @@ -15,6 +15,9 @@ // specific language governing permissions and limitations // under the License. +//! Stop a service on Ctrl+C or an administrative request, wait for worker cleanup, and abort the +//! worker if the shutdown deadline expires. + use std::error::Error; use std::time::Duration; diff --git a/examples/src/lazy_cell_boxed_vs_inline.rs b/examples/src/lazy_cell_boxed_vs_inline.rs index 8feb15e..5bc1d8c 100644 --- a/examples/src/lazy_cell_boxed_vs_inline.rs +++ b/examples/src/lazy_cell_boxed_vs_inline.rs @@ -15,6 +15,9 @@ // specific language governing permissions and limitations // under the License. +//! Choose how to store a lazy initializer: box its future to keep the cell movable, pin the cell +//! locally to store the future inline, or share a pinned cell across tasks. + use std::sync::Arc; use asyncband::once::LazyCell; diff --git a/examples/src/notify_vs_event.rs b/examples/src/notify_vs_event.rs new file mode 100644 index 0000000..84cb529 --- /dev/null +++ b/examples/src/notify_vs_event.rs @@ -0,0 +1,189 @@ +// 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. + +//! Migrate two uses of Tokio Notify: wake a cache worker to flush queued entries, and let search +//! requests wait until an index reaches the revision they need. Each scenario has a Tokio version +//! followed by an Asyncband version with the same application behavior. +//! +//! Asyncband has no single primitive preserving Notify's combined notify_one/notify_waiters +//! contract on the same waiters. The migrations below separate worker and reader notifications. + +use std::collections::VecDeque; +use std::sync::Mutex; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +use asyncband::event::AutoResetEvent; +use asyncband::watch; +use tokio::sync::Notify; + +#[tokio::main(flavor = "current_thread")] +async fn main() { + cache_worker_with_notify().await; + cache_worker_with_event().await; + index_readers_with_notify().await; + index_readers_with_watch().await; +} + +// One worker owns flushing. Producers keep the entries in a queue and notify it after enqueueing. +// This demonstration ends after three entries; a service would keep processing until shutdown. +async fn cache_worker_with_notify() { + let entries = Mutex::new(VecDeque::new()); + let changed = Notify::new(); + let worker = async { + let mut flushed = Vec::new(); + while flushed.len() < 3 { + let entry = entries.lock().unwrap().pop_front(); + if let Some(entry) = entry { + // A real worker writes this entry to storage here. + flushed.push(entry); + } else { + changed.notified().await; + } + } + flushed + }; + let enqueue = async { + for entry in ["alice", "bob", "carol"] { + entries.lock().unwrap().push_back(entry); + changed.notify_one(); + } + }; + + let (flushed, ()) = tokio::join!(biased; worker, enqueue); + assert_eq!(flushed, ["alice", "bob", "carol"]); + println!("Notify worker flushed {flushed:?}"); +} + +// Keep the queue and processing loop. Replace notify_one/notified with set/wait. +// Signals may coalesce, but entries do not: the worker drains the queue before waiting again. +// With competing consumers, migrate the queue to mpmc or redesign its registration protocol: +// AutoResetEvent has no counterpart to Notified::enable(), which registers before checking work. +async fn cache_worker_with_event() { + let entries = Mutex::new(VecDeque::new()); + let changed = AutoResetEvent::new(); + let worker = async { + let mut flushed = Vec::new(); + while flushed.len() < 3 { + let entry = entries.lock().unwrap().pop_front(); + if let Some(entry) = entry { + flushed.push(entry); + } else { + // With one consumer, an enqueue between the empty check and this wait leaves + // a signal for us. Checking the queue again also handles leftover signals. + changed.wait().await; + } + } + flushed + }; + let enqueue = async { + for entry in ["alice", "bob", "carol"] { + entries.lock().unwrap().push_back(entry); + changed.set(); + } + }; + + let (flushed, ()) = tokio::join!(biased; worker, enqueue); + assert_eq!(flushed, ["alice", "bob", "carol"]); + println!("AutoResetEvent worker flushed {flushed:?}"); +} + +// A search request must wait for its writes to become searchable. Each request has a required +// index revision; publishing a revision wakes all requests so they can recheck their own target. +async fn index_readers_with_notify() { + let indexed = AtomicUsize::new(0); + let changed = Notify::new(); + let publish = async { + for revision in 1..=2 { + // The indexer finishes applying this revision before publishing it. + indexed.store(revision, Ordering::Release); + changed.notify_waiters(); + tokio::task::yield_now().await; + } + }; + + let (first, second, ()) = tokio::join!(biased; + wait_for_index_with_notify(&indexed, &changed, 1), + wait_for_index_with_notify(&indexed, &changed, 2), + publish, + ); + assert!(first >= 1); + assert!(second >= 2); + // A late request succeeds by checking the index, even though it missed the notification. + let late = wait_for_index_with_notify(&indexed, &changed, 2).await; + assert_eq!(late, 2); + println!("Notify readers reached revisions {first}, {second}, {late}"); +} + +async fn wait_for_index_with_notify( + indexed: &AtomicUsize, + changed: &Notify, + required: usize, +) -> usize { + loop { + // Create the future BEFORE checking. notify_waiters reaches existing futures even if + // they have not been polled, covering an update between the check and the await. + let notified = changed.notified(); + let revision = indexed.load(Ordering::Acquire); + if revision >= required { + return revision; + } + notified.await; + } +} + +// Publish the indexed revision through watch, replacing both the atomic and the broadcast. +// Each request subscribes before checking and independently waits for its required revision. +// Each request keeps its subscription across retries, so cancelling a changed() wait does not +// lose an unseen revision update. +// A ManualResetEvent set/reset pulse would miss unpolled waits; leaving it set admits future waits. +async fn index_readers_with_watch() { + let (indexed, mut first_request) = watch::channel(0); + let mut second_request = indexed.subscribe(); + let publish = async { + for revision in 1..=2 { + // Publishing also works when there are temporarily no requests listening. + indexed.send_replace(revision); + tokio::task::yield_now().await; + } + }; + + let (first, second, ()) = tokio::join!(biased; + wait_for_index_with_watch(&mut first_request, 1), + wait_for_index_with_watch(&mut second_request, 2), + publish, + ); + assert!(first >= 1); + assert!(second >= 2); + let mut late_request = indexed.subscribe(); + let late = wait_for_index_with_watch(&mut late_request, 2).await; + assert_eq!(late, 2); + println!("Watch readers reached revisions {first}, {second}, {late}"); +} + +async fn wait_for_index_with_watch(indexed: &mut watch::Receiver, required: usize) -> usize { + loop { + let revision = indexed.get(); + if revision >= required { + return revision; + } + // The receiver remembers updates between get() and changed(), including ones published + // before changed() is first polled. Intermediate revisions may coalesce; the target + // predicate, rather than a notification count, determines when this request can proceed. + indexed.changed().await.unwrap(); + } +} diff --git a/examples/src/once_cell_vs_lazy_cell.rs b/examples/src/once_cell_vs_lazy_cell.rs index 726a117..caad33a 100644 --- a/examples/src/once_cell_vs_lazy_cell.rs +++ b/examples/src/once_cell_vs_lazy_cell.rs @@ -15,14 +15,18 @@ // specific language governing permissions and limitations // under the License. +//! Initialize service endpoints and clients with OnceCell or LazyCell. Compare fixed initializers, +//! access-time configuration, and a captured credential initializer that survives caller +//! cancellation. + use std::future::Ready; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; +use asyncband::event::AutoResetEvent; use asyncband::once::LazyCell; use asyncband::once::OnceCell; -use tokio::sync::Notify; static ONCE_ENDPOINT: OnceCell = OnceCell::new(); static LAZY_ENDPOINT: LazyCell> = LazyCell::new(load_default_endpoint); @@ -81,8 +85,8 @@ async fn lazy_cell_owns_a_local_fn_once() { } let attempts = Arc::new(AtomicUsize::new(0)); - let started = Arc::new(Notify::new()); - let resume = Arc::new(Notify::new()); + let started = Arc::new(AutoResetEvent::new()); + let resume = Arc::new(AutoResetEvent::new()); let credentials = Credentials { token: "secret".to_owned(), }; @@ -98,8 +102,8 @@ async fn lazy_cell_owns_a_local_fn_once() { async move { attempts.fetch_add(1, Ordering::SeqCst); - started.notify_one(); - resume.notified().await; + started.set(); + resume.wait().await; Client { token } } } @@ -116,13 +120,13 @@ async fn lazy_cell_owns_a_local_fn_once() { LazyCell::force_pin(client.as_ref()).await; } }); - started.notified().await; + started.wait().await; first_caller.abort(); assert!(first_caller.await.unwrap_err().is_cancelled()); // Cancellation does not consume the captured credentials or restart the initializer. The next // caller resumes the same future. - resume.notify_one(); + resume.set(); assert_eq!(LazyCell::force_pin(client.as_ref()).await.token, "secret"); assert_eq!(attempts.load(Ordering::SeqCst), 1); } diff --git a/tests-integration/tests/auto_reset_event_test.rs b/tests-integration/tests/auto_reset_event_test.rs new file mode 100644 index 0000000..dc0e66b --- /dev/null +++ b/tests-integration/tests/auto_reset_event_test.rs @@ -0,0 +1,361 @@ +// 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::sync::Arc; +use std::sync::Barrier; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Wake; +use std::task::Waker; +use std::thread; + +use asyncband::blocking::FutureExt; +use asyncband::event::AutoResetEvent; +use tests_integration::PanicWake; +use tests_integration::WakeCounter; +use tests_integration::assert_completes_without_deadlock; +use tests_integration::poll_once; + +#[test] +fn unpolled_waits_do_not_reserve_stored_signals() { + let event = AutoResetEvent::new(); + assert!(!event.is_set()); + let mut first = pin!(event.wait()); + let mut second = pin!(event.wait()); + + event.set(); + event.set(); + assert!(event.is_set()); + assert!(poll_once(second.as_mut()).is_ready()); + assert!(!event.is_set()); + assert!(!event.try_wait()); + assert!(poll_once(first.as_mut()).is_pending()); + + event.set(); + assert!(poll_once(first.as_mut()).is_ready()); + assert!(!event.try_wait()); +} + +#[test] +fn reset_discards_only_unassigned_signals() { + let event = AutoResetEvent::new(); + let mut selected = pin!(event.wait()); + assert!(poll_once(selected.as_mut()).is_pending()); + event.set(); + assert!(!event.is_set()); // The signal belongs to the selected wait. + + let mut unpolled = pin!(event.wait()); + event.set(); // One signal is assigned and another is stored. + assert!(event.is_set()); + event.reset(); + assert!(!event.is_set()); + assert!(!event.try_wait()); + assert!(poll_once(unpolled.as_mut()).is_pending()); + assert!(poll_once(selected.as_mut()).is_ready()); + + event.reset(); + event.set(); + assert!(poll_once(unpolled.as_mut()).is_ready()); + assert!(!event.try_wait()); +} + +#[test] +fn reset_preserves_cancellation_handoff() { + let event = AutoResetEvent::new(); + let mut selected = Box::pin(event.wait()); + assert!(poll_once(selected.as_mut()).is_pending()); + event.set(); + event.reset(); + + let mut remaining = Box::pin(event.wait()); + assert!(poll_once(remaining.as_mut()).is_pending()); + drop(selected); + assert!(!event.is_set()); + assert!(!event.try_wait()); + + // The transferred signal also survives reset and can be restored by cancellation. + event.reset(); + drop(remaining); + assert!(event.is_set()); + assert!(event.try_wait()); + assert!(!event.is_set()); + assert!(!event.try_wait()); +} + +#[test] +fn assigned_signals_cannot_be_stolen() { + let event = AutoResetEvent::new(); + let first_wake = Arc::new(WakeCounter::default()); + let second_wake = Arc::new(WakeCounter::default()); + let first_waker = Waker::from(first_wake.clone()); + let second_waker = Waker::from(second_wake.clone()); + let mut first = pin!(event.wait()); + let mut second = pin!(event.wait()); + assert!( + first + .as_mut() + .poll(&mut Context::from_waker(&first_waker)) + .is_pending() + ); + assert!( + second + .as_mut() + .poll(&mut Context::from_waker(&second_waker)) + .is_pending() + ); + + event.set(); + assert_eq!(first_wake.count() + second_wake.count(), 1); + assert!(!event.try_wait()); + let (unselected, waker) = if first_wake.count() == 0 { + (first.as_mut(), &first_waker) + } else { + (second.as_mut(), &second_waker) + }; + assert!( + unselected + .poll(&mut Context::from_waker(waker)) + .is_pending() + ); + + // Another set serves the remaining waiter before the selected wait is polled again. + event.set(); + assert_eq!(first_wake.count(), 1); + assert_eq!(second_wake.count(), 1); + let mut newcomer = pin!(event.wait()); + assert!(poll_once(newcomer.as_mut()).is_pending()); + assert!(!event.try_wait()); + assert!(poll_once(second.as_mut()).is_ready()); + assert!(poll_once(first.as_mut()).is_ready()); + event.set(); + assert!(poll_once(newcomer.as_mut()).is_ready()); + assert!(!event.try_wait()); +} + +#[test] +fn cancelling_selected_waits_transfers_then_restores_the_signal() { + let event = AutoResetEvent::new(); + let mut first = Box::pin(event.wait()); + let mut second = Box::pin(event.wait()); + let mut third = Box::pin(event.wait()); + assert!(poll_once(first.as_mut()).is_pending()); + event.set(); + assert!(poll_once(second.as_mut()).is_pending()); + drop(first); + assert!(!event.try_wait()); + assert!(poll_once(third.as_mut()).is_pending()); + drop(second); + assert!(!event.try_wait()); + drop(third); + assert!(event.try_wait()); + assert!(!event.try_wait()); +} + +#[test] +fn cancelling_unselected_waits_does_not_add_a_signal() { + let event = AutoResetEvent::new(); + let mut cancelled = Box::pin(event.wait()); + let mut remaining = Box::pin(event.wait()); + assert!(poll_once(cancelled.as_mut()).is_pending()); + assert!(poll_once(remaining.as_mut()).is_pending()); + + drop(cancelled); + assert!(!event.try_wait()); + assert!(poll_once(remaining.as_mut()).is_pending()); + event.set(); + assert!(poll_once(remaining.as_mut()).is_ready()); + drop(remaining); + assert!(!event.try_wait()); +} + +#[test] +fn returned_signals_coalesce_with_an_already_stored_signal() { + let event = AutoResetEvent::new(); + let mut first = Box::pin(event.wait()); + let mut second = Box::pin(event.wait()); + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(second.as_mut()).is_pending()); + event.set(); + event.set(); + event.set(); + + drop(first); + drop(second); + assert!(event.try_wait()); + assert!(!event.try_wait()); +} + +#[test] +fn completed_owned_waits_do_not_return_their_signal() { + let event = Arc::new(AutoResetEvent::with_state(true)); + let weak = Arc::downgrade(&event); + let mut completed = Box::pin(event.clone().wait_owned()); + assert!(poll_once(completed.as_mut()).is_ready()); + drop(completed); + assert!(!event.try_wait()); + + let mut pending = Box::pin(event.clone().wait_owned()); + assert!(poll_once(pending.as_mut()).is_pending()); + event.set(); + drop(pending); + assert!(event.try_wait()); + drop(event); + assert!(weak.upgrade().is_none()); +} + +#[test] +fn waker_replacement_and_cancellation_release_registrations() { + let event = AutoResetEvent::new(); + let first = Arc::new(WakeCounter::default()); + let second = Arc::new(WakeCounter::default()); + let first_waker = Waker::from(first.clone()); + let second_waker = Waker::from(second.clone()); + let mut wait = Box::pin(event.wait()); + for _ in 0..2 { + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&first_waker)) + .is_pending() + ); + assert_eq!(Arc::strong_count(&first), 3); + } + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&second_waker)) + .is_pending() + ); + assert_eq!(Arc::strong_count(&first), 2); + assert_eq!(Arc::strong_count(&second), 3); + drop(wait); + assert_eq!(Arc::strong_count(&second), 2); + event.set(); + assert_eq!(first.count(), 0); + assert_eq!(second.count(), 0); + assert!(event.try_wait()); +} + +struct ReentrantWaker(Arc); + +impl Wake for ReentrantWaker { + fn wake(self: Arc) { + self.0.try_wait(); + } +} + +impl Drop for ReentrantWaker { + fn drop(&mut self) { + self.0.try_wait(); + } +} + +#[test] +fn wake_and_waker_destruction_happen_outside_the_event_lock() { + assert_completes_without_deadlock(|| { + let event = Arc::new(AutoResetEvent::new()); + let mut first = Box::pin(event.wait()); + let mut second = Box::pin(event.wait()); + assert!(poll_once(first.as_mut()).is_pending()); + event.set(); + { + let waker = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + second + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + } + drop(first); // The cancellation handoff wakes the second waiter. + assert!(poll_once(second.as_mut()).is_ready()); + + let mut wait = Box::pin(event.wait()); + { + let waker = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + } + event.set(); + assert!(poll_once(wait.as_mut()).is_ready()); + + let mut cancelled = Box::pin(event.wait()); + for _ in 0..2 { + let waker = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + cancelled + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + } + drop(cancelled); // Both replaced and cancelled registrations release their last waker. + }); +} + +#[test] +fn a_panicking_wake_leaves_its_signal_available_for_cancellation_handoff() { + let event = AutoResetEvent::new(); + let waker = Waker::from(Arc::new(PanicWake)); + let mut selected = Box::pin(event.wait()); + let mut next = Box::pin(event.wait()); + assert!( + selected + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + assert!(panic::catch_unwind(AssertUnwindSafe(|| event.set())).is_err()); + assert!(!event.try_wait()); + assert!(poll_once(next.as_mut()).is_pending()); + drop(selected); + assert!(poll_once(next.as_mut()).is_ready()); + assert!(!event.try_wait()); +} + +#[test] +fn concurrent_sets_and_waits_publish_state_without_losing_signals() { + assert_completes_without_deadlock(|| { + let event = AutoResetEvent::new(); + let round = Barrier::new(2); + let value = AtomicUsize::new(0); + thread::scope(|scope| { + scope.spawn(|| { + for expected in 1..=100 { + round.wait(); + value.store(expected, Ordering::Relaxed); + event.set(); + round.wait(); + } + }); + for expected in 1..=100 { + // Registration races with set; the second barrier prevents the next set from + // coalescing before this round's signal has been consumed. + round.wait(); + event.wait().block_on(); + assert_eq!(value.load(Ordering::Relaxed), expected); + round.wait(); + } + }); + }); +} diff --git a/tests-integration/tests/event_test.rs b/tests-integration/tests/event_test.rs index 122a851..f234a7c 100644 --- a/tests-integration/tests/event_test.rs +++ b/tests-integration/tests/event_test.rs @@ -37,12 +37,17 @@ use tests_integration::poll_once; #[test] fn set_is_sticky_and_reset_blocks_new_waiters() { let event = ManualResetEvent::new(); + assert!(!event.try_wait()); event.set(); + assert!(event.try_wait()); + assert!(event.try_wait()); let mut ready = pin!(event.wait()); assert!(poll_once(ready.as_mut()).is_ready()); + assert!(event.is_set()); event.reset(); + assert!(!event.try_wait()); let mut pending = pin!(event.wait()); assert!(poll_once(pending.as_mut()).is_pending()); } @@ -317,6 +322,8 @@ fn set_then_reset_commits_registered_waiters() { event.set(); event.reset(); + assert!(!event.is_set()); + assert!(!event.try_wait()); assert_eq!(tracker.count(), 1); assert!(wait.as_mut().poll(&mut context).is_ready()); diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index 00d44dc..26205e3 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -24,6 +24,7 @@ use asyncband::barrier::Barrier; use asyncband::broadcast; use asyncband::completion; use asyncband::condvar::Condvar; +use asyncband::event::AutoResetEvent; use asyncband::event::ManualResetEvent; use asyncband::latch::Latch; use asyncband::mpmc; @@ -81,6 +82,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::>(); assert_send_and_sync::>(); @@ -173,6 +175,7 @@ fn public_types_are_unpin() { assert_unpin::(); assert_unpin::(); assert_unpin::(); + assert_unpin::(); assert_unpin::>(); assert_unpin::>(); assert_unpin::();