Improvements to the Kademlia system (#688)

substrate/substrate/network-libp2p/Cargo.toml

@@ -11,7 +11,7 @@ bytes = "0.4"
 error-chain = { version = "0.12", default-features = false }
 fnv = "1.0"
 futures = "0.1"
-libp2p = { git = "https://github.com/tomaka/libp2p-rs", rev = "8ac9b65e669ebbd9deeffa9d4566c95f631cfed5", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
+libp2p = { git = "https://github.com/libp2p/rust-libp2p", rev = "304e9c72c88bc97824f2734dc19d1b5f4556d346", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
 ethcore-io = { git = "https://github.com/paritytech/parity.git" }
 ethkey = { git = "https://github.com/paritytech/parity.git" }
 ethereum-types = "0.3"

substrate/substrate/network-libp2p/src/network_state.rs

@@ -253,17 +253,6 @@ impl NetworkState {
 		(list, change)
 	}
 
-	/// Returns all the IDs of the peers on the network we have knowledge of.
-	///
-	/// This includes peers we are not connected to.
-	pub fn known_peers(&self) -> Vec<PeerId> {
-		let topology = self.topology.read();
-		// Note: I have no idea why, but fusing the two lines below fails the
-		// borrow check
-		let out: Vec<_> = topology.peers().cloned().collect();
-		out
-	}
-
 	/// Returns true if we are connected to any peer at all.
 	pub fn has_connected_peer(&self) -> bool {
 		!self.connections.read().peer_by_nodeid.is_empty()
@@ -350,16 +339,21 @@ impl NetworkState {
 
 	/// Adds an address discovered by Kademlia.
 	/// Note that we don't have to be connected to a peer to add an address.
-	pub fn add_kad_discovered_addr(&self, node_id: &PeerId, addr: Multiaddr) {
-		self.topology.write().add_kademlia_discovered_addr(node_id, addr)
+	/// If `connectable` is `true`, that means we have a hint from a remote that this node can be
+	/// connected to.
+	pub fn add_kad_discovered_addr(&self, node_id: &PeerId, addr: Multiaddr, connectable: bool) {
+		self.topology.write().add_kademlia_discovered_addr(node_id, addr, connectable)
 	}
 
 	/// Returns the known multiaddresses of a peer.
-	pub fn addrs_of_peer(&self, node_id: &PeerId) -> Vec<Multiaddr> {
+	///
+	/// The boolean associated to each address indicates whether we're connected to it.
+	pub fn addrs_of_peer(&self, node_id: &PeerId) -> Vec<(Multiaddr, bool)> {
 		let topology = self.topology.read();
 		// Note: I have no idea why, but fusing the two lines below fails the
 		// borrow check
-		let out: Vec<Multiaddr> = topology.addrs_of_peer(node_id).cloned().collect();
+		let out: Vec<_> = topology
+			.addrs_of_peer(node_id).map(|(a, c)| (a.clone(), c)).collect();
 		out
 	}
 
@@ -424,12 +418,6 @@ impl NetworkState {
 		}
 	}
 
-	/// Returns true if we are connected to the given node.
-	pub fn has_connection(&self, node_id: &PeerId) -> bool {
-		let connections = self.connections.read();
-		connections.peer_by_nodeid.contains_key(node_id)
-	}
-
 	/// Reports that we tried to connect to the given address but failed.
 	///
 	/// This decreases the chance this address will be tried again in the future.

substrate/substrate/network-libp2p/src/service.rs

@@ -33,6 +33,7 @@ use libp2p::transport_timeout::TransportTimeout;
 use {PacketId, SessionInfo, TimerToken};
 use rand;
 use std::io::{Error as IoError, ErrorKind as IoErrorKind};
+use std::iter;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::sync::mpsc as sync_mpsc;
@@ -119,7 +120,7 @@ impl NetworkService {
 			local_peer_id: local_peer_id.clone(),
 			kbuckets_timeout: Duration::from_secs(600),
 			request_timeout: Duration::from_secs(10),
-			known_initial_peers: network_state.known_peers().into_iter(),
+			known_initial_peers: iter::empty(),
 		});
 
 		// Channel we use to signal success or failure of the bg thread
@@ -703,8 +704,11 @@ fn handle_kademlia_connection(
 			.and_then(move |(req, rest)| {
 				shared.kad_system.update_kbuckets(node_id);
 				match req {
-					Some(KadIncomingRequest::FindNode { searched, responder }) =>
-						responder.respond(build_kademlia_response(&shared, &searched)),
+					Some(KadIncomingRequest::FindNode { searched, responder }) => {
+						let resp = build_kademlia_response(&shared, &searched);
+						trace!(target: "sub-libp2p", "Responding to Kad {:?} with {:?}", searched, resp);
+						responder.respond(resp)
+					},
 					Some(KadIncomingRequest::PingPong) => (),
 					None => return Ok(future::Loop::Break(()))
 				}
@@ -734,19 +738,25 @@ fn build_kademlia_response(
 					connection_ty: KadConnectionType::Connected,
 				}
 			} else {
-				let addrs = shared.network_state.addrs_of_peer(&who);
-				let connec_ty = if shared.network_state.has_connection(&who) {
-					// TODO: this only checks connections with substrate ; but what
-					//       if we're connected through Kademlia only?
-					KadConnectionType::Connected
-				} else {
-					KadConnectionType::NotConnected
-				};
+				let mut addrs = shared.network_state.addrs_of_peer(&who);
+				let connected = addrs.iter().any(|&(_, conn)| conn);
+				// The Kademlia protocol of libp2p doesn't allow specifying which address is valid
+				// and which is outdated, therefore in order to stay honest towards the network
+				// we only report the addresses we're connected to if we're connected to any.
+				if connected {
+					addrs = addrs.into_iter()
+						.filter_map(|(a, c)| if c { Some((a, c)) } else { None })
+						.collect();
+				}
 
 				KadPeer {
 					node_id: who.clone(),
-					multiaddrs: addrs,
-					connection_ty: connec_ty,
+					multiaddrs: addrs.into_iter().map(|(a, _)| a).collect(),
+					connection_ty: if connected {
+						KadConnectionType::Connected
+					} else {
+						KadConnectionType::NotConnected
+					},
 				}
 			}
 		})
@@ -980,10 +990,21 @@ fn perform_kademlia_query<T, To, St, C>(
 		})
 		.filter_map(move |event|
 			match event {
-				KadQueryEvent::NewKnownMultiaddrs(peers) => {
-					for (peer, addrs) in peers {
-						for addr in addrs {
-							shared.network_state.add_kad_discovered_addr(&peer, addr);
+				KadQueryEvent::PeersReported(peers) => {
+					for peer in peers {
+						let connected = match peer.connection_ty {
+							KadConnectionType::NotConnected => false,
+							KadConnectionType::Connected => true,
+							KadConnectionType::CanConnect => true,
+							KadConnectionType::CannotConnect => continue,
+						};
+
+						for addr in peer.multiaddrs {
+							shared.network_state.add_kad_discovered_addr(
+								&peer.node_id,
+								addr,
+								connected
+							);
 						}
 					}
 					None
@@ -1167,7 +1188,7 @@ fn obtain_kad_connection<T, To, St, C>(
 			Ok((kad, addr))
 		})
 		.and_then(move |(unique_connec, addr)| {
-			unique_connec.dial(&swarm_controller, &addr, transport.clone())
+			unique_connec.dial(&swarm_controller, &addr.0, transport.clone())
 		})
 		.then(|result| -> Result<_, ()> { Ok(result.ok()) })
 		.filter_map(|result| result)
@@ -1215,7 +1236,7 @@ fn process_identify_info(
 
 	for addr in info.listen_addrs.iter() {
 		if let Some(node_id) = shared.network_state.node_id_from_index(node_index) {
-			shared.network_state.add_kad_discovered_addr(&node_id, addr.clone());
+			shared.network_state.add_kad_discovered_addr(&node_id, addr.clone(), true);
 		}
 	}
 }

substrate/substrate/network-libp2p/src/topology.rs

@@ -27,7 +27,12 @@ use std::time::{Duration, Instant, SystemTime};
 const CONNEC_DURATION_PER_SCORE: Duration = Duration::from_secs(10);
 /// Maximum value for the score.
 const MAX_SCORE: u32 = 100;
-/// Initial score that a node discovered through Kademlia receives.
+/// When we successfully connect to a node, raises its score to the given minimum value.
+const CONNECTED_MINIMUM_SCORE: u32 = 20;
+/// Initial score that a node discovered through Kademlia receives, where we have a hint that the
+/// node is reachable.
+const KADEMLIA_DISCOVERY_INITIAL_SCORE_CONNECTABLE: u32 = 15;
+/// Initial score that a node discovered through Kademlia receives, without any hint.
 const KADEMLIA_DISCOVERY_INITIAL_SCORE: u32 = 10;
 /// Score adjustement when we fail to connect to an address.
 const SCORE_DIFF_ON_FAILED_TO_CONNECT: i32 = -1;
@@ -112,23 +117,17 @@ impl NetTopology {
 		let now_systime = SystemTime::now();
 		self.store.retain(|_, peer| {
 			peer.addrs.retain(|a| {
-				a.expires > now_systime
+				a.expires > now_systime || a.is_connected()
 			});
 			!peer.addrs.is_empty()
 		});
 	}
 
-	/// Returns a list of all the known peers.
-	pub fn peers(&self) -> impl Iterator<Item = &PeerId> {
-		self.store.keys()
-	}
-
 	/// Returns the known potential addresses of a peer, ordered by score.
 	///
-	/// If we're already connected to that peer, the address(es) we're connected with will be at
-	/// the top of the list.
+	/// The boolean associated to each address indicates whether we're connected to it.
 	// TODO: filter out backed off ones?
-	pub fn addrs_of_peer(&self, peer: &PeerId) -> impl Iterator<Item = &Multiaddr> {
+	pub fn addrs_of_peer(&self, peer: &PeerId) -> impl Iterator<Item = (&Multiaddr, bool)> {
 		let peer = if let Some(peer) = self.store.get(peer) {
 			peer
 		} else {
@@ -140,14 +139,14 @@ impl NetTopology {
 		let mut list = peer.addrs.iter().filter_map(move |addr| {
 			let (score, connected) = addr.score_and_is_connected();
 			if (addr.expires >= now && score > 0) || connected {
-				Some((score, &addr.addr))
+				Some((score, connected, &addr.addr))
 			} else {
 				None
 			}
 		}).collect::<Vec<_>>();
 		list.sort_by(|a, b| a.0.cmp(&b.0));
 		// TODO: meh, optimize
-		let l = list.into_iter().map(|(_, addr)| addr).collect::<Vec<_>>();
+		let l = list.into_iter().map(|(_, connec, addr)| (addr, connec)).collect::<Vec<_>>();
 		l.into_iter()
 	}
 
@@ -195,7 +194,7 @@ impl NetTopology {
 
 		let mut found = false;
 		peer.addrs.retain(|a| {
-			if a.expires < now_systime {
+			if a.expires < now_systime && !a.is_connected() {
 				return false;
 			}
 			if a.addr == addr {
@@ -222,7 +221,15 @@ impl NetTopology {
 	/// Adds an address discovered through the Kademlia DHT.
 	///
 	/// This address is not necessarily valid and should expire after a TTL.
-	pub fn add_kademlia_discovered_addr(&mut self, peer_id: &PeerId, addr: Multiaddr) {
+	///
+	/// If `connectable` is true, that means we have some sort of hint that this node can
+	/// be reached.
+	pub fn add_kademlia_discovered_addr(
+		&mut self,
+		peer_id: &PeerId,
+		addr: Multiaddr,
+		connectable: bool
+	) {
 		let now_systime = SystemTime::now();
 		let now = Instant::now();
 
@@ -230,7 +237,7 @@ impl NetTopology {
 
 		let mut found = false;
 		peer.addrs.retain(|a| {
-			if a.expires < now_systime {
+			if a.expires < now_systime && !a.is_connected() {
 				return false;
 			}
 			if a.addr == addr {
@@ -240,7 +247,20 @@ impl NetTopology {
 		});
 
 		if !found {
-			trace!(target: "sub-libp2p", "Peer store: adding address {} for {:?}", addr, peer_id);
+			trace!(
+				target: "sub-libp2p",
+				"Peer store: adding address {} for {:?} (connectable hint: {:?})",
+				addr,
+				peer_id,
+				connectable
+			);
+
+			let initial_score = if connectable {
+				KADEMLIA_DISCOVERY_INITIAL_SCORE_CONNECTABLE
+			} else {
+				KADEMLIA_DISCOVERY_INITIAL_SCORE
+			};
+
 			peer.addrs.push(Addr {
 				addr,
 				expires: now_systime + KADEMLIA_DISCOVERY_EXPIRATION,
@@ -248,7 +268,7 @@ impl NetTopology {
 				next_back_off: FIRST_CONNECT_FAIL_BACKOFF,
 				score: Mutex::new(AddrScore {
 					connected_since: None,
-					score: KADEMLIA_DISCOVERY_INITIAL_SCORE,
+					score: initial_score,
 					latest_score_update: now,
 				}),
 			});
@@ -269,7 +289,7 @@ impl NetTopology {
 		for (peer_in_store, info_in_store) in self.store.iter_mut() {
 			if peer == peer_in_store {
 				if let Some(addr) = info_in_store.addrs.iter_mut().find(|a| &a.addr == addr) {
-					addr.connected_now();
+					addr.connected_now(CONNECTED_MINIMUM_SCORE);
 					addr.back_off_until = now;
 					addr.next_back_off = FIRST_CONNECT_FAIL_BACKOFF;
 					continue;
@@ -284,7 +304,7 @@ impl NetTopology {
 					score: Mutex::new(AddrScore {
 						connected_since: Some(now),
 						latest_score_update: now,
-						score: KADEMLIA_DISCOVERY_INITIAL_SCORE
+						score: CONNECTED_MINIMUM_SCORE,
 					}),
 				});
 
@@ -394,12 +414,16 @@ struct AddrScore {
 }
 
 impl Addr {
-	/// Sets the addr to connected.
-	fn connected_now(&self) {
+	/// Sets the addr to connected. If the score is lower than the given value, raises it to this
+	/// value.
+	fn connected_now(&self, raise_to_min: u32) {
 		let mut score = self.score.lock();
 		let now = Instant::now();
 		Addr::flush(&mut score, now);
 		score.connected_since = Some(now);
+		if score.score < raise_to_min {
+			score.score = raise_to_min;
+		}
 	}
 
 	/// Applies a modification to the score.
@@ -425,6 +449,12 @@ impl Addr {
 		}
 	}
 
+	/// Returns true if we are connected to this addr.
+	fn is_connected(&self) -> bool {
+		let score = self.score.lock();
+		score.connected_since.is_some()
+	}
+
 	/// Returns the score, and true if we are connected to this addr.
 	fn score_and_is_connected(&self) -> (u32, bool) {
 		let mut score = self.score.lock();
@@ -612,7 +642,7 @@ fn serialize<W: Write>(out: W, map: &FnvHashMap<PeerId, PeerInfo>) -> Result<(),
 		let peer = peer.to_base58();
 		let info = SerializedPeerInfo {
 			addrs: info.addrs.iter()
-				.filter(|a| a.expires > now)
+				.filter(|a| a.expires > now || a.is_connected())
 				.map(Into::into)
 				.collect(),
 		};