diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/paths.rs | 56 | ||||
| -rw-r--r-- | src/transport/stats.rs | 6 |
2 files changed, 45 insertions, 17 deletions
diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 6d1f9b4..b9dfc96 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -96,8 +96,8 @@ const PROBE_TIMEOUT: Duration = Duration::from_secs(15); const UNANSWERED_DOWN: u32 = 3; /// How often the probe task looks for paths due. const PROBE_TICK: Duration = Duration::from_secs(5); -/// Paths this node had a choice of within this long are probed; ones it didn't use for -/// twice as long are forgotten. +/// Paths this node had a choice of, or sent over, within this long are probed; ones it +/// didn't use for twice as long are forgotten. const ACTIVE_FOR: Duration = Duration::from_secs(300); /// The path in use is changed for one this much faster (a fraction of its round-trip /// time), and at least `SWITCH_MIN`. @@ -111,6 +111,9 @@ const MAX_PATHS: usize = 4096; struct PathState { /// When it was last among several to choose from (`rank`). used: Option<Instant>, + /// When a message last went over it: a peer's only path is measured too, if its + /// keepalives don't (a busy pair has none: `PeerState::take_keepalive_due`). + carried: Option<Instant>, /// Smoothed round-trip time of its PINGs. rtt: Option<Duration>, /// Where the transport went when `rtt` was measured (`Paths::location`). @@ -235,15 +238,22 @@ impl Table { } /// A send to `address` worked. - fn sent(&mut self, address: &str) { - if let Some(path) = self.paths.get_mut(address) { - path.failed_at = None; + fn sent(&mut self, address: &str, now: Instant) { + 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.failed_at = None; + path.carried = Some(now); } - /// The PINGs due now, with their nonces (from `nonce`): to every path used within - /// `ACTIVE_FOR` that wasn't pinged within `PROBE_EVERY` and has no PING out. A PING - /// out for `PROBE_TIMEOUT` counts as unanswered. + /// The PINGs due now, with their nonces (from `nonce`): to every path chosen among + /// others or sent over within `ACTIVE_FOR` that wasn't pinged (by us, or by a + /// keepalive) within `PROBE_EVERY` and has no PING out. A PING out for + /// `PROBE_TIMEOUT` counts as unanswered. fn due(&mut self, now: Instant, mut nonce: impl FnMut() -> u64) -> Vec<(String, u64)> { self.forget_unused(now); let mut due = Vec::new(); @@ -258,9 +268,10 @@ impl Table { info!(address = %address, "A path stopped answering PINGs; using others"); } } - let active = path - .used - .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR); + let lately = |at: Option<Instant>| { + at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR) + }; + let active = lately(path.used) || lately(path.carried); let rested = path .last_probe .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY); @@ -347,7 +358,7 @@ impl Table { at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR * 2) }; self.paths - .retain(|_, path| recent(path.used) || recent(path.last_probe)); + .retain(|_, path| recent(path.used) || recent(path.carried) || recent(path.last_probe)); let paths = &self.paths; self.chosen .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address))); @@ -649,7 +660,7 @@ impl Paths { match self.send(&encoded).await { Ok(()) => { let mut table = self.table.lock_or_recover(); - table.sent(address); + table.sent(address, Instant::now()); if !failures.is_empty() { table.counts.failovers += 1; drop(table); @@ -1039,7 +1050,7 @@ mod tests { table.failed("mycelium:x", now); assert!(table.info("mycelium:x", now).down); assert!(!table.info("mycelium:x", now + DOWN_FOR).down); - table.sent("mycelium:x"); + table.sent("mycelium:x", now); assert!(!table.info("mycelium:x", now).down); } @@ -1066,6 +1077,23 @@ mod tests { assert!(table.paths.is_empty() && table.chosen.is_empty()); } + /// A peer's only path is measured too once messages go over it, unless its + /// keepalives just did. + #[test] + fn a_path_sent_over_is_pinged() { + let mut table = Table::default(); + let now = Instant::now(); + table.sent("libp2p:z", now); + assert_eq!(table.due(now, || 1), [("libp2p:z".to_owned(), 1)]); + table.pong("libp2p:z", 1, now + Duration::from_millis(20)); + assert!(table.info("libp2p:z", now).rtt_ms.is_some()); + // A keepalive's PING within PROBE_EVERY: none of ours. + let later = now + PROBE_EVERY; + table.pinged("libp2p:z", 2, later); + table.sent("libp2p:z", later); + assert!(table.due(later, || 3).is_empty()); + } + /// A PING's round trip counts from when it goes, not from when it was found due. #[test] fn round_trips_count_from_the_send() { diff --git a/src/transport/stats.rs b/src/transport/stats.rs index ed8b247..68fc3c6 100644 --- a/src/transport/stats.rs +++ b/src/transport/stats.rs @@ -34,12 +34,12 @@ impl TransportStats { } } -/// How the paths to peers with several addresses do (`transport::paths`). +/// How the paths to peers do (`transport::paths`). #[derive(Debug, Clone, Default, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct PathsStats { - /// Every path measured: those of peers this node chose among lately, and those its - /// keepalives went over. + /// Every path measured: those of peers this node chose among or sent to lately, and + /// those its keepalives went over. pub(crate) paths: Vec<PathInfo>, #[serde(flatten)] pub(crate) counts: PathCounts, |