#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, vec::Vec};
use crate::sync::{Arc, ArcSlice, new_arc_slice};
use crate::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering};
use core::hash::Hash;
use crate::arena::Arena;
use crate::storage::{Cache, Node};
use crate::filters::{T1, T2};
use crate::cache::{WorkerState, GLOBAL_EPOCH};
const MAX_RANK: u8 = 3;
pub struct CoreCache<K, V> {
pub arena: Arena,
pub t1: Arc<T1<K, V>>,
pub t2: Arc<T2<K, V>>,
pub cache: Arc<Cache<K, V>>,
pub admission: Arc<AdmissionFilter>,
pub garbage_queue: Vec<(*mut Node<K, V>, usize)>,
pub hit_accumulator: Vec<usize>,
pub is_cold_start: Arc<AtomicBool>,
pub epoch: Arc<AtomicU32>,
pub duration: u32,
pub worker_states: ArcSlice<WorkerState>,
}
unsafe impl<K: Send, V: Send> Send for CoreCache<K, V> {}
impl<K, V> CoreCache<K, V>
where
K: Hash + Eq + Send + Sync + Clone + 'static,
V: Send + Sync + Clone + 'static,
{
#[allow(clippy::too_many_arguments)]
pub fn new(
capacity: usize,
t1: Arc<T1<K, V>>,
t2: Arc<T2<K, V>>,
cache: Arc<Cache<K, V>>,
epoch: Arc<AtomicU32>,
duration: u32,
worker_states: ArcSlice<WorkerState>,
is_cold_start: Arc<AtomicBool>,
) -> Self {
Self {
arena: Arena::new(capacity),
t1,
t2,
cache,
admission: Arc::new(AdmissionFilter::new(capacity)),
garbage_queue: Vec::new(),
hit_accumulator: Vec::with_capacity(8192),
is_cold_start,
epoch,
duration,
worker_states,
}
}
#[inline(always)]
pub fn handle_admission_insert(&mut self, k: K, v: V, hash: u64, is_t1: bool) {
if is_t1 {
self.handle_insert_t1(k, v, hash);
return;
}
let cold_start = self.arena.free_list_len() > self.arena.capacity / 20;
self.is_cold_start.store(cold_start, Ordering::Relaxed);
if cold_start || self.admission.check_ghost(hash) {
self.handle_insert_with_hash(k, v, hash);
}
}
#[inline(always)]
pub fn handle_insert_with_hash(&mut self, k: K, v: V, hash: u64) {
let tag = (hash >> 48) as u16;
let global_idx = if let Some(existing_idx) = self.cache.index_probe(hash, tag) {
existing_idx
} else {
if self.arena.free_list_empty() {
self.evict_batch();
}
if let Some(new_idx) = self.arena.pop_free_slot() {
new_idx
} else {
return;
}
};
let entry = (tag as u64) << 48 | (global_idx as u64 & 0x0000_FFFF_FFFF_FFFF);
let node_ptr = Box::into_raw(Box::new(Node {
key: k,
value: v,
expire_at: self.epoch.load(Ordering::Relaxed) + self.duration,
g_idx: global_idx as u32,
}));
let old_ptr = self.cache.nodes[global_idx].swap(node_ptr, Ordering::Release);
if !old_ptr.is_null() {
let epoch = GLOBAL_EPOCH.load(Ordering::Relaxed);
self.garbage_queue.push((old_ptr, epoch));
}
self.cache.index_store(hash, tag, entry);
self.arena.set_hash(global_idx, hash);
self.arena.set_rank(global_idx, MAX_RANK);
}
#[inline(always)]
pub fn handle_insert_t1(&mut self, k: K, v: V, hash: u64) {
let tag = (hash >> 48) as u16;
let global_idx = if let Some(existing_idx) = self.cache.index_probe(hash, tag) {
existing_idx
} else {
if self.arena.free_list_empty() {
self.evict_batch();
}
if let Some(new_idx) = self.arena.pop_free_slot() {
new_idx
} else {
return;
}
};
let entry = (tag as u64) << 48 | (global_idx as u64 & 0x0000_FFFF_FFFF_FFFF);
let node_ptr = Box::into_raw(Box::new(Node {
key: k,
value: v,
expire_at: self.epoch.load(Ordering::Relaxed) + self.duration,
g_idx: global_idx as u32,
}));
let old_ptr = self.cache.nodes[global_idx].swap(node_ptr, Ordering::Release);
if !old_ptr.is_null() {
let epoch = GLOBAL_EPOCH.load(Ordering::Relaxed);
self.garbage_queue.push((old_ptr, epoch));
}
self.cache.index_store(hash, tag, entry);
self.arena.set_hash(global_idx, hash);
self.arena.set_rank(global_idx, 255);
self.t1.store_slot(hash, node_ptr);
}
#[inline(always)]
pub fn handle_remove(&mut self, _k: K, hash: u64) {
let tag = (hash >> 48) as u16;
if let Some(g_idx) = self.cache.index_probe(hash, tag) {
let old_ptr =
self.cache.nodes[g_idx].swap(core::ptr::null_mut(), Ordering::Release);
if !old_ptr.is_null() {
let epoch = GLOBAL_EPOCH.load(Ordering::Relaxed);
self.garbage_queue.push((old_ptr, epoch));
self.t1.clear_if_matches(hash, old_ptr);
self.t2.clear_if_matches(hash, old_ptr);
}
self.cache.index_remove(hash, tag, g_idx);
self.arena.set_rank(g_idx, 0);
}
}
pub fn handle_clear(&mut self) {
self.cache.clear();
for i in 0..self.t1.len() {
self.t1.clear_at(i);
}
for i in 0..self.t2.len() {
self.t2.clear_at(i);
}
self.admission.clear();
self.arena.clear();
self.is_cold_start.store(true, Ordering::Relaxed);
}
pub fn maintenance(&mut self) {
if !self.garbage_queue.is_empty() {
let current_global = GLOBAL_EPOCH.load(Ordering::Relaxed);
GLOBAL_EPOCH.store(current_global + 1, Ordering::Release);
let mut min_active_epoch = current_global + 1;
for state in self.worker_states.iter() {
let local = state.local_epoch.load(Ordering::Acquire);
if local != 0 && local < min_active_epoch {
min_active_epoch = local;
}
}
self.garbage_queue.retain(|&(ptr, epoch)| {
if epoch < min_active_epoch {
unsafe { drop(Box::from_raw(ptr)) };
false
} else {
true
}
});
}
if !self.hit_accumulator.is_empty() {
self.hit_accumulator.sort_unstable();
for &g_idx in &self.hit_accumulator {
self.arena.set_rank(g_idx, MAX_RANK);
let hash = self.arena.get_hash(g_idx);
let ptr = self.cache.nodes[g_idx].load(Ordering::Acquire);
if !ptr.is_null() && self.t1.load_slot(hash) != ptr {
self.t1.store_slot(hash, ptr);
}
}
self.hit_accumulator.clear();
}
if self.arena.free_list_len() < self.arena.capacity / 10 {
self.evict_batch();
}
let cold_start = self.arena.free_list_len() > self.arena.capacity / 20;
if self.is_cold_start.load(Ordering::Relaxed) != cold_start {
self.is_cold_start.store(cold_start, Ordering::Relaxed);
}
}
#[inline(always)]
pub fn process_hits(&mut self, batch: &[usize]) {
for &g_idx in batch.iter() {
if g_idx < self.arena.capacity && g_idx != usize::MAX {
self.hit_accumulator.push(g_idx);
}
}
}
pub fn evict_batch(&mut self) {
let count = 128;
let avg = (self.arena.count_sum() / self.arena.capacity as u64) as u8;
let threshold = avg.max(1);
for _ in 0..count {
if self.arena.free_list_len() > self.arena.capacity / 10 {
break;
}
let idx = self.arena.cursor();
let r = self.arena.get_rank(idx);
if r <= threshold {
let hash = self.arena.get_hash(idx);
let tag = (hash >> 48) as u16;
let old_ptr =
self.cache.nodes[idx].swap(core::ptr::null_mut(), Ordering::Release);
if !old_ptr.is_null() {
let epoch = GLOBAL_EPOCH.load(Ordering::Relaxed);
self.garbage_queue.push((old_ptr, epoch));
self.t1.clear_if_matches(hash, old_ptr);
self.t2.clear_if_matches(hash, old_ptr);
}
self.cache.index_remove(hash, tag, idx);
self.admission.record_death(hash);
self.arena.push_free_slot(idx);
self.arena.set_rank(idx, 0);
} else {
self.arena.decrement_rank(idx);
}
self.arena.advance_cursor();
}
}
}
pub struct AdmissionFilter {
pub ghost_mask: usize,
pub ghost_set: ArcSlice<AtomicU16>,
}
impl AdmissionFilter {
pub fn new(capacity: usize) -> Self {
let ghost_size = capacity.max(256);
let mut ghost_vec = Vec::with_capacity(ghost_size);
for _ in 0..ghost_size {
ghost_vec.push(AtomicU16::new(0));
}
Self {
ghost_mask: ghost_size - 1,
ghost_set: new_arc_slice(ghost_vec),
}
}
#[inline(always)]
pub fn record_death(&self, hash: u64) {
let fp = (hash >> 48) as u16;
let idx = (hash as usize) & self.ghost_mask;
self.ghost_set[idx].store(fp, Ordering::Relaxed);
}
#[inline(always)]
pub fn check_ghost(&self, hash: u64) -> bool {
let fp = (hash >> 48) as u16;
let ghost_idx = (hash as usize) & self.ghost_mask;
self.ghost_set[ghost_idx].load(Ordering::Relaxed) == fp
}
pub fn clear(&self) {
for val in self.ghost_set.iter() {
val.store(0, Ordering::Relaxed);
}
}
}