From b315cbca560a31e798286970dc471f816caeb378 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 21 Jul 2026 15:05:25 +0700 Subject: [PATCH] refactor(baseliner): Replace shared Mutex state with thread-safe control snapshots Eliminate lock contention between the baseliner, ratecontroller, and selector by moving to a message-passing architecture using immutable snapshots of reflector state. This improves data consistency and simplifies cross-thread communication. --- src/baseliner.rs | 441 +++++++++++++++++++++++--------------- src/main.rs | 22 +- src/ratecontroller.rs | 213 +++++++++++++----- src/reflector_selector.rs | 276 ++++++++++++++++++++++-- src/util.rs | 1 - 5 files changed, 703 insertions(+), 250 deletions(-) diff --git a/src/baseliner.rs b/src/baseliner.rs index e8a28da..a2b00fa 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -9,29 +9,73 @@ use crate::Config; use crate::SHUTDOWN; use crate::metrics::{Metric, MetricsSender}; use crate::pinger::PingReply; -use crate::util::ArcMutex; -use crate::util::MutexExt; use flume::{Receiver, Sender}; use log::{debug, info}; use std::collections::HashMap; use std::net::IpAddr; use std::sync::atomic::Ordering; +use std::time::Duration; use std::time::Instant; -#[derive(Copy, Clone)] -pub struct ReflectorStats { - pub down_ewma: f64, - pub up_ewma: f64, - pub last_receive_time_s: Instant, +const STALE_AFTER: Duration = Duration::from_secs(30); +const ANOMALY_THRESHOLD_MS: f64 = 5_000.0; +const BASELINE_HALF_LIFE_SECS: f64 = 135.0; +const RECENT_HALF_LIFE_SECS: f64 = 0.4; + +#[derive(Clone, Copy, Debug)] +pub struct EwmaStats { + pub down: f64, + pub up: f64, +} + +impl EwmaStats { + fn from_reply(reply: &PingReply) -> Self { + Self { + down: reply.down_time, + up: reply.up_time, + } + } + + fn update(&mut self, sample: Self, factor: f64) { + self.down = self.down * factor + sample.down * (1.0 - factor); + self.up = self.up * factor + sample.up * (1.0 - factor); + } +} + +#[derive(Clone, Copy, Debug)] +pub struct ReflectorState { + pub baseline: EwmaStats, + pub recent: EwmaStats, + pub last_receive_at: Instant, +} + +impl ReflectorState { + fn new(sample: EwmaStats, received_at: Instant) -> Self { + Self { + baseline: sample, + recent: sample, + last_receive_at: received_at, + } + } + + fn reset(&mut self, sample: EwmaStats, received_at: Instant) { + *self = Self::new(sample, received_at); + } +} + +#[derive(Clone, Debug)] +pub struct ControlSnapshot { + pub generated_at: Instant, + pub reflectors: HashMap, } pub struct Baseliner { pub config: Config, - pub owd_baseline: ArcMutex>, - pub owd_recent: ArcMutex>, pub reselect_trigger: Sender, pub start_time: Instant, pub stats_rx: Receiver, + pub control_tx: Sender, + pub selection_tx: Option>, pub baseline_metrics: MetricsSender, pub event_metrics: MetricsSender, } @@ -40,146 +84,127 @@ fn ewma_factor(tick: f64, dur: f64) -> f64 { ((0.5_f64).ln() / (dur / tick)).exp() } -fn update_reflector_stats( - mut baseline: ReflectorStats, - mut recent: ReflectorStats, - down_time: f64, - up_time: f64, - receive_time: Instant, +fn update_reflector_state( + state: &mut ReflectorState, + sample: EwmaStats, + received_at: Instant, start_time: Instant, slow_factor: f64, fast_factor: f64, -) -> (ReflectorStats, ReflectorStats, bool) { - if receive_time - .duration_since(baseline.last_receive_time_s) - .as_secs_f64() - > 30.0 - || receive_time - .duration_since(recent.last_receive_time_s) - .as_secs_f64() - > 30.0 +) -> bool { + if received_at.saturating_duration_since(state.last_receive_at) > STALE_AFTER { + state.reset(sample, received_at); + } + + if sample.up > state.baseline.up + ANOMALY_THRESHOLD_MS + || sample.down > state.baseline.down + ANOMALY_THRESHOLD_MS { - baseline.down_ewma = down_time; - baseline.up_ewma = up_time; - recent.down_ewma = down_time; - recent.up_ewma = up_time; + state.last_receive_at = start_time; + return true; } - baseline.last_receive_time_s = receive_time; - recent.last_receive_time_s = receive_time; + state.baseline.update(sample, slow_factor); + state.recent.update(sample, fast_factor); - if up_time > baseline.up_ewma + 5000.0 || down_time > baseline.down_ewma + 5000.0 { - baseline.last_receive_time_s = start_time; - recent.last_receive_time_s = start_time; - return (baseline, recent, true); + state.baseline.down = state.baseline.down.min(state.recent.down); + state.baseline.up = state.baseline.up.min(state.recent.up); + state.last_receive_at = received_at; + + false +} + +fn process_reply( + reflectors: &mut HashMap, + reply: &PingReply, + start_time: Instant, + slow_factor: f64, + fast_factor: f64, +) -> (ReflectorState, bool) { + let sample = EwmaStats::from_reply(reply); + let state = reflectors + .entry(reply.reflector) + .or_insert_with(|| ReflectorState::new(sample, reply.last_receive_time_s)); + let anomaly = update_reflector_state( + state, + sample, + reply.last_receive_time_s, + start_time, + slow_factor, + fast_factor, + ); + + (*state, anomaly) +} + +fn control_snapshot( + reflectors: &HashMap, + generated_at: Instant, +) -> ControlSnapshot { + ControlSnapshot { + generated_at, + reflectors: reflectors.clone(), } - - baseline.down_ewma = baseline.down_ewma * slow_factor + (1.0 - slow_factor) * down_time; - baseline.up_ewma = baseline.up_ewma * slow_factor + (1.0 - slow_factor) * up_time; - - recent.down_ewma = recent.down_ewma * fast_factor + (1.0 - fast_factor) * down_time; - recent.up_ewma = recent.up_ewma * fast_factor + (1.0 - fast_factor) * up_time; - - if baseline.down_ewma > recent.down_ewma { - baseline.down_ewma = recent.down_ewma; - } - - if baseline.up_ewma > recent.up_ewma { - baseline.up_ewma = recent.up_ewma; - } - - (baseline, recent, false) } impl Baseliner { pub fn run(&self) -> anyhow::Result<()> { - /* - * 135 seconds to decay to 50% for the slow factor and - * 0.4 seconds to decay to 50% for the fast factor. - * The fast one can be adjusted to tune, try anything from 0.01 to 3.0 to get more or less sensitivity - * with more sensitivity we respond faster to bloat, but are at risk from triggering due to lag spikes that - * aren't bloat related, with less sensitivity (bigger numbers) we smooth through quick spikes - * but take longer to respond to real bufferbloat - */ - let slow_factor = ewma_factor(self.config.tick_interval, 135.0); - let fast_factor = ewma_factor(self.config.tick_interval, 0.4); + let mut reflectors = HashMap::::new(); + + let slow_factor = ewma_factor(self.config.tick_interval, BASELINE_HALF_LIFE_SECS); + let fast_factor = ewma_factor(self.config.tick_interval, RECENT_HALF_LIFE_SECS); loop { if SHUTDOWN.load(Ordering::Relaxed) { info!("Baseliner shutting down"); return Ok(()); } - let time_data = self.stats_rx.recv()?; - let mut owd_baseline_map = self.owd_baseline.lock_anyhow()?; - let mut owd_recent_map = self.owd_recent.lock_anyhow()?; - - let owd_baseline_new = ReflectorStats { - down_ewma: time_data.down_time, - up_ewma: time_data.up_time, - last_receive_time_s: time_data.last_receive_time_s, - }; - - let owd_recent_new = ReflectorStats { - down_ewma: time_data.down_time, - up_ewma: time_data.up_time, - last_receive_time_s: time_data.last_receive_time_s, - }; - - let owd_baseline = owd_baseline_map - .entry(time_data.reflector) - .or_insert(owd_baseline_new); - - let owd_recent = owd_recent_map - .entry(time_data.reflector) - .or_insert(owd_recent_new); - - let (updated_baseline, updated_recent, anomaly) = update_reflector_stats( - *owd_baseline, - *owd_recent, - time_data.down_time, - time_data.up_time, - time_data.last_receive_time_s, + let reply = self.stats_rx.recv()?; + let reflector = reply.reflector; + let (state, anomaly) = process_reply( + &mut reflectors, + &reply, self.start_time, slow_factor, fast_factor, ); - *owd_baseline = updated_baseline; - *owd_recent = updated_recent; - // if this reflection is more than 5 seconds higher than baseline... mark it no good and trigger a reselection if anomaly { info!( "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", - time_data.reflector + reflector ); self.event_metrics.send(Metric::Event { name: "reselection", reason: "anomaly", - reflector: Some(time_data.reflector), + reflector: Some(reflector), tags: &[], }); - // The reselect channel is bounded to 1, - // so we use try_send to avoid blocking if the channel is full let _ = self.reselect_trigger.try_send(true); } self.baseline_metrics.send(Metric::Baseline { - reflector: time_data.reflector, - baseline_up_ewma: owd_baseline.up_ewma, - baseline_down_ewma: owd_baseline.down_ewma, - recent_up_ewma: owd_recent.up_ewma, - recent_down_ewma: owd_recent.down_ewma, + reflector, + baseline_up_ewma: state.baseline.up, + baseline_down_ewma: state.baseline.down, + recent_up_ewma: state.recent.up, + recent_down_ewma: state.recent.down, }); debug!( "Reflector {} up baseline = {} down baseline = {}", - time_data.reflector, owd_baseline.up_ewma, owd_baseline.down_ewma + reflector, state.baseline.up, state.baseline.down ); debug!( "Reflector {} up recent = {} down recent = {}", - time_data.reflector, owd_recent.up_ewma, owd_recent.down_ewma + reflector, state.recent.up, state.recent.down ); + + let snapshot = control_snapshot(&reflectors, Instant::now()); + if let Some(selection_tx) = &self.selection_tx { + let _ = selection_tx.send(snapshot.clone()); + } + self.control_tx.send(snapshot)?; } } } @@ -187,13 +212,46 @@ impl Baseliner { #[cfg(test)] mod tests { use super::*; + use crate::config::MeasurementType; use std::time::Duration; - fn stats(down_ewma: f64, up_ewma: f64, last_receive_time_s: Instant) -> ReflectorStats { - ReflectorStats { - down_ewma, - up_ewma, - last_receive_time_s, + fn state( + baseline_down: f64, + baseline_up: f64, + recent_down: f64, + recent_up: f64, + last_receive_at: Instant, + ) -> ReflectorState { + ReflectorState { + baseline: EwmaStats { + down: baseline_down, + up: baseline_up, + }, + recent: EwmaStats { + down: recent_down, + up: recent_up, + }, + last_receive_at, + } + } + + fn sample(down: f64, up: f64) -> EwmaStats { + EwmaStats { down, up } + } + + fn reply(reflector: IpAddr, down_time: f64, up_time: f64, received_at: Instant) -> PingReply { + PingReply { + reflector, + measurement_type: MeasurementType::Icmp, + seq: 0, + rtt: down_time + up_time, + current_time: 0, + down_time, + up_time, + originate_timestamp: 0, + receive_timestamp: 0, + transmit_timestamp: 0, + last_receive_time_s: received_at, } } @@ -208,11 +266,10 @@ mod tests { fn updates_ewmas() { let start = Instant::now(); let receive_time = start + Duration::from_secs(1); - let (baseline, recent, anomaly) = update_reflector_stats( - stats(100.0, 100.0, start), - stats(100.0, 100.0, start), - 200.0, - 300.0, + let mut state = state(100.0, 100.0, 100.0, 100.0, start); + let anomaly = update_reflector_state( + &mut state, + sample(200.0, 300.0), receive_time, start, 0.9, @@ -220,22 +277,20 @@ mod tests { ); assert!(!anomaly); - assert_close(baseline.down_ewma, 110.0); - assert_close(baseline.up_ewma, 120.0); - assert_close(recent.down_ewma, 150.0); - assert_close(recent.up_ewma, 200.0); - assert_eq!(baseline.last_receive_time_s, receive_time); - assert_eq!(recent.last_receive_time_s, receive_time); + assert_close(state.baseline.down, 110.0); + assert_close(state.baseline.up, 120.0); + assert_close(state.recent.down, 150.0); + assert_close(state.recent.up, 200.0); + assert_eq!(state.last_receive_at, receive_time); } #[test] fn keeps_baseline_at_or_below_recent() { let start = Instant::now(); - let (baseline, recent, anomaly) = update_reflector_stats( - stats(200.0, 200.0, start), - stats(100.0, 100.0, start), - 100.0, - 100.0, + let mut state = state(200.0, 200.0, 100.0, 100.0, start); + let anomaly = update_reflector_state( + &mut state, + sample(100.0, 100.0), start + Duration::from_secs(1), start, 0.9, @@ -243,21 +298,20 @@ mod tests { ); assert!(!anomaly); - assert_close(baseline.down_ewma, 100.0); - assert_close(baseline.up_ewma, 100.0); - assert_close(recent.down_ewma, 100.0); - assert_close(recent.up_ewma, 100.0); + assert_close(state.baseline.down, 100.0); + assert_close(state.baseline.up, 100.0); + assert_close(state.recent.down, 100.0); + assert_close(state.recent.up, 100.0); } #[test] fn resets_stats_after_a_long_gap() { let start = Instant::now(); let receive_time = start + Duration::from_secs(31); - let (baseline, recent, anomaly) = update_reflector_stats( - stats(100.0, 200.0, start), - stats(300.0, 400.0, start), - 500.0, - 600.0, + let mut state = state(100.0, 200.0, 300.0, 400.0, start); + let anomaly = update_reflector_state( + &mut state, + sample(500.0, 600.0), receive_time, start, 0.9, @@ -265,22 +319,41 @@ mod tests { ); assert!(!anomaly); - assert_close(baseline.down_ewma, 500.0); - assert_close(baseline.up_ewma, 600.0); - assert_close(recent.down_ewma, 500.0); - assert_close(recent.up_ewma, 600.0); - assert_eq!(baseline.last_receive_time_s, receive_time); - assert_eq!(recent.last_receive_time_s, receive_time); + assert_close(state.baseline.down, 500.0); + assert_close(state.baseline.up, 600.0); + assert_close(state.recent.down, 500.0); + assert_close(state.recent.up, 600.0); + assert_eq!(state.last_receive_at, receive_time); + } + + #[test] + fn does_not_reset_at_exactly_thirty_seconds() { + let start = Instant::now(); + let receive_time = start + STALE_AFTER; + let mut state = state(100.0, 200.0, 100.0, 200.0, start); + let anomaly = update_reflector_state( + &mut state, + sample(200.0, 300.0), + receive_time, + start, + 0.9, + 0.5, + ); + + assert!(!anomaly); + assert_close(state.baseline.down, 110.0); + assert_close(state.baseline.up, 210.0); + assert_close(state.recent.down, 150.0); + assert_close(state.recent.up, 250.0); } #[test] fn marks_anomalies_for_reselection() { let start = Instant::now(); - let (baseline, recent, anomaly) = update_reflector_stats( - stats(100.0, 200.0, start), - stats(300.0, 400.0, start), - 5_101.0, - 200.0, + let mut state = state(100.0, 200.0, 300.0, 400.0, start); + let anomaly = update_reflector_state( + &mut state, + sample(5_101.0, 200.0), start + Duration::from_secs(1), start, 0.9, @@ -288,23 +361,21 @@ mod tests { ); assert!(anomaly); - assert_close(baseline.down_ewma, 100.0); - assert_close(baseline.up_ewma, 200.0); - assert_close(recent.down_ewma, 300.0); - assert_close(recent.up_ewma, 400.0); - assert_eq!(baseline.last_receive_time_s, start); - assert_eq!(recent.last_receive_time_s, start); + assert_close(state.baseline.down, 100.0); + assert_close(state.baseline.up, 200.0); + assert_close(state.recent.down, 300.0); + assert_close(state.recent.up, 400.0); + assert_eq!(state.last_receive_at, start); } #[test] fn does_not_mark_exact_anomaly_threshold() { let start = Instant::now(); let receive_time = start + Duration::from_secs(1); - let (baseline, recent, anomaly) = update_reflector_stats( - stats(100.0, 200.0, start), - stats(100.0, 200.0, start), - 5_100.0, - 5_200.0, + let mut state = state(100.0, 200.0, 100.0, 200.0, start); + let anomaly = update_reflector_state( + &mut state, + sample(5_100.0, 5_200.0), receive_time, start, 0.9, @@ -312,30 +383,58 @@ mod tests { ); assert!(!anomaly); - assert_eq!(baseline.last_receive_time_s, receive_time); - assert_eq!(recent.last_receive_time_s, receive_time); + assert_eq!(state.last_receive_at, receive_time); } #[test] - fn resets_both_tracks_when_either_track_is_stale() { + fn processed_reply_updates_combined_reflector_state() { let start = Instant::now(); - let recent_time = start + Duration::from_secs(20); - let receive_time = start + Duration::from_secs(31); - let (baseline, recent, anomaly) = update_reflector_stats( - stats(100.0, 200.0, start), - stats(300.0, 400.0, recent_time), - 500.0, - 600.0, - receive_time, + let received_at = start + Duration::from_secs(1); + let reflector = "192.0.2.1".parse().unwrap(); + let mut reflectors = HashMap::new(); + reflectors.insert(reflector, state(100.0, 100.0, 100.0, 100.0, start)); + + let (updated, anomaly) = process_reply( + &mut reflectors, + &reply(reflector, 200.0, 300.0, received_at), start, 0.9, 0.5, ); assert!(!anomaly); - assert_close(baseline.down_ewma, 500.0); - assert_close(baseline.up_ewma, 600.0); - assert_close(recent.down_ewma, 500.0); - assert_close(recent.up_ewma, 600.0); + assert_close(updated.baseline.down, 110.0); + assert_close(updated.baseline.up, 120.0); + assert_close(updated.recent.down, 150.0); + assert_close(updated.recent.up, 200.0); + assert_eq!(updated.last_receive_at, received_at); + assert_close(reflectors[&reflector].recent.up, 200.0); + } + + #[test] + fn snapshot_contains_updated_combined_state() { + let start = Instant::now(); + let received_at = start + Duration::from_secs(1); + let generated_at = received_at + Duration::from_millis(1); + let reflector = "192.0.2.1".parse().unwrap(); + let mut reflectors = HashMap::new(); + + let (updated, anomaly) = process_reply( + &mut reflectors, + &reply(reflector, 20.0, 30.0, received_at), + start, + 0.9, + 0.5, + ); + let snapshot = control_snapshot(&reflectors, generated_at); + + assert!(!anomaly); + assert_eq!(snapshot.generated_at, generated_at); + assert_close( + snapshot.reflectors[&reflector].baseline.down, + updated.baseline.down, + ); + assert_close(snapshot.reflectors[&reflector].recent.up, updated.recent.up); + assert_eq!(snapshot.reflectors[&reflector].last_receive_at, received_at); } } diff --git a/src/main.rs b/src/main.rs index 8a52cda..a8a0b51 100644 --- a/src/main.rs +++ b/src/main.rs @@ -20,7 +20,7 @@ mod reflector_selector; mod time; mod util; -use crate::baseliner::{Baseliner, ReflectorStats}; +use crate::baseliner::Baseliner; use crate::metrics::{Metric, Metrics, MetricsSender}; use ::log::{debug, info, warn}; use flume::RecvTimeoutError; @@ -255,10 +255,14 @@ fn run(config: Config) -> anyhow::Result<()> { active_count: active_reflector_count, } = setup_reflectors(&config)?; - let owd_baseline = Arc::new(Mutex::new(HashMap::::new())); - let owd_recent = Arc::new(Mutex::new(HashMap::::new())); - let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded(); + let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded(); + let (selection_snapshot_tx, selection_snapshot_rx) = if reselection_enabled { + let (tx, rx) = flume::unbounded(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; let (error_tx, error_rx) = flume::unbounded::(); let (reselect_tx, reselect_rx) = flume::bounded(1); @@ -322,11 +326,11 @@ fn run(config: Config) -> anyhow::Result<()> { let baseliner = Baseliner { config: config.clone(), - owd_baseline: Arc::clone(&owd_baseline), - owd_recent: Arc::clone(&owd_recent), reselect_trigger: reselect_tx.clone(), start_time, stats_rx: baseliner_stats_rx, + control_tx: control_snapshot_tx, + selection_tx: selection_snapshot_tx, baseline_metrics, event_metrics: event_metrics.clone(), }; @@ -386,7 +390,8 @@ fn run(config: Config) -> anyhow::Result<()> { if reselection_enabled { let reflector_selector = ReflectorSelector { config: config.clone(), - owd_recent: Arc::clone(&owd_recent), + snapshot_rx: selection_snapshot_rx + .expect("reselection snapshot receiver must exist when reselection is enabled"), reflector_peers_lock: Arc::clone(&reflector_peers), reflector_pool, trigger_channel: reselect_rx, @@ -418,8 +423,7 @@ fn run(config: Config) -> anyhow::Result<()> { let mut ratecontroller = Ratecontroller::new( config.clone(), - owd_baseline, - owd_recent, + control_snapshot_rx, reflector_peers, reselect_tx, dl_direction, diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index 32ecc61..8b0bf49 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -5,19 +5,19 @@ // // SPDX-License-Identifier: MPL-2.0 +use crate::Config; use crate::SHUTDOWN; +use crate::baseliner::ControlSnapshot; use crate::metrics::{Metric, MetricsSender}; use crate::platform::{ InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend, interface_stats_provider, traffic_control_backend, }; use crate::time::Time; -use crate::util::{ArcMutex, ArcRwLock, MutexExt, RwLockExt}; -use crate::{Config, ReflectorStats}; -use flume::Sender; +use crate::util::{ArcRwLock, RwLockExt}; +use flume::{Receiver, Sender}; use log::{debug, info, warn}; use rustix::time::ClockId; -use std::collections::HashMap; use std::fs::File; use std::io::Write; use std::net::IpAddr; @@ -105,11 +105,62 @@ impl State { } } +fn drain_latest_snapshot( + snapshot_rx: &Receiver, + latest: &mut Option, +) { + while let Ok(snapshot) = snapshot_rx.try_recv() { + if latest + .as_ref() + .is_none_or(|current| snapshot.generated_at >= current.generated_at) + { + *latest = Some(snapshot); + } + } +} + +fn deltas_from_snapshot( + snapshot: Option<&ControlSnapshot>, + reflectors: &[IpAddr], + now: Instant, + tick_interval: f64, +) -> (Vec, Vec) { + let mut down = Vec::new(); + let mut up = Vec::new(); + + let Some(snapshot) = snapshot else { + return (down, up); + }; + + for reflector in reflectors { + let Some(state) = snapshot.reflectors.get(reflector) else { + continue; + }; + + if now.duration_since(state.last_receive_at).as_secs_f64() < tick_interval * 2.0 { + let down_delta = state.recent.down - state.baseline.down; + let up_delta = state.recent.up - state.baseline.up; + down.push(down_delta); + up.push(up_delta); + + debug!( + "Reflector: {} down_delay: {} up_delay: {}", + reflector, down_delta, up_delta + ); + } + } + + down.sort_by(|a, b| a.total_cmp(b)); + up.sort_by(|a, b| a.total_cmp(b)); + + (down, up) +} + pub struct Ratecontroller { config: Config, + control_snapshot: Option, + control_snapshot_rx: Receiver, down_direction: StatsDirection, - owd_baseline: ArcMutex>, - owd_recent: ArcMutex>, reflectors_lock: ArcRwLock>, reselect_trigger: Sender, state_dl: State, @@ -123,8 +174,7 @@ pub struct Ratecontroller { impl Ratecontroller { fn new_with_backends( config: Config, - owd_baseline: ArcMutex>, - owd_recent: ArcMutex>, + control_snapshot_rx: Receiver, reflectors_lock: ArcRwLock>, reselect_trigger: Sender, down_direction: StatsDirection, @@ -150,9 +200,9 @@ impl Ratecontroller { Ok(Self { config, + control_snapshot: None, + control_snapshot_rx, down_direction, - owd_baseline, - owd_recent, reflectors_lock, reselect_trigger, state_dl: State::new(dl_shaper, cur_rx, dl_safe_rates), @@ -261,52 +311,24 @@ impl Ratecontroller { } fn update_deltas(&mut self) -> anyhow::Result<()> { - let state_dl = &mut self.state_dl; - let state_ul = &mut self.state_ul; - - state_dl.deltas.clear(); - state_ul.deltas.clear(); - + drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot); let now_t = Instant::now(); - let owd_baseline = self.owd_baseline.lock_anyhow()?; - let owd_recent = self.owd_recent.lock_anyhow()?; let reflectors = self.reflectors_lock.read_anyhow()?; + let (down_deltas, up_deltas) = deltas_from_snapshot( + self.control_snapshot.as_ref(), + &reflectors, + now_t, + self.config.tick_interval, + ); + self.state_dl.deltas = down_deltas; + self.state_ul.deltas = up_deltas; - for reflector in reflectors.iter() { - // only consider this data if it's less than 2 * tick_duration seconds old - if owd_baseline.contains_key(reflector) - && owd_recent.contains_key(reflector) - && now_t - .duration_since(owd_recent[reflector].last_receive_time_s) - .as_secs_f64() - < self.config.tick_interval * 2.0 - { - state_dl - .deltas - .push(owd_recent[reflector].down_ewma - owd_baseline[reflector].down_ewma); - state_ul - .deltas - .push(owd_recent[reflector].up_ewma - owd_baseline[reflector].up_ewma); - - debug!( - "Reflector: {} down_delay: {} up_delay: {}", - reflector, - state_dl.deltas.last().unwrap(), - state_ul.deltas.last().unwrap() - ); - } - } - - // sort owd's lowest to highest - state_dl.deltas.sort_by(|a, b| a.total_cmp(b)); - state_ul.deltas.sort_by(|a, b| a.total_cmp(b)); - - if state_dl.deltas.len() < 5 || state_ul.deltas.len() < 5 { + if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 { // trigger reselection warn!( "Not enough delta values (D: {}, U: {}, need 5), triggering reselection", - state_dl.deltas.len(), - state_ul.deltas.len() + self.state_dl.deltas.len(), + self.state_ul.deltas.len() ); let _ = self.reselect_trigger.try_send(true); } @@ -318,8 +340,7 @@ impl Ratecontroller { impl Ratecontroller { pub fn new( config: Config, - owd_baseline: ArcMutex>, - owd_recent: ArcMutex>, + control_snapshot_rx: Receiver, reflectors_lock: ArcRwLock>, reselect_trigger: Sender, down_direction: StatsDirection, @@ -328,8 +349,7 @@ impl Ratecontroller { ) -> anyhow::Result { Self::new_with_backends( config, - owd_baseline, - owd_recent, + control_snapshot_rx, reflectors_lock, reselect_trigger, down_direction, @@ -530,3 +550,88 @@ impl Ratecontroller { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::baseliner::{EwmaStats, ReflectorState}; + use std::collections::HashMap; + + fn reflector_state( + baseline_down: f64, + baseline_up: f64, + recent_down: f64, + recent_up: f64, + last_receive_at: Instant, + ) -> ReflectorState { + ReflectorState { + baseline: EwmaStats { + down: baseline_down, + up: baseline_up, + }, + recent: EwmaStats { + down: recent_down, + up: recent_up, + }, + last_receive_at, + } + } + + fn snapshot(generated_at: Instant, marker: f64) -> ControlSnapshot { + let reflector = "192.0.2.1".parse().unwrap(); + ControlSnapshot { + generated_at, + reflectors: HashMap::from([( + reflector, + reflector_state(marker, marker, marker, marker, generated_at), + )]), + } + } + + #[test] + fn drains_queued_snapshots_and_keeps_the_newest() { + let start = Instant::now(); + let newest_at = start + Duration::from_secs(2); + let (tx, rx) = flume::unbounded(); + tx.send(snapshot(start + Duration::from_secs(1), 1.0)) + .unwrap(); + tx.send(snapshot(newest_at, 2.0)).unwrap(); + let mut latest = None; + + drain_latest_snapshot(&rx, &mut latest); + + let latest = latest.unwrap(); + assert!(rx.is_empty()); + assert_eq!(latest.generated_at, newest_at); + assert_eq!( + latest.reflectors[&"192.0.2.1".parse().unwrap()].recent.down, + 2.0 + ); + } + + #[test] + fn calculates_sorted_deltas_from_active_fresh_snapshot_state() { + let now = Instant::now(); + let first: IpAddr = "192.0.2.1".parse().unwrap(); + let second: IpAddr = "192.0.2.2".parse().unwrap(); + let stale: IpAddr = "192.0.2.3".parse().unwrap(); + let inactive: IpAddr = "192.0.2.4".parse().unwrap(); + let snapshot = ControlSnapshot { + generated_at: now, + reflectors: HashMap::from([ + (first, reflector_state(10.0, 20.0, 14.0, 28.0, now)), + (second, reflector_state(10.0, 20.0, 11.0, 22.0, now)), + ( + stale, + reflector_state(10.0, 20.0, 110.0, 220.0, now - Duration::from_secs(3)), + ), + (inactive, reflector_state(10.0, 20.0, 60.0, 80.0, now)), + ]), + }; + + let (down, up) = deltas_from_snapshot(Some(&snapshot), &[first, second, stale], now, 1.0); + + assert_eq!(down, vec![1.0, 4.0]); + assert_eq!(up, vec![2.0, 8.0]); + } +} diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index c88bd02..5effe0c 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -5,22 +5,108 @@ // // SPDX-License-Identifier: MPL-2.0 +use crate::Config; use crate::SHUTDOWN; +use crate::baseliner::ControlSnapshot; use crate::metrics::{Metric, MetricsSender}; -use crate::util::{ArcMutex, MutexExt, RwLockExt}; -use crate::{Config, ReflectorStats}; -use flume::Receiver; +use crate::util::RwLockExt; +use flume::{Receiver, RecvError, RecvTimeoutError, Selector}; use log::{debug, info}; -use std::collections::HashMap; use std::net::IpAddr; use std::sync::atomic::Ordering; use std::sync::{Arc, RwLock}; use std::thread::sleep; -use std::time::Duration; +use std::time::{Duration, Instant}; + +enum SelectorEvent { + Trigger(Result), + Snapshot(Result), +} + +#[derive(Debug, Eq, PartialEq)] +enum WaitOutcome { + Triggered, + DeadlineReached, + ChannelsClosed, +} + +fn wait_for_trigger_or_deadline( + trigger_rx: &Receiver, + snapshot_rx: &Receiver, + latest_snapshot: &mut Option, + deadline: Instant, +) -> WaitOutcome { + let mut trigger_connected = true; + let mut snapshot_connected = true; + + loop { + if Instant::now() >= deadline { + return WaitOutcome::DeadlineReached; + } + + let event = match (trigger_connected, snapshot_connected) { + (true, true) => Selector::new() + .recv(trigger_rx, SelectorEvent::Trigger) + .recv(snapshot_rx, SelectorEvent::Snapshot) + .wait_deadline(deadline) + .ok(), + (true, false) => match trigger_rx.recv_deadline(deadline) { + Ok(trigger) => Some(SelectorEvent::Trigger(Ok(trigger))), + Err(RecvTimeoutError::Timeout) => None, + Err(RecvTimeoutError::Disconnected) => { + Some(SelectorEvent::Trigger(Err(RecvError::Disconnected))) + } + }, + (false, true) => match snapshot_rx.recv_deadline(deadline) { + Ok(snapshot) => Some(SelectorEvent::Snapshot(Ok(snapshot))), + Err(RecvTimeoutError::Timeout) => None, + Err(RecvTimeoutError::Disconnected) => { + Some(SelectorEvent::Snapshot(Err(RecvError::Disconnected))) + } + }, + (false, false) => return WaitOutcome::ChannelsClosed, + }; + + match event { + Some(SelectorEvent::Trigger(Ok(_))) => return WaitOutcome::Triggered, + Some(SelectorEvent::Trigger(Err(_))) => trigger_connected = false, + Some(SelectorEvent::Snapshot(Ok(snapshot))) => *latest_snapshot = Some(snapshot), + Some(SelectorEvent::Snapshot(Err(_))) => snapshot_connected = false, + None => return WaitOutcome::DeadlineReached, + } + } +} + +fn receive_snapshots_until( + snapshot_rx: &Receiver, + latest_snapshot: &mut Option, + deadline: Instant, +) { + loop { + if Instant::now() >= deadline { + return; + } + + match snapshot_rx.recv_deadline(deadline) { + Ok(snapshot) => *latest_snapshot = Some(snapshot), + Err(RecvTimeoutError::Timeout) => return, + Err(RecvTimeoutError::Disconnected) => { + sleep(deadline.saturating_duration_since(Instant::now())); + return; + } + } + } +} + +fn recent_rtt_for_peer(snapshot: Option<&ControlSnapshot>, peer: &IpAddr) -> Option { + snapshot + .and_then(|snapshot| snapshot.reflectors.get(peer)) + .map(|state| (state.recent.down + state.recent.up) as u64) +} pub struct ReflectorSelector { pub config: Config, - pub owd_recent: ArcMutex>, + pub snapshot_rx: Receiver, pub reflector_peers_lock: Arc>>, pub reflector_pool: Vec, pub trigger_channel: Receiver, @@ -31,10 +117,15 @@ impl ReflectorSelector { pub fn run(&self) -> anyhow::Result<()> { let mut selector_sleep_time = Duration::new(30, 0); let mut reselection_count = 0; + let mut latest_snapshot = None; let baseline_sleep_time = Duration::from_secs_f64(self.config.tick_interval * std::f64::consts::PI); // Initial wait of several seconds to allow some OWD data to build up - sleep(baseline_sleep_time); + receive_snapshots_until( + &self.snapshot_rx, + &mut latest_snapshot, + Instant::now() + baseline_sleep_time, + ); loop { if SHUTDOWN.load(Ordering::Relaxed) { @@ -46,10 +137,19 @@ impl ReflectorSelector { * or it passes the timeout. In any case we don't care about the result of this function, * so we ignore the result of it. */ - let triggered = self - .trigger_channel - .recv_timeout(selector_sleep_time) - .is_ok(); + let triggered = match wait_for_trigger_or_deadline( + &self.trigger_channel, + &self.snapshot_rx, + &mut latest_snapshot, + Instant::now() + selector_sleep_time, + ) { + WaitOutcome::Triggered => true, + WaitOutcome::DeadlineReached => false, + WaitOutcome::ChannelsClosed => { + info!("Reflector selector channels closed, shutting down"); + return Ok(()); + } + }; reselection_count += 1; info!("Starting reselection [#{}]", reselection_count); self.metrics.send(Metric::Event { @@ -93,17 +193,19 @@ impl ReflectorSelector { debug!("Waiting for candidates to be baselined"); // Wait for several seconds to allow all reflectors to be re-baselined - sleep(baseline_sleep_time); + receive_snapshots_until( + &self.snapshot_rx, + &mut latest_snapshot, + Instant::now() + baseline_sleep_time, + ); // Re-acquire the lock when we wake up again reflectors_peers = self.reflector_peers_lock.write_anyhow()?; let mut candidates = Vec::new(); - let owd_recent = self.owd_recent.lock_anyhow()?; for peer in next_peers { - if owd_recent.contains_key(&peer) { - let rtt = (owd_recent[&peer].down_ewma + owd_recent[&peer].up_ewma) as u64; + if let Some(rtt) = recent_rtt_for_peer(latest_snapshot.as_ref(), &peer) { candidates.push((peer, rtt)); info!("Candidate reflector: {} RTT: {}", peer, rtt); } else { @@ -153,3 +255,147 @@ impl ReflectorSelector { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::baseliner::{EwmaStats, ReflectorState}; + use std::collections::HashMap; + use std::thread; + + fn snapshot(generated_at: Instant) -> ControlSnapshot { + ControlSnapshot { + generated_at, + reflectors: HashMap::new(), + } + } + + fn snapshot_with_rtt(generated_at: Instant, peer: IpAddr, rtt: f64) -> ControlSnapshot { + let mut reflectors = HashMap::new(); + reflectors.insert( + peer, + ReflectorState { + baseline: EwmaStats { + down: rtt / 2.0, + up: rtt / 2.0, + }, + recent: EwmaStats { + down: rtt / 2.0, + up: rtt / 2.0, + }, + last_receive_at: generated_at, + }, + ); + ControlSnapshot { + generated_at, + reflectors, + } + } + + #[test] + fn wait_retains_the_newest_snapshot() { + let (_trigger_tx, trigger_rx) = flume::unbounded(); + let (snapshot_tx, snapshot_rx) = flume::unbounded(); + let generated_at = Instant::now(); + snapshot_tx.send(snapshot(generated_at)).unwrap(); + snapshot_tx + .send(snapshot(generated_at + Duration::from_millis(1))) + .unwrap(); + snapshot_tx + .send(snapshot(generated_at + Duration::from_millis(2))) + .unwrap(); + let mut latest_snapshot = None; + + let result = wait_for_trigger_or_deadline( + &trigger_rx, + &snapshot_rx, + &mut latest_snapshot, + Instant::now() + Duration::from_millis(20), + ); + + assert_eq!(result, WaitOutcome::DeadlineReached); + assert_eq!( + latest_snapshot.unwrap().generated_at, + generated_at + Duration::from_millis(2) + ); + } + + #[test] + fn snapshots_do_not_extend_the_deadline() { + let (_trigger_tx, trigger_rx) = flume::unbounded(); + let (snapshot_tx, snapshot_rx) = flume::unbounded(); + let producer = thread::spawn(move || { + let stop_at = Instant::now() + Duration::from_millis(500); + while Instant::now() < stop_at { + if snapshot_tx.send(snapshot(Instant::now())).is_err() { + break; + } + sleep(Duration::from_millis(1)); + } + }); + let mut latest_snapshot = None; + let started_at = Instant::now(); + + let result = wait_for_trigger_or_deadline( + &trigger_rx, + &snapshot_rx, + &mut latest_snapshot, + started_at + Duration::from_millis(30), + ); + let elapsed = started_at.elapsed(); + drop(snapshot_rx); + producer.join().unwrap(); + + assert_eq!(result, WaitOutcome::DeadlineReached); + assert!(latest_snapshot.is_some()); + assert!(elapsed < Duration::from_millis(250), "elapsed: {elapsed:?}"); + } + + #[test] + fn trigger_interrupts_the_wait() { + let (trigger_tx, trigger_rx) = flume::unbounded(); + let (_snapshot_tx, snapshot_rx) = flume::unbounded(); + trigger_tx.send(true).unwrap(); + let mut latest_snapshot = None; + let started_at = Instant::now(); + + let result = wait_for_trigger_or_deadline( + &trigger_rx, + &snapshot_rx, + &mut latest_snapshot, + started_at + Duration::from_secs(1), + ); + + assert_eq!(result, WaitOutcome::Triggered); + assert!(started_at.elapsed() < Duration::from_millis(250)); + } + + #[test] + fn baseline_wait_retains_snapshot_for_evaluation() { + let (snapshot_tx, snapshot_rx) = flume::unbounded(); + let peer = "192.0.2.1".parse().unwrap(); + let generated_at = Instant::now(); + snapshot_tx + .send(snapshot_with_rtt(generated_at, peer, 20.0)) + .unwrap(); + snapshot_tx + .send(snapshot_with_rtt( + generated_at + Duration::from_millis(1), + peer, + 42.0, + )) + .unwrap(); + let mut latest_snapshot = None; + + receive_snapshots_until( + &snapshot_rx, + &mut latest_snapshot, + Instant::now() + Duration::from_millis(20), + ); + + assert_eq!( + recent_rtt_for_peer(latest_snapshot.as_ref(), &peer), + Some(42) + ); + } +} diff --git a/src/util.rs b/src/util.rs index 725a908..d7d1855 100644 --- a/src/util.rs +++ b/src/util.rs @@ -5,7 +5,6 @@ use anyhow::anyhow; use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard}; -pub type ArcMutex = Arc>; pub type ArcRwLock = Arc>; pub trait MutexExt {