velesdb_core/cache/

lockfree.rs

//! Lock-Free LRU Cache with DashMap L1 (US-CORE-003-15)
//!
//! Two-tier cache architecture for maximum concurrent throughput:
//! - **L1**: DashMap (lock-free concurrent HashMap) for hot keys
//! - **L2**: LruCache (with LRU eviction) for capacity management
//!
//! # Performance
//!
//! | Operation | L1 Hit | L1 Miss + L2 Hit |
//! |-----------|--------|------------------|
//! | get() | ~50ns (lock-free) | ~500ns (with promotion) |
//! | peek() | ~30ns (L1 only) | N/A |
//! | insert() | ~100ns (write-through) | - |

use dashmap::DashMap;
use std::hash::Hash;
use std::sync::atomic::{AtomicU64, Ordering};

use super::LruCache;

/// Lock-free two-tier cache with DashMap L1 and LruCache L2.
///
/// Optimized for read-heavy workloads with hot keys.
pub struct LockFreeLruCache<K, V>
where
    K: Hash + Eq + Clone + Send + Sync + 'static,
    V: Clone + Send + Sync + 'static,
{
    /// L1: Lock-free concurrent cache for hot keys.
    l1: DashMap<K, V>,
    /// L2: LRU cache with eviction for capacity management.
    l2: LruCache<K, V>,
    /// Maximum L1 entries before eviction to L2.
    l1_capacity: usize,
    /// L1 hit counter.
    l1_hits: AtomicU64,
    /// L2 hit counter (L1 miss, L2 hit).
    l2_hits: AtomicU64,
    /// Total miss counter.
    misses: AtomicU64,
}

/// Statistics for the two-tier cache.
#[derive(Debug, Clone, Default)]
pub struct LockFreeCacheStats {
    /// L1 cache hits.
    pub l1_hits: u64,
    /// L2 cache hits (L1 miss, L2 hit).
    pub l2_hits: u64,
    /// Total misses.
    pub misses: u64,
    /// L1 current size.
    pub l1_size: usize,
    /// L2 current size.
    pub l2_size: usize,
}

impl LockFreeCacheStats {
    /// Calculate L1 hit rate.
    #[must_use]
    pub fn l1_hit_rate(&self) -> f64 {
        let total = self.l1_hits + self.l2_hits + self.misses;
        if total == 0 {
            0.0
        } else {
            self.l1_hits as f64 / total as f64
        }
    }

    /// Calculate total hit rate (L1 + L2).
    #[must_use]
    pub fn total_hit_rate(&self) -> f64 {
        let total = self.l1_hits + self.l2_hits + self.misses;
        if total == 0 {
            0.0
        } else {
            (self.l1_hits + self.l2_hits) as f64 / total as f64
        }
    }
}

