Backport the current networking to v0.2

substrate/cli/src/lib.rs

@@ -294,10 +294,12 @@ where
 			None => 30333,
 		};
 
-		config.network.listen_address = iter::once(AddrComponent::IP4(Ipv4Addr::new(0, 0, 0, 0)))
-			.chain(iter::once(AddrComponent::TCP(port)))
-			.collect();
-		config.network.public_address = None;
+		config.network.listen_addresses = vec![
+			iter::once(AddrComponent::IP4(Ipv4Addr::new(0, 0, 0, 0)))
+				.chain(iter::once(AddrComponent::TCP(port)))
+				.collect()
+		];
+		config.network.public_addresses = Vec::new();
 		config.network.client_version = config.client_id();
 		config.network.use_secret = match matches.value_of("node-key").map(|s| s.parse()) {
 			Some(Ok(secret)) => Some(secret),

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 = "cc79d91706e7f1ce40a3a30626e192539ae993c0", default-features = false, features = ["libp2p-secio", "libp2p-secio-secp256k1"] }
+libp2p = { git = "https://github.com/tomaka/libp2p-rs", rev = "4ff16806970cad950619af8735d934f4b326996c", 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"
@@ -22,10 +22,10 @@ rand = "0.5.0"
 tokio = "0.1"
 tokio-io = "0.1"
 tokio-timer = "0.2"
-varint = { git = "https://github.com/tomaka/libp2p-rs", branch = "polkadot-2" }
+unsigned-varint = { version = "0.1", features = ["codec"] }
 
 [dev-dependencies]
 assert_matches = "1.2"
-ethcore-bytes = { git = "https://github.com/paritytech/parity.git" }
+parity-bytes = { git = "https://github.com/paritytech/parity-common.git" }
 ethcore-io = { git = "https://github.com/paritytech/parity.git" }
 ethcore-logger = { git = "https://github.com/paritytech/parity.git" }

substrate/network-libp2p/src/custom_proto.rs

@@ -12,7 +12,7 @@
 // GNU General Public License for more details.
 
 // You should have received a copy of the GNU General Public License
-// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.?
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use bytes::{Bytes, BytesMut};
 use ProtocolId;
@@ -23,7 +23,7 @@ use std::vec::IntoIter as VecIntoIter;
 use futures::{future, Future, stream, Stream, Sink};
 use futures::sync::mpsc;
 use tokio_io::{AsyncRead, AsyncWrite};
-use varint::VarintCodec;
+use unsigned_varint::codec::UviBytes;
 
 /// Connection upgrade for a single protocol.
 ///
@@ -157,7 +157,7 @@ where C: AsyncRead + AsyncWrite + 'static,		// TODO: 'static :-/
 		}
 
 		let (sink, stream) = {
-			let framed = AsyncRead::framed(socket, VarintCodec::default());
+			let framed = AsyncRead::framed(socket, UviBytes::default());
 			let msg_rx = msg_rx.map(Message::SendReq)
 				.map_err(|()| unreachable!("mpsc::UnboundedReceiver never errors"));
 			let (sink, stream) = framed.split();

substrate/network-libp2p/src/lib.rs

@@ -12,7 +12,7 @@
 // GNU General Public License for more details.
 
 // You should have received a copy of the GNU General Public License
-// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.?
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 #![recursion_limit="128"]
 #![type_length_limit = "268435456"]
@@ -28,7 +28,7 @@ extern crate libc;
 extern crate libp2p;
 extern crate rand;
 extern crate bytes;
-extern crate varint;
+extern crate unsigned_varint;
 
 extern crate ethcore_io as io;
 extern crate ethereum_types;

substrate/network-libp2p/src/network_state.rs

@@ -12,7 +12,7 @@
 // GNU General Public License for more details.
 
 // You should have received a copy of the GNU General Public License
-// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.?
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use bytes::Bytes;
 use fnv::{FnvHashMap, FnvHashSet};
@@ -677,7 +677,7 @@ impl NetworkState {
 	}
 
 	/// Disables a peer for `PEER_DISABLE_DURATION`. This adds the peer to the
-	/// list of disabled peers, and  drops any existing connections if
+	/// list of disabled peers, and drops any existing connections if
 	/// necessary (ie. drops the sender that was stored in the `UniqueConnec`
 	/// of `custom_proto`).
 	pub fn ban_peer(&self, who: NodeIndex, reason: &str) {
@@ -838,7 +838,7 @@ fn parse_and_add_to_node_store(
 
 	let mut addr = addr_str.to_multiaddr().map_err(|_| ErrorKind::AddressParse)?;
 	let who = match addr.pop() {
-		Some(AddrComponent::P2P(key)) | Some(AddrComponent::IPFS(key)) =>
+		Some(AddrComponent::P2P(key)) =>
 			PeerId::from_bytes(key).map_err(|_| ErrorKind::AddressParse)?,
 		_ => return Err(ErrorKind::AddressParse.into()),
 	};

substrate/network-libp2p/src/service.rs

@@ -12,7 +12,7 @@
 // GNU General Public License for more details.
 
 // You should have received a copy of the GNU General Public License
-// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.?
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use bytes::Bytes;
 use {Error, ErrorKind, NetworkConfiguration, NetworkProtocolHandler};
@@ -109,9 +109,10 @@ impl NetworkService {
 
 		let local_peer_id = network_state.local_public_key().clone()
 			.into_peer_id();
-		let mut listen_addr = config.listen_address.clone();
-		listen_addr.append(AddrComponent::P2P(local_peer_id.clone().into_bytes()));
-		info!(target: "sub-libp2p", "Local node address is: {}", listen_addr);
+		for mut addr in config.listen_addresses.iter().cloned() {
+			addr.append(AddrComponent::P2P(local_peer_id.clone().into_bytes()));
+			info!(target: "sub-libp2p", "Local node address is: {}", addr);
+		}
 
 		let kad_system = KadSystem::without_init(KadSystemConfig {
 			parallelism: 3,
@@ -129,10 +130,7 @@ impl NetworkService {
 		let (close_tx, close_rx) = oneshot::channel();
 		let (timeouts_register_tx, timeouts_register_rx) = mpsc::unbounded();
 
-		let mut listened_addrs = Vec::new();
-		if let Some(ref addr) = config.public_address {
-			listened_addrs.push(addr.clone());
-		}
+		let listened_addrs = config.public_addresses.clone();
 
 		let shared = Arc::new(Shared {
 			network_state,
@@ -431,7 +429,7 @@ fn init_thread(
 					let listener_upgrade = upgrade::or(upgrade::or(upgrade::or(
 						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_with_addr(ping::Ping, |(p, f), addr| FinalUpgrade::Ping(p, f, addr.clone()))),
+						upgrade::map_with_addr(ping::Ping, |out, addr| FinalUpgrade::from((out, addr.clone())))),
 						upgrade::map_with_addr(DelayedProtosList(shared), |c, a| FinalUpgrade::Custom(c, a.clone())));
 					upgrade::apply(out.socket, listener_upgrade, endpoint, client_addr)
 				}
@@ -445,17 +443,18 @@ fn init_thread(
 		)
 	};
 
-	// Listen on multiaddress.
-	match swarm_controller.listen_on(shared.config.listen_address.clone()) {
-		Ok(new_addr) => {
-			debug!(target: "sub-libp2p", "Libp2p listening on {}", new_addr);
-			*shared.original_listened_addr.write() = Some(new_addr.clone());
-		},
-		Err(_) => {
-			warn!(target: "sub-libp2p", "Can't listen on {}, protocol not supported",
-				shared.config.listen_address);
-			return Err(ErrorKind::BadProtocol.into())
-		},
+	// Listen on multiaddresses.
+	for addr in &shared.config.listen_addresses {
+		match swarm_controller.listen_on(addr.clone()) {
+			Ok(new_addr) => {
+				debug!(target: "sub-libp2p", "Libp2p listening on {}", new_addr);
+				*shared.original_listened_addr.write() = Some(new_addr.clone());
+			},
+			Err(_) => {
+				warn!(target: "sub-libp2p", "Can't listen on {}, protocol not supported", addr);
+				return Err(ErrorKind::BadProtocol.into())
+			},
+		}
 	}
 
 	// Explicitely connect to _all_ the boostrap nodes as a temporary measure.
@@ -571,12 +570,23 @@ enum FinalUpgrade<C> {
 	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>>, Multiaddr),
+	PingDialer(ping::Pinger, Box<Future<Item = (), Error = IoError>>, Multiaddr),
+	PingListener(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>>, Multiaddr),
 }
 
+impl<C> From<(ping::PingOutput, Multiaddr)> for FinalUpgrade<C> {
+	fn from((out, addr): (ping::PingOutput, Multiaddr)) -> FinalUpgrade<C> {
+		match out {
+			ping::PingOutput::Ponger(processing) => FinalUpgrade::PingListener(processing, addr),
+			ping::PingOutput::Pinger { pinger, processing } =>
+				FinalUpgrade::PingDialer(pinger, processing, addr),
+		}
+	}
+}
+
 /// Called whenever we successfully open a multistream with a remote.
 fn listener_handle<'a, C>(
 	shared: Arc<Shared>,
@@ -611,7 +621,12 @@ fn listener_handle<'a, C>(
 		FinalUpgrade::Identify(IdentifyOutput::RemoteInfo { .. }, _) =>
 			unreachable!("We are never dialing with the identify protocol"),
 
-		FinalUpgrade::Ping(pinger, future, client_addr) => {
+		FinalUpgrade::PingListener(future, client_addr) => {
+			trace!(target: "sub-libp2p", "Received ping substream from {:?}", client_addr);
+			future
+		},
+
+		FinalUpgrade::PingDialer(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)) => {
@@ -761,17 +776,20 @@ fn handle_custom_connection(
 		who: NodeIndex,
 		node_id: PeerstorePeerId,
 		handler: Arc<NetworkProtocolHandler + Send + Sync>,
-		protocol: ProtocolId
+		protocol: ProtocolId,
+		print_log_message: bool,
 	}
 
 	impl Drop for ProtoDisconnectGuard {
 		fn drop(&mut self) {
-			info!(target: "sub-libp2p",
-				"Node {:?} with peer ID {} through protocol {:?} disconnected",
-				self.node_id,
-				self.who,
-				self.protocol
-			);
+			if self.print_log_message {
+				info!(target: "sub-libp2p",
+					"Node {:?} with peer ID {} through protocol {:?} disconnected",
+					self.node_id,
+					self.who,
+					self.protocol
+				);
+			}
 			self.handler.disconnected(&NetworkContextImpl {
 				inner: self.inner.clone(),
 				protocol: self.protocol,
@@ -784,12 +802,13 @@ fn handle_custom_connection(
 		}
 	}
 
-	let dc_guard = ProtoDisconnectGuard {
+	let mut dc_guard = ProtoDisconnectGuard {
 		inner: shared.clone(),
 		who,
 		node_id: node_id.clone(),
 		handler: handler.clone(),
 		protocol: protocol_id,
+		print_log_message: true,
 	};
 
 	let fut = custom_proto_out.incoming
@@ -816,6 +835,7 @@ fn handle_custom_connection(
 				info!(target: "sub-libp2p", "Finishing future for proto {:?} with {:?} => {:?}",
 					protocol_id, node_id, val);
 				// Makes sure that `dc_guard` is kept alive until here.
+				dc_guard.print_log_message = false;
 				drop(dc_guard);
 				val
 			}
@@ -844,7 +864,7 @@ fn handle_custom_connection(
 /// nodes and only accept incoming connections.
 fn start_kademlia_discovery<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,
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
 		T::MultiaddrFuture: 'static,
 		To: AsyncRead + AsyncWrite + 'static,
 		St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
@@ -911,7 +931,7 @@ fn perform_kademlia_query<T, To, St, C>(
 	transport: T,
 	swarm_controller: SwarmController<St>
 ) -> impl Future<Item = (), Error = IoError>
-	where T: MuxedTransport<Output =  TransportOutput<To>> + Clone + 'static,
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
 		T::MultiaddrFuture: 'static,
 		To: AsyncRead + AsyncWrite + 'static,
 		St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
@@ -960,7 +980,7 @@ fn connect_to_nodes<T, To, St, C>(
 	base_transport: T,
 	swarm_controller: &SwarmController<St>
 )
-	where T: MuxedTransport<Output =  TransportOutput<To>> + Clone + 'static,
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
 		T::MultiaddrFuture: 'static,
 		To: AsyncRead + AsyncWrite + 'static,
 		St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
@@ -1005,7 +1025,7 @@ fn connect_with_query_peer_id<T, To, St, C>(
 	addr: Multiaddr,
 	swarm_controller: &SwarmController<St>
 )
-	where T: MuxedTransport<Output =  TransportOutput<To>> + Clone + 'static,
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
 		T::MultiaddrFuture: 'static,
 		To: AsyncRead + AsyncWrite + 'static,
 		St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
@@ -1067,7 +1087,7 @@ fn open_peer_custom_proto<T, To, St, C>(
 	expected_peer_id: PeerstorePeerId,
 	swarm_controller: &SwarmController<St>
 )
-	where T: MuxedTransport<Output =  TransportOutput<To>> + Clone + 'static,
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
 		T::MultiaddrFuture: 'static,
 		To: AsyncRead + AsyncWrite + 'static,
 		St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
@@ -1157,7 +1177,7 @@ fn open_peer_custom_proto<T, To, St, C>(
 fn obtain_kad_connection<T, To, St, C>(shared: Arc<Shared>,
 	who: PeerstorePeerId, transport: T, swarm_controller: SwarmController<St>)
 	-> impl Future<Item = KadConnecController, Error = IoError>
-	where T: MuxedTransport<Output =  TransportOutput<To>> + Clone + 'static,
+	where T: MuxedTransport<Output = TransportOutput<To>> + Clone + 'static,
 		T::MultiaddrFuture: 'static,
 		To: AsyncRead + AsyncWrite + 'static,
 		St: MuxedTransport<Output = FinalUpgrade<C>> + Clone + 'static,
@@ -1245,9 +1265,9 @@ fn start_pinger<T, To, St, C>(
 		.and_then(move |out, endpoint, client_addr|
 			upgrade::apply(out.socket, ping::Ping, endpoint, client_addr)
 		)
-		.and_then(move |(ctrl, fut), _, client_addr| {
+		.and_then(move |out, _, client_addr| {
 			client_addr.map(|client_addr| {
-				let out = FinalUpgrade::Ping(ctrl, fut, client_addr.clone());
+				let out = FinalUpgrade::from((out, client_addr.clone()));
 				(out, future::ok(client_addr))
 			})
 		});

substrate/network-libp2p/src/timeouts.rs

@@ -12,7 +12,7 @@
 // GNU General Public License for more details.
 
 // You should have received a copy of the GNU General Public License
-// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.?
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use futures::{Async, future, Future, Poll, stream, Stream, sync::mpsc};
 use std::io::{Error as IoError, ErrorKind as IoErrorKind};

substrate/network-libp2p/src/traits.rs

@@ -104,10 +104,10 @@ pub struct NetworkConfiguration {
 	pub config_path: Option<String>,
 	/// Directory path to store network-specific configuration. None means nothing will be saved
 	pub net_config_path: Option<String>,
-	/// IP address to listen for incoming connections. Listen to all connections by default
-	pub listen_address: Multiaddr,
-	/// IP address to advertise. Detected automatically if none.
-	pub public_address: Option<Multiaddr>,
+	/// Multiaddresses to listen for incoming connections.
+	pub listen_addresses: Vec<Multiaddr>,
+	/// Multiaddresses to advertise. Detected automatically if empty.
+	pub public_addresses: Vec<Multiaddr>,
 	/// List of initial node addresses
 	pub boot_nodes: Vec<String>,
 	/// Use provided node key instead of default
@@ -136,10 +136,12 @@ impl NetworkConfiguration {
 		NetworkConfiguration {
 			config_path: None,
 			net_config_path: None,
-			listen_address: iter::once(AddrComponent::IP4(Ipv4Addr::new(0, 0, 0, 0)))
-				.chain(iter::once(AddrComponent::TCP(30333)))
-				.collect(),
-			public_address: None,
+			listen_addresses: vec![
+				iter::once(AddrComponent::IP4(Ipv4Addr::new(0, 0, 0, 0)))
+					.chain(iter::once(AddrComponent::TCP(30333)))
+					.collect()
+			],
+			public_addresses: Vec::new(),
 			boot_nodes: Vec::new(),
 			use_secret: None,
 			min_peers: 25,
@@ -153,9 +155,11 @@ impl NetworkConfiguration {
 	/// Create new default configuration for localhost-only connection with random port (useful for testing)
 	pub fn new_local() -> NetworkConfiguration {
 		let mut config = NetworkConfiguration::new();
-		config.listen_address = iter::once(AddrComponent::IP4(Ipv4Addr::new(127, 0, 0, 1)))
-			.chain(iter::once(AddrComponent::TCP(0)))
-			.collect();
+		config.listen_addresses = vec![
+			iter::once(AddrComponent::IP4(Ipv4Addr::new(127, 0, 0, 1)))
+				.chain(iter::once(AddrComponent::TCP(0)))
+				.collect()
+		];
 		config
 	}
 }