Backport networking changes 2018-09-07 edition

substrate/network-libp2p/src/service.rs

@@ -38,7 +38,7 @@ use std::sync::Arc;
 use std::sync::mpsc as sync_mpsc;
 use std::thread;
 use std::time::{Duration, Instant};
-use futures::{future, Future, stream, Stream};
+use futures::{future, Future, stream, Stream, select_all};
 use futures::sync::{mpsc, oneshot};
 use tokio::runtime::current_thread;
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -546,19 +546,24 @@ fn init_thread(
 	// Start the process of pinging the active nodes on the network.
 	let periodic = start_periodic_updates(shared.clone(), transport, swarm_controller);
 
-	// Merge all the futures into one!
-	Ok(swarm_events.for_each(|_| Ok(()))
-		.select(discovery).map_err(|(err, _)| err).and_then(|(_, rest)| rest)
-		.select(periodic).map_err(|(err, _)| err).and_then(|(_, rest)| rest)
-		.select(outgoing_connections).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)
-
+	let futures: Vec<Box<Future<Item = (), Error = IoError>>> = vec![
+		Box::new(swarm_events.for_each(|_| Ok(()))),
+		Box::new(discovery),
+		Box::new(periodic),
+		Box::new(outgoing_connections),
+		Box::new(timeouts),
+		Box::new(close_rx.map_err(|err| IoError::new(IoErrorKind::Other, err))),
+	];
+
+	Ok(
+		select_all(futures)
 		.and_then(move |_| {
 			debug!(target: "sub-libp2p", "Networking ended ; disconnecting all peers");
 			shared.network_state.disconnect_all();
 			Ok(())
-		}))
+		})
+		.map_err(|(r, _, _)| r)
+	)
 }
 
 /// Output of the common transport layer.

substrate/network-libp2p/src/topology.rs

@@ -19,7 +19,7 @@ use parking_lot::Mutex;
 use libp2p::{Multiaddr, PeerId};
 use serde_json;
 use std::{cmp, fs};
-use std::io::{Read, Cursor, Error as IoError, ErrorKind as IoErrorKind, Write};
+use std::io::{Read, Cursor, Error as IoError, ErrorKind as IoErrorKind, Write, BufReader, BufWriter};
 use std::path::{Path, PathBuf};
 use std::time::{Duration, Instant, SystemTime};
 
@@ -46,7 +46,7 @@ const FIRST_CONNECT_FAIL_BACKOFF: Duration = Duration::from_secs(2);
 /// Every time we fail to connect to an address, multiply the backoff by this constant.
 const FAIL_BACKOFF_MULTIPLIER: u32 = 2;
 /// We need a maximum value for the backoff, overwise we risk an overflow.
-const MAX_BACKOFF: Duration = Duration::from_secs(60);
+const MAX_BACKOFF: Duration = Duration::from_secs(30 * 60);
 
 // TODO: should be merged with the Kademlia k-buckets
 
@@ -101,7 +101,7 @@ impl NetTopology {
 		};
 
 		let file = fs::File::create(path)?;
-		serialize(file, &self.store)
+		serialize(BufWriter::with_capacity(1024 * 1024, file), &self.store)
 	}
 
 	/// Perform a cleanup pass, removing all obsolete addresses and peers.
@@ -497,7 +497,7 @@ fn try_load(path: impl AsRef<Path>) -> FnvHashMap<PeerId, PeerInfo> {
 	}
 
 	let mut file = match fs::File::open(path) {
-		Ok(f) => f,
+		Ok(f) => BufReader::with_capacity(1024 * 1024, f),
 		Err(err) => {
 			warn!(target: "sub-libp2p", "Failed to open peer storage file: {:?}", err);
 			info!(target: "sub-libp2p", "Deleting peer storage file {:?}", path);