libp2p-next (#3076)

* Changes for the next libp2p release:

  * Updates to the Kademlia APIs.
  * Updated imports due to the extracted libp2p-swarm crate.
  * ...

Still pending at least the following:

  * rust-libp2p/#1189
  * rust-libp2p/#1191
  * rust-libp2p/#1194

* Use Quorum::One.

The previous choice was apparently arbitrary.

* Use libp2p-0.11 from crates.io. Address feedback.

* Correct imports after merge.

substrate/core/network/src/behaviour.rs

@@ -23,7 +23,7 @@ use crate::protocol::{CustomMessageOutcome, Protocol};
 use futures::prelude::*;
 use libp2p::NetworkBehaviour;
 use libp2p::core::{Multiaddr, PeerId, PublicKey};
-use libp2p::core::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess};
+use libp2p::swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess};
 use libp2p::core::{nodes::Substream, muxing::StreamMuxerBox};
 use libp2p::multihash::Multihash;
 use log::warn;
@@ -150,6 +150,12 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviourEventPr
 	for Behaviour<B, S, H> {
 	fn inject_event(&mut self, out: DiscoveryOut) {
 		match out {
+			DiscoveryOut::UnroutablePeer(_peer_id) => {
+				// Obtaining and reporting listen addresses for unroutable peers back
+				// to Kademlia is handled by the `Identify` protocol, part of the
+				// `DebugInfoBehaviour`. See the `NetworkBehaviourEventProcess`
+				// implementation for `DebugInfoEvent`.
+			}
 			DiscoveryOut::Discovered(peer_id) => {
 				self.substrate.add_discovered_nodes(iter::once(peer_id));
 			}

substrate/core/network/src/custom_proto/behaviour.rs

@@ -20,8 +20,8 @@ use crate::custom_proto::upgrade::{CustomMessage, RegisteredProtocol};
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use futures03::{compat::Compat, TryFutureExt as _, StreamExt as _, TryStreamExt as _};
-use libp2p::core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
-use libp2p::core::{Multiaddr, PeerId};
+use libp2p::core::{ConnectedPoint, Multiaddr, PeerId};
+use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
 use log::{debug, error, trace, warn};
 use smallvec::SmallVec;
 use std::{borrow::Cow, collections::hash_map::Entry, cmp, error, marker::PhantomData, mem, pin::Pin};

substrate/core/network/src/custom_proto/handler.rs

@@ -19,13 +19,14 @@ use crate::custom_proto::upgrade::{RegisteredProtocolEvent, RegisteredProtocolSu
 use futures::prelude::*;
 use futures03::{compat::Compat, TryFutureExt as _};
 use futures_timer::Delay;
-use libp2p::core::{
-	ConnectedPoint, PeerId, Endpoint, ProtocolsHandler, ProtocolsHandlerEvent,
-	protocols_handler::IntoProtocolsHandler,
-	protocols_handler::KeepAlive,
-	protocols_handler::ProtocolsHandlerUpgrErr,
-	protocols_handler::SubstreamProtocol,
-	upgrade::{InboundUpgrade, OutboundUpgrade}
+use libp2p::core::{ConnectedPoint, PeerId, Endpoint};
+use libp2p::core::upgrade::{InboundUpgrade, OutboundUpgrade};
+use libp2p::swarm::{
+	ProtocolsHandler, ProtocolsHandlerEvent,
+	IntoProtocolsHandler,
+	KeepAlive,
+	ProtocolsHandlerUpgrErr,
+	SubstreamProtocol,
 };
 use log::{debug, error};
 use smallvec::{smallvec, SmallVec};

substrate/core/network/src/custom_proto/tests.rs

@@ -17,11 +17,10 @@
 #![cfg(test)]
 
 use futures::{future, prelude::*, try_ready};
-use libp2p::core::{nodes::Substream, swarm::Swarm};
-use libp2p::core::{transport::boxed::Boxed, muxing::StreamMuxerBox};
-use libp2p::core::{ProtocolsHandler, protocols_handler::IntoProtocolsHandler};
-use libp2p::core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction};
-use libp2p::core::swarm::PollParameters;
+use libp2p::core::nodes::Substream;
+use libp2p::core::{ConnectedPoint, transport::boxed::Boxed, muxing::StreamMuxerBox};
+use libp2p::swarm::{Swarm, ProtocolsHandler, IntoProtocolsHandler};
+use libp2p::swarm::{PollParameters, NetworkBehaviour, NetworkBehaviourAction};
 use libp2p::{PeerId, Multiaddr, Transport};
 use rand::seq::SliceRandom;
 use std::{io, time::Duration, time::Instant};
@@ -84,7 +83,7 @@ fn build_nodes<T: CustomMessage + Send + 'static>()
 				.collect(),
 		};
 
-		let mut swarm = libp2p::core::swarm::Swarm::new(
+		let mut swarm = Swarm::new(
 			transport,
 			behaviour,
 			keypairs[index].public().into_peer_id()

substrate/core/network/src/debug_info.rs

@@ -18,10 +18,9 @@ use fnv::FnvHashMap;
 use futures::prelude::*;
 use futures03::{StreamExt as _, TryStreamExt as _};
 use libp2p::Multiaddr;
-use libp2p::core::{either::EitherOutput, PeerId, PublicKey};
-use libp2p::core::protocols_handler::{IntoProtocolsHandler, IntoProtocolsHandlerSelect, ProtocolsHandler};
-use libp2p::core::nodes::ConnectedPoint;
-use libp2p::core::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
+use libp2p::core::{ConnectedPoint, either::EitherOutput, PeerId, PublicKey};
+use libp2p::swarm::{IntoProtocolsHandler, IntoProtocolsHandlerSelect, ProtocolsHandler};
+use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
 use libp2p::identify::{Identify, IdentifyEvent, protocol::IdentifyInfo};
 use libp2p::ping::{Ping, PingConfig, PingEvent, PingSuccess};
 use log::{debug, trace, error};

substrate/core/network/src/discovery.rs

@@ -48,18 +48,21 @@
 use futures::prelude::*;
 use futures_timer::Delay;
 use futures03::{compat::Compat, TryFutureExt as _};
-use libp2p::core::{Multiaddr, PeerId, ProtocolsHandler, PublicKey};
-use libp2p::core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction};
-use libp2p::core::swarm::PollParameters;
+use libp2p::core::{ConnectedPoint, Multiaddr, PeerId, PublicKey};
+use libp2p::swarm::{ProtocolsHandler, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
+use libp2p::kad::{Kademlia, KademliaEvent, Quorum, Record};
+use libp2p::kad::GetClosestPeersError;
+use libp2p::kad::record::store::MemoryStore;
 #[cfg(not(target_os = "unknown"))]
-use libp2p::core::{swarm::toggle::Toggle, nodes::Substream, muxing::StreamMuxerBox};
-use libp2p::kad::{GetValueResult, Kademlia, KademliaOut, PutValueResult};
+use libp2p::{swarm::toggle::Toggle};
+#[cfg(not(target_os = "unknown"))]
+use libp2p::core::{nodes::Substream, muxing::StreamMuxerBox};
 #[cfg(not(target_os = "unknown"))]
 use libp2p::mdns::{Mdns, MdnsEvent};
 use libp2p::multihash::Multihash;
 use libp2p::multiaddr::Protocol;
 use log::{debug, info, trace, warn};
-use std::{cmp, collections::VecDeque, num::NonZeroU8, time::Duration};
+use std::{cmp, collections::VecDeque, time::Duration};
 use tokio_io::{AsyncRead, AsyncWrite};
 
 /// Implementation of `NetworkBehaviour` that discovers the nodes on the network.
@@ -68,7 +71,7 @@ pub struct DiscoveryBehaviour<TSubstream> {
 	/// reserved nodes.
 	user_defined: Vec<(PeerId, Multiaddr)>,
 	/// Kademlia requests and answers.
-	kademlia: Kademlia<TSubstream>,
+	kademlia: Kademlia<TSubstream, MemoryStore>,
 	/// Discovers nodes on the local network.
 	#[cfg(not(target_os = "unknown"))]
 	mdns: Toggle<Mdns<Substream<StreamMuxerBox>>>,
@@ -98,7 +101,9 @@ impl<TSubstream> DiscoveryBehaviour<TSubstream> {
 			warn!(target: "sub-libp2p", "mDNS is not available on this platform");
 		}
 
-		let mut kademlia = Kademlia::new(local_public_key.clone().into_peer_id());
+		let local_id = local_public_key.clone().into_peer_id();
+		let store = MemoryStore::new(local_id.clone());
+		let mut kademlia = Kademlia::new(local_id.clone(), store);
 		for (peer_id, addr) in &user_defined {
 			kademlia.add_address(peer_id, addr.clone());
 		}
@@ -155,8 +160,7 @@ impl<TSubstream> DiscoveryBehaviour<TSubstream> {
 	///
 	/// A corresponding `ValueFound` or `ValueNotFound` event will later be generated.
 	pub fn get_value(&mut self, key: &Multihash) {
-		self.kademlia.get_value(key, NonZeroU8::new(10)
-								.expect("Casting 10 to NonZeroU8 should succeed; qed"));
+		self.kademlia.get_record(key, Quorum::One)
 	}
 
 	/// Start putting a record into the DHT. Other nodes can later fetch that value with
@@ -164,15 +168,24 @@ impl<TSubstream> DiscoveryBehaviour<TSubstream> {
 	///
 	/// A corresponding `ValuePut` or `ValuePutFailed` event will later be generated.
 	pub fn put_value(&mut self, key: Multihash, value: Vec<u8>) {
-		self.kademlia.put_value(key, value);
+		self.kademlia.put_record(Record::new(key, value), Quorum::All);
 	}
 }
 
 /// Event generated by the `DiscoveryBehaviour`.
 pub enum DiscoveryOut {
-	/// We have discovered a node. Can be called multiple times with the same identity.
+	/// The address of a peer has been added to the Kademlia routing table.
+	///
+	/// Can be called multiple times with the same identity.
 	Discovered(PeerId),
 
+	/// A peer connected to this node for whom no listen address is known.
+	///
+	/// In order for the peer to be added to the Kademlia routing table, a known
+	/// listen address must be added via [`DiscoveryBehaviour::add_self_reported_address`],
+	/// e.g. obtained through the `identify` protocol.
+	UnroutablePeer(PeerId),
+
 	/// The DHT yeided results for the record request, grouped in (key, value) pairs.
 	ValueFound(Vec<(Multihash, Vec<u8>)>),
 
@@ -190,7 +203,7 @@ impl<TSubstream> NetworkBehaviour for DiscoveryBehaviour<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
 {
-	type ProtocolsHandler = <Kademlia<TSubstream> as NetworkBehaviour>::ProtocolsHandler;
+	type ProtocolsHandler = <Kademlia<TSubstream, MemoryStore> as NetworkBehaviour>::ProtocolsHandler;
 	type OutEvent = DiscoveryOut;
 
 	fn new_handler(&mut self) -> Self::ProtocolsHandler {
@@ -272,9 +285,11 @@ where
 					let random_peer_id = PeerId::random();
 					debug!(target: "sub-libp2p", "Libp2p <= Starting random Kademlia request for \
 						{:?}", random_peer_id);
-					self.kademlia.find_node(random_peer_id);
 
-					// Reset the `Delay` to the next random.
+					self.kademlia.get_closest_peers(random_peer_id);
+
+					// Schedule the next random query with exponentially increasing delay,
+					// capped at 60 seconds.
 					self.next_kad_random_query = Delay::new(self.duration_to_next_kad).compat();
 					self.duration_to_next_kad = cmp::min(self.duration_to_next_kad * 2,
 						Duration::from_secs(60));
@@ -290,50 +305,74 @@ where
 		loop {
 			match self.kademlia.poll(params) {
 				Async::NotReady => break,
-				Async::Ready(NetworkBehaviourAction::GenerateEvent(ev)) => {
-					match ev {
-						KademliaOut::Discovered { .. } => {}
-						KademliaOut::KBucketAdded { peer_id, .. } => {
-							let ev = DiscoveryOut::Discovered(peer_id);
-							return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
-						}
-						KademliaOut::FindNodeResult { key, closer_peers } => {
-							trace!(target: "sub-libp2p", "Libp2p => Query for {:?} yielded {:?} results",
-								key, closer_peers.len());
-							if closer_peers.is_empty() && self.num_connections != 0 {
-								warn!(target: "sub-libp2p", "Libp2p => Random Kademlia query has yielded empty \
-									results");
+				Async::Ready(NetworkBehaviourAction::GenerateEvent(ev)) => match ev {
+					KademliaEvent::UnroutablePeer { peer, .. } => {
+						let ev = DiscoveryOut::UnroutablePeer(peer);
+						return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
+					}
+					KademliaEvent::RoutingUpdated { peer, .. } => {
+						let ev = DiscoveryOut::Discovered(peer);
+						return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
+					}
+					KademliaEvent::GetClosestPeersResult(res) => {
+						match res {
+							Err(GetClosestPeersError::Timeout { key, peers }) => {
+								warn!(target: "sub-libp2p",
+									"Libp2p => Query for {:?} timed out with {:?} results",
+									key, peers.len());
+							},
+							Ok(ok) => {
+								trace!(target: "sub-libp2p",
+									"Libp2p => Query for {:?} yielded {:?} results",
+									ok.key, ok.peers.len());
+								if ok.peers.is_empty() && self.num_connections != 0 {
+									warn!(target: "sub-libp2p", "Libp2p => Random Kademlia query has yielded empty \
+										results");
+								}
 							}
 						}
-						KademliaOut::GetValueResult(res) => {
-							let ev = match res {
-								GetValueResult::Found { results } => {
-									let results = results
-											.into_iter()
-											.map(|r| (r.key, r.value))
-											.collect();
+					}
+					KademliaEvent::GetRecordResult(res) => {
+						let ev = match res {
+							Ok(ok) => {
+								let results = ok.records
+									.into_iter()
+									.map(|r| (r.key, r.value))
+									.collect();
 
-									DiscoveryOut::ValueFound(results)
-								}
-								GetValueResult::NotFound { key, .. } => {
-									DiscoveryOut::ValueNotFound(key)
-								}
-							};
-							return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
+								DiscoveryOut::ValueFound(results)
+							}
+							Err(e) => {
+								DiscoveryOut::ValueNotFound(e.into_key())
+							}
+						};
+						return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
+					}
+					KademliaEvent::PutRecordResult(res) => {
+						let ev = match res {
+							Ok(ok) => DiscoveryOut::ValuePut(ok.key),
+							Err(e) => {
+								DiscoveryOut::ValuePutFailed(e.into_key())
+							}
+						};
+						return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
+					}
+					KademliaEvent::RepublishRecordResult(res) => {
+						match res {
+							Ok(ok) => debug!(target: "sub-libp2p",
+								"Libp2p => Record republished: {:?}",
+								ok.key),
+							Err(e) => warn!(target: "sub-libp2p",
+								"Libp2p => Republishing of record {:?} failed with: {:?}",
+								e.key(), e)
 						}
-						KademliaOut::PutValueResult(res) => {
-							let ev = match res {
-								PutValueResult::Ok{ key, .. } =>  {
-									DiscoveryOut::ValuePut(key)
-								}
-								PutValueResult::Err { key, .. } => {
-									DiscoveryOut::ValuePutFailed(key)
-								}
-							};
-							return Async::Ready(NetworkBehaviourAction::GenerateEvent(ev));
-						}
-						// We never start any other type of query.
-						KademliaOut::GetProvidersResult { .. } => {}
+					}
+					KademliaEvent::Discovered { .. } => {
+						// We are not interested in these events at the moment.
+					}
+					// We never start any other type of query.
+					e => {
+						warn!(target: "sub-libp2p", "Libp2p => Unhandled Kademlia event: {:?}", e)
 					}
 				},
 				Async::Ready(NetworkBehaviourAction::DialAddress { address }) =>
@@ -384,9 +423,10 @@ mod tests {
 	use futures::prelude::*;
 	use libp2p::identity::Keypair;
 	use libp2p::Multiaddr;
-	use libp2p::core::{upgrade, Swarm};
+	use libp2p::core::upgrade;
 	use libp2p::core::transport::{Transport, MemoryTransport};
 	use libp2p::core::upgrade::{InboundUpgradeExt, OutboundUpgradeExt};
+	use libp2p::swarm::Swarm;
 	use std::collections::HashSet;
 	use super::{DiscoveryBehaviour, DiscoveryOut};
 
@@ -428,28 +468,34 @@ mod tests {
 				.collect::<HashSet<_>>()
 		}).collect::<Vec<_>>();
 
-		let fut = futures::future::poll_fn(move || -> Result<_, ()> {
-			loop {
-				let mut keep_polling = false;
-
+		let fut = futures::future::poll_fn::<_, (), _>(move || {
+			'polling: loop {
 				for swarm_n in 0..swarms.len() {
-					if let Async::Ready(Some(DiscoveryOut::Discovered(other))) =
-						swarms[swarm_n].0.poll().unwrap() {
-						if to_discover[swarm_n].remove(&other) {
-							keep_polling = true;
-							// Call `add_self_reported_address` to simulate identify happening.
-							let addr = swarms.iter()
-								.find(|s| *Swarm::local_peer_id(&s.0) == other)
-								.unwrap()
-								.1.clone();
-							swarms[swarm_n].0.add_self_reported_address(&other, addr);
+					match swarms[swarm_n].0.poll().unwrap() {
+						Async::Ready(Some(e)) => {
+							match e {
+								DiscoveryOut::UnroutablePeer(other) => {
+									// Call `add_self_reported_address` to simulate identify happening.
+									let addr = swarms.iter().find_map(|(s, a)|
+										if s.local_peer_id == other {
+											Some(a.clone())
+										} else {
+											None
+										})
+										.unwrap();
+									swarms[swarm_n].0.add_self_reported_address(&other, addr);
+								},
+								DiscoveryOut::Discovered(other) => {
+									to_discover[swarm_n].remove(&other);
+								}
+								_ => {}
+							}
+							continue 'polling
 						}
+						_ => {}
 					}
 				}
-
-				if !keep_polling {
-					break;
-				}
+				break
 			}
 
 			if to_discover.iter().all(|l| l.is_empty()) {

substrate/core/network/src/lib.rs

@@ -204,7 +204,7 @@ pub use on_demand_layer::{OnDemand, RemoteResponse};
 #[doc(hidden)]
 pub use runtime_primitives::traits::Block as BlockT;
 
-use libp2p::core::nodes::ConnectedPoint;
+use libp2p::core::ConnectedPoint;
 use serde::{Deserialize, Serialize};
 use slog_derive::SerdeValue;
 use std::{collections::{HashMap, HashSet}, time::Duration};

substrate/core/network/src/protocol.rs

@@ -19,9 +19,9 @@ use crate::custom_proto::{CustomProto, CustomProtoOut};
 use futures::prelude::*;
 use futures03::{StreamExt as _, TryStreamExt as _};
 use libp2p::{Multiaddr, PeerId};
-use libp2p::core::swarm::{ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
-use libp2p::core::{nodes::Substream, muxing::StreamMuxerBox};
-use libp2p::core::protocols_handler::{ProtocolsHandler, IntoProtocolsHandler};
+use libp2p::core::{ConnectedPoint, nodes::Substream, muxing::StreamMuxerBox};
+use libp2p::swarm::{ProtocolsHandler, IntoProtocolsHandler};
+use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
 use primitives::storage::StorageKey;
 use consensus::{import_queue::IncomingBlock, import_queue::Origin, BlockOrigin};
 use runtime_primitives::{generic::BlockId, ConsensusEngineId, Justification};

substrate/core/network/src/service.rs

@@ -32,8 +32,9 @@ use consensus::import_queue::{ImportQueue, Link};
 use consensus::import_queue::{BlockImportResult, BlockImportError};
 use futures::{prelude::*, sync::mpsc};
 use log::{warn, error, info};
-use libp2p::core::{swarm::NetworkBehaviour, transport::boxed::Boxed, muxing::StreamMuxerBox};
 use libp2p::{PeerId, Multiaddr, multihash::Multihash};
+use libp2p::core::{transport::boxed::Boxed, muxing::StreamMuxerBox};
+use libp2p::swarm::NetworkBehaviour;
 use parking_lot::Mutex;
 use peerset::PeersetHandle;
 use runtime_primitives::{traits::{Block as BlockT, NumberFor}, ConsensusEngineId};
@@ -675,7 +676,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Future for Ne
 }
 
 /// The libp2p swarm, customized for our needs.
-type Swarm<B, S, H> = libp2p::core::Swarm<
+type Swarm<B, S, H> = libp2p::swarm::Swarm<
 	Boxed<(PeerId, StreamMuxerBox), io::Error>,
 	Behaviour<B, S, H>
 >;