use atomic_float::AtomicF64;
use crossbeam_channel::{Receiver, Sender, bounded};
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use std::hash::Hash;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use tracing::{debug, info};
pub struct Cache<K, V> {
data: DashMap<K, CacheEntry<V>>,
capacity: AtomicUsize,
current_size: AtomicUsize,
global_access_counter: AtomicU64,
total_requests: AtomicU64,
hits: AtomicU64,
misses: AtomicU64,
evictions: AtomicU64,
total_lookup_time_ns: AtomicU64,
hit_rate: AtomicF64,
avg_lookup_time_ns: AtomicU64,
name: String,
created_at: AtomicU64,
eviction_sender: Sender<EvictionEvent<K>>,
eviction_receiver: Receiver<EvictionEvent<K>>,
}
#[derive(Debug)]
struct CacheEntry<V> {
value: V,
access_count: AtomicU64,
last_accessed: AtomicU64,
#[allow(dead_code)]
created_at: AtomicU64,
}
#[derive(Debug, Clone)]
pub enum EvictionEvent<K> {
#[allow(dead_code)]
ShouldEvict {
key: K,
access_count: u64,
last_accessed: u64,
},
Evicted {
#[allow(dead_code)]
key: K,
#[allow(dead_code)]
timestamp: u64,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheStats {
pub name: String,
pub current_size: usize,
pub capacity: usize,
pub utilization: f64,
pub total_requests: u64,
pub hits: u64,
pub misses: u64,
pub evictions: u64,
pub hit_rate: f64,
pub miss_rate: f64,
pub avg_lookup_time_ns: u64,
pub created_at: u64,
pub uptime_seconds: u64,
}
impl<V> CacheEntry<V> {
fn new_with_logical_time(value: V, logical_time: u64) -> Self {
Self {
value,
access_count: AtomicU64::new(1),
last_accessed: AtomicU64::new(logical_time),
created_at: AtomicU64::new(logical_time),
}
}
fn touch(&self, global_counter: &AtomicU64) -> u64 {
let logical_time = global_counter.fetch_add(1, Ordering::Relaxed) + 1;
self.last_accessed.store(logical_time, Ordering::Relaxed);
self.access_count.fetch_add(1, Ordering::Relaxed) + 1
}
fn priority_score(&self, current_logical_time: u64) -> u64 {
let access_count = self.access_count.load(Ordering::Relaxed);
let last_accessed = self.last_accessed.load(Ordering::Relaxed);
let recency_factor = current_logical_time.saturating_sub(last_accessed);
let frequency_factor = if access_count > 0 {
1000 / access_count
} else {
1000
};
recency_factor + frequency_factor
}
}
impl<K, V> Cache<K, V>
where
K: Hash + Eq + Clone + Send + Sync + 'static,
V: Clone + Send + Sync + 'static,
{
pub fn new(capacity: usize, name: String) -> Result<Self, String> {
if capacity == 0 {
return Err("Cache capacity must be greater than 0".to_string());
}
let (eviction_sender, eviction_receiver) = bounded(1000);
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
Ok(Self {
data: DashMap::new(),
capacity: AtomicUsize::new(capacity),
current_size: AtomicUsize::new(0),
global_access_counter: AtomicU64::new(0),
total_requests: AtomicU64::new(0),
hits: AtomicU64::new(0),
misses: AtomicU64::new(0),
evictions: AtomicU64::new(0),
total_lookup_time_ns: AtomicU64::new(0),
hit_rate: AtomicF64::new(0.0),
avg_lookup_time_ns: AtomicU64::new(0),
name,
created_at: AtomicU64::new(now),
eviction_sender,
eviction_receiver,
})
}
#[allow(clippy::option_if_let_else)]
pub fn get(&self, key: &K) -> Option<V> {
let start = Instant::now();
let total_requests = self.total_requests.fetch_add(1, Ordering::Relaxed) + 1;
let result = if let Some(entry_ref) = self.data.get(key) {
let access_count = entry_ref.value().touch(&self.global_access_counter);
let value = entry_ref.value().value.clone();
let hits = self.hits.fetch_add(1, Ordering::Relaxed) + 1;
#[allow(clippy::cast_precision_loss)]
let hit_rate = hits as f64 / total_requests as f64;
self.hit_rate.store(hit_rate, Ordering::Relaxed);
debug!(
"{} Cache: HIT (access_count: {})",
self.name, access_count
);
Some(value)
} else {
self.misses.fetch_add(1, Ordering::Relaxed);
let hits = self.hits.load(Ordering::Relaxed);
#[allow(clippy::cast_precision_loss)]
let hit_rate = hits as f64 / total_requests as f64;
self.hit_rate.store(hit_rate, Ordering::Relaxed);
debug!("{} Cache: MISS", self.name);
None
};
#[allow(clippy::cast_possible_truncation)]
let lookup_time_ns = start.elapsed().as_nanos() as u64;
let total_lookup_time = self
.total_lookup_time_ns
.fetch_add(lookup_time_ns, Ordering::Relaxed)
+ lookup_time_ns;
let avg_lookup_time = total_lookup_time / total_requests;
self.avg_lookup_time_ns
.store(avg_lookup_time, Ordering::Relaxed);
result
}
pub fn put(&self, key: K, value: V) -> Option<V> {
let capacity = self.capacity.load(Ordering::Relaxed);
let logical_time = self.global_access_counter.fetch_add(1, Ordering::Relaxed) + 1;
let entry = CacheEntry::new_with_logical_time(value, logical_time);
let current_size = self.current_size.load(Ordering::Relaxed);
if current_size >= capacity {
self.try_evict();
}
let old_value = self.data.insert(key, entry).map(|old_entry| {
old_entry.value
});
if old_value.is_none() {
self.current_size.fetch_add(1, Ordering::Relaxed);
}
old_value
}
fn try_evict(&self) {
let capacity = self.capacity.load(Ordering::Relaxed);
let current_size = self.current_size.load(Ordering::Relaxed);
if current_size < capacity {
return; }
let current_logical_time = self.global_access_counter.load(Ordering::Relaxed);
let mut eviction_candidate: Option<(K, u64)> = None;
let mut highest_eviction_priority = 0;
let sample_size = std::cmp::min(20, self.data.len());
for (sampled, entry_ref) in self.data.iter().enumerate() {
if sampled >= sample_size {
break;
}
let priority = entry_ref.value().priority_score(current_logical_time);
if priority > highest_eviction_priority {
highest_eviction_priority = priority;
eviction_candidate = Some((entry_ref.key().clone(), priority));
}
}
if let Some((key, _priority)) = eviction_candidate
&& let Some((_key, _old_entry)) = self.data.remove(&key)
{
self.current_size.fetch_sub(1, Ordering::Relaxed);
self.evictions.fetch_add(1, Ordering::Relaxed);
let _ = self.eviction_sender.try_send(EvictionEvent::Evicted {
key,
timestamp: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs(),
});
debug!("{} Cache: Evicted entry", self.name);
}
}
pub fn contains_key(&self, key: &K) -> bool {
self.data.contains_key(key)
}
pub fn remove(&self, key: &K) -> Option<V> {
if let Some((_key, entry)) = self.data.remove(key) {
self.current_size.fetch_sub(1, Ordering::Relaxed);
Some(entry.value)
} else {
None
}
}
pub fn clear(&self) {
let old_size = self.current_size.load(Ordering::Relaxed);
self.data.clear();
self.current_size.store(0, Ordering::Relaxed);
if old_size > 0 {
info!("{} Cache: Cleared {} entries", self.name, old_size);
}
}
pub fn len(&self) -> usize {
self.current_size.load(Ordering::Relaxed)
}
pub fn is_empty(&self) -> bool {
self.current_size.load(Ordering::Relaxed) == 0
}
pub fn capacity(&self) -> usize {
self.capacity.load(Ordering::Relaxed)
}
pub fn get_stats(&self) -> CacheStats {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
let created_at = self.created_at.load(Ordering::Relaxed);
let current_size = self.current_size.load(Ordering::Relaxed);
let capacity = self.capacity.load(Ordering::Relaxed);
CacheStats {
name: self.name.clone(),
current_size,
capacity,
utilization: if capacity > 0 {
#[allow(clippy::cast_precision_loss)]
{
current_size as f64 / capacity as f64
}
} else {
0.0
},
total_requests: self.total_requests.load(Ordering::Relaxed),
hits: self.hits.load(Ordering::Relaxed),
misses: self.misses.load(Ordering::Relaxed),
evictions: self.evictions.load(Ordering::Relaxed),
hit_rate: self.hit_rate.load(Ordering::Relaxed),
miss_rate: 1.0 - self.hit_rate.load(Ordering::Relaxed),
avg_lookup_time_ns: self.avg_lookup_time_ns.load(Ordering::Relaxed),
created_at,
uptime_seconds: now.saturating_sub(created_at),
}
}
pub fn reset_metrics(&self) {
self.total_requests.store(0, Ordering::Relaxed);
self.hits.store(0, Ordering::Relaxed);
self.misses.store(0, Ordering::Relaxed);
self.evictions.store(0, Ordering::Relaxed);
self.total_lookup_time_ns.store(0, Ordering::Relaxed);
self.hit_rate.store(0.0, Ordering::Relaxed);
self.avg_lookup_time_ns.store(0, Ordering::Relaxed);
info!("{} Cache: Metrics reset", self.name);
}
pub fn resize(&self, new_capacity: usize) -> Result<(), String> {
if new_capacity == 0 {
return Err("Cache capacity must be greater than 0".to_string());
}
let old_capacity = self.capacity.swap(new_capacity, Ordering::Relaxed);
let current_size = self.current_size.load(Ordering::Relaxed);
if new_capacity < current_size {
let evictions_needed = current_size - new_capacity;
for _ in 0..evictions_needed {
self.try_evict();
}
}
info!(
"{} Cache: Resized from {} to {} capacity",
self.name, old_capacity, new_capacity
);
Ok(())
}
pub fn has_performance_issues(&self) -> bool {
let stats = self.get_stats();
if stats.hit_rate < 0.3 && stats.total_requests > 100 {
return true;
}
if stats.avg_lookup_time_ns > 1_000_000 {
return true;
}
let eviction_rate = if stats.total_requests > 0 {
#[allow(clippy::cast_precision_loss)]
{
stats.evictions as f64 / stats.total_requests as f64
}
} else {
0.0
};
if eviction_rate > 0.5 {
return true;
}
false
}
pub fn get_recommendations(&self) -> Vec<String> {
let mut recommendations = Vec::new();
let stats = self.get_stats();
if stats.hit_rate < 0.5 && stats.total_requests > 100 {
recommendations.push(format!(
"Consider increasing cache size. Current hit rate: {:.1}%",
stats.hit_rate * 100.0
));
}
if stats.utilization > 0.9 {
recommendations.push("Cache is nearly full, consider increasing capacity".to_string());
}
if stats.avg_lookup_time_ns > 500_000 {
recommendations.push(format!(
"Slow cache lookups detected: {}µs average",
stats.avg_lookup_time_ns / 1000
));
}
let eviction_rate = if stats.total_requests > 0 {
#[allow(clippy::cast_precision_loss)]
{
stats.evictions as f64 / stats.total_requests as f64
}
} else {
0.0
};
if eviction_rate > 0.2 {
recommendations.push(format!(
"High eviction rate: {:.1}%, consider increasing capacity",
eviction_rate * 100.0
));
}
recommendations
}
pub fn keys(&self) -> Vec<K> {
self.data.iter().map(|entry| entry.key().clone()).collect()
}
pub fn values(&self) -> Vec<V> {
self.data
.iter()
.map(|entry| entry.value().value.clone())
.collect()
}
pub fn for_each<F>(&self, mut f: F)
where
F: FnMut(&K, &V),
{
for entry_ref in &self.data {
f(entry_ref.key(), &entry_ref.value().value);
}
}
pub fn drain_eviction_events(&self) -> Vec<EvictionEvent<K>> {
let mut events = Vec::new();
while let Ok(event) = self.eviction_receiver.try_recv() {
events.push(event);
}
events
}
}
pub type SessionCache<K, V> = Cache<K, V>;
pub struct CacheManager {
caches: DashMap<String, Arc<dyn CacheProvider + Send + Sync>>,
total_requests: AtomicU64,
total_hits: AtomicU64,
total_misses: AtomicU64,
total_evictions: AtomicU64,
}
pub trait CacheProvider {
fn get_stats(&self) -> CacheStats;
fn reset_metrics(&self);
fn has_performance_issues(&self) -> bool;
fn get_recommendations(&self) -> Vec<String>;
}
impl<K, V> CacheProvider for Cache<K, V>
where
K: Hash + Eq + Clone + Send + Sync + 'static,
V: Clone + Send + Sync + 'static,
{
fn get_stats(&self) -> CacheStats {
self.get_stats()
}
fn reset_metrics(&self) {
self.total_requests.store(0, Ordering::Relaxed);
self.hits.store(0, Ordering::Relaxed);
self.misses.store(0, Ordering::Relaxed);
self.evictions.store(0, Ordering::Relaxed);
self.total_lookup_time_ns.store(0, Ordering::Relaxed);
self.hit_rate.store(0.0, Ordering::Relaxed);
self.avg_lookup_time_ns.store(0, Ordering::Relaxed);
}
fn has_performance_issues(&self) -> bool {
self.has_performance_issues()
}
fn get_recommendations(&self) -> Vec<String> {
self.get_recommendations()
}
}
impl CacheManager {
#[must_use]
pub fn new() -> Self {
Self {
caches: DashMap::new(),
total_requests: AtomicU64::new(0),
total_hits: AtomicU64::new(0),
total_misses: AtomicU64::new(0),
total_evictions: AtomicU64::new(0),
}
}
pub fn register_cache<K, V>(&self, name: &str, cache: Cache<K, V>)
where
K: Hash + Eq + Clone + Send + Sync + 'static,
V: Clone + Send + Sync + 'static,
{
self.caches.insert(name.to_string(), Arc::new(cache));
info!("Registered cache: {}", name);
}
pub fn get_all_stats(&self) -> Vec<CacheStats> {
self.caches
.iter()
.map(|entry| entry.value().get_stats())
.collect()
}
pub fn has_any_performance_issues(&self) -> bool {
self.caches
.iter()
.any(|entry| entry.value().has_performance_issues())
}
pub fn reset_all_metrics(&self) {
for entry in &self.caches {
entry.value().reset_metrics();
}
self.total_requests.store(0, Ordering::Relaxed);
self.total_hits.store(0, Ordering::Relaxed);
self.total_misses.store(0, Ordering::Relaxed);
self.total_evictions.store(0, Ordering::Relaxed);
info!("Reset all cache metrics");
}
pub fn get_summary(&self) -> CacheManagerSummary {
let stats: Vec<CacheStats> = self.get_all_stats();
let cache_count = stats.len();
let total_requests: u64 = stats.iter().map(|s| s.total_requests).sum();
let total_hits: u64 = stats.iter().map(|s| s.hits).sum();
let total_evictions: u64 = stats.iter().map(|s| s.evictions).sum();
let avg_hit_rate = if total_requests > 0 {
#[allow(clippy::cast_precision_loss)]
{
total_hits as f64 / total_requests as f64
}
} else {
0.0
};
let problematic_caches: Vec<String> = self
.caches
.iter()
.filter_map(|entry| {
if entry.value().has_performance_issues() {
Some(entry.key().clone())
} else {
None
}
})
.collect();
CacheManagerSummary {
cache_count,
total_requests,
total_hits,
total_evictions,
average_hit_rate: avg_hit_rate,
problematic_caches,
individual_stats: stats,
}
}
}
impl Default for CacheManager {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CacheManagerSummary {
pub cache_count: usize,
pub total_requests: u64,
pub total_hits: u64,
pub total_evictions: u64,
pub average_hit_rate: f64,
pub problematic_caches: Vec<String>,
pub individual_stats: Vec<CacheStats>,
}
static GLOBAL_CACHE_MANAGER: std::sync::OnceLock<CacheManager> =
std::sync::OnceLock::new();
pub fn init_global_cache_manager() {
let _ = GLOBAL_CACHE_MANAGER.set(CacheManager::new());
}
pub fn get_global_cache_manager() -> &'static CacheManager {
GLOBAL_CACHE_MANAGER.get_or_init(CacheManager::new)
}
#[macro_export]
macro_rules! new_cache {
($capacity:expr, $name:expr) => {
$crate::core::cache::Cache::new($capacity, $name.to_string())
};
}