diff --git a/CHANGELOG.md b/CHANGELOG.md index ea2f1e3c..0fc85334 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file. ### New features +* 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`. * Add bounded MPSC `reserve` and `try_reserve` methods returning a `Permit`, allowing callers to wait for capacity before constructing a message; pending sends and reservations receive capacity in wait-queue order, and unused permits release capacity without claiming message order. ### Bug fixes diff --git a/README.md b/README.md index 1203ae01..47bb7629 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon | | [`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. | | | [`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. | | | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Request shutdown and wait until all completion guards are dropped. | | Work coalescing | [`Once`](https://docs.rs/asyncband/*/asyncband/once/struct.Once.html) | `once` | Complete one asynchronous initialization; cancelled or panicked attempts may be retried. | diff --git a/asyncband/Cargo.toml b/asyncband/Cargo.toml index d2d74f45..19c97562 100644 --- a/asyncband/Cargo.toml +++ b/asyncband/Cargo.toml @@ -58,6 +58,7 @@ once = ["semaphore"] once-cell = ["semaphore"] once-map = ["dep:hashbrown", "once-cell"] oneshot = [] +phaser = [] pool = ["semaphore"] rwlock = [] semaphore = [] diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 0252aa83..29d3a072 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -57,6 +57,7 @@ pub(crate) fn wake_all(mut wakers: impl Iterator) { feature = "latch", feature = "mpsc", feature = "mutex", + feature = "phaser", feature = "rwlock", feature = "semaphore", feature = "waitgroup", @@ -84,6 +85,7 @@ pub(crate) mod value_cell; feature = "latch", feature = "mpsc", feature = "mutex", + feature = "phaser", feature = "rwlock", feature = "semaphore", feature = "waitgroup", @@ -120,6 +122,7 @@ pub(crate) mod waitlist; feature = "mpsc", feature = "mutex", feature = "once", + feature = "phaser", feature = "rwlock", feature = "semaphore", feature = "waitgroup", @@ -136,6 +139,7 @@ pub(crate) mod waker_batch; feature = "completion", feature = "latch", feature = "once", + feature = "phaser", feature = "waitgroup", feature = "watch", ))] diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs index 93ce485e..cb645cee 100644 --- a/asyncband/src/lib.rs +++ b/asyncband/src/lib.rs @@ -63,6 +63,7 @@ //! | | [`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. | //! | | [`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. | //! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Request shutdown and wait until all completion guards are dropped. | //! | Work coalescing | [`Once`](once::Once) | `once` | Complete one asynchronous initialization; cancelled or panicked attempts may be retried. | @@ -145,6 +146,8 @@ pub mod mutex; pub mod once; #[cfg(feature = "oneshot")] pub mod oneshot; +#[cfg(feature = "phaser")] +pub mod phaser; #[cfg(feature = "pool")] pub mod pool; #[cfg(feature = "rwlock")] @@ -160,5 +163,8 @@ pub mod waitgroup; #[cfg(feature = "watch")] pub mod watch; -#[cfg(all(test, any(feature = "once-map", feature = "singleflight")))] +#[cfg(all( + test, + any(feature = "once-map", feature = "phaser", feature = "singleflight") +))] mod test_support; diff --git a/asyncband/src/mutex/mod.rs b/asyncband/src/mutex/mod.rs index 97772eea..ed049641 100644 --- a/asyncband/src/mutex/mod.rs +++ b/asyncband/src/mutex/mod.rs @@ -19,9 +19,9 @@ // Copyright (c) Tokio Contributors // The Tokio-derived portions remain licensed under the MIT License. // Asyncband independently built the mutex on its own semaphore and substantially changed the -// incorporated guard implementation: the try-lock error type and semaphore closure are absent, -// projected guards use NonNull pointers with explicit invariance, and projected guards can be -// mapped repeatedly in both borrowed and owned forms. +// incorporated guard implementation: the try-lock error type and support for closing the semaphore +// are absent, projected guards use NonNull pointers with explicit invariance, and projected guards +// can be mapped repeatedly in both borrowed and owned forms. // Upstream source: // https://github.com/tokio-rs/tokio/blob/01e04daaa162ce6122bb894fdda0b6803dd32093/tokio/src/sync/mutex.rs diff --git a/asyncband/src/phaser/mod.rs b/asyncband/src/phaser/mod.rs new file mode 100644 index 00000000..ae7fbe12 --- /dev/null +++ b/asyncband/src/phaser/mod.rs @@ -0,0 +1,552 @@ +// 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. + +//! Coordinate repeated rounds of work with a dynamic participant set. +//! +//! A [`Phaser`] is a shared coordination handle. Cloning it creates another observer, without +//! registering a participant. Each [`PhaserParticipant`] owns one arrival obligation per phase. +//! Register participants before starting their tasks, or keep a coordinator participant registered +//! while setting up a group so that the first workers cannot finish the phase prematurely. +//! +//! # Example: build a shared dictionary before encoding documents +//! +//! An indexing job needs the same numeric ID for a word in every document. Workers first collect +//! vocabulary in parallel. Once all documents have contributed their words, the coordinator assigns +//! IDs in sorted order. A second rendezvous keeps workers from encoding documents before that +//! shared dictionary is ready. The same participants coordinate both steps. +//! +//! ``` +//! use std::collections::BTreeMap; +//! use std::sync::Arc; +//! use std::sync::Mutex; +//! +//! use asyncband::phaser::Closed; +//! use asyncband::phaser::Phaser; +//! +//! # #[tokio::main(flavor = "current_thread")] +//! # async fn main() -> Result<(), Closed> { +//! let documents = ["rust async rust", "async tasks"]; +//! let dictionary = Arc::new(Mutex::new(BTreeMap::new())); +//! let phaser = Phaser::new(); +//! let mut coordinator = phaser.register_one()?; +//! let participants = phaser.register(documents.len())?; +//! let mut tasks = Vec::new(); +//! +//! for (document, mut participant) in documents.into_iter().zip(participants) { +//! let dictionary = dictionary.clone(); +//! tasks.push(tokio::spawn(async move { +//! let words: Vec<_> = document.split_whitespace().collect(); +//! { +//! let mut dictionary = dictionary.lock().unwrap(); +//! for &word in &words { +//! dictionary.entry(word).or_insert(0); +//! } +//! } +//! participant.wait().await?; // All vocabulary has been collected. +//! participant.wait().await?; // The coordinator has assigned the IDs. +//! +//! let dictionary = dictionary.lock().unwrap(); +//! let encoded: Vec<_> = words.iter().map(|word| dictionary[word]).collect(); +//! Ok::<_, Closed>(encoded) +//! })); +//! } +//! +//! coordinator.wait().await?; +//! for (id, value) in dictionary.lock().unwrap().values_mut().enumerate() { +//! *value = id; +//! } +//! coordinator.wait().await?; +//! +//! let mut encoded_documents = Vec::new(); +//! for task in tasks { +//! encoded_documents.push(task.await.unwrap()?); +//! } +//! // Every document uses the same dictionary: async = 0, rust = 1, tasks = 2. +//! assert_eq!(encoded_documents, [vec![1, 0, 1], vec![0, 2]]); +//! # Ok(()) +//! # } +//! ``` +//! +//! # Arriving and waiting +//! +//! [`PhaserParticipant::wait`] arrives and waits for the other participants. To overlap independent +//! work with that wait, call [`arrive`](PhaserParticipant::arrive) first. The subsequent `wait` +//! observes that arrival's phase even if it has already completed. Explicitly arriving again +//! replaces the pending observation with the current phase; repeated arrivals within one phase +//! do not count twice. +//! +//! # Membership and cancellation +//! +//! Registration joins the phase current at the registration's synchronization point. In +//! particular, [`register`](Phaser::register) registers its entire batch in one phase. +//! Dropping or [`deregistering`](PhaserParticipant::deregister) a participant removes its future +//! obligations and discharges any outstanding arrival in the current phase. An empty phaser is +//! dormant and can be reused; it does not advance repeatedly or close automatically. +//! +//! A participant's `wait` records arrival on its first poll. Dropping an unpolled future has no +//! effect. Cancelling a polled wait preserves its arrival and pending phase: retrying `wait` on +//! that participant observes the same phase instead of arriving in a later one. Dropping the +//! participant itself withdraws it from the group. Withdrawal does not certify successful work; +//! applications that require all workers to succeed should close the phaser on failure. +//! +//! [`Phaser::wait`] is an independent, cancel-safe observation. It never registers a +//! participant or records an arrival. Observers may miss intermediate phases; this is not an +//! event stream with one notification per phase. +//! +//! # Closing and synchronization +//! +//! [`Phaser::close`] permanently freezes the current phase, rejects registration and arrival, +//! and releases waits for the unfinished phase with [`Closed`]. A previously completed phase +//! remains successful even if the phaser closes before its waiter is polled again. +//! +//! Work performed before an arrival or deregistration happens before work performed after a +//! successful wait for that phase's completion. No such all-participants-completed guarantee is +//! provided by a wait that returns `Closed`. +//! +//! Phase numbers start at zero and wrap from `u64::MAX` to zero. Pass a value previously obtained +//! from the same phaser to `wait`; it tests for a different phase, not a target number +//! or numeric threshold. An observation must not be retained across a full counter cycle. + +use std::fmt; +use std::future::Future; +use std::iter::FusedIterator; +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::wake_all; +use crate::internal::wakerset::WakerSet; +use crate::internal::wakerset::WakerToken; + +#[cfg(test)] +mod tests; + +/// The phaser was closed before this operation could complete. +/// +/// This error is returned by phaser operations and cannot be constructed directly by callers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Closed(()); + +impl fmt::Display for Closed { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("Phaser is closed") + } +} + +impl std::error::Error for Closed {} + +/// A shared handle to a reusable phase barrier with dynamic participants. +/// +/// Cloning this handle does not register a participant. Use [`register_one`](Self::register_one) to +/// create a participant that can be moved into an independently spawned task. +/// +/// See the [module level documentation](self) for usage examples and synchronization semantics. +#[derive(Clone)] +pub struct Phaser { + state: Arc>, +} + +struct State { + phase: u64, + closed: bool, + registered: usize, + unarrived: usize, + waiters: WakerSet, +} + +impl State { + fn advance_if_ready(&mut self) -> Option + 'static> { + if self.closed || self.unarrived != 0 { + return None; + } + self.phase = self.phase.wrapping_add(1); + self.unarrived = self.registered; + Some(self.waiters.drain()) + } + + fn completion(&self, observed: u64) -> Poll> { + // Completion wins over a later close; the unfinished phase itself never advances on close. + if self.phase != observed { + Poll::Ready(Ok(self.phase)) + } else if self.closed { + Poll::Ready(Err(Closed(()))) + } else { + Poll::Pending + } + } +} + +impl fmt::Debug for Phaser { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let state = self.state.lock(); + f.debug_struct("Phaser") + .field("phase", &state.phase) + .field("closed", &state.closed) + .field("registered", &state.registered) + .field("unarrived", &state.unarrived) + .finish_non_exhaustive() + } +} + +impl Default for Phaser { + fn default() -> Self { + Self::new() + } +} + +impl Phaser { + /// Creates an open, dormant phaser at phase zero with no registered participants. + pub fn new() -> Self { + Self { + state: Arc::new(Mutex::new(State { + phase: 0, + closed: false, + registered: 0, + unarrived: 0, + waiters: WakerSet::new(), + })), + } + } + + /// Returns the current phase number, which remains fixed once the phaser is closed. + pub fn phase(&self) -> u64 { + self.state.lock().phase + } + + /// Returns whether this phaser has been permanently closed. + pub fn is_closed(&self) -> bool { + self.state.lock().closed + } + + /// Closes this phaser and wakes all pending observers without completing the current phase. + /// + /// This operation is idempotent and affects every handle and participant. Existing participants + /// may still deregister or be dropped; their removal no longer advances the phase. + /// + /// # Panics + /// + /// If a waker panics, the phaser remains closed and notification is attempted for the other + /// waiters before the panic resumes. + pub fn close(&self) { + let wakers = { + let mut state = self.state.lock(); + if state.closed { + return; + } + state.closed = true; + state.waiters.take_all() + }; + wake_all(wakers); + } + + /// Returns an instantaneous count of registered participants, including those already arrived. + /// + /// Counts can change between separate queries. This is not a synchronization operation. + pub fn registered_parties(&self) -> usize { + self.state.lock().registered + } + + /// Returns an instantaneous count of participants that have arrived in the current phase. + pub fn arrived_parties(&self) -> usize { + let state = self.state.lock(); + state.registered - state.unarrived + } + + /// Returns an instantaneous count of outstanding arrivals in the current phase. + pub fn unarrived_parties(&self) -> usize { + self.state.lock().unarrived + } + + /// Registers one participant in the current phase, or returns [`Closed`]. + /// + /// Registration racing with advancement joins the phase before or after that advancement. + /// The returned participant owns a shared handle and does not borrow this one. + /// + /// # Panics + /// + /// Panics if the registered count would exceed `usize::MAX`. + pub fn register_one(&self) -> Result { + let phaser = self.clone(); + self.do_register(1)?; + Ok(PhaserParticipant::new(phaser)) + } + + /// Registers an entire batch in one phase and returns an iterator over its participants. + /// + /// Registration is immediate, including participants not yet yielded by the iterator. No + /// storage is allocated for the batch. Dropping the iterator withdraws its remaining + /// participants; yielded handles keep their registrations. The iterator owns a shared handle + /// and does not borrow this phaser. + /// + /// Returns [`Closed`] without registering anyone if the phaser is closed. On an open phaser, + /// a zero-sized batch does nothing. Collect this iterator into a collection of your choice. + /// + /// # Panics + /// + /// Panics if the registered count would exceed `usize::MAX`. + pub fn register(&self, parties: usize) -> Result { + let phaser = self.clone(); + self.do_register(parties)?; + Ok(PhaserParticipants { + phaser, + remaining: parties, + }) + } + + fn do_register(&self, parties: usize) -> Result<(), Closed> { + let mut state = self.state.lock(); + if state.closed { + return Err(Closed(())); + } + let registered = state + .registered + .checked_add(parties) + .expect("Phaser registered-party count overflow"); + state.registered = registered; + // unarrived <= registered, so the registered-count check also covers this addition. + state.unarrived += parties; + Ok(()) + } + + /// Waits until the current phase differs from a phase previously observed on this phaser. + /// + /// Returns the current phase, possibly skipping intermediate phases. This does not wait for + /// a future target number: a number different from the current phase returns immediately. + /// It neither registers a participant nor records an arrival. + /// + /// Returns [`Closed`] if the observed phase is still current when the phaser closes. A phase + /// completed before the phaser was closed remains successful. Phase numbers wrap; do not retain + /// an observation across a full `u64` cycle or use a number obtained from another phaser. + /// + /// # Cancel safety + /// + /// Cancelling only unregisters this wait's waker. The same observation can be retried. + pub async fn wait(&self, observed: u64) -> Result { + PhaserWait { + phaser: self, + observed, + token: None, + } + .await + } +} + +/// An owning iterator over a batch registered by [`Phaser::register`]. +/// +/// All participants are already registered, so even those not yet yielded hold back phase +/// advancement. Dropping this iterator withdraws the remaining participants in one state +/// transition. Yielded participants are independent and retain their registrations. +/// +/// Closing the phaser does not prevent iteration over this existing batch, but arrival and waiting +/// on the yielded participants return [`Closed`]. This iterator is not cloneable. +#[must_use = "dropping the iterator withdraws participants that have not been yielded"] +#[derive(Debug)] +pub struct PhaserParticipants { + phaser: Phaser, + remaining: usize, +} + +impl Iterator for PhaserParticipants { + type Item = PhaserParticipant; + + fn next(&mut self) -> Option { + if self.remaining == 0 { + return None; + } + let participant = PhaserParticipant::new(self.phaser.clone()); + self.remaining -= 1; + Some(participant) + } + + fn size_hint(&self) -> (usize, Option) { + (self.remaining, Some(self.remaining)) + } +} + +impl ExactSizeIterator for PhaserParticipants {} + +impl FusedIterator for PhaserParticipants {} + +impl Drop for PhaserParticipants { + fn drop(&mut self) { + if self.remaining == 0 { + return; + } + let wakers = { + let mut state = self.phaser.state.lock(); + // Unyielded participants have never arrived and prevent their phase from advancing. + state.registered -= self.remaining; + state.unarrived -= self.remaining; + self.remaining = 0; + state.advance_if_ready() + }; + wake_all(wakers.into_iter().flatten()); + } +} + +/// One participant's arrival obligation in every phase until it deregisters or is dropped. +/// +/// This handle is not cloneable. Dropping it withdraws the participant, including any outstanding +/// current arrival. It does not report successful work or close the other participants. +/// +/// See the [module level documentation](self) for usage examples and cancellation semantics. +#[must_use = "dropping a participant withdraws it from the Phaser"] +#[derive(Debug)] +pub struct PhaserParticipant { + phaser: Phaser, + // Cleared only after advancement, so a pending current phase also records arrival. + pending: Option, + registered: bool, +} + +impl PhaserParticipant { + fn new(phaser: Phaser) -> Self { + Self { + phaser, + pending: None, + registered: true, + } + } + + /// Returns the shared coordination handle without registering another participant. + pub fn phaser(&self) -> &Phaser { + &self.phaser + } + + /// Records this participant's arrival and remembers that phase for [`wait`](Self::wait). + /// + /// Returns the phase in which arrival was recorded, or [`Closed`] without recording one. + /// Repeated calls within one phase count only once. After advancement, an explicit new call + /// arrives in the new phase and replaces any previous pending observation. + /// + /// Arrival and the pending observation remain committed if notifying a waker panics. + pub fn arrive(&mut self) -> Result { + let (phase, wakers) = { + let mut state = self.phaser.state.lock(); + if state.closed { + return Err(Closed(())); + } + let phase = state.phase; + if self.pending != Some(phase) { + state.unarrived -= 1; + } + self.pending = Some(phase); + (phase, state.advance_if_ready()) + }; + wake_all(wakers.into_iter().flatten()); + Ok(phase) + } + + /// Arrives if necessary and waits for this participant's pending phase to complete. + /// + /// After an explicit [`arrive`](Self::arrive), this observes that arrival even if the phase + /// already completed. With no pending observation, the first poll arrives in the current + /// phase. Success consumes the pending observation and returns the new current phase. + /// + /// # Cancel safety + /// + /// An unpolled call has no effect. Once polled, arrival is committed. Cancelling preserves + /// the pending observation, so retrying this method waits for the same phase without + /// counting another arrival. An explicit new `arrive` replaces that pending observation. + pub async fn wait(&mut self) -> Result { + let observed = match self.pending { + Some(phase) => phase, + None => self.arrive()?, + }; + let next = self.phaser.wait(observed).await?; + self.pending = None; + Ok(next) + } + + /// Withdraws this participant and returns the phase from which it withdrew, or [`Closed`]. + /// + /// Any outstanding arrival is discharged without counting an already-arrived participant + /// twice. This always removes the registration, even if the phaser is closed. The pending + /// observation is abandoned. Dropping a participant has the same membership effect. + pub fn deregister(mut self) -> Result { + self.do_deregister() + } + + fn do_deregister(&mut self) -> Result { + let (result, wakers) = { + let mut state = self.phaser.state.lock(); + self.registered = false; + state.registered -= 1; + if self.pending != Some(state.phase) { + state.unarrived -= 1; + } + let result = if state.closed { + Err(Closed(())) + } else { + Ok(state.phase) + }; + (result, state.advance_if_ready()) + }; + wake_all(wakers.into_iter().flatten()); + result + } +} + +impl Drop for PhaserParticipant { + fn drop(&mut self) { + if self.registered { + let _ = self.do_deregister(); + } + } +} + +#[must_use = "futures do nothing unless you .await or poll them"] +struct PhaserWait<'a> { + phaser: &'a Phaser, + observed: u64, + token: Option, +} + +impl Future for PhaserWait<'_> { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.get_mut(); + let mut state = this.phaser.state.lock(); + if let ready @ Poll::Ready(_) = state.completion(this.observed) { + this.token = None; + return ready; + } + let retired = state.waiters.register(&mut this.token, cx.waker()); + drop(state); + drop(retired); + Poll::Pending + } +} + +impl Drop for PhaserWait<'_> { + fn drop(&mut self) { + if self.token.is_none() { + return; + } + let mut state = self.phaser.state.lock(); + if state.completion(self.observed).is_ready() { + return; + } + let retired = state.waiters.unregister(&mut self.token); + drop(state); + drop(retired); + } +} diff --git a/asyncband/src/phaser/tests.rs b/asyncband/src/phaser/tests.rs new file mode 100644 index 00000000..48cffa89 --- /dev/null +++ b/asyncband/src/phaser/tests.rs @@ -0,0 +1,46 @@ +// 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::task::Poll; + +use super::Phaser; +use crate::test_support::poll_once; + +#[test] +fn phase_identity_wraps_without_an_ordering_contract() { + let phaser = Phaser::new(); + phaser.state.lock().phase = u64::MAX; + let observed = phaser.phase(); + let mut participant = phaser.register_one().unwrap(); + + assert_eq!(participant.arrive().unwrap(), observed); + assert_eq!(phaser.phase(), 0); + assert_ne!(phaser.phase(), observed); +} + +#[test] +fn a_late_waiter_observes_completion_across_counter_wraparound() { + let phaser = Phaser::new(); + phaser.state.lock().phase = u64::MAX; + let mut participant = phaser.register_one().unwrap(); + participant.arrive().unwrap(); + phaser.close(); + assert_eq!( + poll_once(Box::pin(participant.wait()).as_mut()), + Poll::Ready(Ok(0)) + ); +} diff --git a/benchmarks/Cargo.toml b/benchmarks/Cargo.toml index 19e5ac33..4e98b791 100644 --- a/benchmarks/Cargo.toml +++ b/benchmarks/Cargo.toml @@ -39,6 +39,7 @@ asyncband = { workspace = true, features = [ "once-cell", "once-map", "oneshot", + "phaser", "pool", "rwlock", "semaphore", diff --git a/benchmarks/asyncband/main.rs b/benchmarks/asyncband/main.rs index 907a07ef..8d0c8e08 100644 --- a/benchmarks/asyncband/main.rs +++ b/benchmarks/asyncband/main.rs @@ -27,6 +27,7 @@ mod mutex; mod once; mod once_map; mod oneshot; +mod phaser; mod pool; mod rwlock; mod semaphore; diff --git a/benchmarks/asyncband/phaser/mod.rs b/benchmarks/asyncband/phaser/mod.rs new file mode 100644 index 00000000..00b12391 --- /dev/null +++ b/benchmarks/asyncband/phaser/mod.rs @@ -0,0 +1,119 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::pin::pin; + +use asyncband::phaser::Phaser; +use divan::Bencher; +use divan::black_box; + +use crate::support::bench_context; +use crate::support::poll_pending; +use crate::support::poll_pinned_ready; +use crate::support::poll_ready; + +const PARTICIPANT_COUNTS: &[usize] = &[2, 8, 32]; + +#[divan::bench] +fn ready_wait(bencher: Bencher) { + let mut context = bench_context(); + let phaser = Phaser::new(); + let observed = phaser.phase(); + phaser.register_one().unwrap().arrive().unwrap(); + + bencher.bench_local(|| black_box(poll_ready(phaser.wait(observed), &mut context).unwrap())); +} + +#[divan::bench] +fn repoll_pending(bencher: Bencher) { + let mut context = bench_context(); + let phaser = Phaser::new(); + let mut wait = pin!(phaser.wait(phaser.phase())); + poll_pending(wait.as_mut(), &mut context); + + bencher.bench_local(|| poll_pending(wait.as_mut(), &mut context)); +} + +#[divan::bench] +fn cancel_pending(bencher: Bencher) { + let mut context = bench_context(); + let phaser = Phaser::new(); + let observed = phaser.phase(); + + bencher.bench_local(|| { + let mut wait = pin!(phaser.wait(observed)); + poll_pending(wait.as_mut(), &mut context); + }); + black_box(phaser); +} + +#[divan::bench(args = PARTICIPANT_COUNTS)] +fn register_batch(bencher: Bencher, parties: usize) { + let phaser = Phaser::new(); + + bencher.bench_local(|| { + let participants: Vec<_> = phaser.register(black_box(parties)).unwrap().collect(); + drop(black_box(participants)); + }); +} + +#[divan::bench(args = PARTICIPANT_COUNTS)] +fn register_individually(bencher: Bencher, parties: usize) { + let phaser = Phaser::new(); + + bencher.bench_local(|| { + let participants: Vec<_> = (0..black_box(parties)) + .map(|_| phaser.register_one().unwrap()) + .collect(); + drop(black_box(participants)); + }); +} + +#[divan::bench(args = PARTICIPANT_COUNTS)] +fn arrive_then_wait(bencher: Bencher, parties: usize) { + let mut context = bench_context(); + let phaser = Phaser::new(); + let mut participants: Vec<_> = phaser.register(parties).unwrap().collect(); + + bencher.bench_local(|| { + for participant in &mut participants { + black_box(participant.arrive().unwrap()); + } + for participant in &mut participants { + black_box(poll_ready(participant.wait(), &mut context).unwrap()); + } + }); +} + +#[divan::bench(args = PARTICIPANT_COUNTS)] +fn notify_pending_fanout(bencher: Bencher, parties: usize) { + let mut context = bench_context(); + let phaser = Phaser::new(); + let mut participants: Vec<_> = phaser.register(parties).unwrap().collect(); + let (last, others) = participants.split_last_mut().unwrap(); + + bencher.bench_local(|| { + let mut waiters: Vec<_> = others.iter_mut().map(|p| Box::pin(p.wait())).collect(); + for waiter in &mut waiters { + poll_pending(waiter.as_mut(), &mut context); + } + black_box(poll_ready(last.wait(), &mut context).unwrap()); + for waiter in &mut waiters { + black_box(poll_pinned_ready(waiter.as_mut(), &mut context).unwrap()); + } + }); +} diff --git a/examples/Cargo.toml b/examples/Cargo.toml index 6e079a89..47a82de5 100644 --- a/examples/Cargo.toml +++ b/examples/Cargo.toml @@ -30,6 +30,7 @@ asyncband = { workspace = true, features = [ "completion", "lazy-cell", "once-cell", + "phaser", "shutdown", ] } tokio = { workspace = true, features = [ @@ -58,3 +59,15 @@ path = "src/graceful_shutdown.rs" [[example]] name = "shared_completion" path = "src/shared_completion.rs" + +[[example]] +name = "phaser_rounds" +path = "src/phaser_rounds.rs" + +[[example]] +name = "phaser_completion" +path = "src/phaser_completion.rs" + +[[example]] +name = "phaser_groups" +path = "src/phaser_groups.rs" diff --git a/examples/src/phaser_completion.rs b/examples/src/phaser_completion.rs new file mode 100644 index 00000000..cef5d205 --- /dev/null +++ b/examples/src/phaser_completion.rs @@ -0,0 +1,170 @@ +// 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. + +//! Publish consistent progress snapshots from parallel import workers. +//! +//! Each worker contributes its cumulative record count after a batch. The coordinator sums those +//! counts and publishes a snapshot before workers process the next batch. Separate ready/resume +//! phasers prevent a fast worker from updating its count while the snapshot is being prepared. +//! The coordinator can also await an asynchronous checkpoint before releasing the workers. +//! +//! The first scenario stops once the total reaches a target. The other two show how a failed or +//! cancelled worker stops its peers, including a task cancelled before its first poll. Import work +//! is represented by counters, and checkpoint I/O by a yield. +//! +//! Membership is fixed within this protocol; changes must update both groups at a common batch +//! boundary. Each phaser has its own counter, distinct from the application's batch number. + +use std::sync::Arc; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; + +use asyncband::phaser::Closed; +use asyncband::phaser::Phaser; +use asyncband::phaser::PhaserParticipant; + +// Own this guard before constructing a task future so cancelling an unpolled task also aborts. +struct CloseOnDrop([Phaser; 2]); + +impl Drop for CloseOnDrop { + fn drop(&mut self) { + for phaser in &self.0 { + phaser.close(); + } + } +} + +struct Member { + // Fields drop in declaration order: close before withdrawing any arrival obligation. + _close: CloseOnDrop, + ready: PhaserParticipant, + resume: PhaserParticipant, +} + +impl Member { + fn register(ready: &Phaser, resume: &Phaser) -> Result { + Ok(Self { + _close: CloseOnDrop([ready.clone(), resume.clone()]), + ready: ready.register_one()?, + resume: resume.register_one()?, + }) + } +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Closed> { + publish_until_target().await?; + failure_closes_the_group().await?; + cancelling_an_unpolled_task_closes_the_group().await?; + Ok(()) +} + +async fn compute( + mut member: Member, + id: usize, + values: Arc>, + published: Arc, +) -> Result<(), Closed> { + for round in 1_u64.. { + // The resume rendezvous must publish the previous aggregate before this read. + assert_eq!(published.load(Ordering::Relaxed), (round - 1) * 6); + values[id].store((id as u64 + 1) * round, Ordering::Relaxed); + member.ready.wait().await?; + member.resume.wait().await?; + } + unreachable!() +} + +async fn publish_until_target() -> Result<(), Closed> { + let ready = Phaser::new(); + let resume = Phaser::new(); + let mut coordinator = Member::register(&ready, &resume)?; + let values = Arc::new((0..3).map(|_| AtomicU64::new(0)).collect::>()); + let published = Arc::new(AtomicU64::new(0)); + let mut tasks = Vec::new(); + // Register everyone before polling any worker; the coordinator also keeps both phases open. + for id in 0..3 { + tasks.push(tokio::spawn(compute( + Member::register(&ready, &resume)?, + id, + values.clone(), + published.clone(), + ))); + } + + loop { + coordinator.ready.wait().await?; + let sum: u64 = values + .iter() + .map(|value| value.load(Ordering::Relaxed)) + .sum(); + // An async checkpoint can be awaited here while workers wait at resume. + tokio::task::yield_now().await; + published.store(sum, Ordering::Relaxed); + println!("coordinator: published aggregate {sum}"); + if sum >= 18 { + // Stop without releasing anyone into another computation round. + ready.close(); + resume.close(); + break; + } + coordinator.resume.wait().await?; + } + for task in tasks { + assert!(task.await.expect("worker panicked").is_err()); + } + assert_eq!(published.load(Ordering::Relaxed), 18); + println!("target reached: all workers stopped after 18 imported records"); + Ok(()) +} + +async fn wait_once(mut member: Member) -> Result<(), Closed> { + member.ready.wait().await?; + member.resume.wait().await?; + Ok(()) +} + +async fn fail(_member: Member) -> Result<(), &'static str> { + // The job error stays in its result; Closed tells peers that no next round is available. + Err("input validation failed") +} + +async fn failure_closes_the_group() -> Result<(), Closed> { + let ready = Phaser::new(); + let resume = Phaser::new(); + let healthy = Member::register(&ready, &resume)?; + let failing = Member::register(&ready, &resume)?; + let (peer, failure) = tokio::join!(wait_once(healthy), fail(failing)); + assert!(peer.is_err()); + assert_eq!(failure, Err("input validation failed")); + assert_eq!(ready.phase(), 0); + assert_eq!(resume.phase(), 0); + println!("failure: peers observed Closed, not successful phase completion"); + Ok(()) +} + +async fn cancelling_an_unpolled_task_closes_the_group() -> Result<(), Closed> { + let ready = Phaser::new(); + let resume = Phaser::new(); + let peer = Member::register(&ready, &resume)?; + let cancelled = wait_once(Member::register(&ready, &resume)?); + drop(cancelled); + assert!(wait_once(peer).await.is_err()); + assert_eq!(ready.phase(), 0); + println!("cancellation: dropping an unpolled task closed both gates"); + Ok(()) +} diff --git a/examples/src/phaser_groups.rs b/examples/src/phaser_groups.rs new file mode 100644 index 00000000..9db8c35f --- /dev/null +++ b/examples/src/phaser_groups.rs @@ -0,0 +1,199 @@ +// 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. + +//! Synchronize the time steps of a simulation whose workers are grouped by region. +//! +//! This example models progress tracking for two regions with two workers each. Workers publish +//! their completed time step, then wait for their region's driver. Each driver represents its +//! region at a root phaser and releases local workers only after every region is ready. No region +//! can start the next step while another is still processing the current one. +//! +//! The first run completes three steps. The second fails a worker during step two and verifies +//! that all groups stop with only step one completed globally. The simulation's domain calculation +//! is omitted; the shared counters record each worker's completed step. +//! +//! The groups and drivers are explicit application code, with independent local phase counters +//! and fixed membership. This does not provide automatic parent registration or arbitrary +//! concurrent membership changes. Performance relative to a flat phaser depends on the workload. + +use std::error::Error; +use std::sync::Arc; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::Ordering; + +use asyncband::phaser::Closed; +use asyncband::phaser::Phaser; +use asyncband::phaser::PhaserParticipant; + +const GROUPS: usize = 2; +const WORKERS_PER_GROUP: usize = 2; +const ROUNDS: u64 = 3; + +struct AbortOnDrop { + phasers: [Phaser; 3], + armed: bool, +} + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + if self.armed { + for phaser in &self.phasers { + phaser.close(); + } + } + } +} + +struct LocalMember { + // Abort before dropping either participant, including when a task is never polled. + abort: AbortOnDrop, + ready: PhaserParticipant, + resume: PhaserParticipant, +} + +impl LocalMember { + fn register(root: &Phaser, ready: &Phaser, resume: &Phaser) -> Result { + Ok(Self { + abort: AbortOnDrop { + phasers: [root.clone(), ready.clone(), resume.clone()], + armed: true, + }, + ready: ready.register_one()?, + resume: resume.register_one()?, + }) + } +} + +async fn worker( + mut member: LocalMember, + id: usize, + values: Arc>, + fail: bool, +) -> Result<(), Box> { + for round in 1..=ROUNDS { + if fail && round == 2 { + // The abort guard closes the root before any participant is withdrawn. + return Err("input validation failed".into()); + } + values[id].store(round, Ordering::Relaxed); + member.ready.wait().await?; + member.resume.wait().await?; + // Every group, not merely this worker's local group, must have published this round. + assert!( + values + .iter() + .all(|value| value.load(Ordering::Relaxed) >= round) + ); + } + member.abort.armed = false; + Ok(()) +} + +struct GroupDriver { + // Keep the root obligation behind the local abort guard in the same owned task argument. + local: LocalMember, + root: PhaserParticipant, +} + +async fn drive_group(mut driver: GroupDriver) -> Result<(), Box> { + for _ in 0..ROUNDS { + driver.local.ready.wait().await?; + driver.root.wait().await?; + driver.local.resume.wait().await?; + } + driver.local.abort.armed = false; + Ok(()) +} + +struct CloseRootOnDrop(Phaser); + +impl Drop for CloseRootOnDrop { + fn drop(&mut self) { + self.0.close(); + } +} + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Box> { + run_groups(false).await?; + assert!(run_groups(true).await.unwrap_err().is::()); + println!("group failure: every local group stopped after the root was closed"); + Ok(()) +} + +async fn run_groups(fail_one_worker: bool) -> Result<(), Box> { + let root = Phaser::new(); + let mut coordinator = root.register_one()?; + // Created after the participant so cancellation closes the root before withdrawing it. + let _close_root = CloseRootOnDrop(root.clone()); + let values = Arc::new( + (0..GROUPS * WORKERS_PER_GROUP) + .map(|_| AtomicU64::new(0)) + .collect::>(), + ); + let mut tasks = Vec::new(); + for group in 0..GROUPS { + let ready = Phaser::new(); + let resume = Phaser::new(); + let driver = LocalMember::register(&root, &ready, &resume)?; + let representative = root.register_one()?; + for worker_id in 0..WORKERS_PER_GROUP { + let member = LocalMember::register(&root, &ready, &resume)?; + let id = group * WORKERS_PER_GROUP + worker_id; + tasks.push(tokio::spawn(worker( + member, + id, + values.clone(), + fail_one_worker && id == 0, + ))); + } + tasks.push(tokio::spawn(drive_group(GroupDriver { + local: driver, + root: representative, + }))); + } + assert_eq!(root.registered_parties(), GROUPS + 1); + + for round in 1..=ROUNDS { + if let Err(error) = coordinator.wait().await { + // After the root is closed, group drivers close their local phasers. + root.close(); + let mut failures = 0; + for task in tasks { + let error = task.await.expect("group task panicked").unwrap_err(); + if !error.is::() { + assert_eq!(error.to_string(), "input validation failed"); + failures += 1; + } + } + assert_eq!(failures, 1); + assert_eq!(root.phase(), 1); + return Err(error.into()); + } + println!("root: all {GROUPS} groups completed round {round}"); + } + for task in tasks { + task.await.expect("group task panicked")?; + } + assert!( + values + .iter() + .all(|value| value.load(Ordering::Relaxed) == ROUNDS) + ); + println!("grouped fan-in: four workers synchronized through two root representatives"); + Ok(()) +} diff --git a/examples/src/phaser_rounds.rs b/examples/src/phaser_rounds.rs new file mode 100644 index 00000000..132ee335 --- /dev/null +++ b/examples/src/phaser_rounds.rs @@ -0,0 +1,138 @@ +// 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. + +//! Coordinate a batch job while its worker pool changes. +//! +//! The start-gate scenario keeps workers waiting until the coordinator has finished setting up +//! the job. The changing-membership scenario adds a worker after the first batch and lets an +//! independent task report progress without holding up the workers. The cancellation scenario +//! retries a wait that lost a select race, without counting the worker in the next batch early. +//! +//! These are coordination patterns for jobs such as parallel imports. The work within a batch is +//! represented by a yield; the examples focus on when workers can join, proceed, and leave. + +use asyncband::phaser::Closed; +use asyncband::phaser::Phaser; +use asyncband::phaser::PhaserParticipant; + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Closed> { + start_gate().await?; + changing_membership().await?; + cancellation_retry().await?; + Ok(()) +} + +async fn start_gate() -> Result<(), Closed> { + let phaser = Phaser::new(); + let setup = phaser.register_one()?; + let mut tasks = Vec::new(); + for mut participant in phaser.register(3)? { + tasks.push(tokio::spawn(async move { + participant.wait().await?; + // Initialization is complete; real work may now start. + Ok::<_, Closed>(()) + })); + } + setup.deregister()?; + for task in tasks { + task.await.expect("worker panicked")?; + } + assert_eq!(phaser.registered_parties(), 0); + assert!(!phaser.is_closed()); + println!("start gate: all three workers released; the empty phaser remains reusable"); + Ok(()) +} + +async fn work(mut participant: PhaserParticipant, rounds: usize) -> Result<(), Closed> { + for _ in 0..rounds { + let observed = participant.arrive()?; + // This work does not hold up the other participants' arrivals. + tokio::task::yield_now().await; + let next = participant.wait().await?; + assert_ne!(observed, next); + } + participant.deregister()?; + Ok(()) +} + +async fn changing_membership() -> Result<(), Closed> { + let phaser = Phaser::new(); + let mut coordinator = phaser.register_one()?; + let worker = tokio::spawn(work(phaser.register_one()?, 3)); + + let progress = phaser.clone(); + let observer = tokio::spawn(async move { + let mut observed = progress.phase(); + while let Ok(next) = progress.wait(observed).await { + println!("observer: phase {observed} -> {next}"); + // A slow observer may skip phases; it never delays workers. + observed = next; + } + }); + let target = phaser.clone(); + let target_wait = tokio::spawn(async move { wait_until(&target, 2).await }); + + coordinator.wait().await?; + // The coordinator has not arrived in the next phase, so this registration joins that phase. + let joining_worker = tokio::spawn(work(phaser.register_one()?, 2)); + assert_eq!(phaser.registered_parties(), 3); + coordinator.wait().await?; + coordinator.wait().await?; + + worker.await.expect("worker panicked")?; + joining_worker.await.expect("joining worker panicked")?; + assert!(target_wait.await.expect("target observer panicked")? >= 2); + coordinator.deregister()?; + phaser.close(); + observer.await.expect("progress observer panicked"); + println!("dynamic membership: a second worker joined after the first round"); + Ok(()) +} + +/// A caller-side numeric threshold, for a run known not to cross counter wraparound. +/// This observer does not register, drive the computation, or guarantee one notification for +/// every intermediate phase. +async fn wait_until(phaser: &Phaser, target: u64) -> Result { + let mut observed = phaser.phase(); + while observed < target { + observed = phaser.wait(observed).await?; + } + Ok(observed) +} + +async fn cancellation_retry() -> Result<(), Closed> { + let phaser = Phaser::new(); + let mut participant = phaser.register_one()?; + let mut peer = phaser.register_one()?; + let observed = phaser.phase(); + + tokio::select! { + biased; + result = participant.wait() => panic!("peer has not arrived: {result:?}"), + // Poll the wait once, then cancel it deterministically without a timer or sleep. + _ = std::future::ready(()) => {} + } + assert_eq!(phaser.arrived_parties(), 1); + peer.arrive()?; + assert_ne!(phaser.phase(), observed); + + assert_eq!(participant.wait().await?, phaser.phase()); + assert_eq!(phaser.arrived_parties(), 0); + println!("cancellation: retry observed the completed round without arriving in the next one"); + Ok(()) +} diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 04df5894..eb419c32 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -41,6 +41,7 @@ asyncband = { workspace = true, features = [ "once-cell", "once-map", "oneshot", + "phaser", "pool", "rwlock", "semaphore", diff --git a/tests-integration/tests/phaser_test.rs b/tests-integration/tests/phaser_test.rs new file mode 100644 index 00000000..9730c859 --- /dev/null +++ b/tests-integration/tests/phaser_test.rs @@ -0,0 +1,788 @@ +// 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::sync::Arc; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::task::Context; +use std::task::Poll; +use std::task::Wake; +use std::task::Waker; + +use asyncband::phaser::Phaser; + +fn poll_once(future: std::pin::Pin<&mut F>) -> Poll { + future.poll(&mut Context::from_waker(Waker::noop())) +} + +struct CountWake(AtomicUsize); + +impl Wake for CountWake { + fn wake(self: Arc) { + self.0.fetch_add(1, Ordering::Relaxed); + } +} + +struct PanicWake; + +impl Wake for PanicWake { + fn wake(self: Arc) { + panic!("wake failed"); + } +} + +struct CloseOnDrop(Phaser); + +impl Wake for CloseOnDrop { + fn wake(self: Arc) { + self.0.close(); + } +} + +impl Drop for CloseOnDrop { + fn drop(&mut self) { + self.0.close(); + } +} + +#[test] +fn batch_registration_joins_one_observed_phase() { + let phaser = Phaser::new(); + let mut participants = phaser.register(3).unwrap(); + + assert_eq!(participants.len(), 3); + assert_eq!(phaser.registered_parties(), 3); + assert_eq!(phaser.unarrived_parties(), 3); + + let mut first = participants.next().unwrap(); + let observed = first.arrive().unwrap(); + let mut second = participants.next().unwrap(); + second.arrive().unwrap(); + assert_eq!(participants.len(), 1); + assert_eq!(phaser.phase(), observed); + + let counter = Arc::new(CountWake(AtomicUsize::new(0))); + let waker = Waker::from(counter.clone()); + let mut wait = Box::pin(first.wait()); + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + drop(participants); + assert_eq!(counter.0.load(Ordering::Relaxed), 1); + assert_eq!(poll_once(wait.as_mut()), Poll::Ready(Ok(phaser.phase()))); + assert_ne!(phaser.phase(), observed); + assert_eq!(phaser.registered_parties(), 2); + assert_eq!(phaser.unarrived_parties(), 2); +} + +#[test] +fn collecting_a_batch_can_unwind_without_leaking_registrations() { + let phaser = Phaser::new(); + let mut coordinator = phaser.register_one().unwrap(); + let observed = phaser.phase(); + + assert!( + panic::catch_unwind(|| { + let _: Vec<_> = phaser + .register(3) + .unwrap() + .enumerate() + .map(|(index, participant)| { + assert_ne!(index, 1, "task setup failed"); + participant + }) + .collect(); + }) + .is_err() + ); + assert_eq!(phaser.phase(), observed); + assert_eq!(phaser.registered_parties(), 1); + assert_eq!(phaser.unarrived_parties(), 1); + coordinator.arrive().unwrap(); + assert_ne!(phaser.phase(), observed); +} + +#[test] +fn exhausted_batch_does_not_advance_a_dormant_phaser() { + let phaser = Phaser::new(); + let mut participants = phaser.register(1).unwrap(); + drop(participants.next().unwrap()); + let completed = phaser.phase(); + + assert_eq!(participants.len(), 0); + assert!(participants.next().is_none()); + drop(participants); + assert_eq!(phaser.phase(), completed); + assert_eq!(phaser.registered_parties(), 0); +} + +#[test] +fn an_existing_batch_can_be_iterated_and_withdrawn_after_close() { + let phaser = Phaser::new(); + let mut participants = phaser.register(3).unwrap(); + let observed = phaser.phase(); + phaser.close(); + + let mut participant = participants.next().unwrap(); + drop(participants); + assert_eq!(phaser.registered_parties(), 1); + assert_eq!(phaser.unarrived_parties(), 1); + assert!(participant.arrive().is_err()); + drop(participant); + assert_eq!(phaser.phase(), observed); + assert_eq!(phaser.registered_parties(), 0); + assert_eq!(phaser.unarrived_parties(), 0); +} + +#[test] +fn registering_zero_parties_is_a_noop() { + let phaser = Phaser::new(); + let phase = phaser.phase(); + + assert_eq!(phaser.register(0).unwrap().len(), 0); + assert_eq!(phaser.phase(), phase); + assert_eq!(phaser.registered_parties(), 0); + assert_eq!(phaser.unarrived_parties(), 0); +} + +#[test] +fn participants_advance_across_repeated_phases() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + + assert_eq!(first.arrive().unwrap(), phase0); + assert_eq!(phaser.arrived_parties(), 1); + assert_eq!(second.arrive().unwrap(), phase0); + let phase1 = phaser.phase(); + assert_ne!(phase1, phase0); + assert_eq!(phaser.arrived_parties(), 0); + + assert_eq!(first.arrive().unwrap(), phase1); + assert_eq!(second.arrive().unwrap(), phase1); + assert_ne!(phaser.phase(), phase1); +} + +#[test] +fn unpolled_wait_future_does_not_arrive() { + let phaser = Phaser::new(); + let mut participant = phaser.register_one().unwrap(); + + let wait = participant.wait(); + + assert_eq!(phaser.arrived_parties(), 0); + drop(wait); + assert_eq!(phaser.arrived_parties(), 0); +} + +#[test] +fn cancelled_wait_retry_waits_for_original_phase_after_advance() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + + { + let mut cancelled = Box::pin(first.wait()); + assert!(poll_once(cancelled.as_mut()).is_pending()); + } + + assert_eq!(phaser.arrived_parties(), 1); + assert_eq!(second.arrive().unwrap(), phase0); + let phase1 = phaser.phase(); + assert_ne!(phase1, phase0); + assert_eq!(phaser.arrived_parties(), 0); + + let mut retry = Box::pin(first.wait()); + assert_eq!(poll_once(retry.as_mut()), Poll::Ready(Ok(phase1))); + assert_eq!(phaser.arrived_parties(), 0); +} + +#[test] +fn cancelled_wait_retry_before_advance_does_not_arrive_twice() { + let phaser = Phaser::new(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + + { + let mut cancelled = Box::pin(first.wait()); + assert!(poll_once(cancelled.as_mut()).is_pending()); + } + + let mut retry = Box::pin(first.wait()); + assert!(poll_once(retry.as_mut()).is_pending()); + assert_eq!(phaser.arrived_parties(), 1); + + second.arrive().unwrap(); + assert!(poll_once(retry.as_mut()).is_ready()); +} + +#[test] +fn dropping_last_participant_advances_once_and_dormant_phaser_can_be_reused() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let participant = phaser.register_one().unwrap(); + + drop(participant); + let phase1 = phaser.phase(); + assert_ne!(phase1, phase0); + assert_eq!(phaser.registered_parties(), 0); + assert_eq!(phaser.arrived_parties(), 0); + + let mut participant = phaser.register_one().unwrap(); + assert_eq!(participant.arrive().unwrap(), phase1); + assert_ne!(phaser.phase(), phase1); +} + +#[test] +fn dropping_an_arrived_participant_only_removes_its_next_phase_registration() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + + first.arrive().unwrap(); + drop(first); + assert_eq!(phaser.phase(), phase0); + assert_eq!(phaser.registered_parties(), 1); + assert_eq!(phaser.unarrived_parties(), 1); + + second.arrive().unwrap(); + assert_ne!(phaser.phase(), phase0); +} + +#[test] +fn registration_before_last_arrival_joins_and_delays_current_phase() { + let phaser = Phaser::new(); + let phase = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + + first.arrive().unwrap(); + let mut joining = phaser.register_one().unwrap(); + second.arrive().unwrap(); + + assert_eq!(phaser.phase(), phase); + assert_eq!(phaser.unarrived_parties(), 1); + joining.arrive().unwrap(); + assert_ne!(phaser.phase(), phase); +} + +#[test] +fn registration_after_last_arrival_joins_the_advanced_phase() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + + first.arrive().unwrap(); + let phase1 = phaser.phase(); + assert_ne!(phase1, phase0); + + let mut joining = phaser.register_one().unwrap(); + assert_eq!(phaser.registered_parties(), 2); + assert_eq!(phaser.unarrived_parties(), 2); + assert_eq!(joining.arrive().unwrap(), phase1); + assert_eq!(phaser.phase(), phase1); +} + +#[test] +fn registration_before_last_participant_drop_joins_the_current_phase() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let participant = phaser.register_one().unwrap(); + let joining = phaser.register_one().unwrap(); + + drop(participant); + + assert_eq!(phaser.phase(), phase0); + assert_eq!(phaser.registered_parties(), 1); + assert_eq!(phaser.unarrived_parties(), 1); + assert_eq!(joining.deregister().unwrap(), phase0); + assert_ne!(phaser.phase(), phase0); +} + +#[test] +fn registration_after_last_participant_drop_joins_the_advanced_phase() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let participant = phaser.register_one().unwrap(); + + drop(participant); + let phase1 = phaser.phase(); + let joining = phaser.register_one().unwrap(); + + assert_ne!(phase1, phase0); + assert_eq!(phaser.registered_parties(), 1); + assert_eq!(phaser.unarrived_parties(), 1); + assert_eq!(joining.deregister().unwrap(), phase1); + assert_ne!(phaser.phase(), phase1); +} + +#[test] +fn observer_wait_is_cancel_safe_and_does_not_participate() { + let phaser = Phaser::new(); + let phase = phaser.phase(); + let counter = Arc::new(CountWake(AtomicUsize::new(0))); + let waker = Waker::from(Arc::clone(&counter)); + let mut context = Context::from_waker(&waker); + + { + let mut wait = Box::pin(phaser.wait(phase)); + assert_eq!(Future::poll(wait.as_mut(), &mut context), Poll::Pending); + assert_eq!(phaser.registered_parties(), 0); + } + + assert_eq!(phaser.registered_parties(), 0); + let participant = phaser.register_one().unwrap(); + drop(participant); + assert_eq!(counter.0.load(Ordering::Relaxed), 0); +} + +#[test] +fn advancing_a_phase_wakes_every_registered_waiter_once() { + let phaser = Phaser::new(); + let observed = phaser.phase(); + let participant = phaser.register_one().unwrap(); + let first_counter = Arc::new(CountWake(AtomicUsize::new(0))); + let second_counter = Arc::new(CountWake(AtomicUsize::new(0))); + let first_waker = Waker::from(Arc::clone(&first_counter)); + let second_waker = Waker::from(Arc::clone(&second_counter)); + let mut first_context = Context::from_waker(&first_waker); + let mut second_context = Context::from_waker(&second_waker); + let mut first_wait = Box::pin(phaser.wait(observed)); + let mut second_wait = Box::pin(phaser.wait(observed)); + + assert_eq!( + Future::poll(first_wait.as_mut(), &mut first_context), + Poll::Pending + ); + assert_eq!( + Future::poll(second_wait.as_mut(), &mut second_context), + Poll::Pending + ); + + drop(participant); + assert_eq!(first_counter.0.load(Ordering::Relaxed), 1); + assert_eq!(second_counter.0.load(Ordering::Relaxed), 1); + assert!(matches!( + Future::poll(first_wait.as_mut(), &mut first_context), + Poll::Ready(_) + )); + assert!(matches!( + Future::poll(second_wait.as_mut(), &mut second_context), + Poll::Ready(_) + )); +} + +#[test] +fn repolling_updates_the_task_that_will_be_notified() { + let phaser = Phaser::new(); + let participant = phaser.register_one().unwrap(); + let first = Arc::new(CountWake(AtomicUsize::new(0))); + let second = Arc::new(CountWake(AtomicUsize::new(0))); + let first_waker = Waker::from(first.clone()); + let second_waker = Waker::from(second.clone()); + let mut wait = Box::pin(phaser.wait(phaser.phase())); + + for waker in [&first_waker, &first_waker, &second_waker, &second_waker] { + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(waker)) + .is_pending() + ); + } + drop(participant); + + assert_eq!(first.0.load(Ordering::Relaxed), 0); + assert_eq!(second.0.load(Ordering::Relaxed), 1); + assert_eq!(poll_once(wait.as_mut()), Poll::Ready(Ok(phaser.phase()))); +} + +#[test] +fn cancelling_a_woken_waiter_does_not_unregister_a_next_phase_waiter() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let participant = phaser.register_one().unwrap(); + let stale_counter = Arc::new(CountWake(AtomicUsize::new(0))); + let stale_waker = Waker::from(Arc::clone(&stale_counter)); + let mut stale_context = Context::from_waker(&stale_waker); + let mut stale_wait = Box::pin(phaser.wait(phase0)); + + assert_eq!( + Future::poll(stale_wait.as_mut(), &mut stale_context), + Poll::Pending + ); + drop(participant); + let phase1 = phaser.phase(); + assert_ne!(phase1, phase0); + assert_eq!(stale_counter.0.load(Ordering::Relaxed), 1); + + let participant = phaser.register_one().unwrap(); + let current_counter = Arc::new(CountWake(AtomicUsize::new(0))); + let current_waker = Waker::from(Arc::clone(¤t_counter)); + let mut current_context = Context::from_waker(¤t_waker); + let mut current_wait = Box::pin(phaser.wait(phase1)); + assert_eq!( + Future::poll(current_wait.as_mut(), &mut current_context), + Poll::Pending + ); + + drop(stale_wait); + drop(participant); + assert_eq!(current_counter.0.load(Ordering::Relaxed), 1); + assert!(matches!( + Future::poll(current_wait.as_mut(), &mut current_context), + Poll::Ready(_) + )); +} + +#[test] +fn panicking_waker_does_not_lose_a_pending_phase() { + let phaser = Phaser::new(); + let phase0 = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + let panic_waker = Waker::from(Arc::new(PanicWake)); + let mut panic_context = Context::from_waker(&panic_waker); + let mut observer = Box::pin(phaser.wait(phase0)); + + assert_eq!( + Future::poll(observer.as_mut(), &mut panic_context), + Poll::Pending + ); + assert_eq!(first.arrive().unwrap(), phase0); + + let polling_waker = Waker::from(Arc::new(CountWake(AtomicUsize::new(0)))); + let mut polling_context = Context::from_waker(&polling_waker); + let mut wait = Box::pin(second.wait()); + let result = panic::catch_unwind(panic::AssertUnwindSafe(|| { + Future::poll(wait.as_mut(), &mut polling_context) + })); + + assert!(result.is_err()); + drop(wait); + drop(observer); + + let phase1 = phaser.phase(); + assert_ne!(phase1, phase0); + assert_eq!(phaser.arrived_parties(), 0); + + let mut retry = Box::pin(second.wait()); + assert_eq!(poll_once(retry.as_mut()), Poll::Ready(Ok(phase1))); + assert_eq!(phaser.arrived_parties(), 0); +} + +#[test] +fn a_late_waiter_for_a_completed_phase_is_immediately_ready() { + let phaser = Phaser::new(); + let observed = phaser.phase(); + let participant = phaser.register_one().unwrap(); + drop(participant); + + let mut wait = Box::pin(phaser.wait(observed)); + assert_eq!(poll_once(wait.as_mut()), Poll::Ready(Ok(phaser.phase()))); +} + +#[test] +fn registration_overflow_panics_without_partially_updating_state() { + let phaser = Phaser::new(); + let participants = phaser.register(usize::MAX).unwrap(); + + assert!(panic::catch_unwind(|| phaser.register_one().unwrap()).is_err()); + assert!(panic::catch_unwind(|| phaser.register(2).unwrap()).is_err()); + assert_eq!(phaser.registered_parties(), usize::MAX); + assert_eq!(phaser.unarrived_parties(), usize::MAX); + drop(participants); + assert_eq!(phaser.registered_parties(), 0); + assert_eq!(phaser.unarrived_parties(), 0); +} + +#[test] +fn explicit_arrival_and_wait_observe_the_same_completed_phase() { + let phaser = Phaser::new(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + let observed = first.arrive().unwrap(); + second.arrive().unwrap(); + let next = phaser.phase(); + + assert_ne!(observed, next); + assert_eq!( + poll_once(Box::pin(first.wait()).as_mut()), + Poll::Ready(Ok(next)) + ); + assert_eq!( + poll_once(Box::pin(second.wait()).as_mut()), + Poll::Ready(Ok(next)) + ); + assert_eq!(phaser.arrived_parties(), 0); + + let mut wait = Box::pin(first.wait()); + assert!(poll_once(wait.as_mut()).is_pending()); + second.arrive().unwrap(); + assert_eq!(poll_once(wait.as_mut()), Poll::Ready(Ok(phaser.phase()))); +} + +#[test] +fn explicit_arrival_replaces_a_cancelled_pending_observation() { + let phaser = Phaser::new(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + assert!(poll_once(Box::pin(first.wait()).as_mut()).is_pending()); + second.arrive().unwrap(); + let next = first.arrive().unwrap(); + assert_eq!(next, phaser.phase()); + + let mut wait = Box::pin(first.wait()); + assert!(poll_once(wait.as_mut()).is_pending()); + second.arrive().unwrap(); + assert_eq!(poll_once(wait.as_mut()), Poll::Ready(Ok(phaser.phase()))); +} + +#[test] +fn cloned_handles_observe_without_registering_and_participants_own_the_state() { + let phaser = Phaser::new(); + let observer = phaser.clone(); + let mut participant = phaser.register_one().unwrap(); + drop(phaser); + assert_eq!(observer.registered_parties(), 1); + let observed = observer.phase(); + participant.arrive().unwrap(); + assert_eq!( + poll_once(Box::pin(observer.wait(observed)).as_mut()), + Poll::Ready(Ok(observer.phase())) + ); +} + +#[test] +fn closing_wakes_all_waiters_once_and_rejects_new_obligations() { + let phaser = Phaser::new(); + let mut first = phaser.register_one().unwrap(); + let second = phaser.register_one().unwrap(); + let observed = first.arrive().unwrap(); + let counter = Arc::new(CountWake(AtomicUsize::new(0))); + let waker = Waker::from(counter.clone()); + let mut context = Context::from_waker(&waker); + let mut observer = Box::pin(phaser.wait(observed)); + let mut wait = Box::pin(first.wait()); + assert!(observer.as_mut().poll(&mut context).is_pending()); + assert!(wait.as_mut().poll(&mut context).is_pending()); + + phaser.clone().close(); + phaser.close(); + assert!(phaser.is_closed()); + assert_eq!(counter.0.load(Ordering::Relaxed), 2); + assert!(matches!(poll_once(observer.as_mut()), Poll::Ready(Err(_)))); + assert!(matches!(poll_once(wait.as_mut()), Poll::Ready(Err(_)))); + drop(wait); + assert!(first.arrive().is_err()); + assert!(phaser.register_one().is_err()); + assert!(phaser.register(2).is_err()); + assert!(phaser.register(0).is_err()); + assert!(first.deregister().is_err()); + drop(second); + assert_eq!(phaser.registered_parties(), 0); + assert_eq!(phaser.unarrived_parties(), 0); + assert_eq!(phaser.phase(), observed); +} + +#[test] +fn completed_arrival_remains_successful_after_close_but_cannot_start_another_round() { + let phaser = Phaser::new(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + let observed = first.arrive().unwrap(); + let mut observer = Box::pin(phaser.wait(observed)); + assert!(poll_once(observer.as_mut()).is_pending()); + second.arrive().unwrap(); + let completed = phaser.phase(); + phaser.close(); + + assert_eq!(poll_once(observer.as_mut()), Poll::Ready(Ok(completed))); + assert_eq!( + poll_once(Box::pin(first.wait()).as_mut()), + Poll::Ready(Ok(completed)) + ); + assert!(matches!( + poll_once(Box::pin(first.wait()).as_mut()), + Poll::Ready(Err(_)) + )); + drop(first); + drop(second); + assert_eq!(phaser.phase(), completed); +} + +#[test] +fn close_survives_a_panicking_waker_and_notifies_other_waiters() { + let phaser = Phaser::new(); + let observed = phaser.phase(); + let panic_waker = Waker::from(Arc::new(PanicWake)); + let counter = Arc::new(CountWake(AtomicUsize::new(0))); + let count_waker = Waker::from(counter.clone()); + let mut first = Box::pin(phaser.wait(observed)); + let mut second = Box::pin(phaser.wait(observed)); + assert!( + first + .as_mut() + .poll(&mut Context::from_waker(&panic_waker)) + .is_pending() + ); + assert!( + second + .as_mut() + .poll(&mut Context::from_waker(&count_waker)) + .is_pending() + ); + + assert!(panic::catch_unwind(|| phaser.close()).is_err()); + assert!(phaser.is_closed()); + assert_eq!(counter.0.load(Ordering::Relaxed), 1); + assert!(matches!(poll_once(first.as_mut()), Poll::Ready(Err(_)))); + assert!(matches!(poll_once(second.as_mut()), Poll::Ready(Err(_)))); +} + +#[test] +fn replacing_a_waiter_waker_can_close_the_phaser_from_its_destructor() { + let phaser = Phaser::new(); + let waker = Waker::from(Arc::new(CloseOnDrop(phaser.clone()))); + let mut wait = Box::pin(phaser.wait(phaser.phase())); + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + drop(waker); + assert!(!phaser.is_closed()); + + let counter = Arc::new(CountWake(AtomicUsize::new(0))); + let replacement = Waker::from(counter.clone()); + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&replacement)) + .is_pending() + ); + assert!(phaser.is_closed()); + assert_eq!(counter.0.load(Ordering::Relaxed), 1); + assert!(matches!(poll_once(wait.as_mut()), Poll::Ready(Err(_)))); +} + +#[test] +fn cancelling_a_waiter_can_close_the_phaser_from_its_waker_destructor() { + let phaser = Phaser::new(); + let waker = Waker::from(Arc::new(CloseOnDrop(phaser.clone()))); + let mut wait = Box::pin(phaser.wait(phaser.phase())); + assert!( + wait.as_mut() + .poll(&mut Context::from_waker(&waker)) + .is_pending() + ); + drop(waker); + assert!(!phaser.is_closed()); + + drop(wait); + assert!(phaser.is_closed()); +} + +#[tokio::test] +#[cfg_attr(miri, ignore = "requires an OS-backed Tokio runtime")] +async fn participant_can_wait_from_a_spawned_task() { + let phaser = Phaser::new(); + let mut first = phaser.register_one().unwrap(); + let mut second = phaser.register_one().unwrap(); + + let first_wait = tokio::spawn(async move { first.wait().await }); + + tokio::task::yield_now().await; + let phase = second.arrive().unwrap(); + assert_eq!(first_wait.await.unwrap().unwrap(), phaser.phase()); + assert_ne!(phase, phaser.phase()); +} + +#[tokio::test] +#[cfg_attr(miri, ignore = "requires an OS-backed Tokio runtime")] +async fn observer_waits_without_becoming_a_party() { + let phaser = Phaser::new(); + let observed = phaser.phase(); + let mut first = phaser.register_one().unwrap(); + let second = phaser.register_one().unwrap(); + let observer_phaser = phaser.clone(); + let observer = tokio::spawn(async move { observer_phaser.wait(observed).await }); + + tokio::task::yield_now().await; + assert_eq!(phaser.registered_parties(), 2); + first.arrive().unwrap(); + second.deregister().unwrap(); + + assert_eq!(observer.await.unwrap().unwrap(), phaser.phase()); + assert_eq!(phaser.registered_parties(), 1); +} + +#[test] +fn arrivals_publish_each_workers_writes_across_threads() { + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + + let phaser = Phaser::new(); + let values = std::array::from_fn::<_, 4, _>(|_| AtomicUsize::new(0)); + let participants = phaser.register(values.len()).unwrap(); + std::thread::scope(|scope| { + for (id, mut participant) in participants.enumerate() { + let values = &values; + scope.spawn(move || { + pollster::block_on(async { + for round in 1..=16 { + values[id].store(round, Ordering::Relaxed); + participant.wait().await.unwrap(); + assert!( + values + .iter() + .all(|value| value.load(Ordering::Relaxed) == round) + ); + // Keep the next round's writers behind this read-side rendezvous. + participant.wait().await.unwrap(); + } + }); + }); + } + }); +} + +#[tokio::test] +#[cfg_attr(miri, ignore = "requires an OS-backed Tokio runtime")] +async fn a_failed_task_can_close_the_group_without_reporting_phase_completion() { + let phaser = Phaser::new(); + let mut worker = phaser.register_one().unwrap(); + let failing = phaser.register_one().unwrap(); + let observed = phaser.phase(); + let (arrived, arrival) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + worker.arrive().unwrap(); + arrived.send(()).unwrap(); + worker.wait().await + }); + arrival.await.unwrap(); + failing.phaser().close(); + drop(failing); + assert!(task.await.unwrap().is_err()); + assert_eq!(phaser.phase(), observed); + assert_eq!(phaser.registered_parties(), 0); +} diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs index dfbefb8c..2ddcad88 100644 --- a/tests-integration/tests/traits_test.rs +++ b/tests-integration/tests/traits_test.rs @@ -34,6 +34,10 @@ use asyncband::once::Once; use asyncband::once::OnceCell; use asyncband::once::OnceMap; use asyncband::oneshot; +use asyncband::phaser::Closed; +use asyncband::phaser::Phaser; +use asyncband::phaser::PhaserParticipant; +use asyncband::phaser::PhaserParticipants; use asyncband::pool; use asyncband::pool::ManageObject; use asyncband::pool::ObjectStatus; @@ -103,6 +107,10 @@ fn public_types_are_send_and_sync() { assert_send_and_sync::(); assert_send_and_sync::>(); assert_send_and_sync::>(); + assert_send_and_sync::(); + assert_send_and_sync::(); + assert_send_and_sync::(); + assert_send_and_sync::(); assert_send_and_sync::>(); assert_send_and_sync::>(); assert_send_and_sync::>(); @@ -172,6 +180,10 @@ fn public_types_are_unpin() { assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); + assert_unpin::(); + assert_unpin::(); + assert_unpin::(); + assert_unpin::(); assert_unpin::>(); assert_unpin::>(); assert_unpin::>(); diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 41c5a877..51fb9866 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -121,6 +121,10 @@ impl CommandMiri { &["--test", "unsafe_paths_test"], )); run_command(make_miri_cmd("tests-integration", &["--test", "mpsc_test"])); + run_command(make_miri_cmd( + "tests-integration", + &["--test", "phaser_test"], + )); } }