featherdb_storage/

eviction.rs

//! Eviction policies for the buffer pool
//!
//! This module provides different eviction policies:
//! - Clock: Simple clock-sweep algorithm (original)
//! - LRU-2: Tracks last 2 access times, evicts by oldest second-to-last access
//! - LIRS: Low Inter-reference Recency Set - better scan resistance

use featherdb_core::PageId;
use std::collections::{HashMap, VecDeque};
use std::time::Instant;

/// Eviction policy trait
///
/// Implementations of this trait decide which page to evict when the buffer pool is full.
pub trait EvictionPolicy: Send + Sync {
    /// Called when a page is accessed (cache hit)
    fn on_access(&mut self, page_id: PageId);

    /// Called when a new page is loaded into the buffer pool
    fn on_load(&mut self, page_id: PageId);

    /// Select a victim page for eviction
    /// Returns None if no page can be evicted
    fn select_victim(&mut self) -> Option<PageId>;

    /// Remove a page from tracking (called when page is evicted or freed)
    fn remove(&mut self, page_id: PageId);

    /// Get the name of this eviction policy
    fn name(&self) -> &'static str;
}

// =============================================================================
// Clock Eviction Policy
// =============================================================================

/// Clock eviction policy (second-chance algorithm)
///
/// Simple and efficient. Each page has a reference bit that is set on access.
/// The clock hand sweeps through pages, evicting the first page with reference bit = 0,
/// and clearing reference bits as it goes.
pub struct ClockEviction {
    /// Ordered list of page IDs in clock order
    pages: VecDeque<PageId>,
    /// Reference bits for each page
    reference_bits: HashMap<PageId, bool>,
    /// Current clock hand position
    hand: usize,
}

impl ClockEviction {
    pub fn new() -> Self {
        ClockEviction {
            pages: VecDeque::new(),
            reference_bits: HashMap::new(),
            hand: 0,
        }
    }
}

impl Default for ClockEviction {
    fn default() -> Self {
        Self::new()
    }
}

impl EvictionPolicy for ClockEviction {
    fn on_access(&mut self, page_id: PageId) {
        self.reference_bits.insert(page_id, true);
    }

    fn on_load(&mut self, page_id: PageId) {
        self.pages.push_back(page_id);
        self.reference_bits.insert(page_id, true);
    }

    fn select_victim(&mut self) -> Option<PageId> {
        let len = self.pages.len();
        if len == 0 {
            return None;
        }

        // Try up to 2 full sweeps
        for _ in 0..len * 2 {
            self.hand %= len;
            let page_id = self.pages[self.hand];

            if let Some(ref_bit) = self.reference_bits.get_mut(&page_id) {
                if *ref_bit {
                    // Give second chance
                    *ref_bit = false;
                    self.hand = (self.hand + 1) % len;
                } else {
                    // Found victim
                    return Some(page_id);
                }
            } else {
                // Page not tracked, can be evicted
                return Some(page_id);
            }
        }

        // All pages have reference bit set, just pick current one
        Some(self.pages[self.hand])
    }

    fn remove(&mut self, page_id: PageId) {
        if let Some(pos) = self.pages.iter().position(|&id| id == page_id) {
            self.pages.remove(pos);
            // Adjust hand if necessary
            if self.hand >= self.pages.len() && !self.pages.is_empty() {
                self.hand = 0;
            }
        }
        self.reference_bits.remove(&page_id);
    }

    fn name(&self) -> &'static str {
        "Clock"
    }
}

// =============================================================================
// LRU-2 Eviction Policy
// =============================================================================

/// Access history for a page in LRU-K algorithm
#[derive(Debug, Clone)]
struct Lru2Entry {
    /// Most recent access time
    last_access: Instant,
    /// Second-to-last access time (K=2 distance)
    prev_access: Option<Instant>,
    /// Number of accesses
    access_count: u32,
    /// Timestamp when this entry was created (used for cold page ordering)
    created_at: Instant,
}

impl Lru2Entry {
    fn new() -> Self {
        let now = Instant::now();
        Lru2Entry {
            last_access: now,
            prev_access: None,
            access_count: 1,
            created_at: now,
        }
    }

    fn record_access(&mut self) {
        self.prev_access = Some(self.last_access);
        self.last_access = Instant::now();
        self.access_count = self.access_count.saturating_add(1);
    }

