refactor: replace waker batch iteration with wake_all

asyncband/src/barrier/mod.rs

@@ -52,7 +52,6 @@ use std::future::Future;
 use std::pin::Pin;
 use std::task::Context;
 use std::task::Poll;
-use std::task::Waker;
 
 use crate::internal::mutex::Mutex;
 use crate::internal::wakerset::WakerSet;
@@ -190,7 +189,7 @@ impl Barrier {
                 state.generation += 1;
                 let mut wakers = state.waiters.take_all();
                 drop(state);
-                wakers.by_ref().for_each(Waker::wake);
+                wakers.wake_all();
                 return BarrierWaitResult(true);
             }
 

asyncband/src/broadcast/mpmc/bounded/mod.rs

@@ -108,7 +108,6 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::task::Context;
 use std::task::Poll;
-use std::task::Waker;
 
 use super::common;
 use super::common::Backlog;
@@ -422,7 +421,7 @@ impl<T> BoundedSender<T> {
         }
         let mut wakers = inner.waiters.take_all();
         drop(inner);
-        wakers.by_ref().for_each(Waker::wake);
+        wakers.wake_all();
         drop(discarded);
         Ok(())
     }

asyncband/src/broadcast/mpmc/common.rs

@@ -343,7 +343,7 @@ impl<T> Backlog<T> {
             .take_while(|slot| slot.cursors == 0)
             .count();
         let rest = if extra == 0 {
-            Vec::new()
+            vec![]
         } else {
             self.buffer.drain(..extra).map(|slot| slot.msg).collect()
         };

asyncband/src/broadcast/mpmc/unbounded/mod.rs

@@ -62,7 +62,6 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 use std::task::Context;
 use std::task::Poll;
-use std::task::Waker;
 
 use super::common;
 use super::common::Backlog;
@@ -180,7 +179,7 @@ impl<T> UnboundedSender<T> {
         drop(inner);
 
         // Wake callbacks and payload destruction may reenter the channel.
-        wakers.by_ref().for_each(Waker::wake);
+        wakers.wake_all();
         drop(unretained);
     }
 

asyncband/src/condvar/mod.rs

@@ -170,7 +170,7 @@ impl Condvar {
             }
         }
 
-        wakers.by_ref().for_each(Waker::wake);
+        wakers.wake_all();
     }
 
     /// Waits for a notification, atomically releasing and then reacquiring the mutex.

asyncband/src/event/manual_reset.rs

@@ -116,7 +116,7 @@ impl ManualResetEvent {
                 }
             }
         }
-        wakers.by_ref().for_each(Waker::wake);
+        wakers.wake_all();
     }
 
     /// Clears the set state.

asyncband/src/internal/mod.rs

@@ -124,6 +124,8 @@ pub(crate) mod waitlist;
     feature = "waitgroup",
     feature = "watch",
 ))]
+// Terminal-only primitives wake WakerSet directly, leaving WakerBatch::wake_all unused.
+#[allow(dead_code)]
 pub(crate) mod waker_batch;
 
 #[cfg(any(

asyncband/src/internal/semaphore.rs

@@ -172,7 +172,7 @@ impl Semaphore {
             }
         }
         drop(waiters);
-        wakers.by_ref().for_each(Waker::wake);
+        wakers.wake_all();
     }
 
     fn insert_permits_with_lock<'a>(
@@ -211,7 +211,7 @@ impl Semaphore {
             }
 
             drop(waiters);
-            batch.by_ref().for_each(Waker::wake);
+            batch.wake_all();
             if rem == 0 {
                 return;
             }

asyncband/src/internal/waker_batch.rs

@@ -15,27 +15,21 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::VecDeque;
+use std::mem;
 use std::mem::MaybeUninit;
 use std::ptr;
 use std::task::Waker;
 
 /// An owning FIFO of wakers that stores the first [`Self::STACK_SIZE`] entries without allocating.
 ///
