diff --git a/Cargo.lock b/Cargo.lock index c32d59e..80f3a37 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 891ae77..734430d 100644 --- a/Cargo.toml +++ b/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"] diff --git a/src/baseliner.rs b/src/baseliner.rs index 6dd48bf..b879067 100644 --- a/src/baseliner.rs +++ b/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)] diff --git a/src/main.rs b/src/main.rs index 53eb1da..2384613 100644 --- a/src/main.rs +++ b/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 (error_tx, error_rx) = channel::(); - let (reselect_tx, reselect_rx) = channel(); + let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded(); + let (error_tx, error_rx) = flume::unbounded::(); + 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, diff --git a/src/metrics.rs b/src/metrics.rs index 2ca2a05..76301c7 100644 --- a/src/metrics.rs +++ b/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>, + tx: Option>, dropped: Arc, } impl MetricsSender { - pub fn new(tx: SyncSender<(Metric, u64)>, dropped: Arc) -> Self { + pub fn new(tx: Sender<(Metric, u64)>, dropped: Arc) -> Self { Self { tx: Some(tx), dropped, diff --git a/src/pinger.rs b/src/pinger.rs index 0ee960a..d6fbadf 100644 --- a/src/pinger.rs +++ b/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}; diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index aecbf19..75a15e5 100644 --- a/src/ratecontroller.rs +++ b/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; diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index 332b2cd..b8af45b 100644 --- a/src/reflector_selector.rs +++ b/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;