    /// Check if this page is "hot" (has been accessed more than once)
    fn is_hot(&self) -> bool {
        self.prev_access.is_some()
    }

    /// Get the K-distance for this page (for sorting hot pages)
    /// For LRU-2, this is the time of the second-to-last access
    fn k_distance(&self) -> Option<Instant> {
        self.prev_access
    }
}

/// LRU-2 eviction policy
///
/// Tracks the last 2 access times for each page. Pages are evicted based on
/// the oldest second-to-last access time. This provides better resistance to
/// scan pollution compared to plain LRU.
///
/// Key insight: A page that was accessed only once recently (like during a scan)
/// will be evicted before a page that was accessed twice (indicating actual use).
pub struct Lru2Eviction {
    /// Access history for each page
    entries: HashMap<PageId, Lru2Entry>,
    /// Correlated reference period (CRP) - pages within this window are not evicted
    /// This prevents recently loaded pages from being immediately evicted
    correlated_reference_period_ms: u64,
}

impl Lru2Eviction {
    pub fn new() -> Self {
        Lru2Eviction {
            entries: HashMap::new(),
            // 10ms default CRP - pages accessed within 10ms are considered correlated
            correlated_reference_period_ms: 10,
        }
    }

    /// Create with custom correlated reference period
    pub fn with_crp(crp_ms: u64) -> Self {
        Lru2Eviction {
            entries: HashMap::new(),
            correlated_reference_period_ms: crp_ms,
        }
    }
}

impl Default for Lru2Eviction {
    fn default() -> Self {
        Self::new()
    }
}

impl EvictionPolicy for Lru2Eviction {
    fn on_access(&mut self, page_id: PageId) {
        let now = Instant::now();

        if let Some(entry) = self.entries.get_mut(&page_id) {
            // Only record access if outside correlated reference period
            let elapsed_ms = now.duration_since(entry.last_access).as_millis() as u64;
            if elapsed_ms > self.correlated_reference_period_ms {
                entry.record_access();
            }
        } else {
            // New page
            self.entries.insert(page_id, Lru2Entry::new());
        }
    }

    fn on_load(&mut self, page_id: PageId) {
        self.entries.insert(page_id, Lru2Entry::new());
    }

    fn select_victim(&mut self) -> Option<PageId> {
        if self.entries.is_empty() {
            return None;
        }

        // LRU-2 algorithm provides scan resistance by treating cold pages (1 access)
        // differently from hot pages (2+ accesses).
        //
        // Priority 1: Evict cold pages first (pages accessed only once)
        //             Sort by creation time (oldest first)
        // Priority 2: If no cold pages, evict hot page with oldest K-distance
        //             (oldest second-to-last access time)

        // Separate cold and hot pages
        let mut cold_pages: Vec<_> = self
            .entries
            .iter()
            .filter(|(_, entry)| !entry.is_hot())
            .map(|(&id, entry)| (id, entry.created_at))
            .collect();

        let mut hot_pages: Vec<_> = self
            .entries
            .iter()
            .filter(|(_, entry)| entry.is_hot())
            .map(|(&id, entry)| (id, entry.k_distance().unwrap()))
            .collect();

        // Prefer evicting cold pages (scan pollution victims)
        if !cold_pages.is_empty() {
            // Sort by creation time (oldest first)
            cold_pages.sort_by_key(|(_, created_at)| *created_at);
            return Some(cold_pages[0].0);
        }

        // No cold pages, evict the hot page with oldest K-distance
        if !hot_pages.is_empty() {
            // Sort by K-distance (oldest prev_access first)
            hot_pages.sort_by_key(|(_, k_distance)| *k_distance);
            return Some(hot_pages[0].0);
        }

        None
    }

    fn remove(&mut self, page_id: PageId) {
        self.entries.remove(&page_id);
    }

    fn name(&self) -> &'static str {
        "LRU-2"
    }
}

// =============================================================================
// LIRS Eviction Policy
// =============================================================================

/// Block status in LIRS algorithm
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LirsStatus {
    /// Low Inter-reference Recency (hot)
    Lir,
    /// High Inter-reference Recency, resident in cache
    HirResident,
    /// High Inter-reference Recency, not resident (only metadata)
    HirNonResident,
}

