From baae7a01c4e74adae4b76ef703922686e9f746c6 Mon Sep 17 00:00:00 2001 From: tison Date: Tue, 1 Sep 2026 01:28:44 +0800 Subject: [PATCH] refactor(internal): lift panic-safe fan-out waking wake_all is shared by WaitSet, WaitList, countdown, and semaphore-backed primitives, so keeping it inside waitset couples unrelated feature builds to the waiter arena. Move the helper to the internal module boundary and restrict waitset compilation to features that actually use WaitSet or WakerToken. Signed-off-by: tison --- asyncband/src/barrier/mod.rs | 2 +- asyncband/src/broadcast/mpmc/unbounded/mod.rs | 2 +- asyncband/src/completion/mod.rs | 2 +- asyncband/src/condvar/mod.rs | 2 +- asyncband/src/event/mod.rs | 2 +- asyncband/src/internal/countdown.rs | 2 +- asyncband/src/internal/mod.rs | 43 +++++++++++++++---- asyncband/src/internal/semaphore.rs | 2 +- asyncband/src/internal/waitset.rs | 33 ++------------ asyncband/src/watch/mod.rs | 2 +- 10 files changed, 46 insertions(+), 46 deletions(-) diff --git a/asyncband/src/barrier/mod.rs b/asyncband/src/barrier/mod.rs index a0ee4b1b..aca2be36 100644 --- a/asyncband/src/barrier/mod.rs +++ b/asyncband/src/barrier/mod.rs @@ -56,7 +56,7 @@ use std::task::Poll; use crate::internal::mutex::Mutex; use crate::internal::waitset::WaitSet; use crate::internal::waitset::WakerToken; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; /// A synchronization primitive for multiple tasks that need to wait for each other. /// diff --git a/asyncband/src/broadcast/mpmc/unbounded/mod.rs b/asyncband/src/broadcast/mpmc/unbounded/mod.rs index 1e57028c..1feb1053 100644 --- a/asyncband/src/broadcast/mpmc/unbounded/mod.rs +++ b/asyncband/src/broadcast/mpmc/unbounded/mod.rs @@ -105,7 +105,7 @@ use crate::internal::arena::SlotId; use crate::internal::mutex::Mutex; use crate::internal::waitset::WaitSet; use crate::internal::waitset::WakerToken; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; #[cfg(test)] mod tests; diff --git a/asyncband/src/completion/mod.rs b/asyncband/src/completion/mod.rs index 024f1c90..c8ab8207 100644 --- a/asyncband/src/completion/mod.rs +++ b/asyncband/src/completion/mod.rs @@ -60,7 +60,7 @@ use std::task::Poll; use crate::internal::mutex::Mutex; use crate::internal::waitset::WaitSet; use crate::internal::waitset::WakerToken; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; /// Creates a single-use [`Completer`] and a cloneable [`Completion`] observer. pub fn new() -> (Completer, Completion) { diff --git a/asyncband/src/condvar/mod.rs b/asyncband/src/condvar/mod.rs index 69db906f..18d3a648 100644 --- a/asyncband/src/condvar/mod.rs +++ b/asyncband/src/condvar/mod.rs @@ -68,7 +68,7 @@ use std::task::Waker; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; use crate::mutex; use crate::mutex::MutexGuard; use crate::mutex::OwnedMutexGuard; diff --git a/asyncband/src/event/mod.rs b/asyncband/src/event/mod.rs index 07b81b34..312a3a17 100644 --- a/asyncband/src/event/mod.rs +++ b/asyncband/src/event/mod.rs @@ -62,7 +62,7 @@ use std::task::Waker; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; /// A reusable event that remains set until explicitly reset. /// diff --git a/asyncband/src/internal/countdown.rs b/asyncband/src/internal/countdown.rs index 70a8187f..9e331648 100644 --- a/asyncband/src/internal/countdown.rs +++ b/asyncband/src/internal/countdown.rs @@ -23,7 +23,7 @@ use std::task::Poll; use crate::internal::mutex::Mutex; use crate::internal::waitset::WaitSet; use crate::internal::waitset::WakerToken; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; #[derive(Debug)] pub struct CountdownState { diff --git a/asyncband/src/internal/mod.rs b/asyncband/src/internal/mod.rs index 2369c5ec..a0605117 100644 --- a/asyncband/src/internal/mod.rs +++ b/asyncband/src/internal/mod.rs @@ -15,6 +15,40 @@ // specific language governing permissions and limitations // under the License. +use std::panic; +use std::panic::AssertUnwindSafe; +use std::task::Waker; + +/// Wakes every waker while preserving the first panic. +/// +/// If a wake callback panics, the remaining callbacks are still attempted during unwinding. Any +/// later panic is suppressed so the first panic can continue to the caller. +#[inline] +// A no-feature or blocking-only build has no primitive that fans notifications out. +#[allow(dead_code)] +pub(crate) fn wake_all(mut wakers: impl Iterator) { + struct WakeRemaining<'a, I: Iterator> { + wakers: &'a mut I, + } + + impl> Drop for WakeRemaining<'_, I> { + fn drop(&mut self) { + // This iterator is empty after normal completion. During unwinding, attempt every + // callback left after the one that panicked without replacing the original panic. + for waker in self.wakers.by_ref() { + let _ = panic::catch_unwind(AssertUnwindSafe(|| waker.wake())); + } + } + } + + let remaining = WakeRemaining { + wakers: &mut wakers, + }; + for waker in remaining.wakers.by_ref() { + waker.wake(); + } +} + #[cfg(feature = "mpsc")] pub(crate) mod atomic_waker; @@ -90,20 +124,13 @@ pub(crate) mod waitlist; #[cfg(any( feature = "barrier", feature = "broadcast", - feature = "condvar", - feature = "event", feature = "completion", feature = "latch", - feature = "mpsc", - feature = "mutex", feature = "once", - feature = "rwlock", - feature = "semaphore", feature = "waitgroup", feature = "watch", ))] // `barrier` constructs a wait set with `with_capacity`, while completion and countdown-based -// primitives use `new`; `condvar`, `event`, and semaphore-backed primitives use only the free -// `wake_all` helper. One constructor is therefore unused in every single-primitive build. +// primitives use `new`. One constructor is therefore unused in every single-primitive build. #[allow(dead_code)] pub(crate) mod waitset; diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs index 1873d703..bb0afe56 100644 --- a/asyncband/src/internal/semaphore.rs +++ b/asyncband/src/internal/semaphore.rs @@ -29,7 +29,7 @@ use std::task::Waker; use crate::internal::mutex::Mutex; use crate::internal::waitlist::WaitList; use crate::internal::waitlist::WaiterId; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; /// The internal semaphore that provides low-level async primitives. #[derive(Debug)] diff --git a/asyncband/src/internal/waitset.rs b/asyncband/src/internal/waitset.rs index 36cf96c0..13f580dd 100644 --- a/asyncband/src/internal/waitset.rs +++ b/asyncband/src/internal/waitset.rs @@ -22,41 +22,11 @@ //! wake them after unlocking. use std::mem; -use std::panic; -use std::panic::AssertUnwindSafe; use std::task::Waker; use crate::internal::arena::Arena; use crate::internal::arena::SlotId; -/// Wakes every waker while preserving the first panic. -/// -/// If a wake callback panics, the remaining callbacks are still attempted during unwinding. Any -/// later panic is suppressed so the first panic can continue to the caller. -#[inline] -pub fn wake_all(mut wakers: impl Iterator) { - struct WakeRemaining<'a, I: Iterator> { - wakers: &'a mut I, - } - - impl> Drop for WakeRemaining<'_, I> { - fn drop(&mut self) { - // This iterator is empty after normal completion. During unwinding, attempt every - // callback left after the one that panicked without replacing the original panic. - for waker in self.wakers.by_ref() { - let _ = panic::catch_unwind(AssertUnwindSafe(|| waker.wake())); - } - } - } - - let remaining = WakeRemaining { - wakers: &mut wakers, - }; - for waker in remaining.wakers.by_ref() { - waker.wake(); - } -} - /// An exclusive handle to one waiter slot in a [`WaitSet`]. /// /// The wait set owns the registered waker; this token only lets its future update or cancel that @@ -159,6 +129,8 @@ impl WaitSet { #[cfg(test)] mod tests { + use std::panic; + use std::panic::AssertUnwindSafe; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicUsize; @@ -166,6 +138,7 @@ mod tests { use std::task::Wake; use super::*; + use crate::internal::wake_all; #[test] fn waker_token_preserves_the_option_niche() { diff --git a/asyncband/src/watch/mod.rs b/asyncband/src/watch/mod.rs index 5d9e393d..cf68a6ba 100644 --- a/asyncband/src/watch/mod.rs +++ b/asyncband/src/watch/mod.rs @@ -76,7 +76,7 @@ pub use self::error::SendError; use crate::internal::mutex::Mutex; use crate::internal::waitset::WaitSet; use crate::internal::waitset::WakerToken; -use crate::internal::waitset::wake_all; +use crate::internal::wake_all; /// Creates a watch channel with an initial value. ///