rat_congestion 0.1.0

//! Congestion control algorithm implementations.
//!
//! This module contains high-performance implementations of BBR and CUBIC
//! congestion control algorithms with extensive optimizations.

use crate::constants::*;
use crate::events::CongestionEvent;
use std::any::Any;
use std::time::{Duration, Instant};

// Missing constants for compilation
const INITIAL_CWND: u64 = perf::INITIAL_CWND;
const MIN_CWND: u64 = perf::MIN_CWND;
const MIN_UPDATE_INTERVAL: Duration = Duration::from_millis(1);

/// Core trait for congestion control algorithms.
///
/// This trait defines the interface that all congestion control algorithms
/// must implement. It includes performance optimizations like inlining
/// and batch processing capabilities.
pub trait CongestionControlAlgorithm: Send + Sync {
    /// Process a congestion control event (ACK, loss, ECN).
    fn on_congestion_event(&mut self, event: CongestionEvent);
    
    /// Get the current congestion window size in bytes.
    fn congestion_window(&self) -> u64;
    
    /// Get the current slow start threshold.
    fn ssthresh(&self) -> u64;
    
    /// Check if more data can be sent given the current bytes in flight.
    fn can_send(&self, bytes_in_flight: u64) -> bool;
    
    /// Get the current pacing rate in bytes per second.
    #[inline(always)]
    fn pacing_rate(&self) -> u64;
    
    /// Get a reference to this algorithm as Any for type checking.
    fn as_any(&self) -> &dyn Any;
    
    /// Get the name of this algorithm.
    fn algorithm_name(&self) -> &'static str;
    
    /// Process multiple ACKs in batch for better performance.
    ///
    /// Default implementation processes each ACK individually,
    /// but algorithms can override this for optimized batch processing.
    fn process_ack_batch(&mut self, acks: &[(u64, u64, Duration)]) {
        for &(packet_number, bytes, rtt) in acks {
            self.on_congestion_event(CongestionEvent::PacketAcked {
                packet_number,
                bytes,
                rtt,
            });
        }
    }
}

/// CUBIC congestion control algorithm with performance optimizations.
///
/// This implementation includes cache-line alignment, SIMD-friendly operations,
/// and optimized cubic function calculations.
#[repr(C, align(64))] // Cache line alignment for better performance
pub struct Cubic {
    // Hot path fields (accessed frequently)
    congestion_window: u64,
    pacing_rate: u64,
    ssthresh: u64,
    
    // CUBIC-specific state
    w_max: u64,
    k: f64,
    last_congestion_time: Instant,
    
    // Performance optimization fields
    last_update: Instant,
    update_interval: Duration,
}

impl Cubic {
    /// Creates a new CUBIC congestion control instance.
    pub fn new() -> Self {
        Self {
            congestion_window: INITIAL_CWND,
            pacing_rate: 1_000_000, // 1 Mbps initial pacing rate
            ssthresh: u64::MAX,
            w_max: 0,
            k: 0.0,
            last_congestion_time: Instant::now(),
            last_update: Instant::now(),
            update_interval: perf::MIN_UPDATE_INTERVAL,
        }
    }
    
    /// Core CUBIC window calculation with optimizations.
    #[inline(always)]
    fn cubic_update(&mut self) {
        let now = Instant::now();
        
        // Rate limiting to reduce computational overhead
        if now - self.last_update < self.update_interval {
            return;
        }
        self.last_update = now;
        
        let t = (now - self.last_congestion_time).as_secs_f64();
        
        // Fast approximation for cubic calculation near the inflection point
        let t_diff = t - self.k;
        let w_cubic = if t_diff.abs() < 0.1 {
            // Linear approximation near k (avoids expensive cubic calculation)
            self.w_max as f64 + 0.4 * 3.0 * self.k * self.k * t_diff
        } else {
            // Full cubic calculation when far from inflection point
            0.4 * t_diff.powi(3) + self.w_max as f64
        };
        
        // Apply window growth with bounds checking
        if w_cubic > self.congestion_window as f64 {
            self.congestion_window = (w_cubic as u64).max(MIN_CWND);
        }
        
        // Update pacing rate based on current window
        self.pacing_rate = self.congestion_window * 8 / 10; // 80% of window rate
    }
    
