Network improvements (#374)

* Ping active connections

* Bump rust-libp2p

* Deny connection attempts if we're already connected

* Properly check max_peers against custom connections

* Fix set_until not immediately called

* Clean up unusued connections and stop pinging them

* Disconnect Kad connections after 20 seconds of inactivity

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 = "727e0e099b53a4032a7e2330994c819fe866add7", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
+libp2p = { git = "https://github.com/tomaka/libp2p-rs", rev = "38c9238a74504a6149909eb80f73608f63ab46c7", 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

@@ -23,6 +23,7 @@ use libp2p::kad::KadConnecController;
 use libp2p::peerstore::{Peerstore, PeerAccess};
 use libp2p::peerstore::json_peerstore::JsonPeerstore;
 use libp2p::peerstore::memory_peerstore::MemoryPeerstore;
+use libp2p::ping::Pinger;
 use libp2p::secio;
 use {Error, ErrorKind, NetworkConfiguration, NonReservedPeerMode};
 use {PeerId, ProtocolId, SessionInfo};
@@ -99,6 +100,9 @@ struct PeerConnectionInfo {
 	/// The Kademlia connection to this node.
 	kad_connec: UniqueConnec<KadConnecController>,
 
+	/// The ping connection to this node.
+	ping_connec: UniqueConnec<Pinger>,
+
 	/// Id of the peer.
 	id: PeerstorePeerId,
 
@@ -245,8 +249,7 @@ impl NetworkState {
 
 	/// Reports the ping of the peer. Returned later by `session_info()`.
 	/// No-op if the `peer_id` is not valid/expired.
-	#[allow(dead_code)]
-	pub fn report_ping(&self, peer_id: PeerId, ping: Duration) {
+	pub fn report_ping_duration(&self, peer_id: PeerId, ping: Duration) {
 		let connections = self.connections.read();
 		let info = match connections.info_by_peer.get(&peer_id) {
 			Some(info) => info,
@@ -426,19 +429,7 @@ impl NetworkState {
 	/// Returns the number of open and pending connections with
 	/// custom protocols.
 	pub fn num_open_custom_connections(&self) -> u32 {
-		self.connections
-			.read()
-			.info_by_peer
-			.values()
-			.filter(|info|
-				info.protocols.iter().any(|&(_, ref connec)|
-					match connec.state() {
-						UniqueConnecState::Pending | UniqueConnecState::Full => true,
-						_ => false
-					}
-				)
-			)
-			.count() as u32
+		num_open_custom_connections(&self.connections.read())
 	}
 
 	/// Returns the number of new outgoing custom connections to peers to
@@ -457,7 +448,7 @@ impl NetworkState {
 		connections.peer_by_nodeid.contains_key(node_id)
 	}
 
-	/// Obtains the `UniqueConnec` corresponding to the Kademlia connect to a peer.
+	/// Obtains the `UniqueConnec` corresponding to the Kademlia connection to a peer.
 	pub fn kad_connection(
 		&self,
 		node_id: PeerstorePeerId
@@ -473,6 +464,46 @@ impl NetworkState {
 		Ok((peer_id, connec))
 	}
 
+	/// Obtains the `UniqueConnec` corresponding to the Ping connection to a peer.
+	pub fn ping_connection(
+		&self,
+		node_id: PeerstorePeerId
+	) -> Result<(PeerId, UniqueConnec<Pinger>), IoError> {
+		let mut connections = self.connections.write();
+		let peer_id = accept_connection(&mut connections, &self.next_peer_id,
+			node_id, Endpoint::Listener)?;
+		let infos = connections.info_by_peer.get_mut(&peer_id)
+			.expect("Newly-created peer id is always valid");
+		let connec = infos.ping_connec.clone();
+		Ok((peer_id, connec))
+	}
+
+	/// Cleans up inactive connections and returns a list of
+	/// connections to ping.
+	pub fn cleanup_and_prepare_ping(
+		&self
+	) -> Vec<(PeerId, PeerstorePeerId, UniqueConnec<Pinger>)> {
+		let mut connections = self.connections.write();
+		let connections = &mut *connections;
+		let peer_by_nodeid = &mut connections.peer_by_nodeid;
+		let info_by_peer = &mut connections.info_by_peer;
+
+		let mut ret = Vec::with_capacity(info_by_peer.len());
+		info_by_peer.retain(|&peer_id, infos| {
+			// Remove the peer if neither Kad nor any protocol is alive.
+			if !infos.kad_connec.is_alive() &&
+				!infos.protocols.iter().any(|(_, conn)| conn.is_alive())
+			{
+				peer_by_nodeid.remove(&infos.id);
+				return false;
+			}
+
+			ret.push((peer_id, infos.id.clone(), infos.ping_connec.clone()));
+			true
+		});
+		ret
+	}
+
 	/// Try to add a new connection to a node in the list.
 	///
 	/// Returns a `PeerId` to allow further interfacing with this connection.
@@ -503,16 +534,15 @@ impl NetworkState {
 		let peer_id = accept_connection(&mut connections, &self.next_peer_id,
 			node_id.clone(), endpoint)?;
 
-		let connections = &mut *connections;
-		let info_by_peer = &mut connections.info_by_peer;
-		let peer_by_nodeid = &mut connections.peer_by_nodeid;
-		let infos = info_by_peer.get_mut(&peer_id)
+		let num_open_connections = num_open_custom_connections(&connections);
+
+		let infos = connections.info_by_peer.get_mut(&peer_id)
 			.expect("Newly-created peer id is always valid");
 
 		let node_is_reserved = self.reserved_peers.read().contains(&infos.id);
 		if !node_is_reserved {
 			if self.reserved_only.load(atomic::Ordering::Relaxed) ||
-				peer_by_nodeid.len() >= self.max_peers as usize
+				num_open_connections >= self.max_peers
 			{
 				debug!(target: "sub-libp2p", "Refusing node {:?} because we \
 					reached the max number of peers", node_id);
@@ -660,6 +690,7 @@ fn accept_connection(
 		info_by_peer.insert(new_id, PeerConnectionInfo {
 			protocols: Vec::new(),    // TODO: Vec::with_capacity(num_registered_protocols),
 			kad_connec: UniqueConnec::empty(),
+			ping_connec: UniqueConnec::empty(),
 			id: node_id.clone(),
 			originated: endpoint == Endpoint::Dialer,
 			ping: Mutex::new(None),
@@ -691,6 +722,23 @@ fn is_peer_disabled(
 	}
 }
 
+/// Returns the number of open and pending connections with
+/// custom protocols.
+fn num_open_custom_connections(connections: &Connections) -> u32 {
+	connections
+		.info_by_peer
+		.values()
+		.filter(|info|
+			info.protocols.iter().any(|&(_, ref connec)|
+				match connec.state() {
+					UniqueConnecState::Pending | UniqueConnecState::Full => true,
+					_ => false
+				}
+			)
+		)
+		.count() as u32
+}
+
 /// Parses an address of the form `/ip4/x.x.x.x/tcp/x/p2p/xxxxxx`, and adds it
 /// to the given peerstore. Returns the corresponding peer ID.
 fn parse_and_add_to_peerstore(addr_str: &str, peerstore: &PeersStorage)

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

@@ -27,7 +27,7 @@ use libp2p::identify::{IdentifyInfo, IdentifyOutput, IdentifyTransportOutcome};
 use libp2p::identify::{IdentifyProtocolConfig, PeerIdTransport};
 use libp2p::core::{upgrade, Transport, MuxedTransport, ConnectionUpgrade};
 use libp2p::core::{Endpoint, PeerId as PeerstorePeerId, PublicKey};
-use libp2p::core::SwarmController;
+use libp2p::core::{SwarmController, UniqueConnecState};
 use libp2p::ping;
 use libp2p::transport_timeout::TransportTimeout;
 use {PacketId, SessionInfo, ConnectionFilter, TimerToken};
@@ -43,7 +43,7 @@ use futures::{future, Future, Stream, IntoFuture};
 use futures::sync::{mpsc, oneshot};
 use tokio_core::reactor::{Core, Handle};
 use tokio_io::{AsyncRead, AsyncWrite};
-use tokio_timer;
+use tokio_timer::{Interval, Deadline};
 
 use custom_proto::{RegisteredProtocol, RegisteredProtocols};
 use custom_proto::RegisteredProtocolOutput;
@@ -445,10 +445,10 @@ fn init_thread(
 				move |out, endpoint, client_addr| {
 					let original_addr = out.original_addr;
 					let listener_upgrade = upgrade::or(upgrade::or(upgrade::or(
-						upgrade::map(shared.kad_upgrade.clone(), FinalUpgrade::Kad),
+						upgrade::map_with_addr(shared.kad_upgrade.clone(), |(c, f), a| FinalUpgrade::Kad(c, f, a.clone())),
 						upgrade::map(IdentifyProtocolConfig, |id| FinalUpgrade::Identify(id, original_addr))),
-						upgrade::map(ping::Ping, |(p, f)| FinalUpgrade::Ping(p, f))),
-						upgrade::map(DelayedProtosList(shared), FinalUpgrade::Custom));
+						upgrade::map_with_addr(ping::Ping, |(p, f), addr| FinalUpgrade::Ping(p, f, addr.clone()))),
+						upgrade::map_with_addr(DelayedProtosList(shared), |c, a| FinalUpgrade::Custom(c, a.clone())));
 					upgrade::apply(out.socket, listener_upgrade, endpoint, client_addr)
 				}
 			})
@@ -457,8 +457,8 @@ fn init_thread(
 
 		libp2p::core::swarm(
 			upgraded_transport,
-			move |(upgrade, endpoint), client_addr|
-				listener_handle(shared.clone(), upgrade, endpoint, client_addr)
+			move |(upgrade, endpoint), _client_addr|
+				listener_handle(shared.clone(), upgrade, endpoint)
 		)
 	};
 
@@ -523,12 +523,16 @@ fn init_thread(
 		});
 
 	// Start the process of periodically discovering nodes to connect to.
-	let discovery = start_kademlia_discovery(shared.clone(), transport,
-		swarm_controller);
+	let discovery = start_kademlia_discovery(shared.clone(),
+		transport.clone(), swarm_controller.clone());
+
+	// Start the process of pinging the active nodes on the network.
+	let pinger = start_pinger(shared.clone(), transport, swarm_controller);
 
 	// Merge all the futures into one!
 	Ok(swarm_future
 		.select(discovery).map_err(|(err, _)| err).and_then(|(_, rest)| rest)
+		.select(pinger).map_err(|(err, _)| err).and_then(|(_, rest)| rest)
 		.select(timeouts).map_err(|(err, _)| err).and_then(|(_, rest)| rest)
 		.select(close_rx.then(|_| Ok(()))).map(|_| ()).map_err(|(err, _)| err)
 
@@ -549,33 +553,30 @@ struct TransportOutput<S> {
 
 /// Enum of all the possible protocols our service handles.
 enum FinalUpgrade<C> {
-	Kad((KadConnecController, Box<Stream<Item = KadIncomingRequest, Error = IoError>>)),
+	Kad(KadConnecController, Box<Stream<Item = KadIncomingRequest, Error = IoError>>, Multiaddr),
 	/// The remote identification system, and the multiaddress we see the remote as.
 	Identify(IdentifyOutput<C>, Multiaddr),
-	Ping(ping::Pinger, Box<Future<Item = (), Error = IoError>>),
+	Ping(ping::Pinger, Box<Future<Item = (), Error = IoError>>, Multiaddr),
 	/// `Custom` means anything not in the core libp2p and is handled
 	/// by `CustomProtoConnectionUpgrade`.
-	Custom(RegisteredProtocolOutput<Arc<NetworkProtocolHandler + Send + Sync>>),
+	Custom(RegisteredProtocolOutput<Arc<NetworkProtocolHandler + Send + Sync>>, Multiaddr),
 }
 
 /// Called whenever we successfully open a multistream with a remote.
-fn listener_handle<'a, C, F>(
+fn listener_handle<'a, C>(
 	shared: Arc<Shared>,
 	upgrade: FinalUpgrade<C>,
 	endpoint: Endpoint,
-	client_addr: F,
 ) -> Box<Future<Item = (), Error = IoError> + 'a>
-	where C: AsyncRead + AsyncWrite + 'a,
-		F: Future<Item = Multiaddr, Error = IoError> + 'a {
+	where C: AsyncRead + AsyncWrite + 'a {
 	match upgrade {
-		FinalUpgrade::Kad((controller, kademlia_stream)) => {
+		FinalUpgrade::Kad(controller, kademlia_stream, client_addr) => {
 			trace!(target: "sub-libp2p", "Opened kademlia substream with \
-				remote as {:?}", endpoint);
-
-			let shared = shared.clone();
-			Box::new(client_addr.and_then(move |client_addr|
-				handle_kademlia_connection(shared, client_addr, controller, kademlia_stream)
-			).flatten())
+				{:?} as {:?}", client_addr, endpoint);
+			match handle_kademlia_connection(shared, client_addr, controller, kademlia_stream) {
+				Ok(fut) => Box::new(fut) as Box<_>,
+				Err(err) => Box::new(future::err(err)) as Box<_>,
+			}
 		},
 
 		FinalUpgrade::Identify(IdentifyOutput::Sender { sender }, original_addr) => {
@@ -597,13 +598,23 @@ fn listener_handle<'a, C, F>(
 		FinalUpgrade::Identify(IdentifyOutput::RemoteInfo { .. }, _) =>
 			unreachable!("We are never dialing with the identify protocol"),
 
-		FinalUpgrade::Ping(_pinger, future) => future,
+		FinalUpgrade::Ping(pinger, future, client_addr) => {
+			let node_id = p2p_multiaddr_to_node_id(client_addr);
+			match shared.network_state.ping_connection(node_id.clone()) {
+				Ok((_, ping_connec)) => {
+					trace!(target: "sub-libp2p", "Successfully opened ping \
+						substream with {:?}", node_id);
+					let fut = ping_connec.set_until(pinger, future);
+					Box::new(fut) as Box<_>
+				},
+				Err(err) => Box::new(future::err(err)) as Box<_>
+			}
+		},
 
-		FinalUpgrade::Custom(custom_proto_out) => {
+		FinalUpgrade::Custom(custom_proto_out, client_addr) => {
 			// A "custom" protocol is one that is part of substrate and not part of libp2p.
 			let shared = shared.clone();
-			let fut = client_addr.and_then(move |client_addr|
-				handle_custom_connection(shared, client_addr, endpoint, custom_proto_out));
+			let fut = handle_custom_connection(shared, client_addr, endpoint, custom_proto_out);
 			Box::new(fut) as Box<_>
 		},
 	}
@@ -619,19 +630,31 @@ fn handle_kademlia_connection(
 	let node_id = p2p_multiaddr_to_node_id(client_addr);
 	let (_peer_id, kad_connec) = shared.network_state
 		.kad_connection(node_id.clone())?;
-
-	let future = kademlia_stream.for_each({
+	
+	let future = future::loop_fn(kademlia_stream, move |kademlia_stream| {
 		let shared = shared.clone();
-		move |req| {
-			let shared = shared.clone();
-			shared.kad_system.update_kbuckets(node_id.clone());
-			match req {
-				KadIncomingRequest::FindNode { searched, responder } =>
-					responder.respond(build_kademlia_response(&shared, &searched)),
-				KadIncomingRequest::PingPong => (),
-			}
-			Ok(())
-		}
+		let node_id = node_id.clone();
+
+		let next = kademlia_stream
+			.into_future()
+			.map_err(|(err, _)| err);
+		let deadline = Instant::now() + Duration::from_secs(20);
+
+		Deadline::new(next, deadline)
+			.map_err(|err|
+				// TODO: improve the error reporting here, but tokio-timer's API is bad
+				IoError::new(IoErrorKind::Other, err)
+			)
+			.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::PingPong) => (),
+					None => return Ok(future::Loop::Break(()))
+				}
+				Ok(future::Loop::Continue(rest))
+			})
 	});
 
 	Ok(kad_connec.set_until(controller, future))
@@ -709,6 +732,12 @@ fn handle_custom_connection(
 		Err(err) => return future::Either::A(future::err(err.into())),
 	};
 
+	if let UniqueConnecState::Full = unique_connec.state() {
+		debug!(target: "sub-libp2p", "Interrupting connection attempt to {:?} \
+			with {:?} because we're already connected", node_id, custom_proto_out.protocol_id);
+		return future::Either::A(future::ok(()))
+	}
+
 	struct ProtoDisconnectGuard {
 		inner: Arc<Shared>,
 		peer_id: PeerId,
@@ -820,7 +849,7 @@ fn start_kademlia_discovery<T, To, St, C>(shared: Arc<Shared>, transport: T,
 			)
 	});
 
-	let discovery = tokio_timer::Interval::new(Instant::now(), Duration::from_secs(30))
+	let discovery = Interval::new(Instant::now(), Duration::from_secs(30))
 		// TODO: add a timeout to the lookups
 		.map_err(|err| IoError::new(IoErrorKind::Other, err))
 		.and_then({
@@ -1011,7 +1040,10 @@ fn open_peer_custom_proto<T, To, St, C>(
 				)
 		})
 		.and_then(move |out, endpoint, client_addr|
-			future::ok(((FinalUpgrade::Custom(out), endpoint), client_addr))
+			client_addr.map(move |client_addr| {
+				let out = FinalUpgrade::Custom(out, client_addr.clone());
+				((out, endpoint), future::ok(client_addr))
+			})
 		);
 	
 	let with_timeout = TransportTimeout::new(with_proto,
@@ -1048,9 +1080,12 @@ fn obtain_kad_connection<T, To, St, C>(shared: Arc<Shared>,
 			upgrade::apply(out.socket, kad_upgrade.clone(),
 				endpoint, client_addr)
 		)
-		.map(move |(kad_ctrl, stream), _|
-			(FinalUpgrade::Kad((kad_ctrl, stream)), Endpoint::Dialer)
-		);
+		.and_then(move |(ctrl, fut), _, client_addr| {
+			client_addr.map(|client_addr| {
+				let out = FinalUpgrade::Kad(ctrl, fut, client_addr.clone());
+				((out, Endpoint::Dialer), future::ok(client_addr))
+			})
+		});
 	
 	shared.network_state
 		.kad_connection(peer_id.clone())
@@ -1103,6 +1138,106 @@ fn process_identify_info(
 	Ok(())
 }
 
+/// Returns a future that regularly pings every peer we're connected to.
+/// If a peer doesn't respond after a while, we disconnect it.
+fn start_pinger<T, To, St, C>(
+	shared: Arc<Shared>,
+	transport: T,
+	swarm_controller: SwarmController<St>
+) -> impl Future<Item = (), Error = IoError>
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
+		T::MultiaddrFuture: 'static,
+		To: AsyncRead + AsyncWrite + 'static,
+		St: MuxedTransport<Output = (FinalUpgrade<C>, Endpoint)> + Clone + 'static,
+		C: 'static {
+	let transport = transport
+		.and_then(move |out, endpoint, client_addr|
+			upgrade::apply(out.socket, ping::Ping, endpoint, client_addr)
+		)
+		.and_then(move |(ctrl, fut), _, client_addr| {
+			client_addr.map(|client_addr| {
+				let out = FinalUpgrade::Ping(ctrl, fut, client_addr.clone());
+				((out, Endpoint::Dialer), future::ok(client_addr))
+			})
+		});
+
+	let fut = Interval::new(Instant::now(), Duration::from_secs(30))
+		.map_err(|err| IoError::new(IoErrorKind::Other, err))
+		.for_each(move |_|
+			ping_all(shared.clone(), transport.clone(), &swarm_controller))
+		.then(|val| {
+			warn!(target: "sub-libp2p", "Pinging stream has stopped: {:?}", val);
+			val
+		});
+
+	// Note that we use a Box in order to speed compilation time.
+	Box::new(fut) as Box<Future<Item = _, Error = _>>
+}
+
+/// Pings all the nodes we're connected to and disconnects any node that
+/// doesn't respond. Returns a `Future` when all the pings have either
+/// suceeded or timed out.
+fn ping_all<T, St, C>(
+	shared: Arc<Shared>,
+	transport: T,
+	swarm_controller: &SwarmController<St>
+) -> impl Future<Item = (), Error = IoError>
+	where T: MuxedTransport<Output = (FinalUpgrade<C>, Endpoint)> + Clone + 'static,
+		T::MultiaddrFuture: 'static,
+		St: MuxedTransport<Output = (FinalUpgrade<C>, Endpoint)> + Clone + 'static,
+		C: 'static {
+	let mut ping_futures = Vec::new();
+
+	for (peer, peer_id, pinger) in shared.network_state.cleanup_and_prepare_ping() {
+		let shared = shared.clone();
+
+		let addr = Multiaddr::from(AddrComponent::P2P(peer_id.clone().into_bytes()));
+		let fut = pinger
+			.get_or_dial(&swarm_controller, &addr, transport.clone())
+			.and_then(move |mut p| {
+				trace!(target: "sub-libp2p",
+					"Pinging active connection with #{} {:?}", peer, peer_id);
+				p.ping()
+					.map(|()| peer_id)
+					.map_err(|err| IoError::new(IoErrorKind::Other, err))
+			});
+		let ping_start_time = Instant::now();
+		let fut = Deadline::new(fut, ping_start_time + Duration::from_secs(30))
+			.then(move |val|
+				match val {
+					Err(err) => {
+						trace!(target: "sub-libp2p",
+							"Error while pinging #{:?} => {:?}", peer, err);
+						shared.network_state.disconnect_peer(peer);
+						// Return Ok, otherwise we would close the ping service
+						Ok(())
+					},
+					Ok(peer_id) => {
+						let elapsed = ping_start_time.elapsed();
+						trace!(target: "sub-libp2p", "Pong from #{:?} in {:?}",
+							peer, elapsed);
+						shared.network_state.report_ping_duration(peer, elapsed);
+						shared.kad_system.update_kbuckets(peer_id);
+						Ok(())
+					}
+				}
+			);
+		ping_futures.push(fut);
+	}
+
+	future::loop_fn(ping_futures, |ping_futures| {
+		if ping_futures.is_empty() {
+			let fut = future::ok(future::Loop::Break(()));
+			return future::Either::A(fut)
+		}
+
+		let fut = future::select_all(ping_futures)
+			.map(|((), _, rest)| future::Loop::Continue(rest))
+			.map_err(|(err, _, _)| err);
+		future::Either::B(fut)
+	})
+}
+
 /// Expects a multiaddr of the format `/p2p/<node_id>` and returns the node ID.
 /// Panics if the format is not correct.
 fn p2p_multiaddr_to_node_id(client_addr: Multiaddr) -> PeerstorePeerId {

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

@@ -35,7 +35,7 @@ pub fn build_transport(
 
 			let mut plaintext = upgrade::toggleable(upgrade::PlainTextConfig);
 			match unencrypted_allowed {
-				UnencryptedAllowed::Allowed => plaintext.disable(),
+				UnencryptedAllowed::Allowed => (),//plaintext.disable(),
 				UnencryptedAllowed::Denied => (),
 			};