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

@@ -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)]
+const STALE_AFTER: Duration = Duration::from_secs(30);
-pub struct ReflectorStats {
+const ANOMALY_THRESHOLD_MS: f64 = 5_000.0;
-    pub down_ewma: f64,
+const BASELINE_HALF_LIFE_SECS: f64 = 135.0;
-    pub up_ewma: f64,
+const RECENT_HALF_LIFE_SECS: f64 = 0.4;
-    pub last_receive_time_s: Instant,
+
+#[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<IpAddr, ReflectorState>,
 }
 
 pub struct Baseliner {
     pub config: Config,
-    pub owd_baseline: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
-    pub owd_recent: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
     pub stats_rx: Receiver<PingReply>,
+    pub control_tx: Sender<ControlSnapshot>,
+    pub selection_tx: Option<Sender<ControlSnapshot>>,
     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(
+fn update_reflector_state(
-    mut baseline: ReflectorStats,
+    state: &mut ReflectorState,
-    mut recent: ReflectorStats,
+    sample: EwmaStats,
-    down_time: f64,
+    received_at: Instant,
-    up_time: f64,
-    receive_time: Instant,
     start_time: Instant,
     slow_factor: f64,
     fast_factor: f64,
-) -> (ReflectorStats, ReflectorStats, bool) {
+) -> bool {
-    if receive_time
+    if received_at.saturating_duration_since(state.last_receive_at) > STALE_AFTER {
-        .duration_since(baseline.last_receive_time_s)
+        state.reset(sample, received_at);
-        .as_secs_f64()
+    }
-        > 30.0
+
-        || receive_time
+    if sample.up > state.baseline.up + ANOMALY_THRESHOLD_MS
-            .duration_since(recent.last_receive_time_s)
+        || sample.down > state.baseline.down + ANOMALY_THRESHOLD_MS
-            .as_secs_f64()
-            > 30.0
     {
-        baseline.down_ewma = down_time;
+        state.last_receive_at = start_time;
-        baseline.up_ewma = up_time;
+        return true;
-        recent.down_ewma = down_time;
-        recent.up_ewma = up_time;
     }
 
-    baseline.last_receive_time_s = receive_time;
+    state.baseline.update(sample, slow_factor);
-    recent.last_receive_time_s = receive_time;
+    state.recent.update(sample, fast_factor);
 
-    if up_time > baseline.up_ewma + 5000.0 || down_time > baseline.down_ewma + 5000.0 {
+    state.baseline.down = state.baseline.down.min(state.recent.down);
-        baseline.last_receive_time_s = start_time;
+    state.baseline.up = state.baseline.up.min(state.recent.up);
-        recent.last_receive_time_s = start_time;
+    state.last_receive_at = received_at;
-        return (baseline, recent, true);
+
+    false
+}
+
+fn process_reply(
+    reflectors: &mut HashMap<IpAddr, ReflectorState>,
+    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<IpAddr, ReflectorState>,
+    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<()> {
-        /*
+        let mut reflectors = HashMap::<IpAddr, ReflectorState>::new();
-         * 135 seconds to decay to 50% for the slow factor and
+
-         * 0.4 seconds to decay to 50% for the fast factor.
+        let slow_factor = ewma_factor(self.config.tick_interval, BASELINE_HALF_LIFE_SECS);
-         * The fast one can be adjusted to tune, try anything from 0.01 to 3.0 to get more or less sensitivity
+        let fast_factor = ewma_factor(self.config.tick_interval, RECENT_HALF_LIFE_SECS);
-         * 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);
 
         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 reply = self.stats_rx.recv()?;
-            let mut owd_recent_map = self.owd_recent.lock_anyhow()?;
+            let reflector = reply.reflector;
-
+            let (state, anomaly) = process_reply(
-            let owd_baseline_new = ReflectorStats {
+                &mut reflectors,
-                down_ewma: time_data.down_time,
+                &reply,
-                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,
                 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,
+                reflector,
-                baseline_up_ewma: owd_baseline.up_ewma,
+                baseline_up_ewma: state.baseline.up,
-                baseline_down_ewma: owd_baseline.down_ewma,
+                baseline_down_ewma: state.baseline.down,
-                recent_up_ewma: owd_recent.up_ewma,
+                recent_up_ewma: state.recent.up,
-                recent_down_ewma: owd_recent.down_ewma,
+                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 {
+    fn state(
-        ReflectorStats {
+        baseline_down: f64,
-            down_ewma,
+        baseline_up: f64,
-            up_ewma,
+        recent_down: f64,
-            last_receive_time_s,
+        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(
+        let mut state = state(100.0, 100.0, 100.0, 100.0, start);
-            stats(100.0, 100.0, start),
+        let anomaly = update_reflector_state(
-            stats(100.0, 100.0, start),
+            &mut state,
-            200.0,
+            sample(200.0, 300.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(state.baseline.down, 110.0);
-        assert_close(baseline.up_ewma, 120.0);
+        assert_close(state.baseline.up, 120.0);
-        assert_close(recent.down_ewma, 150.0);
+        assert_close(state.recent.down, 150.0);
-        assert_close(recent.up_ewma, 200.0);
+        assert_close(state.recent.up, 200.0);
-        assert_eq!(baseline.last_receive_time_s, receive_time);
+        assert_eq!(state.last_receive_at, receive_time);
-        assert_eq!(recent.last_receive_time_s, receive_time);
     }
 
     #[test]
     fn keeps_baseline_at_or_below_recent() {
         let start = Instant::now();
-        let (baseline, recent, anomaly) = update_reflector_stats(
+        let mut state = state(200.0, 200.0, 100.0, 100.0, start);
-            stats(200.0, 200.0, start),
+        let anomaly = update_reflector_state(
-            stats(100.0, 100.0, start),
+            &mut state,
-            100.0,
+            sample(100.0, 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(state.baseline.down, 100.0);
-        assert_close(baseline.up_ewma, 100.0);
+        assert_close(state.baseline.up, 100.0);
-        assert_close(recent.down_ewma, 100.0);
+        assert_close(state.recent.down, 100.0);
-        assert_close(recent.up_ewma, 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(
+        let mut state = state(100.0, 200.0, 300.0, 400.0, start);
-            stats(100.0, 200.0, start),
+        let anomaly = update_reflector_state(
-            stats(300.0, 400.0, start),
+            &mut state,
-            500.0,
+            sample(500.0, 600.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(state.baseline.down, 500.0);
-        assert_close(baseline.up_ewma, 600.0);
+        assert_close(state.baseline.up, 600.0);
-        assert_close(recent.down_ewma, 500.0);
+        assert_close(state.recent.down, 500.0);
-        assert_close(recent.up_ewma, 600.0);
+        assert_close(state.recent.up, 600.0);
-        assert_eq!(baseline.last_receive_time_s, receive_time);
+        assert_eq!(state.last_receive_at, receive_time);
-        assert_eq!(recent.last_receive_time_s, 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(
+        let mut state = state(100.0, 200.0, 300.0, 400.0, start);
-            stats(100.0, 200.0, start),
+        let anomaly = update_reflector_state(
-            stats(300.0, 400.0, start),
+            &mut state,
-            5_101.0,
+            sample(5_101.0, 200.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(state.baseline.down, 100.0);
-        assert_close(baseline.up_ewma, 200.0);
+        assert_close(state.baseline.up, 200.0);
-        assert_close(recent.down_ewma, 300.0);
+        assert_close(state.recent.down, 300.0);
-        assert_close(recent.up_ewma, 400.0);
+        assert_close(state.recent.up, 400.0);
-        assert_eq!(baseline.last_receive_time_s, start);
+        assert_eq!(state.last_receive_at, start);
-        assert_eq!(recent.last_receive_time_s, 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(
+        let mut state = state(100.0, 200.0, 100.0, 200.0, start);
-            stats(100.0, 200.0, start),
+        let anomaly = update_reflector_state(
-            stats(100.0, 200.0, start),
+            &mut state,
-            5_100.0,
+            sample(5_100.0, 5_200.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!(state.last_receive_at, receive_time);
-        assert_eq!(recent.last_receive_time_s, 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 received_at = start + Duration::from_secs(1);
-        let receive_time = start + Duration::from_secs(31);
+        let reflector = "192.0.2.1".parse().unwrap();
-        let (baseline, recent, anomaly) = update_reflector_stats(
+        let mut reflectors = HashMap::new();
-            stats(100.0, 200.0, start),
+        reflectors.insert(reflector, state(100.0, 100.0, 100.0, 100.0, start));
-            stats(300.0, 400.0, recent_time),
+
-            500.0,
+        let (updated, anomaly) = process_reply(
-            600.0,
+            &mut reflectors,
-            receive_time,
+            &reply(reflector, 200.0, 300.0, received_at),
             start,
             0.9,
             0.5,
         );
 
         assert!(!anomaly);
-        assert_close(baseline.down_ewma, 500.0);
+        assert_close(updated.baseline.down, 110.0);
-        assert_close(baseline.up_ewma, 600.0);
+        assert_close(updated.baseline.up, 120.0);
-        assert_close(recent.down_ewma, 500.0);
+        assert_close(updated.recent.down, 150.0);
-        assert_close(recent.up_ewma, 600.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);
     }
 }

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::<IpAddr, ReflectorStats>::new()));
-    let owd_recent = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::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::<anyhow::Error>();
     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,
+        control_snapshot_rx,
-        owd_recent,
         reflector_peers,
         reselect_tx,
         dl_direction,

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::util::{ArcRwLock, RwLockExt};
-use crate::{Config, ReflectorStats};
+use flume::{Receiver, Sender};
-use flume::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<H> State<H> {
     }
 }
 
+fn drain_latest_snapshot(
+    snapshot_rx: &Receiver<ControlSnapshot>,
+    latest: &mut Option<ControlSnapshot>,
+) {
+    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<f64>, Vec<f64>) {
+    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<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     config: Config,
+    control_snapshot: Option<ControlSnapshot>,
+    control_snapshot_rx: Receiver<ControlSnapshot>,
     down_direction: StatsDirection,
-    owd_baseline: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
-    owd_recent: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
     reflectors_lock: ArcRwLock<Vec<IpAddr>>,
     reselect_trigger: Sender<bool>,
     state_dl: State<T::Handle>,
@@ -123,8 +174,7 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
 impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     fn new_with_backends(
         config: Config,
-        owd_baseline: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
+        control_snapshot_rx: Receiver<ControlSnapshot>,
-        owd_recent: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
         reflectors_lock: ArcRwLock<Vec<IpAddr>>,
         reselect_trigger: Sender<bool>,
         down_direction: StatsDirection,
@@ -150,9 +200,9 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 
         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<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     }
 
     fn update_deltas(&mut self) -> anyhow::Result<()> {
-        let state_dl = &mut self.state_dl;
+        drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
-        let state_ul = &mut self.state_ul;
-
-        state_dl.deltas.clear();
-        state_ul.deltas.clear();
-
         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() {
+        if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 {
-            // 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 {
             // trigger reselection
             warn!(
                 "Not enough delta values (D: {}, U: {}, need 5), triggering reselection",
-                state_dl.deltas.len(),
+                self.state_dl.deltas.len(),
-                state_ul.deltas.len()
+                self.state_ul.deltas.len()
             );
             let _ = self.reselect_trigger.try_send(true);
         }
@@ -318,8 +340,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
     pub fn new(
         config: Config,
-        owd_baseline: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
+        control_snapshot_rx: Receiver<ControlSnapshot>,
-        owd_recent: ArcMutex<HashMap<IpAddr, ReflectorStats>>,
         reflectors_lock: ArcRwLock<Vec<IpAddr>>,
         reselect_trigger: Sender<bool>,
         down_direction: StatsDirection,
@@ -328,8 +349,7 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
     ) -> anyhow::Result<Self> {
         Self::new_with_backends(
             config,
-            owd_baseline,
+            control_snapshot_rx,
-            owd_recent,
             reflectors_lock,
             reselect_trigger,
             down_direction,
@@ -530,3 +550,88 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         }
     }
 }
+
+#[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]);
+    }
+}

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::util::RwLockExt;
-use crate::{Config, ReflectorStats};
+use flume::{Receiver, RecvError, RecvTimeoutError, Selector};
-use flume::Receiver;
 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<bool, RecvError>),
+    Snapshot(Result<ControlSnapshot, RecvError>),
+}
+
+#[derive(Debug, Eq, PartialEq)]
+enum WaitOutcome {
+    Triggered,
+    DeadlineReached,
+    ChannelsClosed,
+}
+
+fn wait_for_trigger_or_deadline(
+    trigger_rx: &Receiver<bool>,
+    snapshot_rx: &Receiver<ControlSnapshot>,
+    latest_snapshot: &mut Option<ControlSnapshot>,
+    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<ControlSnapshot>,
+    latest_snapshot: &mut Option<ControlSnapshot>,
+    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<u64> {
+    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<HashMap<IpAddr, ReflectorStats>>,
+    pub snapshot_rx: Receiver<ControlSnapshot>,
     pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
     pub reflector_pool: Vec<IpAddr>,
     pub trigger_channel: Receiver<bool>,
@@ -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
+            let triggered = match wait_for_trigger_or_deadline(
-                .trigger_channel
+                &self.trigger_channel,
-                .recv_timeout(selector_sleep_time)
+                &self.snapshot_rx,
-                .is_ok();
+                &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) {
+                if let Some(rtt) = recent_rtt_for_peer(latest_snapshot.as_ref(), &peer) {
-                    let rtt = (owd_recent[&peer].down_ewma + owd_recent[&peer].up_ewma) as u64;
                     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)
+        );
+    }
+}

src/util.rs

@@ -5,7 +5,6 @@
 use anyhow::anyhow;
 use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard};
 
-pub type ArcMutex<T> = Arc<Mutex<T>>;
 pub type ArcRwLock<T> = Arc<RwLock<T>>;
 
 pub trait MutexExt<T> {