diff --git a/Cargo.lock b/Cargo.lock index 470f9d2..80f3a37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -155,12 +155,6 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" -[[package]] -name = "hashlru" -version = "0.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d2c782df656366ccb42fbbac7c7c180535a43d906a138ffb491ec3ed56a239ab" - [[package]] name = "icmp-socket2" version = "0.4.2" @@ -426,7 +420,6 @@ dependencies = [ "anyhow", "fastrand", "flume", - "hashlru", "icmp-socket2", "libc", "log", diff --git a/Cargo.toml b/Cargo.toml index 80a5247..734430d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,7 +29,6 @@ rust-uci = { git = "https://github.com/namib-project/rust-uci.git", rev = "71ea7 thiserror = "2.0" libc = "0.2" flume = { version = "0.12.0", default-features = false, features = ["eventual-fairness", "select"] } -hashlru = { version = "0.11.1", default-features = false } [dependencies.log] features = ["std"] diff --git a/src/main.rs b/src/main.rs index e9f0d23..b893653 100644 --- a/src/main.rs +++ b/src/main.rs @@ -195,7 +195,8 @@ fn main() -> anyhow::Result<()> { "In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})", inflight_probe_capacity, active_reflector_count, reselection_enabled, config.tick_interval ); - let inflight: InFlightProbeCache = InFlightProbeCache::new(inflight_probe_capacity); + let inflight: InFlightProbeCache = + Arc::new(Mutex::new(HashMap::with_capacity(inflight_probe_capacity))); let baseliner = Baseliner { config: config.clone(), diff --git a/src/pinger.rs b/src/pinger.rs index ccfc591..c030039 100644 --- a/src/pinger.rs +++ b/src/pinger.rs @@ -1,15 +1,15 @@ use crate::MeasurementType; use crate::SHUTDOWN; use crate::metrics::{Metric, MetricsSender}; -use crate::util::RwLockExt; +use crate::util::{MutexExt, RwLockExt}; use flume::Sender; -use hashlru::SyncCache; use icmp_socket2::socket::IcmpSocket; use icmp_socket2::{IcmpSocket4, Icmpv4Packet}; use log::{debug, info, warn}; +use std::collections::HashMap; use std::net::{IpAddr, Ipv4Addr}; use std::sync::atomic::Ordering; -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, Instant}; use std::{io, thread}; use thiserror::Error; @@ -47,7 +47,8 @@ pub struct InFlightProbe { } pub type InFlightProbeKey = (IpAddr, MeasurementType, u16); -pub type InFlightProbeCache = SyncCache; +pub type InFlightProbeCache = Arc>>; +const INFLIGHT_PROBE_TTL: Duration = Duration::from_secs(30); fn open_socket(type_: MeasurementType) -> io::Result { match type_ { @@ -97,15 +98,12 @@ pub trait PingListener { }; let key = (addr, type_, reply.seq); - let Some(probe) = inflight.remove(&key) else { + let Some(probe) = inflight.lock_anyhow()?.remove(&key) else { continue; }; if reply.originate_timestamp != probe.originate_timestamp { continue; } - if probe.sent_at.elapsed() > Duration::from_secs(30) { - continue; - } ping_metrics.send(Metric::Ping { reflector: reply.reflector, @@ -174,6 +172,10 @@ pub trait PingSender { continue; } + inflight + .lock_anyhow()? + .retain(|_, probe| probe.sent_at.elapsed() < INFLIGHT_PROBE_TTL); + let sleep_duration = Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64); @@ -185,25 +187,22 @@ pub trait PingSender { let (packet, originate_timestamp) = self.craft_packet(id, seq); let sent_at = Instant::now(); - inflight.insert( - (*reflector, type_, seq), + let key = (*reflector, type_, seq); + inflight.lock_anyhow()?.insert( + key, InFlightProbe { sent_at, originate_timestamp, }, ); if let Err(e) = socket.send_to(addr, packet) { - inflight.remove(&(*reflector, type_, seq)); + inflight.lock_anyhow()?.remove(&key); return Err(e.into()); } thread::sleep(sleep_duration); } - if seq == u16::MAX { - seq = 0; - } else { - seq += 1; - } + seq = seq.wrapping_add(1); } }