/// Entry in the LIRS stack
#[derive(Debug, Clone)]
struct LirsEntry {
    status: LirsStatus,
    /// Recency (position in S stack, lower = more recent)
    recency: u64,
    /// True if in the S stack (LIRS stack)
    in_stack_s: bool,
    /// True if in the Q list (resident HIR blocks)
    in_list_q: bool,
}

/// LIRS (Low Inter-reference Recency Set) eviction policy
///
/// LIRS separates pages into two sets:
/// - LIR set: Pages with low inter-reference recency (hot, frequently accessed)
/// - HIR set: Pages with high inter-reference recency (cold, infrequently accessed)
///
/// Only HIR resident pages can be evicted. When a HIR page is accessed twice
/// within the LIR set's recency window, it becomes LIR.
///
/// This provides excellent scan resistance and adapts to workload changes.
pub struct LirsEviction {
    /// All tracked pages
    entries: HashMap<PageId, LirsEntry>,
    /// Stack S - LIRS stack ordered by recency (most recent at end)
    stack_s: VecDeque<PageId>,
    /// List Q - Resident HIR pages in FIFO order
    list_q: VecDeque<PageId>,
    /// Maximum size of LIR set (typically ~99% of cache size)
    lir_size_limit: usize,
    /// Current size of LIR set
    lir_count: usize,
    /// Current size of HIR resident set
    hir_count: usize,
    /// Global recency counter
    recency_counter: u64,
}

impl LirsEviction {
    pub fn new(capacity: usize) -> Self {
        // LIR set is ~99% of cache, HIR is ~1%
        let lir_limit = (capacity as f64 * 0.99) as usize;

        LirsEviction {
            entries: HashMap::new(),
            stack_s: VecDeque::new(),
            list_q: VecDeque::new(),
            lir_size_limit: lir_limit.max(1),
            lir_count: 0,
            hir_count: 0,
            recency_counter: 0,
        }
    }

    /// Set the LIR size ratio (0.0 to 1.0)
    pub fn with_lir_ratio(mut self, ratio: f64) -> Self {
        let total = self.lir_size_limit + 1; // Approximate capacity
        self.lir_size_limit = ((total as f64) * ratio) as usize;
        self
    }

    /// Move a page to the top of stack S (most recent)
    fn move_to_stack_top(&mut self, page_id: PageId) {
        // Remove from current position if present
        self.stack_s.retain(|&id| id != page_id);
        // Add to end (most recent)
        self.stack_s.push_back(page_id);

        self.recency_counter += 1;
        if let Some(entry) = self.entries.get_mut(&page_id) {
            entry.recency = self.recency_counter;
            entry.in_stack_s = true;
        }
    }

    /// Prune the bottom of stack S (remove non-LIR pages from bottom)
    fn prune_stack(&mut self) {
        while let Some(&page_id) = self.stack_s.front() {
            // First, check the status
            let status = self.entries.get(&page_id).map(|e| e.status);

            match status {
                Some(LirsStatus::Lir) => {
                    // Stop at first LIR page
                    break;
                }
                Some(status) => {
                    // Remove non-LIR page from bottom of stack
                    self.stack_s.pop_front();

                    // Update in_stack_s flag
                    if let Some(e) = self.entries.get_mut(&page_id) {
                        e.in_stack_s = false;
                    }

                    // If it's non-resident HIR, we can remove the entry entirely
                    if status == LirsStatus::HirNonResident {
                        self.entries.remove(&page_id);
                    }
                }
                None => {
                    // Entry doesn't exist, remove from stack
                    self.stack_s.pop_front();
                }
            }
        }
    }

    /// Convert the bottom LIR block to HIR (when promoting a HIR to LIR)
    fn demote_bottom_lir(&mut self) {
        // Find the bottom LIR block in stack S
        let mut bottom_lir = None;
        for &page_id in &self.stack_s {
            if let Some(entry) = self.entries.get(&page_id) {
                if entry.status == LirsStatus::Lir {
                    bottom_lir = Some(page_id);
                    break;
                }
            }
        }

        if let Some(page_id) = bottom_lir {
            if let Some(entry) = self.entries.get_mut(&page_id) {
                entry.status = LirsStatus::HirResident;
                entry.in_list_q = true;
                self.lir_count = self.lir_count.saturating_sub(1);
                self.hir_count += 1;
                self.list_q.push_back(page_id);
            }

            // Prune stack after demotion
            self.prune_stack();
        }
    }
}

impl Default for LirsEviction {
    fn default() -> Self {
        Self::new(1000) // Default capacity, should be set properly
    }
}

