diff --git a/Cargo.lock b/Cargo.lock index bde2836..c996483 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,17 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + [[package]] name = "autocfg" version = "1.5.1" @@ -55,6 +66,12 @@ version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "cc" version = "1.3.0" @@ -183,9 +200,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "fastrand", + "futures-core", + "futures-sink", "spin", ] +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + [[package]] name = "getrandom" version = "0.4.3" @@ -217,9 +248,11 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "816dc56a4f7f76e42b0616818600b05528cd346cb3843276e82df982d5c3ee26" dependencies = [ + "async-trait", "nix", "socket2", "thiserror", + "tokio", ] [[package]] @@ -325,6 +358,17 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + [[package]] name = "netlink-bindings" version = "0.3.5" @@ -371,12 +415,41 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "pathdiff" version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3" +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "prettyplease" version = "0.2.37" @@ -411,6 +484,15 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.13.1" @@ -552,6 +634,22 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + +[[package]] +name = "smallvec" +version = "1.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e" + [[package]] name = "socket2" version = "0.6.5" @@ -589,6 +687,7 @@ dependencies = [ "rustix", "serde", "thiserror", + "tokio", "uci-derive", ] @@ -655,6 +754,34 @@ dependencies = [ "syn 3.0.6", ] +[[package]] +name = "tokio" +version = "1.53.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044" +dependencies = [ + "bytes", + "libc", + "mio", + "parking_lot", + "pin-project-lite", + "signal-hook-registry", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + [[package]] name = "toml" version = "1.1.3+spec-1.1.0" @@ -725,6 +852,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasm-bindgen" version = "0.2.126" diff --git a/Cargo.toml b/Cargo.toml index 120c8d5..cd61932 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,15 +20,16 @@ strip = "symbols" [dependencies] anyhow = "1.0" -icmp-socket2 = "0.6.4" -fastrand = "2" -rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } -thiserror = "2.0" -libc = "0.2" -flume = { version = "0.12.0", default-features = false, features = ["eventual-fairness", "select"] } config = { version = "0.15.25", default-features = false } csv = "1.4.0" +fastrand = "2" +flume = { version = "0.12.0", default-features = false, features = ["async", "eventual-fairness", "select"] } +icmp-socket2 = { version = "0.6.4", features = ["tokio"] } +libc = "0.2" +rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } serde = { version = "1", features = ["derive"] } +thiserror = "2.0" +tokio = { version = "1", features = ["full"] } uci-derive = { path = "crates/uci-derive" } [dependencies.log] diff --git a/src/baseliner.rs b/src/baseliner.rs index 1f05116..4908c43 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -6,8 +6,8 @@ // SPDX-License-Identifier: MPL-2.0 use crate::SHUTDOWN; +use crate::measurement::model::{MeasurementEvent, MeasurementObservation}; use crate::metrics::{Metric, MetricsSender}; -use crate::pinger::PingReply; use crate::settings::Settings; use flume::{Receiver, Sender}; use log::{debug, info}; @@ -29,10 +29,16 @@ pub struct EwmaStats { } impl EwmaStats { - fn from_reply(reply: &PingReply) -> Self { + fn from_reply(reply: &MeasurementObservation) -> Self { Self { - down: reply.down_time, - up: reply.up_time, + down: reply + .one_way + .map(|o| o.downlink.as_nanos() as f64 / 1000000.0) + .unwrap_or(reply.rtt_ms() / 2.0), + up: reply + .one_way + .map(|o| o.uplink.as_nanos() as f64 / 1000000.0) + .unwrap_or(reply.rtt_ms() / 2.0), } } @@ -73,11 +79,12 @@ pub struct Baseliner { pub settings: Settings, pub reselect_trigger: Sender, pub start_time: Instant, - pub stats_rx: Receiver, + pub stats_rx: Receiver, pub control_tx: Sender, pub selection_tx: Option>, pub baseline_metrics: MetricsSender, pub event_metrics: MetricsSender, + pub ping_metrics: MetricsSender, } fn ewma_factor(tick: f64, dur: f64) -> f64 { @@ -115,19 +122,19 @@ fn update_reflector_state( fn process_reply( reflectors: &mut HashMap, - reply: &PingReply, + reply: &MeasurementObservation, start_time: Instant, slow_factor: f64, fast_factor: f64, ) -> (ReflectorState, bool) { let sample = EwmaStats::from_reply(reply); let state = reflectors - .entry(reply.reflector) - .or_insert_with(|| ReflectorState::new(sample, reply.last_receive_time_s)); + .entry(reply.stream.peer) + .or_insert_with(|| ReflectorState::new(sample, reply.observed_at)); let anomaly = update_reflector_state( state, sample, - reply.last_receive_time_s, + reply.observed_at, start_time, slow_factor, fast_factor, @@ -165,8 +172,14 @@ impl Baseliner { return Ok(()); } - let reply = self.stats_rx.recv()?; - let reflector = reply.reflector; + let event = self.stats_rx.recv()?; + + let reply = match event { + MeasurementEvent::Observation(reply) => reply, + MeasurementEvent::Loss(_) => continue, + }; + + let reflector = reply.stream.peer; let (state, anomaly) = process_reply( &mut reflectors, &reply, @@ -189,6 +202,20 @@ impl Baseliner { let _ = self.reselect_trigger.try_send(true); } + self.ping_metrics.send(Metric::Ping { + reflector, + measurement_type: reply.stream.source, + rtt: reply.rtt_ms(), + up_time: reply + .one_way + .map(|o| o.uplink.as_nanos() as f64 / 1000000.0) + .unwrap_or(reply.rtt_ms() / 2.0), + down_time: reply + .one_way + .map(|o| o.downlink.as_nanos() as f64 / 1000000.0) + .unwrap_or(reply.rtt_ms() / 2.0), + }); + self.baseline_metrics.send(Metric::Baseline { reflector, baseline_up_ewma: state.baseline.up, @@ -218,7 +245,10 @@ impl Baseliner { #[cfg(test)] mod tests { use super::*; - use crate::settings::MeasurementType; + use crate::measurement::model::{ + MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, + SignedDuration, + }; use std::time::Duration; fn state( @@ -245,19 +275,25 @@ mod tests { EwmaStats { down, up } } - fn reply(reflector: IpAddr, down_time: f64, up_time: f64, received_at: Instant) -> PingReply { - PingReply { - reflector, - measurement_type: MeasurementType::Icmp, - seq: 0, - rtt: down_time + up_time, - current_time: 0, - down_time, - up_time, - originate_timestamp: 0, - receive_timestamp: 0, - transmit_timestamp: 0, - last_receive_time_s: received_at, + fn reply( + reflector: IpAddr, + down_time: f64, + up_time: f64, + received_at: Instant, + ) -> MeasurementObservation { + MeasurementObservation { + stream: MeasurementStream { + id: MeasurementStreamId(0), + peer: reflector, + source: MeasurementSource::IcmpEcho, + }, + started_at: received_at, + observed_at: received_at, + one_way: Some(OneWayLatency { + uplink: SignedDuration::from_millis(up_time as i64), + downlink: SignedDuration::from_millis(down_time as i64), + clock: OneWayClock::Uncalibrated, + }), } } diff --git a/src/controller.rs b/src/controller.rs new file mode 100644 index 0000000..11408a8 --- /dev/null +++ b/src/controller.rs @@ -0,0 +1,102 @@ +use std::{collections::HashMap, net::IpAddr, time::Duration}; + +use flume::{Receiver, SendError, Sender}; +use thiserror::Error; + +use crate::measurement::{ + icmp::IcmpBinding, + model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule}, +}; + +#[derive(Debug, Error)] +pub enum ControllerError { + #[error("ICMP channel error")] + IcmpChannelError(#[from] SendError>), +} + +pub struct Controller { + reflector_rx: Receiver>, + icmp_binding_tx: Sender>, + + source: MeasurementSource, + period: Duration, + + next_stream_id: u64, + stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>, +} + +impl Controller { + pub fn new( + reflector_rx: Receiver>, + icmp_binding_tx: Sender>, + ) -> Self { + Self { + reflector_rx, + icmp_binding_tx, + source: MeasurementSource::IcmpTimestamp, + period: Duration::from_millis(500), + next_stream_id: 0, + stream_ids: HashMap::new(), + } + } + + pub async fn run(mut self, initial_reflectors: Vec) -> Result<(), ControllerError> { + self.update_reflectors(initial_reflectors).await?; + + while let Ok(reflectors) = self.reflector_rx.recv_async().await { + self.update_reflectors(reflectors).await?; + } + + Ok(()) + } + + fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream { + let key = (peer, self.source); + + let id = *self.stream_ids.entry(key).or_insert_with(|| { + let id = MeasurementStreamId(self.next_stream_id); + self.next_stream_id += 1; + id + }); + + MeasurementStream { + id, + peer, + source: self.source, + } + } + + async fn update_reflectors(&mut self, reflectors: Vec) -> Result<(), ControllerError> { + let count = reflectors.len(); + + let bindings = reflectors + .into_iter() + .enumerate() + .map(|(index, peer)| { + let stream = self.stream_for(peer); + + IcmpBinding::new( + stream, + ProbeSchedule { + period: self.period, + offset: evenly_spaced_offset(self.period, index, count), + }, + ) + }) + .collect(); + + self.icmp_binding_tx + .send_async(bindings) + .await + .map_err(|e| ControllerError::IcmpChannelError(e)) + } +} + +fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration { + let nanos = period.as_nanos() * index as u128 / count as u128; + + Duration::new( + (nanos / 1_000_000_000) as u64, + (nanos % 1_000_000_000) as u32, + ) +} diff --git a/src/main.rs b/src/main.rs index ad6c000..560cab9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,11 +8,10 @@ extern crate core; mod baseliner; +mod controller; mod log; +mod measurement; mod metrics; -mod pinger; -mod pinger_icmp; -mod pinger_icmp_ts; mod platform; mod ratecontroller; mod reflector_selector; @@ -21,36 +20,29 @@ mod time; mod util; use crate::baseliner::Baseliner; +use crate::controller::Controller; +use crate::measurement::icmp::IcmpEngine; use crate::metrics::{Metric, Metrics, MetricsSender}; -use crate::pinger::{InFlightProbeCache, PingListener, PingSender}; -use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender}; -use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender}; use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations}; use crate::ratecontroller::Ratecontroller; use crate::reflector_selector::ReflectorSelector; -use crate::settings::MeasurementType; use crate::settings::Settings; +use crate::util::RwLockExt; use ::log::{info, warn}; 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::{Arc, Mutex, RwLock}; -use std::thread::sleep; +use std::sync::{Arc, RwLock}; +use std::thread; +use std::thread::{JoinHandle, sleep}; use std::time::{Duration, Instant}; -use std::{process, thread}; struct ReflectorSetup { peers: Arc>>, pool: Vec, reselection_enabled: bool, - active_count: usize, } - -type PingListenerBox = Box; -type PingSenderBox = Box; - pub static SHUTDOWN: AtomicBool = AtomicBool::new(false); extern "C" fn signal_handler(_: libc::c_int) { @@ -58,47 +50,6 @@ extern "C" fn signal_handler(_: libc::c_int) { } const VERSION: &str = env!("CARGO_PKG_VERSION"); -const RESELECTION_CANDIDATE_BURST: usize = 20; -const INFLIGHT_CAPACITY_DUPLICATE_FACTOR: usize = 2; -const INFLIGHT_CAPACITY_MIN: usize = 256; - -fn compute_inflight_probe_capacity( - settings: &Settings, - active_reflector_count: usize, - reselection_enabled: bool, -) -> usize { - let tick_interval_ms = (settings.advanced_settings.tick_interval * 1000.0).max(1.0); - let expected_path_delay_ms = (settings.advanced_settings.download_delay_ms - + settings.advanced_settings.upload_delay_ms) - .max(10.0); - // Keep enough room for severe queueing events while still adapting to configured delay budgets. - let max_rtt_ms = (expected_path_delay_ms * 20.0).clamp(1000.0, 10000.0); - - let burst_reflector_count = if reselection_enabled { - active_reflector_count + RESELECTION_CANDIDATE_BURST - } else { - active_reflector_count - }; - let probes_per_reflector = (max_rtt_ms / tick_interval_ms).ceil() as usize; - - (burst_reflector_count - .saturating_mul(probes_per_reflector) - .saturating_mul(INFLIGHT_CAPACITY_DUPLICATE_FACTOR)) - .max(INFLIGHT_CAPACITY_MIN) -} - -fn create_pinger(measurement_type: MeasurementType) -> (PingListenerBox, PingSenderBox) { - match measurement_type { - MeasurementType::Icmp => ( - Box::new(PingerICMPEchoListener {}), - Box::new(PingerICMPEchoSender {}), - ), - MeasurementType::IcmpTimestamps => ( - Box::new(PingerICMPTimestampListener {}), - Box::new(PingerICMPTimestampSender {}), - ), - } -} fn initialize_shaper( settings: &Settings, @@ -155,11 +106,6 @@ fn install_signal_handlers() { } } -fn probe_identifier() -> u16 { - // The ICMP identifier is only two bytes. - (process::id() & 0xffff) as u16 -} - fn restore_shaper( settings: &Settings, traffic_control: &mut T, @@ -212,17 +158,52 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result { peers: Arc::new(RwLock::new(default_reflectors.to_vec())), pool, reselection_enabled, - active_count: default_reflectors - .len() - .max(settings.advanced_settings.num_reflectors as usize), }) } +fn spawn_task( + rt: &tokio::runtime::Runtime, + error_tx: &flume::Sender, + task: F, +) -> tokio::task::JoinHandle<()> +where + F: Future> + Send + 'static, + E: Into + Send + 'static, +{ + let error_tx = error_tx.clone(); + + rt.spawn(async move { + if let Err(error) = task.await { + let _ = error_tx.send(error.into()); + } + }) +} + +fn spawn_worker( + name: &str, + error_tx: &flume::Sender, + worker: F, +) -> anyhow::Result> +where + F: FnOnce() -> anyhow::Result<()> + Send + 'static, +{ + let error_tx = error_tx.clone(); + + thread::Builder::new() + .name(name.to_owned()) + .spawn(move || { + if let Err(error) = worker() { + let _ = error_tx.send(error); + } + }) + .map_err(|e| anyhow::anyhow!(e)) +} + fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<()> { loop { match error_rx.recv_timeout(Duration::from_secs(1)) { Ok(error) => { - return Err(anyhow::anyhow!("thread exited with error: {error}")); + return Err(anyhow::anyhow!("worker exited with error: {error}")); } Err(RecvTimeoutError::Disconnected) => { return Ok(()); @@ -238,30 +219,45 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } fn run(settings: &Settings) -> anyhow::Result<()> { + let rt = tokio::runtime::Runtime::new()?; + if settings.advanced_settings.dry_run { info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***"); } let start_time = Instant::now(); - let probe_id = probe_identifier(); let ReflectorSetup { peers: reflector_peers, pool: reflector_pool, reselection_enabled, - active_count: active_reflector_count, } = setup_reflectors(settings)?; - let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded(); let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded(); + let (error_tx, error_rx) = flume::unbounded::(); + let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded(); + let (measurement_tx, measurement_rx) = flume::unbounded(); + let (reflector_tx, reflector_rx) = flume::unbounded(); + let (reselect_tx, reselect_rx) = flume::bounded(1); let (selection_snapshot_tx, selection_snapshot_rx) = if reselection_enabled { let (tx, rx) = flume::unbounded(); (Some(tx), Some(rx)) } else { (None, None) }; - let (error_tx, error_rx) = flume::unbounded::(); - let (reselect_tx, reselect_rx) = flume::bounded(1); + + let controller = Controller::new(reflector_rx, icmp_binding_tx); + let engine = { + let _guard = rt.enter(); + IcmpEngine::new( + start_time, + &settings.network, + measurement_tx, + icmp_binding_rx, + )? + }; + + spawn_task(&rt, &error_tx, engine.run()); let dropped = Arc::new(AtomicU32::new(0)); @@ -272,14 +268,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { metrics_rx: rx, metrics_dropped: Arc::clone(&dropped), }; - let err_tx = error_tx.clone(); - let handle = thread::Builder::new() - .name("metrics".to_string()) - .spawn(move || { - if let Err(error) = metrics.run() { - let _ = err_tx.send(error); - } - })?; + let handle = spawn_worker("metrics", &error_tx, move || metrics.run())?; (Some(tx), Some(handle)) } else { (None, None) @@ -309,82 +298,27 @@ fn run(settings: &Settings) -> anyhow::Result<()> { }, }); - let (mut ping_listener, mut ping_sender) = - create_pinger(settings.advanced_settings.measurement_type); - - let inflight_probe_capacity = - compute_inflight_probe_capacity(settings, active_reflector_count, reselection_enabled); - info!( - "In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})", - inflight_probe_capacity, - active_reflector_count, - reselection_enabled, - settings.advanced_settings.tick_interval - ); - - let inflight: InFlightProbeCache = - Arc::new(Mutex::new(HashMap::with_capacity(inflight_probe_capacity))); - let baseliner = Baseliner { settings: settings.clone(), reselect_trigger: reselect_tx.clone(), start_time, - stats_rx: baseliner_stats_rx, + stats_rx: measurement_rx, control_tx: control_snapshot_tx, selection_tx: selection_snapshot_tx, baseline_metrics, event_metrics: event_metrics.clone(), + ping_metrics, }; let mut main_traffic_control = traffic_control_backend(); let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?; - let err_tx = error_tx.clone(); - let listener_peers = Arc::clone(&reflector_peers); - let listener_inflight = Arc::clone(&inflight); - let measurement_type = settings.advanced_settings.measurement_type; - thread::Builder::new() - .name("receiver".to_string()) - .spawn(move || { - if let Err(error) = ping_listener.listen( - probe_id, - measurement_type, - listener_peers, - listener_inflight, - baseliner_stats_tx, - ping_metrics, - ) { - let _ = err_tx.send(error); - } - })?; - - let err_tx = error_tx.clone(); - thread::Builder::new() - .name("baseliner".to_string()) - .spawn(move || { - if let Err(error) = baseliner.run() { - let _ = err_tx.send(error); - } - })?; - - let err_tx = error_tx.clone(); - let sender_peers = Arc::clone(&reflector_peers); - let sender_inflight = Arc::clone(&inflight); - let measurement_type = settings.advanced_settings.measurement_type; - let tick_interval = settings.advanced_settings.tick_interval; - thread::Builder::new() - .name("sender".to_string()) - .spawn(move || { - if let Err(error) = ping_sender.send( - probe_id, - measurement_type, - sender_peers, - sender_inflight, - tick_interval, - ) { - let _ = err_tx.send(error); - } - })?; + spawn_worker("baseliner", &error_tx, move || baseliner.run())?; + spawn_task( + &rt, + &error_tx, + controller.run(reflector_peers.read_anyhow()?.clone()), + ); let main_event_metrics = event_metrics.clone(); @@ -397,15 +331,9 @@ fn run(settings: &Settings) -> anyhow::Result<()> { reflector_pool, trigger_channel: reselect_rx, metrics: event_metrics, + reflector_tx, }; - let err_tx = error_tx.clone(); - thread::Builder::new() - .name("reselection".to_string()) - .spawn(move || { - if let Err(error) = reflector_selector.run() { - let _ = err_tx.send(error); - } - })?; + spawn_worker("reselection", &error_tx, move || reflector_selector.run())?; } // Give the baseliner time to collect initial samples before adjusting rates. @@ -419,14 +347,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { rate_metrics, )?; - let err_tx = error_tx.clone(); - thread::Builder::new() - .name("ratecontroller".to_string()) - .spawn(move || { - if let Err(error) = ratecontroller.run() { - let _ = err_tx.send(error); - } - })?; + spawn_worker("ratecontroller", &error_tx, move || ratecontroller.run())?; // Drop the original sender so the channel disconnects if all workers exit cleanly. drop(error_tx); diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs new file mode 100644 index 0000000..3d96b3b --- /dev/null +++ b/src/measurement/icmp.rs @@ -0,0 +1,505 @@ +use std::{ + collections::{HashMap, HashSet}, + io, + net::IpAddr, + time::{Duration, Instant, SystemTime}, +}; + +use flume::{Receiver, Sender}; +use icmp_socket2::{ + AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, + packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest}, + tokio::AsyncIcmpV4Socket, +}; +use log::debug; +use thiserror::Error; + +use crate::{ + measurement::model::{ + MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementSource, + MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, + SignedDuration, + }, + settings::NetworkSettings, + time::{ClockSample, Realtime}, +}; + +pub struct IcmpBinding { + pub stream: MeasurementStream, + pub schedule: ProbeSchedule, + + next_probe_at: Instant, + next_sequence: u16, +} + +impl IcmpBinding { + pub fn new(stream: MeasurementStream, schedule: ProbeSchedule) -> Self { + Self { + stream, + schedule, + next_probe_at: Instant::now(), + next_sequence: 0, + } + } +} + +#[derive(Debug, Error)] +pub enum IcmpError { + #[error("I/O error")] + Io(#[from] io::Error), + + #[error("failed to build ICMP packet")] + PacketBuild(#[from] IcmpPacketBuildError), + + #[error("invalid ICMP packet: {0}")] + InvalidPacket(String), + + #[error("unsupported measurement source: {0:?}")] + UnsupportedSource(MeasurementSource), + + #[error("probe interval must be non-zero for stream {0:?}")] + InvalidInterval(MeasurementStreamId), + + #[error("probe offset must be less than the period for stream {0:?}")] + InvalidOffset(MeasurementStreamId), + + #[error( + "no matching probe found for peer {:?}, source: {:?}, sequence: {}", + .0.0, + .0.1, + .0.2 + )] + NoMatchingProbe(InFlightProbeKey), + + #[error("wrong identifier: expected {expected}, found {found}")] + WrongIdentifier { expected: u16, found: u16 }, + + #[error("unsupported ICMP peer address: {0}")] + UnsupportedPeer(IpAddr), + + #[error("event channel closed")] + EventChannelClosed, + + #[error("binding channel closed")] + BindingChannelClosed, +} + +#[derive(Debug, Clone, Copy)] +struct InFlightProbe { + stream: MeasurementStream, + sent_at: Instant, +} + +type InFlightProbeKey = (IpAddr, MeasurementSource, u16); +type InFlightProbes = HashMap; + +pub struct IcmpEngine { + socket: AsyncIcmpV4Socket, + + identifier: u16, + + bindings: HashMap, + inflight: HashMap, + + origin: Instant, + + binding_rx: Receiver>, + observation_tx: Sender, +} + +async fn wait_for_deadline(deadline: Option) { + match deadline { + Some(deadline) => { + tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await; + } + + None => { + std::future::pending::<()>().await; + } + } +} + +impl IcmpEngine { + pub fn new( + origin: Instant, + #[allow(unused)] settings: &NetworkSettings, + observation_tx: Sender, + binding_rx: Receiver>, + ) -> Result { + #[allow(unused_mut)] + let mut socket = IcmpSocket4::new()?; + + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + if let Some(device) = &settings.measurement_bind_device { + icmp_socket2::IcmpSocket::bind_device(&mut socket, device)?; + } + + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + if let Some(mark) = settings.measurement_mark { + icmp_socket2::IcmpSocket::set_mark(&mut socket, mark)?; + } + + #[cfg(target_os = "freebsd")] + if let Some(fib) = settings.measurement_fib { + icmp_socket2::IcmpSocket::set_fib(&mut socket, fib)?; + } + + socket.enable_receive_metadata()?; + let identifier = (std::process::id() & 0xffff) as u16; + let inflight: InFlightProbes = HashMap::new(); + + Ok(Self { + socket: socket.into_tokio()?, + identifier, + bindings: HashMap::new(), + inflight, + origin, + binding_rx, + observation_tx, + }) + } + + fn set_bindings(&mut self, desired: Vec) -> Result<(), IcmpError> { + // Validate the complete desired state before mutating anything. + for binding in &desired { + if binding.schedule.period.is_zero() { + return Err(IcmpError::InvalidInterval(binding.stream.id)); + } + + if binding.schedule.offset >= binding.schedule.period { + return Err(IcmpError::InvalidOffset(binding.stream.id)); + } + + if !matches!( + binding.stream.source, + MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp + ) { + return Err(IcmpError::UnsupportedSource(binding.stream.source)); + } + } + + let now = Instant::now(); + let desired_ids = desired + .iter() + .map(|binding| binding.stream.id) + .collect::>(); + + // Remove bindings that are no longer desired. + self.bindings.retain(|id, _| desired_ids.contains(id)); + self.inflight + .retain(|_, probe| desired_ids.contains(&probe.stream.id)); + + for desired in desired { + match self.bindings.get_mut(&desired.stream.id) { + Some(current) if current.schedule == desired.schedule => { + // Same schedule, so leave next_probe_at alone. + current.stream = desired.stream; + } + + Some(current) => { + // Same stream, but its schedule changed. Rejoin the new grid. + current.stream = desired.stream; + current.schedule = desired.schedule; + current.next_probe_at = desired.schedule.next_probe_at(self.origin, now); + } + + None => { + // New stream. Join its configured schedule at the next + // occurrence after now. + let mut binding = desired; + binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now); + + self.bindings.insert(binding.stream.id, binding); + } + } + } + + Ok(()) + } + + async fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> { + let due = self + .bindings + .iter() + .filter_map(|(&stream_id, binding)| (binding.next_probe_at <= now).then_some(stream_id)) + .collect::>(); + + for stream_id in due { + self.send_probe(stream_id).await?; + + let binding = self + .bindings + .get_mut(&stream_id) + .expect("due binding disappeared"); + + binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now); + } + + Ok(()) + } + + async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> { + let (stream, sequence) = { + let binding = self + .bindings + .get_mut(&stream_id) + .expect("binding disappeared"); + + let sequence = binding.next_sequence; + binding.next_sequence = binding.next_sequence.wrapping_add(1); + + (binding.stream, sequence) + }; + + let IpAddr::V4(peer) = stream.peer else { + return Err(IcmpError::UnsupportedPeer(stream.peer)); + }; + + let packet = match stream.source { + MeasurementSource::IcmpEcho => { + Icmpv4Packet::with_echo_request(self.identifier, sequence, Vec::new())? + } + MeasurementSource::IcmpTimestamp => { + let originate = ClockSample::::now().as_time_since_midnight() as u32; + Icmpv4Packet::with_timestamp_request(self.identifier, sequence, originate, 0, 0)? + } + source => return Err(IcmpError::UnsupportedSource(source)), + }; + + let sent_at = Instant::now(); + let key = (stream.peer, stream.source, sequence); + + self.socket.send_to(peer, packet).await?; + self.inflight.insert(key, InFlightProbe { stream, sent_at }); + + Ok(()) + } + + fn next_deadline(&self) -> Option { + self.bindings + .values() + .map(|binding| binding.next_probe_at) + .min() + } + + fn handle_reply( + &mut self, + packet: Icmpv4Packet, + peer: IpAddr, + observed_at: Instant, + kernel_rx_timestamp: Option, + ) -> Result { + match packet.message { + Icmpv4Message::EchoReply { + identifier, + sequence, + .. + } => { + if identifier != self.identifier { + return Err(IcmpError::WrongIdentifier { + expected: self.identifier, + found: identifier, + }); + } + + let key = (peer, MeasurementSource::IcmpEcho, sequence); + let probe = self + .inflight + .remove(&key) + .ok_or(IcmpError::NoMatchingProbe(key))?; + + debug!( + "ICMP engine > Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:>8} | Received time: {:>8} | Transmit time: {:>8} | RTT: {:8.3} ms | UL time: {:>8} | DL time: {:>8}", + "ICMP", + peer, + sequence, + ClockSample::::now().as_time_since_midnight(), + "N/A", + "N/A", + "N/A", + observed_at.duration_since(probe.sent_at).as_secs_f64() * 1_000.0, + "N/A", + "N/A", + ); + + Ok(MeasurementObservation { + stream: probe.stream, + started_at: probe.sent_at, + observed_at, + one_way: None, + }) + } + Icmpv4Message::TimestampReply { + identifier, + sequence, + originate, + receive, + transmit, + } => { + if identifier != self.identifier { + return Err(IcmpError::WrongIdentifier { + expected: self.identifier, + found: identifier, + }); + } + + if originate >= MS_PER_DAY as u32 + || receive >= MS_PER_DAY as u32 + || transmit >= MS_PER_DAY as u32 + { + return Err(IcmpError::InvalidPacket( + "non-standard RFC 792 timestamp".into(), + )); + } + + let key = (peer, MeasurementSource::IcmpTimestamp, sequence); + let probe = self + .inflight + .remove(&key) + .ok_or(IcmpError::NoMatchingProbe(key))?; + + let now_ms = if let Some(timestamp) = kernel_rx_timestamp { + ClockSample::::from(timestamp).as_time_since_midnight() + } else { + ClockSample::::now().as_time_since_midnight() + }; + + let uplink_ms = timestamp_delta(receive as i64, originate as i64); + let downlink_ms = timestamp_delta(now_ms, transmit as i64); + + debug!( + "ICMP engine > Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:8} | Received time: {:8} | Transmit time: {:8} | RTT: {:8.3} ms | UL time: {:5} ms | DL time: {:5} ms", + "ICMP", + peer, + sequence, + now_ms, + originate, + receive, + transmit, + observed_at.duration_since(probe.sent_at).as_secs_f64() * 1_000.0, + uplink_ms, + downlink_ms, + ); + + Ok(MeasurementObservation { + stream: probe.stream, + started_at: probe.sent_at, + observed_at, + one_way: Some(OneWayLatency { + uplink: SignedDuration::from_millis(uplink_ms), + downlink: SignedDuration::from_millis(downlink_ms), + clock: OneWayClock::Uncalibrated, + }), + }) + } + _ => Err(IcmpError::InvalidPacket( + "unsupported ICMP message type".to_string(), + )), + } + } + + async fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> { + const PROBE_TIMEOUT: Duration = Duration::from_secs(30); + + let expired = { + let mut expired = Vec::new(); + + self.inflight.retain(|_, probe| { + if now.saturating_duration_since(probe.sent_at) >= PROBE_TIMEOUT { + expired.push(*probe); + false + } else { + true + } + }); + + expired + }; + + for probe in expired { + self.observation_tx + .send_async(MeasurementEvent::Loss(MeasurementLoss { + stream: probe.stream, + started_at: probe.sent_at, + detected_at: now, + })) + .await + .map_err(|_| IcmpError::EventChannelClosed)?; + } + + Ok(()) + } + + pub async fn run(mut self) -> Result<(), IcmpError> { + const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500); + + let mut housekeeping = tokio::time::interval_at( + tokio::time::Instant::now() + HOUSEKEEPING_INTERVAL, + HOUSEKEEPING_INTERVAL, + ); + + housekeeping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + let deadline = self.next_deadline(); + + tokio::select! { + bindings = self.binding_rx.recv_async() => { + let bindings = bindings + .map_err(|_| IcmpError::BindingChannelClosed)?; + + self.set_bindings(bindings)?; + } + + _ = wait_for_deadline(deadline) => { + self.send_due_probes(Instant::now()).await?; + } + + result = self.socket.rcv_from_with_meta() => { + let result = result?; + + match self.handle_reply( + result.packet, + result.peer.as_socket().unwrap().ip(), + result.received_at, + result.kernel_rx_timestamp, + ) { + Ok(observation) => { + self.observation_tx + .send_async(MeasurementEvent::Observation(observation)) + .await + .map_err(|_| IcmpError::EventChannelClosed)?; + } + + Err( + IcmpError::WrongIdentifier { .. } + | IcmpError::NoMatchingProbe { .. } + | IcmpError::InvalidPacket(_), + ) => { + // Not ours, malformed, or late. + } + + Err(err) => return Err(err), + } + } + + _ = housekeeping.tick() => { + self.expire_inflight(Instant::now()).await?; + } + } + } + } +} + +const MS_PER_DAY: i64 = 86_400_000; + +fn timestamp_delta(later: i64, earlier: i64) -> i64 { + let delta = later - earlier; + if delta < -MS_PER_DAY / 2 { + delta + MS_PER_DAY + } else if delta > MS_PER_DAY / 2 { + delta - MS_PER_DAY + } else { + delta + } +} diff --git a/src/measurement/mod.rs b/src/measurement/mod.rs new file mode 100644 index 0000000..0ee60f2 --- /dev/null +++ b/src/measurement/mod.rs @@ -0,0 +1,2 @@ +pub(crate) mod icmp; +pub(crate) mod model; diff --git a/src/measurement/model.rs b/src/measurement/model.rs new file mode 100644 index 0000000..12b7bf5 --- /dev/null +++ b/src/measurement/model.rs @@ -0,0 +1,149 @@ +use std::fmt::{self, Display}; +use std::{ + net::IpAddr, + time::{Duration, Instant}, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct MeasurementStreamId(pub u64); + +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum MeasurementSource { + IcmpEcho, + IcmpTimestamp, + Irtt, + TcpSeqAck, + TcpTimestamp, +} + +impl Display for MeasurementSource { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + MeasurementSource::IcmpEcho => write!(f, "icmp"), + MeasurementSource::IcmpTimestamp => write!(f, "icmp-timestamps"), + MeasurementSource::Irtt => write!(f, "irtt"), + MeasurementSource::TcpSeqAck => write!(f, "tcp-seqack"), + MeasurementSource::TcpTimestamp => write!(f, "tcp-timestamp"), + } + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct MeasurementStream { + pub id: MeasurementStreamId, + pub peer: IpAddr, + pub source: MeasurementSource, +} + +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OneWayClock { + Uncalibrated, + Calibrated, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ProbeSchedule { + pub period: Duration, + pub offset: Duration, +} + +impl ProbeSchedule { + pub(crate) fn next_probe_at(&self, origin: Instant, now: Instant) -> Instant { + const NANOS_PER_SEC: u128 = 1_000_000_000; + + let first = origin + self.offset; + + if now < first { + return first; + } + + let elapsed_ns = now.duration_since(first).as_nanos(); + let period_ns = self.period.as_nanos(); + let remainder = elapsed_ns % period_ns; + + let until_next = if remainder == 0 { + period_ns + } else { + period_ns - remainder + }; + + let until_next = Duration::new( + (until_next / NANOS_PER_SEC) as u64, + (until_next % NANOS_PER_SEC) as u32, + ); + + now + until_next + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct SignedDuration { + nanos: i64, +} + +#[allow(dead_code)] +impl SignedDuration { + pub const fn from_nanos(nanos: i64) -> Self { + Self { nanos } + } + + pub const fn from_millis(millis: i64) -> Self { + Self { + nanos: millis * 1_000_000, + } + } + + pub const fn as_nanos(self) -> i64 { + self.nanos + } + + pub fn as_secs_f64(self) -> f64 { + self.nanos as f64 / 1_000_000_000.0 + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct OneWayLatency { + pub uplink: SignedDuration, + pub downlink: SignedDuration, + pub clock: OneWayClock, +} + +#[derive(Debug, Clone, Copy)] +pub struct MeasurementObservation { + pub stream: MeasurementStream, + + /// Local monotonic time at which measurement started. + pub started_at: Instant, + + /// Local monotonic time at which the result was observed. + pub observed_at: Instant, + + /// Directional timing, when the measurement mechanism can provide it. + pub one_way: Option, +} + +impl MeasurementObservation { + pub fn rtt(&self) -> Duration { + self.observed_at.duration_since(self.started_at) + } + + pub fn rtt_ms(&self) -> f64 { + self.rtt().as_secs_f64() * 1000.0 + } +} + +#[allow(dead_code)] +pub struct MeasurementLoss { + pub stream: MeasurementStream, + pub started_at: Instant, + pub detected_at: Instant, +} + +#[allow(dead_code)] +pub enum MeasurementEvent { + Observation(MeasurementObservation), + Loss(MeasurementLoss), +} diff --git a/src/metrics.rs b/src/metrics.rs index a989161..fac27c9 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -2,8 +2,9 @@ // // SPDX-License-Identifier: MPL-2.0 +use crate::measurement::model::MeasurementSource; +use crate::settings::ObservabilityProtocol; use crate::settings::Settings; -use crate::settings::{MeasurementType, ObservabilityProtocol}; use crate::time::{ClockSample, Realtime}; use flume::{Receiver, RecvTimeoutError, Sender}; use log::{error, info, warn}; @@ -93,7 +94,7 @@ impl Transport { pub enum Metric { Ping { reflector: IpAddr, - measurement_type: MeasurementType, + measurement_type: MeasurementSource, rtt: f64, up_time: f64, down_time: f64, diff --git a/src/pinger.rs b/src/pinger.rs deleted file mode 100644 index 54e83a6..0000000 --- a/src/pinger.rs +++ /dev/null @@ -1,214 +0,0 @@ -// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot) -// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan) -// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe) -// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) -// -// SPDX-License-Identifier: MPL-2.0 - -use crate::MeasurementType; -use crate::SHUTDOWN; -use crate::metrics::{Metric, MetricsSender}; -use crate::util::{MutexExt, RwLockExt}; -use flume::Sender; -use icmp_socket2::socket::IcmpSocket; -use icmp_socket2::{IcmpSocket4, Icmpv4Packet}; -use log::{debug, info, warn}; -use std::collections::HashMap; -use std::net::{IpAddr, Ipv4Addr}; -use std::sync::atomic::Ordering; -use std::sync::{Arc, Mutex, RwLock}; -use std::time::{Duration, Instant}; -use std::{io, thread}; -use thiserror::Error; - -#[derive(Error, Debug)] -pub enum PingError { - #[error("Couldn't parse number")] - InvalidNumber(#[from] io::Error), - #[error("Error parsing packet")] - InvalidPacket(String), - #[error("Invalid packet type")] - InvalidType(String), - #[error("Wrong ICMP identifier (expected {expected:?}, found {found:?})")] - WrongID { expected: u16, found: u16 }, -} - -pub struct PingReply { - pub reflector: IpAddr, - pub measurement_type: MeasurementType, - pub seq: u16, - pub rtt: f64, - pub current_time: i64, - pub down_time: f64, - pub up_time: f64, - pub originate_timestamp: i64, - pub receive_timestamp: i64, - pub transmit_timestamp: i64, - pub last_receive_time_s: Instant, -} - -#[derive(Clone, Copy, Debug)] -pub struct InFlightProbe { - pub sent_at: Instant, - pub originate_timestamp: i64, -} - -pub type InFlightProbeKey = (IpAddr, MeasurementType, u16); -pub type InFlightProbeCache = Arc>>; -const INFLIGHT_PROBE_TTL: Duration = Duration::from_secs(30); - -fn open_socket(type_: MeasurementType) -> io::Result { - match type_ { - MeasurementType::Icmp | MeasurementType::IcmpTimestamps => IcmpSocket4::new(), - } -} - -pub trait PingListener { - fn listen( - &mut self, - id: u16, - type_: MeasurementType, - reflectors_lock: Arc>>, - inflight: InFlightProbeCache, - stats_tx: Sender, - ping_metrics: MetricsSender, - ) -> anyhow::Result<()> { - let socket = &mut open_socket(type_)?; - - loop { - if SHUTDOWN.load(Ordering::Relaxed) { - info!("Ping listener shutting down"); - return Ok(()); - } - let (pkt, sender) = match socket.rcv_from() { - Ok(val) => val, - Err(_) => continue, - }; - - let addr: IpAddr = sender.as_socket().unwrap().ip(); - - let reflectors = reflectors_lock.read_anyhow()?; - if !reflectors.contains(&addr) { - continue; - } - - let reply = match self.parse_packet(id, addr, type_, pkt) { - Ok(val) => val, - Err(_) => { - // parse_packet will throw an error if it's an unknown protocol etc. - // so just quietly move on - continue; - } - }; - - let key = (addr, type_, reply.seq); - let Some(probe) = inflight.lock_anyhow()?.remove(&key) else { - continue; - }; - if reply.originate_timestamp != probe.originate_timestamp { - continue; - } - - ping_metrics.send(Metric::Ping { - reflector: reply.reflector, - measurement_type: reply.measurement_type, - rtt: reply.rtt, - up_time: reply.up_time, - down_time: reply.down_time, - }); - - debug!( - "Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:8} | Received time: {:8} | Transmit time : {:8} | RTT: {:8} | UL time: {:5} | DL time: {:5}", - "ICMP", - addr.to_string(), - reply.seq, - reply.current_time, - reply.originate_timestamp, - reply.receive_timestamp, - reply.transmit_timestamp, - reply.rtt, - reply.up_time, - reply.down_time - ); - if let Err(e) = stats_tx.send(reply) { - warn!("Stats channel closed, stopping listener: {}", e); - break Ok(()); - } - } - } - - fn parse_packet( - &self, - id: u16, - reflector: IpAddr, - measurement_type: MeasurementType, - pkt: Icmpv4Packet, - ) -> Result; -} - -pub trait PingSender { - fn send( - &mut self, - id: u16, - type_: MeasurementType, - reflectors_lock: Arc>>, - inflight: InFlightProbeCache, - tick_interval: f64, - ) -> anyhow::Result<()> { - let mut socket = open_socket(type_)?; - - let mut seq: u16 = 0; - let tick_duration_ms: u16 = (tick_interval * 1000.0) as u16; - - loop { - if SHUTDOWN.load(Ordering::Relaxed) { - info!("Ping sender shutting down"); - return Ok(()); - } - // Clone the reflectors vec and drop the read lock immediately — - // holding it for the entire tick would starve the reflector selector's write lock. - let reflectors_unlocked = reflectors_lock.read_anyhow()?; - let reflectors = reflectors_unlocked.clone(); - drop(reflectors_unlocked); - - if reflectors.is_empty() { - thread::sleep(Duration::from_millis(tick_duration_ms as u64)); - continue; - } - - inflight - .lock_anyhow()? - .retain(|_, probe| probe.sent_at.elapsed() < INFLIGHT_PROBE_TTL); - - let sleep_duration = - Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64); - - for reflector in reflectors.iter() { - let addr: Ipv4Addr = match reflector { - IpAddr::V4(ipv4) => *ipv4, - IpAddr::V6(_) => unimplemented!(), - }; - - let (packet, originate_timestamp) = self.craft_packet(id, seq); - let sent_at = Instant::now(); - let key = (*reflector, type_, seq); - inflight.lock_anyhow()?.insert( - key, - InFlightProbe { - sent_at, - originate_timestamp, - }, - ); - if let Err(e) = socket.send_to(addr, packet) { - inflight.lock_anyhow()?.remove(&key); - return Err(e.into()); - } - thread::sleep(sleep_duration); - } - - seq = seq.wrapping_add(1); - } - } - - fn craft_packet(&self, id: u16, seq: u16) -> (icmp_socket2::packet::Icmpv4Packet, i64); -} diff --git a/src/pinger_icmp.rs b/src/pinger_icmp.rs deleted file mode 100644 index 7dfceea..0000000 --- a/src/pinger_icmp.rs +++ /dev/null @@ -1,96 +0,0 @@ -// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot) -// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan) -// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe) -// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) -// -// SPDX-License-Identifier: MPL-2.0 - -use std::net::IpAddr; -use std::time::Instant; - -use crate::pinger::{PingError, PingListener, PingReply, PingSender}; -use crate::settings::MeasurementType; -use crate::time::{ClockSample, Monotonic}; -use icmp_socket2::Icmpv4Message; -use icmp_socket2::Icmpv4Packet; -use icmp_socket2::packet::WithEchoRequest; - -pub struct PingerICMPEchoListener {} - -pub struct PingerICMPEchoSender {} - -impl PingListener for PingerICMPEchoListener { - // Result: RTT, down time, up time - fn parse_packet( - &self, - id: u16, - reflector: IpAddr, - measurement_type: MeasurementType, - pkt: Icmpv4Packet, - ) -> Result { - match pkt.typ { - 0 => { - if let Icmpv4Message::EchoReply { - identifier, - sequence, - payload, - } = pkt.message - { - if identifier != id { - return Err(PingError::WrongID { - expected: id, - found: identifier, - }); - } - - let time_sent = match payload.as_slice().try_into() { - Ok(bytes) => u64::from_be_bytes(bytes) as i64, - Err(_) => { - return Err(PingError::InvalidPacket(format!( - "Expected 8 bytes payload, but found {}", - payload.len() - ))); - } - }; - - let clock = ClockSample::::now(); - let time_ms = clock.to_milliseconds() as i64; - - let rtt = (time_ms - time_sent) as f64; - Ok(PingReply { - reflector, - measurement_type, - seq: sequence, - rtt, - current_time: time_ms, - down_time: (rtt / 2.0) as f64, - up_time: (rtt / 2.0) as f64, - originate_timestamp: time_sent, - receive_timestamp: 0, - transmit_timestamp: 0, - last_receive_time_s: Instant::now(), - }) - } else { - Err(PingError::InvalidPacket(format!( - "Packet had type {:?}, but did not match the structure", - pkt.typ - ))) - } - } - type_ => Err(PingError::InvalidType(format!("{:?}", type_))), - } - } -} - -impl PingSender for PingerICMPEchoSender { - fn craft_packet(&self, id: u16, seq: u16) -> (Icmpv4Packet, i64) { - let clock = ClockSample::::now(); - let time_ms = clock.to_milliseconds(); - let payload = time_ms.to_be_bytes().to_vec(); - - ( - Icmpv4Packet::with_echo_request(id, seq, payload).unwrap(), - time_ms as i64, - ) - } -} diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs deleted file mode 100644 index 79246da..0000000 --- a/src/pinger_icmp_ts.rs +++ /dev/null @@ -1,132 +0,0 @@ -// SPDX-FileCopyrightText: 2022-Present Charles Corrigan mailto:chas-iot@runegate.org (github @chas-iot) -// SPDX-FileCopyrightText: 2022-Present Daniel Lakeland mailto:dlakelan@street-artists.org (github @dlakelan) -// SPDX-FileCopyrightText: 2022-Present Mark Baker mailto:mark@vpost.net (github @Fail-Safe) -// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) -// -// SPDX-License-Identifier: MPL-2.0 - -use crate::pinger::{PingError, PingListener, PingReply, PingSender}; -use crate::settings::MeasurementType; -use crate::time::{ClockSample, Realtime}; -use icmp_socket2::Icmpv4Message; -use icmp_socket2::Icmpv4Packet; -use icmp_socket2::packet::WithTimestampRequest; -use std::net::IpAddr; -use std::time::Instant; - -pub struct PingerICMPTimestampListener {} - -pub struct PingerICMPTimestampSender {} - -const MS_PER_DAY: i64 = 86_400_000; - -fn timestamp_delta(later: i64, earlier: i64) -> i64 { - let delta = later - earlier; - if delta < -MS_PER_DAY / 2 { - delta + MS_PER_DAY - } else if delta > MS_PER_DAY / 2 { - delta - MS_PER_DAY - } else { - delta - } -} - -impl PingListener for PingerICMPTimestampListener { - // Result: RTT, down time, up time - fn parse_packet( - &self, - id: u16, - reflector: IpAddr, - measurement_type: MeasurementType, - pkt: Icmpv4Packet, - ) -> Result { - match pkt.typ { - // 14 = Timestamp reply - 14 => { - if let Icmpv4Message::TimestampReply { - identifier, - sequence, - originate, - receive, - transmit, - } = pkt.message - { - if identifier != id { - return Err(PingError::WrongID { - expected: id, - found: identifier, - }); - } - - if receive & 0x8000_0000 != 0 || transmit & 0x8000_0000 != 0 { - return Err(PingError::InvalidPacket( - "non-standard RFC 792 timestamp".into(), - )); - } - - let time_since_midnight = - ClockSample::::now().as_time_since_midnight(); - - let rtt = timestamp_delta(time_since_midnight, originate as i64) as f64; - let dl_time = timestamp_delta(time_since_midnight, transmit as i64) as f64; - let ul_time = timestamp_delta(receive as i64, originate as i64) as f64; - - Ok(PingReply { - reflector, - measurement_type, - seq: sequence, - rtt, - current_time: time_since_midnight, - down_time: dl_time, - up_time: ul_time, - originate_timestamp: originate as i64, - receive_timestamp: receive as i64, - transmit_timestamp: transmit as i64, - last_receive_time_s: Instant::now(), - }) - } else { - Err(PingError::InvalidPacket(format!( - "Packet had type {:?}, but did not match the structure", - pkt.typ - ))) - } - } - type_ => Err(PingError::InvalidType(format!("{:?}", type_))), - } - } -} - -impl PingSender for PingerICMPTimestampSender { - fn craft_packet(&self, id: u16, seq: u16) -> (Icmpv4Packet, i64) { - let time_since_midnight = ClockSample::::now().as_time_since_midnight(); - ( - Icmpv4Packet::with_timestamp_request(id, seq, time_since_midnight as u32, 0, 0) - .unwrap(), - time_since_midnight as i64, - ) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use icmp_socket2::packet::WithTimestampReply; - - #[test] - fn timestamps_wrap_at_midnight_in_both_directions() { - assert_eq!(timestamp_delta(20, 86_399_900), 120); - assert_eq!(timestamp_delta(86_399_900, 20), -120); - } - - #[test] - fn non_standard_rfc_792_timestamps_are_rejected() { - let packet = Icmpv4Packet::with_timestamp_reply(7, 1, 10, 0x8000_0001, 20).unwrap(); - let result = PingerICMPTimestampListener {}.parse_packet( - 7, - "192.0.2.1".parse().unwrap(), - MeasurementType::IcmpTimestamps, - packet, - ); - assert!(matches!(result, Err(PingError::InvalidPacket(_)))); - } -} diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index ab78889..92c1227 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -10,7 +10,7 @@ use crate::baseliner::ControlSnapshot; use crate::metrics::{Metric, MetricsSender}; use crate::settings::Settings; use crate::util::RwLockExt; -use flume::{Receiver, RecvError, RecvTimeoutError, Selector}; +use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender}; use log::{debug, info}; use std::net::IpAddr; use std::sync::atomic::Ordering; @@ -108,6 +108,7 @@ pub struct ReflectorSelector { pub settings: Settings, pub snapshot_rx: Receiver, pub reflector_peers_lock: Arc>>, + pub reflector_tx: Sender>, pub reflector_pool: Vec, pub trigger_channel: Receiver, pub metrics: MetricsSender, @@ -186,6 +187,7 @@ impl ReflectorSelector { // Clone next_peers because we need it again after the baseline sleep // to iterate over candidates for RTT measurement. *reflectors_peers = next_peers.clone(); + self.reflector_tx.send(next_peers.clone())?; // Drop the MutexGuard explicitly, as Rust won't unlock the mutex by default // until the guard goes out of scope @@ -251,6 +253,7 @@ impl ReflectorSelector { }); } + self.reflector_tx.send(new_peers.clone())?; *reflectors_peers = new_peers; } } diff --git a/src/settings/mod.rs b/src/settings/mod.rs index 577ee55..06f7c71 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -218,6 +218,13 @@ pub(crate) struct NetworkSettings { pub(crate) upload_base_kbits: f64, #[serde(default = "default_min_percent")] pub(crate) upload_min_percent: f64, + + #[allow(dead_code)] + pub(crate) measurement_bind_device: Option, + #[allow(dead_code)] + pub(crate) measurement_mark: Option, + #[allow(dead_code)] + pub(crate) measurement_fib: Option, } impl NetworkSettings { diff --git a/src/settings/validation.rs b/src/settings/validation.rs index f392b01..14a54cd 100644 --- a/src/settings/validation.rs +++ b/src/settings/validation.rs @@ -399,6 +399,9 @@ mod tests { download_min_percent: 1.0, upload_base_kbits, upload_min_percent: 1.0, + measurement_bind_device: None, + measurement_mark: None, + measurement_fib: None, } } diff --git a/src/time.rs b/src/time.rs index d6da605..13f3b78 100644 --- a/src/time.rs +++ b/src/time.rs @@ -2,7 +2,10 @@ // // SPDX-License-Identifier: MPL-2.0 -use std::marker::PhantomData; +use std::{ + marker::PhantomData, + time::{SystemTime, UNIX_EPOCH}, +}; use rustix::time::{ClockId, clock_gettime}; @@ -64,3 +67,17 @@ impl ClockSample { (self.time_s as i64 % 86_400 * 1000) + self.time_ns as i64 / 1_000_000 } } + +impl From for ClockSample { + fn from(time: SystemTime) -> Self { + let duration = time + .duration_since(UNIX_EPOCH) + .expect("realtime timestamp predates Unix epoch"); + + Self { + time_s: duration.as_secs(), + time_ns: u64::from(duration.subsec_nanos()), + _clock: PhantomData, + } + } +}