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. ///