impl EvictionPolicy for LirsEviction {
    fn on_access(&mut self, page_id: PageId) {
        if let Some(entry) = self.entries.get(&page_id).cloned() {
            match entry.status {
                LirsStatus::Lir => {
                    // LIR hit: move to top of stack S
                    self.move_to_stack_top(page_id);
                    self.prune_stack();
                }
                LirsStatus::HirResident => {
                    if entry.in_stack_s {
                        // HIR hit and in stack S: promote to LIR
                        if let Some(e) = self.entries.get_mut(&page_id) {
                            e.status = LirsStatus::Lir;
                            e.in_list_q = false;
                            self.list_q.retain(|&id| id != page_id);
                            self.lir_count += 1;
                            self.hir_count = self.hir_count.saturating_sub(1);
                        }

                        // Move to stack top
                        self.move_to_stack_top(page_id);

                        // Demote bottom LIR if over limit
                        if self.lir_count > self.lir_size_limit {
                            self.demote_bottom_lir();
                        }
                    } else {
                        // HIR hit but not in stack S: move to end of Q and add to S
                        self.list_q.retain(|&id| id != page_id);
                        self.list_q.push_back(page_id);
                        self.move_to_stack_top(page_id);
                    }
                }
                LirsStatus::HirNonResident => {
                    // This shouldn't happen for on_access (page should be loaded first)
                    // But handle it by promoting to HIR resident
                    if let Some(e) = self.entries.get_mut(&page_id) {
                        e.status = LirsStatus::HirResident;
                        e.in_list_q = true;
                        self.list_q.push_back(page_id);
                        self.hir_count += 1;
                    }
                    self.move_to_stack_top(page_id);
                }
            }
        }
    }

    fn on_load(&mut self, page_id: PageId) {
        // Check if this page was previously tracked (as non-resident HIR)
        let was_in_stack = self
            .entries
            .get(&page_id)
            .map(|e| e.in_stack_s)
            .unwrap_or(false);

        if self.lir_count < self.lir_size_limit {
            // Space in LIR set, add as LIR
            self.entries.insert(
                page_id,
                LirsEntry {
                    status: LirsStatus::Lir,
                    recency: self.recency_counter,
                    in_stack_s: true,
                    in_list_q: false,
                },
            );
            self.lir_count += 1;
            self.move_to_stack_top(page_id);
        } else if was_in_stack {
            // Was in stack as non-resident, promote to LIR
            if let Some(entry) = self.entries.get_mut(&page_id) {
                entry.status = LirsStatus::Lir;
                entry.in_list_q = false;
            }
            self.lir_count += 1;
            self.move_to_stack_top(page_id);

            // Demote bottom LIR to maintain limit
            if self.lir_count > self.lir_size_limit {
                self.demote_bottom_lir();
            }
        } else {
            // New page, add as HIR resident
            self.entries.insert(
                page_id,
                LirsEntry {
                    status: LirsStatus::HirResident,
                    recency: self.recency_counter,
                    in_stack_s: true,
                    in_list_q: true,
                },
            );
            self.hir_count += 1;
            self.list_q.push_back(page_id);
            self.move_to_stack_top(page_id);
        }
    }

    fn select_victim(&mut self) -> Option<PageId> {
        // Evict from front of Q (oldest HIR resident)
        while let Some(page_id) = self.list_q.pop_front() {
            if let Some(entry) = self.entries.get_mut(&page_id) {
                if entry.status == LirsStatus::HirResident {
                    entry.in_list_q = false;

                    // If in stack S, keep as non-resident HIR for future detection
                    if entry.in_stack_s {
                        entry.status = LirsStatus::HirNonResident;
                        self.hir_count = self.hir_count.saturating_sub(1);
                    }

                    return Some(page_id);
                }
            }
        }

        // Fallback: if Q is empty, evict bottom LIR (shouldn't normally happen)
        if let Some(&page_id) = self.stack_s.front() {
            return Some(page_id);
        }

        None
    }

    fn remove(&mut self, page_id: PageId) {
        if let Some(entry) = self.entries.remove(&page_id) {
            if entry.status == LirsStatus::Lir {
                self.lir_count = self.lir_count.saturating_sub(1);
            } else if entry.status == LirsStatus::HirResident {
                self.hir_count = self.hir_count.saturating_sub(1);
            }
        }
        self.stack_s.retain(|&id| id != page_id);
        self.list_q.retain(|&id| id != page_id);
    }