impl<K, V> LockFreeLruCache<K, V>
where
    K: Hash + Eq + Clone + Send + Sync + 'static,
    V: Clone + Send + Sync + 'static,
{
    /// Create a new lock-free LRU cache.
    ///
    /// # Arguments
    ///
    /// * `l1_capacity` - Maximum entries in L1 (hot cache)
    /// * `l2_capacity` - Maximum entries in L2 (LRU backing store)
    #[must_use]
    pub fn new(l1_capacity: usize, l2_capacity: usize) -> Self {
        Self {
            l1: DashMap::with_capacity(l1_capacity),
            l2: LruCache::new(l2_capacity),
            l1_capacity,
            l1_hits: AtomicU64::new(0),
            l2_hits: AtomicU64::new(0),
            misses: AtomicU64::new(0),
        }
    }

    /// Get a value, checking L1 first then L2.
    ///
    /// If found in L2, promotes to L1 for faster subsequent access.
    #[must_use]
    pub fn get(&self, key: &K) -> Option<V> {
        // Fast path: L1 lookup (lock-free)
        if let Some(entry) = self.l1.get(key) {
            self.l1_hits.fetch_add(1, Ordering::Relaxed);
            return Some(entry.value().clone());
        }

        // Slow path: L2 lookup
        if let Some(value) = self.l2.get(key) {
            self.l2_hits.fetch_add(1, Ordering::Relaxed);
            // Promote to L1
            self.promote_to_l1(key.clone(), value.clone());
            return Some(value);
        }

        self.misses.fetch_add(1, Ordering::Relaxed);
        None
    }

    /// Peek at L1 only (ultra-fast, no promotion).
    ///
    /// Returns None if not in L1, even if present in L2.
    #[must_use]
    pub fn peek_l1(&self, key: &K) -> Option<V> {
        self.l1.get(key).map(|entry| entry.value().clone())
    }

    /// Peek at L2 only (no L1 check, no promotion).
    #[must_use]
    pub fn peek_l2(&self, key: &K) -> Option<V> {
        self.l2.peek(key)
    }

    /// Insert a key-value pair (write-through to L1 and L2).
    pub fn insert(&self, key: K, value: V) {
        // Write to L1
        self.l1.insert(key.clone(), value.clone());

        // Evict from L1 if over capacity
        self.maybe_evict_l1();

        // Write to L2 (backing store)
        self.l2.insert(key, value);
    }

    /// Remove a key from both L1 and L2.
    pub fn remove(&self, key: &K) {
        self.l1.remove(key);
        self.l2.remove(key);
    }

    /// Clear both L1 and L2.
    pub fn clear(&self) {
        self.l1.clear();
        self.l2.clear();
    }

    /// Get cache statistics.
    #[must_use]
    pub fn stats(&self) -> LockFreeCacheStats {
        LockFreeCacheStats {
            l1_hits: self.l1_hits.load(Ordering::Relaxed),
            l2_hits: self.l2_hits.load(Ordering::Relaxed),
            misses: self.misses.load(Ordering::Relaxed),
            l1_size: self.l1.len(),
            l2_size: self.l2.len(),
        }
    }

    /// Get L1 capacity.
    #[must_use]
    pub fn l1_capacity(&self) -> usize {
        self.l1_capacity
    }

    /// Get L2 capacity.
    #[must_use]
    pub fn l2_capacity(&self) -> usize {
        self.l2.capacity()
    }

    /// Promote a key from L2 to L1.
    fn promote_to_l1(&self, key: K, value: V) {
        self.l1.insert(key, value);
        self.maybe_evict_l1();
    }

    /// Evict entries from L1 if over capacity.
    /// Uses a bounded loop to prevent infinite spinning under contention.
    fn maybe_evict_l1(&self) {
        // Bounded eviction: max attempts to prevent infinite loop under contention
        let mut attempts = 0;
        let max_attempts = 10;

        while self.l1.len() > self.l1_capacity && attempts < max_attempts {
            attempts += 1;

            // Collect keys to remove (avoid holding iterator while removing)
            let keys_to_remove: Vec<K> = self
                .l1
                .iter()
                .take(self.l1.len().saturating_sub(self.l1_capacity).max(1))
                .map(|entry| entry.key().clone())
                .collect();

            if keys_to_remove.is_empty() {
                break;
            }

            for key in keys_to_remove {
                self.l1.remove(&key);
            }
        }
    }
}

