use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(feature = "cuda")]
use std::sync::Arc;
use dashmap::DashMap;
use tensor_wasm_core::types::{InstanceId, KernelId};
use crate::abi::{AbiError, MAX_KERNELS_PER_INSTANCE, MAX_PTX_BYTES};
pub const MAX_TOTAL_PTX_BYTES: usize = 64 * MAX_PTX_BYTES;
#[derive(Debug)]
pub struct KernelEntry {
pub owner: InstanceId,
pub entry: String,
pub ptx_bytes_len: usize,
#[cfg(feature = "cuda")]
pub module: Option<Arc<cust::module::Module>>,
}
pub struct KernelRegistry {
next_id: AtomicU64,
entries: DashMap<KernelId, KernelEntry>,
total_ptx_bytes: AtomicU64,
}
impl Default for KernelRegistry {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone, Debug)]
pub struct KernelHandle {
pub owner: InstanceId,
pub entry: String,
pub ptx_bytes_len: usize,
#[cfg(feature = "cuda")]
pub module: Option<Arc<cust::module::Module>>,
}
fn random_kernel_id_seed() -> u64 {
use std::collections::hash_map::RandomState;
use std::hash::{BuildHasher, Hasher};
let mut h = RandomState::new().build_hasher();
h.write_u64(0xa5a5_a5a5_a5a5_a5a5);
let raw = h.finish();
(raw & 0x7fff_ffff_ffff_0000) | 1
}
impl KernelRegistry {
pub fn new() -> Self {
Self {
next_id: AtomicU64::new(random_kernel_id_seed()),
entries: DashMap::new(),
total_ptx_bytes: AtomicU64::new(0),
}
}
pub fn register(&self, entry: KernelEntry) -> Result<KernelId, AbiError> {
if self.entries.len() >= MAX_KERNELS_PER_INSTANCE {
return Err(AbiError::QuotaExceeded);
}
let add = entry.ptx_bytes_len as u64;
let mut current = self.total_ptx_bytes.load(Ordering::Acquire);
loop {
let next = current.saturating_add(add);
if next > MAX_TOTAL_PTX_BYTES as u64 {
return Err(AbiError::QuotaExceeded);
}
match self.total_ptx_bytes.compare_exchange_weak(
current,
next,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => break,
Err(observed) => current = observed,
}
}
let id = KernelId(self.next_id.fetch_add(1, Ordering::Relaxed));
self.entries.insert(id, entry);
Ok(id)
}
pub fn lookup(&self, id: KernelId, owner: InstanceId) -> Result<KernelHandle, AbiError> {
let r = self.entries.get(&id).ok_or(AbiError::InvalidKernel)?;
if r.owner != owner {
return Err(AbiError::InvalidKernel);
}
Ok(KernelHandle {
owner: r.owner,
entry: r.entry.clone(),
ptx_bytes_len: r.ptx_bytes_len,
#[cfg(feature = "cuda")]
module: r.module.clone(),
})
}
pub fn remove(&self, id: KernelId) -> Option<KernelEntry> {
let (_, entry) = self.entries.remove(&id)?;
let _ = self
.total_ptx_bytes
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |cur| {
Some(cur.saturating_sub(entry.ptx_bytes_len as u64))
});
Some(entry)
}
pub fn len(&self) -> usize {
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.entries.is_empty()
}
pub fn total_ptx_bytes(&self) -> u64 {
self.total_ptx_bytes.load(Ordering::Acquire)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_entry(owner: InstanceId, entry: &str) -> KernelEntry {
KernelEntry {
owner,
entry: entry.into(),
ptx_bytes_len: 1024,
#[cfg(feature = "cuda")]
module: None,
}
}
fn make_entry_sized(owner: InstanceId, entry: &str, bytes: usize) -> KernelEntry {
KernelEntry {
owner,
entry: entry.into(),
ptx_bytes_len: bytes,
#[cfg(feature = "cuda")]
module: None,
}
}
#[test]
fn register_then_lookup() {
let reg = KernelRegistry::new();
let id = reg
.register(make_entry(InstanceId(1), "vector_add"))
.unwrap();
let entry = reg.lookup(id, InstanceId(1)).unwrap();
assert_eq!(entry.owner, InstanceId(1));
assert_eq!(entry.entry, "vector_add");
}
#[test]
fn lookup_wrong_owner_rejected() {
let reg = KernelRegistry::new();
let id = reg
.register(make_entry(InstanceId(1), "vector_add"))
.unwrap();
let err = reg.lookup(id, InstanceId(2)).unwrap_err();
assert_eq!(err, AbiError::InvalidKernel);
}
#[test]
fn lookup_unknown_rejected() {
let reg = KernelRegistry::new();
let err = reg.lookup(KernelId(42), InstanceId(1)).unwrap_err();
assert_eq!(err, AbiError::InvalidKernel);
}
#[test]
fn remove_drops_entry() {
let reg = KernelRegistry::new();
let id = reg
.register(make_entry(InstanceId(1), "vector_add"))
.unwrap();
assert!(reg.remove(id).is_some());
assert!(reg.lookup(id, InstanceId(1)).is_err());
}
#[test]
fn ids_are_unique_and_increasing() {
let reg = KernelRegistry::new();
let a = reg.register(make_entry(InstanceId(1), "k1")).unwrap();
let b = reg.register(make_entry(InstanceId(1), "k2")).unwrap();
let c = reg.register(make_entry(InstanceId(1), "k3")).unwrap();
assert_ne!(a, b);
assert_ne!(b, c);
assert_eq!(a.0 + 1, b.0);
assert_eq!(b.0 + 1, c.0);
}
#[test]
fn len_tracks_entries() {
let reg = KernelRegistry::new();
assert!(reg.is_empty());
let id = reg.register(make_entry(InstanceId(1), "k")).unwrap();
assert_eq!(reg.len(), 1);
reg.remove(id);
assert!(reg.is_empty());
}
#[test]
fn aggregate_byte_cap_enforced() {
let reg = KernelRegistry::new();
let per = MAX_PTX_BYTES; let cap_count = MAX_TOTAL_PTX_BYTES / per; for i in 0..cap_count {
reg.register(make_entry_sized(InstanceId(1), &format!("k{i}"), per))
.expect("under aggregate cap");
}
let err = reg
.register(make_entry_sized(InstanceId(1), "k_over", per))
.unwrap_err();
assert_eq!(err, AbiError::QuotaExceeded);
}
#[test]
fn aggregate_bytes_counter_decreases_on_remove() {
let reg = KernelRegistry::new();
let id = reg
.register(make_entry_sized(InstanceId(1), "k", 4096))
.unwrap();
assert_eq!(reg.total_ptx_bytes(), 4096);
reg.remove(id);
assert_eq!(reg.total_ptx_bytes(), 0);
}
#[test]
fn lookup_returns_independent_handle() {
let reg = KernelRegistry::new();
let id = reg.register(make_entry(InstanceId(1), "v")).unwrap();
let handle = reg.lookup(id, InstanceId(1)).unwrap();
assert!(reg.remove(id).is_some());
assert_eq!(handle.entry, "v");
assert_eq!(handle.owner, InstanceId(1));
}
}