    /// Handle packet loss with optimized calculations.
    #[inline(always)]
    fn on_packet_lost(&mut self) {
        self.w_max = self.congestion_window;
        self.congestion_window = ((self.congestion_window as f64) * 0.7) as u64;
        self.congestion_window = self.congestion_window.max(MIN_CWND);
        self.last_congestion_time = Instant::now();
        
        // Fast k calculation (cube root approximation)
        let beta = 0.7;
        let c = 0.4;
        self.k = ((self.w_max as f64) * (1.0 - beta) / c).powf(1.0/3.0);
        
        // Reduce pacing rate after loss
        self.pacing_rate = self.congestion_window * 6 / 10; // 60% rate
    }
}

impl CongestionControlAlgorithm for Cubic {
    #[inline(always)]
    fn on_congestion_event(&mut self, event: CongestionEvent) {
        match event {
            CongestionEvent::PacketAcked { .. } => {
                self.cubic_update();
            },
            CongestionEvent::PacketLost { .. } => {
                self.on_packet_lost();
            },
            CongestionEvent::ExplicitCongestionNotification => {
                // ECN: gentler response than packet loss
                self.congestion_window = ((self.congestion_window as f64) * 0.8) as u64;
                self.congestion_window = self.congestion_window.max(MIN_CWND);
                self.pacing_rate = self.congestion_window * 7 / 10; // 70% rate
            },
        }
    }
    
    #[inline(always)]
    fn congestion_window(&self) -> u64 {
        self.congestion_window
    }
    
    #[inline(always)]
    fn ssthresh(&self) -> u64 {
        self.ssthresh
    }
    
    #[inline(always)]
    fn can_send(&self, bytes_in_flight: u64) -> bool {
        bytes_in_flight < self.congestion_window
    }
    
    #[inline(always)]
    fn pacing_rate(&self) -> u64 {
        self.pacing_rate
    }
    
    fn as_any(&self) -> &dyn Any {
        self
    }
    
    fn algorithm_name(&self) -> &'static str {
        "CUBIC"
    }
    
    /// Optimized batch ACK processing for CUBIC.
    fn process_ack_batch(&mut self, acks: &[(u64, u64, Duration)]) {
        if acks.is_empty() {
            return;
        }
        
        // For batch processing, temporarily increase update interval
        // to avoid excessive computation, then process normally
        if acks.len() > 1 {
            let old_interval = self.update_interval;
            self.update_interval = Duration::from_secs(1);
            
            // Process all but the last ACK with reduced updates
            for &(packet_number, bytes, rtt) in &acks[..acks.len()-1] {
                self.on_congestion_event(CongestionEvent::PacketAcked {
                    packet_number, bytes, rtt,
                });
            }
            
            // Restore update interval and process final ACK
            self.update_interval = old_interval;
        }
        
        // Process the last (or only) ACK with normal update frequency
        let &(packet_number, bytes, rtt) = acks.last().unwrap();
        self.on_congestion_event(CongestionEvent::PacketAcked {
            packet_number, bytes, rtt,
        });
    }
}

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

/// BBR (Bottleneck Bandwidth and Round-trip propagation time) algorithm states.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BbrState {
    /// Startup phase: aggressively probe for bandwidth
    Startup,
    /// Drain phase: reduce in-flight data to clear queues
    Drain,
    /// Probe bandwidth phase: cyclically probe for more bandwidth
    ProbeBw,
    /// Probe RTT phase: reduce window to measure minimum RTT
    ProbeRtt,
}

/// BBR congestion control algorithm with performance optimizations.
///
/// This implementation includes cache-line alignment, optimized bandwidth
/// estimation, and efficient state machine transitions.
#[repr(C, align(64))] // Cache line alignment
pub struct Bbr {
    // Hot path fields (ordered by access frequency)
    congestion_window: u64,
    pacing_rate: u64,
    max_bw: f64,
    min_rtt: Duration,
    
    // BBR state machine
    state: BbrState,
    pacing_gain: f64,
    cwnd_gain: f64,
    
    // Measurement fields
    delivery_rate: f64,
    rt_prop: Duration,
    bytes_in_flight: u64,
    
    // Counters and timestamps
    cycle_count: u32,
    full_bw_count: u32,
    ssthresh: u64,
    