impl<K, V> Default for LockFreeLruCache<K, V>
where
    K: Hash + Eq + Clone + Send + Sync + 'static,
    V: Clone + Send + Sync + 'static,
{
    fn default() -> Self {
        // Default: L1 = 1K hot entries, L2 = 10K LRU entries
        Self::new(1_000, 10_000)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::thread;
    use std::time::Instant;

    #[test]
    fn test_lockfree_cache_new() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        assert_eq!(cache.l1_capacity(), 100);
        assert_eq!(cache.l2_capacity(), 1000);
        assert_eq!(cache.stats().l1_size, 0);
        assert_eq!(cache.stats().l2_size, 0);
    }

    #[test]
    fn test_lockfree_cache_insert_and_get() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        cache.insert(1, "value_1".to_string());

        // Should be in both L1 and L2
        assert_eq!(cache.get(&1), Some("value_1".to_string()));
        assert_eq!(cache.peek_l1(&1), Some("value_1".to_string()));
        assert_eq!(cache.peek_l2(&1), Some("value_1".to_string()));
    }

    #[test]
    fn test_lockfree_cache_get_l1_hit() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        cache.insert(1, "value_1".to_string());

        // First get
        let _ = cache.get(&1);
        // Second get (should hit L1)
        let _ = cache.get(&1);

        let stats = cache.stats();
        assert_eq!(stats.l1_hits, 2);
        assert_eq!(stats.l2_hits, 0);
        assert_eq!(stats.misses, 0);
    }

    #[test]
    fn test_lockfree_cache_get_l2_promotion() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(10, 100);

        // Insert via L2 directly (simulate L1 eviction scenario)
        cache.l2.insert(99, "from_l2".to_string());

        // Get should find in L2 and promote to L1
        let result = cache.get(&99);
        assert_eq!(result, Some("from_l2".to_string()));

        let stats = cache.stats();
        assert_eq!(stats.l1_hits, 0);
        assert_eq!(stats.l2_hits, 1);

        // Now should be in L1
        assert!(cache.peek_l1(&99).is_some());

        // Second get should hit L1
        let _ = cache.get(&99);
        let stats = cache.stats();
        assert_eq!(stats.l1_hits, 1);
    }

    #[test]
    fn test_lockfree_cache_miss() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        let result = cache.get(&999);

        assert_eq!(result, None);
        assert_eq!(cache.stats().misses, 1);
    }

    #[test]
    fn test_lockfree_cache_remove() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        cache.insert(1, "value_1".to_string());
        assert!(cache.get(&1).is_some());

        cache.remove(&1);

        assert!(cache.peek_l1(&1).is_none());
        assert!(cache.peek_l2(&1).is_none());
    }

    #[test]
    fn test_lockfree_cache_clear() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        for i in 0..50 {
            cache.insert(i, format!("value_{i}"));
        }

        assert_eq!(cache.stats().l1_size, 50);

        cache.clear();

        assert_eq!(cache.stats().l1_size, 0);
        assert_eq!(cache.stats().l2_size, 0);
    }

    #[test]
    fn test_lockfree_cache_l1_eviction() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(10, 100);

        // Insert more than L1 capacity
        for i in 0..20 {
            cache.insert(i, format!("value_{i}"));
        }

        // L1 should be at or near capacity (bounded eviction may leave slightly over)
        assert!(
            cache.stats().l1_size <= 15,
            "L1 size {} should be <= 15",
            cache.stats().l1_size
        );
        // L2 should have all entries
        assert_eq!(cache.stats().l2_size, 20);
    }

    #[test]
    fn test_lockfree_cache_concurrent_reads() {
        let cache = Arc::new(LockFreeLruCache::<u64, String>::new(100, 1000));

        // Pre-populate
        for i in 0..100 {
            cache.insert(i, format!("value_{i}"));
        }

        let mut handles = vec![];

        // 8 threads doing concurrent reads
        for _ in 0..8 {
            let cache_clone = Arc::clone(&cache);
            handles.push(thread::spawn(move || {
                for i in 0..1000 {
                    let key = i % 100;
                    let _ = cache_clone.get(&key);
                }
            }));
        }

        for h in handles {
            h.join().expect("Thread panicked");
        }

        // Should have many L1 hits
        let stats = cache.stats();
        assert!(stats.l1_hits > 0);
    }

    #[test]
    fn test_lockfree_cache_scaling_8_threads() {
        // Measure throughput with 1 thread vs 8 threads
        let cache = Arc::new(LockFreeLruCache::<u64, String>::new(1000, 10000));

        // Pre-populate
        for i in 0..1000 {
            cache.insert(i, format!("value_{i}"));
        }

        let ops_per_thread = 5_000;

        // Single thread baseline
        let start = Instant::now();
        for i in 0..ops_per_thread {
            let _ = cache.get(&(i % 1000));
        }
        let single_thread_time = start.elapsed();

        // 8 threads
        let start = Instant::now();
        let mut handles = vec![];
        for t in 0..8 {
            let cache_clone = Arc::clone(&cache);
            handles.push(thread::spawn(move || {
                // Each thread accesses different key range to reduce L2 promotion contention
                let offset = t * 100;
                for i in 0..ops_per_thread {
                    let key = ((i + offset) % 1000) as u64;
                    let _ = cache_clone.get(&key);
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        let eight_thread_time = start.elapsed();

        // Calculate throughput
        let single_throughput = ops_per_thread as f64 / single_thread_time.as_secs_f64();
        let eight_throughput = (8 * ops_per_thread) as f64 / eight_thread_time.as_secs_f64();
        let scaling_factor = eight_throughput / single_throughput;

        println!("LockFreeLruCache scaling test:");
        println!("  1 thread:  {:.0} ops/sec", single_throughput);
        println!("  8 threads: {:.0} ops/sec", eight_throughput);
        println!("  Scaling:   {:.2}x", scaling_factor);

        // DashMap L1 should not cause severe regression
        // Note: L2 promotion uses locks, scaling depends on L1 hit rate and system load
        // On CI/loaded systems, contention may reduce throughput slightly
        assert!(
            scaling_factor > 0.5,
            "Scaling factor {scaling_factor:.2}x should be > 0.5x (no severe regression)"
        );
    }

    #[test]
    fn test_lockfree_cache_hit_rate() {
        let cache: LockFreeLruCache<u64, String> = LockFreeLruCache::new(100, 1000);

        // Insert some values
        for i in 0..50 {
            cache.insert(i, format!("value_{i}"));
        }

        // Access existing keys (hits)
        for i in 0..50 {
            let _ = cache.get(&i);
        }

        // Access non-existing keys (misses)
        for i in 100..110 {
            let _ = cache.get(&i);
        }

        let stats = cache.stats();
        assert_eq!(stats.l1_hits, 50);
        assert_eq!(stats.misses, 10);

        // Total hit rate should be 50/60 ≈ 0.833
        let hit_rate = stats.total_hit_rate();
        assert!((hit_rate - 0.833).abs() < 0.01);
    }
}