From 085390437eba05a76efdcfd7e4388b1ea0023dbc Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 19:41:08 +0800 Subject: [PATCH 1/8] feat(event): add AutoResetEvent --- CHANGELOG.md | 1 + README.md | 1 + asyncband/src/event/auto_reset.rs | 270 +++++++++++++++ asyncband/src/event/mod.rs | 30 +- asyncband/src/lib.rs | 1 + examples/Cargo.toml | 5 + examples/src/coalesced_worker.rs | 77 +++++ .../tests/auto_reset_event_test.rs | 308 ++++++++++++++++++ tests-integration/tests/traits_test.rs | 3 + 9 files changed, 693 insertions(+), 3 deletions(-) create mode 100644 asyncband/src/event/auto_reset.rs create mode 100644 examples/src/coalesced_worker.rs create mode 100644 tests-integration/tests/auto_reset_event_test.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 087014c..19263c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ 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, and transfers assigned signals when waits are cancelled. * 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..57c451d --- /dev/null +++ b/asyncband/src/event/auto_reset.rs @@ -0,0 +1,270 @@ +// 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 the oldest 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. +/// +/// 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. +/// +/// # 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: when it +/// is used to notify changes to an external predicate, callers must synchronize access to that +/// predicate 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 event without a stored signal. + pub const fn new() -> Self { + Self::with_state(false) + } + + /// Creates an event with one stored signal if `is_set` is `true`, or none otherwise. + pub const fn with_state(is_set: bool) -> Self { + Self { + state: Mutex::new(State { + is_set, + waiters: WaitList::new(), + }), + } + } + + /// Signals the oldest 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 the 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(); + } + } + + /// Consumes a stored signal without waiting, returning whether one was available. + /// + /// This never takes a signal assigned to another wait or bypasses a queued wait. A `false` + /// result is only a snapshot; use [`wait`](Self::wait) to wait for a future signal. + /// + /// ``` + /// use asyncband::event::AutoResetEvent; + /// + /// let event = AutoResetEvent::with_state(true); + /// assert!(event.try_wait()); + /// assert!(!event.try_wait()); + /// ``` + 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 joins the FIFO waiter queue. + /// Merely creating this future neither reserves a signal nor establishes a queue position. + /// 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 the oldest queued wait or stored for a future + /// wait, coalescing with any signal already stored. Retrying a cancelled wait joins the back + /// of the queue. Dropping a completed wait does not return its consumed signal. + pub async fn wait(&self) { + Wait { + event: self, + waiter: None, + } + .await + } + + /// Waits for and consumes one signal without borrowing the event. + /// + /// The future owns the [`Arc`], making it suitable for spawned tasks. Its registration, + /// fairness, and cancellation semantics match [`wait`](Self::wait). + /// + /// ``` + /// # #[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, +} + +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/mod.rs b/asyncband/src/event/mod.rs index 1afa051..d658900 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -15,20 +15,40 @@ // 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`] releases all registered waits and remains ready until explicitly reset. +//! An [`AutoResetEvent`] releases one wait per assigned signal and consumes that signal when the +//! wait completes. With no queued waits, it retains at most one signal, coalescing further sets. +//! +//! 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. +//! +//! # Manual-reset events //! //! 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. +//! 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. //! +//! # Auto-reset events +//! +//! An auto-reset event is useful 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 on an auto-reset event compete for signals. This does not broadcast a predicate +//! change to every observer, and the simple check-then-wait loop is not a general multi-consumer +//! queue protocol: several changes can coalesce before those consumers register their waits. +//! //! # Examples //! //! ``` @@ -65,6 +85,10 @@ use crate::internal::waitlist::WaiterId; use crate::internal::wake_all; use crate::internal::waker_batch::WakerBatch; +mod auto_reset; + +pub use self::auto_reset::AutoResetEvent; + /// A reusable event that remains set until explicitly reset. /// /// See the [module-level documentation](self) for its waiting semantics. 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/Cargo.toml b/examples/Cargo.toml index 47a82de..31663a6 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -28,6 +28,7 @@ release = false [dependencies] asyncband = { workspace = true, features = [ "completion", + "event", "lazy-cell", "once-cell", "phaser", @@ -44,6 +45,10 @@ tokio = { workspace = true, features = [ [lints] workspace = true +[[example]] +name = "coalesced_worker" +path = "src/coalesced_worker.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..4e3d34c --- /dev/null +++ b/examples/src/coalesced_worker.rs @@ -0,0 +1,77 @@ +// 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. + +//! Run with `cargo run --package examples --example coalesced_worker`. +//! +//! 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); + rebuilder.stop(); + }); +} 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..7045be1 --- /dev/null +++ b/tests-integration/tests/auto_reset_event_test.rs @@ -0,0 +1,308 @@ +// 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(); + let mut first = pin!(event.wait()); + let mut second = pin!(event.wait()); + + event.set(); + event.set(); + assert!(poll_once(second.as_mut()).is_ready()); + 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 assigned_signals_are_fifo_and_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(), 1); + assert_eq!(second_wake.count(), 0); + assert!(!event.try_wait()); + assert!(poll_once(second.as_mut()).is_pending()); + + // A second set serves the next waiter even though the first has not been polled again. + event.set(); + 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()); + assert!(poll_once(second.as_mut()).is_pending()); + assert!(poll_once(third.as_mut()).is_pending()); + + event.set(); + 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_preserves_fifo_without_adding_a_signal() { + let event = AutoResetEvent::new(); + let mut first = Box::pin(event.wait()); + let mut cancelled = Box::pin(event.wait()); + let mut last = Box::pin(event.wait()); + assert!(poll_once(first.as_mut()).is_pending()); + assert!(poll_once(cancelled.as_mut()).is_pending()); + assert!(poll_once(last.as_mut()).is_pending()); + + drop(cancelled); + assert!(!event.try_wait()); + event.set(); + assert!(poll_once(last.as_mut()).is_pending()); + assert!(poll_once(first.as_mut()).is_ready()); + drop(first); + assert!(poll_once(last.as_mut()).is_pending()); + event.set(); + assert!(poll_once(last.as_mut()).is_ready()); +} + +#[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()); + { + let waker = Waker::from(Arc::new(ReentrantWaker(event.clone()))); + assert!( + second + .as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + } + event.set(); + 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!(poll_once(next.as_mut()).is_pending()); + assert!(panic::catch_unwind(AssertUnwindSafe(|| event.set())).is_err()); + assert!(!event.try_wait()); + 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/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::(); From ba303882c7e6adc9649d7da4b29dd37b54bc25e7 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 21:36:57 +0800 Subject: [PATCH 2/8] docs: demonstrate Notify and event semantics --- examples/Cargo.toml | 5 ++ examples/src/notify_vs_event.rs | 111 +++++++++++++++++++++++++ examples/src/once_cell_vs_lazy_cell.rs | 14 ++-- 3 files changed, 123 insertions(+), 7 deletions(-) create mode 100644 examples/src/notify_vs_event.rs diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 31663a6..61f060e 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -33,6 +33,7 @@ asyncband = { workspace = true, features = [ "once-cell", "phaser", "shutdown", + "watch", ] } tokio = { workspace = true, features = [ "macros", @@ -49,6 +50,10 @@ workspace = true 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/notify_vs_event.rs b/examples/src/notify_vs_event.rs new file mode 100644 index 0000000..0fbcf39 --- /dev/null +++ b/examples/src/notify_vs_event.rs @@ -0,0 +1,111 @@ +// 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. + +//! Run with `cargo run --package examples --example notify_vs_event`. +//! +//! Three notification needs: wake one worker, notify existing observers of a change, and keep a +//! readiness gate open. Tokio Notify combines the first two; ManualResetEvent expresses the third. +//! See https://docs.rs/tokio/1.53.1/tokio/sync/struct.Notify.html for Tokio's contracts. + +use std::future::Future; +use std::pin::Pin; +use std::pin::pin; +use std::task::Context; +use std::task::Poll; +use std::task::Waker; + +use asyncband::event::AutoResetEvent; +use asyncband::event::ManualResetEvent; +use asyncband::watch; +use tokio::sync::Notify; + +#[tokio::main(flavor = "current_thread")] +async fn main() { + wake_one_worker().await; + broadcast_vs_readiness().await; + broadcast_change_with_watch().await; +} + +async fn wake_one_worker() { + // A worker rechecks external work after waking. Repeated signals can coalesce while idle. + // See coalesced_worker.rs for a complete worker using this pattern. + let notify = Notify::new(); + notify.notify_one(); + notify.notify_one(); + notify.notified().await; + assert!(poll_once(pin!(notify.notified())).is_pending()); + + let event = AutoResetEvent::new(); + event.set(); + event.set(); + event.wait().await; + assert!(!event.try_wait()); + println!("Wake one: both Notify and AutoResetEvent retain one unused signal"); + + // This comparison assumes one consumer. Tokio also has Notified::enable() for registering + // before checking an external queue; AutoResetEvent registers on first poll only. +} + +async fn broadcast_vs_readiness() { + let notify = Notify::new(); + let first = notify.notified(); + let second = notify.notified(); + notify.notify_waiters(); + // Both futures existed before the broadcast, so even these unpolled futures receive it. + tokio::join!(first, second); + assert!(poll_once(pin!(notify.notified())).is_pending()); // A late waiter misses it. + + let gate = ManualResetEvent::new(); + let mut registered = pin!(gate.wait()); + let mut unpolled = pin!(gate.wait()); + assert!(poll_once(registered.as_mut()).is_pending()); + gate.set(); + gate.reset(); + assert!(poll_once(registered.as_mut()).is_ready()); + assert!(poll_once(unpolled.as_mut()).is_pending()); + // A set/reset pulse misses an unpolled wait, so it cannot replace notify_waiters(). + + gate.set(); + unpolled.await; + gate.wait().await; // A late waiter also passes while the gate remains set. + gate.reset(); + println!("Notify broadcasts once; ManualResetEvent stays ready until reset"); +} + +async fn broadcast_change_with_watch() { + // For "the state changed; all existing observers should recheck", subscribe before checking + // the external state. Each observer has its own progress, so they do not compete for a signal. + let (changes, mut first) = watch::channel(()); + let mut second = changes.subscribe(); + changes.send(()).unwrap(); + let (a, b) = tokio::join!(first.changed(), second.changed()); + a.unwrap(); + b.unwrap(); + + let mut late = changes.subscribe(); + assert!(poll_once(pin!(late.changed())).is_pending()); + println!("Broadcast change: watch<()> notifies each prior subscription; late ones wait"); + + // This is an explicit subscription protocol: a retained receiver remembers unseen changes + // across waits and cancellation. Tokio establishes a new boundary for each notified() future. + // Repeated changes may coalesce. A fresh subscription starts observing from its creation. +} + +// Poll once to show a wait stays pending, without hanging the example or relying on a timeout. +fn poll_once(future: Pin<&mut F>) -> Poll { + future.poll(&mut Context::from_waker(Waker::noop())) +} diff --git a/examples/src/once_cell_vs_lazy_cell.rs b/examples/src/once_cell_vs_lazy_cell.rs index 726a117..165dcd0 100644 --- a/examples/src/once_cell_vs_lazy_cell.rs +++ b/examples/src/once_cell_vs_lazy_cell.rs @@ -20,9 +20,9 @@ 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 +81,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 +98,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 +116,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); } From 016f183c2ba54c4d9a0e6ae433f7da00a029f68e Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 21:42:23 +0800 Subject: [PATCH 3/8] docs: clarify example documentation style --- examples/AGENTS.md | 25 +++++++++++++++++++++++ examples/src/coalesced_worker.rs | 2 -- examples/src/graceful_shutdown.rs | 3 +++ examples/src/lazy_cell_boxed_vs_inline.rs | 3 +++ examples/src/notify_vs_event.rs | 2 -- examples/src/once_cell_vs_lazy_cell.rs | 4 ++++ 6 files changed, 35 insertions(+), 4 deletions(-) create mode 100644 examples/AGENTS.md diff --git a/examples/AGENTS.md b/examples/AGENTS.md new file mode 100644 index 0000000..a75f58d --- /dev/null +++ b/examples/AGENTS.md @@ -0,0 +1,25 @@ + + +# 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. +- Keep example explanations in the example source. The repository README only needs a concise entry point. diff --git a/examples/src/coalesced_worker.rs b/examples/src/coalesced_worker.rs index 4e3d34c..37fff0b 100644 --- a/examples/src/coalesced_worker.rs +++ b/examples/src/coalesced_worker.rs @@ -15,8 +15,6 @@ // specific language governing permissions and limitations // under the License. -//! Run with `cargo run --package examples --example coalesced_worker`. -//! //! 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. 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 index 0fbcf39..da9356e 100644 --- a/examples/src/notify_vs_event.rs +++ b/examples/src/notify_vs_event.rs @@ -15,8 +15,6 @@ // specific language governing permissions and limitations // under the License. -//! Run with `cargo run --package examples --example notify_vs_event`. -//! //! Three notification needs: wake one worker, notify existing observers of a change, and keep a //! readiness gate open. Tokio Notify combines the first two; ManualResetEvent expresses the third. //! See https://docs.rs/tokio/1.53.1/tokio/sync/struct.Notify.html for Tokio's contracts. diff --git a/examples/src/once_cell_vs_lazy_cell.rs b/examples/src/once_cell_vs_lazy_cell.rs index 165dcd0..caad33a 100644 --- a/examples/src/once_cell_vs_lazy_cell.rs +++ b/examples/src/once_cell_vs_lazy_cell.rs @@ -15,6 +15,10 @@ // 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; From 263da2c7dd850e402a8dbcdaa05d388f7c3aa7ed Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 21:52:25 +0800 Subject: [PATCH 4/8] docs: demonstrate application migrations from Tokio Notify --- examples/AGENTS.md | 1 + examples/src/notify_vs_event.rs | 220 ++++++++++++++++++++++---------- 2 files changed, 151 insertions(+), 70 deletions(-) diff --git a/examples/AGENTS.md b/examples/AGENTS.md index a75f58d..fc6cb48 100644 --- a/examples/AGENTS.md +++ b/examples/AGENTS.md @@ -22,4 +22,5 @@ under the License. - 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/src/notify_vs_event.rs b/examples/src/notify_vs_event.rs index da9356e..373c128 100644 --- a/examples/src/notify_vs_event.rs +++ b/examples/src/notify_vs_event.rs @@ -15,95 +15,175 @@ // specific language governing permissions and limitations // under the License. -//! Three notification needs: wake one worker, notify existing observers of a change, and keep a -//! readiness gate open. Tokio Notify combines the first two; ManualResetEvent expresses the third. -//! See https://docs.rs/tokio/1.53.1/tokio/sync/struct.Notify.html for Tokio's contracts. +//! 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::future::Future; -use std::pin::Pin; -use std::pin::pin; -use std::task::Context; -use std::task::Poll; -use std::task::Waker; +use std::collections::VecDeque; +use std::sync::Mutex; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; use asyncband::event::AutoResetEvent; -use asyncband::event::ManualResetEvent; use asyncband::watch; use tokio::sync::Notify; #[tokio::main(flavor = "current_thread")] async fn main() { - wake_one_worker().await; - broadcast_vs_readiness().await; - broadcast_change_with_watch().await; + cache_worker_with_notify().await; + cache_worker_with_event().await; + index_readers_with_notify().await; + index_readers_with_watch().await; } -async fn wake_one_worker() { - // A worker rechecks external work after waking. Repeated signals can coalesce while idle. - // See coalesced_worker.rs for a complete worker using this pattern. - let notify = Notify::new(); - notify.notify_one(); - notify.notify_one(); - notify.notified().await; - assert!(poll_once(pin!(notify.notified())).is_pending()); +// 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 event = AutoResetEvent::new(); - event.set(); - event.set(); - event.wait().await; - assert!(!event.try_wait()); - println!("Wake one: both Notify and AutoResetEvent retain one unused signal"); - - // This comparison assumes one consumer. Tokio also has Notified::enable() for registering - // before checking an external queue; AutoResetEvent registers on first poll only. + let (flushed, ()) = tokio::join!(biased; worker, enqueue); + assert_eq!(flushed, ["alice", "bob", "carol"]); + println!("Notify worker flushed {flushed:?}"); } -async fn broadcast_vs_readiness() { - let notify = Notify::new(); - let first = notify.notified(); - let second = notify.notified(); - notify.notify_waiters(); - // Both futures existed before the broadcast, so even these unpolled futures receive it. - tokio::join!(first, second); - assert!(poll_once(pin!(notify.notified())).is_pending()); // A late waiter misses it. +// 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:?}"); +} - let gate = ManualResetEvent::new(); - let mut registered = pin!(gate.wait()); - let mut unpolled = pin!(gate.wait()); - assert!(poll_once(registered.as_mut()).is_pending()); - gate.set(); - gate.reset(); - assert!(poll_once(registered.as_mut()).is_ready()); - assert!(poll_once(unpolled.as_mut()).is_pending()); - // A set/reset pulse misses an unpolled wait, so it cannot replace notify_waiters(). +// 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; + } + }; - gate.set(); - unpolled.await; - gate.wait().await; // A late waiter also passes while the gate remains set. - gate.reset(); - println!("Notify broadcasts once; ManualResetEvent stays ready until reset"); + 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 broadcast_change_with_watch() { - // For "the state changed; all existing observers should recheck", subscribe before checking - // the external state. Each observer has its own progress, so they do not compete for a signal. - let (changes, mut first) = watch::channel(()); - let mut second = changes.subscribe(); - changes.send(()).unwrap(); - let (a, b) = tokio::join!(first.changed(), second.changed()); - a.unwrap(); - b.unwrap(); +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; + } +} - let mut late = changes.subscribe(); - assert!(poll_once(pin!(late.changed())).is_pending()); - println!("Broadcast change: watch<()> notifies each prior subscription; late ones wait"); +// 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. +// A retained receiver remembers unseen changes across cancelled waits. That suits a revision +// predicate; preserving Notify's per-wait broadcast boundary instead requires fresh subscriptions. +// 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; + } + }; - // This is an explicit subscription protocol: a retained receiver remembers unseen changes - // across waits and cancellation. Tokio establishes a new boundary for each notified() future. - // Repeated changes may coalesce. A fresh subscription starts observing from its creation. + 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}"); } -// Poll once to show a wait stays pending, without hanging the example or relying on a timeout. -fn poll_once(future: Pin<&mut F>) -> Poll { - future.poll(&mut Context::from_waker(Waker::noop())) +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(); + } } From 08649e3de1cd0aab85d427a63fc7b664e766b6a5 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 22:54:42 +0800 Subject: [PATCH 5/8] docs: remove event wake ordering promises --- asyncband/src/event/auto_reset.rs | 22 ++++----- asyncband/src/event/mod.rs | 2 +- .../tests/auto_reset_event_test.rs | 46 ++++++++++--------- 3 files changed, 37 insertions(+), 33 deletions(-) diff --git a/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs index 57c451d..f1ae621 100644 --- a/asyncband/src/event/auto_reset.rs +++ b/asyncband/src/event/auto_reset.rs @@ -30,7 +30,7 @@ use crate::internal::waitlist::WaiterId; /// A reusable signal that releases one waiter and resets automatically. /// -/// Each [`set`](Self::set) assigns a signal to the oldest registered wait, or stores one signal +/// 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. @@ -82,7 +82,7 @@ impl AutoResetEvent { } } - /// Signals the oldest registered wait, or stores one signal if no wait is queued. + /// 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 @@ -101,8 +101,8 @@ impl AutoResetEvent { /// Consumes a stored signal without waiting, returning whether one was available. /// - /// This never takes a signal assigned to another wait or bypasses a queued wait. A `false` - /// result is only a snapshot; use [`wait`](Self::wait) to wait for a future signal. + /// 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. /// /// ``` /// use asyncband::event::AutoResetEvent; @@ -117,16 +117,16 @@ impl AutoResetEvent { /// Waits for and consumes one signal. /// - /// The first poll consumes a stored signal immediately, or joins the FIFO waiter queue. - /// Merely creating this future neither reserves a signal nor establishes a queue position. + /// 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 the oldest queued wait or stored for a future - /// wait, coalescing with any signal already stored. Retrying a cancelled wait joins the back - /// of the queue. Dropping a completed wait does not return its consumed signal. + /// 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, @@ -137,8 +137,8 @@ impl AutoResetEvent { /// Waits for and consumes one signal without borrowing the event. /// - /// The future owns the [`Arc`], making it suitable for spawned tasks. Its registration, - /// fairness, and cancellation semantics match [`wait`](Self::wait). + /// The future owns the [`Arc`], making it suitable for spawned tasks. Its waiting and + /// cancellation semantics match [`wait`](Self::wait). /// /// ``` /// # #[tokio::main] diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index d658900..4626c85 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -164,7 +164,7 @@ impl ManualResetEvent { /// 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. + /// already set has no effect. /// /// # Panics /// diff --git a/tests-integration/tests/auto_reset_event_test.rs b/tests-integration/tests/auto_reset_event_test.rs index 7045be1..e8b599f 100644 --- a/tests-integration/tests/auto_reset_event_test.rs +++ b/tests-integration/tests/auto_reset_event_test.rs @@ -53,7 +53,7 @@ fn unpolled_waits_do_not_reserve_stored_signals() { } #[test] -fn assigned_signals_are_fifo_and_cannot_be_stolen() { +fn assigned_signals_cannot_be_stolen() { let event = AutoResetEvent::new(); let first_wake = Arc::new(WakeCounter::default()); let second_wake = Arc::new(WakeCounter::default()); @@ -75,13 +75,23 @@ fn assigned_signals_are_fifo_and_cannot_be_stolen() { ); event.set(); - assert_eq!(first_wake.count(), 1); - assert_eq!(second_wake.count(), 0); + assert_eq!(first_wake.count() + second_wake.count(), 1); assert!(!event.try_wait()); - assert!(poll_once(second.as_mut()).is_pending()); + 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() + ); - // A second set serves the next waiter even though the first has not been polled again. + // 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()); @@ -99,10 +109,8 @@ fn cancelling_selected_waits_transfers_then_restores_the_signal() { let mut second = Box::pin(event.wait()); let mut third = Box::pin(event.wait()); assert!(poll_once(first.as_mut()).is_pending()); - assert!(poll_once(second.as_mut()).is_pending()); - assert!(poll_once(third.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()); @@ -114,24 +122,20 @@ fn cancelling_selected_waits_transfers_then_restores_the_signal() { } #[test] -fn cancelling_unselected_waits_preserves_fifo_without_adding_a_signal() { +fn cancelling_unselected_waits_does_not_add_a_signal() { let event = AutoResetEvent::new(); - let mut first = Box::pin(event.wait()); let mut cancelled = Box::pin(event.wait()); - let mut last = Box::pin(event.wait()); - assert!(poll_once(first.as_mut()).is_pending()); + let mut remaining = Box::pin(event.wait()); assert!(poll_once(cancelled.as_mut()).is_pending()); - assert!(poll_once(last.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(last.as_mut()).is_pending()); - assert!(poll_once(first.as_mut()).is_ready()); - drop(first); - assert!(poll_once(last.as_mut()).is_pending()); - event.set(); - assert!(poll_once(last.as_mut()).is_ready()); + assert!(poll_once(remaining.as_mut()).is_ready()); + drop(remaining); + assert!(!event.try_wait()); } #[test] @@ -221,6 +225,7 @@ fn wake_and_waker_destruction_happen_outside_the_event_lock() { 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!( @@ -230,7 +235,6 @@ fn wake_and_waker_destruction_happen_outside_the_event_lock() { .is_pending() ); } - event.set(); drop(first); // The cancellation handoff wakes the second waiter. assert!(poll_once(second.as_mut()).is_ready()); @@ -272,9 +276,9 @@ fn a_panicking_wake_leaves_its_signal_available_for_cancellation_handoff() { .poll(&mut Context::from_waker(&waker)) .is_pending() ); - assert!(poll_once(next.as_mut()).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()); From 9eb5fbb491f7361faddce9e233e772a8ac3b0b1e Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 22:58:44 +0800 Subject: [PATCH 6/8] feat(event): add AutoResetEvent reset --- CHANGELOG.md | 2 +- asyncband/src/event/auto_reset.rs | 9 +++++ .../tests/auto_reset_event_test.rs | 40 +++++++++++++++++++ 3 files changed, 50 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 19263c0..213eb3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,7 +25,7 @@ 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, and transfers assigned signals when waits are cancelled. +* 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. * 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/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs index f1ae621..b7e8c4f 100644 --- a/asyncband/src/event/auto_reset.rs +++ b/asyncband/src/event/auto_reset.rs @@ -34,6 +34,7 @@ use crate::internal::waitlist::WaiterId; /// 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 @@ -99,6 +100,14 @@ impl AutoResetEvent { } } + /// 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; + } + /// Consumes a stored signal without waiting, returning whether one was available. /// /// This never takes a signal assigned to another wait. A `false` result is only a snapshot; diff --git a/tests-integration/tests/auto_reset_event_test.rs b/tests-integration/tests/auto_reset_event_test.rs index e8b599f..a0381d7 100644 --- a/tests-integration/tests/auto_reset_event_test.rs +++ b/tests-integration/tests/auto_reset_event_test.rs @@ -52,6 +52,46 @@ fn unpolled_waits_do_not_reserve_stored_signals() { 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(); + + let mut unpolled = pin!(event.wait()); + event.set(); // One signal is assigned and another is stored. + event.reset(); + 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.try_wait()); + + // The transferred signal also survives reset and can be restored by cancellation. + event.reset(); + drop(remaining); + assert!(event.try_wait()); + assert!(!event.try_wait()); +} + #[test] fn assigned_signals_cannot_be_stolen() { let event = AutoResetEvent::new(); From 29b2692fc2e6db9f89f15a86139afbc30ed6883c Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 23:03:45 +0800 Subject: [PATCH 7/8] refactor(event): align event modules and documentation --- asyncband/src/event/auto_reset.rs | 35 ++- asyncband/src/event/manual_reset.rs | 320 +++++++++++++++++++++ asyncband/src/event/mod.rs | 431 +--------------------------- 3 files changed, 352 insertions(+), 434 deletions(-) create mode 100644 asyncband/src/event/manual_reset.rs diff --git a/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs index b7e8c4f..173a2e7 100644 --- a/asyncband/src/event/auto_reset.rs +++ b/asyncband/src/event/auto_reset.rs @@ -40,13 +40,25 @@ use crate::internal::waitlist::WaiterId; /// [`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: when it -/// is used to notify changes to an external predicate, callers must synchronize access to that -/// predicate separately. +/// signal and signals passed on after cancellation. +/// +/// The event carries no application state: callers must synchronize access to external predicates +/// separately. /// /// # Examples /// @@ -68,12 +80,14 @@ pub struct AutoResetEvent { } impl AutoResetEvent { - /// Creates an event without a stored signal. + /// Creates an unset event. pub const fn new() -> Self { Self::with_state(false) } - /// Creates an event with one stored signal if `is_set` is `true`, or none otherwise. + /// 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 { @@ -91,7 +105,7 @@ impl AutoResetEvent { /// /// # Panics /// - /// Panics if waking the selected task panics. Its signal remains assigned and can still be + /// 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(); @@ -113,6 +127,8 @@ impl AutoResetEvent { /// 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; /// @@ -144,11 +160,13 @@ impl AutoResetEvent { .await } - /// Waits for and consumes one signal without borrowing the event. + /// 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() { @@ -256,6 +274,7 @@ enum Waiter { Notified, } +#[must_use = "futures do nothing unless you `.await` or poll them"] struct Wait<'a> { event: &'a AutoResetEvent, waiter: Option, @@ -264,7 +283,7 @@ struct Wait<'a> { impl Future for Wait<'_> { type Output = (); - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let this = self.get_mut(); this.event.poll_wait(&mut this.waiter, cx) } diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs new file mode 100644 index 0000000..43ca40b --- /dev/null +++ b/asyncband/src/event/manual_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; +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 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. +/// +/// 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 reserve or consume the set state. + /// + /// # 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 + } + + /// 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 4626c85..a67859f 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -17,437 +17,16 @@ //! Reusable signals for coordinating tasks without carrying a value. //! -//! A [`ManualResetEvent`] releases all registered waits and remains ready until explicitly reset. -//! An [`AutoResetEvent`] releases one wait per assigned signal and consumes that signal when the -//! wait completes. With no queued waits, it retains at most one signal, coalescing further sets. +//! 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. //! //! 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. -//! -//! # Manual-reset events -//! -//! 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. -//! -//! 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. -//! -//! # Auto-reset events -//! -//! An auto-reset event is useful 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 on an auto-reset event compete for signals. This does not broadcast a predicate -//! change to every observer, and the simple check-then-wait loop is not a general multi-consumer -//! queue protocol: several changes can coalesce before those consumers register their waits. -//! -//! # 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; mod auto_reset; +mod manual_reset; pub use self::auto_reset::AutoResetEvent; - -/// 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. - /// - /// # 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 = (); - - 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); - } -} +pub use self::manual_reset::ManualResetEvent; From 1f973c4d143f43130ebfbdee16faee0df5a386b2 Mon Sep 17 00:00:00 2001 From: tison Date: Mon, 14 Sep 2026 23:19:26 +0800 Subject: [PATCH 8/8] feat(event): align state queries and immediate waits --- CHANGELOG.md | 1 + asyncband/src/event/auto_reset.rs | 30 ++++++++++++++++--- asyncband/src/event/manual_reset.rs | 29 ++++++++++++++---- asyncband/src/event/mod.rs | 3 ++ examples/src/coalesced_worker.rs | 4 +++ examples/src/notify_vs_event.rs | 4 +-- .../tests/auto_reset_event_test.rs | 9 ++++++ tests-integration/tests/event_test.rs | 7 +++++ 8 files changed, 76 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 213eb3e..9b4aa46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ 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/asyncband/src/event/auto_reset.rs b/asyncband/src/event/auto_reset.rs index 173a2e7..af72eae 100644 --- a/asyncband/src/event/auto_reset.rs +++ b/asyncband/src/event/auto_reset.rs @@ -122,10 +122,32 @@ impl AutoResetEvent { self.state.lock().is_set = false; } - /// Consumes a stored signal without waiting, returning whether one was available. + /// Returns whether the event is currently set. /// - /// 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. + /// 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 /// @@ -134,7 +156,7 @@ impl AutoResetEvent { /// /// let event = AutoResetEvent::with_state(true); /// assert!(event.try_wait()); - /// 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) diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs index 43ca40b..d51b16a 100644 --- a/asyncband/src/event/manual_reset.rs +++ b/asyncband/src/event/manual_reset.rs @@ -45,12 +45,12 @@ use crate::internal::waker_batch::WakerBatch; /// /// # 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. +/// 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. -/// [`is_set`](Self::is_set) is only a snapshot and cannot replace a wait or support check-then-act. /// /// The event carries no application state: callers must synchronize access to external predicates /// separately. @@ -137,7 +137,8 @@ impl ManualResetEvent { /// Returns whether the event is currently set. /// - /// This is a snapshot only; it does not reserve or consume the set 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 /// @@ -152,6 +153,24 @@ impl ManualResetEvent { 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. diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index a67859f..8f94320 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -21,6 +21,9 @@ //! it retains at most one signal, coalescing further sets. A [`ManualResetEvent`] releases all //! registered waits and remains set until explicitly reset. //! +//! 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. +//! //! 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. diff --git a/examples/src/coalesced_worker.rs b/examples/src/coalesced_worker.rs index 37fff0b..52a1cd3 100644 --- a/examples/src/coalesced_worker.rs +++ b/examples/src/coalesced_worker.rs @@ -70,6 +70,10 @@ async fn main() { 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/notify_vs_event.rs b/examples/src/notify_vs_event.rs index 373c128..84cb529 100644 --- a/examples/src/notify_vs_event.rs +++ b/examples/src/notify_vs_event.rs @@ -148,8 +148,8 @@ async fn wait_for_index_with_notify( // 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. -// A retained receiver remembers unseen changes across cancelled waits. That suits a revision -// predicate; preserving Notify's per-wait broadcast boundary instead requires fresh subscriptions. +// 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); diff --git a/tests-integration/tests/auto_reset_event_test.rs b/tests-integration/tests/auto_reset_event_test.rs index a0381d7..dc0e66b 100644 --- a/tests-integration/tests/auto_reset_event_test.rs +++ b/tests-integration/tests/auto_reset_event_test.rs @@ -38,12 +38,15 @@ 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()); @@ -58,10 +61,13 @@ fn reset_discards_only_unassigned_signals() { 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()); @@ -83,12 +89,15 @@ fn reset_preserves_cancellation_handoff() { 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()); } 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());