diff --git a/src/main.rs b/src/main.rs index 40b0281..5bc51c7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -243,6 +243,8 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } fn run(settings: &Settings) -> anyhow::Result<()> { + let rt = tokio::runtime::Runtime::new().unwrap(); + if settings.advanced_settings.dry_run { info!("*** MONITORING MODE ACTIVE — qdisc rates will NOT be changed ***"); } @@ -252,7 +254,10 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let (ie_txt, _) = flume::unbounded(); let (binding_tx, binding_rx) = flume::unbounded(); - let engine = IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)?; + let engine = { + let _guard = rt.enter(); + IcmpEngine::new(start_time, &settings.network, ie_txt, binding_rx)? + }; binding_tx.send(vec![IcmpBinding::new( MeasurementStream { id: MeasurementStreamId(0), @@ -265,7 +270,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> { }, )])?; - let rt = tokio::runtime::Runtime::new().unwrap(); rt.spawn(engine.run()); let ReflectorSetup { diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs index 22a6d49..cc2b59d 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -229,9 +229,7 @@ impl IcmpEngine { .get_mut(&stream_id) .expect("due binding disappeared"); - while binding.next_probe_at <= now { - binding.next_probe_at += binding.schedule.period; - } + binding.next_probe_at = binding.schedule.next_probe_at(self.origin, now); } Ok(()) @@ -318,8 +316,6 @@ impl IcmpEngine { receive, transmit, } => { - const NON_STANDARD_TIMESTAMP: u32 = 1 << 31; - if identifier != self.identifier { return Err(IcmpError::WrongIdentifier { expected: self.identifier, @@ -327,7 +323,10 @@ impl IcmpEngine { }); } - if receive & NON_STANDARD_TIMESTAMP != 0 || transmit & NON_STANDARD_TIMESTAMP != 0 { + 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(), )); @@ -365,8 +364,8 @@ impl IcmpEngine { } } - fn expire_inflight(&mut self, now: Instant) -> Result<(), IcmpError> { - const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + 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(); @@ -385,11 +384,12 @@ impl IcmpEngine { for probe in expired { self.observation_tx - .send(MeasurementEvent::Loss(MeasurementLoss { + .send_async(MeasurementEvent::Loss(MeasurementLoss { stream: probe.stream, started_at: probe.sent_at, detected_at: now, })) + .await .map_err(|_| IcmpError::EventChannelClosed)?; } @@ -432,7 +432,8 @@ impl IcmpEngine { ) { Ok(observation) => { self.observation_tx - .send(MeasurementEvent::Observation(observation)) + .send_async(MeasurementEvent::Observation(observation)) + .await .map_err(|_| IcmpError::EventChannelClosed)?; } @@ -449,7 +450,7 @@ impl IcmpEngine { } _ = housekeeping.tick() => { - self.expire_inflight(Instant::now())?; + self.expire_inflight(Instant::now()).await?; } } }