commonware_runtime/iobuf/pool/class.rs
1//! Buffer-pool size classes and thread-local caching.
2//!
3//! A size class owns buffers of one fixed capacity and a shared global
4//! freelist. Each thread may cache a bounded number of buffers per size class,
5//! allocations check that cache before the global freelist, and returned
6//! buffers spill back to the global freelist when the cache is full or the
7//! thread exits.
8
9use super::Freelist;
10use crate::iobuf::owner::{PooledBuffer, PooledOwner};
11use std::{
12 cell::{Cell, UnsafeCell},
13 mem::MaybeUninit,
14 num::{NonZeroU32, NonZeroUsize},
15 ptr,
16};
17
18cfg_if::cfg_if! {
19 if #[cfg(feature = "loom")] {
20 use loom::sync::Arc;
21 } else {
22 use std::sync::Arc;
23 }
24}
25
26/// Minimum thread-local cache capacity required before refill/spill batches.
27///
28/// Below this threshold TLS still provides same-thread locality, but batching
29/// would degrade to single-buffer moves and add policy complexity without
30/// amortizing shared-queue traffic.
31const MIN_TLS_BATCH_CAPACITY: usize = 4;
32
33/// Per-size-class state.
34///
35/// Each class is a small two-level allocator:
36/// - a shared global freelist for tracked buffers visible to all threads
37/// - a per-thread local cache for same-thread reuse
38///
39/// The global freelist owns the allocation layout, slot reservation counter,
40/// and side-table slots for this class. A tracked buffer can be globally parked,
41/// owned by a pooled backing, or parked in one thread's local cache, but the
42/// slot always belongs to this `SizeClass`.
43///
44/// Liveness follows the buffer ownership state. Global freelist entries carry
45/// no per-buffer strong reference and rely on the pool's [`SizeClassHandle`]
46/// while the pool is alive. Pooled backing values and thread-local cache
47/// entries carry a live [`SizeClassLease`] in the pooled slot. Those
48/// non-global states are what allow a buffer to outlive the public
49/// [`super::BufferPool`] handle and still return to the correct freelist.
50///
51/// The freelist is the only place that deallocates tracked buffers. Returning a
52/// buffer to the freelist transfers buffer ownership back to that freelist and
53/// releases the slot lease that kept the class alive while the buffer was
54/// outside the global freelist.
55///
56/// Allocation prefers the local cache, then refills from the global freelist,
57/// and only creates a new tracked buffer when no free buffer is available and
58/// the class still has remaining capacity.
59pub(super) struct SizeClass {
60 /// Dense global identifier for the TLS cache registry.
61 class_id: usize,
62 /// The buffer size for this class.
63 size: usize,
64 /// Global free list of tracked buffers available for reuse.
65 global: Freelist,
66 /// Maximum number of buffers retained in the current thread's local bin.
67 thread_cache_capacity: usize,
68}
69
70// SAFETY: shared state in `SizeClass` is synchronized through atomics and the
71// global free set. Per-thread bins are stored in thread-local registries and only
72// accessed by the current thread.
73unsafe impl Send for SizeClass {}
74// SAFETY: see above.
75unsafe impl Sync for SizeClass {}
76
77/// Non-owning raw identity for a size class.
78///
79/// # Size-class lifetime model
80///
81/// A [`SizeClass`] owns the [`Freelist`] for one buffer size class. The
82/// freelist creates tracked [`PooledBuffer`]s, owns the allocation layout
83/// needed to deallocate them, and is the only place that releases their memory.
84/// A `PooledBuffer` outside the freelist does not carry enough information to
85/// deallocate itself, so it must keep its originating `SizeClass` alive until
86/// it can return to that freelist.
87///
88/// The pool has three buffer states, and those states determine where the
89/// strong size-class references live.
90///
91/// - Global freelist: the buffer is parked in [`SizeClass::global`] and carries
92/// no per-buffer strong reference. While the public pool exists, the
93/// [`SizeClassHandle`] in [`super::BufferPoolInner::classes`] keeps the class alive.
94/// - Pooled view: the buffer is owned by mutable or immutable I/O view state
95/// and carries one [`SizeClassLease`], which is one strong reference to the
96/// class.
97/// - Thread-local cache: each initialized [`TlsSizeClassCacheEntry`] owns a
98/// [`PooledBuffer`] whose side-table slot contains a live
99/// [`SizeClassLease`]. Increasing or decreasing the cache `len` moves entries
100/// into or out of the initialized prefix, but does not touch the `Arc` strong
101/// count.
102///
103/// Moving a buffer from the global freelist to pooled view or TLS state retains
104/// one class reference. Moving it back to the global freelist releases that
105/// reference. Moving between pooled view and TLS state transfers the same
106/// reference without touching the refcount:
107///
108/// ```text
109/// lease_into: retain class ref into slot lease
110/// +-------------------+ +-----------------+
111/// | parked in global | --------------------------> | checked out |
112/// | freelist | | (pooled view) |
113/// | (no per-buffer | +-----------------+
114/// | ref: the pool's | cache pop ^ | move buffer,
115/// | SizeClassHandle | | v lease stays
116/// | keeps class | +-----------------+
117/// | alive) | <-------------------------- | parked in TLS |
118/// +-------------------+ return_global[_batch]: | cache |
119/// park, THEN release lease +-----------------+
120/// ```
121///
122/// A checked-out buffer can also return directly to the global freelist
123/// (thread caching disabled, tiny-cache overflow, or TLS unavailable during
124/// thread teardown) through the same park-then-release transition.
125///
126/// Dropping the public [`super::BufferPool`] drains globally parked buffers, then
127/// drops its `SizeClassHandle`s. Pooled views and non-empty TLS caches may keep
128/// the `SizeClass` alive after that point. Empty TLS caches keep no size-class
129/// reference. A later return of an outstanding buffer can recreate the cache
130/// from the live lease in that buffer's slot.
131///
132/// This is the one raw pointer shape used by all pool-owned, pooled view, and
133/// thread-local references to a [`SizeClass`]. The pointer is always derived
134/// from [`Arc::into_raw`].
135///
136/// `SizeClassToken` itself owns nothing. It is only an identity token and raw
137/// pointer accepted by the `Arc` refcount APIs:
138/// - [`SizeClassHandle`] pairs a token with ownership of one strong reference.
139/// - [`SizeClassLease`] pairs a token with ownership of one strong reference.
140/// - [`TlsSizeClassCache`] stores entries whose pooled slots own strong
141/// references through live leases.
142///
143/// Because the token is non-owning, code may dereference it or adjust the
144/// strong count only when another invariant proves the allocation is still
145/// live. [`SizeClassHandle`] and [`SizeClassLease`] prove liveness through
146/// owned strong references. A non-empty [`TlsSizeClassCache`] proves liveness
147/// through the live leases stored in its entries' pooled slots.
148#[derive(Clone, Copy, Debug, PartialEq, Eq)]
149#[repr(transparent)]
150struct SizeClassToken {
151 ptr: ptr::NonNull<SizeClass>,
152}
153
154impl SizeClassToken {
155 /// Creates a raw token for a newly allocated size class.
156 ///
157 /// `Arc::into_raw` leaves one outstanding strong reference behind the
158 /// pointer. The returned token is still non-owning: the caller must
159 /// immediately place it in an owning wrapper, such as [`SizeClassHandle`],
160 /// or otherwise arrange for that strong reference to be released.
161 fn new(class: SizeClass) -> Self {
162 let ptr = Arc::into_raw(Arc::new(class)).cast_mut();
163 // SAFETY: `Arc::into_raw` never returns null.
164 let ptr = unsafe { ptr::NonNull::new_unchecked(ptr) };
165 Self { ptr }
166 }
167
168 /// Returns the referenced size class.
169 ///
170 /// # Safety
171 ///
172 /// Some owner must currently hold a strong reference for this token.
173 #[inline(always)]
174 const unsafe fn as_ref(&self) -> &SizeClass {
175 // SAFETY: guaranteed by the caller.
176 unsafe { self.ptr.as_ref() }
177 }
178
179 /// Retains one strong reference for this token.
180 ///
181 /// # Safety
182 ///
183 /// Some owner must currently hold a strong reference for this token.
184 #[inline(always)]
185 unsafe fn retain(self) {
186 // SAFETY: guaranteed by the caller.
187 unsafe { Arc::increment_strong_count(self.ptr.as_ptr()) };
188 }
189
190 /// Releases one owned strong reference for this token.
191 ///
192 /// # Safety
193 ///
194 /// The caller must own one strong reference represented by this token.
195 #[inline(always)]
196 unsafe fn release(self) {
197 // SAFETY: guaranteed by the caller.
198 unsafe { Arc::decrement_strong_count(self.ptr.as_ptr()) };
199 }
200}
201
202/// Owning pool reference to a size class.
203///
204/// This is the pool's strong `Arc<SizeClass>` reference represented by a
205/// [`SizeClassToken`]. `SizeClassHandle` is the long-lived owner for a class
206/// while the [`super::BufferPoolInner`] exists. Dropping the handle releases that
207/// pool-owned strong reference. A class may still outlive the handle if pooled
208/// backing values or thread-local cache entries own additional references
209/// through live [`SizeClassLease`] values in pooled slots.
210///
211/// Functionally this is an `Arc<SizeClass>` stored in raw-token form. It exists
212/// to keep the pool-owned reference alive and to provide a live token for
213/// allocation paths that need to retain pooled-slot lease references. The raw
214/// form keeps the already-loaded class pointer usable for
215/// explicit refcount operations without calling [`Arc::as_ptr`] or storing a
216/// second token alongside an `Arc`.
217#[repr(transparent)]
218pub(super) struct SizeClassHandle {
219 token: SizeClassToken,
220}
221
222// SAFETY: `SizeClassHandle` owns a strong reference to a `SizeClass`, which is
223// `Send`.
224unsafe impl Send for SizeClassHandle {}
225// SAFETY: same argument as `Send`, shared access to `SizeClass` is synchronized.
226unsafe impl Sync for SizeClassHandle {}
227
228impl SizeClassHandle {
229 /// Creates a new size class and takes ownership of its initial strong ref.
230 ///
231 /// If `prefill` is true, the global freelist creates `max` buffers upfront
232 /// and makes them immediately available for reuse.
233 pub(super) fn new(
234 class_id: usize,
235 size: usize,
236 alignment: usize,
237 max: NonZeroU32,
238 parallelism: NonZeroUsize,
239 thread_cache_capacity: usize,
240 prefill: bool,
241 ) -> Self {
242 let layout = PooledOwner::layout(size, alignment);
243 let freelist = Freelist::new(max, parallelism, layout, prefill);
244 let class = SizeClass {
245 class_id,
246 size,
247 global: freelist,
248 thread_cache_capacity,
249 };
250 Self {
251 token: SizeClassToken::new(class),
252 }
253 }
254
255 /// Creates a new tracked buffer and initializes its live lease.
256 #[inline(always)]
257 pub(super) fn try_create(&self, zeroed: bool) -> Option<PooledBuffer> {
258 let buffer = self.global.try_create(zeroed)?;
259 Some(self.lease_into(buffer))
260 }
261
262 /// Takes a parked buffer from the class-global freelist and installs its
263 /// live lease.
264 #[inline(always)]
265 fn take_global(&self) -> Option<PooledBuffer> {
266 let buffer = self.global.take()?;
267 Some(self.lease_into(buffer))
268 }
269
270 /// Installs a retained class reference as a buffer's live lease.
271 ///
272 /// This is the single transition from lease-free pool states (just
273 /// created, or parked in the global freelist) to checked-out state, so
274 /// the retain plus `init_lease` pairing cannot drift between call sites.
275 #[inline(always)]
276 fn lease_into(&self, mut buffer: PooledBuffer) -> PooledBuffer {
277 let lease = SizeClassLease::retain(self);
278 // SAFETY: freshly created buffers and buffers taken from the global
279 // freelist do not carry a live lease.
280 unsafe { buffer.init_lease(lease) };
281 buffer
282 }
283
284 /// Returns the allocation size represented by this class.
285 #[inline(always)]
286 pub(super) fn size(&self) -> usize {
287 self.size
288 }
289
290 /// Returns whether two handles refer to the same size class.
291 #[inline(always)]
292 pub(super) fn same_class(&self, other: &Self) -> bool {
293 self.token == other.token
294 }
295
296 /// Drains all buffers currently parked in the global freelist.
297 #[inline(always)]
298 pub(super) fn drain_global(&self) {
299 self.global.drain();
300 }
301}
302
303impl Clone for SizeClassHandle {
304 fn clone(&self) -> Self {
305 // SAFETY: this handle owns one strong reference for `self.token`, so
306 // the class is live and a new strong reference can be retained for
307 // the returned handle.
308 unsafe { self.token.retain() };
309 Self { token: self.token }
310 }
311}
312
313impl Drop for SizeClassHandle {
314 fn drop(&mut self) {
315 // SAFETY: this handle owns one strong reference for `self.token`.
316 unsafe { self.token.release() };
317 }
318}
319
320impl std::ops::Deref for SizeClassHandle {
321 type Target = SizeClass;
322
323 #[inline(always)]
324 fn deref(&self) -> &Self::Target {
325 // SAFETY: this handle owns one strong reference for `self.token`.
326 unsafe { self.token.as_ref() }
327 }
328}
329
330/// Owned size-class reference for a pooled buffer outside the global freelist.
331///
332/// A pooled buffer outside the global freelist must keep its originating
333/// [`SizeClass`] alive so it can be returned after the [`super::BufferPool`] handle is
334/// dropped. This is one strong `Arc<SizeClass>` reference represented by a
335/// [`SizeClassToken`], with retain and release performed explicitly at the
336/// boundaries where a buffer enters or leaves global pool state.
337///
338/// Lifetime-wise this is the same kind of reference as [`SizeClassHandle`]:
339/// both own exactly one strong reference for a token. The types are separate
340/// because they live in different state machines. `SizeClassHandle` is ordinary
341/// RAII ownership for the pool's class vector. `SizeClassLease` is hot-path
342/// pooled ownership that lives in the pooled slot while a buffer is
343/// checked out or parked in a thread-local cache. It must be explicitly
344/// returned to the global freelist when the buffer leaves local ownership.
345///
346/// The raw representation matters because the hot path mostly transfers
347/// ownership between pooled view state and this thread's local cache. A real
348/// `Arc<SizeClass>` field is pointer-sized too, but it is a non-`Copy` value
349/// with drop glue. Even when the strong count would not change, moving it
350/// through pooled buffer and cache-entry structs makes the compiler preserve
351/// destructor paths for those structs. `SizeClassLease` has no automatic drop:
352/// moving between checked-out view and local-cache state is just moving the
353/// pooled buffer whose slot contains the lease. Only explicit calls such as
354/// [`Self::return_global`] adjust the strong count.
355///
356/// A lease must be consumed when a buffer returns to the global freelist.
357/// Because this type intentionally has no `Drop` implementation, simply
358/// dropping a lease value would leak the strong reference. The hot local
359/// alloc/drop loop therefore keeps the lease in the slot and avoids moving a
360/// separate lease value at all.
361///
362/// Thread-local cache entries store a [`PooledBuffer`] whose slot lease stays
363/// live. Popping from the local cache hands that same live lease back to the
364/// caller without touching the strong count.
365///
366/// Globally parked buffers do not carry a class reference: taking from the
367/// global freelist retains the class, and returning to the global freelist
368/// releases it.
369#[must_use]
370pub(crate) struct SizeClassLease {
371 token: SizeClassToken,
372 class_id: usize,
373 thread_cache_capacity: usize,
374}
375
376// SAFETY: `SizeClassLease` owns one strong reference to a `SizeClass`, which is
377// `Send`.
378unsafe impl Send for SizeClassLease {}
379// SAFETY: same argument as `Send`, shared access to `SizeClass` is synchronized.
380unsafe impl Sync for SizeClassLease {}
381
382impl SizeClassLease {
383 /// Retains `class` for a buffer leaving the global freelist.
384 #[inline(always)]
385 fn retain(class: &SizeClassHandle) -> Self {
386 let token = class.token;
387 // The route is copied into the lease so buffer return can pick the
388 // thread-local cache without dereferencing the class object.
389 // SAFETY: the borrowed `class` owns one strong reference for `token`.
390 unsafe { token.retain() };
391 Self {
392 token,
393 class_id: class.class_id,
394 thread_cache_capacity: class.thread_cache_capacity,
395 }
396 }
397
398 /// Returns the TLS cache registry id for the owning size class.
399 #[inline(always)]
400 const fn class_id(&self) -> usize {
401 self.class_id
402 }
403
404 /// Returns the per-thread cache capacity for the owning size class.
405 #[inline(always)]
406 const fn thread_cache_capacity(&self) -> usize {
407 self.thread_cache_capacity
408 }
409
410 /// Returns the referenced size class.
411 ///
412 /// The token is valid because `SizeClassLease` owns one strong reference.
413 #[inline(always)]
414 const fn class(&self) -> &SizeClass {
415 // SAFETY: guaranteed by the ownership invariant documented on
416 // `SizeClassLease`.
417 unsafe { self.token.as_ref() }
418 }
419
420 /// Returns a buffer to this class's global freelist and releases the class
421 /// reference.
422 ///
423 /// The buffer is parked before the strong reference is released. If this is
424 /// the last outstanding reference after the public pool has been dropped,
425 /// dropping the `SizeClass` will then drain the just-parked buffer.
426 #[inline(always)]
427 fn return_global(self, buffer: PooledBuffer) {
428 // The lease proves this buffer was checked out from this class's
429 // freelist, and checked-out slots are not available in the freelist.
430 self.class().global.put(buffer);
431 // SAFETY: this lease owns one strong reference.
432 unsafe { self.token.release() };
433 }
434
435 /// Consumes the lease without releasing its strong reference.
436 ///
437 /// The returned token still represents one owned strong reference. The
438 /// caller takes over responsibility for releasing it. Batch returns use
439 /// this to park many buffers first and release their references together
440 /// afterwards.
441 #[inline(always)]
442 const fn into_token(self) -> SizeClassToken {
443 self.token
444 }
445}
446
447/// Free tracked buffer owned by a thread-local size-class cache.
448///
449/// This is allocator cache state, not a caller-visible pooled view. While an
450/// entry is held here, the buffer is owned by the current thread and is not
451/// visible to the class-global freelist.
452///
453/// The buffer's side-table slot identifies the stable slot within its
454/// [`SizeClass`] and contains the live lease that keeps that class alive. The
455/// entry itself intentionally stores only the buffer so local pop/push does
456/// not move separate slot or class metadata per buffer.
457struct TlsSizeClassCacheEntry {
458 buffer: PooledBuffer,
459}
460
461impl TlsSizeClassCacheEntry {
462 /// Returns this entry to its class-global freelist.
463 #[inline(always)]
464 fn return_global(mut self) {
465 // SAFETY: local cache entries keep a live lease in the pooled slot.
466 let lease = unsafe { self.buffer.take_lease() };
467 lease.return_global(self.buffer);
468 }
469}
470
471/// Per-thread cache for one size class's tracked buffers.
472///
473/// Each instance is stored in [`TlsSizeClassCaches`] under one global
474/// [`SizeClass::class_id`], so all entries in the cache belong to the same size
475/// class. The cache owns full [`PooledBuffer`] values while they are local.
476/// Interaction with the global freelist happens only on miss refill (take),
477/// overflow spill, explicit flush, or thread exit (return).
478///
479/// When `len > 0`, each initialized entry in `entries[..len]` owns one live
480/// slot lease, which keeps the pointed-to class alive. An empty cache owns no
481/// class reference. It is only an allocated local stack for a class id.
482///
483/// The hot steady-state allocation path pops an entry from `entries`, and the
484/// hot return path pushes one back while there is room.
485struct TlsSizeClassCache {
486 entries: Box<[MaybeUninit<TlsSizeClassCacheEntry>]>,
487 len: usize,
488 capacity: usize,
489}
490
491impl TlsSizeClassCache {
492 /// Creates a new empty cache with the given maximum thread-cache size.
493 fn new(capacity: usize) -> Self {
494 let entries = (0..capacity)
495 .map(|_| MaybeUninit::uninit())
496 .collect::<Vec<_>>()
497 .into_boxed_slice();
498 Self {
499 entries,
500 len: 0,
501 capacity,
502 }
503 }
504
505 /// Removes and returns one reusable buffer entry.
506 ///
507 /// Local hits are served directly from the cache. On a local miss, small
508 /// caches take only the buffer being returned to the caller. Larger caches
509 /// batch-take from the global freelist, return the first claimed buffer,
510 /// and retain the rest locally for future allocations.
511 ///
512 /// The returned entry carries a live lease in its pooled slot.
513 #[inline(always)]
514 fn pop(&mut self, class: &SizeClassHandle) -> Option<TlsSizeClassCacheEntry> {
515 if let Some(entry) = self.pop_local() {
516 return Some(entry);
517 }
518
519 // Take from the class-global freelist on a local miss.
520 self.pop_global(class)
521 }
522
523 /// Removes and returns one entry from this thread's local stack.
524 ///
525 /// This touches only thread-local cache state. A returned entry consumes
526 /// one live checked-out pooled buffer from this cache.
527 #[inline(always)]
528 fn pop_local(&mut self) -> Option<TlsSizeClassCacheEntry> {
529 if self.len == 0 {
530 return None;
531 }
532
533 self.len -= 1;
534 // SAFETY: entries in `0..self.len` are initialized. Decrementing `len`
535 // above makes this slot uninitialized again.
536 Some(unsafe { self.entries.get_unchecked(self.len).assume_init_read() })
537 }
538
539 /// Takes from the class-global freelist after the local stack misses.
540 ///
541 /// Every claimed global entry gets one retained class reference installed
542 /// as a live slot lease. The first claimed entry is returned to the
543 /// caller, and additional claimed entries are parked in this cache and
544 /// counted by `len`.
545 ///
546 /// This is separate from [`Self::pop`] so the steady-state allocation hot
547 /// path can inline only the local cache hit. We annotate with `inline(never)`
548 /// to keep the refill and batching code out of `BufferPoolInner::try_alloc`,
549 /// reducing hot-path code size and register pressure.
550 #[inline(never)]
551 fn pop_global(&mut self, class: &SizeClassHandle) -> Option<TlsSizeClassCacheEntry> {
552 // Tiny caches do not batch enough to justify the wider global claim.
553 // Keep their miss path equivalent to a single take.
554 if self.capacity < MIN_TLS_BATCH_CAPACITY {
555 return class
556 .take_global()
557 .map(|buffer| TlsSizeClassCacheEntry { buffer });
558 }
559
560 // Refill larger caches to half capacity. That leaves room for future
561 // same-thread returns while still amortizing the global stripe locks
562 // over several future local pops.
563 let mut entry = None;
564 let take = self.capacity / 2;
565 class.global.take_batch(take, |buffer| {
566 // Each claimed global entry becomes either the returned allocation
567 // or a local cache entry, so each needs one retained class
568 // reference stored in its slot lease.
569 let buffer = class.lease_into(buffer);
570 let cache_entry = TlsSizeClassCacheEntry { buffer };
571 if entry.is_none() {
572 // Hand the first claimed buffer to the allocation that missed
573 // locally. Additional claimed buffers refill the local cache.
574 entry = Some(cache_entry);
575 } else {
576 // The take count is derived from the target occupancy, so
577 // refill cannot overflow the local cache. Push directly to
578 // avoid the spill checks used by return-to-cache.
579 self.push_local(cache_entry);
580 }
581 });
582
583 entry
584 }
585
586 /// Pushes an entry into the local cache, spilling to global if full.
587 ///
588 /// Small local caches prioritize same-thread locality and route overflow
589 /// directly to the global freelist. Once the local cache is large enough to
590 /// batch effectively, half the entries are drained to amortize global queue
591 /// traffic across future returns.
592 #[inline(always)]
593 fn push(&mut self, buffer: PooledBuffer) {
594 let entry = TlsSizeClassCacheEntry { buffer };
595
596 if self.len < self.capacity {
597 // Keep the returned entry local while there is room.
598 self.push_local(entry);
599 return;
600 }
601
602 // Handle overflow when the local stack is full.
603 self.push_full(entry);
604 }
605
606 /// Pushes one entry onto this thread's local stack.
607 ///
608 /// The caller must ensure the stack has room.
609 #[inline(always)]
610 fn push_local(&mut self, entry: TlsSizeClassCacheEntry) {
611 // SAFETY: the caller ensured `self.len < self.capacity`, so this slot
612 // is in bounds and currently uninitialized.
613 unsafe {
614 self.entries.get_unchecked_mut(self.len).write(entry);
615 }
616 self.len += 1;
617 }
618
619 /// Handles a push after the local stack fills.
620 ///
621 /// Very small caches return the incoming entry directly to the global
622 /// freelist. Larger caches spill the top half of the local stack (the
623 /// most recently returned entries, which are contiguous and cheap to
624 /// drain without shifting the rest), then keep the incoming entry local
625 /// so the dropping thread retains the freshest buffer.
626 ///
627 /// This is separate from [`Self::push`] so the steady-state return hot path
628 /// can inline only the local cache push. We annotate with `inline(never)`
629 /// to keep the spill and batching code out of pooled buffer drop when the
630 /// local cache has room.
631 #[inline(never)]
632 fn push_full(&mut self, entry: TlsSizeClassCacheEntry) {
633 // Very small caches cannot spill enough entries to amortize a batch
634 // insert, so overflow goes straight to the global freelist.
635 if self.capacity < MIN_TLS_BATCH_CAPACITY {
636 entry.return_global();
637 return;
638 }
639
640 // Spill half the cache to global to make room.
641 let spill = self.len.min(self.capacity / 2).max(1);
642 let end = self.len;
643 let start = end - spill;
644 // Stop tracking slots before moving them out.
645 self.len = start;
646 self.return_global_batch(start, end);
647
648 // Keep the incoming entry local after making room.
649 self.push_local(entry);
650 }
651
652 /// Returns the initialized entries in `start..end` to their class-global
653 /// freelist as one batch.
654 ///
655 /// All entries in one cache belong to the same size class, so their slot
656 /// leases own strong references represented by one shared token. Each
657 /// lease is consumed without being released, then the whole batch parks
658 /// under its freelist stripe locks. The strong references are released
659 /// only after every buffer is parked. Parking before releasing matters: if
660 /// the public pool is already gone and these leases are the last
661 /// references, releasing first would drop the freelist before the buffers
662 /// returned to it.
663 ///
664 /// The caller must have already lowered `len` to at most `start`,
665 /// transferring ownership of the entries in `start..end` to this function.
666 #[inline(never)]
667 fn return_global_batch(&mut self, start: usize, end: usize) {
668 assert!(start < end && end <= self.capacity);
669 assert!(self.len <= start);
670 let count = end - start;
671 let entries = self.entries.as_mut_ptr();
672
673 // Read the shared token from the first entry's live lease. The
674 // not-yet-released lease references keep the class (and its freelist)
675 // alive until the releases below.
676 // SAFETY: `start..end` was initialized and ownership transferred to
677 // this function. The entry is only borrowed here.
678 let token = unsafe { (*entries.add(start)).assume_init_ref().buffer.lease() }.token;
679
680 // SAFETY: the lease strong references consumed below are not released
681 // until after the batch insert completes.
682 let class = unsafe { token.as_ref() };
683 let batch = (start..end).map(|index| {
684 // SAFETY: `start..end` was initialized before `len` was lowered.
685 // Reading moves each entry out and leaves the slot uninitialized.
686 let mut entry = unsafe { entries.add(index).read().assume_init() };
687 // SAFETY: local cache entries keep a live lease in the pooled
688 // slot. The strong reference it owns is intentionally not
689 // released here (leases have no drop glue). The token releases
690 // below settle it.
691 let _ = unsafe { entry.buffer.take_lease() }.into_token();
692 entry.buffer
693 });
694 // Cache entries are distinct checked-out buffers from this class, and
695 // the iterator body cannot panic after yielding an entry.
696 class.global.put_batch(batch);
697
698 // Release the strong references only now that every buffer is parked.
699 for _ in 0..count {
700 // SAFETY: each lease consumed above owned one strong reference
701 // that has not been released yet.
702 unsafe { token.release() };
703 }
704 }
705}
706
707impl Drop for TlsSizeClassCache {
708 fn drop(&mut self) {
709 if self.len == 0 {
710 return;
711 }
712
713 // Flush remaining entries (thread exit or explicit flush) with one
714 // coalesced batch insert.
715 let end = self.len;
716 // Stop tracking slots before moving them out.
717 self.len = 0;
718 self.return_global_batch(0, end);
719 }
720}
721
722/// Registry of one thread's per-size-class caches.
723///
724/// A [`super::BufferPool`] keeps its size classes in a vector, so allocation resolves
725/// a request to an index within that pool. Thread-local caches need a different
726/// key because a thread can use more than one pool. They use the process-global
727/// [`SizeClass::class_id`] assigned by [`super::NEXT_SIZE_CLASS_ID`], so index `0` in
728/// one pool cannot collide with index `0` in another pool.
729///
730/// The registry is a sparse vector indexed by `class_id`. Each initialized
731/// entry is a [`TlsSizeClassCache`] for that global size class. Missing entries
732/// mean this thread has not used that size class yet. Holes can remain for the
733/// lifetime of the thread because class ids are monotonic and never reused.
734/// Empty initialized caches can also remain after their pool has been dropped.
735/// They own no class reference while empty. If the class is still live because
736/// a pooled buffer is outstanding, a later return of that buffer to this same
737/// thread can use the buffer's live slot lease to make the cache usable
738/// again.
739///
740/// We intentionally use `Vec<Option<...>>` because class ids are dense enough
741/// for direct indexing to be cheaper than hashing, but a thread may initialize
742/// only a subset of live size classes. This keeps the TLS-hit path to a bounds
743/// check and an initialized-entry check, with no synchronization.
744struct TlsSizeClassCaches {
745 bins: Vec<Option<TlsSizeClassCache>>,
746}
747
748impl TlsSizeClassCaches {
749 /// Creates an empty registry.
750 const fn new() -> Self {
751 Self { bins: Vec::new() }
752 }
753
754 /// Returns the cache for the given class, creating it lazily on first use.
755 ///
756 /// The caller must provide a live class id from a [`SizeClassHandle`] or
757 /// [`SizeClassLease`]. A missing cache starts empty and owns no class
758 /// reference. The first local push or global refill stores entries whose
759 /// pooled slots contain live leases.
760 #[inline(always)]
761 fn get_or_init(&mut self, class_id: usize, capacity: usize) -> &mut TlsSizeClassCache {
762 // The initialized arm is defensively kept but not reachable today:
763 // callers route through this method only when the fast TLS pointer
764 // misses, and the fast pointer is published before any cache is
765 // created, so an existing cache is always found through the fast path.
766 if class_id < self.bins.len() && self.bins[class_id].is_some() {
767 return self.bins[class_id]
768 .as_mut()
769 .expect("class cache was checked as initialized");
770 }
771
772 self.init(class_id, capacity)
773 }
774
775 /// Initializes and returns the cache for `class_id`.
776 ///
777 /// This is separate from [`Self::get_or_init`] so the steady-state TLS hit
778 /// can inline only the existing-cache lookup. We annotate with
779 /// `inline(never)` to keep the resize and allocation path out of pooled
780 /// allocation and drop.
781 #[inline(never)]
782 fn init(&mut self, class_id: usize, capacity: usize) -> &mut TlsSizeClassCache {
783 if class_id >= self.bins.len() {
784 self.bins.resize_with(class_id + 1, || None);
785 }
786 self.bins[class_id].get_or_insert_with(|| TlsSizeClassCache::new(capacity))
787 }
788
789 /// Returns an initialized cache without creating a missing one.
790 #[inline(always)]
791 fn get(&mut self, class_id: usize) -> Option<&mut TlsSizeClassCache> {
792 self.bins.get_mut(class_id).and_then(Option::as_mut)
793 }
794}
795
796impl Drop for TlsSizeClassCaches {
797 fn drop(&mut self) {
798 // The registry lives only in `TLS_SIZE_CLASS_CACHES`' static storage
799 // (its const initializer is the sole constructor), and std destroys
800 // const-initialized TLS values in place, so a published fast pointer
801 // can only refer to this instance. Clear it unconditionally rather
802 // than comparing identities: a null fast pointer is always safe (the
803 // hot paths fall back to checked TLS access), while a stale one would
804 // be a use-after-destroy if std ever moved the value before dropping.
805 let this: *mut Self = self;
806 BufferPoolThreadCache::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| {
807 assert!(fast.get().is_null() || fast.get() == this);
808 fast.set(ptr::null_mut());
809 });
810 }
811}
812
813/// Access to the calling thread's local [`BufferPool`](super::BufferPool) caches.
814///
815/// This type hides the TLS layout used by pooled allocation and return. The
816/// main TLS key owns the registry. It has a destructor, so thread exit drops
817/// the registry and each `TlsSizeClassCache` flushes its remaining entries to
818/// the class-global freelist.
819///
820/// Steady-state allocation and return first read `TLS_SIZE_CLASS_CACHES_FAST`.
821/// If it points at this thread's registry and the requested class cache is
822/// initialized, the cache lookup itself touches only thread-local memory:
823/// local hits and non-spilling returns complete without shared state, while a
824/// local miss refills from the global freelist and a full cache spills to it.
825/// Missing TLS state routes through `cache_slow` or `push_slow`, which access
826/// the owning TLS key, install the fast pointer, and lazily initialize the
827/// class cache.
828///
829/// Rust's access path for TLS values with destructors includes checks for
830/// access during or after destruction. Those checks are correct, but they are
831/// expensive on the hot pooled allocation/drop path. After first checked
832/// access, we cache a raw pointer to the same registry in a destructor-free TLS
833/// key and use that pointer for steady-state access.
834///
835/// If the checked key is unavailable during thread-local destruction, cache
836/// access returns `None` and callers use the class-global freelist instead.
837pub struct BufferPoolThreadCache;
838
839impl BufferPoolThreadCache {
840 thread_local! {
841 // Owns this thread's cache registry and drops it during thread exit.
842 static TLS_SIZE_CLASS_CACHES: UnsafeCell<TlsSizeClassCaches> =
843 const { UnsafeCell::new(TlsSizeClassCaches::new()) };
844
845 // Performance-only pointer to the same registry. This key has no
846 // destructor, so the hot allocation/drop path avoids Rust's
847 // destructor-aware access path for `TLS_SIZE_CLASS_CACHES`.
848 static TLS_SIZE_CLASS_CACHES_FAST: Cell<*mut TlsSizeClassCaches> =
849 const { Cell::new(ptr::null_mut()) };
850 }
851
852 /// Flushes all local caches for the current thread into the global freelists.
853 pub fn flush() {
854 // If the owning TLS registry is unavailable during thread exit, this
855 // is a no-op. The registry's own drop path will flush any remaining
856 // entries.
857 let _ = Self::TLS_SIZE_CLASS_CACHES.try_with(|caches| {
858 // SAFETY: this TLS value is only ever accessed by the current thread.
859 let caches = unsafe { &mut *caches.get() };
860 for cache in caches.bins.iter_mut() {
861 let _ = cache.take();
862 }
863 });
864 }
865
866 /// Returns a buffer to the current thread's local cache for the given
867 /// size class, spilling to the global freelist if the cache is full.
868 ///
869 /// The hot path uses only an already-initialized cache from the fast TLS
870 /// pointer. If the fast pointer is missing, or this thread has not
871 /// initialized the size class yet, [`Self::push_slow`] performs the checked
872 /// TLS access and creates the local cache. The buffer's live slot lease
873 /// proves that initialization is safe. During thread-local teardown,
874 /// checked TLS access can fail, in that case the buffer falls back to the
875 /// global freelist.
876 ///
877 /// Cache routing reads `class_id` and `thread_cache_capacity` from the
878 /// live slot lease (on the slot line the release path has already loaded)
879 /// instead of dereferencing the class object, keeping the dependent-load
880 /// chain on the return fast path one level shorter.
881 #[inline(always)]
882 pub(in crate::iobuf) fn push(buffer: PooledBuffer) {
883 // SAFETY: pooled buffers entering the pool return path have an
884 // initialized live lease.
885 let lease = unsafe { buffer.lease() };
886 let class_id = lease.class_id();
887 let thread_cache_capacity = lease.thread_cache_capacity();
888 if thread_cache_capacity == 0 {
889 TlsSizeClassCacheEntry { buffer }.return_global();
890 return;
891 }
892
893 let caches = Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.get());
894 if !caches.is_null() {
895 // SAFETY: the fast pointer is set only from this thread's
896 // `TLS_SIZE_CLASS_CACHES` value and cleared before that value
897 // drops.
898 if let Some(cache) = unsafe { (&mut *caches).get(class_id) } {
899 cache.push(buffer);
900 return;
901 }
902 }
903
904 Self::push_slow(buffer);
905 }
906
907 /// Returns a buffer to the current thread's local cache after the fast
908 /// lookup misses.
909 ///
910 /// This is called when the fast TLS pointer is not initialized, or when
911 /// that pointer exists but this size class has no local cache yet. It
912 /// installs the fast TLS pointer after successfully accessing the owning
913 /// TLS key, then initializes the size-class cache if needed.
914 ///
915 /// This is separate from [`Self::push`] so the steady-state return hot path
916 /// only contains the initialized-cache lookup and local push.
917 #[inline(never)]
918 fn push_slow(buffer: PooledBuffer) {
919 // SAFETY: pooled buffers entering the pool return path have an
920 // initialized live lease.
921 let lease = unsafe { buffer.lease() };
922 let class_id = lease.class_id();
923 let thread_cache_capacity = lease.thread_cache_capacity();
924 // Returning a pooled buffer can happen from arbitrary Drop code,
925 // including during thread-local destruction. If the local cache is
926 // unavailable, fall back to the global freelist instead of panicking.
927 match Self::TLS_SIZE_CLASS_CACHES
928 .try_with(|caches| {
929 let caches = caches.get();
930
931 // Publish the checked owner TLS address to the fast key.
932 Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.set(caches));
933
934 // SAFETY: this TLS value is only ever accessed by the current thread.
935 ptr::NonNull::from(unsafe {
936 (&mut *caches).get_or_init(class_id, thread_cache_capacity)
937 })
938 })
939 .ok()
940 {
941 Some(mut cache) => {
942 // SAFETY: `cache` points to this thread's initialized TLS cache.
943 unsafe { cache.as_mut().push(buffer) };
944 }
945 None => TlsSizeClassCacheEntry { buffer }.return_global(),
946 }
947 }
948
949 /// Takes a buffer from the current thread's local cache for the given
950 /// size class, refilling from the global freelist if the cache is empty.
951 ///
952 /// The hot path uses only an already-initialized cache from the fast TLS
953 /// pointer. On a local miss, the global freelist is queried once. The first
954 /// claimed buffer is returned to the caller, and any additional claimed
955 /// buffers are appended directly to the local cache.
956 #[inline(always)]
957 pub(super) fn pop(class: &SizeClassHandle) -> Option<PooledBuffer> {
958 if class.thread_cache_capacity == 0 {
959 return class.take_global();
960 }
961
962 let caches = Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.get());
963 if !caches.is_null() {
964 // SAFETY: the fast pointer is set only from this thread's
965 // `TLS_SIZE_CLASS_CACHES` value and cleared before that value
966 // drops.
967 if let Some(cache) = unsafe { (&mut *caches).get(class.class_id) } {
968 return cache.pop(class).map(|entry| entry.buffer);
969 }
970 }
971
972 // Resolve the cache and fall back to the global freelist if
973 // unavailable.
974 let Some(mut cache) = Self::cache_slow(class) else {
975 return class.take_global();
976 };
977
978 // SAFETY: `cache` points to this thread's initialized TLS cache.
979 unsafe { cache.as_mut() }
980 .pop(class)
981 .map(|entry| entry.buffer)
982 }
983
984 /// Resolves the local cache after the fast TLS or class-cache lookup
985 /// misses.
986 ///
987 /// This is called when the fast TLS pointer is not initialized, or when
988 /// that pointer exists but this size class has no local cache yet. It
989 /// installs the fast TLS pointer after successfully accessing the owning
990 /// TLS key, then initializes the size-class cache if needed.
991 #[inline(never)]
992 fn cache_slow(class: &SizeClassHandle) -> Option<ptr::NonNull<TlsSizeClassCache>> {
993 // Allocation can happen from caller-owned TLS destructors during thread
994 // teardown. Return `None` instead of panicking if the owning TLS key is
995 // unavailable.
996 Self::TLS_SIZE_CLASS_CACHES
997 .try_with(|caches| {
998 let caches = caches.get();
999
1000 // Publish the checked owner TLS address to the fast key.
1001 Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.set(caches));
1002
1003 // SAFETY: this TLS value is only ever accessed by the current thread.
1004 ptr::NonNull::from(unsafe {
1005 (&mut *caches).get_or_init(class.class_id, class.thread_cache_capacity)
1006 })
1007 })
1008 .ok()
1009 }
1010}
1011
1012#[cfg(all(test, not(feature = "loom")))]
1013pub(super) mod tests {
1014 use super::{
1015 super::{BufferPool, BufferPoolConfig, NEXT_SIZE_CLASS_ID},
1016 *,
1017 };
1018 use crate::{
1019 iobuf::{IoBuf, page_size},
1020 telemetry::metrics::Registry,
1021 };
1022 use bytes::BufMut;
1023 use commonware_utils::{NZU32, NZUsize};
1024 use std::{
1025 cell::Cell,
1026 sync::{Arc, atomic::Ordering, mpsc},
1027 thread,
1028 };
1029
1030 fn test_size_class(size: usize, alignment: usize) -> SizeClassHandle {
1031 SizeClassHandle::new(
1032 NEXT_SIZE_CLASS_ID.fetch_add(1, Ordering::Relaxed),
1033 size,
1034 alignment,
1035 NZU32!(8),
1036 NZUsize!(4),
1037 4,
1038 false,
1039 )
1040 }
1041
1042 fn test_pool(config: BufferPoolConfig) -> BufferPool {
1043 let mut registry = Registry::default();
1044 BufferPool::new(config, &mut registry)
1045 }
1046
1047 fn test_config(min_size: usize, max_size: usize, max_per_class: u32) -> BufferPoolConfig {
1048 BufferPoolConfig::for_network()
1049 .with_pool_min_size(0)
1050 .with_size_class_range(
1051 NZUsize!(min_size),
1052 NZUsize!(max_size),
1053 NZU32!(max_per_class),
1054 )
1055 .with_alignment(NZUsize!(page_size()))
1056 }
1057
1058 /// Returns the current strong count without changing it after the helper
1059 /// returns.
1060 fn size_class_strong_count(class: &SizeClassHandle) -> usize {
1061 // SAFETY: the borrowed handle owns one strong reference for `class.token`
1062 // for the duration of this call.
1063 unsafe { class.token.retain() };
1064 // SAFETY: the increment above created the strong reference consumed by
1065 // this temporary Arc.
1066 let arc = unsafe { Arc::from_raw(class.token.ptr.as_ptr()) };
1067 Arc::strong_count(&arc) - 1
1068 }
1069
1070 fn get_available(pool: &BufferPool, size: usize) -> i64 {
1071 let class_index = pool.class_index(size).unwrap();
1072 let class = &pool.inner.classes[class_index];
1073 (get_global_len(class) + get_local_len(class)) as i64
1074 }
1075
1076 /// Returns the configured per-thread cache capacity.
1077 pub const fn get_thread_cache_capacity(class: &SizeClass) -> usize {
1078 class.thread_cache_capacity
1079 }
1080
1081 /// Helper to get the number of free buffers parked in the global freelist.
1082 pub fn get_global_len(class: &SizeClass) -> usize {
1083 super::super::freelist::tests::len(&class.global)
1084 }
1085
1086 /// Helper to get the number of buffers created by the global freelist.
1087 pub fn get_global_created(class: &SizeClass) -> usize {
1088 super::super::freelist::tests::created(&class.global)
1089 }
1090
1091 /// Returns the number of global freelist stripes for tests.
1092 pub fn get_global_num_stripes(class: &SizeClass) -> usize {
1093 super::super::freelist::tests::num_stripes(&class.global)
1094 }
1095
1096 /// Helper to get the number of free buffers parked in the current thread's
1097 /// local cache for a size class.
1098 pub fn get_local_len(class: &SizeClass) -> usize {
1099 BufferPoolThreadCache::TLS_SIZE_CLASS_CACHES.with(|caches| {
1100 // SAFETY: this TLS value is only ever accessed by the current thread.
1101 let caches = unsafe { &*caches.get() };
1102 caches
1103 .bins
1104 .get(class.class_id)
1105 .and_then(Option::as_ref)
1106 .map_or(0, |cache| cache.len)
1107 })
1108 }
1109
1110 #[test]
1111 fn test_thread_cache_flush_moves_local_entries_to_global() {
1112 let page = page_size();
1113 let pool =
1114 test_pool(test_config(page, page * 2, 8).with_max_thread_cache_capacity(NZUsize!(4)));
1115
1116 // Use two distinct size classes so the test exercises the whole TLS
1117 // registry, not just a single per-class cache entry.
1118 let small_index = pool.class_index(page).unwrap();
1119 let large_index = pool.class_index(page + 1).unwrap();
1120 let small_class = &pool.inner.classes[small_index];
1121 let large_class = &pool.inner.classes[large_index];
1122
1123 // Return one buffer from each class to the current thread. With local
1124 // caching enabled, both drops should stay in the thread-local bins.
1125 let small = pool.try_alloc(page).expect("tracked allocation");
1126 let large = pool.try_alloc(page + 1).expect("tracked allocation");
1127 drop(small);
1128 drop(large);
1129
1130 // Before flushing, both buffers are only visible via the current
1131 // thread's local caches, nothing has been pushed to the global queues.
1132 assert_eq!(get_local_len(small_class), 1);
1133 assert_eq!(get_local_len(large_class), 1);
1134 assert_eq!(get_global_len(small_class), 0);
1135 assert_eq!(get_global_len(large_class), 0);
1136
1137 // Flushing should walk the entire TLS registry, drop every local cache,
1138 // and let each cache's drop implementation return its buffers to the
1139 // shared global freelists.
1140 BufferPoolThreadCache::flush();
1141
1142 // After flush, the current thread retains nothing locally and both
1143 // buffers are once again visible through their class-global queues.
1144 assert_eq!(get_local_len(small_class), 0);
1145 assert_eq!(get_local_len(large_class), 0);
1146 assert_eq!(get_global_len(small_class), 1);
1147 assert_eq!(get_global_len(large_class), 1);
1148 }
1149
1150 #[test]
1151 fn test_return_buffer_local_overflow_spills_to_global() {
1152 let page = page_size();
1153 let pool = test_pool(test_config(page, page, 2));
1154 let class_index = pool
1155 .class_index(page)
1156 .expect("class exists for page-sized buffer");
1157
1158 let tracked1 = pool.try_alloc(page).expect("first tracked allocation");
1159 let tracked2 = pool.try_alloc(page).expect("second tracked allocation");
1160
1161 // The first return should stay entirely in the current thread's local cache.
1162 drop(tracked1);
1163 assert_eq!(get_global_len(&pool.inner.classes[class_index]), 0);
1164 assert_eq!(get_local_len(&pool.inner.classes[class_index]), 1);
1165
1166 // Returning another tracked buffer should route overflow to the global
1167 // freelist and retain one in the current thread's local bin.
1168 drop(tracked2);
1169 assert_eq!(get_global_len(&pool.inner.classes[class_index]), 1);
1170 assert_eq!(get_local_len(&pool.inner.classes[class_index]), 1);
1171 assert_eq!(get_available(&pool, page), 2);
1172 }
1173
1174 #[test]
1175 fn test_small_local_cache_overflow_preserves_locality() {
1176 let page = page_size();
1177 let pool = test_pool(test_config(page, page, 2));
1178
1179 // With `thread_cache_capacity == 1`, the first return stays local and the
1180 // second overflows directly to global instead of spilling the hot
1181 // local entry through the shared queue.
1182 let mut tracked1 = pool.try_alloc(page).expect("first tracked allocation");
1183 let ptr1 = tracked1.as_mut_ptr();
1184 let mut tracked2 = pool.try_alloc(page).expect("second tracked allocation");
1185 let ptr2 = tracked2.as_mut_ptr();
1186
1187 drop(tracked1);
1188 drop(tracked2);
1189
1190 let mut reused_local = pool.try_alloc(page).expect("reuse from local cache");
1191 assert_eq!(reused_local.as_mut_ptr(), ptr1);
1192
1193 let mut reused_global = pool.try_alloc(page).expect("reuse from global freelist");
1194 assert_eq!(reused_global.as_mut_ptr(), ptr2);
1195 }
1196
1197 #[test]
1198 fn test_large_local_cache_batches_overflow_and_refill() {
1199 let page = page_size();
1200 let threads = std::thread::available_parallelism().map_or(1, NonZeroUsize::get);
1201 let max_per_class =
1202 u32::try_from(threads * 8).expect("test capacity must fit in u32 slot ids");
1203 let pool = test_pool(test_config(page, page, max_per_class));
1204 let class_index = pool
1205 .class_index(page)
1206 .expect("class exists for page-sized buffer");
1207 let class = &pool.inner.classes[class_index];
1208
1209 assert!(class.thread_cache_capacity >= MIN_TLS_BATCH_CAPACITY);
1210
1211 // Drop enough distinct pooled buffers to force an overflow from a
1212 // full local cache. Large bins should spill half the entries to global
1213 // and keep the remainder local for fast same-thread reuse.
1214 let mut bufs = Vec::new();
1215 for _ in 0..class.thread_cache_capacity + 1 {
1216 bufs.push(pool.try_alloc(page).expect("tracked allocation"));
1217 }
1218 for buf in bufs {
1219 drop(buf);
1220 }
1221
1222 assert_eq!(get_local_len(class), class.thread_cache_capacity / 2 + 1);
1223 assert_eq!(get_global_len(class), class.thread_cache_capacity / 2);
1224
1225 // Drain the local half, then hit global once. That global take should
1226 // batch-refill the local cache back up to the configured target.
1227 let mut reused = Vec::new();
1228 for _ in 0..class.thread_cache_capacity / 2 + 1 {
1229 reused.push(pool.try_alloc(page).expect("local reuse"));
1230 }
1231 assert_eq!(get_local_len(class), 0);
1232 assert_eq!(get_global_len(class), class.thread_cache_capacity / 2);
1233
1234 let _global = pool.try_alloc(page).expect("global reuse with refill");
1235 assert_eq!(get_local_len(class), class.thread_cache_capacity / 2 - 1);
1236 assert_eq!(get_global_len(class), 0);
1237 }
1238
1239 #[test]
1240 fn test_global_batch_alloc_stops_when_global_runs_empty() {
1241 let class = test_size_class(64, 64);
1242 let buffer = class.global.try_create(false).expect("slot reservation");
1243
1244 // A short global freelist should return the allocation and stop
1245 // without filling the local cache to its batch target.
1246 class.global.put(buffer);
1247 let buffer = BufferPoolThreadCache::pop(&class).expect("global allocation");
1248
1249 assert_eq!(get_local_len(&class), 0);
1250 assert_eq!(get_global_len(&class), 0);
1251
1252 // Return the manually popped entry so the freelist owns and deallocates
1253 // the buffer at test teardown.
1254 TlsSizeClassCacheEntry { buffer }.return_global();
1255 }
1256
1257 #[test]
1258 fn test_size_class_leases_use_raw_arc_tokens_across_cache_paths() {
1259 let class = test_size_class(64, 64);
1260 let mut cache = TlsSizeClassCache::new(MIN_TLS_BATCH_CAPACITY);
1261 assert_eq!(size_class_strong_count(&class), 1);
1262
1263 let mut buffer = class.global.try_create(false).expect("slot reservation");
1264 let lease = SizeClassLease::retain(&class);
1265 // SAFETY: this buffer was just created and has no live lease.
1266 unsafe { buffer.init_lease(lease) };
1267 assert_eq!(size_class_strong_count(&class), 2);
1268
1269 // Moving a live pooled buffer into the local cache keeps the same
1270 // strong reference in the header, it should not clone the class.
1271 cache.push(buffer);
1272 assert_eq!(size_class_strong_count(&class), 2);
1273
1274 let entry = cache.pop(&class).expect("local cache pop");
1275 assert_eq!(size_class_strong_count(&class), 2);
1276 entry.return_global();
1277 assert_eq!(size_class_strong_count(&class), 1);
1278
1279 for _ in 0..2 {
1280 let buffer = class.global.try_create(false).expect("slot reservation");
1281 class.global.put(buffer);
1282 }
1283
1284 let entry = cache.pop(&class).expect("global refill");
1285 assert_eq!(size_class_strong_count(&class), 3);
1286
1287 entry.return_global();
1288 assert_eq!(size_class_strong_count(&class), 2);
1289
1290 // Dropping the cache returns the live refill entry and releases its
1291 // size-class reference.
1292 drop(cache);
1293 assert_eq!(size_class_strong_count(&class), 1);
1294 }
1295
1296 #[test]
1297 fn test_tls_size_class_cache_push_tolerates_empty_spill() {
1298 let class = test_size_class(64, 64);
1299 let mut buffer = class.global.try_create(false).expect("slot reservation");
1300 let lease = SizeClassLease::retain(&class);
1301 // SAFETY: this buffer was just created and has no live lease.
1302 unsafe { buffer.init_lease(lease) };
1303 let mut cache = TlsSizeClassCache::new(0);
1304
1305 // Small local capacities should bypass batching and push straight to
1306 // global. The retained reference above is represented by the live
1307 // slot lease and transferred into `cache.push`.
1308 cache.push(buffer);
1309 assert_eq!(cache.len, 0);
1310 drop(cache);
1311 }
1312
1313 #[test]
1314 fn test_global_freelist_returns_each_slot_once() {
1315 // Use a two-slot class with TLS capacity one so this test can exercise
1316 // the class-global freelist directly without involving local-cache
1317 // refill or spill behavior.
1318 let class = SizeClassHandle::new(
1319 NEXT_SIZE_CLASS_ID.fetch_add(1, Ordering::Relaxed),
1320 64,
1321 64,
1322 NZU32!(2),
1323 NZUsize!(1),
1324 1,
1325 false,
1326 );
1327
1328 // Create both slot ids and keep each allocation's pointer so we can
1329 // verify that the freelist returns the same buffer parked for that slot.
1330 let buffer0 = class.global.try_create(false).expect("first slot");
1331 let slot0 = buffer0.slot();
1332 let ptr0 = buffer0.as_ptr();
1333 let buffer1 = class.global.try_create(false).expect("second slot");
1334 let slot1 = buffer1.slot();
1335 let ptr1 = buffer1.as_ptr();
1336 let mut expected = [(slot0, ptr0), (slot1, ptr1)];
1337 expected.sort_by_key(|(slot, _)| *slot);
1338
1339 class.global.put(buffer0);
1340 class.global.put(buffer1);
1341
1342 // The freelist does not preserve insertion order, so normalize by slot
1343 // before asserting identity. The important property is that each slot is
1344 // returned exactly once with its original parked buffer.
1345 let mut popped = [
1346 class.global.take().expect("first pop"),
1347 class.global.take().expect("second pop"),
1348 ];
1349 popped.sort_by_key(PooledBuffer::slot);
1350
1351 assert_eq!(popped[0].slot(), expected[0].0);
1352 assert_eq!(popped[0].as_ptr(), expected[0].1);
1353 assert_eq!(popped[1].slot(), expected[1].0);
1354 assert_eq!(popped[1].as_ptr(), expected[1].1);
1355
1356 // Both slots were claimed above, so the global freelist is empty.
1357 assert!(class.global.take().is_none());
1358
1359 // Return the buffers so the freelist owns and deallocates them when the
1360 // test size class is dropped.
1361 for buffer in popped {
1362 class.global.put(buffer);
1363 }
1364 }
1365
1366 #[test]
1367 fn test_thread_exit_flushes_local_bin() {
1368 // When a thread exits, its TLS cache Drop flushes buffers back to the
1369 // global freelist, making them available to other threads.
1370 let page = page_size();
1371 let pool = Arc::new(test_pool(test_config(page, page, 1)));
1372
1373 // Allocate and return a buffer on a worker thread, then let it exit.
1374 let worker_pool = pool.clone();
1375 thread::spawn(move || {
1376 let buf = worker_pool
1377 .try_alloc(page)
1378 .expect("worker should allocate tracked buffer");
1379 drop(buf);
1380 })
1381 .join()
1382 .expect("worker thread should exit cleanly");
1383
1384 // After thread exit, the buffer should be in the global freelist (not
1385 // stuck in a dead thread's local cache).
1386 let class_index = pool
1387 .class_index(page)
1388 .expect("class exists for page-sized buffer");
1389 assert_eq!(get_global_len(&pool.inner.classes[class_index]), 1);
1390 assert_eq!(get_local_len(&pool.inner.classes[class_index]), 0);
1391
1392 // The flushed buffer should be reusable from the main thread.
1393 let _buf = pool
1394 .try_alloc(page)
1395 .expect("thread-exited local buffer should be reusable");
1396 }
1397
1398 #[test]
1399 fn test_thread_exit_batch_flush_outlives_pool() {
1400 // The batch return path (TlsSizeClassCache::drop -> return_global_batch)
1401 // must park every buffer before releasing the lease references. Drop
1402 // the pool while a worker's TLS cache holds several entries so the
1403 // flush's lease releases are the last strong references: the final
1404 // release drops the SizeClass, whose freelist must reclaim the
1405 // just-parked buffers.
1406 let page = page_size();
1407 let pool = test_pool(test_config(page, page, 8));
1408
1409 let (cached_tx, cached_rx) = mpsc::channel();
1410 let (release_tx, release_rx) = mpsc::channel::<()>();
1411 let worker_pool = pool.clone();
1412 let handle = thread::spawn(move || {
1413 let class_index = worker_pool
1414 .class_index(page)
1415 .expect("class exists for page-sized buffer");
1416 let class = &worker_pool.inner.classes[class_index];
1417 assert!(class.thread_cache_capacity >= MIN_TLS_BATCH_CAPACITY);
1418
1419 // Fill this thread's local cache so the exit flush takes the
1420 // multi-entry batch path.
1421 let bufs = (0..MIN_TLS_BATCH_CAPACITY)
1422 .map(|_| worker_pool.try_alloc(page).expect("tracked allocation"))
1423 .collect::<Vec<_>>();
1424 drop(bufs);
1425 assert_eq!(get_local_len(class), MIN_TLS_BATCH_CAPACITY);
1426
1427 drop(worker_pool);
1428 cached_tx.send(()).expect("signal cached buffers");
1429 release_rx.recv().expect("wait for pool drop");
1430 });
1431
1432 cached_rx.recv().expect("worker cached buffers");
1433 // Every pool handle is gone before the worker exits, so the worker's
1434 // cached leases are the only remaining size-class references.
1435 drop(pool);
1436 release_tx.send(()).expect("release worker");
1437 handle.join().expect("worker thread should exit cleanly");
1438 }
1439
1440 #[test]
1441 fn test_pooled_ops_inside_tls_destructor_fall_back_to_global() {
1442 // Pooled operations that run inside another thread_local's destructor
1443 // may find the pool's TLS registry already destroyed. The push/pop
1444 // slow paths must then fall back to the global freelist instead of
1445 // panicking or stranding buffers. Destructor order is
1446 // platform-dependent, so this test asserts the outcome invariant
1447 // (a clean exit with every buffer reusable) rather than the path.
1448 struct ExitReleaser {
1449 pool: BufferPool,
1450 size: usize,
1451 held: Option<IoBuf>,
1452 }
1453
1454 impl Drop for ExitReleaser {
1455 fn drop(&mut self) {
1456 // An allocation inside a TLS destructor exercises the pop
1457 // fallback. The drops exercise the push fallback.
1458 let extra = self
1459 .pool
1460 .try_alloc(self.size)
1461 .expect("pool must serve allocations from TLS destructors");
1462 drop(extra);
1463 drop(self.held.take());
1464 }
1465 }
1466
1467 thread_local! {
1468 static EXIT_RELEASER: Cell<Option<ExitReleaser>> = const { Cell::new(None) };
1469 }
1470
1471 let page = page_size();
1472 let pool = test_pool(test_config(page, page, 4));
1473
1474 thread::spawn({
1475 let pool = pool.clone();
1476 move || {
1477 // Register the releaser's thread_local before first touching
1478 // the pool: destructors run in reverse registration order on
1479 // the platforms we target, so the releaser's pooled operations
1480 // run after the pool's TLS registry is gone (on platforms with
1481 // a different order the outcome invariant still holds).
1482 EXIT_RELEASER.with(|cell| {
1483 cell.set(Some(ExitReleaser {
1484 pool: pool.clone(),
1485 size: page,
1486 held: None,
1487 }));
1488 });
1489
1490 drop(pool.try_alloc(page).expect("first allocation"));
1491 let mut buf = pool.try_alloc(page).expect("second allocation");
1492 buf.put_u8(1);
1493 let held = buf.freeze();
1494 EXIT_RELEASER.with(|cell| {
1495 let mut releaser = cell.take().expect("releaser installed above");
1496 releaser.held = Some(held);
1497 cell.set(Some(releaser));
1498 });
1499 }
1500 })
1501 .join()
1502 .expect("worker thread must exit cleanly");
1503
1504 // Nothing may remain in the dead thread's cache, and every buffer the
1505 // worker touched must be reusable: with a class capacity of four, all
1506 // four allocations succeed only if none were stranded.
1507 let class_index = pool
1508 .class_index(page)
1509 .expect("class exists for page-sized buffer");
1510 assert_eq!(get_local_len(&pool.inner.classes[class_index]), 0);
1511 let bufs = (0..4)
1512 .map(|i| {
1513 pool.try_alloc(page)
1514 .unwrap_or_else(|_| panic!("buffer {i} was stranded at thread exit"))
1515 })
1516 .collect::<Vec<_>>();
1517 drop(bufs);
1518 }
1519
1520 #[test]
1521 fn test_pool_drop_drains_global_freelist() {
1522 // Dropping the pool should immediately reclaim globally-visible free
1523 // tracked buffers, while leaving TLS-cached buffers alone.
1524 let page = page_size();
1525 let pool = test_pool(test_config(page, page, 2));
1526 let class_index = pool
1527 .class_index(page)
1528 .expect("class exists for page-sized buffer");
1529 let class = &pool.inner.classes[class_index];
1530 // Keep a test-owned handle so the class remains inspectable after
1531 // dropping the public pool below.
1532 // SAFETY: `class` owns one strong reference for `class.token`.
1533 unsafe { class.token.retain() };
1534 let class = SizeClassHandle { token: class.token };
1535
1536 // Return one buffer to the current thread's local cache and overflow
1537 // the other into the shared global freelist.
1538 let buf1 = pool.try_alloc(page).unwrap();
1539 let buf2 = pool.try_alloc(page).unwrap();
1540 drop(buf1);
1541 drop(buf2);
1542
1543 assert_eq!(get_global_len(&class), 1);
1544 assert_eq!(get_local_len(&class), 1);
1545
1546 // Pool drop should drain only the global freelist. The thread-local
1547 // cache remains untouched until thread exit.
1548 drop(pool);
1549
1550 assert_eq!(get_global_len(&class), 0);
1551 assert_eq!(get_local_len(&class), 1);
1552 assert_eq!(get_global_created(&class), 2);
1553 }
1554}
1555
1556#[cfg(all(test, feature = "loom"))]
1557mod loom_tests {
1558 use super::*;
1559 use commonware_utils::{NZU32, NZUsize};
1560 use loom::thread;
1561
1562 // Models the multi-entry TLS teardown edge without using OS thread-local
1563 // state, which loom cannot reset between model executions. Dropping the
1564 // production cache takes each real pooled-slot lease, batch-parks both
1565 // slots, and only then releases the lease references. The final class-handle
1566 // release races that batch return, so either side may perform the final
1567 // SizeClass drop and reclaim the parked buffers.
1568 #[test]
1569 fn tls_batch_drop_races_pool_teardown() {
1570 loom::model(|| {
1571 let class = SizeClassHandle::new(1, 64, 64, NZU32!(2), NZUsize!(1), 2, false);
1572 let mut cache = TlsSizeClassCache::new(2);
1573
1574 for _ in 0..2 {
1575 let buffer = class.try_create(false).expect("tracked slot");
1576 cache.push(buffer);
1577 }
1578 assert_eq!(cache.len, 2);
1579
1580 let t = thread::spawn(move || drop(cache));
1581 drop(class);
1582 t.join().unwrap();
1583 });
1584 }
1585}