Lochnair-Labs/sqm-autorate-rust
2
0 Fork 0
Code Issues 5 Pull Requests Actions 14 Packages Projects Releases Wiki Activity

refactor(measurement): remove legacy pinger and migrate production to

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2026-10-08 14:13:59 +07:00
1 parent f9b0d9f7ab
commit a97fdebb83
7 files changed
+95 -596
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/baseliner.rs
+46 -25
View File
@@ -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<bool>,
pub start_time: Instant,
pub stats_rx: Receiver<PingReply>,
pub stats_rx: Receiver<MeasurementEvent>,
pub control_tx: Sender<ControlSnapshot>,
pub selection_tx: Option<Sender<ControlSnapshot>>,
pub baseline_metrics: MetricsSender,
@@ -115,19 +121,19 @@ fn update_reflector_state(
fn process_reply(
reflectors: &mut HashMap<IpAddr, ReflectorState>,
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,
}),
}
}
src/main.rs
+6 -118
View File
@@ -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<RwLock<Vec<IpAddr>>>,
pool: Vec<IpAddr>,
reselection_enabled: bool,
active_count: usize,
}
type PingListenerBox = Box<dyn PingListener + Send>;
type PingSenderBox = Box<dyn PingSender + Send>;
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<T: TrafficControlBackend>(
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<T: TrafficControlBackend>(
settings: &Settings,
traffic_control: &mut T,
@@ -220,9 +158,6 @@ fn setup_reflectors(settings: &Settings) -> anyhow::Result<ReflectorSetup> {
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::<anyhow::Error>();
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 {
src/measurement/icmp.rs
+43 -9
View File
@@ -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<MeasurementStreamId, IcmpBinding>,
inflight: HashMap<InFlightProbeKey, InFlightProbe>,
@@ -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
let (stream, sequence) = {
let binding = self
.bindings
.get(&stream_id)
.expect("binding disappeared")
.stream;
.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::<Realtime>::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,
src/measurement/model.rs
-2
View File
@@ -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<OneWayLatency>,
}
#[allow(dead_code)]
impl MeasurementObservation {
pub fn rtt(&self) -> Duration {
self.observed_at.duration_since(self.started_at)
src/pinger.rs
-214
View File
@@ -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<Mutex<HashMap<InFlightProbeKey, InFlightProbe>>>;
const INFLIGHT_PROBE_TTL: Duration = Duration::from_secs(30);
fn open_socket(type_: MeasurementType) -> io::Result<IcmpSocket4> {
match type_ {
MeasurementType::Icmp | MeasurementType::IcmpTimestamps => IcmpSocket4::new(),
}
}
pub trait PingListener {
fn listen(
&mut self,
id: u16,
type_: MeasurementType,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
inflight: InFlightProbeCache,
stats_tx: Sender<PingReply>,
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<PingReply, PingError>;
}
pub trait PingSender {
fn send(
&mut self,
id: u16,
type_: MeasurementType,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
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);
}
src/pinger_icmp.rs
-96
View File
@@ -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<PingReply, PingError> {
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::<Monotonic>::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::<Monotonic>::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,
)
}
}
src/pinger_icmp_ts.rs
-132
View File
@@ -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<PingReply, PingError> {
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::<Realtime>::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::<Realtime>::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(_))));
}
}