diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/identity.rs | 6 | ||||
| -rw-r--r-- | src/transport/mod.rs | 7 | ||||
| -rw-r--r-- | src/transport/paths.rs | 138 |
3 files changed, 134 insertions, 17 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 1df84ec..0b36967 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -610,6 +610,12 @@ impl Transport for IdentityTransport { self.shared.paths.rtt(&address) } + /// Its proven paths, the one in use first. + fn suspect(&self, peer: &str) { + let addresses = self.shared.proven_at(peer, false); + self.shared.paths.suspect(&addresses); + } + /// The narrowest of its proven paths'. fn clock_offset(&self, peer: &str) -> Option<crate::transport::paths::ClockOffset> { self.shared diff --git a/src/transport/mod.rs b/src/transport/mod.rs index eeb9d34..474e386 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -1004,6 +1004,13 @@ pub(crate) trait Transport: Send + Sync { None } + /// What went to `peer` went unanswered for a while (`swarm::reach`): the path in use + /// may take messages and drop them. The paths layer tries the peer's other paths + /// first and pings them all now (`paths::Paths::suspect`). By default nothing. + fn suspect(&self, peer: &str) { + let _ = peer; + } + /// Whether the transport answers PINGs with PONGs itself (the paths layer, /// `transport::paths`); if not (a bare transport, in tests), the node does. By /// default not. diff --git a/src/transport/paths.rs b/src/transport/paths.rs index 6f0a892..8c03b1f 100644 --- a/src/transport/paths.rs +++ b/src/transport/paths.rs @@ -21,7 +21,9 @@ //! 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 +//! send that failed within `DOWN_FOR`, make a path down. What went to a peer going +//! unanswered makes the path in use suspect (`Paths::suspect`): the others first, all +//! pinged at once. `rank` puts paths that are up //! first, the one in use first among them, and the rest by round-trip time; it changes //! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least //! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the @@ -160,6 +162,9 @@ struct PathState { /// The peer's clock as its PONGs on this path say (the narrowest bound lately), and /// when that sample came. clock: Option<(ClockOffset, Instant)>, + /// What went over it went unanswered (`Paths::suspect`): the peer's other paths, if + /// up, come first until it answers a PING again. + suspect: bool, } impl PathState { @@ -233,6 +238,13 @@ impl Table { .get(address) .is_some_and(|path| path.is_down(now)) }); + // Suspect paths after the others that are up, as long as any isn't. + let suspect = + |table: &Self, address: &str| table.paths.get(address).is_some_and(|path| path.suspect); + let (fine, doubted): (Vec<_>, Vec<_>) = up + .into_iter() + .partition(|(address, _)| !suspect(self, address)); + let up: Vec<(String, usize)> = fine.into_iter().chain(doubted).collect(); let mut set: Vec<&str> = up .iter() .chain(down.iter()) @@ -246,10 +258,12 @@ impl Table { .chosen .get(&key) .filter(|chosen| up.iter().any(|(address, _)| address == *chosen)) + .filter(|chosen| !suspect(self, chosen) || suspect(self, &up[0].0)) .cloned() .unwrap_or_else(|| up[0].0.clone()); let fastest = up .iter() + .filter(|(address, _)| !suspect(self, address) || suspect(self, ¤t)) .filter_map(|(address, _)| Some((address, rtt(self, address)?))) .min_by_key(|(_, rtt)| *rtt); if let (Some((fastest, fastest_rtt)), Some(current_rtt)) = @@ -351,6 +365,40 @@ impl Table { due } + /// What went to the peer at `addresses` went unanswered: the first, the one in use, + /// is suspect (`PathState::suspect`), and each without a PING out gets one now. + /// Returns those PINGs, with their nonces (from `nonce`). + fn suspect( + &mut self, + addresses: &[String], + now: Instant, + mut nonce: impl FnMut() -> u64, + ) -> Vec<(String, u64)> { + if let Some(first) = addresses.first() + && addresses.len() > 1 + && let Some(path) = self.paths.get_mut(first) + && !path.suspect + { + info!(address = %first, "Answers over a path are overdue; trying the peer's others first"); + path.suspect = true; + } + let mut due = Vec::new(); + for address in addresses { + if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) { + continue; + } + let path = self.paths.entry(address.clone()).or_default(); + if path.probe.is_none() { + let nonce = nonce(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; + due.push((address.clone(), nonce)); + } + } + due + } + /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that /// path, which its PONG measures, even if it is a peer's only path; `due` then /// doesn't ping it again within `probe_every`. @@ -408,6 +456,7 @@ impl Table { } path.unanswered = 0; path.failed_at = None; + path.suspect = false; path.rtt = Some(match path.rtt { Some(rtt) => (rtt * 3 + sample) / 4, None => sample, @@ -913,9 +962,7 @@ impl Paths { sent } - /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. Each PING - /// goes on its own, timed from when it goes: a send that takes long (a dial) neither - /// holds up the others nor counts in their round trips. + /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. async fn probe(self: Arc<Self>) { let mut tick = tokio::time::interval(PROBE_TICK); tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -925,25 +972,45 @@ impl Paths { .table .lock_or_recover() .due(Instant::now(), random_nonce); - for (address, nonce) in due { - let paths = Arc::clone(&self); - tokio::spawn(async move { + Arc::clone(&self).ping(due); + } + } + + /// Sends these PINGs. Each goes on its own, timed from when it goes: a send that + /// takes long (a dial) neither holds up the others nor counts in their round trips. + fn ping(self: Arc<Self>, due: Vec<(String, u64)>) { + for (address, nonce) in due { + let paths = Arc::clone(&self); + tokio::spawn(async move { + paths + .table + .lock_or_recover() + .sending(&address, nonce, Instant::now()); + if let Err(error) = paths.send_probe(&address, false, nonce).await { + debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING"); paths .table .lock_or_recover() - .sending(&address, nonce, Instant::now()); - if let Err(error) = paths.send_probe(&address, false, nonce).await { - debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING"); - paths - .table - .lock_or_recover() - .failed(&address, Instant::now()); - } - }); - } + .failed(&address, Instant::now()); + } + }); } } + /// What went to a peer at `addresses` (best first, as `rank` gives them) went + /// unanswered (`swarm::reach`): the path in use is suspect, and the peer's other + /// paths come first while they are up, until it answers a PING again; every path to + /// it gets a PING now, rather than at its next probe (up to `PROBE_EVERY`). A path + /// that takes messages and drops them (a transport that only queues them locally) + /// is otherwise down only after three unanswered probes, a minute and a half. + pub(crate) fn suspect(self: &Arc<Self>, addresses: &[String]) { + let due = self + .table + .lock_or_recover() + .suspect(addresses, Instant::now(), random_nonce); + Arc::clone(self).ping(due); + } + /// A message from `address`, handed on to `handler`: a PING answered and a PONG /// noted first; dropped if the firewall blocks it. async fn receive( @@ -1193,6 +1260,43 @@ mod tests { assert_eq!(table.rank(lan, now, true), ["libp2p:y", "mycelium:y"]); } + /// A path whose messages went unanswered gives way to the peer's other paths, each + /// pinged at once; answering a PING, it is trusted again, but the choice stays + /// where it went until it is clearly faster. + #[test] + fn suspect_paths_give_way() { + let mut table = Table::default(); + let now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + assert_eq!(table.rank(set.clone(), now, true)[0], "libp2p:x"); + let mut nonces = 0; + let pinged = table.suspect( + &["libp2p:x".to_string(), "mycelium:x".to_string()], + now, + || { + nonces += 1; + nonces + }, + ); + assert_eq!(pinged.len(), 2, "every path pinged"); + assert_eq!( + table.rank(set.clone(), now, true), + ["mycelium:x", "libp2p:x"] + ); + // It answers: no longer suspect, and nothing faster than the other. + for (address, nonce) in pinged { + table.pong(&address, nonce, now + Duration::from_millis(100)); + } + assert_eq!( + table.rank(set.clone(), now, true), + ["mycelium:x", "libp2p:x"] + ); + // A peer's only path: nothing to give way to, but pinged. + let only = table.suspect(&["mycelium:y".to_string()], now, || 9); + assert_eq!(only.len(), 1); + assert!(!table.paths["mycelium:y"].suspect); + } + #[test] fn unanswered_or_failing_paths_are_down() { let mut table = Table::default(); |