refactor(pinger): replace hashlru with standard HashMap and manual TTL cleanup
Remove the hashlru dependency to reduce bloat, replacing it with a Mutex-wrapped HashMap and manual entry expiration logic.
This commit is contained in:
4 files changed
+17
-25
No files matched your search
Generated
-7
@@ -155,12 +155,6 @@ version = "0.3.3"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
|
checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
|
||||||
|
|
||||||
[[package]]
|
|
||||||
name = "hashlru"
|
|
||||||
version = "0.11.1"
|
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
|
||||||
checksum = "d2c782df656366ccb42fbbac7c7c180535a43d906a138ffb491ec3ed56a239ab"
|
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "icmp-socket2"
|
name = "icmp-socket2"
|
||||||
version = "0.4.2"
|
version = "0.4.2"
|
||||||
@@ -426,7 +420,6 @@ dependencies = [
|
|||||||
"anyhow",
|
"anyhow",
|
||||||
"fastrand",
|
"fastrand",
|
||||||
"flume",
|
"flume",
|
||||||
"hashlru",
|
|
||||||
"icmp-socket2",
|
"icmp-socket2",
|
||||||
"libc",
|
"libc",
|
||||||
"log",
|
"log",
|
||||||
|
|||||||
@@ -29,7 +29,6 @@ rust-uci = { git = "https://github.com/namib-project/rust-uci.git", rev = "71ea7
|
|||||||
thiserror = "2.0"
|
thiserror = "2.0"
|
||||||
libc = "0.2"
|
libc = "0.2"
|
||||||
flume = { version = "0.12.0", default-features = false, features = ["eventual-fairness", "select"] }
|
flume = { version = "0.12.0", default-features = false, features = ["eventual-fairness", "select"] }
|
||||||
hashlru = { version = "0.11.1", default-features = false }
|
|
||||||
|
|
||||||
[dependencies.log]
|
[dependencies.log]
|
||||||
features = ["std"]
|
features = ["std"]
|
||||||
|
|||||||
+2
-1
@@ -195,7 +195,8 @@ fn main() -> anyhow::Result<()> {
|
|||||||
"In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})",
|
"In-flight probe cache capacity: {} (active_reflectors={}, reselection_enabled={}, tick_interval_s={})",
|
||||||
inflight_probe_capacity, active_reflector_count, reselection_enabled, config.tick_interval
|
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 {
|
let baseliner = Baseliner {
|
||||||
config: config.clone(),
|
config: config.clone(),
|
||||||
|
|||||||
+15
-16
@@ -1,15 +1,15 @@
|
|||||||
use crate::MeasurementType;
|
use crate::MeasurementType;
|
||||||
use crate::SHUTDOWN;
|
use crate::SHUTDOWN;
|
||||||
use crate::metrics::{Metric, MetricsSender};
|
use crate::metrics::{Metric, MetricsSender};
|
||||||
use crate::util::RwLockExt;
|
use crate::util::{MutexExt, RwLockExt};
|
||||||
use flume::Sender;
|
use flume::Sender;
|
||||||
use hashlru::SyncCache;
|
|
||||||
use icmp_socket2::socket::IcmpSocket;
|
use icmp_socket2::socket::IcmpSocket;
|
||||||
use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
|
use icmp_socket2::{IcmpSocket4, Icmpv4Packet};
|
||||||
use log::{debug, info, warn};
|
use log::{debug, info, warn};
|
||||||
|
use std::collections::HashMap;
|
||||||
use std::net::{IpAddr, Ipv4Addr};
|
use std::net::{IpAddr, Ipv4Addr};
|
||||||
use std::sync::atomic::Ordering;
|
use std::sync::atomic::Ordering;
|
||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, Mutex, RwLock};
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
use std::{io, thread};
|
use std::{io, thread};
|
||||||
use thiserror::Error;
|
use thiserror::Error;
|
||||||
@@ -47,7 +47,8 @@ pub struct InFlightProbe {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub type InFlightProbeKey = (IpAddr, MeasurementType, u16);
|
pub type InFlightProbeKey = (IpAddr, MeasurementType, u16);
|
||||||
pub type InFlightProbeCache = SyncCache<InFlightProbeKey, InFlightProbe>;
|
pub type InFlightProbeCache = Arc<Mutex<HashMap<InFlightProbeKey, InFlightProbe>>>;
|
||||||
|
const INFLIGHT_PROBE_TTL: Duration = Duration::from_secs(30);
|
||||||
|
|
||||||
fn open_socket(type_: MeasurementType) -> io::Result<IcmpSocket4> {
|
fn open_socket(type_: MeasurementType) -> io::Result<IcmpSocket4> {
|
||||||
match type_ {
|
match type_ {
|
||||||
@@ -97,15 +98,12 @@ pub trait PingListener {
|
|||||||
};
|
};
|
||||||
|
|
||||||
let key = (addr, type_, reply.seq);
|
let key = (addr, type_, reply.seq);
|
||||||
let Some(probe) = inflight.remove(&key) else {
|
let Some(probe) = inflight.lock_anyhow()?.remove(&key) else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
if reply.originate_timestamp != probe.originate_timestamp {
|
if reply.originate_timestamp != probe.originate_timestamp {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
if probe.sent_at.elapsed() > Duration::from_secs(30) {
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
ping_metrics.send(Metric::Ping {
|
ping_metrics.send(Metric::Ping {
|
||||||
reflector: reply.reflector,
|
reflector: reply.reflector,
|
||||||
@@ -174,6 +172,10 @@ pub trait PingSender {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
inflight
|
||||||
|
.lock_anyhow()?
|
||||||
|
.retain(|_, probe| probe.sent_at.elapsed() < INFLIGHT_PROBE_TTL);
|
||||||
|
|
||||||
let sleep_duration =
|
let sleep_duration =
|
||||||
Duration::from_millis((tick_duration_ms / reflectors.len() as u16) as u64);
|
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 (packet, originate_timestamp) = self.craft_packet(id, seq);
|
||||||
let sent_at = Instant::now();
|
let sent_at = Instant::now();
|
||||||
inflight.insert(
|
let key = (*reflector, type_, seq);
|
||||||
(*reflector, type_, seq),
|
inflight.lock_anyhow()?.insert(
|
||||||
|
key,
|
||||||
InFlightProbe {
|
InFlightProbe {
|
||||||
sent_at,
|
sent_at,
|
||||||
originate_timestamp,
|
originate_timestamp,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
if let Err(e) = socket.send_to(addr, packet) {
|
if let Err(e) = socket.send_to(addr, packet) {
|
||||||
inflight.remove(&(*reflector, type_, seq));
|
inflight.lock_anyhow()?.remove(&key);
|
||||||
return Err(e.into());
|
return Err(e.into());
|
||||||
}
|
}
|
||||||
thread::sleep(sleep_duration);
|
thread::sleep(sleep_duration);
|
||||||
}
|
}
|
||||||
|
|
||||||
if seq == u16::MAX {
|
seq = seq.wrapping_add(1);
|
||||||
seq = 0;
|
|
||||||
} else {
|
|
||||||
seq += 1;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user