refactor(internal): lift panic-safe fan-out waking

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

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;

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<T>() -> (Completer<T>, Completion<T>) {

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;

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

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 {

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<Item = Waker>) {
+    struct WakeRemaining<'a, I: Iterator<Item = Waker>> {
+        wakers: &'a mut I,
+    }
+
+    impl<I: Iterator<Item = Waker>> 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;

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)]

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<Item = Waker>) {
-    struct WakeRemaining<'a, I: Iterator<Item = Waker>> {
-        wakers: &'a mut I,
-    }
-
-    impl<I: Iterator<Item = Waker>> 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,13 +129,16 @@ 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;
     use std::sync::atomic::Ordering;
     use std::task::Wake;
 
     use super::*;
+    use crate::internal::wake_all;
 
     #[test]
     fn waker_token_preserves_the_option_niche() {

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