refactor(measurement): migrate ICMP engine to Tokio async runtime

Cargo.lock

@@ -17,6 +17,17 @@ version = "1.0.104"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
 
+[[package]]
+name = "async-trait"
+version = "0.1.92"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.5.1"
@@ -55,6 +66,12 @@ version = "3.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
 
+[[package]]
+name = "bytes"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
+
 [[package]]
 name = "cc"
 version = "1.3.0"
@@ -183,9 +200,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
 dependencies = [
  "fastrand",
+ "futures-core",
+ "futures-sink",
  "spin",
 ]
 
+[[package]]
+name = "futures-core"
+version = "0.3.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e"
+
+[[package]]
+name = "futures-sink"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
+
 [[package]]
 name = "getrandom"
 version = "0.4.3"
@@ -217,9 +248,11 @@ version = "0.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "816dc56a4f7f76e42b0616818600b05528cd346cb3843276e82df982d5c3ee26"
 dependencies = [
+ "async-trait",
  "nix",
  "socket2",
  "thiserror",
+ "tokio",
 ]
 
 [[package]]
@@ -325,6 +358,17 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
+[[package]]
+name = "mio"
+version = "1.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee"
+dependencies = [
+ "libc",
+ "wasi",
+ "windows-sys",
+]
+
 [[package]]
 name = "netlink-bindings"
 version = "0.3.5"
@@ -371,12 +415,41 @@ version = "1.21.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
 
+[[package]]
+name = "parking_lot"
+version = "0.12.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a"
+dependencies = [
+ "lock_api",
+ "parking_lot_core",
+]
+
+[[package]]
+name = "parking_lot_core"
+version = "0.9.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "redox_syscall",
+ "smallvec",
+ "windows-link",
+]
+
 [[package]]
 name = "pathdiff"
 version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "df94ce210e5bc13cb6651479fa48d14f601d9858cfe0467f43ae157023b938d3"
 
+[[package]]
+name = "pin-project-lite"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+
 [[package]]
 name = "prettyplease"
 version = "0.2.37"
@@ -411,6 +484,15 @@ version = "6.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
+[[package]]
+name = "redox_syscall"
+version = "0.5.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
+dependencies = [
+ "bitflags",
+]
+
 [[package]]
 name = "regex"
 version = "1.13.1"
@@ -552,6 +634,22 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
 
+[[package]]
+name = "signal-hook-registry"
+version = "1.4.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b"
+dependencies = [
+ "errno",
+ "libc",
+]
+
+[[package]]
+name = "smallvec"
+version = "1.16.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
+
 [[package]]
 name = "socket2"
 version = "0.6.5"
@@ -589,6 +687,7 @@ dependencies = [
  "rustix",
  "serde",
  "thiserror",
+ "tokio",
  "uci-derive",
 ]
 
@@ -655,6 +754,34 @@ dependencies = [
  "syn 3.0.6",
 ]
 
+[[package]]
+name = "tokio"
+version = "1.53.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044"
+dependencies = [
+ "bytes",
+ "libc",
+ "mio",
+ "parking_lot",
+ "pin-project-lite",
+ "signal-hook-registry",
+ "socket2",
+ "tokio-macros",
+ "windows-sys",
+]
+
+[[package]]
+name = "tokio-macros"
+version = "2.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
 [[package]]
 name = "toml"
 version = "1.1.3+spec-1.1.0"
@@ -725,6 +852,12 @@ version = "1.0.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
 
+[[package]]
+name = "wasi"
+version = "0.11.1+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
+
 [[package]]
 name = "wasm-bindgen"
 version = "0.2.126"

Cargo.toml

@@ -20,15 +20,16 @@ strip = "symbols"
 
 [dependencies]
 anyhow = "1.0"
-icmp-socket2 = "0.6.4"
-fastrand = "2"
-rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
-thiserror = "2.0"
-libc = "0.2"
-flume = { version = "0.12.0",  default-features = false, features = ["eventual-fairness", "select"] }
 config = { version = "0.15.25", default-features = false }
 csv = "1.4.0"
+fastrand = "2"
+flume = { version = "0.12.0",  default-features = false, features = ["async", "eventual-fairness", "select"] }
+icmp-socket2 = { version = "0.6.4", features = ["tokio"] }
+libc = "0.2"
+rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] }
 serde = { version = "1", features = ["derive"] }
