You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asyncband.apache.org by ti...@apache.org on 2026/09/03 14:20:55 UTC
(asyncband) 01/01: refactor(waitgroup): separate coordinator and worker roles
This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch codex/split-waitgroup-roles
in repository https://gitbox.apache.org/repos/asf/asyncband.git
commit 4bd97a228fe4b4e7beb508c584d945d04fae1f16
Author: tison <wa...@gmail.com>
AuthorDate: Thu Sep 3 22:20:38 2026 +0800
refactor(waitgroup): separate coordinator and worker roles
---
CHANGELOG.md | 1 +
HISTORY.md | 2 +-
README.md | 2 +-
asyncband/src/lib.rs | 2 +-
asyncband/src/shutdown/mod.rs | 14 ++--
asyncband/src/waitgroup/mod.rs | 106 ++++++++++++++++++++----------
benchmarks/asyncband/waitgroup/wait.rs | 17 +++--
tests-integration/tests/traits_test.rs | 3 +
tests-integration/tests/waitgroup_test.rs | 32 +++++++--
9 files changed, 120 insertions(+), 59 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index c51ffff..13f57b1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ All notable changes to this project will be documented in this file.
* Remove the lossy `broadcast::overflow` channel; use the new lossless `broadcast::mpmc::unbounded` channel instead.
* Remove `OnceMap::with_capacity` and `OnceMap::with_capacity_and_hasher`; use `OnceMap::new` or `OnceMap::with_hasher`, which allocate the backing table lazily.
* Remove the unconstructible `LatchWait` and `OwnedLatchWait` types from the public API; `Latch::wait` and `Latch::wait_owned` continue to return anonymous futures through their `async fn` signatures.
+* Redesign `waitgroup::WaitGroup` as a non-cloneable coordinator with explicit `Worker` handles: replace `WaitGroup::clone` with `WaitGroup::worker`, clone workers for nested work, and await or call `WaitGroup::wait` only from the coordinator.
* Rename `oneshot::Sender::is_closed` and `oneshot::Receiver::is_closed` to `is_disconnected`.
* Remove `Semaphore::try_acquire_and_forget`, `Semaphore::acquire_and_forget`, `Semaphore::try_acquire_owned_and_forget`, and `Semaphore::acquire_owned_and_forget`; acquire a permit and call its `forget` method instead.
* Replace `Semaphore::forget` with `Semaphore::drain_permits` and `Semaphore::forget_exact` with `Semaphore::reduce_permits`; permit-level `forget` methods are unchanged.
diff --git a/HISTORY.md b/HISTORY.md
index 020ff63..5b0719c 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -13,4 +13,4 @@ Asyncband collects composable, runtime-agnostic concurrency building blocks info
- `pool` is ported from [`fastpool`](https://github.com/fast/fastpool), which is derived from [`deadpool`](https://github.com/bikeshedder/deadpool), while keeping Fastpool's runtime-agnostic design and caller-side timeout composition.
- `rwlock::RwLock` is derived from [`tokio::sync::RwLock`](https://docs.rs/tokio/latest/tokio/sync/struct.RwLock.html), but accepts any `NonZeroUsize` as `max_readers` instead of Tokio's restricted range.
- `semaphore::Semaphore` is derived from [`tokio::sync::Semaphore`](https://docs.rs/tokio/latest/tokio/sync/struct.Semaphore.html), but omits `close`, avoids Tokio's fixed maximum-permit constant, and adds operations such as `reduce_permits` for Asyncband's use cases.
-- `waitgroup::WaitGroup` is inspired by [`waitgroup-rs`](https://github.com/laizy/waitgroup-rs), with a different API and an implementation based on the internal `CountdownState` primitive.
+- `waitgroup::WaitGroup` adopts the coordinator and worker roles from [`waitgroup-rs`](https://github.com/laizy/waitgroup-rs), but uses Asyncband's internal `CountdownState` to provide cancellable multi-observer waits and explicit completion synchronization.
diff --git a/README.md b/README.md
index b079a6c..34dccfe 100644
--- a/README.md
+++ b/README.md
@@ -75,7 +75,7 @@ Runnable examples live in the [`examples`](examples) workspace crate. They demon
| | [`Completion`](https://docs.rs/asyncband/*/asyncband/completion/struct.Completion.html) | `completion` | Publish one shared result to any number of current and future observers. |
| | [`ManualResetEvent`](https://docs.rs/asyncband/*/asyncband/event/struct.ManualResetEvent.html) | `event` | Signal current and future waits until explicitly reset. |
| | [`Latch`](https://docs.rs/asyncband/*/asyncband/latch/struct.Latch.html) | `latch` | Wait until a fixed one-way countdown reaches zero. |
-| | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Wait until all cloned worker handles are dropped. |
+| | [`WaitGroup`](https://docs.rs/asyncband/*/asyncband/waitgroup/struct.WaitGroup.html) | `waitgroup` | Create explicit worker handles and wait until all are dropped. |
| | [`Shutdown`](https://docs.rs/asyncband/*/asyncband/shutdown/struct.Shutdown.html) | `shutdown` | Request shutdown and wait until all completion guards are dropped. |
| Channels | [`oneshot`](https://docs.rs/asyncband/*/asyncband/oneshot/) | `oneshot` | Send one value from one sender to one receiver. |
| | [`mpsc`](https://docs.rs/asyncband/*/asyncband/mpsc/) | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. |
diff --git a/asyncband/src/lib.rs b/asyncband/src/lib.rs
index 934e28a..f18bbc5 100644
--- a/asyncband/src/lib.rs
+++ b/asyncband/src/lib.rs
@@ -67,7 +67,7 @@
//! | | [`Completion`](completion::Completion) | `completion` | Publish one shared result to any number of current and future observers. |
//! | | [`ManualResetEvent`](event::ManualResetEvent) | `event` | Signal current and future waits until explicitly reset. |
//! | | [`Latch`](latch::Latch) | `latch` | Wait until a fixed one-way countdown reaches zero. |
-//! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Wait until all cloned worker handles are dropped. |
+//! | | [`WaitGroup`](waitgroup::WaitGroup) | `waitgroup` | Create explicit worker handles and wait until all are dropped. |
//! | | [`Shutdown`](shutdown::Shutdown) | `shutdown` | Request shutdown and wait until all completion guards are dropped. |
//! | Channels | [`oneshot`] | `oneshot` | Send one value from one sender to one receiver. |
//! | | [`mpsc`] | `mpsc` | Send each value from multiple producers to one receiver with bounded backpressure or an unbounded queue. |
diff --git a/asyncband/src/shutdown/mod.rs b/asyncband/src/shutdown/mod.rs
index f530750..722b0b6 100644
--- a/asyncband/src/shutdown/mod.rs
+++ b/asyncband/src/shutdown/mod.rs
@@ -61,7 +61,6 @@
//! ```
use std::future::Future;
-use std::future::IntoFuture;
use std::pin::Pin;
use std::sync::Arc;
use std::task::Context;
@@ -70,20 +69,21 @@ use std::task::Poll;
use crate::latch::Latch;
use crate::waitgroup::Wait;
use crate::waitgroup::WaitGroup;
+use crate::waitgroup::Worker;
/// Creates a graceful shutdown coordinator and an initial completion guard.
///
/// See the [module level documentation](self) for more.
pub fn new() -> (Shutdown, ShutdownGuard) {
let latch = Arc::new(Latch::new(1));
- let wg = WaitGroup::new();
- let shutdown = Shutdown {
+ let group = WaitGroup::new();
+ let guard = ShutdownGuard {
latch: latch.clone(),
- wait: wg.clone().into_future(),
+ worker: group.worker(),
};
- let guard = ShutdownGuard {
+ let shutdown = Shutdown {
latch,
- wait_group: wg,
+ wait: group.wait(),
};
(shutdown, guard)
}
@@ -155,7 +155,7 @@ pub struct ShutdownGuard {
dead_code,
reason = "keeps shutdown completion pending until this guard is dropped"
)]
- wait_group: WaitGroup,
+ worker: Worker,
}
impl ShutdownGuard {
diff --git a/asyncband/src/waitgroup/mod.rs b/asyncband/src/waitgroup/mod.rs
index e4be015..4548cd1 100644
--- a/asyncband/src/waitgroup/mod.rs
+++ b/asyncband/src/waitgroup/mod.rs
@@ -17,9 +17,16 @@
//! Wait for a set of worker handles to be dropped.
//!
-//! A [`WaitGroup`] starts with one coordinator handle. Clone that handle once for each unit of work
-//! and move the clones into their workers. Dropping a worker handle marks that worker as complete.
-//! Awaiting the coordinator consumes it and waits until every remaining handle has been dropped.
+//! A [`WaitGroup`] coordinates completion without itself representing unfinished work. Call
+//! [`WaitGroup::worker`] once for each unit of work and move the returned [`Worker`] into that
+//! work. Workers may be cloned for nested work. Dropping the last worker completes the group.
+//!
+//! Awaiting the coordinator consumes it, preventing new top-level workers from being registered
+//! after waiting begins. A worker cannot be awaited, so accidentally waiting from inside a worker
+//! does not remove that worker from the group.
+//!
+//! Completion acquires the state published before every worker was dropped, so work performed by
+//! those workers is visible after the wait returns.
//!
//! # Examples
//!
@@ -34,7 +41,7 @@
//! let mut tasks = vec![];
//!
//! for _ in 0..3 {
-//! let worker = group.clone();
+//! let worker = group.worker();
//! tasks.push(tokio::spawn(async move {
//! do_work().await;
//! drop(worker); // Signals completion. This would also happen at the end of the task.
@@ -59,7 +66,7 @@ use std::task::Poll;
use crate::internal::countdown::CountdownState;
use crate::internal::waitset::WakerToken;
-/// A group of handles whose collective completion can be awaited.
+/// Coordinates a set of [`Worker`] handles whose completion can be awaited.
///
/// See the [module level documentation](self) for more.
pub struct WaitGroup {
@@ -79,45 +86,40 @@ impl Default for WaitGroup {
}
impl WaitGroup {
- /// Creates a new `WaitGroup`.
+ /// Creates an empty `WaitGroup`.
///
/// # Examples
///
/// ```
/// use asyncband::waitgroup::WaitGroup;
///
- /// let wg = WaitGroup::new();
+ /// let group = WaitGroup::new();
/// ```
pub fn new() -> Self {
Self {
- state: Arc::new(CountdownState::new(1)),
+ state: Arc::new(CountdownState::new(0)),
}
}
-}
-impl Clone for WaitGroup {
- /// Creates a new worker handle for the wait group.
+ /// Creates a handle representing one unfinished unit of work.
///
- /// This increments the WaitGroup counter. The counter will be decremented
- /// when the new handle is dropped.
+ /// Clone the returned worker to register nested work. The group completes after every worker
+ /// has been dropped.
///
/// # Panics
///
- /// Panics if the WaitGroup counter would overflow.
- fn clone(&self) -> Self {
- let sync = self.state.clone();
- if sync.increment(1) {
- panic!("WaitGroup counter overflow");
+ /// Panics if the worker count would overflow.
+ pub fn worker(&self) -> Worker {
+ let state = self.state.clone();
+ if state.increment(1) {
+ panic!("WaitGroup worker count overflow");
}
- Self { state: sync }
+ Worker { state }
}
-}
-impl Drop for WaitGroup {
- fn drop(&mut self) {
- if self.state.decrement(1) {
- self.state.wake_all();
- }
+ /// Consumes this coordinator and returns a future that waits for all workers to be dropped.
+ pub fn wait(self) -> Wait {
+ self.into_future()
}
}
@@ -125,18 +127,56 @@ impl IntoFuture for WaitGroup {
type Output = ();
type IntoFuture = Wait;
- /// Consumes this handle and waits for all other handles to be dropped.
+ /// Consumes this coordinator and waits for all workers to be dropped.
fn into_future(self) -> Self::IntoFuture {
+ Wait {
+ token: None,
+ state: self.state,
+ }
+ }
+}
+
+/// Keeps a [`WaitGroup`] pending until this handle and all of its clones are dropped.
+///
+/// A worker may be cloned to represent nested work. Workers deliberately cannot be awaited; only
+/// the coordinating [`WaitGroup`] can begin a wait.
+pub struct Worker {
+ state: Arc<CountdownState>,
+}
+
+impl Clone for Worker {
+ /// Creates a handle representing another unfinished unit of work.
+ ///
+ /// # Panics
+ ///
+ /// Panics if the worker count would overflow.
+ fn clone(&self) -> Self {
let state = self.state.clone();
- drop(self);
- Wait { token: None, state }
+ if state.increment(1) {
+ panic!("WaitGroup worker count overflow");
+ }
+ Worker { state }
+ }
+}
+
+impl fmt::Debug for Worker {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("Worker").finish_non_exhaustive()
+ }
+}
+
+impl Drop for Worker {
+ fn drop(&mut self) {
+ if self.state.decrement(1) {
+ self.state.wake_all();
+ }
}
}
-/// A future that completes when every [`WaitGroup`] handle has been dropped.
+/// A future that completes when every [`Worker`] has been dropped.
///
-/// Awaiting a [`WaitGroup`] creates this future. Cloning a `Wait` creates another observer without
-/// adding a worker to the group.
+/// Awaiting or calling [`WaitGroup::wait`] creates this future. Cloning a `Wait` creates another
+/// observer without adding a worker to the group.
#[must_use = "futures do nothing unless you `.await` or poll them"]
pub struct Wait {
token: Option<WakerToken>,
@@ -144,9 +184,9 @@ pub struct Wait {
}
impl Clone for Wait {
- /// Creates a new future that also completes when the WaitGroup counter reaches zero.
+ /// Creates a new future that observes the same group completion.
///
- /// This does not increment the WaitGroup counter.
+ /// This does not add a worker to the group.
fn clone(&self) -> Self {
Wait {
token: None,
diff --git a/benchmarks/asyncband/waitgroup/wait.rs b/benchmarks/asyncband/waitgroup/wait.rs
index ff4e764..2fa5c1e 100644
--- a/benchmarks/asyncband/waitgroup/wait.rs
+++ b/benchmarks/asyncband/waitgroup/wait.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::future::IntoFuture;
use std::pin::pin;
use asyncband::waitgroup::WaitGroup;
@@ -34,9 +33,9 @@ fn cancel_pending(bencher: Bencher) {
bencher.bench_local(|| {
let root = WaitGroup::new();
- let worker = root.clone();
+ let worker = root.worker();
{
- let mut wait = pin!(root.into_future());
+ let mut wait = pin!(root.wait());
poll_pending(wait.as_mut(), &mut context);
}
drop(worker);
@@ -50,8 +49,8 @@ fn complete_waiter(bencher: Bencher) {
bencher.bench_local(|| {
let root = WaitGroup::new();
- let worker = root.clone();
- let mut wait = pin!(root.into_future());
+ let worker = root.worker();
+ let mut wait = pin!(root.wait());
poll_pending(wait.as_mut(), &mut context);
drop(worker);
@@ -66,8 +65,8 @@ fn complete_worker_batch(bencher: Bencher, worker_count: usize) {
bencher.bench_local(|| {
let root = WaitGroup::new();
- let workers = (0..worker_count).map(|_| root.clone()).collect::<Vec<_>>();
- let mut wait = pin!(root.into_future());
+ let workers = (0..worker_count).map(|_| root.worker()).collect::<Vec<_>>();
+ let mut wait = pin!(root.wait());
poll_pending(wait.as_mut(), &mut context);
drop(workers);
@@ -82,8 +81,8 @@ fn complete_waiter_batch(bencher: Bencher, waiter_count: usize) {
bencher.bench_local(|| {
let root = WaitGroup::new();
- let worker = root.clone();
- let wait = root.into_future();
+ let worker = root.worker();
+ let wait = root.wait();
let mut waiters = (0..waiter_count)
.map(|_| Box::pin(wait.clone()))
.collect::<Vec<_>>();
diff --git a/tests-integration/tests/traits_test.rs b/tests-integration/tests/traits_test.rs
index bf10a0e..17142cd 100644
--- a/tests-integration/tests/traits_test.rs
+++ b/tests-integration/tests/traits_test.rs
@@ -45,6 +45,7 @@ use asyncband::shutdown::ShutdownWatch;
use asyncband::singleflight;
use asyncband::waitgroup::Wait;
use asyncband::waitgroup::WaitGroup;
+use asyncband::waitgroup::Worker;
use asyncband::watch;
struct PoolManager;
@@ -88,6 +89,7 @@ fn public_types_are_send_and_sync() {
assert_send_and_sync::<ShutdownGuard>();
assert_send_and_sync::<ShutdownWatch>();
assert_send_and_sync::<WaitGroup>();
+ assert_send_and_sync::<Worker>();
assert_send_and_sync::<Mutex<i64>>();
assert_send_and_sync::<MutexGuard<'_, i64>>();
assert_send_and_sync::<RwLock<i64>>();
@@ -155,6 +157,7 @@ fn public_types_are_unpin() {
assert_unpin::<ShutdownGuard>();
assert_unpin::<ShutdownWatch>();
assert_unpin::<WaitGroup>();
+ assert_unpin::<Worker>();
assert_unpin::<Wait>();
assert_unpin::<Mutex<i64>>();
assert_unpin::<MutexGuard<'_, i64>>();
diff --git a/tests-integration/tests/waitgroup_test.rs b/tests-integration/tests/waitgroup_test.rs
index c5d77ba..ab6255c 100644
--- a/tests-integration/tests/waitgroup_test.rs
+++ b/tests-integration/tests/waitgroup_test.rs
@@ -15,8 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::future::IntoFuture;
-
use asyncband::waitgroup::WaitGroup;
use tests_integration::poll_once;
@@ -25,7 +23,7 @@ async fn waits_for_all_worker_handles() {
let wg = WaitGroup::new();
let mut tasks = vec![];
for _ in 0..100 {
- let worker = wg.clone();
+ let worker = wg.worker();
tasks.push(tokio::spawn(async move {
drop(worker);
}));
@@ -40,8 +38,8 @@ async fn waits_for_all_worker_handles() {
#[test]
fn wait_is_pending_until_the_last_handle_drops() {
let wg = WaitGroup::new();
- let worker = wg.clone();
- let mut wait = Box::pin(wg.into_future());
+ let worker = wg.worker();
+ let mut wait = Box::pin(wg.wait());
assert!(poll_once(wait.as_mut()).is_pending());
drop(worker);
@@ -51,8 +49,8 @@ fn wait_is_pending_until_the_last_handle_drops() {
#[test]
fn cancelling_one_wait_does_not_cancel_another() {
let wg = WaitGroup::new();
- let worker = wg.clone();
- let first = wg.into_future();
+ let worker = wg.worker();
+ let first = wg.wait();
let mut second = Box::pin(first.clone());
let mut first = Box::pin(first);
@@ -63,3 +61,23 @@ fn cancelling_one_wait_does_not_cancel_another() {
drop(worker);
assert!(poll_once(second.as_mut()).is_ready());
}
+
+#[test]
+fn worker_clones_register_nested_work() {
+ let wg = WaitGroup::new();
+ let worker = wg.worker();
+ let nested = worker.clone();
+ let mut wait = Box::pin(wg.wait());
+
+ drop(worker);
+ assert!(poll_once(wait.as_mut()).is_pending());
+ drop(nested);
+ assert!(poll_once(wait.as_mut()).is_ready());
+}
+
+#[test]
+fn empty_group_is_immediately_ready() {
+ let mut wait = Box::pin(WaitGroup::new().wait());
+
+ assert!(poll_once(wait.as_mut()).is_ready());
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@asyncband.apache.org
For additional commands, e-mail: commits-help@asyncband.apache.org