diff --git a/src/controller.rs b/src/controller.rs index ea3b027..521e729 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -5,9 +5,11 @@ use std::{ time::Duration, }; -use flume::{SendError, Sender}; use thiserror::Error; -use tokio::sync::mpsc::UnboundedReceiver; +use tokio::sync::{ + mpsc::UnboundedReceiver, + watch::{self, error::SendError}, +}; #[cfg(feature = "irtt")] use crate::measurement::irtt::IrttBinding; @@ -30,9 +32,9 @@ pub enum ControllerError { pub struct Controller { reflector_rx: UnboundedReceiver>, - icmp_binding_tx: Sender>, + icmp_binding_tx: watch::Sender>, #[cfg(feature = "irtt")] - irtt_binding_tx: Sender>, + irtt_binding_tx: watch::Sender>, reflectors: HashMap, active_streams: Arc>>, endpoint_streams: Arc>>, @@ -45,8 +47,8 @@ pub struct Controller { impl Controller { pub fn new( reflector_rx: UnboundedReceiver>, - icmp_binding_tx: Sender>, - #[cfg(feature = "irtt")] irtt_binding_tx: Sender>, + icmp_binding_tx: watch::Sender>, + #[cfg(feature = "irtt")] irtt_binding_tx: watch::Sender>, reflectors: HashMap, active_streams: Arc>>, endpoint_streams: Arc>>, @@ -137,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(()) } } diff --git a/src/main.rs b/src/main.rs index 50b0c70..12e4647 100644 --- a/src/main.rs +++ b/src/main.rs @@ -348,9 +348,9 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let (control_snapshot_tx, control_snapshot_rx) = watch::channel(None); let (error_tx, mut error_rx) = tokio::sync::mpsc::unbounded_channel::(); - let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded(); + let (icmp_binding_tx, icmp_binding_rx) = watch::channel(Vec::new()); #[cfg(feature = "irtt")] - let (irtt_binding_tx, irtt_binding_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); diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs index 6a60562..fa980b8 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -5,7 +5,6 @@ use std::{ time::{Duration, Instant, SystemTime}, }; -use flume::Receiver; use icmp_socket2::{ AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest}, @@ -13,7 +12,7 @@ use icmp_socket2::{ }; use log::debug; use thiserror::Error; -use tokio::sync::mpsc::UnboundedSender; +use tokio::sync::{mpsc::UnboundedSender, watch}; use crate::{ measurement::model::{ @@ -25,6 +24,7 @@ use crate::{ time::{ClockSample, Realtime}, }; +#[derive(Clone)] pub struct IcmpBinding { pub stream: MeasurementStream, pub schedule: ProbeSchedule, @@ -104,7 +104,7 @@ pub struct IcmpEngine { origin: Instant, - binding_rx: Receiver>, + binding_rx: watch::Receiver>, observation_tx: UnboundedSender, } @@ -125,7 +125,7 @@ impl IcmpEngine { origin: Instant, #[allow(unused)] settings: &NetworkSettings, observation_tx: UnboundedSender, - binding_rx: Receiver>, + binding_rx: watch::Receiver>, ) -> Result { #[allow(unused_mut)] let mut socket = IcmpSocket4::new()?; @@ -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)?; } diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs index ff87f6a..51957e7 100644 --- a/src/measurement/irtt.rs +++ b/src/measurement/irtt.rs @@ -1,6 +1,5 @@ use std::{collections::HashMap, net::SocketAddr}; -use flume::Receiver; 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, mpsc::UnboundedSender}; +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,7 +69,7 @@ pub struct IrttEngine { bindings: HashMap, targets: Vec, - binding_rx: Receiver>, + binding_rx: watch::Receiver>, observation_tx: UnboundedSender, } @@ -78,7 +78,7 @@ impl IrttEngine { interval: std::time::Duration, settings: &NetworkSettings, observation_tx: UnboundedSender, - binding_rx: Receiver>, + binding_rx: watch::Receiver>, ) -> 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?; }