featherdb_mvcc/

gc.rs

//! Garbage collection metrics and adaptive scheduling for MVCC

use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

/// Configuration for garbage collection
#[derive(Debug, Clone)]
pub struct GcConfig {
    /// Minimum interval between GC runs (milliseconds)
    pub min_interval_ms: u64,
    /// Maximum versions before triggering GC
    pub max_versions_before_gc: u64,
    /// Enable adaptive GC scheduling
    pub adaptive: bool,
}

impl Default for GcConfig {
    fn default() -> Self {
        Self {
            min_interval_ms: 1000, // 1 second minimum
            max_versions_before_gc: 10000,
            adaptive: true,
        }
    }
}

/// GC metrics for monitoring and adaptive scheduling
#[derive(Debug, Default)]
pub struct GcMetrics {
    /// Total versions created
    pub versions_created: AtomicU64,
    /// Total versions cleaned
    pub versions_cleaned: AtomicU64,
    /// Number of GC runs
    pub gc_runs: AtomicU64,
    /// Total time spent in GC (microseconds)
    pub gc_time_us: AtomicU64,
    /// Versions pending cleanup
    pub versions_pending: AtomicU64,
}

impl GcMetrics {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn record_version_created(&self) {
        self.versions_created.fetch_add(1, Ordering::Relaxed);
        self.versions_pending.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_versions_cleaned(&self, count: u64) {
        self.versions_cleaned.fetch_add(count, Ordering::Relaxed);
        self.versions_pending.fetch_sub(
            count.min(self.versions_pending.load(Ordering::Relaxed)),
            Ordering::Relaxed,
        );
    }

    pub fn record_gc_run(&self, duration: Duration) {
        self.gc_runs.fetch_add(1, Ordering::Relaxed);
        self.gc_time_us
            .fetch_add(duration.as_micros() as u64, Ordering::Relaxed);
    }

    /// Get a snapshot of metrics
    pub fn snapshot(&self) -> GcMetricsSnapshot {
        GcMetricsSnapshot {
            versions_created: self.versions_created.load(Ordering::Relaxed),
            versions_cleaned: self.versions_cleaned.load(Ordering::Relaxed),
            gc_runs: self.gc_runs.load(Ordering::Relaxed),
            gc_time_us: self.gc_time_us.load(Ordering::Relaxed),
            versions_pending: self.versions_pending.load(Ordering::Relaxed),
        }
    }
}

#[derive(Debug, Clone)]
pub struct GcMetricsSnapshot {
    pub versions_created: u64,
    pub versions_cleaned: u64,
    pub gc_runs: u64,
    pub gc_time_us: u64,
    pub versions_pending: u64,
}

/// Adaptive GC scheduler
pub struct GcScheduler {
    config: GcConfig,
    last_gc: Instant,
    metrics: GcMetrics,
}

impl GcScheduler {
    pub fn new(config: GcConfig) -> Self {
        Self {
            config,
            last_gc: Instant::now(),
            metrics: GcMetrics::new(),
        }
    }

    /// Check if GC should run based on adaptive heuristics
    pub fn should_run_gc(&self) -> bool {
        let elapsed = self.last_gc.elapsed();
        let min_interval = Duration::from_millis(self.config.min_interval_ms);

        // Always respect minimum interval
        if elapsed < min_interval {
            return false;
        }

        // Check if too many versions pending
        let pending = self.metrics.versions_pending.load(Ordering::Relaxed);
        if pending >= self.config.max_versions_before_gc {
            return true;
        }

        // Adaptive: run more frequently if version creation rate is high
        if self.config.adaptive {
            let snapshot = self.metrics.snapshot();
            if snapshot.gc_runs > 0 {
                let avg_cleaned_per_run = snapshot.versions_cleaned / snapshot.gc_runs;
                // If pending is > 2x average, run GC
                if pending > avg_cleaned_per_run * 2 {
                    return true;
                }
            }
        }

        // Default: run if minimum interval passed
        elapsed >= min_interval
    }

    /// Record that GC has run
    pub fn record_gc_run(&mut self, duration: Duration, versions_cleaned: u64) {
        self.last_gc = Instant::now();
        self.metrics.record_gc_run(duration);
        self.metrics.record_versions_cleaned(versions_cleaned);
    }

    /// Record a version was created
    pub fn record_version_created(&self) {
        self.metrics.record_version_created();
    }

