mod aligned_buf;
mod budget;
mod depot;
mod inbox;
mod tls;
use std::{
alloc::{Layout, dealloc},
mem::forget,
ptr::{self, NonNull, write_bytes},
sync::{
Arc,
atomic::{
AtomicBool, AtomicU64,
Ordering::{AcqRel, Acquire, Relaxed},
},
},
};
pub use aligned_buf::AlignedBuf;
pub(crate) use budget::Budget;
pub(crate) use depot::Depot;
pub(crate) use inbox::{ChainIter, CrossThreadInbox, FreeNode};
pub(crate) use tls::TLS_POOLS;
pub use tls::current_thread_id;
use crate::{
align::{MIN_SECTOR_SIZE, is_valid_sector_size},
error::{Error, Result},
};
#[inline]
pub(crate) fn validate_sector_size(size: usize) -> Result<()> {
if !is_valid_sector_size(size) {
return Err(Error::InvalidAlignment(size, MIN_SECTOR_SIZE));
}
Ok(())
}
pub const NUM_CLASSES: usize = 28;
const TINY_EXACT_CLASSES: usize = 2;
const LINEAR_STRIDE_SECTORS: usize = 4;
const STRIDE_CLASSES: usize = 4;
const LINEAR_CLASSES: usize = TINY_EXACT_CLASSES + STRIDE_CLASSES;
const LINEAR_TOP_SECTORS: usize = LINEAR_STRIDE_SECTORS * STRIDE_CLASSES;
const LOG2_LINEAR_TOP: usize = 4;
const GEOMETRIC_DOUBLINGS: usize = 11;
pub const MAX_POOLED_SECTORS: usize = LINEAR_TOP_SECTORS << GEOMETRIC_DOUBLINGS;
pub const LARGE_TIER_MIN_BYTES: usize = 256 << 10;
pub const DEFAULT_SMALL_BUDGET_BYTES: i64 = 32 << 20;
pub const DEFAULT_LARGE_BUDGET_BYTES: i64 = 128 << 20;
pub const MAX_LOCAL_PER_CLASS: usize = 64;
pub(crate) const DEPOT_STRIPES: usize = 8;
pub(crate) const DEPOT_STRIPE_MASK: usize = DEPOT_STRIPES - 1;
pub const DEPOT_STRIPE_CAP: usize = 8;
#[inline]
#[must_use]
pub const fn class_capacity_sectors(cls: usize) -> usize {
if cls < TINY_EXACT_CLASSES {
cls + 1
} else if cls < LINEAR_CLASSES {
(cls - TINY_EXACT_CLASSES + 1) * LINEAR_STRIDE_SECTORS
} else {
let g = cls - LINEAR_CLASSES;
let octave = LOG2_LINEAR_TOP.saturating_add(g >> 1);
let (base, exp) = if g & 1 == 0 {
(3usize, octave - 1)
} else {
(1usize, octave + 1)
};
if exp >= u32::MAX as usize {
return usize::MAX;
}
match base.checked_shl(exp as u32) {
Some(cap) => cap,
None => usize::MAX,
}
}
}
pub const CLASS_CAPACITIES_SECTORS: [usize; NUM_CLASSES] = {
let mut arr = [0; NUM_CLASSES];
let mut i = 0;
while i < NUM_CLASSES {
arr[i] = class_capacity_sectors(i);
i += 1;
}
arr
};
#[inline]
#[must_use]
pub const fn class_capacity_bytes(cls: usize, sector_size: usize) -> usize {
class_capacity_sectors(cls).saturating_mul(sector_size)
}
#[inline]
#[must_use]
pub const fn class_of_sectors(sectors: usize) -> Option<usize> {
let sectors = if sectors == 0 { 1 } else { sectors };
if sectors <= TINY_EXACT_CLASSES {
return Some(sectors - 1);
}
if sectors <= LINEAR_TOP_SECTORS {
return Some(TINY_EXACT_CLASSES + (sectors - 1) / LINEAR_STRIDE_SECTORS);
}
if sectors > MAX_POOLED_SECTORS {
return None;
}
let octave = (sectors - 1).ilog2() as usize;
let mid = 3 << (octave - 1);
let sub = (sectors > mid) as usize;
Some(LINEAR_CLASSES + 2 * (octave - LOG2_LINEAR_TOP) + sub)
}
pub(crate) struct CachedBuf {
pub(crate) ptr: NonNull<u8>,
pub(crate) cap: usize,
pub(crate) align: usize,
pub(crate) cacheable: bool,
pub(crate) dirty: bool,
}
unsafe impl Send for CachedBuf {}
impl Drop for CachedBuf {
fn drop(&mut self) {
unsafe {
dealloc(
self.ptr.as_ptr(),
Layout::from_size_align_unchecked(self.cap, self.align),
)
};
}
}
pub(crate) struct BufMeta {
pub(crate) cls: u32,
pub(crate) cacheable: bool,
pub(crate) clear_on_return: bool,
pub(crate) required: usize,
pub(crate) owner_tid: u64,
pub(crate) inbox: Option<Arc<CrossThreadInbox>>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PoolStats {
pub reserved_bytes: i64,
pub small_reserved_bytes: i64,
pub large_reserved_bytes: i64,
pub direct_alloc_count: u64,
pub direct_alloc_bytes: u64,
pub bypass_alloc_count: u64,
pub bypass_alloc_bytes: u64,
}
pub struct BufferPool {
pub pool_id: u64,
sector_size: usize,
sector_shift: u32,
small_budget: Budget,
large_budget: Budget,
first_large_class: usize,
depot: Depot,
is_closed: AtomicBool,
direct_alloc_count: AtomicU64,
direct_alloc_bytes: AtomicU64,
bypass_alloc_count: AtomicU64,
bypass_alloc_bytes: AtomicU64,
}
impl BufferPool {
pub fn new(sector_size: usize) -> Result<Arc<Self>> {
Self::with_budgets(
sector_size,
DEFAULT_SMALL_BUDGET_BYTES,
DEFAULT_LARGE_BUDGET_BYTES,
)
}
pub fn with_budgets(
sector_size: usize,
small_budget_bytes: i64,
large_budget_bytes: i64,
) -> Result<Arc<Self>> {
validate_sector_size(sector_size)?;
if small_budget_bytes < 0 {
return Err(Error::InvalidBudget(small_budget_bytes));
}
if large_budget_bytes < 0 {
return Err(Error::InvalidBudget(large_budget_bytes));
}
if sector_size > (i64::MAX as usize) / MAX_POOLED_SECTORS {
return Err(Error::InvalidSize(sector_size));
}
static NEXT_POOL_ID: AtomicU64 = AtomicU64::new(1);
let pool_id = NEXT_POOL_ID.fetch_add(1, Relaxed);
let sector_shift = sector_size.trailing_zeros();
Ok(Arc::new(Self {
pool_id,
sector_size,
sector_shift,
small_budget: Budget::new(small_budget_bytes),
large_budget: Budget::new(large_budget_bytes),
first_large_class: (0..NUM_CLASSES)
.find(|&c| class_capacity_bytes(c, sector_size) > LARGE_TIER_MIN_BYTES)
.unwrap_or(NUM_CLASSES),
depot: Depot::new(),
is_closed: AtomicBool::new(false),
direct_alloc_count: AtomicU64::new(0),
direct_alloc_bytes: AtomicU64::new(0),
bypass_alloc_count: AtomicU64::new(0),
bypass_alloc_bytes: AtomicU64::new(0),
}))
}
#[inline]
#[must_use]
pub fn sector_size(&self) -> usize {
self.sector_size
}
#[inline]
#[must_use]
pub fn is_closed(&self) -> bool {
self.is_closed.load(Acquire)
}
#[inline]
#[must_use]
pub fn reserved_bytes(&self) -> i64 {
self.small_budget.used() + self.large_budget.used()
}
#[inline]
#[must_use]
pub fn small_reserved_bytes(&self) -> i64 {
self.small_budget.used()
}
#[inline]
#[must_use]
pub fn large_reserved_bytes(&self) -> i64 {
self.large_budget.used()
}
#[inline]
#[must_use]
pub fn small_budget_bytes(&self) -> i64 {
self.small_budget.total()
}
#[inline]
#[must_use]
pub fn large_budget_bytes(&self) -> i64 {
self.large_budget.total()
}
pub fn stats(&self) -> PoolStats {
PoolStats {
reserved_bytes: self.reserved_bytes(),
small_reserved_bytes: self.small_reserved_bytes(),
large_reserved_bytes: self.large_reserved_bytes(),
direct_alloc_count: self.direct_alloc_count.load(Relaxed),
direct_alloc_bytes: self.direct_alloc_bytes.load(Relaxed),
bypass_alloc_count: self.bypass_alloc_count.load(Relaxed),
bypass_alloc_bytes: self.bypass_alloc_bytes.load(Relaxed),
}
}
pub fn free(&self) {
if self.is_closed.swap(true, AcqRel) {
return;
}
self.depot.clear(|cls, cap| {
self.budget_for(cls).release(cap as i64);
});
self.drain_tls_self();
}
fn drain_tls_self(&self) {
let _ = TLS_POOLS.try_with(|mgr| {
if let Some(entry) = mgr.borrow_mut().find_mut(self.pool_id) {
entry.drain_and_release(self);
}
});
}
#[must_use]
pub fn cached_len(&self, cls: usize) -> usize {
if cls >= NUM_CLASSES {
return 0;
}
let local_len = TLS_POOLS
.try_with(|mgr| {
mgr
.borrow()
.find(self.pool_id)
.map_or(0, |e| e.local[cls].len())
})
.unwrap_or(0);
local_len + self.depot.total_cached(cls)
}
pub fn get(self: &Arc<Self>, required_bytes: usize) -> Result<AlignedBuf> {
self.get_with_policy(required_bytes, true)
}
pub fn get_from_slice(self: &Arc<Self>, slice: &[u8]) -> Result<AlignedBuf> {
let mut buf = self.get(slice.len())?;
buf.as_allocated_slice_mut()[..slice.len()].copy_from_slice(slice);
buf.set_len(slice.len())?;
Ok(buf)
}
pub fn ensure_size(self: &Arc<Self>, buf: &mut AlignedBuf, size: usize) -> Result<()> {
if buf.capacity() < size {
*buf = self.get(size)?;
} else {
buf.set_required(size);
}
Ok(())
}
pub fn get_with_policy(
self: &Arc<Self>,
required_bytes: usize,
clear_on_return: bool,
) -> Result<AlignedBuf> {
if required_bytes == 0 {
return AlignedBuf::new(0, self.sector_size);
}
let required = match required_bytes.checked_add(self.sector_size - 1) {
Some(v) => v & !(self.sector_size - 1),
None => return Err(Error::Overflow),
};
let Some(cls) = class_of_sectors(required >> self.sector_shift) else {
self.record_bypass(required);
return AlignedBuf::new(required, self.sector_size);
};
if self.is_closed.load(Acquire) {
self.drain_tls_self();
self.record_bypass(required);
return AlignedBuf::new(required, self.sector_size);
}
let tid = current_thread_id();
if cls < self.first_large_class {
let (cached_opt, inbox) = TLS_POOLS
.try_with(|mgr| {
let mut mgr = mgr.borrow_mut();
let entry = mgr.get_or_create(self);
if let Some(node) = entry.local[cls].pop() {
return (Some(node), Some(entry.inbox.clone()));
}
let chain = entry.inbox.claim(cls);
if !chain.is_null() {
let mut iter = ChainIter::new(chain);
if let Some(first) = iter.next() {
for node in iter {
if entry.local[cls].len() < MAX_LOCAL_PER_CLASS {
entry.local[cls].push(node);
} else {
self.spill_to_depot(cls, node, tid);
}
}
return (Some(first), Some(entry.inbox.clone()));
}
}
(None, Some(entry.inbox.clone()))
})
.unwrap_or((None, None));
if let Some(node) = cached_opt {
return Ok(self.reuse_cached(node, cls, required_bytes, clear_on_return, tid, inbox));
}
if let Some(node) = self.depot.pop(cls, tid) {
return Ok(self.reuse_cached(node, cls, required_bytes, clear_on_return, tid, inbox));
}
return self.issue_new(cls, required_bytes, clear_on_return, tid, inbox);
}
if let Some(node) = self.depot.pop(cls, tid) {
return Ok(self.reuse_cached(node, cls, required_bytes, clear_on_return, tid, None));
}
self.issue_new(cls, required_bytes, clear_on_return, tid, None)
}
#[inline]
fn record_bypass(&self, bytes: usize) {
self.bypass_alloc_count.fetch_add(1, Relaxed);
self.bypass_alloc_bytes.fetch_add(bytes as u64, Relaxed);
}
fn reuse_cached(
self: &Arc<Self>,
node: CachedBuf,
cls: usize,
required: usize,
clear_on_return: bool,
tid: u64,
inbox: Option<Arc<CrossThreadInbox>>,
) -> AlignedBuf {
if clear_on_return && node.dirty {
unsafe { write_bytes(node.ptr.as_ptr(), 0, node.cap) };
}
let cacheable = node.cacheable;
AlignedBuf::from_cached(
node,
self.clone(),
BufMeta {
cls: cls as u32,
cacheable,
clear_on_return,
required,
owner_tid: tid,
inbox,
},
)
}
fn issue_new(
self: &Arc<Self>,
cls: usize,
required: usize,
clear_on_return: bool,
tid: u64,
inbox: Option<Arc<CrossThreadInbox>>,
) -> Result<AlignedBuf> {
let cap = class_capacity_bytes(cls, self.sector_size);
let cacheable = self.budget_for(cls).try_reserve(cap as i64);
if !cacheable {
self.direct_alloc_count.fetch_add(1, Relaxed);
self.direct_alloc_bytes.fetch_add(cap as u64, Relaxed);
}
let mut buf = match AlignedBuf::new(cap, self.sector_size) {
Ok(b) => b,
Err(e) => {
if cacheable {
self.budget_for(cls).release(cap as i64);
}
return Err(e);
}
};
buf.attach(
self.clone(),
BufMeta {
cls: cls as u32,
cacheable,
clear_on_return,
required,
owner_tid: tid,
inbox,
},
);
Ok(buf)
}
pub(crate) fn return_buf(&self, ptr: NonNull<u8>, cap: usize, align: usize, meta: BufMeta) {
let BufMeta {
cls,
cacheable,
clear_on_return,
owner_tid,
inbox,
..
} = meta;
let cls = cls as usize;
let dealloc = |ptr: NonNull<u8>| unsafe {
dealloc(ptr.as_ptr(), Layout::from_size_align_unchecked(cap, align))
};
if !cacheable || cls >= NUM_CLASSES {
dealloc(ptr);
return;
}
if self.is_closed.load(Acquire) {
self.budget_for(cls).release(cap as i64);
dealloc(ptr);
return;
}
if clear_on_return {
unsafe { write_bytes(ptr.as_ptr(), 0, cap) };
}
let node = CachedBuf {
ptr,
cap,
align,
cacheable,
dirty: !clear_on_return,
};
let tid = current_thread_id();
if cls >= self.first_large_class {
self.spill_to_depot(cls, node, tid);
return;
}
if tid == owner_tid {
self.return_owner(cls, node, tid);
} else {
self.return_foreign(cls, node, inbox, tid);
}
}
fn spill_to_depot(&self, cls: usize, node: CachedBuf, tid: u64) {
let cap = node.cap;
if !self.depot.push(cls, node, tid) {
self.budget_for(cls).release(cap as i64);
}
}
fn return_owner(&self, cls: usize, node: CachedBuf, tid: u64) {
let mut hold = Some(node);
let _ = TLS_POOLS.try_with(|mgr| match mgr.borrow_mut().find_mut(self.pool_id) {
Some(entry) if entry.local[cls].len() < MAX_LOCAL_PER_CLASS => {
if let Some(node) = hold.take() {
entry.local[cls].push(node);
}
}
_ => {}
});
if let Some(node) = hold {
self.spill_to_depot(cls, node, tid);
}
}
fn return_foreign(
&self,
cls: usize,
node: CachedBuf,
inbox: Option<Arc<CrossThreadInbox>>,
tid: u64,
) {
if let Some(inbox) = inbox {
let node_ptr = node.ptr.as_ptr() as *mut FreeNode;
unsafe {
ptr::write(
node_ptr,
FreeNode {
next: ptr::null_mut(),
cap: node.cap,
align: node.align,
cacheable: node.cacheable,
dirty: node.dirty,
},
);
}
if inbox.try_push(cls, node_ptr) {
forget(node);
return;
}
if !node.dirty {
unsafe { write_bytes(node_ptr, 0, size_of::<FreeNode>()) };
}
}
self.spill_to_depot(cls, node, tid);
}
#[inline]
pub(crate) fn budget_for(&self, cls: usize) -> &Budget {
if cls >= self.first_large_class {
&self.large_budget
} else {
&self.small_budget
}
}
}
impl Drop for BufferPool {
fn drop(&mut self) {
self.free();
}
}