feat(measurement): implement ICMP measurement engine

src/main.rs

@@ -9,6 +9,7 @@ extern crate core;
 
 mod baseliner;
 mod log;
+mod measurement;
 mod metrics;
 mod pinger;
 mod pinger_icmp;

src/measurement/icmp.rs

@@ -0,0 +1,289 @@
+use std::{
+    collections::HashMap,
+    io,
+    net::IpAddr,
+    time::{Duration, Instant, SystemTime},
+};
+
+use flume::Sender;
+use icmp_socket2::{
+    IcmpSocket, IcmpSocket4, Icmpv4Message, Icmpv4Packet, packet::IcmpPacketBuildError,
+};
+use thiserror::Error;
+
+use crate::{
+    measurement::model::{
+        MeasurementEvent, MeasurementObservation, MeasurementSource, MeasurementStream,
+        MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration,
+    },
+    settings::NetworkSettings,
+    time::{ClockSample, Realtime},
+};
+
+pub struct IcmpBinding {
+    pub stream: MeasurementStream,
+    pub interval: Duration,
+
+    next_probe_at: Instant,
+}
+
+#[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("measurement observation channel closed")]
+    ObservationChannelClosed,
+
+    #[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 },
+}
+
+#[derive(Debug, Clone, Copy)]
+struct InFlightProbe {
+    stream: MeasurementStream,
+    sent_at: Instant,
+}
+
+type InFlightProbeKey = (IpAddr, MeasurementSource, u16);
+
+pub struct IcmpEngine {
+    rx_socket: IcmpSocket4,
+    tx_socket: IcmpSocket4,
+    identifier: u16,
+    next_sequence: u16,
+
+    bindings: HashMap<MeasurementStreamId, IcmpBinding>,
+    inflight: HashMap<InFlightProbeKey, InFlightProbe>,
+
+    observation_tx: Sender<MeasurementEvent>,
+}
+
+impl IcmpEngine {
+    pub fn new(
+        #[allow(unused)] settings: NetworkSettings,
+        observation_tx: Sender<MeasurementEvent>,
+    ) -> Result<Self, IcmpError> {
+        #[allow(unused_mut)]
+        let mut rx_socket = IcmpSocket4::new()?;
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(device) = &settings.measurement_bind_device {
+            icmp_socket2::IcmpSocket::bind_device(&mut rx_socket, device)?;
+        }
+
+        #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))]
+        if let Some(mark) = settings.measurement_mark {
+            icmp_socket2::IcmpSocket::set_mark(&mut rx_socket, mark)?;
+        }
+
+        #[cfg(target_os = "freebsd")]
+        if let Some(fib) = settings.measurement_fib {
+            icmp_socket2::IcmpSocket::set_fib(&mut rx_socket, fib)?;
+        }
+
+        rx_socket.enable_receive_metadata()?;
+        let tx_socket = rx_socket.try_clone()?;
+
+        Ok(Self {
+            rx_socket,
+            tx_socket,
+            identifier: (std::process::id() & 0xffff) as u16,
+            next_sequence: 0,
+            bindings: HashMap::new(),
+            inflight: HashMap::new(),
+            observation_tx,
+        })
+    }
+
+    pub fn set_bindings(
+        &mut self,
+        bindings: impl IntoIterator<Item = IcmpBinding>,
+    ) -> Result<(), IcmpError> {
+        todo!()
+    }
+
+    fn recv_loop(&mut self) -> Result<(), IcmpError> {
+        const RECEIVE_TIMEOUT: Duration = Duration::from_millis(50);
+        const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(500);
+
+        let mut next_housekeeping = Instant::now() + HOUSEKEEPING_INTERVAL;
+
+        loop {
+            match self.rx_socket.rcv_from_with_meta() {
+                Ok(result) => {
+                    self.handle_reply(
+                        result.packet,
+                        result.peer.as_socket().unwrap().ip(),
+                        result.received_at,
+                        result.kernel_rx_timestamp,
+                    )?;
+                }
+
+                Err(err)
+                    if matches!(
+                        err.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                    ) => {}
+
+                Err(err) => return Err(err.into()),
+            }
+
+            let now = Instant::now();
+
+            if now >= next_housekeeping {
+                self.expire_inflight(now);
+                next_housekeeping = now + HOUSEKEEPING_INTERVAL;
+            }
+        }
+    }
+
+    pub fn run(&mut self) -> Result<(), IcmpError> {
+        todo!()
+    }
+
+    fn send_due_probes(&mut self, now: Instant) -> Result<(), IcmpError> {
+        todo!()
+    }
+
+    fn send_probe(
+        &mut self,
+        stream_id: MeasurementStreamId,
+        now: Instant,
+    ) -> Result<(), IcmpError> {
+        todo!()
+    }
+
+    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,
+                    });
+                }
+
+                let key = (peer, MeasurementSource::IcmpTimestamp, sequence);
+
+                let probe = self
+                    .inflight
+                    .remove(&key)
+                    .ok_or(IcmpError::NoMatchingProbe(key))?;
+
+                if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 {
+                    return Err(IcmpError::InvalidPacket(
+                        "non-standard RFC 792 timestamp".into(),
+                    ));
+                }
+
+                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,
+                    }),
+                })
+            }
+            _ => {
+                return Err(IcmpError::InvalidPacket(
+                    "unsupported ICMP message type".to_string(),
+                ));
+            }
+        }
+    }
+
+    fn expire_inflight(&mut self, now: Instant) {
+        todo!()
+    }
+
+    fn allocate_sequence(&mut self, peer: IpAddr) -> u16 {
+        todo!()
+    }
+
+    fn next_deadline(&self) -> Option<Instant> {
+        todo!()
+    }
+}
+
+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 @@
+mod icmp;
+mod model;

src/measurement/model.rs

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

src/settings/mod.rs

@@ -218,6 +218,10 @@ pub(crate) struct NetworkSettings {
     pub(crate) upload_base_kbits: f64,
     #[serde(default = "default_min_percent")]
     pub(crate) upload_min_percent: f64,
+
+    pub(crate) measurement_bind_device: Option<String>,
+    pub(crate) measurement_mark: Option<u32>,
+    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,
+        }
+    }
+}