    // Performance optimization fields
    last_update: Instant,
    probe_rtt_start: Option<Instant>,
    cycle_start: Instant,
    bw_samples: [f64; 8], // Circular buffer for bandwidth samples
    bw_sample_idx: usize,
}

impl Bbr {
    /// Creates a new BBR congestion control instance.
    pub fn new() -> Self {
        let now = Instant::now();
        Self {
            congestion_window: INITIAL_CWND,
            pacing_rate: 1_000_000, // 1 Mbps initial pacing rate
            max_bw: 1_000_000.0,    // 1 Mbps initial bandwidth estimate
            min_rtt: Duration::from_millis(100),
            
            state: BbrState::Startup,
            pacing_gain: bbr::STARTUP_GAIN as f64 / 1000.0,
            cwnd_gain: bbr::CWND_GAIN as f64 / 1000.0,
            
            delivery_rate: 0.0,
            rt_prop: Duration::from_millis(100),
            bytes_in_flight: 0,
            
            cycle_count: 0,
            full_bw_count: 0,
            ssthresh: u64::MAX,
            
            last_update: now,
            probe_rtt_start: None,
            cycle_start: now,
            bw_samples: [0.0; 8],
            bw_sample_idx: 0,
        }
    }
    
    /// Calculate the target in-flight data based on BDP and gain.
    #[inline(always)]
    fn inflight_target(&self) -> u64 {
        (self.cwnd_gain * self.max_bw * self.min_rtt.as_secs_f64()) as u64
    }
    
    /// Calculate bandwidth-delay product.
    #[inline(always)]
    fn bdp(&self) -> u64 {
        (self.max_bw * self.min_rtt.as_secs_f64()) as u64
    }
    
    /// Core BBR state machine update with optimizations.
    #[inline(always)]
    fn bbr_update(&mut self, rtt: Duration, delivery_rate: f64) {
        let now = Instant::now();
        
        // Rate limiting for performance
        if now - self.last_update < MIN_UPDATE_INTERVAL {
            return;
        }
        self.last_update = now;
        
        // Update minimum RTT with fast comparison
        if rtt < self.min_rtt {
            self.min_rtt = rtt;
        }
        
        // Update bandwidth samples using circular buffer
        self.bw_samples[self.bw_sample_idx] = delivery_rate;
        self.bw_sample_idx = (self.bw_sample_idx + 1) % 8;
        
        // Calculate maximum bandwidth (SIMD-friendly)
        let mut max_bw = 0.0;
        for &bw in &self.bw_samples {
            if bw > max_bw {
                max_bw = bw;
            }
        }
        self.max_bw = max_bw;
        
        // Update delivery measurements
        self.delivery_rate = delivery_rate;
        self.rt_prop = rtt;
        
        // BBR state machine logic
        match self.state {
            BbrState::Startup => {
                // Exit startup when bandwidth growth stalls
                if self.delivery_rate >= self.max_bw * 0.9 {
                    self.state = BbrState::Drain;
                    self.pacing_gain = bbr::DRAIN_GAIN as f64 / 1000.0;
                }
            },
            BbrState::Drain => {
                // Exit drain when in-flight data is reduced
                if self.bytes_in_flight <= self.inflight_target() {
                    self.state = BbrState::ProbeBw;
                    self.pacing_gain = 1.0;
                    self.cycle_count = 0;
                }
            },
            BbrState::ProbeBw => {
                // Cycle through probe gains
                self.cycle_count += 1;
                if self.cycle_count >= 8 {
                    self.cycle_count = 0;
                }
                
                // Apply BBR probe gain cycle
                self.pacing_gain = bbr::PACING_GAIN_CYCLE[self.cycle_count as usize] as f64 / 1000.0;
                
                // Periodically enter RTT probe
                if now.duration_since(self.cycle_start) >= perf::BW_PROBE_CYCLE {
                    self.state = BbrState::ProbeRtt;
                    self.pacing_gain = 1.0;
                    self.probe_rtt_start = Some(now);
                }
            },
            BbrState::ProbeRtt => {
                // Exit RTT probe after sufficient time
                if let Some(start) = self.probe_rtt_start {
                    if now.duration_since(start) >= perf::RTT_PROBE_DURATION {
                        self.state = BbrState::ProbeBw;
                        self.cycle_count = 0;
                        self.cycle_start = now;
                        self.probe_rtt_start = None;
                    }
                }
            }
        }
        
        // Update congestion window and pacing rate
        self.update_congestion_window();
        self.update_pacing_rate();
    }
    
