Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ All notable changes to this project will be documented in this file.

* Reduce intermediate payload copies in bounded MPSC receives, especially for large inline messages.
* Transfer reclaimed broadcast head messages directly to receivers, avoiding an extra payload `Arc` clone and unnecessary blocked-sender bookkeeping.
* Reduce the cost of acquiring and releasing an uncontended `Mutex` by avoiding its internal wait-queue lock.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Locking takes the same path as before. Only unlocking skips the queue lock now.

Suggested change
* Reduce the cost of acquiring and releasing an uncontended `Mutex` by avoiding its internal wait-queue lock.
* Reduce the cost of unlocking an uncontended `Mutex` by avoiding its internal wait-queue lock.

* Reduce the cost of initializing an uncontended `OnceCell` by avoiding its internal wait-queue lock.
* Reduce the cost of polling a pending oneshot receive again from the same task, which now keeps its registered waker instead of cloning and dropping it.
* Reduce the state kept by futures waiting for a `Mutex`, `RwLock`, `Semaphore`, `Once`, `OnceCell`, or bounded broadcast capacity, making lock, permit acquisition, and bounded broadcast send futures smaller.
Expand Down
29 changes: 0 additions & 29 deletions asyncband/src/internal/semaphore.rs
Original file line number Diff line number Diff line change
Expand Up @@ -164,17 +164,6 @@ impl Semaphore {
self.distribute_permits(n, self.waiters.lock());
}

/// Releases `n` permits by acquiring the queue lock directly.
///
/// Use this when no permits can be available, as when a mutex guard holds its only permit,
/// to skip the speculative attempt in [`release`](Self::release).
/// This method also works when permits are available.
pub fn release_with_lock(&self, n: usize) {
if n != 0 {
self.distribute_permits(n, self.waiters.lock());
}
}

/// Releases `n` permits only if the queue contains waiters when locked.
#[cfg(feature = "broadcast")]
pub fn release_if_waiters(&self, n: usize) {
Expand Down Expand Up @@ -533,24 +522,6 @@ mod tests {
assert!(semaphore.waiters.lock().is_empty());
}

#[test]
fn release_with_lock_hands_permits_to_waiters() {
let semaphore = Semaphore::new(1);
let counter = Arc::new(WakeCounter(AtomicUsize::new(0)));
let waker = Waker::from(counter.clone());
assert!(semaphore.try_acquire(1));
let mut acquire = semaphore.acquire(1);
assert!(acquire.poll_once(&waker).is_pending());

semaphore.release_with_lock(1);
assert_eq!(counter.0.load(Ordering::Relaxed), 1);
assert_eq!(semaphore.available_permits(), 0);
assert!(acquire.poll_once(&waker).is_ready());

semaphore.release_with_lock(1);
assert_eq!(semaphore.available_permits(), 1);
}

#[test]
fn release_distributes_permits_across_multiple_batches() {
const WAITER_COUNT: usize = WakerBatch::STACK_SIZE * 2 + 1;
Expand Down
20 changes: 11 additions & 9 deletions asyncband/src/mutex/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,9 @@ impl<T> Mutex<T> {
/// let mutex = Mutex::new(5);
/// ```
pub const fn new(t: T) -> Self {
let s = semaphore::Semaphore::new(1);
// Each lock takes two of three permits, so only one can run at a time. The spare
// permit lets an uncontended release avoid locking the semaphore's wait queue.
let s = semaphore::Semaphore::new(3);
let c = UnsafeCell::new(t);
Self { s, c }
}
Expand Down Expand Up @@ -162,7 +164,7 @@ impl<T: ?Sized> Mutex<T> {
/// # }
/// ```
pub async fn lock(&self) -> MutexGuard<'_, T> {
self.s.acquire(1).await;
self.s.acquire(2).await;
MutexGuard { lock: self }
}

Expand All @@ -179,7 +181,7 @@ impl<T: ?Sized> Mutex<T> {
/// assert_eq!(2, *guard);
/// ```
pub fn try_lock(&self) -> Option<MutexGuard<'_, T>> {
if self.s.try_acquire(1) {
if self.s.try_acquire(2) {
let guard = MutexGuard { lock: self };
Some(guard)
} else {
Expand Down Expand Up @@ -212,7 +214,7 @@ impl<T: ?Sized> Mutex<T> {
/// # }
/// ```
pub async fn lock_owned(self: Arc<Self>) -> OwnedMutexGuard<T> {
self.s.acquire(1).await;
self.s.acquire(2).await;
OwnedMutexGuard { lock: self }
}

Expand All @@ -233,7 +235,7 @@ impl<T: ?Sized> Mutex<T> {
/// assert_eq!(2, *guard);
/// ```
pub fn try_lock_owned(self: Arc<Self>) -> Option<OwnedMutexGuard<T>> {
if self.s.try_acquire(1) {
if self.s.try_acquire(2) {
let guard = OwnedMutexGuard { lock: self };
Some(guard)
} else {
Expand Down Expand Up @@ -278,7 +280,7 @@ unsafe impl<T: ?Sized + Send + Sync> Sync for MutexGuard<'_, T> {}

impl<T: ?Sized> Drop for MutexGuard<'_, T> {
fn drop(&mut self) {
self.lock.s.release_with_lock(1);
self.lock.s.release(2);
}
}

Expand Down Expand Up @@ -445,7 +447,7 @@ unsafe impl<T: ?Sized + Send + Sync> Sync for OwnedMutexGuard<T> {}

impl<T: ?Sized> Drop for OwnedMutexGuard<T> {
fn drop(&mut self) {
self.lock.s.release_with_lock(1);
self.lock.s.release(2);
}
}

Expand Down Expand Up @@ -641,7 +643,7 @@ unsafe impl<T: ?Sized + Sync> Sync for MappedMutexGuard<'_, T> {}

impl<T: ?Sized> Drop for MappedMutexGuard<'_, T> {
fn drop(&mut self) {
self.s.release_with_lock(1);
self.s.release(2);
}
}

Expand Down Expand Up @@ -840,7 +842,7 @@ unsafe impl<T: ?Sized + Send + Sync, U: ?Sized + Send + Sync> Sync for OwnedMapp

impl<T: ?Sized, U: ?Sized> Drop for OwnedMappedMutexGuard<T, U> {
fn drop(&mut self) {
self.lock.s.release_with_lock(1);
self.lock.s.release(2);
}
}

Expand Down