+thiserror = "2.0"
+tokio = { version = "1", features = ["full"] }
 uci-derive = { path = "crates/uci-derive" }
 
 [dependencies.log]

src/main.rs

@@ -9,6 +9,7 @@ extern crate core;
 
 mod baseliner;
 mod log;
+mod measurement;
 mod metrics;
 mod pinger;
 mod pinger_icmp;
@@ -21,6 +22,10 @@ mod time;
 mod util;
 
 use crate::baseliner::Baseliner;
+use crate::measurement::icmp::{IcmpBinding, IcmpEngine};
+use crate::measurement::model::{
+    MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule,
+};
 use crate::metrics::{Metric, Metrics, MetricsSender};
 use crate::pinger::{InFlightProbeCache, PingListener, PingSender};
 use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender};
@@ -33,7 +38,7 @@ use crate::settings::Settings;
 use ::log::{info, warn};
 use flume::RecvTimeoutError;
 use std::collections::HashMap;
-use std::net::IpAddr;
+use std::net::{IpAddr, Ipv4Addr};
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
@@ -245,6 +250,24 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let start_time = Instant::now();
     let probe_id = probe_identifier();
 
+    let (ie_txt, _) = flume::unbounded();
+    let (binding_tx, binding_rx) = flume::unbounded();
+    let engine = IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?;
+    binding_tx.send(vec![IcmpBinding::new(
+        MeasurementStream {
+            id: MeasurementStreamId(0),
+            peer: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
+            source: MeasurementSource::IcmpEcho,
+        },
+        ProbeSchedule {
+            period: Duration::from_secs(1),
+            offset: Duration::from_secs(0),
+        },
+    )])?;
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    rt.spawn(engine.run());
+
     let ReflectorSetup {
         peers: reflector_peers,
         pool: reflector_pool,

src/measurement/icmp.rs

@@ -0,0 +1,459 @@
+use std::{
+    collections::{HashMap, HashSet},
+    io,
+    net::IpAddr,
+    sync::{Arc, Mutex, atomic::Ordering},
+    thread,
+    time::{Duration, Instant, SystemTime},
+};
+
+use flume::{Receiver, RecvTimeoutError, Sender};
+use icmp_socket2::{
+    AsyncIcmpSocket, IcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet,
+    packet::{IcmpPacketBuildError, WithEchoRequest, WithTimestampRequest},
+    tokio::AsyncIcmpV4Socket,
+};
+use thiserror::Error;
+
+use crate::{
+    SHUTDOWN,
+    measurement::model::{
+        MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementSource,
+        MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, ProbeSchedule,
+        SignedDuration,
+    },
+    settings::NetworkSettings,
+    time::{ClockSample, Realtime},
+};
+
+pub struct IcmpBinding {
+    pub stream: MeasurementStream,
+    pub schedule: ProbeSchedule,
+
+    next_probe_at: Instant,
+}
+
+impl IcmpBinding {
+    pub fn new(stream: MeasurementStream, schedule: ProbeSchedule) -> Self {
+        Self {
+            stream,
+            schedule,
+            next_probe_at: Instant::now(),
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum IcmpError {
+    #[error("I/O error")]
+    Io(#[from] io::Error),
+
+    #[error("failed to build ICMP packet")]
+    PacketBuild(#[from] IcmpPacketBuildError),
+
+    #[error("invalid ICMP packet: {0}")]
+    InvalidPacket(String),
+
+    #[error("unsupported measurement source: {0:?}")]
+    UnsupportedSource(MeasurementSource),
+
+    #[error("probe interval must be non-zero for stream {0:?}")]
+    InvalidInterval(MeasurementStreamId),
+
+    #[error("probe offset must be less than the period for stream {0:?}")]
+    InvalidOffset(MeasurementStreamId),
+
+    #[error(
+        "no matching probe found for peer {:?}, source: {:?}, sequence: {}",
+        .0.0,
+        .0.1,
+        .0.2
+    )]
+    NoMatchingProbe(InFlightProbeKey),
+
+    #[error("wrong identifier: expected {expected}, found {found}")]
+    WrongIdentifier { expected: u16, found: u16 },
+
+    #[error("unsupported ICMP peer address: {0}")]
+    UnsupportedPeer(IpAddr),
+
+    #[error("event channel closed")]
+    EventChannelClosed,
+
+    #[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,
+    next_sequence: 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 {
+            IcmpSocket::bind_device(&mut socket, device)?;
+        }
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(mark) = settings.measurement_mark {
+            IcmpSocket::set_mark(&mut socket, mark)?;
+        }
+
+        #[cfg(target_os = "freebsd")]
+        if let Some(fib) = settings.measurement_fib {
+            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,
+            next_sequence: 0,
+            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));
+
+        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");
+
+            while binding.next_probe_at <= now {
+                binding.next_probe_at += binding.schedule.period;
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn send_probe(&mut self, stream_id: MeasurementStreamId) -> Result<(), IcmpError> {
+        let stream = self
+            .bindings
+            .get(&stream_id)
+            .expect("binding disappeared")
+            .stream;
+
+        let IpAddr::V4(peer) = stream.peer else {
+            return Err(IcmpError::UnsupportedPeer(stream.peer));
+        };
+
+        let sequence = self.next_sequence;
+        self.next_sequence = self.next_sequence.wrapping_add(1);
+
+        let packet = match stream.source {
+            MeasurementSource::IcmpEcho => {
+                Icmpv4Packet::with_echo_request(self.identifier, sequence, Vec::new())?
+            }
+            MeasurementSource::IcmpTimestamp => {
+                let originate = ClockSample::<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.inflight.insert(key, InFlightProbe { stream, sent_at });
+
+        if let Err(err) = self.socket.send_to(peer, packet).await {
+            return Err(err.into());
+        }
+
+        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))?;
+
+                Ok(MeasurementObservation {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    observed_at,
+                    one_way: None,
+                })
+            }
+            Icmpv4Message::TimestampReply {
+                identifier,
+                sequence,
+                originate,
+                receive,
+                transmit,
+            } => {
+                const NON_STANDARD_TIMESTAMP: u32 = 1 << 31;
+
+                if identifier != self.identifier {
+                    return Err(IcmpError::WrongIdentifier {
+                        expected: self.identifier,
+                        found: identifier,
+                    });
+                }
+
+                if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 {
+                    return Err(IcmpError::InvalidPacket(
+                        "non-standard RFC 792 timestamp".into(),
+                    ));
+                }
+
+                let key = (peer, MeasurementSource::IcmpTimestamp, sequence);
+                let probe = 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);
+
+                Ok(MeasurementObservation {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    observed_at,
+                    one_way: Some(OneWayLatency {
+                        uplink: SignedDuration::from_millis(uplink_ms),
+                        downlink: SignedDuration::from_millis(downlink_ms),
+                        clock: OneWayClock::Uncalibrated,
+                    }),
+                })
+            }
+            _ => Err(IcmpError::InvalidPacket(
+                "unsupported ICMP message type".to_string(),
+            )),
+        }
+    }
+
+    fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> {
+        const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+        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(MeasurementEvent::Loss(MeasurementLoss {
+                    stream: probe.stream,
+                    started_at: probe.sent_at,
+                    detected_at: now,
+                }))
+                .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?;
+
+                    self.handle_reply(
+                        result.packet,
+                        result.peer.as_socket().unwrap().ip(),
+                        result.received_at,
+                        result.kernel_rx_timestamp,
+                    )?;
+                }
+
+                _ = housekeeping.tick() => {
+                    self.expire_inflight(Instant::now())?;
+                }
+            }
+        }
+    }
+}
+
+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
+    }
+}