    /// Get current metrics
    pub fn metrics(&self) -> GcMetricsSnapshot {
        self.metrics.snapshot()
    }
}

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

    #[test]
    fn test_gc_config_default() {
        let config = GcConfig::default();
        assert_eq!(config.min_interval_ms, 1000);
        assert_eq!(config.max_versions_before_gc, 10000);
        assert!(config.adaptive);
    }

    #[test]
    fn test_gc_metrics_creation() {
        let metrics = GcMetrics::new();
        metrics.record_version_created();
        metrics.record_version_created();

        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.versions_created, 2);
        assert_eq!(snapshot.versions_pending, 2);
        assert_eq!(snapshot.versions_cleaned, 0);
    }

    #[test]
    fn test_gc_metrics_cleanup() {
        let metrics = GcMetrics::new();

        // Create 10 versions
        for _ in 0..10 {
            metrics.record_version_created();
        }

        // Clean 5 versions
        metrics.record_versions_cleaned(5);

        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.versions_created, 10);
        assert_eq!(snapshot.versions_cleaned, 5);
        assert_eq!(snapshot.versions_pending, 5);
    }

    #[test]
    fn test_gc_metrics_gc_run() {
        let metrics = GcMetrics::new();

        let duration = Duration::from_millis(100);
        metrics.record_gc_run(duration);

        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.gc_runs, 1);
        assert!(snapshot.gc_time_us >= 100_000); // At least 100ms in microseconds
    }

    #[test]
    fn test_gc_scheduler_min_interval() {
        let config = GcConfig {
            min_interval_ms: 100,
            max_versions_before_gc: 10000,
            adaptive: false,
        };

        let scheduler = GcScheduler::new(config);

        // Should not run immediately
        assert!(!scheduler.should_run_gc());

        // Wait for interval
        thread::sleep(Duration::from_millis(150));

        // Should run now
        assert!(scheduler.should_run_gc());
    }

    #[test]
    fn test_gc_scheduler_max_versions() {
        let config = GcConfig {
            min_interval_ms: 10, // Low interval so we can test max_versions trigger
            max_versions_before_gc: 5,
            adaptive: false,
        };

        let scheduler = GcScheduler::new(config);

        // Create versions
        for _ in 0..6 {
            scheduler.record_version_created();
        }

        // Should trigger GC despite short elapsed time (but still respects min interval)
        thread::sleep(Duration::from_millis(1));
        // Won't trigger because min_interval not elapsed
        assert!(!scheduler.should_run_gc());

        // Wait for min interval
        thread::sleep(Duration::from_millis(15));
        // Now should trigger because versions exceeded max_versions_before_gc
        assert!(scheduler.should_run_gc());
    }

    #[test]
    fn test_gc_scheduler_adaptive() {
        let config = GcConfig {
            min_interval_ms: 10,
            max_versions_before_gc: 100,
            adaptive: true,
        };

        let mut scheduler = GcScheduler::new(config);

        // Simulate first GC run that cleaned 10 versions
        thread::sleep(Duration::from_millis(20));
        scheduler.record_gc_run(Duration::from_millis(5), 10);

        // Create 25 versions (> 2x average of 10)
        for _ in 0..25 {
            scheduler.record_version_created();
        }

        // Wait for min interval
        thread::sleep(Duration::from_millis(20));

        // Adaptive mode should trigger because pending > 2x average
        assert!(scheduler.should_run_gc());
    }

    #[test]
    fn test_gc_scheduler_record_operations() {
        let config = GcConfig::default();
        let mut scheduler = GcScheduler::new(config);

        // Record versions created
        scheduler.record_version_created();
        scheduler.record_version_created();

        let metrics = scheduler.metrics();
        assert_eq!(metrics.versions_created, 2);
        assert_eq!(metrics.versions_pending, 2);

        // Record GC run
        scheduler.record_gc_run(Duration::from_millis(50), 1);

        let metrics = scheduler.metrics();
        assert_eq!(metrics.gc_runs, 1);
        assert_eq!(metrics.versions_cleaned, 1);
        assert_eq!(metrics.versions_pending, 1);
    }

    #[test]
    fn test_gc_metrics_pending_doesnt_underflow() {
        let metrics = GcMetrics::new();

        // Clean more than created (shouldn't underflow)
        metrics.record_versions_cleaned(100);

        let snapshot = metrics.snapshot();
        assert_eq!(snapshot.versions_pending, 0);
    }
}