From 30c2ca8537fbc2874ec6e14cdd1ae5477c95f099 Mon Sep 17 00:00:00 2001 From: tv Date: Wed, 28 Oct 2026 03:17:29 +0100 Subject: measure RTT on single link paths, too --- src/simtest/identity.rs | 27 +++++++++++--------------- src/transport/paths.rs | 51 ++++++++++++++++++++++++++++++++++++------------- src/transport/stats.rs | 6 ++++-- 3 files changed, 53 insertions(+), 31 deletions(-) (limited to 'src') diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index e93b1da..61a54d5 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -14,10 +14,7 @@ use crate::swarm::bootstrap::Start; use crate::swarm::members; use crate::transport::identity::{WHO_LABEL, signed}; use crate::transport::sim::SimNet; -use crate::transport::stats::PathInfo; -use crate::transport::{ - Destination, HeadsSummary, IdentityProof, ParsedMessage, Status, Transport, -}; +use crate::transport::{HeadsSummary, IdentityProof, ParsedMessage, Status, Transport}; const SOON: Duration = Duration::from_secs(20); @@ -45,20 +42,18 @@ async fn nodes_are_their_node_ids() { peers_are_node_ids(node).await; } // Where a sends to b: where b proved itself, on the sim's transport. + let destination = a.transport.destination(b.local_id()).unwrap(); assert_eq!( - a.transport.destination(b.local_id()), - Some(Destination { - address: "sim:b".into(), - proven: true, - connection: None, - // One path: nothing to choose, so nothing measured. - paths: vec![PathInfo { - address: "sim:b".into(), - rtt_ms: None, - down: false, - }], - }) + ( + destination.address.as_str(), + destination.proven, + destination.connection + ), + ("sim:b", true, None) ); + // Its only path, measured by keepalives once there were any. + assert_eq!(destination.paths.len(), 1); + assert_eq!(destination.paths[0].address, "sim:b"); let stats = a.transport.stats(); assert_eq!(stats.identity.unwrap().proven["sim"], 2); assert!(stats.libp2p.is_none()); diff --git a/src/transport/paths.rs b/src/transport/paths.rs index dda2d87..6d4b3a3 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -15,8 +15,9 @@ //! next address, the message encoded for its transport: when one transport fails, the //! next carries on within the same send. //! -//! Paths measures the addresses of such nodes itself: every `PROBE_EVERY`, each address -//! it had a choice of within `ACTIVE_FOR` gets a PING with a random nonce, which the +//! Paths measures the addresses of such nodes itself (and every path the node's +//! keepalive PINGs go over, below): every `PROBE_EVERY`, each address it had a choice of +//! within `ACTIVE_FOR` gets a PING with a random nonce, which the //! other side's paths answers at once with a PONG, over the same path. Answers give each //! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a //! send that failed within `DOWN_FOR`, make a path down. `rank` puts paths that are up @@ -266,13 +267,19 @@ impl Table { } /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that - /// path, if paths measures it. + /// path, which its PONG measures, even if it is a peer's only path (then it isn't + /// pinged otherwise: `due` pings only paths there was a choice of). fn pinged(&mut self, address: &str, nonce: u64, now: Instant) { - if let Some(path) = self.paths.get_mut(address) { - path.probe = Some((nonce, now)); - path.last_probe = Some(now); - self.counts.probes_sent += 1; + if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) { + self.forget_unused(now); + if self.paths.len() >= MAX_PATHS { + return; + } } + let path = self.paths.entry(address.to_owned()).or_default(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; } /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out. @@ -297,12 +304,14 @@ impl Table { path.rtt } - /// Forgets the paths not used for twice `ACTIVE_FOR`, and choices among them. + /// Forgets the paths neither used nor pinged for twice `ACTIVE_FOR`, and choices + /// among them. fn forget_unused(&mut self, now: Instant) { - self.paths.retain(|_, path| { - path.used - .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR * 2) - }); + let recent = |at: Option| { + at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR * 2) + }; + self.paths + .retain(|_, path| recent(path.used) || recent(path.last_probe)); let paths = &self.paths; self.chosen .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address))); @@ -965,8 +974,24 @@ mod tests { .is_some() ); assert_eq!(table.counts.probes_sent, 2); - // A peer with one path isn't measured. + // A peer's only path is measured by its keepalives, and not pinged otherwise. table.pinged("libp2p:z", 8, now); + assert!( + table + .pong("libp2p:z", 8, now + Duration::from_millis(30)) + .is_some() + ); + let rtt = table.info("libp2p:z", now).rtt_ms.unwrap(); + assert!((rtt - 30.0).abs() < 1e-6, "{rtt}"); + let later = now + PROBE_EVERY; + assert!( + table + .due(later, &mut next) + .iter() + .all(|(address, _)| address != "libp2p:z") + ); + // Forgotten once the keepalives stop. + table.due(now + ACTIVE_FOR * 2, &mut next); assert!(!table.paths.contains_key("libp2p:z")); } diff --git a/src/transport/stats.rs b/src/transport/stats.rs index 6b1a42f..12b2c99 100644 --- a/src/transport/stats.rs +++ b/src/transport/stats.rs @@ -36,7 +36,8 @@ impl TransportStats { #[derive(Debug, Clone, Default, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct PathsStats { - /// Every path measured: those of peers this node chose among lately. + /// Every path measured: those of peers this node chose among lately, and those its + /// keepalives went over. pub(crate) paths: Vec, #[serde(flatten)] pub(crate) counts: PathCounts, @@ -48,7 +49,8 @@ pub(crate) struct PathsStats { pub(crate) struct PathInfo { /// `:`. pub(crate) address: String, - /// Smoothed round-trip time of its PINGs; absent before an answer. + /// Smoothed round-trip time of its PINGs (ours, and our keepalives); absent before + /// an answer. pub(crate) rtt_ms: Option, /// Its PINGs went unanswered three times in a row, or a send there failed within a /// minute: used only when no other path works. -- cgit v1.3.1