refactor: replace std::sync::mpsc with flume for improved channel performance

Migrate internal message passing to flume to leverage better performance and additional features like eventual fairness and selection.

Cargo.lock

@@ -43,6 +43,12 @@ version = "2.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3"
 
+[[package]]
+name = "bumpalo"
+version = "3.20.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb"
+
 [[package]]
 name = "cc"
 version = "1.2.60"
@@ -109,6 +115,9 @@ name = "fastrand"
 version = "2.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6"
+dependencies = [
+ "getrandom",
+]
 
 [[package]]
 name = "find-msvc-tools"
@@ -116,6 +125,30 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
 
+[[package]]
+name = "flume"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
+dependencies = [
+ "fastrand",
+ "spin",
+]
+
+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "js-sys",
+ "libc",
+ "r-efi",
+ "wasip2",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "glob"
 version = "0.3.3"
@@ -140,6 +173,16 @@ dependencies = [
  "either",
 ]
 
+[[package]]
+name = "js-sys"
+version = "0.3.95"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2964e92d1d9dc3364cae4d718d93f227e3abb088e747d92e0395bfdedf1c12ca"
+dependencies = [
+ "once_cell",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "libc"
 version = "0.2.185"
@@ -171,6 +214,15 @@ version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"
 
+[[package]]
+name = "lock_api"
+version = "0.4.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965"
+dependencies = [
+ "scopeguard",
+]
+
 [[package]]
 name = "log"
 version = "0.4.29"
@@ -225,6 +277,12 @@ dependencies = [
  "minimal-lexical",
 ]
 
+[[package]]
+name = "once_cell"
+version = "1.21.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
+
 [[package]]
 name = "prettyplease"
 version = "0.2.37"
@@ -253,6 +311,12 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
 [[package]]
 name = "regex"
 version = "1.12.3"
@@ -312,6 +376,18 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "rustversion"
+version = "1.0.22"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
+
+[[package]]
+name = "scopeguard"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
+
 [[package]]
 name = "shlex"
 version = "1.3.0"
@@ -328,12 +404,22 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "spin"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
+dependencies = [
+ "lock_api",
+]
+
 [[package]]
 name = "sqm-autorate-rust"
 version = "0.4.1"
 dependencies = [
  "anyhow",
  "fastrand",
+ "flume",
  "icmp-socket2",
  "libc",
  "log",
@@ -381,6 +467,60 @@ version = "1.0.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
 
+[[package]]
+name = "wasip2"
+version = "1.0.3+wasi-0.2.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6"
+dependencies = [
+ "wit-bindgen",
+]
+
+[[package]]
+name = "wasm-bindgen"
+version = "0.2.118"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0bf938a0bacb0469e83c1e148908bd7d5a6010354cf4fb73279b7447422e3a89"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "rustversion",
+ "wasm-bindgen-macro",
+ "wasm-bindgen-shared",
+]
+
+[[package]]
+name = "wasm-bindgen-macro"
+version = "0.2.118"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eeff24f84126c0ec2db7a449f0c2ec963c6a49efe0698c4242929da037ca28ed"
+dependencies = [
+ "quote",
+ "wasm-bindgen-macro-support",
+]
+
+[[package]]
+name = "wasm-bindgen-macro-support"
+version = "0.2.118"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d08065faf983b2b80a79fd87d8254c409281cf7de75fc4b773019824196c904"
+dependencies = [
+ "bumpalo",
+ "proc-macro2",
+ "quote",
+ "syn",
+ "wasm-bindgen-shared",
+]
+
+[[package]]
+name = "wasm-bindgen-shared"
+version = "0.2.118"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5fd04d9e306f1907bd13c6361b5c6bfc7b3b3c095ed3f8a9246390f8dbdee129"
+dependencies = [
+ "unicode-ident",
+]
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"
@@ -395,3 +535,9 @@ checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
 dependencies = [
  "windows-link",
 ]
+
+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"

Cargo.toml

@@ -28,6 +28,7 @@ rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 rust-uci = { git = "https://github.com/namib-project/rust-uci.git", rev = "71ea7864ae5d266100043bfd2f682d4302a5b3fc", optional = true }
 thiserror = "2.0"
 libc = "0.2"
+flume = { version = "0.12.0",  default-features = false, features = ["eventual-fairness", "select"] }
 
 [dependencies.log]
 features = ["std"]

src/baseliner.rs

@@ -4,11 +4,11 @@ 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::sync::mpsc::{Receiver, Sender};
 use std::time::Instant;
 
 #[derive(Copy, Clone)]

src/main.rs

@@ -16,11 +16,11 @@ mod util;
 use crate::baseliner::{Baseliner, ReflectorStats};
 use crate::metrics::{Metric, Metrics, MetricsSender};
 use ::log::{debug, info};
+use flume::RecvTimeoutError;
 use std::collections::HashMap;
 use std::net::IpAddr;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
-use std::sync::mpsc::{RecvTimeoutError, channel, sync_channel};
 use std::sync::{Arc, Mutex, RwLock};
 use std::thread::sleep;
 use std::time::Duration;
@@ -98,14 +98,14 @@ fn main() -> anyhow::Result<()> {
         }
     }
 
