//! MMP algorithmic building blocks. //! //! Pure computational types with no dependency on peer or node state. //! Each is independently testable. use std::collections::VecDeque; use std::time::Instant; use crate::mmp::{EWMA_LONG_ALPHA, EWMA_SHORT_ALPHA}; // ============================================================================ // Jitter Estimator (RFC 3550 §6.4.1) // ============================================================================ /// Interarrival jitter estimator using RFC 3550 algorithm. /// /// Maintains a smoothed jitter estimate (α = 1/16) from the absolute /// difference in one-way transit times between consecutive frames. /// Uses integer arithmetic scaled by 16 to avoid floating-point. pub struct JitterEstimator { /// Scaled jitter estimate (×16 for integer arithmetic). jitter_q4: i64, } impl JitterEstimator { pub fn new() -> Self { Self { jitter_q4: 0 } } /// Update with transit time delta between consecutive frames. /// /// `transit_delta` = (R_i - R_{i-1}) - (S_i - S_{i-1}) in microseconds. pub fn update(&mut self, transit_delta: i32) { // RFC 3550: J = J + (1/16)(|D(i)| - J) // Scaled: J_q4 = J_q4 + (|D| - J_q4/16) // = J_q4 + |D| - J_q4 >> 4 let abs_d = (transit_delta as i64).unsigned_abs() as i64; self.jitter_q4 += abs_d - (self.jitter_q4 >> 4); } /// Current jitter estimate in microseconds. pub fn jitter_us(&self) -> u32 { (self.jitter_q4 >> 4) as u32 } } impl Default for JitterEstimator { fn default() -> Self { Self::new() } } // ============================================================================ // SRTT Estimator (Jacobson, RFC 6298) // ============================================================================ /// Smoothed RTT estimator using Jacobson's algorithm. /// /// SRTT and RTTVAR are maintained in microseconds using integer arithmetic. pub struct SrttEstimator { /// Smoothed RTT (microseconds). srtt_us: i64, /// RTT variance (microseconds). rttvar_us: i64, /// Whether the first sample has been applied. initialized: bool, } impl SrttEstimator { pub fn new() -> Self { Self { srtt_us: 0, rttvar_us: 0, initialized: false, } } /// Feed an RTT sample in microseconds. pub fn update(&mut self, rtt_us: i64) { if !self.initialized { // RFC 6298 §2.2: first measurement self.srtt_us = rtt_us; self.rttvar_us = rtt_us / 2; self.initialized = true; } else { // RFC 6298 §2.3: // RTTVAR = (1 - β) * RTTVAR + β * |SRTT - R'| β = 1/4 // SRTT = (1 - α) * SRTT + α * R' α = 1/8 let err = (self.srtt_us - rtt_us).abs(); self.rttvar_us = self.rttvar_us - (self.rttvar_us >> 2) + (err >> 2); self.srtt_us = self.srtt_us - (self.srtt_us >> 3) + (rtt_us >> 3); } } pub fn srtt_us(&self) -> i64 { self.srtt_us } pub fn rttvar_us(&self) -> i64 { self.rttvar_us } pub fn initialized(&self) -> bool { self.initialized } /// Retransmission timeout = SRTT + max(4 * RTTVAR, 1s), floored at 1s. pub fn rto_us(&self) -> i64 { let rto = self.srtt_us + (self.rttvar_us << 2).max(1_000_000); rto.max(1_000_000) } } impl Default for SrttEstimator { fn default() -> Self { Self::new() } } // ============================================================================ // Dual EWMA Trend Detector // ============================================================================ /// Dual EWMA for trend detection on a single metric. /// /// Short-term (α=1/4) tracks recent conditions; long-term (α=1/32) /// establishes a stable baseline. Divergence indicates trend direction. pub struct DualEwma { short: f64, long: f64, initialized: bool, } impl DualEwma { pub fn new() -> Self { Self { short: 0.0, long: 0.0, initialized: false, } } pub fn update(&mut self, sample: f64) { if !self.initialized { self.short = sample; self.long = sample; self.initialized = true; } else { self.short += EWMA_SHORT_ALPHA * (sample - self.short); self.long += EWMA_LONG_ALPHA * (sample - self.long); } } pub fn short(&self) -> f64 { self.short } pub fn long(&self) -> f64 { self.long } pub fn initialized(&self) -> bool { self.initialized } } impl Default for DualEwma { fn default() -> Self { Self::new() } } // ============================================================================ // One-Way Delay Trend Detector // ============================================================================ /// OWD trend detector using linear regression over a ring buffer. /// /// Stores (sequence, owd_us) samples and computes the slope via /// least-squares regression. The slope (µs/s) indicates whether /// queuing delay is increasing (congestion) or stable. pub struct OwdTrendDetector { samples: VecDeque<(u32, i64)>, capacity: usize, } impl OwdTrendDetector { pub fn new(capacity: usize) -> Self { Self { samples: VecDeque::with_capacity(capacity), capacity, } } /// Clear all samples, keeping the same capacity. pub fn clear(&mut self) { self.samples.clear(); } /// Add an OWD sample. /// /// `seq` is a monotonic sequence number (e.g., truncated frame counter). /// `owd_us` is the relative one-way delay in microseconds (R_i - S_i). pub fn push(&mut self, seq: u32, owd_us: i64) { if self.samples.len() == self.capacity { self.samples.pop_front(); } self.samples.push_back((seq, owd_us)); } /// Compute the OWD trend as a slope in µs/second. /// /// Uses simple linear regression: slope = Σ((x-x̄)(y-ȳ)) / Σ((x-x̄)²) /// where x = sequence number and y = owd_us. /// /// Returns 0 if fewer than 2 samples. pub fn trend_us_per_sec(&self) -> i32 { let n = self.samples.len(); if n < 2 { return 0; } let n_f = n as f64; let sum_x: f64 = self.samples.iter().map(|(s, _)| *s as f64).sum(); let sum_y: f64 = self.samples.iter().map(|(_, y)| *y as f64).sum(); let mean_x = sum_x / n_f; let mean_y = sum_y / n_f; let mut num = 0.0; let mut den = 0.0; for &(x, y) in &self.samples { let dx = x as f64 - mean_x; let dy = y as f64 - mean_y; num += dx * dy; den += dx * dx; } if den.abs() < f64::EPSILON { return 0; } // slope is in µs/packet. Convert to µs/second assuming ~1ms inter-packet // spacing as a rough estimate. The raw slope per packet is more useful // for trend detection than an absolute rate, but the wire format specifies // µs/s. We report the raw per-packet slope scaled by 1000. let slope_per_packet = num / den; (slope_per_packet * 1000.0) as i32 } pub fn len(&self) -> usize { self.samples.len() } pub fn is_empty(&self) -> bool { self.samples.is_empty() } } // ============================================================================ // ETX // ============================================================================ /// Compute Expected Transmission Count from bidirectional delivery ratios. /// /// ETX = 1 / (d_f × d_r) where d_f and d_r are forward and reverse /// delivery probabilities (1.0 = perfect, 0.0 = no delivery). /// /// Clamped to [1.0, 100.0]. pub fn compute_etx(d_forward: f64, d_reverse: f64) -> f64 { let product = d_forward * d_reverse; if product <= 0.0 { return 100.0; } (1.0 / product).clamp(1.0, 100.0) } // ============================================================================ // Spin Bit // ============================================================================ /// Spin bit state for passive RTT estimation. /// /// Uses asymmetric roles (initiator/responder) per the MMP design: /// - **Initiator**: flips spin value on each received frame; measures RTT /// from edge-to-edge intervals. /// - **Responder**: copies received spin bit into outgoing frames, with a /// counter guard to filter reordered frames. pub struct SpinBitState { is_initiator: bool, current_value: bool, /// Highest counter observed with a spin edge (responder guard). highest_counter_for_spin: u64, /// Time of last spin edge (initiator only, for RTT measurement). last_edge_time: Option, } impl SpinBitState { pub fn new(is_initiator: bool) -> Self { Self { is_initiator, current_value: false, highest_counter_for_spin: 0, last_edge_time: None, } } /// Check if this is the spin bit initiator. pub fn is_initiator(&self) -> bool { self.is_initiator } /// Get the spin bit value to set on an outgoing frame. pub fn tx_bit(&self) -> bool { self.current_value } /// Process a received frame's spin bit. /// /// Returns an RTT sample duration if an edge was detected (initiator only). pub fn rx_observe( &mut self, received_bit: bool, counter: u64, now: Instant, ) -> Option { if self.is_initiator { // Initiator: when the reflected bit matches what we sent, // that completes a round trip. Record the edge time, then // flip for the next cycle. if received_bit == self.current_value { let rtt = self.last_edge_time.map(|t| now.duration_since(t)); self.last_edge_time = Some(now); self.current_value = !self.current_value; rtt } else { None } } else { // Responder: copy received bit, but only if counter is higher // (reordering guard) if counter > self.highest_counter_for_spin { self.highest_counter_for_spin = counter; self.current_value = received_bit; } None } } } // ============================================================================ // Tests // ============================================================================ #[cfg(test)] mod tests { use super::*; #[test] fn test_jitter_zero_input() { let mut j = JitterEstimator::new(); j.update(0); assert_eq!(j.jitter_us(), 0); } #[test] fn test_jitter_convergence() { let mut j = JitterEstimator::new(); // Feed constant transit delta of 1000µs for _ in 0..200 { j.update(1000); } // Should converge near 1000µs let jitter = j.jitter_us(); assert!( jitter > 900 && jitter < 1100, "jitter={jitter}, expected ~1000" ); } #[test] fn test_srtt_first_sample() { let mut s = SrttEstimator::new(); s.update(10_000); // 10ms assert_eq!(s.srtt_us(), 10_000); assert_eq!(s.rttvar_us(), 5_000); assert!(s.initialized()); } #[test] fn test_srtt_convergence() { let mut s = SrttEstimator::new(); // Feed constant 50ms RTT for _ in 0..100 { s.update(50_000); } let srtt = s.srtt_us(); assert!((srtt - 50_000).abs() < 1000, "srtt={srtt}, expected ~50000"); } #[test] fn test_dual_ewma_initialization() { let mut e = DualEwma::new(); assert!(!e.initialized()); e.update(100.0); assert!(e.initialized()); assert_eq!(e.short(), 100.0); assert_eq!(e.long(), 100.0); } #[test] fn test_dual_ewma_short_tracks_faster() { let mut e = DualEwma::new(); // Initialize at 0 e.update(0.0); // Jump to 100 for _ in 0..20 { e.update(100.0); } // Short should be closer to 100 than long assert!( e.short() > e.long(), "short={} long={}", e.short(), e.long() ); } #[test] fn test_owd_trend_flat() { let mut d = OwdTrendDetector::new(32); for i in 0..20 { d.push(i, 5000); // constant OWD } let trend = d.trend_us_per_sec(); assert_eq!(trend, 0, "flat OWD should have zero trend"); } #[test] fn test_owd_trend_increasing() { let mut d = OwdTrendDetector::new(32); for i in 0..20 { d.push(i, 5000 + (i as i64) * 100); // increasing by 100µs per packet } let trend = d.trend_us_per_sec(); assert!( trend > 0, "increasing OWD should have positive trend, got {trend}" ); } #[test] fn test_owd_trend_insufficient_samples() { let mut d = OwdTrendDetector::new(32); d.push(0, 5000); assert_eq!(d.trend_us_per_sec(), 0); } #[test] fn test_etx_perfect_link() { assert!((compute_etx(1.0, 1.0) - 1.0).abs() < f64::EPSILON); } #[test] fn test_etx_lossy_link() { // 10% forward loss, 5% reverse loss let etx = compute_etx(0.9, 0.95); assert!(etx > 1.0 && etx < 2.0, "etx={etx}"); } #[test] fn test_etx_zero_delivery() { assert_eq!(compute_etx(0.0, 1.0), 100.0); assert_eq!(compute_etx(1.0, 0.0), 100.0); } #[test] fn test_spin_bit_initiator_rtt() { let mut initiator = SpinBitState::new(true); let mut responder = SpinBitState::new(false); let t0 = Instant::now(); let t1 = t0 + std::time::Duration::from_millis(10); let t2 = t0 + std::time::Duration::from_millis(20); // Initiator sends with spin=false (initial) let bit_to_send = initiator.tx_bit(); assert!(!bit_to_send); // Responder receives, copies bit responder.rx_observe(bit_to_send, 1, t0); assert!(!responder.tx_bit()); // Responder sends back, initiator receives let resp_bit = responder.tx_bit(); let rtt1 = initiator.rx_observe(resp_bit, 2, t1); // First edge: no previous edge to compare assert!(rtt1.is_none()); // Now initiator's spin flipped to true let bit2 = initiator.tx_bit(); assert!(bit2); // Responder receives new bit responder.rx_observe(bit2, 3, t1); assert!(responder.tx_bit()); // Responder sends back, initiator receives let resp_bit2 = responder.tx_bit(); let rtt2 = initiator.rx_observe(resp_bit2, 4, t2); // Second edge: should produce an RTT sample assert!(rtt2.is_some()); } #[test] fn test_spin_bit_responder_counter_guard() { let mut responder = SpinBitState::new(false); // Receive counter=5 with spin=true responder.rx_observe(true, 5, Instant::now()); assert!(responder.tx_bit()); // Reordered packet with counter=3 and spin=false should be ignored responder.rx_observe(false, 3, Instant::now()); assert!(responder.tx_bit()); // unchanged } }