From 671e3f2671ebaf4a17d4c09cc8bce00baf08925f Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 13 Apr 2026 15:07:00 +0700 Subject: [PATCH] refactor: standardize channel naming with tx and rx suffixes Rename internal channel senders and receivers to follow standard Rust conventions for improved code consistency and readability. --- src/baseliner.rs | 10 +++++----- src/main.rs | 40 +++++++++++++++++++-------------------- src/metrics.rs | 6 +++--- src/pinger.rs | 8 ++++---- src/ratecontroller.rs | 6 +++--- src/reflector_selector.rs | 2 +- 6 files changed, 36 insertions(+), 36 deletions(-) diff --git a/src/baseliner.rs b/src/baseliner.rs index 706bc31..df19941 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -24,8 +24,8 @@ pub struct Baseliner { pub owd_recent: Arc>>, pub reselect_trigger: Sender, pub start_time: Instant, - pub stats_receiver: Receiver, - pub metrics_sender: SyncSender, + pub stats_rx: Receiver, + pub metrics_tx: SyncSender, } fn ewma_factor(tick: f64, dur: f64) -> f64 { @@ -49,7 +49,7 @@ impl Baseliner { let clock = Time::new(ClockId::Realtime); loop { - let time_data = self.stats_receiver.recv()?; + 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()?; @@ -107,7 +107,7 @@ impl Baseliner { "Reflector {} has OWD > 5 seconds more than baseline, triggering reselection", time_data.reflector ); - let _ = self.metrics_sender.try_send(Metric::Event { + let _ = self.metrics_tx.try_send(Metric::Event { name: "reselection", reason: "anomaly", reflector: Some(time_data.reflector), @@ -136,7 +136,7 @@ impl Baseliner { } } - let _ = self.metrics_sender.try_send(Metric::Baseline { + let _ = self.metrics_tx.try_send(Metric::Baseline { reflector: time_data.reflector, baseline_up_ewma: owd_baseline.up_ewma, baseline_down_ewma: owd_baseline.down_ewma, diff --git a/src/main.rs b/src/main.rs index 4733271..ab3a837 100644 --- a/src/main.rs +++ b/src/main.rs @@ -79,24 +79,24 @@ fn main() -> anyhow::Result<()> { } } - let (baseliner_stats_sender, baseliner_stats_receiver) = channel(); - let (reselect_sender, reselect_receiver) = channel(); + let (baseliner_stats_tx, baseliner_stats_rx) = channel(); + let (reselect_tx, reselect_rx) = channel(); - let (metrics_sender, metrics_receiver) = sync_channel(1000); + let (metrics_tx, metrics_rx) = sync_channel(1000); let ping_tx = config .observability_export_ping_metrics - .then(|| metrics_sender.clone()); + .then(|| metrics_tx.clone()); let baseline_tx = config .observability_export_baseline_metrics - .then(|| metrics_sender.clone()); + .then(|| metrics_tx.clone()); let event_tx = config .observability_export_events - .then(|| metrics_sender.clone()); + .then(|| metrics_tx.clone()); - let (mut pinger_receiver, mut pinger_sender) = match config.measurement_type { + let (mut ping_listener, mut ping_sender) = match config.measurement_type { MeasurementType::Icmp => ( Box::new(PingerICMPEchoListener {}) as Box, Box::new(PingerICMPEchoSender {}) as Box, @@ -114,10 +114,10 @@ fn main() -> anyhow::Result<()> { config: config.clone(), owd_baseline: owd_baseline.clone(), owd_recent: owd_recent.clone(), - reselect_trigger: reselect_sender.clone(), + reselect_trigger: reselect_tx.clone(), start_time: start_t, - stats_receiver: baseliner_stats_receiver, - metrics_sender: metrics_sender.clone(), + stats_rx: baseliner_stats_rx, + metrics_tx: metrics_tx.clone(), }; let down_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; @@ -146,17 +146,17 @@ fn main() -> anyhow::Result<()> { let (error_tx, error_rx) = channel::(); let err_tx = error_tx.clone(); - let metrics_sender_clone = metrics_sender.clone(); + let metrics_tx_clone = metrics_tx.clone(); let reflector_peers_lock_clone = reflector_peers_lock.clone(); thread::Builder::new() .name("receiver".to_string()) .spawn(move || { - if let Err(e) = pinger_receiver.listen( + if let Err(e) = ping_listener.listen( id, config.measurement_type, reflector_peers_lock_clone, - baseliner_stats_sender, - metrics_sender_clone, + baseliner_stats_tx, + metrics_tx_clone, ) { let _ = err_tx.send(e); } @@ -176,7 +176,7 @@ fn main() -> anyhow::Result<()> { thread::Builder::new() .name("sender".to_string()) .spawn(move || { - if let Err(e) = pinger_sender.send( + if let Err(e) = ping_sender.send( id, config.measurement_type, reflector_peers_lock_clone, @@ -192,8 +192,8 @@ fn main() -> anyhow::Result<()> { owd_recent: owd_recent.clone(), reflector_peers_lock: reflector_peers_lock.clone(), reflector_pool, - trigger_channel: reselect_receiver, - metrics_sender: metrics_sender.clone(), + trigger_channel: reselect_rx, + metrics_tx: metrics_tx.clone(), }; let err_tx = error_tx.clone(); thread::Builder::new() @@ -228,10 +228,10 @@ fn main() -> anyhow::Result<()> { owd_baseline, owd_recent, reflector_peers_lock, - reselect_sender, + reselect_tx, dl_direction, ul_direction, - metrics_sender.clone(), + metrics_tx.clone(), )?; debug!( @@ -255,7 +255,7 @@ fn main() -> anyhow::Result<()> { let metrics = Metrics { config, - metrics_receiver, + metrics_rx, }; let err_tx = error_tx.clone(); diff --git a/src/metrics.rs b/src/metrics.rs index aa3e782..2e0f34e 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -120,7 +120,7 @@ pub enum Metric { pub struct Metrics { pub config: Config, - pub metrics_receiver: Receiver, + pub metrics_rx: Receiver, } impl Metrics { @@ -153,7 +153,7 @@ impl Metrics { loop { // block until first metric or timeout - match self.metrics_receiver.recv_timeout(timeout) { + match self.metrics_rx.recv_timeout(timeout) { Ok(metric) => batch.push(metric), Err(RecvTimeoutError::Timeout) => { if !batch.is_empty() { @@ -167,7 +167,7 @@ impl Metrics { // greedily drain without blocking loop { - match self.metrics_receiver.try_recv() { + match self.metrics_rx.try_recv() { Ok(metric) => batch.push(metric), Err(_) => break, } diff --git a/src/pinger.rs b/src/pinger.rs index 421725f..8f8409e 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -54,8 +54,8 @@ pub trait PingListener { id: u16, type_: MeasurementType, reflectors_lock: Arc>>, - stats_sender: Sender, - metrics_sender: SyncSender, + stats_tx: Sender, + metrics_tx: SyncSender, ) -> anyhow::Result<()> { let socket = &mut open_socket(type_)?; @@ -84,7 +84,7 @@ pub trait PingListener { } }; - let _ = metrics_sender.try_send(Metric::Ping { + let _ = metrics_tx.try_send(Metric::Ping { reflector: reply.reflector, measurement_type: reply.measurement_type, rtt: reply.rtt, @@ -106,7 +106,7 @@ pub trait PingListener { reply.up_time, reply.down_time ); - if let Err(e) = stats_sender.send(reply) { + if let Err(e) = stats_tx.send(reply) { warn!("Stats channel closed, stopping listener: {}", e); break Ok(()); } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index e03908c..fed89eb 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -109,7 +109,7 @@ pub struct Ratecontroller { state_dl: State, state_ul: State, up_direction: StatsDirection, - metrics_sender: SyncSender, + metrics_tx: SyncSender, } impl Ratecontroller { @@ -248,7 +248,7 @@ impl Ratecontroller { reselect_trigger: Sender, down_direction: StatsDirection, up_direction: StatsDirection, - metrics_sender: SyncSender, + metrics_tx: SyncSender, ) -> anyhow::Result { let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; let dl_safe_rates = @@ -269,7 +269,7 @@ impl Ratecontroller { state_dl: State::new(dl_qdisc, cur_rx, dl_safe_rates), state_ul: State::new(ul_qdisc, cur_tx, ul_safe_rates), up_direction, - metrics_sender, + metrics_tx, }) } diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index 9310cfc..dba1ba0 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -15,7 +15,7 @@ pub struct ReflectorSelector { pub reflector_peers_lock: Arc>>, pub reflector_pool: Vec, pub trigger_channel: Receiver, - pub metrics_sender: SyncSender, + pub metrics_tx: SyncSender, } impl ReflectorSelector {