refactor: use tokio watch channel for bindings

src/baseliner.rs

@@ -9,12 +9,14 @@ use crate::SHUTDOWN;
 use crate::measurement::model::{MeasurementEvent, MeasurementObservation, MeasurementStream};
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
-use flume::{Receiver, Sender};
+use anyhow::Context;
+use flume::Sender;
 use log::{debug, info};
 use std::collections::HashMap;
 use std::sync::atomic::Ordering;
 use std::time::Duration;
 use std::time::Instant;
+use tokio::sync::mpsc::UnboundedReceiver;
 
 const STALE_AFTER: Duration = Duration::from_secs(30);
 const ANOMALY_THRESHOLD_MS: f64 = 5_000.0;
@@ -70,7 +72,6 @@ impl ReflectorState {
 
 #[derive(Clone, Debug)]
 pub struct ControlSnapshot {
-    pub generated_at: Instant,
     pub reflectors: HashMap<MeasurementStream, ReflectorState>,
 }
 
@@ -78,8 +79,8 @@ pub struct Baseliner {
     pub settings: Settings,
     pub reselect_trigger: Sender<bool>,
     pub start_time: Instant,
-    pub stats_rx: Receiver<MeasurementEvent>,
-    pub control_tx: Sender<ControlSnapshot>,
+    pub stats_rx: UnboundedReceiver<MeasurementEvent>,
+    pub control_tx: tokio::sync::watch::Sender<Option<ControlSnapshot>>,
     pub selection_tx: Option<Sender<ControlSnapshot>>,
     pub baseline_metrics: MetricsSender,
     pub event_metrics: MetricsSender,
@@ -142,18 +143,14 @@ fn process_reply(
     (*state, anomaly)
 }
 
-fn control_snapshot(
-    reflectors: &HashMap<MeasurementStream, ReflectorState>,
-    generated_at: Instant,
-) -> ControlSnapshot {
+fn control_snapshot(reflectors: &HashMap<MeasurementStream, ReflectorState>) -> ControlSnapshot {
     ControlSnapshot {
-        generated_at,
         reflectors: reflectors.clone(),
     }
 }
 
 impl Baseliner {
-    pub fn run(&self) -> anyhow::Result<()> {
+    pub fn run(&mut self) -> anyhow::Result<()> {
         let mut reflectors = HashMap::<MeasurementStream, ReflectorState>::new();
 
         let slow_factor = ewma_factor(
@@ -171,7 +168,10 @@ impl Baseliner {
                 return Ok(());
             }
 
-            let event = self.stats_rx.recv()?;
+            let event = self
+                .stats_rx
+                .blocking_recv()
+                .context("measurement channel closed")?;
 
             let reply = match event {
                 MeasurementEvent::Observation(reply) => reply,
@@ -232,11 +232,11 @@ impl Baseliner {
                 reflector, state.recent.up, state.recent.down
             );
 
-            let snapshot = control_snapshot(&reflectors, Instant::now());
+            let snapshot = control_snapshot(&reflectors);
             if let Some(selection_tx) = &self.selection_tx {
                 let _ = selection_tx.send(snapshot.clone());
             }
-            self.control_tx.send(snapshot)?;
+            self.control_tx.send(Some(snapshot))?;
         }
     }
 }
@@ -454,32 +454,4 @@ mod tests {
         assert_eq!(updated.last_receive_at, received_at);
         assert_close(reflectors[&stream].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);
-        let stream = reply(reflector, 0.0, 0.0, received_at).stream;
-
-        assert!(!anomaly);
-        assert_eq!(snapshot.generated_at, generated_at);
-        assert_close(
-            snapshot.reflectors[&stream].baseline.down,
-            updated.baseline.down,
-        );
-        assert_close(snapshot.reflectors[&stream].recent.up, updated.recent.up);
-        assert_eq!(snapshot.reflectors[&stream].last_receive_at, received_at);
-    }
 }

src/controller.rs

@@ -5,8 +5,11 @@ use std::{
     time::Duration,
 };
 
-use flume::{Receiver, SendError, Sender};
 use thiserror::Error;
+use tokio::sync::{
+    mpsc::UnboundedReceiver,
+    watch::{self, error::SendError},
+};
 
 #[cfg(feature = "irtt")]
 use crate::measurement::irtt::IrttBinding;
@@ -28,10 +31,10 @@ pub enum ControllerError {
 }
 
 pub struct Controller {
-    reflector_rx: Receiver<Vec<SocketAddr>>,
-    icmp_binding_tx: Sender<Vec<IcmpBinding>>,
+    reflector_rx: UnboundedReceiver<Vec<SocketAddr>>,
+    icmp_binding_tx: watch::Sender<Vec<IcmpBinding>>,
     #[cfg(feature = "irtt")]
-    irtt_binding_tx: Sender<Vec<IrttBinding>>,
+    irtt_binding_tx: watch::Sender<Vec<IrttBinding>>,
     reflectors: HashMap<SocketAddr, ReflectorDefinition>,
     active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
     endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
@@ -43,9 +46,9 @@ pub struct Controller {
 
 impl Controller {
     pub fn new(
-        reflector_rx: Receiver<Vec<SocketAddr>>,
-        icmp_binding_tx: Sender<Vec<IcmpBinding>>,
-        #[cfg(feature = "irtt")] irtt_binding_tx: Sender<Vec<IrttBinding>>,
+        reflector_rx: UnboundedReceiver<Vec<SocketAddr>>,
+        icmp_binding_tx: watch::Sender<Vec<IcmpBinding>>,
+        #[cfg(feature = "irtt")] irtt_binding_tx: watch::Sender<Vec<IrttBinding>>,
         reflectors: HashMap<SocketAddr, ReflectorDefinition>,
         active_streams: Arc<RwLock<Vec<MeasurementStream>>>,
         endpoint_streams: Arc<RwLock<HashMap<SocketAddr, MeasurementStream>>>,
@@ -69,7 +72,7 @@ impl Controller {
 
     pub async fn run(mut self, initial_reflectors: Vec<SocketAddr>) -> Result<(), ControllerError> {
         self.update_reflectors(initial_reflectors).await?;
-        while let Ok(reflectors) = self.reflector_rx.recv_async().await {
+        while let Some(reflectors) = self.reflector_rx.recv().await {
             self.update_reflectors(reflectors).await?;
         }
         Ok(())
@@ -136,9 +139,9 @@ impl Controller {
             .endpoint_streams
             .write()
             .expect("endpoint stream lock poisoned") = endpoint_streams;
-        self.icmp_binding_tx.send_async(icmp).await?;
+        self.icmp_binding_tx.send(icmp)?;
         #[cfg(feature = "irtt")]
-        self.irtt_binding_tx.send_async(irtt).await?;
+        self.irtt_binding_tx.send(irtt)?;
         Ok(())
     }
 }

src/main.rs

@@ -32,7 +32,6 @@ use crate::reflector_selector::ReflectorSelector;
 use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings};
 use crate::util::RwLockExt;
 use ::log::{info, warn};
-use flume::RecvTimeoutError;
 use std::collections::HashMap;
 use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
@@ -40,6 +39,8 @@ use std::sync::{Arc, RwLock};
 use std::thread;
 use std::thread::{JoinHandle, sleep};
 use std::time::{Duration, Instant};
+use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
+use tokio::sync::watch;
 
 struct ReflectorSetup {
     peers: Arc<RwLock<Vec<SocketAddr>>>,
@@ -266,7 +267,7 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
 
 fn spawn_task<F, E>(
     rt: &tokio::runtime::Runtime,
-    error_tx: &flume::Sender<anyhow::Error>,
+    error_tx: &UnboundedSender<anyhow::Error>,
     task: F,
 ) -> tokio::task::JoinHandle<()>
 where
@@ -284,7 +285,7 @@ where
 
 fn spawn_worker<F>(
     name: &str,
-    error_tx: &flume::Sender<anyhow::Error>,
+    error_tx: &UnboundedSender<anyhow::Error>,
     worker: F,
 ) -> anyhow::Result<JoinHandle<()>>
 where
@@ -302,16 +303,16 @@ where
         .map_err(|e| anyhow::anyhow!(e))
 }
 
-fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()> {
+async fn wait_for_exit(error_rx: &mut UnboundedReceiver<anyhow::Error>) -> anyhow::Result<()> {
     loop {
-        match error_rx.recv_timeout(Duration::from_secs(1)) {
-            Ok(error) => {
+        match tokio::time::timeout(Duration::from_secs(1), error_rx.recv()).await {
+            Ok(Some(error)) => {
                 return Err(anyhow::anyhow!("worker exited with error: {error}"));
             }
-            Err(RecvTimeoutError::Disconnected) => {
+            Ok(None) => {
                 return Ok(());
             }
-            Err(RecvTimeoutError::Timeout) => {
+            Err(_) => {
                 if SHUTDOWN.load(Ordering::Relaxed) {
                     info!("Received shutdown signal");
                     return Ok(());
@@ -345,13 +346,13 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         reselection_enabled,
     } = setup_reflectors(settings)?;
 
-    let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded();
-    let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
-    let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded();
+    let (control_snapshot_tx, control_snapshot_rx) = watch::channel(None);
+    let (error_tx, mut error_rx) = tokio::sync::mpsc::unbounded_channel::<anyhow::Error>();
+    let (icmp_binding_tx, icmp_binding_rx) = watch::channel(Vec::new());
     #[cfg(feature = "irtt")]
-    let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded();
-    let (measurement_tx, measurement_rx) = flume::unbounded();
-    let (reflector_tx, reflector_rx) = flume::unbounded();
+    let (irtt_binding_tx, irtt_binding_rx) = watch::channel(Vec::new());
+    let (measurement_tx, measurement_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (reflector_tx, reflector_rx) = tokio::sync::mpsc::unbounded_channel();
     let (reselect_tx, reselect_rx) = flume::bounded(1);
     let (selection_snapshot_tx, selection_snapshot_rx) = if reselection_enabled {
         let (tx, rx) = flume::unbounded();
@@ -439,7 +440,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         },
     });
 
-    let baseliner = Baseliner {
+    let mut baseliner = Baseliner {
         settings: settings.clone(),
         reselect_trigger: reselect_tx.clone(),
         start_time,
@@ -494,7 +495,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     // Drop the original sender so the channel disconnects if all workers exit cleanly.
     drop(error_tx);
 
-    let result = wait_for_exit(&error_rx);
+    let result = rt.block_on(wait_for_exit(&mut error_rx));
 
     // Make the error path request shutdown too, rather than leaving the other workers running.
     SHUTDOWN.store(true, Ordering::Relaxed);

src/measurement/icmp.rs

@@ -5,7 +5,6 @@ use std::{
     time::{Duration, Instant, SystemTime},
 };
 
-use flume::{Receiver, Sender};
 use icmp_socket2::{
     AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet,
     packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest},
@@ -13,6 +12,7 @@ use icmp_socket2::{
 };
 use log::debug;
 use thiserror::Error;
+use tokio::sync::{mpsc::UnboundedSender, watch};
 
 use crate::{
     measurement::model::{
@@ -24,6 +24,7 @@ use crate::{
     time::{ClockSample, Realtime},
 };
 
+#[derive(Clone)]
 pub struct IcmpBinding {
     pub stream: MeasurementStream,
     pub schedule: ProbeSchedule,
@@ -103,8 +104,8 @@ pub struct IcmpEngine {
 
     origin: Instant,
 
-    binding_rx: Receiver<Vec<IcmpBinding>>,
-    observation_tx: Sender<MeasurementEvent>,
+    binding_rx: watch::Receiver<Vec<IcmpBinding>>,
+    observation_tx: UnboundedSender<MeasurementEvent>,
 }
 
 async fn wait_for_deadline(deadline: Option<Instant>) {
@@ -123,8 +124,8 @@ impl IcmpEngine {
     pub fn new(
         origin: Instant,
         #[allow(unused)] settings: &NetworkSettings,
-        observation_tx: Sender<MeasurementEvent>,
-        binding_rx: Receiver<Vec<IcmpBinding>>,
+        observation_tx: UnboundedSender<MeasurementEvent>,
+        binding_rx: watch::Receiver<Vec<IcmpBinding>>,
     ) -> Result<Self, IcmpError> {
         #[allow(unused_mut)]
         let mut socket = IcmpSocket4::new()?;
@@ -420,12 +421,11 @@ impl IcmpEngine {
 
         for probe in expired {
             self.observation_tx
-                .send_async(MeasurementEvent::Loss(MeasurementLoss {
+                .send(MeasurementEvent::Loss(MeasurementLoss {
                     stream: probe.stream,
                     started_at: probe.sent_at,
                     detected_at: now,
                 }))
-                .await
                 .map_err(|_| IcmpError::EventChannelClosed)?;
         }
 
@@ -446,10 +446,10 @@ impl IcmpEngine {
             let deadline = self.next_deadline();
 
             tokio::select! {
-                bindings = self.binding_rx.recv_async() => {
-                    let bindings = bindings
-                        .map_err(|_| IcmpError::BindingChannelClosed)?;
+                bindings = self.binding_rx.changed() => {
+                    bindings.map_err(|_| IcmpError::BindingChannelClosed)?;
 
+                    let bindings = self.binding_rx.borrow_and_update().clone();
                     self.set_bindings(bindings)?;
                 }
 
@@ -468,8 +468,7 @@ impl IcmpEngine {
                     ) {
                         Ok(observation) => {
                             self.observation_tx
-                                .send_async(MeasurementEvent::Observation(observation))
-                                .await
+                                .send(MeasurementEvent::Observation(observation))
                                 .map_err(|_| IcmpError::EventChannelClosed)?;
                         }
 

src/measurement/irtt.rs

@@ -1,6 +1,5 @@
 use std::{collections::HashMap, net::SocketAddr};
 
-use flume::{Receiver, Sender};
 use irtt_client::{
     ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy,
     managed::{
@@ -12,7 +11,7 @@ use irtt_client::{
 };
 use log::{debug, trace, warn};
 use thiserror::Error;
-use tokio::sync::broadcast::error::RecvError;
+use tokio::sync::{broadcast::error::RecvError, mpsc::UnboundedSender, watch};
 
 use crate::SHUTDOWN;
 use crate::measurement::model::{
@@ -51,6 +50,7 @@ pub enum IrttError {
     ManagedAbandoned,
 }
 
+#[derive(Clone)]
 pub struct IrttBinding {
     stream: MeasurementStream,
     port: u16,
@@ -69,16 +69,16 @@ pub struct IrttEngine {
     bindings: HashMap<MeasurementStreamId, IrttBinding>,
     targets: Vec<ManagedTargetConfig>,
 
-    binding_rx: Receiver<Vec<IrttBinding>>,
-    observation_tx: Sender<MeasurementEvent>,
+    binding_rx: watch::Receiver<Vec<IrttBinding>>,
+    observation_tx: UnboundedSender<MeasurementEvent>,
 }
 
 impl IrttEngine {
     pub fn new(
         interval: std::time::Duration,
         settings: &NetworkSettings,
-        observation_tx: Sender<MeasurementEvent>,
-        binding_rx: Receiver<Vec<IrttBinding>>,
+        observation_tx: UnboundedSender<MeasurementEvent>,
+        binding_rx: watch::Receiver<Vec<IrttBinding>>,
     ) -> Result<(ManagedClientTask, Self), IrttError> {
         let config = managed_client_config(interval, settings);
         let (task, handle) = ManagedClient::task(config, Vec::new())?;
@@ -158,10 +158,10 @@ impl IrttEngine {
         retry.tick().await;
         loop {
             tokio::select! {
-                bindings = self.binding_rx.recv_async() => {
-                    let bindings = bindings
-                        .map_err(|_| IrttError::BindingChannelClosed)?;
+                bindings = self.binding_rx.changed() => {
+                    bindings.map_err(|_| IrttError::BindingChannelClosed)?;
 
+                    let bindings = self.binding_rx.borrow_and_update().clone();
                     self.set_bindings(bindings).await?;
                 }
 
@@ -273,8 +273,7 @@ impl IrttEngine {
                                     };
 
                                     self.observation_tx
-                                        .send_async(MeasurementEvent::Observation(observation))
-                                        .await
+                                        .send(MeasurementEvent::Observation(observation))
                                         .map_err(|_| IrttError::EventChannelClosed)?;
                                 }
 
@@ -294,8 +293,7 @@ impl IrttEngine {
                                     };
 
                                     self.observation_tx
-                                        .send_async(MeasurementEvent::Loss(loss))
-                                        .await
+                                        .send(MeasurementEvent::Loss(loss))
                                         .map_err(|_| IrttError::EventChannelClosed)?;
                                 }
 

src/ratecontroller.rs

@@ -16,7 +16,6 @@ use crate::platform::{
 use crate::settings::Settings;
 use crate::time::{ClockSample, Realtime};
 use crate::util::{ArcRwLock, RwLockExt};
-use flume::{Receiver, Sender};
 use log::{debug, info, warn};
 use std::fs::File;
 use std::io::Write;
@@ -84,20 +83,6 @@ 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: &[MeasurementStream],
@@ -137,10 +122,9 @@ fn deltas_from_snapshot(
 
 pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     settings: Settings,
-    control_snapshot: Option<ControlSnapshot>,
-    control_snapshot_rx: Receiver<ControlSnapshot>,
+    control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
     reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
-    reselect_trigger: Sender<bool>,
+    reselect_trigger: flume::Sender<bool>,
     state_dl: State<T::Handle>,
     state_ul: State<T::Handle>,
     rng: fastrand::Rng,
@@ -152,9 +136,9 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
 impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     fn new_with_backends(
         settings: Settings,
-        control_snapshot_rx: Receiver<ControlSnapshot>,
+        control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
         reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
-        reselect_trigger: Sender<bool>,
+        reselect_trigger: flume::Sender<bool>,
         metrics: MetricsSender,
         mut stats_provider: S,
         mut traffic_control: T,
@@ -179,7 +163,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 
         Ok(Self {
             settings,
-            control_snapshot: None,
             control_snapshot_rx,
             reflectors_lock,
             reselect_trigger,
@@ -479,15 +462,17 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
     }
 
     fn update_deltas(&mut self) -> anyhow::Result<()> {
-        drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
         let now_t = Instant::now();
         let reflectors = self.reflectors_lock.read_anyhow()?;
-        let (down_deltas, up_deltas) = deltas_from_snapshot(
-            self.control_snapshot.as_ref(),
-            &reflectors,
-            now_t,
-            self.settings.advanced_settings.tick_interval,
-        );
+        let (down_deltas, up_deltas) = {
+            let snapshot = self.control_snapshot_rx.borrow();
+            deltas_from_snapshot(
+                snapshot.as_ref(),
+                &reflectors,
+                now_t,
+                self.settings.advanced_settings.tick_interval,
+            )
+        };
         self.state_dl.deltas = down_deltas;
         self.state_ul.deltas = up_deltas;
 
@@ -513,9 +498,9 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
     pub fn new(
         settings: Settings,
-        control_snapshot_rx: Receiver<ControlSnapshot>,
+        control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
         reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
-        reselect_trigger: Sender<bool>,
+        reselect_trigger: flume::Sender<bool>,
         metrics: MetricsSender,
     ) -> anyhow::Result<Self> {
         Self::new_with_backends(
@@ -530,111 +515,3 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
         )
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::baseliner::{EwmaStats, ReflectorState};
-    use crate::measurement::model::{MeasurementSource, MeasurementStreamId};
-    use std::collections::HashMap;
-    use std::net::IpAddr;
-
-    fn stream(id: u64, peer: IpAddr) -> MeasurementStream {
-        MeasurementStream {
-            id: MeasurementStreamId(id),
-            peer,
-            source: MeasurementSource::IcmpTimestamp,
-        }
-    }
-
-    #[test]
-    fn fixed_seed_reproduces_initial_safe_rates() {
-        let mut first = fastrand::Rng::with_seed(42);
-        let mut second = fastrand::Rng::with_seed(42);
-
-        let first_rates = generate_initial_speeds(&mut first, 100_000.0, 100);
-        let second_rates = generate_initial_speeds(&mut second, 100_000.0, 100);
-
-        assert_eq!(first_rates, second_rates);
-    }
-
-    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 = stream(0, "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[&stream(0, "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 = stream(0, "192.0.2.1".parse().unwrap());
-        let second = stream(1, "192.0.2.2".parse().unwrap());
-        let stale = stream(2, "192.0.2.3".parse().unwrap());
-        let inactive = stream(3, "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

@@ -11,13 +11,14 @@ use crate::measurement::model::MeasurementStream;
 use crate::metrics::{Metric, MetricsSender};
 use crate::settings::Settings;
 use crate::util::RwLockExt;
-use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender};
+use flume::{Receiver, RecvError, RecvTimeoutError, Selector};
 use log::{debug, info};
 use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
 use std::sync::{Arc, RwLock};
 use std::thread::sleep;
 use std::time::{Duration, Instant};
+use tokio::sync::mpsc::UnboundedSender;
 
 enum SelectorEvent {
     Trigger(Result<bool, RecvError>),
@@ -115,7 +116,7 @@ pub struct ReflectorSelector {
     pub settings: Settings,
     pub snapshot_rx: Receiver<ControlSnapshot>,
     pub reflector_peers_lock: Arc<RwLock<Vec<SocketAddr>>>,
-    pub reflector_tx: Sender<Vec<SocketAddr>>,
+    pub reflector_tx: UnboundedSender<Vec<SocketAddr>>,
     pub reflector_pool: Vec<SocketAddr>,
     pub endpoint_streams: Arc<RwLock<std::collections::HashMap<SocketAddr, MeasurementStream>>>,
     pub trigger_channel: Receiver<bool>,
@@ -285,158 +286,3 @@ impl ReflectorSelector {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::baseliner::{EwmaStats, ReflectorState};
-    use crate::measurement::model::MeasurementSource;
-    use std::collections::HashMap;
-    use std::net::IpAddr;
-    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(
-            crate::measurement::model::MeasurementStream {
-                id: crate::measurement::model::MeasurementStreamId(0),
-                peer,
-                source: MeasurementSource::IcmpTimestamp,
-            },
-            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: IpAddr = "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(),
-                latest_snapshot.as_ref().unwrap().reflectors.keys().next(),
-                generated_at + Duration::from_secs(1),
-                Duration::from_secs(2)
-            ),
-            Some(42)
-        );
-    }
-}