Compare commits
6
Commits
3fc71c050e
...
480976b666
| Author | SHA1 | Date | |
|---|---|---|---|
 Nils Andreas Svee
|
480976b666 | ||
|
Nils Andreas Svee
|
ac349a1897 | ||
|
Nils Andreas Svee
|
a925e60f36 | ||
|
Nils Andreas Svee
|
c7b283d761 | ||
|
Nils Andreas Svee
|
89c7d75947 | ||
|
Nils Andreas Svee
|
afedfbeb65 |
No files matched your search
Generated
+133
@@ -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"
|
||||
|
||||
+7
-6
@@ -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]
|
||||
|
||||
+61
-25
@@ -6,8 +6,8 @@
|
||||
// SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
use crate::SHUTDOWN;
|
||||
use crate::measurement::model::{MeasurementEvent, MeasurementObservation};
|
||||
use crate::metrics::{Metric, MetricsSender};
|
||||
use crate::pinger::PingReply;
|
||||
use crate::settings::Settings;
|
||||
use flume::{Receiver, Sender};
|
||||
use log::{debug, info};
|
||||
@@ -29,10 +29,16 @@ pub struct EwmaStats {
|
||||
}
|
||||
|
||||
impl EwmaStats {
|
||||
fn from_reply(reply: &PingReply) -> Self {
|
||||
fn from_reply(reply: &MeasurementObservation) -> Self {
|
||||
Self {
|
||||
down: reply.down_time,
|
||||
up: reply.up_time,
|
||||
down: reply
|
||||
.one_way
|
||||
.map(|o| o.downlink.as_nanos() as f64 / 1000000.0)
|
||||
.unwrap_or(reply.rtt_ms() / 2.0),
|
||||
up: reply
|
||||
.one_way
|
||||
.map(|o| o.uplink.as_nanos() as f64 / 1000000.0)
|
||||
.unwrap_or(reply.rtt_ms() / 2.0),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -73,11 +79,12 @@ pub struct Baseliner {
|
||||
pub settings: Settings,
|
||||
pub reselect_trigger: Sender<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,
|
||||
pub event_metrics: MetricsSender,
|
||||
pub ping_metrics: MetricsSender,
|
||||
}
|
||||
|
||||
fn ewma_factor(tick: f64, dur: f64) -> f64 {
|
||||
@@ -115,19 +122,19 @@ fn update_reflector_state(
|
||||
|
||||
fn process_reply(
|
||||
reflectors: &mut HashMap<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 +172,14 @@ impl Baseliner {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let reply = self.stats_rx.recv()?;
|
||||
let reflector = reply.reflector;
|
||||
let event = self.stats_rx.recv()?;
|
||||
|
||||
let reply = match event {
|
||||
MeasurementEvent::Observation(reply) => reply,
|
||||
MeasurementEvent::Loss(_) => continue,
|
||||
};
|
||||
|
||||
let reflector = reply.stream.peer;
|
||||
let (state, anomaly) = process_reply(
|
||||
&mut reflectors,
|
||||
&reply,
|
||||
@@ -189,6 +202,20 @@ impl Baseliner {
|
||||
let _ = self.reselect_trigger.try_send(true);
|
||||
}
|
||||
|
||||
self.ping_metrics.send(Metric::Ping {
|
||||
reflector,
|
||||
measurement_type: reply.stream.source,
|
||||
rtt: reply.rtt_ms(),
|
||||
up_time: reply
|
||||
.one_way
|
||||
.map(|o| o.uplink.as_nanos() as f64 / 1000000.0)
|
||||
.unwrap_or(reply.rtt_ms() / 2.0),
|
||||
down_time: reply
|
||||
.one_way
|
||||
.map(|o| o.downlink.as_nanos() as f64 / 1000000.0)
|
||||
.unwrap_or(reply.rtt_ms() / 2.0),
|
||||
});
|
||||
|
||||
self.baseline_metrics.send(Metric::Baseline {
|
||||
reflector,
|
||||
baseline_up_ewma: state.baseline.up,
|
||||
@@ -218,7 +245,10 @@ impl Baseliner {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::settings::MeasurementType;
|
||||
use crate::measurement::model::{
|
||||
MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency,
|
||||
SignedDuration,
|
||||
};
|
||||
use std::time::Duration;
|
||||
|
||||
fn state(
|
||||
@@ -245,19 +275,25 @@ mod tests {
|
||||
EwmaStats { down, up }
|
||||
}
|
||||
|
||||
fn reply(reflector: IpAddr, down_time: f64, up_time: f64, received_at: Instant) -> PingReply {
|
||||
PingReply {
|
||||
reflector,
|
||||
measurement_type: MeasurementType::Icmp,
|
||||
seq: 0,
|
||||
rtt: down_time + up_time,
|
||||
current_time: 0,
|
||||
down_time,
|
||||
up_time,
|
||||
originate_timestamp: 0,
|
||||
receive_timestamp: 0,
|
||||
transmit_timestamp: 0,
|
||||
last_receive_time_s: received_at,
|
||||
fn reply(
|
||||
reflector: IpAddr,
|
||||
down_time: f64,
|
||||
up_time: f64,
|
||||
received_at: Instant,
|
||||
) -> MeasurementObservation {
|
||||
MeasurementObservation {
|
||||
stream: MeasurementStream {
|
||||
id: MeasurementStreamId(0),
|
||||
peer: reflector,
|
||||
source: MeasurementSource::IcmpEcho,
|
||||
},
|
||||
started_at: received_at,
|
||||
observed_at: received_at,
|
||||
one_way: Some(OneWayLatency {
|
||||
uplink: SignedDuration::from_millis(up_time as i64),
|
||||
downlink: SignedDuration::from_millis(down_time as i64),
|
||||
clock: OneWayClock::Uncalibrated,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Vec<IcmpBinding>>),
|
||||
}
|
||||
|
||||
pub struct Controller {
|
||||
reflector_rx: Receiver<Vec<IpAddr>>,
|
||||
icmp_binding_tx: Sender<Vec<IcmpBinding>>,
|
||||
|
||||
source: MeasurementSource,
|
||||
period: Duration,
|
||||
|
||||
next_stream_id: u64,
|
||||
stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>,
|
||||
}
|
||||
|
||||
impl Controller {
|
||||
pub fn new(
|
||||
reflector_rx: Receiver<Vec<IpAddr>>,
|
||||
icmp_binding_tx: Sender<Vec<IcmpBinding>>,
|
||||
) -> 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<IpAddr>) -> 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<IpAddr>) -> 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,
|
||||
)
|
||||
}
|
||||
+56
-157
@@ -8,12 +8,10 @@
|
||||
extern crate core;
|
||||
|
||||
mod baseliner;
|
||||
mod controller;
|
||||
mod log;
|
||||
mod measurement;
|
||||
mod metrics;
|
||||
mod pinger;
|
||||
mod pinger_icmp;
|
||||
mod pinger_icmp_ts;
|
||||
mod platform;
|
||||
mod ratecontroller;
|
||||
mod reflector_selector;
|
||||
@@ -22,36 +20,29 @@ mod time;
|
||||
mod util;
|
||||
|
||||
use crate::baseliner::Baseliner;
|
||||
use crate::controller::Controller;
|
||||
use crate::measurement::icmp::IcmpEngine;
|
||||
use crate::metrics::{Metric, Metrics, MetricsSender};
|
||||
use crate::pinger::{InFlightProbeCache, PingListener, PingSender};
|
||||
use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender};
|
||||
use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender};
|
||||
use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
|
||||
use crate::ratecontroller::Ratecontroller;
|
||||
use crate::reflector_selector::ReflectorSelector;
|
||||
use crate::settings::MeasurementType;
|
||||
use crate::settings::Settings;
|
||||
use crate::util::RwLockExt;
|
||||
use ::log::{info, warn};
|
||||
use flume::RecvTimeoutError;
|
||||
use std::collections::HashMap;
|
||||
use std::net::IpAddr;
|
||||
use std::str::FromStr;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
|
||||
use std::sync::{Arc, Mutex, RwLock};
|
||||
use std::thread::sleep;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::thread;
|
||||
use std::thread::{JoinHandle, sleep};
|
||||
use std::time::{Duration, Instant};
|
||||
use std::{process, thread};
|
||||
|
||||
struct ReflectorSetup {
|
||||
peers: Arc<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) {
|
||||
@@ -59,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,
|
||||
@@ -156,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,
|
||||
@@ -213,12 +158,29 @@ 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),
|
||||
})
|
||||
}
|
||||
|
||||
fn spawn_worker<F>(
|
||||
name: &str,
|
||||
error_tx: &flume::Sender<anyhow::Error>,
|
||||
worker: F,
|
||||
) -> anyhow::Result<JoinHandle<()>>
|
||||
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::Error>) -> anyhow::Result<()> {
|
||||
loop {
|
||||
match error_rx.recv_timeout(Duration::from_secs(1)) {
|
||||
@@ -239,30 +201,47 @@ fn wait_for_exit(error_rx: &flume::Receiver<anyhow::Error>) -> anyhow::Result<()
|
||||
}
|
||||
|
||||
fn run(settings: &Settings) -> anyhow::Result<()> {
|
||||
let rt = tokio::runtime::Runtime::new()?;
|
||||
|
||||
if settings.advanced_settings.dry_run {
|
||||
info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***");
|
||||
}
|
||||
|
||||
let start_time = Instant::now();
|
||||
let probe_id = probe_identifier();
|
||||
|
||||
let ReflectorSetup {
|
||||
peers: reflector_peers,
|
||||
pool: reflector_pool,
|
||||
reselection_enabled,
|
||||
active_count: active_reflector_count,
|
||||
} = setup_reflectors(settings)?;
|
||||
|
||||
let (baseliner_stats_tx, baseliner_stats_rx) = flume::unbounded();
|
||||
let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded();
|
||||
let (error_tx, error_rx) = flume::unbounded::<anyhow::Error>();
|
||||
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::<anyhow::Error>();
|
||||
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));
|
||||
|
||||
@@ -273,14 +252,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)
|
||||
@@ -310,82 +282,22 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
|
||||
},
|
||||
});
|
||||
|
||||
let (mut ping_listener, mut ping_sender) =
|
||||
create_pinger(settings.advanced_settings.measurement_type);
|
||||
|
||||
let inflight_probe_capacity =
|
||||
compute_inflight_probe_capacity(settings, active_reflector_count, reselection_enabled);
|
||||
info!(
|
||||
"In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})",
|
||||
inflight_probe_capacity,
|
||||
active_reflector_count,
|
||||
reselection_enabled,
|
||||
settings.advanced_settings.tick_interval
|
||||
);
|
||||
|
||||
let inflight: InFlightProbeCache =
|
||||
Arc::new(Mutex::new(HashMap::with_capacity(inflight_probe_capacity)));
|
||||
|
||||
let baseliner = Baseliner {
|
||||
settings: settings.clone(),
|
||||
reselect_trigger: reselect_tx.clone(),
|
||||
start_time,
|
||||
stats_rx: baseliner_stats_rx,
|
||||
stats_rx: measurement_rx,
|
||||
control_tx: control_snapshot_tx,
|
||||
selection_tx: selection_snapshot_tx,
|
||||
baseline_metrics,
|
||||
event_metrics: event_metrics.clone(),
|
||||
ping_metrics,
|
||||
};
|
||||
|
||||
let mut main_traffic_control = traffic_control_backend();
|
||||
let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?;
|
||||
|
||||
let err_tx = error_tx.clone();
|
||||
let listener_peers = Arc::clone(&reflector_peers);
|
||||
let listener_inflight = Arc::clone(&inflight);
|
||||
let measurement_type = settings.advanced_settings.measurement_type;
|
||||
thread::Builder::new()
|
||||
.name("receiver".to_string())
|
||||
.spawn(move || {
|
||||
if let Err(error) = ping_listener.listen(
|
||||
probe_id,
|
||||
measurement_type,
|
||||
listener_peers,
|
||||
listener_inflight,
|
||||
baseliner_stats_tx,
|
||||
ping_metrics,
|
||||
) {
|
||||
let _ = err_tx.send(error);
|
||||
}
|
||||
})?;
|
||||
|
||||
let err_tx = error_tx.clone();
|
||||
thread::Builder::new()
|
||||
.name("baseliner".to_string())
|
||||
.spawn(move || {
|
||||
if let Err(error) = baseliner.run() {
|
||||
let _ = err_tx.send(error);
|
||||
}
|
||||
})?;
|
||||
|
||||
let err_tx = error_tx.clone();
|
||||
let sender_peers = Arc::clone(&reflector_peers);
|
||||
let sender_inflight = Arc::clone(&inflight);
|
||||
let measurement_type = settings.advanced_settings.measurement_type;
|
||||
let tick_interval = settings.advanced_settings.tick_interval;
|
||||
thread::Builder::new()
|
||||
.name("sender".to_string())
|
||||
.spawn(move || {
|
||||
if let Err(error) = ping_sender.send(
|
||||
probe_id,
|
||||
measurement_type,
|
||||
sender_peers,
|
||||
sender_inflight,
|
||||
tick_interval,
|
||||
) {
|
||||
let _ = err_tx.send(error);
|
||||
}
|
||||
})?;
|
||||
spawn_worker("baseliner", &error_tx, move || baseliner.run())?;
|
||||
|
||||
let main_event_metrics = event_metrics.clone();
|
||||
|
||||
@@ -398,15 +310,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.
|
||||
@@ -420,14 +326,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);
|
||||
|
||||
@@ -0,0 +1,505 @@
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
io,
|
||||
net::IpAddr,
|
||||
time::{Duration, Instant, SystemTime},
|
||||
};
|
||||
|
||||
use flume::{Receiver, Sender};
|
||||
use icmp_socket2::{
|
||||
AsyncIcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet,
|
||||
packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest},
|
||||
tokio::AsyncIcmpV4Socket,
|
||||
};
|
||||
use log::debug;
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::{
|
||||
measurement::model::{
|
||||
MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementSource,
|
||||
MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule,
|
||||
SignedDuration,
|
||||
},
|
||||
settings::NetworkSettings,
|
||||
time::{ClockSample, Realtime},
|
||||
};
|
||||
|
||||
pub struct IcmpBinding {
|
||||
pub stream: MeasurementStream,
|
||||
pub schedule: ProbeSchedule,
|
||||
|
||||
next_probe_at: Instant,
|
||||
next_sequence: u16,
|
||||
}
|
||||
|
||||
impl IcmpBinding {
|
||||
pub fn new(stream: MeasurementStream, schedule: ProbeSchedule) -> Self {
|
||||
Self {
|
||||
stream,
|
||||
schedule,
|
||||
next_probe_at: Instant::now(),
|
||||
next_sequence: 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum IcmpError {
|
||||
#[error("I/O error")]
|
||||
Io(#[from] io::Error),
|
||||
|
||||
#[error("failed to build ICMP packet")]
|
||||
PacketBuild(#[from] IcmpPacketBuildError),
|
||||
|
||||
#[error("invalid ICMP packet: {0}")]
|
||||
InvalidPacket(String),
|
||||
|
||||
#[error("unsupported measurement source: {0:?}")]
|
||||
UnsupportedSource(MeasurementSource),
|
||||
|
||||
#[error("probe interval must be non-zero for stream {0:?}")]
|
||||
InvalidInterval(MeasurementStreamId),
|
||||
|
||||
#[error("probe offset must be less than the period for stream {0:?}")]
|
||||
InvalidOffset(MeasurementStreamId),
|
||||
|
||||
#[error(
|
||||
"no matching probe found for peer {:?}, source: {:?}, sequence: {}",
|
||||
.0.0,
|
||||
.0.1,
|
||||
.0.2
|
||||
)]
|
||||
NoMatchingProbe(InFlightProbeKey),
|
||||
|
||||
#[error("wrong identifier: expected {expected}, found {found}")]
|
||||
WrongIdentifier { expected: u16, found: u16 },
|
||||
|
||||
#[error("unsupported ICMP peer address: {0}")]
|
||||
UnsupportedPeer(IpAddr),
|
||||
|
||||
#[error("event channel closed")]
|
||||
EventChannelClosed,
|
||||
|
||||
#[error("binding channel closed")]
|
||||
BindingChannelClosed,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
struct InFlightProbe {
|
||||
stream: MeasurementStream,
|
||||
sent_at: Instant,
|
||||
}
|
||||
|
||||
type InFlightProbeKey = (IpAddr, MeasurementSource, u16);
|
||||
type InFlightProbes = HashMap<InFlightProbeKey, InFlightProbe>;
|
||||
|
||||
pub struct IcmpEngine {
|
||||
socket: AsyncIcmpV4Socket,
|
||||
|
||||
identifier: u16,
|
||||
|
||||
bindings: HashMap<MeasurementStreamId, IcmpBinding>,
|
||||
inflight: HashMap<InFlightProbeKey, InFlightProbe>,
|
||||
|
||||
origin: Instant,
|
||||
|
||||
binding_rx: Receiver<Vec<IcmpBinding>>,
|
||||
observation_tx: Sender<MeasurementEvent>,
|
||||
}
|
||||
|
||||
async fn wait_for_deadline(deadline: Option<Instant>) {
|
||||
match deadline {
|
||||
Some(deadline) => {
|
||||
tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
|
||||
}
|
||||
|
||||
None => {
|
||||
std::future::pending::<()>().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl IcmpEngine {
|
||||
pub fn new(
|
||||
origin: Instant,
|
||||
#[allow(unused)] settings: &NetworkSettings,
|
||||
observation_tx: Sender<MeasurementEvent>,
|
||||
binding_rx: Receiver<Vec<IcmpBinding>>,
|
||||
) -> Result<Self, IcmpError> {
|
||||
#[allow(unused_mut)]
|
||||
let mut socket = IcmpSocket4::new()?;
|
||||
|
||||
#[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
|
||||
if let Some(device) = &settings.measurement_bind_device {
|
||||
icmp_socket2::IcmpSocket::bind_device(&mut socket, device)?;
|
||||
}
|
||||
|
||||
#[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
|
||||
if let Some(mark) = settings.measurement_mark {
|
||||
icmp_socket2::IcmpSocket::set_mark(&mut socket, mark)?;
|
||||
}
|
||||
|
||||
#[cfg(target_os = "freebsd")]
|
||||
if let Some(fib) = settings.measurement_fib {
|
||||
icmp_socket2::IcmpSocket::set_fib(&mut socket, fib)?;
|
||||
}
|
||||
|
||||
socket.enable_receive_metadata()?;
|
||||
let identifier = (std::process::id() & 0xffff) as u16;
|
||||
let inflight: InFlightProbes = HashMap::new();
|
||||
|
||||
Ok(Self {
|
||||
socket: socket.into_tokio()?,
|
||||
identifier,
|
||||
bindings: HashMap::new(),
|
||||
inflight,
|
||||
origin,
|
||||
binding_rx,
|
||||
observation_tx,
|
||||
})
|
||||
}
|
||||
|
||||
fn set_bindings(&mut self, desired: Vec<IcmpBinding>) -> 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::<HashSet<_>>();
|
||||
|
||||
// Remove bindings that are no longer desired.
|
||||
self.bindings.retain(|id, _| desired_ids.contains(id));
|
||||
self.inflight
|
||||
.retain(|_, probe| desired_ids.contains(&probe.stream.id));
|
||||
|
||||
for desired in desired {
|
||||
match self.bindings.get_mut(&desired.stream.id) {
|
||||
Some(current) if current.schedule == desired.schedule => {
|
||||
// Same schedule, so leave next_probe_at alone.
|
||||
current.stream = desired.stream;
|
||||
}
|
||||
|
||||
Some(current) => {
|
||||
// Same stream, but its schedule changed. Rejoin the new grid.
|
||||
current.stream = desired.stream;
|
||||
current.schedule = desired.schedule;
|
||||
current.next_probe_at = desired.schedule.next_probe_at(self.origin, now);
|
||||
}
|
||||
|
||||
None => {
|
||||
// New stream. Join its configured schedule at the next
|
||||
// occurrence after now.
|
||||
let mut binding = desired;
|
||||
binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now);
|
||||
|
||||
self.bindings.insert(binding.stream.id, binding);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> {
|
||||
let due = self
|
||||
.bindings
|
||||
.iter()
|
||||
.filter_map(|(&stream_id, binding)| (binding.next_probe_at <= now).then_some(stream_id))
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
for stream_id in due {
|
||||
self.send_probe(stream_id).await?;
|
||||
|
||||
let binding = self
|
||||
.bindings
|
||||
.get_mut(&stream_id)
|
||||
.expect("due binding disappeared");
|
||||
|
||||
binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> {
|
||||
let (stream, sequence) = {
|
||||
let binding = self
|
||||
.bindings
|
||||
.get_mut(&stream_id)
|
||||
.expect("binding disappeared");
|
||||
|
||||
let sequence = binding.next_sequence;
|
||||
binding.next_sequence = binding.next_sequence.wrapping_add(1);
|
||||
|
||||
(binding.stream, sequence)
|
||||
};
|
||||
|
||||
let IpAddr::V4(peer) = stream.peer else {
|
||||
return Err(IcmpError::UnsupportedPeer(stream.peer));
|
||||
};
|
||||
|
||||
let packet = match stream.source {
|
||||
MeasurementSource::IcmpEcho => {
|
||||
Icmpv4Packet::with_echo_request(self.identifier, sequence, Vec::new())?
|
||||
}
|
||||
MeasurementSource::IcmpTimestamp => {
|
||||
let originate = ClockSample::<Realtime>::now().as_time_since_midnight() as u32;
|
||||
Icmpv4Packet::with_timestamp_request(self.identifier, sequence, originate, 0, 0)?
|
||||
}
|
||||
source => return Err(IcmpError::UnsupportedSource(source)),
|
||||
};
|
||||
|
||||
let sent_at = Instant::now();
|
||||
let key = (stream.peer, stream.source, sequence);
|
||||
|
||||
self.socket.send_to(peer, packet).await?;
|
||||
self.inflight.insert(key, InFlightProbe { stream, sent_at });
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn next_deadline(&self) -> Option<Instant> {
|
||||
self.bindings
|
||||
.values()
|
||||
.map(|binding| binding.next_probe_at)
|
||||
.min()
|
||||
}
|
||||
|
||||
fn handle_reply(
|
||||
&mut self,
|
||||
packet: Icmpv4Packet,
|
||||
peer: IpAddr,
|
||||
observed_at: Instant,
|
||||
kernel_rx_timestamp: Option<SystemTime>,
|
||||
) -> Result<MeasurementObservation, IcmpError> {
|
||||
match packet.message {
|
||||
Icmpv4Message::EchoReply {
|
||||
identifier,
|
||||
sequence,
|
||||
..
|
||||
} => {
|
||||
if identifier != self.identifier {
|
||||
return Err(IcmpError::WrongIdentifier {
|
||||
expected: self.identifier,
|
||||
found: identifier,
|
||||
});
|
||||
}
|
||||
|
||||
let key = (peer, MeasurementSource::IcmpEcho, sequence);
|
||||
let probe = self
|
||||
.inflight
|
||||
.remove(&key)
|
||||
.ok_or(IcmpError::NoMatchingProbe(key))?;
|
||||
|
||||
debug!(
|
||||
"ICMP engine > Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:>8} | Received time: {:>8} | Transmit time: {:>8} | RTT: {:8.3} ms | UL time: {:>8} | DL time: {:>8}",
|
||||
"ICMP",
|
||||
peer,
|
||||
sequence,
|
||||
ClockSample::<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,
|
||||
observed_at,
|
||||
one_way: None,
|
||||
})
|
||||
}
|
||||
Icmpv4Message::TimestampReply {
|
||||
identifier,
|
||||
sequence,
|
||||
originate,
|
||||
receive,
|
||||
transmit,
|
||||
} => {
|
||||
if identifier != self.identifier {
|
||||
return Err(IcmpError::WrongIdentifier {
|
||||
expected: self.identifier,
|
||||
found: identifier,
|
||||
});
|
||||
}
|
||||
|
||||
if originate >= MS_PER_DAY as u32
|
||||
|| receive >= MS_PER_DAY as u32
|
||||
|| transmit >= MS_PER_DAY as u32
|
||||
{
|
||||
return Err(IcmpError::InvalidPacket(
|
||||
"non-standard RFC 792 timestamp".into(),
|
||||
));
|
||||
}
|
||||
|
||||
let key = (peer, MeasurementSource::IcmpTimestamp, sequence);
|
||||
let probe = self
|
||||
.inflight
|
||||
.remove(&key)
|
||||
.ok_or(IcmpError::NoMatchingProbe(key))?;
|
||||
|
||||
let now_ms = if let Some(timestamp) = kernel_rx_timestamp {
|
||||
ClockSample::<Realtime>::from(timestamp).as_time_since_midnight()
|
||||
} else {
|
||||
ClockSample::<Realtime>::now().as_time_since_midnight()
|
||||
};
|
||||
|
||||
let uplink_ms = timestamp_delta(receive as i64, originate as i64);
|
||||
let downlink_ms = timestamp_delta(now_ms, transmit as i64);
|
||||
|
||||
debug!(
|
||||
"ICMP engine > Type: {:4} | Reflector IP: {:>15} | Seq: {:5} | Current time: {:8} | Originate: {:8} | Received time: {:8} | Transmit time: {:8} | RTT: {:8.3} ms | UL time: {:5} ms | DL time: {:5} ms",
|
||||
"ICMP",
|
||||
peer,
|
||||
sequence,
|
||||
now_ms,
|
||||
originate,
|
||||
receive,
|
||||
transmit,
|
||||
observed_at.duration_since(probe.sent_at).as_secs_f64() * 1_000.0,
|
||||
uplink_ms,
|
||||
downlink_ms,
|
||||
);
|
||||
|
||||
Ok(MeasurementObservation {
|
||||
stream: probe.stream,
|
||||
started_at: probe.sent_at,
|
||||
observed_at,
|
||||
one_way: Some(OneWayLatency {
|
||||
uplink: SignedDuration::from_millis(uplink_ms),
|
||||
downlink: SignedDuration::from_millis(downlink_ms),
|
||||
clock: OneWayClock::Uncalibrated,
|
||||
}),
|
||||
})
|
||||
}
|
||||
_ => Err(IcmpError::InvalidPacket(
|
||||
"unsupported ICMP message type".to_string(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> {
|
||||
const PROBE_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
let expired = {
|
||||
let mut expired = Vec::new();
|
||||
|
||||
self.inflight.retain(|_, probe| {
|
||||
if now.saturating_duration_since(probe.sent_at) >= PROBE_TIMEOUT {
|
||||
expired.push(*probe);
|
||||
false
|
||||
} else {
|
||||
true
|
||||
}
|
||||
});
|
||||
|
||||
expired
|
||||
};
|
||||
|
||||
for probe in expired {
|
||||
self.observation_tx
|
||||
.send_async(MeasurementEvent::Loss(MeasurementLoss {
|
||||
stream: probe.stream,
|
||||
started_at: probe.sent_at,
|
||||
detected_at: now,
|
||||
}))
|
||||
.await
|
||||
.map_err(|_| IcmpError::EventChannelClosed)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn run(mut self) -> Result<(), IcmpError> {
|
||||
const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500);
|
||||
|
||||
let mut housekeeping = tokio::time::interval_at(
|
||||
tokio::time::Instant::now() + HOUSEKEEPING_INTERVAL,
|
||||
HOUSEKEEPING_INTERVAL,
|
||||
);
|
||||
|
||||
housekeeping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
|
||||
loop {
|
||||
let deadline = self.next_deadline();
|
||||
|
||||
tokio::select! {
|
||||
bindings = self.binding_rx.recv_async() => {
|
||||
let bindings = bindings
|
||||
.map_err(|_| IcmpError::BindingChannelClosed)?;
|
||||
|
||||
self.set_bindings(bindings)?;
|
||||
}
|
||||
|
||||
_ = wait_for_deadline(deadline) => {
|
||||
self.send_due_probes(Instant::now()).await?;
|
||||
}
|
||||
|
||||
result = self.socket.rcv_from_with_meta() => {
|
||||
let result = result?;
|
||||
|
||||
match self.handle_reply(
|
||||
result.packet,
|
||||
result.peer.as_socket().unwrap().ip(),
|
||||
result.received_at,
|
||||
result.kernel_rx_timestamp,
|
||||
) {
|
||||
Ok(observation) => {
|
||||
self.observation_tx
|
||||
.send_async(MeasurementEvent::Observation(observation))
|
||||
.await
|
||||
.map_err(|_| IcmpError::EventChannelClosed)?;
|
||||
}
|
||||
|
||||
Err(
|
||||
IcmpError::WrongIdentifier { .. }
|
||||
| IcmpError::NoMatchingProbe { .. }
|
||||
| IcmpError::InvalidPacket(_),
|
||||
) => {
|
||||
// Not ours, malformed, or late.
|
||||
}
|
||||
|
||||
Err(err) => return Err(err),
|
||||
}
|
||||
}
|
||||
|
||||
_ = housekeeping.tick() => {
|
||||
self.expire_inflight(Instant::now()).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const MS_PER_DAY: i64 = 86_400_000;
|
||||
|
||||
fn timestamp_delta(later: i64, earlier: i64) -> i64 {
|
||||
let delta = later - earlier;
|
||||
if delta < -MS_PER_DAY / 2 {
|
||||
delta + MS_PER_DAY
|
||||
} else if delta > MS_PER_DAY / 2 {
|
||||
delta - MS_PER_DAY
|
||||
} else {
|
||||
delta
|
||||
}
|
||||
}
|
||||
@@ -1,2 +1,2 @@
|
||||
mod icmp;
|
||||
mod model;
|
||||
pub(crate) mod icmp;
|
||||
pub(crate) mod model;
|
||||
@@ -1,3 +1,4 @@
|
||||
use std::fmt::{self, Display};
|
||||
use std::{
|
||||
net::IpAddr,
|
||||
time::{Duration, Instant},
|
||||
@@ -6,6 +7,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,
|
||||
@@ -15,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,
|
||||
@@ -22,17 +36,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 }
|
||||
@@ -84,12 +135,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),
|
||||
|
||||
+3
-2
@@ -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,
|
||||
|
||||
-214
@@ -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);
|
||||
}
|
||||
@@ -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,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -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(_))));
|
||||
}
|
||||
}
|
||||
@@ -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<ControlSnapshot>,
|
||||
pub reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
|
||||
pub reflector_tx: Sender<Vec<IpAddr>>,
|
||||
pub reflector_pool: Vec<IpAddr>,
|
||||
pub trigger_channel: Receiver<bool>,
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String>,
|
||||
#[allow(dead_code)]
|
||||
pub(crate) measurement_mark: Option<u32>,
|
||||
#[allow(dead_code)]
|
||||
pub(crate) measurement_fib: Option<u32>,
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user