use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, OnceLock};
use dashmap::DashMap;
use tensor_wasm_core::types::InstanceId;
use crate::abi::AbiError;
pub const MAX_DEVICE_ALLOC_BYTES: u64 = 256 * 1024 * 1024;
pub const MAX_DEVICE_ALLOCS_PER_INSTANCE: usize = 4096;
pub const MAX_TOTAL_DEVICE_BYTES: u64 = 16 * MAX_DEVICE_ALLOC_BYTES;
pub const MAX_PROCESS_DEVICE_BYTES: u64 = 4 * MAX_TOTAL_DEVICE_BYTES;
pub const MAX_PROCESS_DEVICE_ALLOCS: usize = 4 * MAX_DEVICE_ALLOCS_PER_INSTANCE;
#[derive(Debug, Default)]
pub struct DeviceMemBudget {
total_bytes: AtomicU64,
total_allocs: AtomicU64,
}
impl DeviceMemBudget {
pub fn new() -> Self {
Self {
total_bytes: AtomicU64::new(0),
total_allocs: AtomicU64::new(0),
}
}
fn try_reserve(&self, size: u64) -> Result<(), AbiError> {
let prev_allocs = self.total_allocs.fetch_add(1, Ordering::AcqRel);
if prev_allocs >= MAX_PROCESS_DEVICE_ALLOCS as u64 {
self.total_allocs.fetch_sub(1, Ordering::AcqRel);
return Err(AbiError::QuotaExceeded);
}
let mut current = self.total_bytes.load(Ordering::Acquire);
loop {
let next = current.saturating_add(size);
if next > MAX_PROCESS_DEVICE_BYTES {
self.total_allocs.fetch_sub(1, Ordering::AcqRel);
return Err(AbiError::QuotaExceeded);
}
match self.total_bytes.compare_exchange_weak(
current,
next,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => return Ok(()),
Err(observed) => current = observed,
}
}
}
fn release(&self, size: u64) {
let _ = self
.total_bytes
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
Some(cur.saturating_sub(size))
});
let _ = self
.total_allocs
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
Some(cur.saturating_sub(1))
});
}
pub fn total_bytes(&self) -> u64 {
self.total_bytes.load(Ordering::Acquire)
}
pub fn total_allocs(&self) -> u64 {
self.total_allocs.load(Ordering::Acquire)
}
}
pub fn process_device_budget() -> &'static Arc<DeviceMemBudget> {
static BUDGET: OnceLock<Arc<DeviceMemBudget>> = OnceLock::new();
BUDGET.get_or_init(|| Arc::new(DeviceMemBudget::new()))
}
#[derive(Debug)]
pub struct DeviceMemEntry {
pub owner: InstanceId,
pub size: u64,
#[cfg(feature = "cuda")]
pub device_ptr: cust::sys::CUdeviceptr,
}
#[derive(Clone, Debug)]
pub struct DeviceMemHandle {
pub owner: InstanceId,
pub size: u64,
#[cfg(feature = "cuda")]
pub device_ptr: cust::sys::CUdeviceptr,
}
pub struct DeviceMemRegistry {
next_handle: AtomicU64,
entries: DashMap<u64, DeviceMemEntry>,
total_device_bytes: AtomicU64,
budget: Arc<DeviceMemBudget>,
}
impl Default for DeviceMemRegistry {
fn default() -> Self {
Self::new()
}
}
impl DeviceMemRegistry {
pub fn new() -> Self {
Self::with_budget(Arc::clone(process_device_budget()))
}
pub fn with_budget(budget: Arc<DeviceMemBudget>) -> Self {
Self {
next_handle: AtomicU64::new(1),
entries: DashMap::new(),
total_device_bytes: AtomicU64::new(0),
budget,
}
}
pub fn insert(&self, entry: DeviceMemEntry) -> Result<u64, AbiError> {
if self.entries.len() >= MAX_DEVICE_ALLOCS_PER_INSTANCE {
return Err(AbiError::QuotaExceeded);
}
let add = entry.size;
let mut current = self.total_device_bytes.load(Ordering::Acquire);
loop {
let next = current.saturating_add(add);
if next > MAX_TOTAL_DEVICE_BYTES {
return Err(AbiError::QuotaExceeded);
}
match self.total_device_bytes.compare_exchange_weak(
current,
next,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => break,
Err(observed) => current = observed,
}
}
if let Err(e) = self.budget.try_reserve(add) {
let _ =
self.total_device_bytes
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
Some(cur.saturating_sub(add))
});
return Err(e);
}
let handle = self.next_handle.fetch_add(1, Ordering::Relaxed);
self.entries.insert(handle, entry);
Ok(handle)
}
pub fn lookup(&self, handle: u64, owner: InstanceId) -> Result<DeviceMemHandle, AbiError> {
let r = self.entries.get(&handle).ok_or(AbiError::InvalidHandle)?;
if r.owner != owner {
return Err(AbiError::InvalidHandle);
}
Ok(DeviceMemHandle {
owner: r.owner,
size: r.size,
#[cfg(feature = "cuda")]
device_ptr: r.device_ptr,
})
}
pub fn free(&self, handle: u64, owner: InstanceId) -> Result<DeviceMemEntry, AbiError> {
{
let r = self.entries.get(&handle).ok_or(AbiError::InvalidHandle)?;
if r.owner != owner {
return Err(AbiError::InvalidHandle);
}
}
let (_, entry) = self
.entries
.remove(&handle)
.ok_or(AbiError::InvalidHandle)?;
let _ = self
.total_device_bytes
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
Some(cur.saturating_sub(entry.size))
});
self.budget.release(entry.size);
Ok(entry)
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn total_device_bytes(&self) -> u64 {
self.total_device_bytes.load(Ordering::Acquire)
}
pub fn budget(&self) -> &Arc<DeviceMemBudget> {
&self.budget
}
}
#[cfg(test)]
mod tests {
use super::*;
fn entry(owner: InstanceId, size: u64) -> DeviceMemEntry {
DeviceMemEntry {
owner,
size,
#[cfg(feature = "cuda")]
device_ptr: 0,
}
}
#[test]
fn insert_then_lookup() {
let reg = DeviceMemRegistry::new();
let h = reg.insert(entry(InstanceId(1), 4096)).unwrap();
let found = reg.lookup(h, InstanceId(1)).unwrap();
assert_eq!(found.owner, InstanceId(1));
assert_eq!(found.size, 4096);
}
#[test]
fn lookup_wrong_owner_rejected() {
let reg = DeviceMemRegistry::new();
let h = reg.insert(entry(InstanceId(1), 4096)).unwrap();
assert_eq!(
reg.lookup(h, InstanceId(2)).unwrap_err(),
AbiError::InvalidHandle
);
}
#[test]
fn free_wrong_owner_rejected() {
let reg = DeviceMemRegistry::new();
let h = reg.insert(entry(InstanceId(1), 4096)).unwrap();
assert_eq!(
reg.free(h, InstanceId(2)).unwrap_err(),
AbiError::InvalidHandle
);
assert!(reg.lookup(h, InstanceId(1)).is_ok());
}
#[test]
fn free_unknown_rejected() {
let reg = DeviceMemRegistry::new();
assert_eq!(
reg.free(999, InstanceId(1)).unwrap_err(),
AbiError::InvalidHandle
);
}
#[test]
fn alloc_free_lifecycle_tracks_bytes() {
let reg = DeviceMemRegistry::new();
assert!(reg.is_empty());
let h = reg.insert(entry(InstanceId(1), 8192)).unwrap();
assert_eq!(reg.len(), 1);
assert_eq!(reg.total_device_bytes(), 8192);
let freed = reg.free(h, InstanceId(1)).unwrap();
assert_eq!(freed.size, 8192);
assert!(reg.is_empty());
assert_eq!(reg.total_device_bytes(), 0);
assert_eq!(
reg.free(h, InstanceId(1)).unwrap_err(),
AbiError::InvalidHandle
);
}
fn iso_reg() -> DeviceMemRegistry {
DeviceMemRegistry::with_budget(Arc::new(DeviceMemBudget::new()))
}
#[test]
fn aggregate_byte_cap_enforced() {
let reg = iso_reg();
let per = MAX_DEVICE_ALLOC_BYTES; let cap_count = (MAX_TOTAL_DEVICE_BYTES / per) as usize; for _ in 0..cap_count {
reg.insert(entry(InstanceId(1), per)).expect("under cap");
}
assert_eq!(
reg.insert(entry(InstanceId(1), per)).unwrap_err(),
AbiError::QuotaExceeded
);
}
#[test]
fn per_instance_count_cap_enforced() {
let reg = iso_reg();
for _ in 0..MAX_DEVICE_ALLOCS_PER_INSTANCE {
reg.insert(entry(InstanceId(1), 1))
.expect("under count cap");
}
assert_eq!(
reg.insert(entry(InstanceId(1), 1)).unwrap_err(),
AbiError::QuotaExceeded
);
}
#[test]
fn process_byte_cap_enforced_across_registries() {
let budget = Arc::new(DeviceMemBudget::new());
let per = MAX_DEVICE_ALLOC_BYTES; let per_reg = (MAX_TOTAL_DEVICE_BYTES / per) as usize; let regs: Vec<DeviceMemRegistry> = (0..4)
.map(|_| DeviceMemRegistry::with_budget(Arc::clone(&budget)))
.collect();
for (i, reg) in regs.iter().enumerate() {
for _ in 0..per_reg {
reg.insert(entry(InstanceId(i as u128 + 1), per))
.expect("each registry stays under its own per-instance cap");
}
}
assert_eq!(budget.total_bytes(), MAX_PROCESS_DEVICE_BYTES);
let extra = DeviceMemRegistry::with_budget(Arc::clone(&budget));
assert_eq!(
extra.insert(entry(InstanceId(99), per)).unwrap_err(),
AbiError::QuotaExceeded
);
let freed_handle = 1u64; regs[0]
.free(freed_handle, InstanceId(1))
.expect("free a live allocation");
assert_eq!(budget.total_bytes(), MAX_PROCESS_DEVICE_BYTES - per);
let h = extra
.insert(entry(InstanceId(99), per))
.expect("re-admitted after a free freed shared headroom");
assert!(extra.lookup(h, InstanceId(99)).is_ok());
}
#[test]
fn process_alloc_count_cap_enforced_across_registries() {
let budget = Arc::new(DeviceMemBudget::new());
let regs: Vec<DeviceMemRegistry> = (0..4)
.map(|_| DeviceMemRegistry::with_budget(Arc::clone(&budget)))
.collect();
for reg in ®s {
for _ in 0..MAX_DEVICE_ALLOCS_PER_INSTANCE {
reg.insert(entry(InstanceId(1), 1))
.expect("under count cap");
}
}
assert_eq!(budget.total_allocs(), MAX_PROCESS_DEVICE_ALLOCS as u64);
let extra = DeviceMemRegistry::with_budget(Arc::clone(&budget));
assert_eq!(
extra.insert(entry(InstanceId(1), 1)).unwrap_err(),
AbiError::QuotaExceeded
);
}
#[test]
fn free_credits_shared_budget() {
let budget = Arc::new(DeviceMemBudget::new());
let reg = DeviceMemRegistry::with_budget(Arc::clone(&budget));
let h = reg.insert(entry(InstanceId(1), 8192)).unwrap();
assert_eq!(budget.total_bytes(), 8192);
assert_eq!(budget.total_allocs(), 1);
reg.free(h, InstanceId(1)).unwrap();
assert_eq!(budget.total_bytes(), 0);
assert_eq!(budget.total_allocs(), 0);
}
#[test]
fn handles_are_unique_and_increasing() {
let reg = DeviceMemRegistry::new();
let a = reg.insert(entry(InstanceId(1), 1)).unwrap();
let b = reg.insert(entry(InstanceId(1), 1)).unwrap();
assert_ne!(a, b);
assert_eq!(a + 1, b);
}
}