-/// Only pushed entries are initialized. Iterate by reference to avoid moving the inline storage
-/// into iterator adapters and to reuse the batch, as the semaphore does between notifications.
+/// Only pushed entries are initialized. [`Self::wake_all`] empties the batch in place so the
+/// semaphore can reuse its storage between notifications.
 pub struct WakerBatch {
-    /// The initialized entries are exactly `start..end`.
+    /// The initialized entries are exactly `0..inline_len`.
     inline: [MaybeUninit<Waker>; Self::STACK_SIZE],
-    /// The next inline entry to yield.
-    start: usize,
-    /// The next inline slot to push into.
-    end: usize,
-    /// Entries pushed while the inline storage was full, yielded after it.
-    ///
-    /// While this is non-empty every push lands here, so the batch never yields a later push
-    /// ahead of an earlier one.
-    spilled: VecDeque<Waker>,
+    inline_len: usize,
+    /// Entries pushed after the inline storage filled, woken after it.
+    spilled: Vec<Waker>,
 }
 
 impl WakerBatch {
@@ -47,9 +41,8 @@ impl WakerBatch {
     pub const fn new() -> Self {
         Self {
             inline: [const { MaybeUninit::uninit() }; Self::STACK_SIZE],
-            start: 0,
-            end: 0,
-            spilled: VecDeque::new(),
+            inline_len: 0,
+            spilled: vec![],
         }
     }
 
@@ -59,17 +52,30 @@ impl WakerBatch {
     /// before collecting more.
     #[inline]
     pub fn will_spill(&self) -> bool {
-        self.end == Self::STACK_SIZE || !self.spilled.is_empty()
+        self.inline_len == Self::STACK_SIZE
     }
 
     #[inline]
     pub fn push(&mut self, waker: Waker) {
         if self.will_spill() {
-            self.spilled.push_back(waker);
+            self.spilled.push(waker);
         } else {
-            self.inline[self.end].write(waker);
-            self.end += 1;
+            self.inline[self.inline_len].write(waker);
+            self.inline_len += 1;
+        }
+    }
+
+    /// Wakes every entry in push order, retaining storage for the next batch.
+    ///
+    /// Call after releasing the owning primitive's state lock.
+    #[inline]
+    pub fn wake_all(&mut self) {
+        let len = mem::take(&mut self.inline_len);
+        for slot in &mut self.inline[..len] {
+            // SAFETY: This prefix was initialized by `push` and is no longer owned by the batch.
+            unsafe { slot.assume_init_read() }.wake();
         }
+        self.spilled.drain(..).for_each(Waker::wake);
     }
 }
 
@@ -84,35 +90,14 @@ impl FromIterator<Waker> for WakerBatch {
     }
 }
 
-impl Iterator for WakerBatch {
-    type Item = Waker;
-
-    #[inline]
-    fn next(&mut self) -> Option<Waker> {
-        if self.start < self.end {
-            let index = self.start;
-            self.start += 1;
-            // SAFETY: `index` was within the initialized range before advancing `start`.
-            return Some(unsafe { self.inline[index].assume_init_read() });
-        }
-
-        // Every inline entry has been yielded, so later pushes can start over from the front.
-        self.start = 0;
-        self.end = 0;
-        self.spilled.pop_front()
-    }
-}
-
 impl Drop for WakerBatch {
     #[inline]
     fn drop(&mut self) {
         let initialized = ptr::slice_from_raw_parts_mut(
-            self.inline[self.start..self.end]
-                .as_mut_ptr()
-                .cast::<Waker>(),
-            self.end - self.start,
+            self.inline.as_mut_ptr().cast::<Waker>(),
+            self.inline_len,
         );
-        // SAFETY: The initialized entries are exactly `start..end`.
+        // SAFETY: The initialized entries are exactly `0..inline_len`.
         unsafe { ptr::drop_in_place(initialized) };
     }
 }
@@ -171,70 +156,55 @@ mod tests {
     }
 
     #[test]
-    fn yields_in_push_order_across_the_spill() {
-        let log = log();
-        let count = STACK_SIZE + 8;
-        let mut batch: WakerBatch = (0..count).map(|id| waker(&log, id)).collect();
-        assert!(batch.will_spill());
-
-        for waker in &mut batch {
-            waker.wake();
+    fn wakes_in_push_order_across_the_spill() {
+        for count in [
+            0,
+            1,
+            STACK_SIZE - 1,
+            STACK_SIZE,
+            STACK_SIZE + 1,
+            2 * STACK_SIZE + 1,
+        ] {
+            let log = log();
+            let mut batch: WakerBatch = (0..count).map(|id| waker(&log, id)).collect();
+            batch.wake_all();
+
+            assert_eq!(woken(&log), (0..count).collect::<Vec<_>>());
+            assert_eq!(alive(&log), 0);
         }
-
-        assert_eq!(woken(&log), (0..count).collect::<Vec<_>>());
-        assert!(batch.next().is_none());
-        assert_eq!(alive(&log), 0);
     }
 
     #[test]
-    fn drops_unconsumed_entries_exactly_once() {
-        let log = log();
-        let count = STACK_SIZE + 8;
-        for consumed in [0, 5, STACK_SIZE, STACK_SIZE + 3, count] {
-            let mut batch = WakerBatch::new();
-            push_wakers(&mut batch, &log, count);
-            for _ in 0..consumed {
-                drop(batch.next().unwrap());
+    fn drops_unwoken_entries_exactly_once() {
+        for previously_woken in [0, STACK_SIZE + 1] {
+            for count in [0, 1, STACK_SIZE, STACK_SIZE + 1, 2 * STACK_SIZE + 1] {
+                let log = log();
+                let mut batch = WakerBatch::new();
+                push_wakers(&mut batch, &log, previously_woken);
+                batch.wake_all();
+                push_wakers(&mut batch, &log, count);
+                assert_eq!(alive(&log), count);
+
+                drop(batch);
+                assert_eq!(alive(&log), 0);
+                assert_eq!(woken(&log), (0..previously_woken).collect::<Vec<_>>());
             }
-            assert_eq!(alive(&log), count - consumed);
-
-            drop(batch);
-            assert_eq!(alive(&log), 0, "after consuming {consumed}");
         }
     }
 
     #[test]
-    fn reuses_inline_storage_after_draining() {
+    fn reuses_inline_storage_after_waking() {
         let log = log();
         let mut batch = WakerBatch::new();
-        for _ in 0..3 {
-            push_wakers(&mut batch, &log, STACK_SIZE);
-            assert!(batch.will_spill());
-            assert_eq!(batch.by_ref().count(), STACK_SIZE);
+        let mut expected = vec![];
+        for count in [STACK_SIZE + 8, STACK_SIZE, 1, 0, STACK_SIZE + 1] {
+            push_wakers(&mut batch, &log, count);
+            batch.wake_all();
+            expected.extend(0..count);
+
+            assert_eq!(woken(&log), expected);
             assert!(!batch.will_spill());
             assert_eq!(alive(&log), 0);
         }
     }
-
-    #[test]
-    fn keeps_push_order_while_spilled() {
-        let log = log();
-        let mut batch = WakerBatch::new();
-        push_wakers(&mut batch, &log, STACK_SIZE + 1);
-        // Free inline room; the spilled entry must still come out before anything pushed now.
-        for _ in 0..4 {
-            batch.next().unwrap().wake();
-        }
-        batch.push(waker(&log, 999));
-        assert!(batch.will_spill());
-
-        for waker in &mut batch {
-            waker.wake();
-        }
-
-        let mut expected = (0..STACK_SIZE + 1).collect::<Vec<_>>();
-        expected.push(999);
-        assert_eq!(woken(&log), expected);
-        assert_eq!(alive(&log), 0);
-    }
 }

asyncband/src/internal/wakerset.rs

@@ -35,11 +35,17 @@ use crate::internal::waker_batch::WakerBatch;
 pub struct WakerToken(SlotId);
 
 /// Cancellable waker storage without an implicit lifecycle or generation.
-#[derive(Default, Debug)]
+#[derive(Debug)]
 pub struct WakerSet {
     wakers: Arena<Waker>,
 }
 
+impl Default for WakerSet {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl WakerSet {
     /// Constructs an empty waker set.
     pub const fn new() -> Self {