-    let (baseliner_stats_tx, baseliner_stats_rx) = channel();
+    let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded();
-    let (error_tx, error_rx) = channel::<anyhow::Error>();
+    let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
-    let (reselect_tx, reselect_rx) = channel();
+    let (reselect_tx, reselect_rx) = flume::unbounded();
 
     let dropped = Arc::new(AtomicU32::new(0));
 
     let (metrics_tx, metrics_thread_handle) = if config.observability_enabled {
-        let (tx, rx) = sync_channel(1000);
+        let (tx, rx) = flume::bounded(1000);
         let metrics = Metrics {
             config: config.clone(),
             metrics_rx: rx,

src/metrics.rs

@@ -1,13 +1,13 @@
 use crate::Config;
 use crate::config::{MeasurementType, ObservabilityProtocol};
 use crate::time::Time;
+use flume::{Receiver, RecvTimeoutError, Sender};
 use log::{error, info, warn};
 use rustix::time::ClockId;
 use std::fmt::Write;
 use std::net::{IpAddr, TcpStream, UdpSocket};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, Ordering};
-use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender};
 use std::time::Duration;
 
 const MAX_RECONNECT_BACKOFF: u64 = 60;
@@ -122,12 +122,12 @@ pub enum Metric {
 
 #[derive(Clone)]
 pub struct MetricsSender {
-    tx: Option<SyncSender<(Metric, u64)>>,
+    tx: Option<Sender<(Metric, u64)>>,
     dropped: Arc<AtomicU32>,
 }
 
 impl MetricsSender {
-    pub fn new(tx: SyncSender<(Metric, u64)>, dropped: Arc<AtomicU32>) -> Self {
+    pub fn new(tx: Sender<(Metric, u64)>, dropped: Arc<AtomicU32>) -> Self {
         Self {
             tx: Some(tx),
             dropped,

src/pinger.rs

@@ -2,12 +2,12 @@ use crate::MeasurementType;
 use crate::SHUTDOWN;
 use crate::metrics::{Metric, MetricsSender};
 use crate::util::RwLockExt;
+use flume::Sender;
 use icmp_socket2::socket::IcmpSocket;
 use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
 use log::{debug, info, warn};
 use std::net::{IpAddr, Ipv4Addr};
 use std::sync::atomic::Ordering;
-use std::sync::mpsc::Sender;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 use std::{io, thread};

src/ratecontroller.rs

@@ -4,6 +4,7 @@ use crate::netlink::{Netlink, NetlinkError, Qdisc};
 use crate::time::Time;
 use crate::util::{ArcMutex, ArcRwLock, MutexExt, RwLockExt};
 use crate::{Config, ReflectorStats};
+use flume::Sender;
 use log::{debug, info, warn};
 use rustix::thread::ClockId;
 use std::collections::HashMap;
@@ -11,7 +12,6 @@ use std::fs::File;
 use std::io::Write;
 use std::net::IpAddr;
 use std::sync::atomic::Ordering;
-use std::sync::mpsc::Sender;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
 use thiserror::Error;

src/reflector_selector.rs

@@ -2,11 +2,11 @@ use crate::SHUTDOWN;
 use crate::metrics::{Metric, MetricsSender};
 use crate::util::{ArcMutex, MutexExt, RwLockExt};
 use crate::{Config, ReflectorStats};
+use flume::Receiver;
 use log::{debug, info};
 use std::collections::HashMap;
 use std::net::IpAddr;
 use std::sync::atomic::Ordering;
-use std::sync::mpsc::Receiver;
 use std::sync::{Arc, RwLock};
 use std::thread::sleep;
 use std::time::Duration;