src/measurement/mod.rs

@@ -0,0 +1,2 @@
+pub(crate) mod icmp;
+pub(crate) mod model;

src/measurement/model.rs

@@ -0,0 +1,138 @@
+use std::{
+    net::IpAddr,
+    time::{Duration, Instant},
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MeasurementStreamId(pub u64);
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum MeasurementSource {
+    IcmpEcho,
+    IcmpTimestamp,
+    Irtt,
+    TcpSeqAck,
+    TcpTimestamp,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct MeasurementStream {
+    pub id: MeasurementStreamId,
+    pub peer: IpAddr,
+    pub source: MeasurementSource,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OneWayClock {
+    Uncalibrated,
+    Calibrated,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ProbeSchedule {
+    pub period: Duration,
+    pub offset: Duration,
+}
+
+impl ProbeSchedule {
+    pub(crate) fn next_probe_at(&self, origin: Instant, now: Instant) -> Instant {
+        const NANOS_PER_SEC: u128 = 1_000_000_000;
+
+        let first = origin + self.offset;
+
+        if now < first {
+            return first;
+        }
+
+        let elapsed_ns = now.duration_since(first).as_nanos();
+        let period_ns = self.period.as_nanos();
+        let remainder = elapsed_ns % period_ns;
+
+        let until_next = if remainder == 0 {
+            period_ns
+        } else {
+            period_ns - remainder
+        };
+
+        let until_next = Duration::new(
+            (until_next / NANOS_PER_SEC) as u64,
+            (until_next % NANOS_PER_SEC) as u32,
+        );
+
+        now + until_next
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+pub struct SignedDuration {
+    nanos: i64,
+}
+
+#[allow(dead_code)]
+impl SignedDuration {
+    pub const fn from_nanos(nanos: i64) -> Self {
+        Self { nanos }
+    }
+
+    pub const fn from_millis(millis: i64) -> Self {
+        Self {
+            nanos: millis * 1_000_000,
+        }
+    }
+
+    pub const fn as_nanos(self) -> i64 {
+        self.nanos
+    }
+
+    pub fn as_secs_f64(self) -> f64 {
+        self.nanos as f64 / 1_000_000_000.0
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct OneWayLatency {
+    pub uplink: SignedDuration,
+    pub downlink: SignedDuration,
+    pub clock: OneWayClock,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone, Copy)]
+pub struct MeasurementObservation {
+    pub stream: MeasurementStream,
+
+    /// Local monotonic time at which measurement started.
+    pub started_at: Instant,
+
+    /// Local monotonic time at which the result was observed.
+    pub observed_at: Instant,
+
+    /// Directional timing, when the measurement mechanism can provide it.
+    pub one_way: Option<OneWayLatency>,
+}
+
+#[allow(dead_code)]
+impl MeasurementObservation {
+    pub fn rtt(&self) -> Duration {
+        self.observed_at.duration_since(self.started_at)
+    }
+
+    pub fn rtt_ms(&self) -> f64 {
+        self.rtt().as_secs_f64() * 1000.0
+    }
+}
+
+#[allow(dead_code)]
+pub struct MeasurementLoss {
+    pub stream: MeasurementStream,
+    pub started_at: Instant,
+    pub detected_at: Instant,
+}
+
+#[allow(dead_code)]
+pub enum MeasurementEvent {
+    Observation(MeasurementObservation),
+    Loss(MeasurementLoss),
+}

src/settings/mod.rs

@@ -218,6 +218,13 @@ pub(crate) struct NetworkSettings {
     pub(crate) upload_base_kbits: f64,
     #[serde(default = "default_min_percent")]
     pub(crate) upload_min_percent: f64,
+
+    #[allow(dead_code)]
+    pub(crate) measurement_bind_device: Option<String>,
+    #[allow(dead_code)]
+    pub(crate) measurement_mark: Option<u32>,
+    #[allow(dead_code)]
+    pub(crate) measurement_fib: Option<u32>,
 }
 
 impl NetworkSettings {

src/settings/validation.rs

@@ -399,6 +399,9 @@ mod tests {
             download_min_percent: 1.0,
             upload_base_kbits,
             upload_min_percent: 1.0,
+            measurement_bind_device: None,
+            measurement_mark: None,
+            measurement_fib: None,
         }
     }
 

src/time.rs

@@ -2,7 +2,10 @@
 //
 // SPDX-License-Identifier: MPL-2.0
 
-use std::marker::PhantomData;
+use std::{
+    marker::PhantomData,
+    time::{SystemTime, UNIX_EPOCH},
+};
 
 use rustix::time::{ClockId, clock_gettime};
 
@@ -64,3 +67,17 @@ impl ClockSample<Realtime> {
         (self.time_s as i64 % 86_400 * 1000) + self.time_ns as i64 / 1_000_000
     }
 }
+
+impl From<SystemTime> for ClockSample<Realtime> {
+    fn from(time: SystemTime) -> Self {
+        let duration = time
+            .duration_since(UNIX_EPOCH)
+            .expect("realtime timestamp predates Unix epoch");
+
+        Self {
+            time_s: duration.as_secs(),
+            time_ns: u64::from(duration.subsec_nanos()),
+            _clock: PhantomData,
+        }
+    }
+}