diff options
| -rw-r--r-- | Cargo.toml | 2 | ||||
| -rw-r--r-- | src/transport/libp2p.rs | 86 |
2 files changed, 68 insertions, 20 deletions
@@ -40,7 +40,7 @@ bytes = "1.12" clap = "4.6" dashmap = "6.2" futures = { version = "0.3", optional = true } -libp2p = { version = "0.57", optional = true, features = ["tokio", "tcp", "quic", "dns", "noise", "yamux", "ed25519", "identify", "macros", "relay", "dcutr", "autonat", "mdns", "kad"] } +libp2p = { version = "0.57", optional = true, features = ["tokio", "tcp", "quic", "dns", "noise", "yamux", "ed25519", "identify", "ping", "macros", "relay", "dcutr", "autonat", "mdns", "kad"] } libp2p-stream = { version = "0.5.0-alpha", optional = true } ed25519-dalek = "3" getrandom = "0.4" diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs index b5cee5f..9f83295 100644 --- a/src/transport/libp2p.rs +++ b/src/transport/libp2p.rs @@ -27,7 +27,9 @@ //! //! Of several connections to one peer (both dialled at once, or AutoNAT dialled back), //! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`), -//! by the side with the smaller peer ID. +//! by the side with the smaller peer ID: it keeps the direct one with the shortest round +//! trip (libp2p's ping, on each connection), so a path over the LAN wins over one +//! through a VPN, whichever opened first. //! //! A relay says so where it publishes its addresses: next to each, the same address //! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the @@ -136,8 +138,9 @@ const UPGRADE_RETRY: Duration = Duration::from_secs(30); const STALE_AFTER: Duration = Duration::from_secs(600); const PRUNE_EVERY: Duration = Duration::from_secs(60); /// Several connections to one peer (both dialled at once, or AutoNAT dialled back to -/// see whether it can) are cut to one (`Book::duplicates`) this long after the last one -/// opened: long enough for what it was dialled for. Looked for every `DEDUPE_EVERY`. +/// see whether it can) are cut to one, the fastest (`Book::duplicates`), this long after +/// the last one opened: long enough for what it was dialled for, and for each to be +/// pinged (at once, as it opens). Looked for every `DEDUPE_EVERY`. const DUPLICATE_GRACE: Duration = Duration::from_secs(10); const DEDUPE_EVERY: Duration = Duration::from_secs(5); @@ -269,6 +272,8 @@ struct Book { /// Peers connected only through a relay, dialled directly (`upgrade`): when to again /// at the soonest, and how long to wait after that. upgrades: HashMap<PeerId, (tokio::time::Instant, Duration)>, + /// The shortest round trip measured on each open connection (libp2p's ping). + rtts: HashMap<ConnectionId, Duration>, /// Open connections: to whom, whether through a relay, and the remote address. connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>, /// What AutoNAT says, and the address it confirmed while public. @@ -450,29 +455,30 @@ impl Book { if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) { return Vec::new(); } - let mut connections: Vec<(bool, ConnectionId)> = self + self.ranked(peer).into_iter().skip(1).collect() + } + + /// The connections to `peer`, the one to keep first: direct ones before relayed + /// ones, measured ones by their round trip (`rtts`: over the LAN before a VPN's hub, + /// whichever opened first), then the oldest. + fn ranked(&self, peer: &PeerId) -> Vec<ConnectionId> { + let mut connections: Vec<(bool, Duration, ConnectionId)> = self .connections .iter() .filter(|(_, (connected, _, _))| connected == peer) - .map(|(id, (_, relayed, _))| (*relayed, *id)) + .map(|(id, (_, relayed, _))| { + let rtt = self.rtts.get(id).copied().unwrap_or(Duration::MAX); + (*relayed, rtt, *id) + }) .collect(); - // Direct (false) first, then the oldest. connections.sort(); - connections.into_iter().skip(1).map(|(_, id)| id).collect() + connections.into_iter().map(|(_, _, id)| id).collect() } - /// The remote address of a connection to `peer`: a direct one if there is one. + /// The remote address of the connection to `peer` that is kept (`ranked`). fn location(&self, peer: &PeerId) -> Option<&Multiaddr> { - let mut found = None; - for (connected, relayed, address) in self.connections.values() { - if connected == peer { - if !relayed { - return Some(address); - } - found = Some(address); - } - } - found + let kept = *self.ranked(peer).first()?; + self.connections.get(&kept).map(|(_, _, address)| address) } fn stats(&self) -> Libp2pStats { @@ -920,6 +926,9 @@ struct Behaviour { /// Only with `Libp2pConfig::relay`. relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>, dcutr: ::libp2p::dcutr::Behaviour, + /// Each connection's round trip (`Book::rtts`): of several to one peer, the fastest + /// is kept. + ping: ::libp2p::ping::Behaviour, autonat: ::libp2p::autonat::Behaviour, /// Only with `Libp2pConfig::mdns`. mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>, @@ -1013,6 +1022,7 @@ impl Behaviour { relay_client, relay: relay.into(), dcutr: ::libp2p::dcutr::Behaviour::new(peer_id), + ping: ::libp2p::ping::Behaviour::new(::libp2p::ping::Config::new()), autonat: ::libp2p::autonat::Behaviour::new( peer_id, ::libp2p::autonat::Config { @@ -1428,7 +1438,11 @@ async fn run( } } SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => { - lock(&book).connections.remove(&connection_id); + { + let mut book = lock(&book); + book.connections.remove(&connection_id); + book.rtts.remove(&connection_id); + } let cause = match cause { _ if closing_duplicates.remove(&connection_id) => "a duplicate".to_owned(), // By us, or idle: libp2p doesn't tell which. @@ -1582,6 +1596,23 @@ async fn run( } debug!(event = ?event, "Relay event"); } + SwarmEvent::Behaviour(BehaviourEvent::Ping(::libp2p::ping::Event { + peer, + connection, + result, + })) => match result { + Ok(rtt) => { + let mut book = lock(&book); + if book.connections.contains_key(&connection) { + let shortest = book.rtts.entry(connection).or_insert(rtt); + *shortest = (*shortest).min(rtt); + } + trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, rtt = ?rtt, "Connection's round trip"); + } + Err(error) => { + trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, error = %error, "Connection's ping failed"); + } + }, SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result { Ok(_) => { lock(&book).counts.hole_punches.succeeded += 1; @@ -2698,5 +2729,22 @@ mod tests { book.slots.clear(); book.slots_here.insert(both); assert!(book.duplicates(&both, false).is_empty()); + book.slots_here.clear(); + // Once measured, the fastest direct one stays, however young. + book.rtts + .insert(ConnectionId::new_unchecked(3), Duration::from_millis(40)); + book.rtts + .insert(ConnectionId::new_unchecked(4), Duration::from_millis(2)); + assert_eq!( + book.duplicates(&both, false), + [ + ConnectionId::new_unchecked(3), + ConnectionId::new_unchecked(2) + ] + ); + assert_eq!( + book.location(&both).map(|address| address.to_string()), + Some("/ip4/10.0.0.4/tcp/1".to_owned()) + ); } } |