diff options
Diffstat (limited to 'src/transport/libp2p')
| -rw-r--r-- | src/transport/libp2p/behaviour.rs | 4 | ||||
| -rw-r--r-- | src/transport/libp2p/book.rs | 38 | ||||
| -rw-r--r-- | src/transport/libp2p/driver.rs | 22 | ||||
| -rw-r--r-- | src/transport/libp2p/mod.rs | 8 | ||||
| -rw-r--r-- | src/transport/libp2p/relays.rs | 23 | ||||
| -rw-r--r-- | src/transport/libp2p/tests.rs | 29 |
6 files changed, 118 insertions, 6 deletions
diff --git a/src/transport/libp2p/behaviour.rs b/src/transport/libp2p/behaviour.rs index e9b156c..a2e8e91 100644 --- a/src/transport/libp2p/behaviour.rs +++ b/src/transport/libp2p/behaviour.rs @@ -244,7 +244,9 @@ impl Behaviour { dht }); let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public()) - .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION"))); + .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION"))) + // Connected peers hear of a new interface at once (`explore`). + .with_push_listen_addr_updates(true); let relay = relay.then(|| { ::libp2p::relay::Behaviour::new( peer_id, diff --git a/src/transport/libp2p/book.rs b/src/transport/libp2p/book.rs index 44b2694..7de4be7 100644 --- a/src/transport/libp2p/book.rs +++ b/src/transport/libp2p/book.rs @@ -41,6 +41,9 @@ pub(super) struct Book { /// connection to the peer closes, as a peer restarted elsewhere is found by what /// the others say. pub(super) identified: HashMap<PeerId, Vec<Multiaddr>>, + /// Addresses of directly connected peers dialled for a better path (`explore`): + /// not again while connected, unless our own interfaces change. + pub(super) explored: HashMap<PeerId, std::collections::HashSet<Multiaddr>>, pub(super) listening: Vec<Multiaddr>, /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`). pub(super) mapped: Vec<Multiaddr>, @@ -213,6 +216,7 @@ impl Book { for peer in &stale { self.addresses.remove(peer); self.identified.remove(peer); + self.explored.remove(peer); self.relays.remove(peer); self.public_relays.remove(peer); self.continuum.remove(peer); @@ -475,6 +479,38 @@ impl Book { addresses.truncate(ADDRESSES_PER_PEER); } + /// The addresses `explore` dials `peer` at, noted as dialled: a node of the swarm + /// connected directly, at addresses of its not dialled so yet, and with `again` + /// not only those at an IP it isn't connected at (an inbound connection's remote + /// port isn't where the peer listens). + pub(super) fn to_explore(&mut self, peer: PeerId, again: bool) -> Vec<Multiaddr> { + if !self.continuum.contains(&peer) + || book_connection(&self.connections, &peer) != Some(Connection::Direct) + { + return Vec::new(); + } + let used: Vec<Option<std::net::IpAddr>> = self + .connections + .values() + .filter(|(connected, relayed, _)| *connected == peer && !relayed) + .map(|(_, _, address)| ip_of(address)) + .collect(); + let tried = self.explored.entry(peer).or_default(); + let candidates: Vec<Multiaddr> = self + .addresses + .get(&peer) + .into_iter() + .flatten() + .filter(|address| is_ip(address) && !left_out(&self.ignored, address)) + .filter(|address| again || !used.contains(&ip_of(address))) + .filter(|address| !tried.contains(*address)) + .take(EXPLORED_AT_ONCE) + .cloned() + .collect(); + tried.extend(candidates.iter().cloned()); + candidates + } + /// `peer`, connected, identified itself at `listen` (`published`): its direct /// addresses are those from now on, while connected (`remember`); those through /// relays and by name stay. Kept in order of `preference`: of a peer with many, @@ -520,7 +556,7 @@ impl Book { } /// The IP address `address` starts with, if any. -fn ip_of(address: &Multiaddr) -> Option<std::net::IpAddr> { +pub(super) fn ip_of(address: &Multiaddr) -> Option<std::net::IpAddr> { match address.iter().next() { Some(Protocol::Ip4(ip)) => Some(ip.into()), Some(Protocol::Ip6(ip)) => Some(ip.into()), diff --git a/src/transport/libp2p/driver.rs b/src/transport/libp2p/driver.rs index 145592d..9b4259f 100644 --- a/src/transport/libp2p/driver.rs +++ b/src/transport/libp2p/driver.rs @@ -220,6 +220,7 @@ pub(super) async fn run( book.rtts.remove(&connection_id); if book.connection(&peer_id).is_none() { book.identified.remove(&peer_id); + book.explored.remove(&peer_id); } } let cause = match cause { @@ -267,9 +268,23 @@ pub(super) async fn run( if serving && !left_out && !is_local(&address) && !is_relayed(&address) { swarm.add_external_address(address.clone()); } - let mut book = lock(&book); - if !book.listening.contains(&address) { - book.listening.push(address); + let new_path = !left_out && !is_local(&address) && !is_relayed(&address); + { + let mut book = lock(&book); + if !book.listening.contains(&address) { + book.listening.push(address); + } + // A new interface (ethernet besides the WLAN, say): every + // address may be reached better now. + if new_path { + book.explored.clear(); + } + } + if new_path { + let peers: Vec<PeerId> = swarm.connected_peers().copied().collect(); + for peer in peers { + explore(&mut swarm, &book, peer, true); + } } } SwarmEvent::ExpiredListenAddr { address, .. } => { @@ -501,6 +516,7 @@ pub(super) async fn run( } } upgrade(&mut swarm, &book, peer_id, relaying.local); + explore(&mut swarm, &book, peer_id, false); } SwarmEvent::Behaviour(BehaviourEvent::Mdns( ::libp2p::mdns::Event::Discovered(found), diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs index 293558b..a4fefc3 100644 --- a/src/transport/libp2p/mod.rs +++ b/src/transport/libp2p/mod.rs @@ -32,7 +32,11 @@ //! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`), //! 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. +//! through a VPN, whichever opened first. A peer connected directly is dialled again +//! where it might be reached better (`explore`): at addresses it tells of later (a new +//! interface: identify pushes them at once), and at all of them when this node gains +//! an interface. Otherwise a connection made over the WLAN would stay there once +//! ethernet comes. //! //! 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 @@ -166,6 +170,8 @@ const RELAY_PAUSE: Duration = Duration::from_secs(600); /// (`upgrade`): first at once, then after a failure no sooner than this, doubling up to /// `RELAY_PAUSE`. const UPGRADE_RETRY: Duration = Duration::from_secs(30); +/// A directly connected peer's addresses dialled at once for a better path (`explore`). +const EXPLORED_AT_ONCE: usize = 4; /// A peer nobody mentioned for this long (the member list, identify, mDNS, a /// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted /// without their libp2p key, whose new ones the member list has since. Members are diff --git a/src/transport/libp2p/relays.rs b/src/transport/libp2p/relays.rs index dca031c..41d7e1d 100644 --- a/src/transport/libp2p/relays.rs +++ b/src/transport/libp2p/relays.rs @@ -175,6 +175,29 @@ pub(super) fn upgrade( } } +/// Dials `peer`, connected directly, at its other addresses, in case one is a better +/// path: one that came since (a new interface of its, identify tells), or, `again` +/// (one of ours came), every one, as our side may reach them better now, the one in +/// use too. Each once while connected (`Book::explored`); a connection that opens is +/// pinged at once, and of several the fastest is kept (`Book::duplicates`). Without +/// it, a connection stays on the path it was made on (the WLAN) when a better one comes +/// (ethernet): a connected peer isn't dialled again. +pub(super) fn explore(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, peer: PeerId, again: bool) { + let candidates = lock(book).to_explore(peer, again); + if candidates.is_empty() { + return; + } + debug!(peer = %Addressed(SCHEME, &peer), addresses = ?candidates, "Connected directly; dialling other addresses, for a better path"); + if let Err(error) = swarm.dial( + DialOpts::peer_id(peer) + .addresses(candidates) + .condition(PeerCondition::Always) + .build(), + ) { + debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Couldn't dial for a better path"); + } +} + /// Tries hole punching again with the peers it failed with that are due (`punching`). /// libp2p punches as a relayed connection opens, started by the side that was dialled /// through the relay; that side is where a punch's failure is told, so it is this one. diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs index 3d11913..ed23b34 100644 --- a/src/transport/libp2p/tests.rs +++ b/src/transport/libp2p/tests.rs @@ -696,6 +696,35 @@ fn ignored_addresses_not_kept() { assert!(!book.knows(them, &[via_vpn])); } +/// A node connected directly over the WLAN dials the ethernet address it later hears +/// of, once; and when an interface of ours comes, every address again, the one in use +/// too. +#[test] +fn better_paths_explored() { + let mut book = Book::default(); + let them = peer(); + let wlan: Multiaddr = "/ip4/192.168.1.7/tcp/9661".parse().unwrap(); + let ethernet: Multiaddr = "/ip4/192.168.2.7/tcp/9661".parse().unwrap(); + book.continuum.insert(them); + book.remember(them, wlan.clone()); + // Connected (inbound: from another port) over the WLAN. + book.connections.insert( + ConnectionId::new_unchecked(1), + (them, false, "/ip4/192.168.1.7/tcp/40000".parse().unwrap()), + ); + assert!(book.to_explore(them, false).is_empty()); + book.identified(them, vec![wlan.clone(), ethernet.clone()]); + assert_eq!(book.to_explore(them, false), [ethernet.clone()]); + assert!(book.to_explore(them, false).is_empty()); + // An interface of ours came (`driver`): all of them again. + book.explored.clear(); + let mut again = book.to_explore(them, true); + again.sort(); + let mut both = vec![wlan, ethernet]; + both.sort(); + assert_eq!(again, both); +} + /// Peers nobody mentions any more are forgotten, but connected ones. #[test] fn stale_peers_are_forgotten() { |