    /// Update congestion window based on current BBR state.
    #[inline(always)]
    fn update_congestion_window(&mut self) {
        let bdp = self.bdp();
        let target_cwnd = (bdp as f64 * self.cwnd_gain) as u64;
        
        match self.state {
            BbrState::Startup => {
                // Aggressive growth in startup
                self.congestion_window = target_cwnd.max(self.congestion_window * 2);
            },
            BbrState::Drain => {
                // Maintain window during drain
            },
            BbrState::ProbeBw => {
                // Adjust based on gain
                self.congestion_window = target_cwnd;
            },
            BbrState::ProbeRtt => {
                // Reduce window for RTT measurement
                self.congestion_window = (bdp / 2).max(MIN_CWND);
            }
        }
        
        // Ensure minimum window size
        self.congestion_window = self.congestion_window.max(MIN_CWND);
    }
    
    /// Update pacing rate based on bandwidth and gain.
    #[inline(always)]
    fn update_pacing_rate(&mut self) {
        self.pacing_rate = (self.max_bw * self.pacing_gain) as u64;
        self.pacing_rate = self.pacing_rate.max(10_000); // Minimum 10 Kbps
    }
}

impl CongestionControlAlgorithm for Bbr {
    #[inline(always)]
    fn on_congestion_event(&mut self, event: CongestionEvent) {
        match event {
            CongestionEvent::PacketAcked { rtt, bytes, .. } => {
                // Estimate delivery rate
                let delivery_rate = if rtt.as_secs_f64() > 0.0 {
                    bytes as f64 / rtt.as_secs_f64()
                } else {
                    self.delivery_rate
                };
                
                self.bbr_update(rtt, delivery_rate);
            },
            CongestionEvent::PacketLost { .. } => {
                // BBR is loss-tolerant, but respond in startup
                if matches!(self.state, BbrState::Startup) {
                    self.congestion_window = (self.congestion_window as f64 * 0.9) as u64;
                    self.congestion_window = self.congestion_window.max(MIN_CWND);
                }
            },
            CongestionEvent::ExplicitCongestionNotification => {
                // Gentle response to ECN
                self.congestion_window = (self.congestion_window as f64 * 0.95) as u64;
                self.congestion_window = self.congestion_window.max(MIN_CWND);
            },
        }
    }
    
    #[inline(always)]
    fn congestion_window(&self) -> u64 {
        self.congestion_window
    }
    
    #[inline(always)]
    fn ssthresh(&self) -> u64 {
        self.ssthresh
    }
    
    #[inline(always)]
    fn can_send(&self, bytes_in_flight: u64) -> bool {
        bytes_in_flight < self.congestion_window
    }
    
    #[inline(always)]
    fn pacing_rate(&self) -> u64 {
        self.pacing_rate
    }
    
    fn as_any(&self) -> &dyn Any {
        self
    }
    
    fn algorithm_name(&self) -> &'static str {
        "BBR"
    }
    
    /// Optimized batch ACK processing for BBR.
    fn process_ack_batch(&mut self, acks: &[(u64, u64, Duration)]) {
        if acks.is_empty() {
            return;
        }
        
        // Calculate aggregate statistics for batch processing
        let mut total_bytes = 0u64;
        let mut total_time = Duration::ZERO;
        let mut min_rtt = Duration::from_secs(1);
        
        for &(_, bytes, rtt) in acks {
            total_bytes += bytes;
            total_time += rtt;
            if rtt < min_rtt {
                min_rtt = rtt;
            }
        }
        
        let avg_rtt = total_time / acks.len() as u32;
        let delivery_rate = if avg_rtt.as_secs_f64() > 0.0 {
            total_bytes as f64 / avg_rtt.as_secs_f64()
        } else {
            self.delivery_rate
        };
        
        // Update BBR with aggregated data
        self.bbr_update(min_rtt, delivery_rate);
    }
}

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