    fn name(&self) -> &'static str {
        "LIRS"
    }
}

// =============================================================================
// Eviction Policy Type (re-exported from core)
// =============================================================================

// Re-export the EvictionPolicyType from core
pub use featherdb_core::EvictionPolicyType;

/// Extension trait for creating eviction policies
pub trait EvictionPolicyFactory {
    /// Create an eviction policy of this type
    fn create(&self, capacity: usize) -> Box<dyn EvictionPolicy>;
}

impl EvictionPolicyFactory for EvictionPolicyType {
    /// Create an eviction policy of this type
    fn create(&self, capacity: usize) -> Box<dyn EvictionPolicy> {
        match self {
            EvictionPolicyType::Clock => Box::new(ClockEviction::new()),
            EvictionPolicyType::Lru2 => Box::new(Lru2Eviction::new()),
            EvictionPolicyType::Lirs => Box::new(LirsEviction::new(capacity)),
        }
    }
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_clock_basic() {
        let mut clock = ClockEviction::new();

        // Load some pages
        clock.on_load(PageId(1));
        clock.on_load(PageId(2));
        clock.on_load(PageId(3));

        // All have reference bit set, first sweep clears them
        // Second sweep should find victims
        let victim = clock.select_victim();
        assert!(victim.is_some());

        // Access page 2 to protect it
        clock.on_access(PageId(2));

        // After victim selection, page 2 should be protected
        let victim = clock.select_victim();
        assert!(victim.is_some());
        assert_ne!(victim, Some(PageId(2)));
    }

    #[test]
    fn test_clock_remove() {
        let mut clock = ClockEviction::new();

        clock.on_load(PageId(1));
        clock.on_load(PageId(2));

        clock.remove(PageId(1));

        // Should only find page 2
        let victim = clock.select_victim();
        assert_eq!(victim, Some(PageId(2)));
    }

    #[test]
    fn test_lru2_scan_resistance() {
        let mut lru2 = Lru2Eviction::new();

        // Load a "hot" page and access it multiple times
        lru2.on_load(PageId(1));
        std::thread::sleep(std::time::Duration::from_millis(20));
        lru2.on_access(PageId(1)); // Now has 2 accesses

        // Simulate scan: load many cold pages
        for i in 2..=10 {
            lru2.on_load(PageId(i as u64));
        }

        // Cold pages (single access) should be evicted before hot page
        let victim = lru2.select_victim();
        assert!(victim.is_some());
        assert_ne!(
            victim,
            Some(PageId(1)),
            "Hot page should not be evicted first"
        );
    }

    #[test]
    fn test_lru2_cold_page_eviction() {
        let mut lru2 = Lru2Eviction::new();

        // Load pages
        lru2.on_load(PageId(1));
        std::thread::sleep(std::time::Duration::from_millis(150)); // Past the 100ms threshold
        lru2.on_load(PageId(2));

        // Page 1 should be evicted first (older single access)
        let victim = lru2.select_victim();
        assert_eq!(victim, Some(PageId(1)));
    }

    #[test]
    fn test_lirs_basic() {
        let mut lirs = LirsEviction::new(10);

        // Load pages (they become LIR until limit reached)
        for i in 1..=5 {
            lirs.on_load(PageId(i as u64));
        }

        assert_eq!(lirs.lir_count, 5);

        // Access some pages
        lirs.on_access(PageId(1));
        lirs.on_access(PageId(2));
    }

    #[test]
    fn test_lirs_eviction() {
        let mut lirs = LirsEviction::new(5);

        // Fill up LIR set
        for i in 1..=10 {
            lirs.on_load(PageId(i as u64));
        }

        // Should have some HIR pages
        assert!(lirs.hir_count > 0);

        // Eviction should come from HIR set
        let victim = lirs.select_victim();
        assert!(victim.is_some());
    }

    #[test]
    fn test_policy_type_create() {
        let clock = EvictionPolicyType::Clock.create(100);
        assert_eq!(clock.name(), "Clock");

        let lru2 = EvictionPolicyType::Lru2.create(100);
        assert_eq!(lru2.name(), "LRU-2");

        let lirs = EvictionPolicyType::Lirs.create(100);
        assert_eq!(lirs.name(), "LIRS");
    }
}