feat(measurement): implement ICMP measurement engine

src/main.rs

@@ -22,10 +22,6 @@ 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};
@@ -38,7 +34,7 @@ use crate::settings::Settings;
 use ::log::{info, warn};
 use flume::RecvTimeoutError;
 use std::collections::HashMap;
-use std::net::{IpAddr, Ipv4Addr};
+use std::net::IpAddr;
 use std::str::FromStr;
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
@@ -250,21 +246,6 @@ 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,7 +2,10 @@ use std::{
     collections::{HashMap, HashSet},
     io,
     net::IpAddr,
-    sync::{Arc, Mutex, atomic::Ordering},
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicBool, Ordering},
+    },
     thread,
     time::{Duration, Instant, SystemTime},
 };
@@ -59,7 +62,7 @@ pub enum IcmpError {
     #[error("probe interval must be non-zero for stream {0:?}")]
     InvalidInterval(MeasurementStreamId),
 
-    #[error("probe offset must be less than the period for stream {0:?}")]
+    #[error("probe offset must be non-zero for stream {0:?}")]
     InvalidOffset(MeasurementStreamId),
 
     #[error(
@@ -113,7 +116,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 @@
-pub(crate) mod icmp;
+mod icmp;
-pub(crate) mod model;
+mod model;

src/measurement/model.rs

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

src/metrics.rs

@@ -7,6 +7,7 @@ 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,6 +14,7 @@ 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,6 +11,7 @@ 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,6 +17,7 @@ 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};
 
-pub trait Clock {
+trait Clock {
     const ID: ClockId;
 }