use super::Freelist;
use crate::iobuf::owner::{PooledBuffer, PooledOwner};
use std::{
cell::{Cell, UnsafeCell},
mem::MaybeUninit,
num::{NonZeroU32, NonZeroUsize},
ptr,
};
cfg_if::cfg_if! {
if #[cfg(feature = "loom")] {
use loom::sync::Arc;
} else {
use std::sync::Arc;
}
}
const MIN_TLS_BATCH_CAPACITY: usize = 4;
pub(super) struct SizeClass {
class_id: usize,
size: usize,
global: Freelist,
thread_cache_capacity: usize,
}
unsafe impl Send for SizeClass {}
unsafe impl Sync for SizeClass {}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[repr(transparent)]
struct SizeClassToken {
ptr: ptr::NonNull<SizeClass>,
}
impl SizeClassToken {
fn new(class: SizeClass) -> Self {
let ptr = Arc::into_raw(Arc::new(class)).cast_mut();
let ptr = unsafe { ptr::NonNull::new_unchecked(ptr) };
Self { ptr }
}
#[inline(always)]
const unsafe fn as_ref(&self) -> &SizeClass {
unsafe { self.ptr.as_ref() }
}
#[inline(always)]
unsafe fn retain(self) {
unsafe { Arc::increment_strong_count(self.ptr.as_ptr()) };
}
#[inline(always)]
unsafe fn release(self) {
unsafe { Arc::decrement_strong_count(self.ptr.as_ptr()) };
}
}
#[repr(transparent)]
pub(super) struct SizeClassHandle {
token: SizeClassToken,
}
unsafe impl Send for SizeClassHandle {}
unsafe impl Sync for SizeClassHandle {}
impl SizeClassHandle {
pub(super) fn new(
class_id: usize,
size: usize,
alignment: usize,
max: NonZeroU32,
parallelism: NonZeroUsize,
thread_cache_capacity: usize,
prefill: bool,
) -> Self {
let layout = PooledOwner::layout(size, alignment);
let freelist = Freelist::new(max, parallelism, layout, prefill);
let class = SizeClass {
class_id,
size,
global: freelist,
thread_cache_capacity,
};
Self {
token: SizeClassToken::new(class),
}
}
#[inline(always)]
pub(super) fn try_create(&self, zeroed: bool) -> Option<PooledBuffer> {
let buffer = self.global.try_create(zeroed)?;
Some(self.lease_into(buffer))
}
#[inline(always)]
fn take_global(&self) -> Option<PooledBuffer> {
let buffer = self.global.take()?;
Some(self.lease_into(buffer))
}
#[inline(always)]
fn lease_into(&self, mut buffer: PooledBuffer) -> PooledBuffer {
let lease = SizeClassLease::retain(self);
unsafe { buffer.init_lease(lease) };
buffer
}
#[inline(always)]
pub(super) fn size(&self) -> usize {
self.size
}
#[inline(always)]
pub(super) fn same_class(&self, other: &Self) -> bool {
self.token == other.token
}
#[inline(always)]
pub(super) fn drain_global(&self) {
self.global.drain();
}
}
impl Clone for SizeClassHandle {
fn clone(&self) -> Self {
unsafe { self.token.retain() };
Self { token: self.token }
}
}
impl Drop for SizeClassHandle {
fn drop(&mut self) {
unsafe { self.token.release() };
}
}
impl std::ops::Deref for SizeClassHandle {
type Target = SizeClass;
#[inline(always)]
fn deref(&self) -> &Self::Target {
unsafe { self.token.as_ref() }
}
}
#[must_use]
pub(crate) struct SizeClassLease {
token: SizeClassToken,
class_id: usize,
thread_cache_capacity: usize,
}
unsafe impl Send for SizeClassLease {}
unsafe impl Sync for SizeClassLease {}
impl SizeClassLease {
#[inline(always)]
fn retain(class: &SizeClassHandle) -> Self {
let token = class.token;
unsafe { token.retain() };
Self {
token,
class_id: class.class_id,
thread_cache_capacity: class.thread_cache_capacity,
}
}
#[inline(always)]
const fn class_id(&self) -> usize {
self.class_id
}
#[inline(always)]
const fn thread_cache_capacity(&self) -> usize {
self.thread_cache_capacity
}
#[inline(always)]
const fn class(&self) -> &SizeClass {
unsafe { self.token.as_ref() }
}
#[inline(always)]
fn return_global(self, buffer: PooledBuffer) {
self.class().global.put(buffer);
unsafe { self.token.release() };
}
#[inline(always)]
const fn into_token(self) -> SizeClassToken {
self.token
}
}
struct TlsSizeClassCacheEntry {
buffer: PooledBuffer,
}
impl TlsSizeClassCacheEntry {
#[inline(always)]
fn return_global(mut self) {
let lease = unsafe { self.buffer.take_lease() };
lease.return_global(self.buffer);
}
}
struct TlsSizeClassCache {
entries: Box<[MaybeUninit<TlsSizeClassCacheEntry>]>,
len: usize,
capacity: usize,
}
impl TlsSizeClassCache {
fn new(capacity: usize) -> Self {
let entries = (0..capacity)
.map(|_| MaybeUninit::uninit())
.collect::<Vec<_>>()
.into_boxed_slice();
Self {
entries,
len: 0,
capacity,
}
}
#[inline(always)]
fn pop(&mut self, class: &SizeClassHandle) -> Option<TlsSizeClassCacheEntry> {
if let Some(entry) = self.pop_local() {
return Some(entry);
}
self.pop_global(class)
}
#[inline(always)]
fn pop_local(&mut self) -> Option<TlsSizeClassCacheEntry> {
if self.len == 0 {
return None;
}
self.len -= 1;
Some(unsafe { self.entries.get_unchecked(self.len).assume_init_read() })
}
#[inline(never)]
fn pop_global(&mut self, class: &SizeClassHandle) -> Option<TlsSizeClassCacheEntry> {
if self.capacity < MIN_TLS_BATCH_CAPACITY {
return class
.take_global()
.map(|buffer| TlsSizeClassCacheEntry { buffer });
}
let mut entry = None;
let take = self.capacity / 2;
class.global.take_batch(take, |buffer| {
let buffer = class.lease_into(buffer);
let cache_entry = TlsSizeClassCacheEntry { buffer };
if entry.is_none() {
entry = Some(cache_entry);
} else {
self.push_local(cache_entry);
}
});
entry
}
#[inline(always)]
fn push(&mut self, buffer: PooledBuffer) {
let entry = TlsSizeClassCacheEntry { buffer };
if self.len < self.capacity {
self.push_local(entry);
return;
}
self.push_full(entry);
}
#[inline(always)]
fn push_local(&mut self, entry: TlsSizeClassCacheEntry) {
unsafe {
self.entries.get_unchecked_mut(self.len).write(entry);
}
self.len += 1;
}
#[inline(never)]
fn push_full(&mut self, entry: TlsSizeClassCacheEntry) {
if self.capacity < MIN_TLS_BATCH_CAPACITY {
entry.return_global();
return;
}
let spill = self.len.min(self.capacity / 2).max(1);
let end = self.len;
let start = end - spill;
self.len = start;
self.return_global_batch(start, end);
self.push_local(entry);
}
#[inline(never)]
fn return_global_batch(&mut self, start: usize, end: usize) {
assert!(start < end && end <= self.capacity);
assert!(self.len <= start);
let count = end - start;
let entries = self.entries.as_mut_ptr();
let token = unsafe { (*entries.add(start)).assume_init_ref().buffer.lease() }.token;
let class = unsafe { token.as_ref() };
let batch = (start..end).map(|index| {
let mut entry = unsafe { entries.add(index).read().assume_init() };
let _ = unsafe { entry.buffer.take_lease() }.into_token();
entry.buffer
});
class.global.put_batch(batch);
for _ in 0..count {
unsafe { token.release() };
}
}
}
impl Drop for TlsSizeClassCache {
fn drop(&mut self) {
if self.len == 0 {
return;
}
let end = self.len;
self.len = 0;
self.return_global_batch(0, end);
}
}
struct TlsSizeClassCaches {
bins: Vec<Option<TlsSizeClassCache>>,
}
impl TlsSizeClassCaches {
const fn new() -> Self {
Self { bins: Vec::new() }
}
#[inline(always)]
fn get_or_init(&mut self, class_id: usize, capacity: usize) -> &mut TlsSizeClassCache {
if class_id < self.bins.len() && self.bins[class_id].is_some() {
return self.bins[class_id]
.as_mut()
.expect("class cache was checked as initialized");
}
self.init(class_id, capacity)
}
#[inline(never)]
fn init(&mut self, class_id: usize, capacity: usize) -> &mut TlsSizeClassCache {
if class_id >= self.bins.len() {
self.bins.resize_with(class_id + 1, || None);
}
self.bins[class_id].get_or_insert_with(|| TlsSizeClassCache::new(capacity))
}
#[inline(always)]
fn get(&mut self, class_id: usize) -> Option<&mut TlsSizeClassCache> {
self.bins.get_mut(class_id).and_then(Option::as_mut)
}
}
impl Drop for TlsSizeClassCaches {
fn drop(&mut self) {
let this: *mut Self = self;
BufferPoolThreadCache::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| {
assert!(fast.get().is_null() || fast.get() == this);
fast.set(ptr::null_mut());
});
}
}
pub struct BufferPoolThreadCache;
impl BufferPoolThreadCache {
thread_local! {
static TLS_SIZE_CLASS_CACHES: UnsafeCell<TlsSizeClassCaches> =
const { UnsafeCell::new(TlsSizeClassCaches::new()) };
static TLS_SIZE_CLASS_CACHES_FAST: Cell<*mut TlsSizeClassCaches> =
const { Cell::new(ptr::null_mut()) };
}
pub fn flush() {
let _ = Self::TLS_SIZE_CLASS_CACHES.try_with(|caches| {
let caches = unsafe { &mut *caches.get() };
for cache in caches.bins.iter_mut() {
let _ = cache.take();
}
});
}
#[inline(always)]
pub(in crate::iobuf) fn push(buffer: PooledBuffer) {
let lease = unsafe { buffer.lease() };
let class_id = lease.class_id();
let thread_cache_capacity = lease.thread_cache_capacity();
if thread_cache_capacity == 0 {
TlsSizeClassCacheEntry { buffer }.return_global();
return;
}
let caches = Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.get());
if !caches.is_null() {
if let Some(cache) = unsafe { (&mut *caches).get(class_id) } {
cache.push(buffer);
return;
}
}
Self::push_slow(buffer);
}
#[inline(never)]
fn push_slow(buffer: PooledBuffer) {
let lease = unsafe { buffer.lease() };
let class_id = lease.class_id();
let thread_cache_capacity = lease.thread_cache_capacity();
match Self::TLS_SIZE_CLASS_CACHES
.try_with(|caches| {
let caches = caches.get();
Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.set(caches));
ptr::NonNull::from(unsafe {
(&mut *caches).get_or_init(class_id, thread_cache_capacity)
})
})
.ok()
{
Some(mut cache) => {
unsafe { cache.as_mut().push(buffer) };
}
None => TlsSizeClassCacheEntry { buffer }.return_global(),
}
}
#[inline(always)]
pub(super) fn pop(class: &SizeClassHandle) -> Option<PooledBuffer> {
if class.thread_cache_capacity == 0 {
return class.take_global();
}
let caches = Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.get());
if !caches.is_null() {
if let Some(cache) = unsafe { (&mut *caches).get(class.class_id) } {
return cache.pop(class).map(|entry| entry.buffer);
}
}
let Some(mut cache) = Self::cache_slow(class) else {
return class.take_global();
};
unsafe { cache.as_mut() }
.pop(class)
.map(|entry| entry.buffer)
}
#[inline(never)]
fn cache_slow(class: &SizeClassHandle) -> Option<ptr::NonNull<TlsSizeClassCache>> {
Self::TLS_SIZE_CLASS_CACHES
.try_with(|caches| {
let caches = caches.get();
Self::TLS_SIZE_CLASS_CACHES_FAST.with(|fast| fast.set(caches));
ptr::NonNull::from(unsafe {
(&mut *caches).get_or_init(class.class_id, class.thread_cache_capacity)
})
})
.ok()
}
}
#[cfg(all(test, not(feature = "loom")))]
pub(super) mod tests {
use super::{
super::{BufferPool, BufferPoolConfig, NEXT_SIZE_CLASS_ID},
*,
};
use crate::{
iobuf::{IoBuf, page_size},
telemetry::metrics::Registry,
};
use bytes::BufMut;
use commonware_utils::{NZU32, NZUsize};
use std::{
cell::Cell,
sync::{Arc, atomic::Ordering, mpsc},
thread,
};
fn test_size_class(size: usize, alignment: usize) -> SizeClassHandle {
SizeClassHandle::new(
NEXT_SIZE_CLASS_ID.fetch_add(1, Ordering::Relaxed),
size,
alignment,
NZU32!(8),
NZUsize!(4),
4,
false,
)
}
fn test_pool(config: BufferPoolConfig) -> BufferPool {
let mut registry = Registry::default();
BufferPool::new(config, &mut registry)
}
fn test_config(min_size: usize, max_size: usize, max_per_class: u32) -> BufferPoolConfig {
BufferPoolConfig::for_network()
.with_pool_min_size(0)
.with_size_class_range(
NZUsize!(min_size),
NZUsize!(max_size),
NZU32!(max_per_class),
)
.with_alignment(NZUsize!(page_size()))
}
fn size_class_strong_count(class: &SizeClassHandle) -> usize {
unsafe { class.token.retain() };
let arc = unsafe { Arc::from_raw(class.token.ptr.as_ptr()) };
Arc::strong_count(&arc) - 1
}
fn get_available(pool: &BufferPool, size: usize) -> i64 {
let class_index = pool.class_index(size).unwrap();
let class = &pool.inner.classes[class_index];
(get_global_len(class) + get_local_len(class)) as i64
}
pub const fn get_thread_cache_capacity(class: &SizeClass) -> usize {
class.thread_cache_capacity
}
pub fn get_global_len(class: &SizeClass) -> usize {
super::super::freelist::tests::len(&class.global)
}
pub fn get_global_created(class: &SizeClass) -> usize {
super::super::freelist::tests::created(&class.global)
}
pub fn get_global_num_stripes(class: &SizeClass) -> usize {
super::super::freelist::tests::num_stripes(&class.global)
}
pub fn get_local_len(class: &SizeClass) -> usize {
BufferPoolThreadCache::TLS_SIZE_CLASS_CACHES.with(|caches| {
let caches = unsafe { &*caches.get() };
caches
.bins
.get(class.class_id)
.and_then(Option::as_ref)
.map_or(0, |cache| cache.len)
})
}
#[test]
fn test_thread_cache_flush_moves_local_entries_to_global() {
let page = page_size();
let pool =
test_pool(test_config(page, page * 2, 8).with_max_thread_cache_capacity(NZUsize!(4)));
let small_index = pool.class_index(page).unwrap();
let large_index = pool.class_index(page + 1).unwrap();
let small_class = &pool.inner.classes[small_index];
let large_class = &pool.inner.classes[large_index];
let small = pool.try_alloc(page).expect("tracked allocation");
let large = pool.try_alloc(page + 1).expect("tracked allocation");
drop(small);
drop(large);
assert_eq!(get_local_len(small_class), 1);
assert_eq!(get_local_len(large_class), 1);
assert_eq!(get_global_len(small_class), 0);
assert_eq!(get_global_len(large_class), 0);
BufferPoolThreadCache::flush();
assert_eq!(get_local_len(small_class), 0);
assert_eq!(get_local_len(large_class), 0);
assert_eq!(get_global_len(small_class), 1);
assert_eq!(get_global_len(large_class), 1);
}
#[test]
fn test_return_buffer_local_overflow_spills_to_global() {
let page = page_size();
let pool = test_pool(test_config(page, page, 2));
let class_index = pool
.class_index(page)
.expect("class exists for page-sized buffer");
let tracked1 = pool.try_alloc(page).expect("first tracked allocation");
let tracked2 = pool.try_alloc(page).expect("second tracked allocation");
drop(tracked1);
assert_eq!(get_global_len(&pool.inner.classes[class_index]), 0);
assert_eq!(get_local_len(&pool.inner.classes[class_index]), 1);
drop(tracked2);
assert_eq!(get_global_len(&pool.inner.classes[class_index]), 1);
assert_eq!(get_local_len(&pool.inner.classes[class_index]), 1);
assert_eq!(get_available(&pool, page), 2);
}
#[test]
fn test_small_local_cache_overflow_preserves_locality() {
let page = page_size();
let pool = test_pool(test_config(page, page, 2));
let mut tracked1 = pool.try_alloc(page).expect("first tracked allocation");
let ptr1 = tracked1.as_mut_ptr();
let mut tracked2 = pool.try_alloc(page).expect("second tracked allocation");
let ptr2 = tracked2.as_mut_ptr();
drop(tracked1);
drop(tracked2);
let mut reused_local = pool.try_alloc(page).expect("reuse from local cache");
assert_eq!(reused_local.as_mut_ptr(), ptr1);
let mut reused_global = pool.try_alloc(page).expect("reuse from global freelist");
assert_eq!(reused_global.as_mut_ptr(), ptr2);
}
#[test]
fn test_large_local_cache_batches_overflow_and_refill() {
let page = page_size();
let threads = std::thread::available_parallelism().map_or(1, NonZeroUsize::get);
let max_per_class =
u32::try_from(threads * 8).expect("test capacity must fit in u32 slot ids");
let pool = test_pool(test_config(page, page, max_per_class));
let class_index = pool
.class_index(page)
.expect("class exists for page-sized buffer");
let class = &pool.inner.classes[class_index];
assert!(class.thread_cache_capacity >= MIN_TLS_BATCH_CAPACITY);
let mut bufs = Vec::new();
for _ in 0..class.thread_cache_capacity + 1 {
bufs.push(pool.try_alloc(page).expect("tracked allocation"));
}
for buf in bufs {
drop(buf);
}
assert_eq!(get_local_len(class), class.thread_cache_capacity / 2 + 1);
assert_eq!(get_global_len(class), class.thread_cache_capacity / 2);
let mut reused = Vec::new();
for _ in 0..class.thread_cache_capacity / 2 + 1 {
reused.push(pool.try_alloc(page).expect("local reuse"));
}
assert_eq!(get_local_len(class), 0);
assert_eq!(get_global_len(class), class.thread_cache_capacity / 2);
let _global = pool.try_alloc(page).expect("global reuse with refill");
assert_eq!(get_local_len(class), class.thread_cache_capacity / 2 - 1);
assert_eq!(get_global_len(class), 0);
}
#[test]
fn test_global_batch_alloc_stops_when_global_runs_empty() {
let class = test_size_class(64, 64);
let buffer = class.global.try_create(false).expect("slot reservation");
class.global.put(buffer);
let buffer = BufferPoolThreadCache::pop(&class).expect("global allocation");
assert_eq!(get_local_len(&class), 0);
assert_eq!(get_global_len(&class), 0);
TlsSizeClassCacheEntry { buffer }.return_global();
}
#[test]
fn test_size_class_leases_use_raw_arc_tokens_across_cache_paths() {
let class = test_size_class(64, 64);
let mut cache = TlsSizeClassCache::new(MIN_TLS_BATCH_CAPACITY);
assert_eq!(size_class_strong_count(&class), 1);
let mut buffer = class.global.try_create(false).expect("slot reservation");
let lease = SizeClassLease::retain(&class);
unsafe { buffer.init_lease(lease) };
assert_eq!(size_class_strong_count(&class), 2);
cache.push(buffer);
assert_eq!(size_class_strong_count(&class), 2);
let entry = cache.pop(&class).expect("local cache pop");
assert_eq!(size_class_strong_count(&class), 2);
entry.return_global();
assert_eq!(size_class_strong_count(&class), 1);
for _ in 0..2 {
let buffer = class.global.try_create(false).expect("slot reservation");
class.global.put(buffer);
}
let entry = cache.pop(&class).expect("global refill");
assert_eq!(size_class_strong_count(&class), 3);
entry.return_global();
assert_eq!(size_class_strong_count(&class), 2);
drop(cache);
assert_eq!(size_class_strong_count(&class), 1);
}
#[test]
fn test_tls_size_class_cache_push_tolerates_empty_spill() {
let class = test_size_class(64, 64);
let mut buffer = class.global.try_create(false).expect("slot reservation");
let lease = SizeClassLease::retain(&class);
unsafe { buffer.init_lease(lease) };
let mut cache = TlsSizeClassCache::new(0);
cache.push(buffer);
assert_eq!(cache.len, 0);
drop(cache);
}
#[test]
fn test_global_freelist_returns_each_slot_once() {
let class = SizeClassHandle::new(
NEXT_SIZE_CLASS_ID.fetch_add(1, Ordering::Relaxed),
64,
64,
NZU32!(2),
NZUsize!(1),
1,
false,
);
let buffer0 = class.global.try_create(false).expect("first slot");
let slot0 = buffer0.slot();
let ptr0 = buffer0.as_ptr();
let buffer1 = class.global.try_create(false).expect("second slot");
let slot1 = buffer1.slot();
let ptr1 = buffer1.as_ptr();
let mut expected = [(slot0, ptr0), (slot1, ptr1)];
expected.sort_by_key(|(slot, _)| *slot);
class.global.put(buffer0);
class.global.put(buffer1);
let mut popped = [
class.global.take().expect("first pop"),
class.global.take().expect("second pop"),
];
popped.sort_by_key(PooledBuffer::slot);
assert_eq!(popped[0].slot(), expected[0].0);
assert_eq!(popped[0].as_ptr(), expected[0].1);
assert_eq!(popped[1].slot(), expected[1].0);
assert_eq!(popped[1].as_ptr(), expected[1].1);
assert!(class.global.take().is_none());
for buffer in popped {
class.global.put(buffer);
}
}
#[test]
fn test_thread_exit_flushes_local_bin() {
let page = page_size();
let pool = Arc::new(test_pool(test_config(page, page, 1)));
let worker_pool = pool.clone();
thread::spawn(move || {
let buf = worker_pool
.try_alloc(page)
.expect("worker should allocate tracked buffer");
drop(buf);
})
.join()
.expect("worker thread should exit cleanly");
let class_index = pool
.class_index(page)
.expect("class exists for page-sized buffer");
assert_eq!(get_global_len(&pool.inner.classes[class_index]), 1);
assert_eq!(get_local_len(&pool.inner.classes[class_index]), 0);
let _buf = pool
.try_alloc(page)
.expect("thread-exited local buffer should be reusable");
}
#[test]
fn test_thread_exit_batch_flush_outlives_pool() {
let page = page_size();
let pool = test_pool(test_config(page, page, 8));
let (cached_tx, cached_rx) = mpsc::channel();
let (release_tx, release_rx) = mpsc::channel::<()>();
let worker_pool = pool.clone();
let handle = thread::spawn(move || {
let class_index = worker_pool
.class_index(page)
.expect("class exists for page-sized buffer");
let class = &worker_pool.inner.classes[class_index];
assert!(class.thread_cache_capacity >= MIN_TLS_BATCH_CAPACITY);
let bufs = (0..MIN_TLS_BATCH_CAPACITY)
.map(|_| worker_pool.try_alloc(page).expect("tracked allocation"))
.collect::<Vec<_>>();
drop(bufs);
assert_eq!(get_local_len(class), MIN_TLS_BATCH_CAPACITY);
drop(worker_pool);
cached_tx.send(()).expect("signal cached buffers");
release_rx.recv().expect("wait for pool drop");
});
cached_rx.recv().expect("worker cached buffers");
drop(pool);
release_tx.send(()).expect("release worker");
handle.join().expect("worker thread should exit cleanly");
}
#[test]
fn test_pooled_ops_inside_tls_destructor_fall_back_to_global() {
struct ExitReleaser {
pool: BufferPool,
size: usize,
held: Option<IoBuf>,
}
impl Drop for ExitReleaser {
fn drop(&mut self) {
let extra = self
.pool
.try_alloc(self.size)
.expect("pool must serve allocations from TLS destructors");
drop(extra);
drop(self.held.take());
}
}
thread_local! {
static EXIT_RELEASER: Cell<Option<ExitReleaser>> = const { Cell::new(None) };
}
let page = page_size();
let pool = test_pool(test_config(page, page, 4));
thread::spawn({
let pool = pool.clone();
move || {
EXIT_RELEASER.with(|cell| {
cell.set(Some(ExitReleaser {
pool: pool.clone(),
size: page,
held: None,
}));
});
drop(pool.try_alloc(page).expect("first allocation"));
let mut buf = pool.try_alloc(page).expect("second allocation");
buf.put_u8(1);
let held = buf.freeze();
EXIT_RELEASER.with(|cell| {
let mut releaser = cell.take().expect("releaser installed above");
releaser.held = Some(held);
cell.set(Some(releaser));
});
}
})
.join()
.expect("worker thread must exit cleanly");
let class_index = pool
.class_index(page)
.expect("class exists for page-sized buffer");
assert_eq!(get_local_len(&pool.inner.classes[class_index]), 0);
let bufs = (0..4)
.map(|i| {
pool.try_alloc(page)
.unwrap_or_else(|_| panic!("buffer {i} was stranded at thread exit"))
})
.collect::<Vec<_>>();
drop(bufs);
}
#[test]
fn test_pool_drop_drains_global_freelist() {
let page = page_size();
let pool = test_pool(test_config(page, page, 2));
let class_index = pool
.class_index(page)
.expect("class exists for page-sized buffer");
let class = &pool.inner.classes[class_index];
unsafe { class.token.retain() };
let class = SizeClassHandle { token: class.token };
let buf1 = pool.try_alloc(page).unwrap();
let buf2 = pool.try_alloc(page).unwrap();
drop(buf1);
drop(buf2);
assert_eq!(get_global_len(&class), 1);
assert_eq!(get_local_len(&class), 1);
drop(pool);
assert_eq!(get_global_len(&class), 0);
assert_eq!(get_local_len(&class), 1);
assert_eq!(get_global_created(&class), 2);
}
}
#[cfg(all(test, feature = "loom"))]
mod loom_tests {
use super::*;
use commonware_utils::{NZU32, NZUsize};
use loom::thread;
#[test]
fn tls_batch_drop_races_pool_teardown() {
loom::model(|| {
let class = SizeClassHandle::new(1, 64, 64, NZU32!(2), NZUsize!(1), 2, false);
let mut cache = TlsSizeClassCache::new(2);
for _ in 0..2 {
let buffer = class.try_create(false).expect("tracked slot");
cache.push(buffer);
}
assert_eq!(cache.len, 2);
let t = thread::spawn(move || drop(cache));
drop(class);
t.join().unwrap();
});
}
}