diff --git a/CHANGELOG.md b/CHANGELOG.md index c51ffff9..72937cc9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,7 @@ All notable changes to this project will be documented in this file. ### Improvements +* Keep broadcast reclaim scans proportional to retained receivers after a tail cohort is dropped instead of the channel's historical receiver peak. * Reduce fan-out notification overhead by avoiding heap allocation for a single waiter and transferring terminal waiter storage out of state locks. * Reduce MPSC receiver registration and wake latency by storing receiver wakers inline instead of allocating them on the heap. * Reduce semaphore and mutex hot-path overhead by avoiding wake-buffer allocation when no tasks are queued and batching queued wakes on the stack. diff --git a/asyncband/src/internal/arena.rs b/asyncband/src/internal/arena.rs index 290a32cd..ffddbb53 100644 --- a/asyncband/src/internal/arena.rs +++ b/asyncband/src/internal/arena.rs @@ -18,6 +18,8 @@ use std::mem; use std::num::NonZeroUsize; +const NO_VACANT_SLOT: usize = usize::MAX; + /// Identifies a reusable slot while that slot is occupied. /// /// The non-zero representation lets wrappers such as `WaiterId` retain a niche when stored in an @@ -50,11 +52,13 @@ impl std::fmt::Debug for SlotId { /// /// The occupied length equals the number of `Occupied` slots. Every `Vacant` slot appears exactly /// once in the singly linked vacant list, which starts at `next_vacant` and terminates at -/// `slots.len()`. Removing a value makes its slot ID available for immediate reuse. +/// [`NO_VACANT_SLOT`]. Removing a value makes its slot ID available for immediate reuse. Vacant +/// slots at the logical tail are removed when they are reachable from the head of the vacant list; +/// this shortens later scans without releasing the vector's allocation. #[derive(Debug)] pub struct Arena { slots: Vec>, - /// The next reusable slot, or `slots.len()` when every slot is occupied. + /// The next reusable slot, or [`NO_VACANT_SLOT`] when every slot is occupied. next_vacant: usize, len: usize, } @@ -69,7 +73,7 @@ impl Arena { pub const fn new() -> Self { Self { slots: vec![], - next_vacant: 0, + next_vacant: NO_VACANT_SLOT, len: 0, } } @@ -77,19 +81,20 @@ impl Arena { pub fn with_capacity(capacity: usize) -> Self { Self { slots: Vec::with_capacity(capacity), - next_vacant: 0, + next_vacant: NO_VACANT_SLOT, len: 0, } } pub fn insert(&mut self, value: T) -> SlotId { - let index = self.next_vacant; self.len += 1; - if index == self.slots.len() { + let index = if self.next_vacant == NO_VACANT_SLOT { + let index = self.slots.len(); self.slots.push(Slot::Occupied(value)); - self.next_vacant = index + 1; + index } else { + let index = self.next_vacant; self.next_vacant = match self.slots.get(index) { Some(Slot::Vacant { next }) => *next, Some(Slot::Occupied(_)) | None => { @@ -97,7 +102,8 @@ impl Arena { } }; self.slots[index] = Slot::Occupied(value); - } + index + }; SlotId::from_index(index) } @@ -140,6 +146,19 @@ impl Arena { #[track_caller] pub fn remove(&mut self, id: SlotId) -> T { let index = id.index(); + if index + 1 == self.slots.len() { + let value = match self.slots.pop().expect("arena slot ID must be in bounds") { + Slot::Occupied(value) => value, + vacant @ Slot::Vacant { .. } => { + self.slots.push(vacant); + panic!("arena slot ID must be occupied"); + } + }; + self.len -= 1; + self.trim_vacant_tail(); + return value; + } + let slot = self .slots .get_mut(index) @@ -161,6 +180,19 @@ impl Arena { value } + fn trim_vacant_tail(&mut self) { + while self.next_vacant != NO_VACANT_SLOT && self.next_vacant + 1 == self.slots.len() { + let Slot::Vacant { next } = self + .slots + .pop() + .expect("vacant-list head must refer to a slot") + else { + unreachable!("arena free list must point to a vacant slot") + }; + self.next_vacant = next; + } + } + /// 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 @@ -168,7 +200,7 @@ impl Arena { /// their own epoch check. #[inline] pub fn drain(&mut self) -> impl Iterator + '_ { - self.next_vacant = 0; + self.next_vacant = NO_VACANT_SLOT; self.len = 0; self.slots.drain(..).filter_map(|slot| match slot { Slot::Occupied(value) => Some(value), @@ -179,7 +211,7 @@ impl Arena { /// Takes every occupied value and the backing allocation in slot order. #[inline] pub fn take_all(&mut self) -> impl Iterator + use { - self.next_vacant = 0; + self.next_vacant = NO_VACANT_SLOT; self.len = 0; mem::take(&mut self.slots) .into_iter() @@ -213,6 +245,47 @@ mod tests { assert_eq!(arena.get(second), Some(&"second")); } + #[test] + fn removing_tail_slots_shortens_the_logical_storage() { + let mut arena = Arena::with_capacity(4); + let ids = [ + arena.insert(0), + arena.insert(1), + arena.insert(2), + arena.insert(3), + ]; + let capacity = arena.slots.capacity(); + + arena.remove(ids[2]); + assert_eq!(arena.slots.len(), 4); + arena.remove(ids[3]); + + assert_eq!(arena.slots.len(), 2); + assert_eq!(arena.slots.capacity(), capacity); + assert_eq!(arena.values().copied().collect::>(), vec![0, 1]); + } + + #[test] + fn tail_trimming_preserves_the_remaining_vacant_list() { + let mut arena = Arena::new(); + let ids = [ + arena.insert(0), + arena.insert(1), + arena.insert(2), + arena.insert(3), + arena.insert(4), + arena.insert(5), + ]; + + arena.remove(ids[4]); + arena.remove(ids[2]); + arena.remove(ids[5]); + + assert_eq!(arena.insert(20), ids[2]); + assert_eq!(arena.insert(40), ids[4]); + assert_eq!(arena.insert(50), ids[5]); + } + #[test] fn drain_restarts_slot_id_allocation() { let mut arena = Arena::with_capacity(3); diff --git a/benchmarks/asyncband/broadcast/mpmc/unbounded.rs b/benchmarks/asyncband/broadcast/mpmc/unbounded.rs index fa35c27e..86b5d4b0 100644 --- a/benchmarks/asyncband/broadcast/mpmc/unbounded.rs +++ b/benchmarks/asyncband/broadcast/mpmc/unbounded.rs @@ -40,9 +40,8 @@ const CONCURRENT_BATCH_SIZE: usize = 4096; /// A channel that peaked at `peak` receivers and currently has `live` of them. /// -/// The two are measured separately because a dropped receiver leaves its slot behind: the reclaim -/// scan walks every slot the channel ever handed out, so a channel that shed receivers keeps -/// paying for the peak. Pairing each peak with a drained arena is what makes that visible. +/// The two are measured separately to verify that dropping a tail cohort shortens later reclaim +/// scans instead of making a channel that shed receivers keep paying for its historical peak. #[derive(Clone, Copy)] struct Fanout { peak: usize, @@ -257,7 +256,7 @@ fn send_and_try_recv_owned_shared(bencher: Bencher) { } // Measures the reclaim scan, which runs when the slowest cursor advances. Comparing a peak against -// the same peak drained down to fewer receivers shows what the slots left behind still cost. +// the same peak drained down to fewer receivers shows whether tail slot trimming tracks live state. #[divan::bench(args = RECLAIM_FANOUTS)] fn drain_with_receivers(bencher: Bencher, fanout: Fanout) { let (sender, receiver) = mpmc::unbounded();