feat(measurement): implement ICMP measurement engine

Remove direct `rustix::time::ClockId` imports by centralizing its dependency within the `time` module.

src/main.rs

@@ -22,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};
@@ -34,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};
@@ -246,6 +250,21 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
     let start_time = Instant::now();
     let probe_id = probe_identifier();
 
+    let (ie_txt, _) = flume::unbounded();
+    let mut engine = IcmpEngine::new(start_time, &settings.network, ie_txt)?;
+    engine.set_bindings(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),
+        },
+    )])?;
+    engine.run()?;
+
     let ReflectorSetup {
         peers: reflector_peers,
         pool: reflector_pool,

src/measurement/icmp.rs

@@ -2,10 +2,7 @@ use std::{
     collections::{HashMap, HashSet},
     io,
     net::IpAddr,
-    sync::{
+    sync::{Arc, Mutex, atomic::Ordering},
-        Arc, Mutex,
-        atomic::{AtomicBool, Ordering},
-    },
     thread,
     time::{Duration, Instant, SystemTime},
 };
@@ -62,7 +59,7 @@ pub enum IcmpError {
     #[error("probe interval must be non-zero for stream {0:?}")]
     InvalidInterval(MeasurementStreamId),
 
-    #[error("probe offset must be non-zero for stream {0:?}")]
+    #[error("probe offset must be less than the period for stream {0:?}")]
     InvalidOffset(MeasurementStreamId),
 
     #[error(
@@ -116,7 +113,7 @@ pub struct IcmpEngine {
 impl IcmpEngine {
     pub fn new(
         origin: Instant,
-        #[allow(unused)] settings: NetworkSettings,
+        #[allow(unused)] settings: &NetworkSettings,
         observation_tx: Sender<MeasurementEvent>,
     ) -> Result<Self, IcmpError> {
         #[allow(unused_mut)]

src/measurement/mod.rs

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

src/measurement/model.rs

@@ -1,5 +1,4 @@
 use std::{
-    intrinsics::simd::SimdAlign::Element,
     net::IpAddr,
     time::{Duration, Instant},
 };
@@ -37,14 +36,15 @@ pub struct ProbeSchedule {
 
 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 = now.duration_since(first);
+        let elapsed_ns = now.duration_since(first).as_nanos();
-        let elapsed_ns = elapsed.as_nanos();
         let period_ns = self.period.as_nanos();
         let remainder = elapsed_ns % period_ns;
 
@@ -54,7 +54,12 @@ impl ProbeSchedule {
             period_ns - remainder
         };
 
-        now + Duration::from_nanos_u128(until_next)
+        let until_next = Duration::new(
+            (until_next / NANOS_PER_SEC) as u64,
+            (until_next % NANOS_PER_SEC) as u32,
+        );
+
+        now + until_next
     }
 }
 

src/metrics.rs

@@ -7,7 +7,6 @@ use crate::settings::{MeasurementType, ObservabilityProtocol};
 use crate::time::{ClockSample, Realtime};
 use flume::{Receiver, RecvTimeoutError, Sender};
 use log::{error, info, warn};
-use rustix::time::ClockId;
 use std::fmt::Write;
 use std::net::{IpAddr, TcpStream, UdpSocket};
 use std::sync::Arc;

src/pinger_icmp.rs

@@ -14,7 +14,6 @@ use crate::time::{ClockSample, Monotonic};
 use icmp_socket2::Icmpv4Message;
 use icmp_socket2::Icmpv4Packet;
 use icmp_socket2::packet::WithEchoRequest;
-use rustix::time::ClockId;
 
 pub struct PingerICMPEchoListener {}
 

src/pinger_icmp_ts.rs

@@ -11,7 +11,6 @@ use crate::time::{ClockSample, Realtime};
 use icmp_socket2::Icmpv4Message;
 use icmp_socket2::Icmpv4Packet;
 use icmp_socket2::packet::WithTimestampRequest;
-use rustix::time::ClockId;
 use std::net::IpAddr;
 use std::time::Instant;
 

src/ratecontroller.rs

@@ -17,7 +17,6 @@ use crate::time::{ClockSample, Realtime};
 use crate::util::{ArcRwLock, RwLockExt};
 use flume::{Receiver, Sender};
 use log::{debug, info, warn};
-use rustix::time::ClockId;
 use std::fs::File;
 use std::io::Write;
 use std::net::IpAddr;

src/time.rs

@@ -9,7 +9,7 @@ use std::{
 
 use rustix::time::{ClockId, clock_gettime};
 
-trait Clock {
+pub trait Clock {
     const ID: ClockId;
 }