use std::fmt;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
pub const DEFAULT_MEMORY_BUDGET_BYTES: usize = 100_000_000;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MemoryBudget {
bytes: usize,
}
impl MemoryBudget {
pub const fn new(bytes: usize) -> Result<Self, MemoryBudgetError> {
if bytes == 0 {
return Err(MemoryBudgetError::Zero);
}
Ok(Self { bytes })
}
#[must_use]
pub const fn bytes(self) -> usize {
self.bytes
}
}
impl Default for MemoryBudget {
fn default() -> Self {
Self {
bytes: DEFAULT_MEMORY_BUDGET_BYTES,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MemoryBudgetError {
Zero,
Exhausted {
requested: usize,
available: usize,
},
}
impl fmt::Display for MemoryBudgetError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Zero => formatter.write_str("execution memory budget must be non-zero"),
Self::Exhausted {
requested,
available,
} => write!(
formatter,
"execution requested {requested} bytes with {available} bytes available"
),
}
}
}
impl std::error::Error for MemoryBudgetError {}
#[derive(Debug)]
pub(crate) struct MemoryAccount {
budget: MemoryBudget,
reserved: AtomicUsize,
peak_reserved: AtomicUsize,
live_batches: AtomicUsize,
peak_live_batches: AtomicUsize,
}
impl MemoryAccount {
pub(crate) fn new(budget: MemoryBudget) -> Self {
Self {
budget,
reserved: AtomicUsize::new(0),
peak_reserved: AtomicUsize::new(0),
live_batches: AtomicUsize::new(0),
peak_live_batches: AtomicUsize::new(0),
}
}
pub(crate) const fn budget(&self) -> MemoryBudget {
self.budget
}
pub(crate) fn reserve(
account: &Arc<Self>,
bytes: usize,
) -> Result<MemoryReservation, MemoryBudgetError> {
let mut current = account.reserved.load(Ordering::Acquire);
loop {
let Some(next) = current.checked_add(bytes) else {
return Err(MemoryBudgetError::Exhausted {
requested: bytes,
available: account.budget.bytes().saturating_sub(current),
});
};
if next > account.budget.bytes() {
return Err(MemoryBudgetError::Exhausted {
requested: bytes,
available: account.budget.bytes().saturating_sub(current),
});
}
match account.reserved.compare_exchange_weak(
current,
next,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => {
account.peak_reserved.fetch_max(next, Ordering::AcqRel);
return Ok(MemoryReservation {
account: Arc::clone(account),
bytes,
});
}
Err(observed) => current = observed,
}
}
}
pub(crate) fn reserved_bytes(&self) -> usize {
self.reserved.load(Ordering::Acquire)
}
pub(crate) fn peak_reserved_bytes(&self) -> usize {
self.peak_reserved.load(Ordering::Acquire)
}
pub(crate) fn live_batches(&self) -> usize {
self.live_batches.load(Ordering::Acquire)
}
pub(crate) fn peak_live_batches(&self) -> usize {
self.peak_live_batches.load(Ordering::Acquire)
}
fn begin_batch(account: &Arc<Self>) -> BatchGuard {
let live = account.live_batches.fetch_add(1, Ordering::AcqRel) + 1;
account.peak_live_batches.fetch_max(live, Ordering::AcqRel);
BatchGuard {
account: Arc::clone(account),
}
}
}
#[derive(Debug)]
pub struct MemoryReservation {
account: Arc<MemoryAccount>,
bytes: usize,
}
impl MemoryReservation {
#[must_use]
pub fn belongs_to(&self, context: &super::ExecutionContext) -> bool {
Arc::ptr_eq(&self.account, &context.memory)
}
#[must_use]
pub const fn bytes(&self) -> usize {
self.bytes
}
pub fn shrink_to(&mut self, bytes: usize) {
if bytes >= self.bytes {
return;
}
let released = self.bytes - bytes;
self.bytes = bytes;
self.account.reserved.fetch_sub(released, Ordering::AcqRel);
}
pub fn try_grow(&mut self, additional: usize) -> Result<(), MemoryBudgetError> {
if additional == 0 {
return Ok(());
}
let mut extra = MemoryAccount::reserve(&self.account, additional)?;
let Some(bytes) = self.bytes.checked_add(additional) else {
return Err(MemoryBudgetError::Exhausted {
requested: additional,
available: 0,
});
};
self.bytes = bytes;
extra.bytes = 0;
Ok(())
}
pub(crate) fn begin_batch(&self) -> BatchGuard {
MemoryAccount::begin_batch(&self.account)
}
}
impl Drop for MemoryReservation {
fn drop(&mut self) {
self.account
.reserved
.fetch_sub(self.bytes, Ordering::AcqRel);
}
}
#[derive(Debug)]
pub(crate) struct BatchGuard {
account: Arc<MemoryAccount>,
}
impl Drop for BatchGuard {
fn drop(&mut self) {
self.account.live_batches.fetch_sub(1, Ordering::AcqRel);
}
}
#[cfg(test)]
#[path = "memory_tests.rs"]
mod tests;