From 3016f4b88781d1d312b06f559d3bea3de5049ede Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 30 Sep 2026 15:34:54 +0700 Subject: [PATCH 01/11] feat(settings): add network settings for measurement sockets --- src/settings/mod.rs | 4 ++++ src/settings/validation.rs | 3 +++ 2 files changed, 7 insertions(+) diff --git a/src/settings/mod.rs b/src/settings/mod.rs index 577ee55..ecc1331 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -218,6 +218,10 @@ pub(crate) struct NetworkSettings { pub(crate) upload_base_kbits: f64, #[serde(default = "default_min_percent")] pub(crate) upload_min_percent: f64, + + pub(crate) measurement_bind_device: Option, + pub(crate) measurement_mark: Option, + 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, } } From 60fdea4dff438dc03c07105f1404d2ad2af9a9a5 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 30 Sep 2026 15:34:54 +0700 Subject: [PATCH 02/11] refactor(time): implement ClockSample from SystemTime --- src/time.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) 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, + } + } +} From f51ff19f0484510b418e0d0c8f148925e129cabd Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 30 Sep 2026 15:34:54 +0700 Subject: [PATCH 03/11] feat(measurement): add core data models --- src/main.rs | 1 + src/measurement/mod.rs | 2 + src/measurement/model.rs | 96 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 99 insertions(+) create mode 100644 src/measurement/mod.rs create mode 100644 src/measurement/model.rs diff --git a/src/main.rs b/src/main.rs index ad6c000..a6f92a6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ extern crate core; mod baseliner; mod log; +mod measurement; mod metrics; mod pinger; mod pinger_icmp; diff --git a/src/measurement/mod.rs b/src/measurement/mod.rs new file mode 100644 index 0000000..e152830 --- /dev/null +++ b/src/measurement/mod.rs @@ -0,0 +1,2 @@ +mod icmp; +mod model; diff --git a/src/measurement/model.rs b/src/measurement/model.rs new file mode 100644 index 0000000..2d8d959 --- /dev/null +++ b/src/measurement/model.rs @@ -0,0 +1,96 @@ +use std::{ + net::IpAddr, + time::{Duration, Instant}, +}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct MeasurementStreamId(pub u64); + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum MeasurementSource { + IcmpEcho, + IcmpTimestamp, + Irtt, + TcpSeqAck, + TcpTimestamp, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct MeasurementStream { + pub id: MeasurementStreamId, + pub peer: IpAddr, + pub source: MeasurementSource, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OneWayClock { + Uncalibrated, + Calibrated, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct SignedDuration { + nanos: i64, +} + +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 + } +} + +pub struct MeasurementLoss { + pub stream: MeasurementStream, + pub started_at: Instant, + pub detected_at: Instant, +} + +pub enum MeasurementEvent { + Observation(MeasurementObservation), + Loss(MeasurementLoss), +} From afedfbeb65156b927a552fb1c32bc76c7ad85f21 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 30 Sep 2026 15:34:54 +0700 Subject: [PATCH 04/11] feat(measurement): implement ICMP measurement engine --- src/main.rs | 22 +- src/measurement/icmp.rs | 543 +++++++++++++++++++++++++++++++++++++++ src/measurement/mod.rs | 4 +- src/measurement/model.rs | 42 +++ src/settings/mod.rs | 3 + 5 files changed, 611 insertions(+), 3 deletions(-) create mode 100644 src/measurement/icmp.rs diff --git a/src/main.rs b/src/main.rs index a6f92a6..2c15b45 100644 --- a/src/main.rs +++ b/src/main.rs @@ -22,6 +22,10 @@ mod time; mod util; use crate::baseliner::Baseliner; +use crate::measurement::icmp::{IcmpBinding, IcmpEngine}; +use crate::measurement::model::{ + MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule, +}; use crate::metrics::{Metric, Metrics, MetricsSender}; use crate::pinger::{InFlightProbeCache, PingListener, PingSender}; use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender}; @@ -34,7 +38,7 @@ use crate::settings::Settings; use ::log::{info, warn}; use flume::RecvTimeoutError; use std::collections::HashMap; -use std::net::IpAddr; +use std::net::{IpAddr, Ipv4Addr}; use std::str::FromStr; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, Mutex, RwLock}; @@ -246,6 +250,22 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let start_time = Instant::now(); let probe_id = probe_identifier(); + let (ie_txt, _) = flume::unbounded(); + let (binding_tx, binding_rx) = flume::unbounded(); + let engine = IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?; + binding_tx.send(vec![IcmpBinding::new( + MeasurementStream { + id: MeasurementStreamId(0), + peer: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), + source: MeasurementSource::IcmpEcho, + }, + ProbeSchedule { + period: Duration::from_secs(1), + offset: Duration::from_secs(0), + }, + )])?; + engine.run()?; + let ReflectorSetup { peers: reflector_peers, pool: reflector_pool, diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs new file mode 100644 index 0000000..80d2099 --- /dev/null +++ b/src/measurement/icmp.rs @@ -0,0 +1,543 @@ +use std::{ + collections::{HashMap, HashSet}, + io, + net::IpAddr, + sync::{Arc, Mutex, atomic::Ordering}, + thread, + time::{Duration, Instant, SystemTime}, +}; + +use flume::{Receiver, RecvTimeoutError, Sender}; +use icmp_socket2::{ + IcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, + packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest}, +}; +use thiserror::Error; + +use crate::{ + SHUTDOWN, + 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, +} + +impl IcmpBinding { + pub fn new(stream: MeasurementStream, schedule: ProbeSchedule) -> Self { + Self { + stream, + schedule, + next_probe_at: Instant::now(), + } + } +} + +#[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, +} + +#[derive(Debug, Clone, Copy)] +struct InFlightProbe { + stream: MeasurementStream, + sent_at: Instant, +} + +type InFlightProbeKey = (IpAddr, MeasurementSource, u16); +type InFlightProbes = Mutex>; + +struct IcmpSender { + socket: IcmpSocket4, + identifier: u16, + next_sequence: u16, + bindings: HashMap, + inflight: Arc, + origin: Instant, + binding_rx: Receiver>, +} + +struct IcmpReceiver { + socket: IcmpSocket4, + identifier: u16, + inflight: Arc, + observation_tx: Sender, +} + +pub struct IcmpEngine { + sender: IcmpSender, + receiver: IcmpReceiver, +} + +impl IcmpEngine { + pub fn new( + origin: Instant, + #[allow(unused)] settings: &NetworkSettings, + observation_tx: Sender, + binding_rx: Receiver>, + ) -> Result { + #[allow(unused_mut)] + let mut rx_socket = IcmpSocket4::new()?; + + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + if let Some(device) = &settings.measurement_bind_device { + IcmpSocket::bind_device(&mut rx_socket, device)?; + } + + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + if let Some(mark) = settings.measurement_mark { + IcmpSocket::set_mark(&mut rx_socket, mark)?; + } + + #[cfg(target_os = "freebsd")] + if let Some(fib) = settings.measurement_fib { + IcmpSocket::set_fib(&mut rx_socket, fib)?; + } + + rx_socket.enable_receive_metadata()?; + let tx_socket = rx_socket.try_clone()?; + let identifier = (std::process::id() & 0xffff) as u16; + let inflight: Arc = Arc::new(Mutex::new(HashMap::new())); + + Ok(Self { + sender: IcmpSender { + socket: tx_socket, + identifier, + next_sequence: 0, + bindings: HashMap::new(), + inflight: Arc::clone(&inflight), + origin, + binding_rx, + }, + receiver: IcmpReceiver { + socket: rx_socket, + identifier, + inflight, + observation_tx, + }, + }) + } + + pub fn run(self) -> Result<(), IcmpError> { + let Self { sender, receiver } = self; + + let sender_handle = thread::spawn(move || { + let result = sender.run(); + SHUTDOWN.store(true, Ordering::Relaxed); + result + }); + let sender_thread = sender_handle.thread().clone(); + + let receiver_handle = thread::spawn(move || { + let result = receiver.run(); + SHUTDOWN.store(true, Ordering::Relaxed); + sender_thread.unpark(); + result + }); + + let sender_result = sender_handle.join().expect("ICMP sender thread panicked"); + + SHUTDOWN.store(true, Ordering::Relaxed); + + let receiver_result = receiver_handle + .join() + .expect("ICMP receiver thread panicked"); + + sender_result?; + receiver_result?; + + Ok(()) + } +} + +impl IcmpSender { + 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)); + + 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(()) + } + + fn run(mut self) -> Result<(), IcmpError> { + while !SHUTDOWN.load(Ordering::Relaxed) { + let deadline = self.next_deadline(); + + match deadline { + Some(deadline) => match self.binding_rx.recv_deadline(deadline) { + Ok(bindings) => { + self.set_bindings(bindings)?; + } + + Err(RecvTimeoutError::Timeout) => { + self.send_due_probes(Instant::now())?; + } + + Err(RecvTimeoutError::Disconnected) => { + break; + } + }, + + None => { + // Nothing scheduled. Don't exit. + // Just sleep until configuration arrives. + let bindings = self + .binding_rx + .recv() + .map_err(|_| IcmpError::EventChannelClosed)?; + self.set_bindings(bindings)?; + } + } + } + + Ok(()) + } + + 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)?; + + let binding = self + .bindings + .get_mut(&stream_id) + .expect("due binding disappeared"); + + while binding.next_probe_at <= now { + binding.next_probe_at += binding.schedule.period; + } + } + + Ok(()) + } + + fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> { + let stream = self + .bindings + .get(&stream_id) + .expect("binding disappeared") + .stream; + + let IpAddr::V4(peer) = stream.peer else { + return Err(IcmpError::UnsupportedPeer(stream.peer)); + }; + + let sequence = self.next_sequence; + self.next_sequence = self.next_sequence.wrapping_add(1); + + 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); + let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); + inflight.insert(key, InFlightProbe { stream, sent_at }); + + if let Err(err) = self.socket.send_to(peer, packet) { + return Err(err.into()); + } + + Ok(()) + } + + fn next_deadline(&self) -> Option { + self.bindings + .values() + .map(|binding| binding.next_probe_at) + .min() + } +} + +impl IcmpReceiver { + fn run(mut self) -> Result<(), IcmpError> { + const RECEIVE_TIMEOUT: Duration = Duration::from_millis(50); + const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500); + + self.socket.set_timeout(Some(RECEIVE_TIMEOUT)); + + let mut next_housekeeping = Instant::now() + HOUSEKEEPING_INTERVAL; + + while !SHUTDOWN.load(Ordering::Relaxed) { + match self.socket.rcv_from_with_meta() { + Ok(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(MeasurementEvent::Observation(observation)) + .map_err(|_| IcmpError::EventChannelClosed)?; + } + Err( + IcmpError::WrongIdentifier { .. } + | IcmpError::NoMatchingProbe { .. } + | IcmpError::InvalidPacket(_), + ) => { + // Not one of ours, malformed, or a late reply. + } + Err(err) => return Err(err), + } + } + Err(err) + if matches!( + err.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ) => {} + Err(err) => return Err(err.into()), + } + + let now = Instant::now(); + + if now >= next_housekeeping { + self.expire_inflight(now)?; + next_housekeeping = now + HOUSEKEEPING_INTERVAL; + } + } + + Ok(()) + } + + 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 = { + let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); + + inflight + .remove(&key) + .ok_or(IcmpError::NoMatchingProbe(key))? + }; + + Ok(MeasurementObservation { + stream: probe.stream, + started_at: probe.sent_at, + observed_at, + one_way: None, + }) + } + Icmpv4Message::TimestampReply { + identifier, + sequence, + originate, + receive, + transmit, + } => { + const NON_STANDARD_TIMESTAMP: u32 = 1 << 31; + + if identifier != self.identifier { + return Err(IcmpError::WrongIdentifier { + expected: self.identifier, + found: identifier, + }); + } + + if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 { + return Err(IcmpError::InvalidPacket( + "non-standard RFC 792 timestamp".into(), + )); + } + + let key = (peer, MeasurementSource::IcmpTimestamp, sequence); + let probe = { + let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); + + 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); + + 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(), + )), + } + } + + fn expire_inflight(&self, now: Instant) -> Result<(), IcmpError> { + const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + + let expired = { + let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); + let mut expired = Vec::new(); + + 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(MeasurementEvent::Loss(MeasurementLoss { + stream: probe.stream, + started_at: probe.sent_at, + detected_at: now, + })) + .map_err(|_| IcmpError::EventChannelClosed)?; + } + + Ok(()) + } +} + +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 index e152830..0ee60f2 100644 --- a/src/measurement/mod.rs +++ b/src/measurement/mod.rs @@ -1,2 +1,2 @@ -mod icmp; -mod model; +pub(crate) mod icmp; +pub(crate) mod model; diff --git a/src/measurement/model.rs b/src/measurement/model.rs index 2d8d959..1bf59cd 100644 --- a/src/measurement/model.rs +++ b/src/measurement/model.rs @@ -6,6 +6,7 @@ use std::{ #[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, @@ -22,17 +23,54 @@ pub struct MeasurementStream { 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 } @@ -60,6 +98,7 @@ pub struct OneWayLatency { pub clock: OneWayClock, } +#[allow(dead_code)] #[derive(Debug, Clone, Copy)] pub struct MeasurementObservation { pub stream: MeasurementStream, @@ -74,6 +113,7 @@ pub struct MeasurementObservation { pub one_way: Option, } +#[allow(dead_code)] impl MeasurementObservation { pub fn rtt(&self) -> Duration { self.observed_at.duration_since(self.started_at) @@ -84,12 +124,14 @@ impl MeasurementObservation { } } +#[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/settings/mod.rs b/src/settings/mod.rs index ecc1331..06f7c71 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -219,8 +219,11 @@ pub(crate) struct NetworkSettings { #[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, } From 89c7d75947b818ff7a2b0ff9f3d5bee070c53029 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 5 Oct 2026 01:15:19 +0700 Subject: [PATCH 05/11] refactor(measurement): migrate ICMP engine to Tokio async runtime --- Cargo.lock | 133 ++++++++++++++++++ Cargo.toml | 13 +- src/main.rs | 4 +- src/measurement/icmp.rs | 301 +++++++++++++++------------------------- 4 files changed, 257 insertions(+), 194 deletions(-) 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/main.rs b/src/main.rs index 2c15b45..40b0281 100644 --- a/src/main.rs +++ b/src/main.rs @@ -264,7 +264,9 @@ fn run(settings: &Settings) -> anyhow::Result<()> { offset: Duration::from_secs(0), }, )])?; - engine.run()?; + + let rt = tokio::runtime::Runtime::new().unwrap(); + rt.spawn(engine.run()); let ReflectorSetup { peers: reflector_peers, diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs index 80d2099..22a6d49 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -2,20 +2,18 @@ use std::{ collections::{HashMap, HashSet}, io, net::IpAddr, - sync::{Arc, Mutex, atomic::Ordering}, - thread, time::{Duration, Instant, SystemTime}, }; -use flume::{Receiver, RecvTimeoutError, Sender}; +use flume::{Receiver, Sender}; use icmp_socket2::{ - IcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, + AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest}, + tokio::AsyncIcmpV4Socket, }; use thiserror::Error; use crate::{ - SHUTDOWN, measurement::model::{ MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule, @@ -78,6 +76,9 @@ pub enum IcmpError { #[error("event channel closed")] EventChannelClosed, + + #[error("binding channel closed")] + BindingChannelClosed, } #[derive(Debug, Clone, Copy)] @@ -87,28 +88,33 @@ struct InFlightProbe { } type InFlightProbeKey = (IpAddr, MeasurementSource, u16); -type InFlightProbes = Mutex>; +type InFlightProbes = HashMap; + +pub struct IcmpEngine { + socket: AsyncIcmpV4Socket, -struct IcmpSender { - socket: IcmpSocket4, identifier: u16, next_sequence: u16, - bindings: HashMap, - inflight: Arc, - origin: Instant, - binding_rx: Receiver>, -} -struct IcmpReceiver { - socket: IcmpSocket4, - identifier: u16, - inflight: Arc, + bindings: HashMap, + inflight: HashMap, + + origin: Instant, + + binding_rx: Receiver>, observation_tx: Sender, } -pub struct IcmpEngine { - sender: IcmpSender, - receiver: IcmpReceiver, +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 { @@ -119,80 +125,39 @@ impl IcmpEngine { binding_rx: Receiver>, ) -> Result { #[allow(unused_mut)] - let mut rx_socket = IcmpSocket4::new()?; + let mut socket = IcmpSocket4::new()?; #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] if let Some(device) = &settings.measurement_bind_device { - IcmpSocket::bind_device(&mut rx_socket, 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 { - IcmpSocket::set_mark(&mut rx_socket, mark)?; + icmp_socket2::IcmpSocket::set_mark(&mut socket, mark)?; } #[cfg(target_os = "freebsd")] if let Some(fib) = settings.measurement_fib { - IcmpSocket::set_fib(&mut rx_socket, fib)?; + icmp_socket2::IcmpSocket::set_fib(&mut socket, fib)?; } - rx_socket.enable_receive_metadata()?; - let tx_socket = rx_socket.try_clone()?; + socket.enable_receive_metadata()?; let identifier = (std::process::id() & 0xffff) as u16; - let inflight: Arc = Arc::new(Mutex::new(HashMap::new())); + let inflight: InFlightProbes = HashMap::new(); Ok(Self { - sender: IcmpSender { - socket: tx_socket, - identifier, - next_sequence: 0, - bindings: HashMap::new(), - inflight: Arc::clone(&inflight), - origin, - binding_rx, - }, - receiver: IcmpReceiver { - socket: rx_socket, - identifier, - inflight, - observation_tx, - }, + socket: socket.into_tokio()?, + identifier, + next_sequence: 0, + bindings: HashMap::new(), + inflight, + origin, + binding_rx, + observation_tx, }) } - pub fn run(self) -> Result<(), IcmpError> { - let Self { sender, receiver } = self; - - let sender_handle = thread::spawn(move || { - let result = sender.run(); - SHUTDOWN.store(true, Ordering::Relaxed); - result - }); - let sender_thread = sender_handle.thread().clone(); - - let receiver_handle = thread::spawn(move || { - let result = receiver.run(); - SHUTDOWN.store(true, Ordering::Relaxed); - sender_thread.unpark(); - result - }); - - let sender_result = sender_handle.join().expect("ICMP sender thread panicked"); - - SHUTDOWN.store(true, Ordering::Relaxed); - - let receiver_result = receiver_handle - .join() - .expect("ICMP receiver thread panicked"); - - sender_result?; - receiver_result?; - - Ok(()) - } -} - -impl IcmpSender { fn set_bindings(&mut self, desired: Vec) -> Result<(), IcmpError> { // Validate the complete desired state before mutating anything. for binding in &desired { @@ -249,41 +214,7 @@ impl IcmpSender { Ok(()) } - fn run(mut self) -> Result<(), IcmpError> { - while !SHUTDOWN.load(Ordering::Relaxed) { - let deadline = self.next_deadline(); - - match deadline { - Some(deadline) => match self.binding_rx.recv_deadline(deadline) { - Ok(bindings) => { - self.set_bindings(bindings)?; - } - - Err(RecvTimeoutError::Timeout) => { - self.send_due_probes(Instant::now())?; - } - - Err(RecvTimeoutError::Disconnected) => { - break; - } - }, - - None => { - // Nothing scheduled. Don't exit. - // Just sleep until configuration arrives. - let bindings = self - .binding_rx - .recv() - .map_err(|_| IcmpError::EventChannelClosed)?; - self.set_bindings(bindings)?; - } - } - } - - Ok(()) - } - - fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> { + async fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> { let due = self .bindings .iter() @@ -291,7 +222,7 @@ impl IcmpSender { .collect::>(); for stream_id in due { - self.send_probe(stream_id)?; + self.send_probe(stream_id).await?; let binding = self .bindings @@ -306,7 +237,7 @@ impl IcmpSender { Ok(()) } - fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> { + async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> { let stream = self .bindings .get(&stream_id) @@ -333,12 +264,9 @@ impl IcmpSender { let sent_at = Instant::now(); let key = (stream.peer, stream.source, sequence); - let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); - inflight.insert(key, InFlightProbe { stream, sent_at }); - if let Err(err) = self.socket.send_to(peer, packet) { - return Err(err.into()); - } + self.socket.send_to(peer, packet).await?; + self.inflight.insert(key, InFlightProbe { stream, sent_at }); Ok(()) } @@ -349,59 +277,6 @@ impl IcmpSender { .map(|binding| binding.next_probe_at) .min() } -} - -impl IcmpReceiver { - fn run(mut self) -> Result<(), IcmpError> { - const RECEIVE_TIMEOUT: Duration = Duration::from_millis(50); - const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500); - - self.socket.set_timeout(Some(RECEIVE_TIMEOUT)); - - let mut next_housekeeping = Instant::now() + HOUSEKEEPING_INTERVAL; - - while !SHUTDOWN.load(Ordering::Relaxed) { - match self.socket.rcv_from_with_meta() { - Ok(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(MeasurementEvent::Observation(observation)) - .map_err(|_| IcmpError::EventChannelClosed)?; - } - Err( - IcmpError::WrongIdentifier { .. } - | IcmpError::NoMatchingProbe { .. } - | IcmpError::InvalidPacket(_), - ) => { - // Not one of ours, malformed, or a late reply. - } - Err(err) => return Err(err), - } - } - Err(err) - if matches!( - err.kind(), - io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut - ) => {} - Err(err) => return Err(err.into()), - } - - let now = Instant::now(); - - if now >= next_housekeeping { - self.expire_inflight(now)?; - next_housekeeping = now + HOUSEKEEPING_INTERVAL; - } - } - - Ok(()) - } fn handle_reply( &mut self, @@ -424,13 +299,10 @@ impl IcmpReceiver { } let key = (peer, MeasurementSource::IcmpEcho, sequence); - let probe = { - let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); - - inflight - .remove(&key) - .ok_or(IcmpError::NoMatchingProbe(key))? - }; + let probe = self + .inflight + .remove(&key) + .ok_or(IcmpError::NoMatchingProbe(key))?; Ok(MeasurementObservation { stream: probe.stream, @@ -462,13 +334,10 @@ impl IcmpReceiver { } let key = (peer, MeasurementSource::IcmpTimestamp, sequence); - let probe = { - let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); - - inflight - .remove(&key) - .ok_or(IcmpError::NoMatchingProbe(key))? - }; + 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() @@ -496,14 +365,13 @@ impl IcmpReceiver { } } - fn expire_inflight(&self, now: Instant) -> Result<(), IcmpError> { + fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> { const PROBE_TIMEOUT: Duration = Duration::from_secs(2); let expired = { - let mut inflight = self.inflight.lock().expect("ICMP inflight mutex poisoned"); let mut expired = Vec::new(); - inflight.retain(|_, probe| { + self.inflight.retain(|_, probe| { if now.saturating_duration_since(probe.sent_at) >= PROBE_TIMEOUT { expired.push(*probe); false @@ -527,6 +395,65 @@ impl IcmpReceiver { 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(MeasurementEvent::Observation(observation)) + .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())?; + } + } + } + } } const MS_PER_DAY: i64 = 86_400_000; From c7b283d7617d0cfcb08389ff9f3cb3f006e0fa1f Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 5 Oct 2026 01:47:03 +0700 Subject: [PATCH 06/11] refactor(measurement): enhance ICMP engine's async integration and robustness Ensure IcmpEngine initializes within a Tokio runtime, improve timestamp validation, and refine probe management. --- src/main.rs | 8 ++++++-- src/measurement/icmp.rs | 23 ++++++++++++----------- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/src/main.rs b/src/main.rs index 40b0281..5bc51c7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -243,6 +243,8 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } fn run(settings: &Settings) -> anyhow::Result<()> { + let rt = tokio::runtime::Runtime::new().unwrap(); + if settings.advanced_settings.dry_run { info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***"); } @@ -252,7 +254,10 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let (ie_txt, _) = flume::unbounded(); let (binding_tx, binding_rx) = flume::unbounded(); - let engine = IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?; + let engine = { + let _guard = rt.enter(); + IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)? + }; binding_tx.send(vec![IcmpBinding::new( MeasurementStream { id: MeasurementStreamId(0), @@ -265,7 +270,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> { }, )])?; - let rt = tokio::runtime::Runtime::new().unwrap(); rt.spawn(engine.run()); let ReflectorSetup { diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs index 22a6d49..cc2b59d 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -229,9 +229,7 @@ impl IcmpEngine { .get_mut(&stream_id) .expect("due binding disappeared"); - while binding.next_probe_at <= now { - binding.next_probe_at += binding.schedule.period; - } + binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now); } Ok(()) @@ -318,8 +316,6 @@ impl IcmpEngine { receive, transmit, } => { - const NON_STANDARD_TIMESTAMP: u32 = 1 << 31; - if identifier != self.identifier { return Err(IcmpError::WrongIdentifier { expected: self.identifier, @@ -327,7 +323,10 @@ impl IcmpEngine { }); } - if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 { + 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(), )); @@ -365,8 +364,8 @@ impl IcmpEngine { } } - fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> { - const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + 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(); @@ -385,11 +384,12 @@ impl IcmpEngine { for probe in expired { self.observation_tx - .send(MeasurementEvent::Loss(MeasurementLoss { + .send_async(MeasurementEvent::Loss(MeasurementLoss { stream: probe.stream, started_at: probe.sent_at, detected_at: now, })) + .await .map_err(|_| IcmpError::EventChannelClosed)?; } @@ -432,7 +432,8 @@ impl IcmpEngine { ) { Ok(observation) => { self.observation_tx - .send(MeasurementEvent::Observation(observation)) + .send_async(MeasurementEvent::Observation(observation)) + .await .map_err(|_| IcmpError::EventChannelClosed)?; } @@ -449,7 +450,7 @@ impl IcmpEngine { } _ = housekeeping.tick() => { - self.expire_inflight(Instant::now())?; + self.expire_inflight(Instant::now()).await?; } } } From a925e60f361ea77329433637be4846f7b4559264 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Tue, 6 Oct 2026 01:14:32 +0700 Subject: [PATCH 07/11] refactor(measurement): introduce Controller for dynamic ICMP probe configuration --- src/controller.rs | 102 +++++++++++++++++++++++++ src/main.rs | 154 +++++++++++++++++--------------------- src/reflector_selector.rs | 5 +- 3 files changed, 175 insertions(+), 86 deletions(-) create mode 100644 src/controller.rs 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 5bc51c7..fdf34b0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,7 @@ extern crate core; mod baseliner; +mod controller; mod log; mod measurement; mod metrics; @@ -22,6 +23,7 @@ mod time; mod util; use crate::baseliner::Baseliner; +use crate::controller::Controller; use crate::measurement::icmp::{IcmpBinding, IcmpEngine}; use crate::measurement::model::{ MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule, @@ -35,6 +37,7 @@ 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; @@ -42,7 +45,7 @@ use std::net::{IpAddr, Ipv4Addr}; use std::str::FromStr; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, Mutex, RwLock}; -use std::thread::sleep; +use std::thread::{JoinHandle, sleep}; use std::time::{Duration, Instant}; use std::{process, thread}; @@ -223,6 +226,26 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result { }) } +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)) { @@ -243,7 +266,7 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } fn run(settings: &Settings) -> anyhow::Result<()> { - let rt = tokio::runtime::Runtime::new().unwrap(); + let rt = tokio::runtime::Runtime::new()?; if settings.advanced_settings.dry_run { info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***"); @@ -252,26 +275,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let start_time = Instant::now(); let probe_id = probe_identifier(); - let (ie_txt, _) = flume::unbounded(); - let (binding_tx, binding_rx) = flume::unbounded(); - let engine = { - let _guard = rt.enter(); - IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)? - }; - binding_tx.send(vec![IcmpBinding::new( - MeasurementStream { - id: MeasurementStreamId(0), - peer: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), - source: MeasurementSource::IcmpEcho, - }, - ProbeSchedule { - period: Duration::from_secs(1), - offset: Duration::from_secs(0), - }, - )])?; - - rt.spawn(engine.run()); - let ReflectorSetup { peers: reflector_peers, pool: reflector_pool, @@ -281,14 +284,32 @@ fn run(settings: &Settings) -> anyhow::Result<()> { 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 engine = { + let _guard = rt.enter(); + IcmpEngine::new( + start_time, + &settings.network, + measurement_tx, + icmp_binding_rx, + )? + }; + + rt.spawn(engine.run()); + + let controller = Controller::new(reflector_rx, icmp_binding_tx); + rt.spawn(controller.run(reflector_peers.read_anyhow()?.clone())); let dropped = Arc::new(AtomicU32::new(0)); @@ -299,14 +320,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) @@ -366,52 +380,35 @@ fn run(settings: &Settings) -> anyhow::Result<()> { 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); - } - })?; + spawn_worker("receiver", &error_tx, move || { + ping_listener.listen( + probe_id, + measurement_type, + listener_peers, + listener_inflight, + baseliner_stats_tx, + ping_metrics, + ) + })?; - 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); - } - })?; + spawn_worker("baseliner", &error_tx, move || baseliner.run())?; - 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("sender", &error_tx, move || { + ping_sender.send( + probe_id, + measurement_type, + sender_peers, + sender_inflight, + tick_interval, + ) + })?; let main_event_metrics = event_metrics.clone(); @@ -424,15 +421,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. @@ -446,14 +437,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/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; } } From ac349a189770bb01c32a60349fe35a73fcda58b8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 8 Oct 2026 01:17:56 +0700 Subject: [PATCH 08/11] refactor(measurement): remove legacy pinger and migrate production to --- src/baseliner.rs | 71 ++++++++----- src/main.rs | 124 ++--------------------- src/measurement/icmp.rs | 54 ++++++++-- src/measurement/model.rs | 2 - src/pinger.rs | 214 --------------------------------------- src/pinger_icmp.rs | 96 ------------------ src/pinger_icmp_ts.rs | 132 ------------------------ 7 files changed, 96 insertions(+), 597 deletions(-) delete mode 100644 src/pinger.rs delete mode 100644 src/pinger_icmp.rs delete mode 100644 src/pinger_icmp_ts.rs diff --git a/src/baseliner.rs b/src/baseliner.rs index 1f05116..b16ea63 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,7 +79,7 @@ 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, @@ -115,19 +121,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 +171,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, @@ -218,7 +230,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 +260,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/main.rs b/src/main.rs index fdf34b0..1059ae8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -12,9 +12,6 @@ mod controller; mod log; mod measurement; mod metrics; -mod pinger; -mod pinger_icmp; -mod pinger_icmp_ts; mod platform; mod ratecontroller; mod reflector_selector; @@ -24,41 +21,28 @@ mod util; use crate::baseliner::Baseliner; use crate::controller::Controller; -use crate::measurement::icmp::{IcmpBinding, IcmpEngine}; -use crate::measurement::model::{ - MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule, -}; +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, Ipv4Addr}; +use std::net::IpAddr; use std::str::FromStr; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; -use std::sync::{Arc, Mutex, RwLock}; +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) { @@ -66,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, @@ -163,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, @@ -220,9 +158,6 @@ 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), }) } @@ -273,16 +208,13 @@ fn run(settings: &Settings) -> anyhow::Result<()> { } 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(); @@ -334,7 +266,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { .unwrap_or_else(MetricsSender::disabled) }; - let ping_metrics = make_sender(settings.observability.export_ping_metrics); + let _ping_metrics = make_sender(settings.observability.export_ping_metrics); let baseline_metrics = make_sender(settings.observability.export_baseline_metrics); let event_metrics = make_sender(settings.observability.export_events); let rate_metrics = make_sender(settings.observability.export_rate_metrics); @@ -350,27 +282,11 @@ 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, @@ -380,36 +296,8 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let mut main_traffic_control = traffic_control_backend(); let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?; - let listener_peers = Arc::clone(&reflector_peers); - let listener_inflight = Arc::clone(&inflight); - let measurement_type = settings.advanced_settings.measurement_type; - spawn_worker("receiver", &error_tx, move || { - ping_listener.listen( - probe_id, - measurement_type, - listener_peers, - listener_inflight, - baseliner_stats_tx, - ping_metrics, - ) - })?; - spawn_worker("baseliner", &error_tx, move || baseliner.run())?; - 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; - spawn_worker("sender", &error_tx, move || { - ping_sender.send( - probe_id, - measurement_type, - sender_peers, - sender_inflight, - tick_interval, - ) - })?; - let main_event_metrics = event_metrics.clone(); if reselection_enabled { diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs index cc2b59d..3d96b3b 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -11,6 +11,7 @@ use icmp_socket2::{ packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest}, tokio::AsyncIcmpV4Socket, }; +use log::debug; use thiserror::Error; use crate::{ @@ -28,6 +29,7 @@ pub struct IcmpBinding { pub schedule: ProbeSchedule, next_probe_at: Instant, + next_sequence: u16, } impl IcmpBinding { @@ -36,6 +38,7 @@ impl IcmpBinding { stream, schedule, next_probe_at: Instant::now(), + next_sequence: 0, } } } @@ -94,7 +97,6 @@ pub struct IcmpEngine { socket: AsyncIcmpV4Socket, identifier: u16, - next_sequence: u16, bindings: HashMap, inflight: HashMap, @@ -149,7 +151,6 @@ impl IcmpEngine { Ok(Self { socket: socket.into_tokio()?, identifier, - next_sequence: 0, bindings: HashMap::new(), inflight, origin, @@ -185,6 +186,8 @@ impl IcmpEngine { // 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) { @@ -236,19 +239,22 @@ impl IcmpEngine { } async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> { - let stream = self - .bindings - .get(&stream_id) - .expect("binding disappeared") - .stream; + 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 sequence = self.next_sequence; - self.next_sequence = self.next_sequence.wrapping_add(1); - let packet = match stream.source { MeasurementSource::IcmpEcho => { Icmpv4Packet::with_echo_request(self.identifier, sequence, Vec::new())? @@ -302,6 +308,20 @@ impl IcmpEngine { .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, @@ -347,6 +367,20 @@ impl IcmpEngine { 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, diff --git a/src/measurement/model.rs b/src/measurement/model.rs index 1bf59cd..a835c63 100644 --- a/src/measurement/model.rs +++ b/src/measurement/model.rs @@ -98,7 +98,6 @@ pub struct OneWayLatency { pub clock: OneWayClock, } -#[allow(dead_code)] #[derive(Debug, Clone, Copy)] pub struct MeasurementObservation { pub stream: MeasurementStream, @@ -113,7 +112,6 @@ pub struct MeasurementObservation { pub one_way: Option, } -#[allow(dead_code)] impl MeasurementObservation { pub fn rtt(&self) -> Duration { self.observed_at.duration_since(self.started_at) 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(_)))); - } -} From 480976b6668658cd5ef9254aa4f06bc3390d2748 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 8 Oct 2026 11:21:26 +0700 Subject: [PATCH 09/11] fix(metrics): move ping metrics sending to baseliner --- src/baseliner.rs | 15 +++++++++++++++ src/main.rs | 3 ++- src/measurement/model.rs | 13 +++++++++++++ src/metrics.rs | 5 +++-- 4 files changed, 33 insertions(+), 3 deletions(-) diff --git a/src/baseliner.rs b/src/baseliner.rs index b16ea63..4908c43 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -84,6 +84,7 @@ pub struct Baseliner { pub selection_tx: Option>, pub baseline_metrics: MetricsSender, pub event_metrics: MetricsSender, + pub ping_metrics: MetricsSender, } fn ewma_factor(tick: f64, dur: f64) -> f64 { @@ -201,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, diff --git a/src/main.rs b/src/main.rs index 1059ae8..19bbabf 100644 --- a/src/main.rs +++ b/src/main.rs @@ -266,7 +266,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { .unwrap_or_else(MetricsSender::disabled) }; - let _ping_metrics = make_sender(settings.observability.export_ping_metrics); + let ping_metrics = make_sender(settings.observability.export_ping_metrics); let baseline_metrics = make_sender(settings.observability.export_baseline_metrics); let event_metrics = make_sender(settings.observability.export_events); let rate_metrics = make_sender(settings.observability.export_rate_metrics); @@ -291,6 +291,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { selection_tx: selection_snapshot_tx, baseline_metrics, event_metrics: event_metrics.clone(), + ping_metrics, }; let mut main_traffic_control = traffic_control_backend(); diff --git a/src/measurement/model.rs b/src/measurement/model.rs index a835c63..12b7bf5 100644 --- a/src/measurement/model.rs +++ b/src/measurement/model.rs @@ -1,3 +1,4 @@ +use std::fmt::{self, Display}; use std::{ net::IpAddr, time::{Duration, Instant}, @@ -16,6 +17,18 @@ pub enum MeasurementSource { 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, 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, From e80940ba6d01e494279899c55e8643d74d5302a1 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 8 Oct 2026 14:28:12 +0700 Subject: [PATCH 10/11] refactor(main): add spawn task helper --- src/main.rs | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/src/main.rs b/src/main.rs index 19bbabf..537ec77 100644 --- a/src/main.rs +++ b/src/main.rs @@ -161,6 +161,24 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result { }) } +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, @@ -185,7 +203,7 @@ 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(()); @@ -228,6 +246,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { (None, None) }; + let controller = Controller::new(reflector_rx, icmp_binding_tx); let engine = { let _guard = rt.enter(); IcmpEngine::new( @@ -238,10 +257,12 @@ fn run(settings: &Settings) -> anyhow::Result<()> { )? }; - rt.spawn(engine.run()); - - let controller = Controller::new(reflector_rx, icmp_binding_tx); - rt.spawn(controller.run(reflector_peers.read_anyhow()?.clone())); + spawn_task(&rt, &error_tx, engine.run()); + spawn_task( + &rt, + &error_tx, + controller.run(reflector_peers.read_anyhow()?.clone()), + ); let dropped = Arc::new(AtomicU32::new(0)); From 0e2e2f9d38a125c65e5eac667dcc1ab870663a04 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 8 Oct 2026 14:31:59 +0700 Subject: [PATCH 11/11] fix: start controller after the baseliner --- src/main.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/main.rs b/src/main.rs index 537ec77..560cab9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -258,11 +258,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> { }; spawn_task(&rt, &error_tx, engine.run()); - spawn_task( - &rt, - &error_tx, - controller.run(reflector_peers.read_anyhow()?.clone()), - ); let dropped = Arc::new(AtomicU32::new(0)); @@ -319,6 +314,11 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?; 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();