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/10/02 13:36:31 UTC
(asyncband) branch fixup created (now 91c5b15)
This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a change to branch fixup
in repository https://gitbox.apache.org/repos/asf/asyncband.git
at 91c5b15 chore: fine tune waker wake
This branch includes the following new commits:
new 91c5b15 chore: fine tune waker wake
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@asyncband.apache.org
For additional commands, e-mail: commits-help@asyncband.apache.org
(asyncband) 01/01: chore: fine tune waker wake
Posted by ti...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch fixup
in repository https://gitbox.apache.org/repos/asf/asyncband.git
commit 91c5b151ece66b2815ff84dfd6b8ee1c9b6d549b
Author: tison <wa...@gmail.com>
AuthorDate: Fri Oct 2 21:35:19 2026 +0800
chore: fine tune waker wake
Signed-off-by: tison <wa...@gmail.com>
---
asyncband/src/barrier/mod.rs | 6 +-
asyncband/src/broadcast/mpmc/bounded/mod.rs | 42 ++++++--------
asyncband/src/broadcast/mpmc/unbounded/mod.rs | 14 ++---
asyncband/src/condvar/mod.rs | 2 +-
asyncband/src/event/manual_reset.rs | 3 +-
asyncband/src/internal/arena.rs | 17 +-----
asyncband/src/internal/semaphore.rs | 2 +-
asyncband/src/internal/wakerset.rs | 15 +----
asyncband/src/phaser/mod.rs | 83 ++++++++++++---------------
asyncband/src/watch/mod.rs | 54 ++++++++---------
10 files changed, 92 insertions(+), 146 deletions(-)
diff --git a/asyncband/src/barrier/mod.rs b/asyncband/src/barrier/mod.rs
index b3a0637..c5760c9 100644
--- a/asyncband/src/barrier/mod.rs
+++ b/asyncband/src/barrier/mod.rs
@@ -55,7 +55,6 @@ use std::task::Poll;
use crate::internal::mutex::Mutex;
use crate::internal::wake_all;
-use crate::internal::waker_batch::WakerBatch;
use crate::internal::wakerset::WakerSet;
use crate::internal::wakerset::WakerToken;
@@ -189,10 +188,9 @@ impl Barrier {
if state.arrived == self.n {
state.arrived = 0;
state.generation += 1;
- let mut wakers = WakerBatch::new();
- state.waiters.drain_into(&mut wakers);
+ let wakers = state.waiters.take_all();
drop(state);
- wake_all(&mut wakers);
+ wake_all(wakers);
return BarrierWaitResult(true);
}
diff --git a/asyncband/src/broadcast/mpmc/bounded/mod.rs b/asyncband/src/broadcast/mpmc/bounded/mod.rs
index 5cc64ed..a187551 100644
--- a/asyncband/src/broadcast/mpmc/bounded/mod.rs
+++ b/asyncband/src/broadcast/mpmc/bounded/mod.rs
@@ -120,7 +120,6 @@ use crate::internal::mutex::Mutex;
use crate::internal::semaphore::Acquire;
use crate::internal::semaphore::Semaphore;
use crate::internal::wake_all;
-use crate::internal::waker_batch::WakerBatch;
use crate::internal::wakerset::WakerToken;
#[cfg(test)]
@@ -212,9 +211,9 @@ impl<T> Shared<T> {
// backlog, which would otherwise leave nearly `capacity` permits behind.
//
// Capping cannot lose a wake-up, by the same argument that lets this read the count at all:
- // a producer this load observes is one the release covers, and one it misses incremented
- // after the load, which it does before taking the channel lock to recheck — so its recheck
- // runs after the reclaim and finds the capacity itself.
+ // a producer this load observes is the one the release covers, and one it misses
+ // incremented after the load, which it does before taking the channel lock to recheck — so
+ // its recheck runs after the reclaim and finds the capacity itself.
let waiting = self.waiting_senders();
if freed > 0 && waiting > 0 {
self.tx_permits.release_if_nonempty(freed.min(waiting));
@@ -429,7 +428,7 @@ impl<T> BoundedSender<T> {
self.publish(msg, |msg| msg)
}
- /// The publish step both send paths share.
+ /// The publishing step both send paths share.
///
/// `into_msg` is called only once this decides the message will actually be retained, which is
/// what lets `try_send` defer its allocation past the capacity check while `try_publish` hands
@@ -439,26 +438,21 @@ impl<T> BoundedSender<T> {
/// observe an empty buffer and park after this message became visible.
fn publish<P>(&self, payload: P, into_msg: impl FnOnce(P) -> Arc<T>) -> Result<(), P> {
let mut discarded = None;
- let mut wakers = WakerBatch::new();
- {
- let mut inner = self.shared.inner.lock();
-
- if !inner.log.has_receivers() {
- // Nothing can read this message. The payload leaves the critical section with us
- // and is dropped below, so `T::drop` never runs under the lock.
- inner.log.publish_discarded();
- discarded = Some(payload);
- } else if inner.log.retained() == self.shared.capacity {
- // Nothing was published, so there is no wait set to drain.
- return Err(payload);
- } else {
- inner.log.publish_retained(into_msg(payload));
- }
-
- inner.waiters.drain_into(&mut wakers);
+ let mut inner = self.shared.inner.lock();
+ if !inner.log.has_receivers() {
+ // Nothing can read this message. The payload leaves the critical section with us
+ // and is dropped below, so `T::drop` never runs under the lock.
+ inner.log.publish_discarded();
+ discarded = Some(payload);
+ } else if inner.log.retained() == self.shared.capacity {
+ // Nothing was published, so there is no wait set to drain.
+ return Err(payload);
+ } else {
+ inner.log.publish_retained(into_msg(payload));
}
-
- wake_all(&mut wakers);
+ let wakers = inner.waiters.take_all();
+ drop(inner);
+ wake_all(wakers);
drop(discarded);
Ok(())
}
diff --git a/asyncband/src/broadcast/mpmc/unbounded/mod.rs b/asyncband/src/broadcast/mpmc/unbounded/mod.rs
index 663e5a6..19422b9 100644
--- a/asyncband/src/broadcast/mpmc/unbounded/mod.rs
+++ b/asyncband/src/broadcast/mpmc/unbounded/mod.rs
@@ -71,7 +71,6 @@ use super::error::TryRecvError;
use crate::internal::arena::SlotId;
use crate::internal::mutex::Mutex;
use crate::internal::wake_all;
-use crate::internal::waker_batch::WakerBatch;
use crate::internal::wakerset::WakerToken;
#[cfg(test)]
@@ -177,17 +176,14 @@ impl<T> UnboundedSender<T> {
// Publishing and draining the wait set share one critical section, so a receiver can never
// observe an empty buffer and park after this message became visible.
- let mut wakers = WakerBatch::new();
- let unretained = {
- let mut inner = self.shared.inner.lock();
- let unretained = inner.log.publish(msg);
- inner.waiters.drain_into(&mut wakers);
- unretained
- };
+ let mut inner = self.shared.inner.lock();
+ let unretained = inner.log.publish(msg);
+ let wakers = inner.waiters.take_all();
+ drop(inner);
// Notify all waiting receivers. An unsent message is dropped here too, once the lock is
// released.
- wake_all(&mut wakers);
+ wake_all(wakers);
drop(unretained);
}
diff --git a/asyncband/src/condvar/mod.rs b/asyncband/src/condvar/mod.rs
index f2b0652..47567d3 100644
--- a/asyncband/src/condvar/mod.rs
+++ b/asyncband/src/condvar/mod.rs
@@ -175,7 +175,7 @@ impl Condvar {
{}
}
- wake_all(&mut wakers);
+ wake_all(wakers);
}
/// Waits for a notification, atomically releasing and then reacquiring the mutex.
diff --git a/asyncband/src/event/manual_reset.rs b/asyncband/src/event/manual_reset.rs
index 4e0e2ce..d2b0bd4 100644
--- a/asyncband/src/event/manual_reset.rs
+++ b/asyncband/src/event/manual_reset.rs
@@ -117,8 +117,7 @@ impl ManualResetEvent {
}
}
}
-
- wake_all(&mut wakers);
+ wake_all(wakers);
}
/// Clears the set state.
diff --git a/asyncband/src/internal/arena.rs b/asyncband/src/internal/arena.rs
index ce0bc7b..fcd11de 100644
--- a/asyncband/src/internal/arena.rs
+++ b/asyncband/src/internal/arena.rs
@@ -166,21 +166,6 @@ impl<T> Arena<T> {
value
}
- /// Drains every occupied value in slot order while retaining the allocation for reuse.
- ///
- /// After a non-empty drain, every previously issued slot ID becomes invalid, including IDs for
- /// slots that were already vacant. Consumers that retain IDs across this operation must supply
- /// their own epoch check.
- #[inline]
- pub fn drain(&mut self) -> impl Iterator<Item = T> + '_ {
- self.vacant_head = None;
- self.len = 0;
- self.slots.drain(..).filter_map(|slot| match slot {
- Slot::Occupied(value) => Some(value),
- Slot::Vacant { .. } => None,
- })
- }
-
/// Takes every occupied value and the backing allocation in slot order.
#[inline]
pub fn take_all(&mut self) -> impl Iterator<Item = T> + use<T> {
@@ -227,7 +212,7 @@ mod tests {
let capacity = arena.slots.capacity();
arena.remove(second);
- assert_eq!(arena.drain().collect::<Vec<_>>(), vec![1, 3]);
+ assert_eq!(arena.take_all().collect::<Vec<_>>(), vec![1, 3]);
assert_eq!(arena.len(), 0);
assert_eq!(arena.slots.capacity(), capacity);
diff --git a/asyncband/src/internal/semaphore.rs b/asyncband/src/internal/semaphore.rs
index 51c7bae..065d2f0 100644
--- a/asyncband/src/internal/semaphore.rs
+++ b/asyncband/src/internal/semaphore.rs
@@ -173,7 +173,7 @@ impl Semaphore {
}
}
drop(waiters);
- wake_all(&mut wakers);
+ wake_all(wakers);
}
fn insert_permits_with_lock<'a>(
diff --git a/asyncband/src/internal/wakerset.rs b/asyncband/src/internal/wakerset.rs
index c74a89b..e6cc16f 100644
--- a/asyncband/src/internal/wakerset.rs
+++ b/asyncband/src/internal/wakerset.rs
@@ -26,13 +26,11 @@ use std::task::Waker;
use crate::internal::arena::Arena;
use crate::internal::arena::SlotId;
-use crate::internal::waker_batch::WakerBatch;
/// An exclusive handle to one waker slot in a [`WakerSet`].
///
/// This token deliberately does not implement `Clone` or `Copy`. Its owner must not pass it back
-/// to the set after the registration has been detached by [`WakerSet::drain_into`] or
-/// [`WakerSet::take_all`].
+/// to the set after the registration has been detached by [`WakerSet::take_all`].
#[derive(Debug)]
pub struct WakerToken(SlotId);
@@ -57,17 +55,6 @@ impl WakerSet {
}
}
- /// Drains all registered wakers into `batch` while retaining slot capacity.
- ///
- /// The batch is filled in place because its inline storage is too large to move for free:
- /// returning it by value costs every publish about 6ns even when nothing is registered. The
- /// caller must invalidate every outstanding token and consume or drop the batch after
- /// releasing the lock that protects this set.
- #[inline]
- pub fn drain_into(&mut self, batch: &mut WakerBatch) {
- batch.extend(self.wakers.drain());
- }
-
/// Takes all registered wakers together with the set's backing allocation.
///
/// The caller must invalidate every outstanding token and consume or drop the iterator after
diff --git a/asyncband/src/phaser/mod.rs b/asyncband/src/phaser/mod.rs
index 9b00e7c..78f2be7 100644
--- a/asyncband/src/phaser/mod.rs
+++ b/asyncband/src/phaser/mod.rs
@@ -131,7 +131,6 @@ use std::task::Poll;
use crate::internal::mutex::Mutex;
use crate::internal::wake_all;
-use crate::internal::waker_batch::WakerBatch;
use crate::internal::wakerset::WakerSet;
use crate::internal::wakerset::WakerToken;
@@ -172,14 +171,13 @@ struct State {
}
impl State {
- /// Completes the phase once every participant has arrived, moving its waiters into `wakers`.
- fn advance_if_ready(&mut self, wakers: &mut WakerBatch) {
+ /// Completes the phase once every participant has arrived.
+ fn advance_if_ready(&mut self) {
if self.closed || self.unarrived != 0 {
return;
}
self.phase = self.phase.wrapping_add(1);
self.unarrived = self.registered;
- self.waiters.drain_into(wakers);
}
fn completion(&self, observed: u64) -> Poll<Result<u64, Closed>> {
@@ -385,16 +383,15 @@ impl Drop for PhaserParticipants {
if self.remaining == 0 {
return;
}
- let mut wakers = WakerBatch::new();
- {
- let mut state = self.phaser.state.lock();
- // Unyielded participants have never arrived and prevent their phase from advancing.
- state.registered -= self.remaining;
- state.unarrived -= self.remaining;
- self.remaining = 0;
- state.advance_if_ready(&mut wakers);
- }
- wake_all(&mut wakers);
+ let mut state = self.phaser.state.lock();
+ // Unyielded participants have never arrived and prevent their phase from advancing.
+ state.registered -= self.remaining;
+ state.unarrived -= self.remaining;
+ self.remaining = 0;
+ state.advance_if_ready();
+ let wakers = state.waiters.take_all();
+ drop(state);
+ wake_all(wakers);
}
}
@@ -433,21 +430,19 @@ impl PhaserParticipant {
/// Repeated calls within one phase count only once. After advancement, an explicit new call
/// arrives in the new phase and replaces any previous pending observation.
pub fn arrive(&mut self) -> Result<u64, Closed> {
- let mut wakers = WakerBatch::new();
- let phase = {
- let mut state = self.phaser.state.lock();
- if state.closed {
- return Err(Closed(()));
- }
- let phase = state.phase;
- if self.pending != Some(phase) {
- state.unarrived -= 1;
- }
- self.pending = Some(phase);
- state.advance_if_ready(&mut wakers);
- phase
- };
- wake_all(&mut wakers);
+ let mut state = self.phaser.state.lock();
+ if state.closed {
+ return Err(Closed(()));
+ }
+ let phase = state.phase;
+ if self.pending != Some(phase) {
+ state.unarrived -= 1;
+ }
+ self.pending = Some(phase);
+ state.advance_if_ready();
+ let wakers = state.waiters.take_all();
+ drop(state);
+ wake_all(wakers);
Ok(phase)
}
@@ -482,23 +477,21 @@ impl PhaserParticipant {
}
fn do_deregister(&mut self) -> Result<u64, Closed> {
- let mut wakers = WakerBatch::new();
- let result = {
- let mut state = self.phaser.state.lock();
- self.registered = false;
- state.registered -= 1;
- if self.pending != Some(state.phase) {
- state.unarrived -= 1;
- }
- let result = if state.closed {
- Err(Closed(()))
- } else {
- Ok(state.phase)
- };
- state.advance_if_ready(&mut wakers);
- result
+ let mut state = self.phaser.state.lock();
+ self.registered = false;
+ state.registered -= 1;
+ if self.pending != Some(state.phase) {
+ state.unarrived -= 1;
+ }
+ let result = if state.closed {
+ Err(Closed(()))
+ } else {
+ Ok(state.phase)
};
- wake_all(&mut wakers);
+ state.advance_if_ready();
+ let wakers = state.waiters.take_all();
+ drop(state);
+ wake_all(wakers);
result
}
}
diff --git a/asyncband/src/watch/mod.rs b/asyncband/src/watch/mod.rs
index 90b7192..48d3513 100644
--- a/asyncband/src/watch/mod.rs
+++ b/asyncband/src/watch/mod.rs
@@ -75,7 +75,6 @@ pub use self::error::RecvError;
pub use self::error::SendError;
use crate::internal::mutex::Mutex;
use crate::internal::wake_all;
-use crate::internal::waker_batch::WakerBatch;
use crate::internal::wakerset::WakerSet;
use crate::internal::wakerset::WakerToken;
@@ -169,23 +168,21 @@ impl<T> Sender<T> {
///
/// Panics if the channel has already published `u64::MAX` updates.
pub fn send(&self, value: T) -> Result<(), SendError<T>> {
- let mut wakers = WakerBatch::new();
- let replaced = {
- let mut state = self.shared.state.lock();
- if state.receivers == 0 {
- return Err(SendError::new(value));
- }
- let version = state
- .version
- .checked_add(1)
- .expect("watch channel version counter overflowed");
- let replaced = mem::replace(&mut state.value, value);
- state.version = version;
- state.waiters.drain_into(&mut wakers);
- replaced
- };
+ let mut state = self.shared.state.lock();
+ if state.receivers == 0 {
+ return Err(SendError::new(value));
+ }
+ let version = state
+ .version
+ .checked_add(1)
+ .expect("watch channel version counter overflowed");
+ let replaced = mem::replace(&mut state.value, value);
+ state.version = version;
+ let wakers = state.waiters.take_all();
+ drop(state);
+
// Waker callbacks and the replaced value's destructor may reenter this channel.
- wake_all(&mut wakers);
+ wake_all(wakers);
drop(replaced);
Ok(())
}
@@ -199,19 +196,16 @@ impl<T> Sender<T> {
///
/// Panics if the channel has already published `u64::MAX` updates.
pub fn send_replace(&self, value: T) -> T {
- let mut wakers = WakerBatch::new();
- let replaced = {
- let mut state = self.shared.state.lock();
- let version = state
- .version
- .checked_add(1)
- .expect("watch channel version counter overflowed");
- let replaced = mem::replace(&mut state.value, value);
- state.version = version;
- state.waiters.drain_into(&mut wakers);
- replaced
- };
- wake_all(&mut wakers);
+ let mut state = self.shared.state.lock();
+ let version = state
+ .version
+ .checked_add(1)
+ .expect("watch channel version counter overflowed");
+ let replaced = mem::replace(&mut state.value, value);
+ state.version = version;
+ let wakers = state.waiters.take_all();
+ drop(state);
+ wake_all(wakers);
replaced
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@asyncband.apache.org
For additional commands, e-mail: commits-help@asyncband.apache.org