From 4acce614fee9eff68def5409290ee1af7f46f2af Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 8 Oct 2026 01:17:56 +0700 Subject: [PATCH] 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(_)))); - } -}