Merge remote-tracking branch 'parity/master' into ashley-service-futures

Cargo.toml

@@ -153,3 +153,6 @@ members = [
 [profile.release]
 # Substrate runtime requires unwinding.
 panic = "unwind"
+
+[replace]
+'zstd:0.5.1' = { git = "https://github.com/gyscos/zstd-rs" }

bin/node/cli/src/service.rs

@@ -113,11 +113,10 @@ macro_rules! new_full_start {
 /// concrete types instead.
 macro_rules! new_full {
 	($config:expr, $with_startup_data: expr) => {{
-		use sc_network::DhtEvent;
 		use futures::{
-			compat::Future01CompatExt,
+			compat::{Future01CompatExt, Stream01CompatExt},
 			stream::StreamExt,
-			future::FutureExt,
+			future::{self, FutureExt},
 		};
 		use sc_network::Event;
 
@@ -142,13 +141,6 @@ macro_rules! new_full {
 
 		let (builder, mut import_setup, inherent_data_providers) = new_full_start!($config);
 
-		// Dht event channel from the network to the authority discovery module. Use bounded channel to ensure
-		// back-pressure. Authority discovery is triggering one event per authority within the current authority set.
-		// This estimates the authority set size to be somewhere below 10 000 thereby setting the channel buffer size to
-		// 10 000.
-		let (dht_event_tx, dht_event_rx) =
-			futures::channel::mpsc::channel::<DhtEvent>(10_000);
-
 		let service = builder.with_network_protocol(|_| Ok(crate::service::NodeProtocol::new()))?
 			.with_finality_proof_provider(|client, backend|
 				Ok(Arc::new(grandpa::FinalityProofProvider::new(backend, client)) as _)
@@ -189,12 +181,20 @@ macro_rules! new_full {
 			let babe = sc_consensus_babe::start_babe(babe_config)?;
 			service.spawn_essential_task(babe);
 
+			let network = service.network();
+			let dht_event_stream = network.event_stream().compat()
+				.map(|x| x.expect("<mpsc::channel::Receiver as Stream> never returns an error; qed"))
+				.filter_map(|e| match e {
+					Event::Dht(e) => future::ready(Some(e)),
+					_ => future::ready(None),
+				})
+				.boxed();
 			let authority_discovery = sc_authority_discovery::AuthorityDiscovery::new(
 				service.client(),
 				network,
 				sentry_nodes,
 				service.keystore(),
-				dht_event_rx.boxed(),
+				dht_event_stream,
 			);
 
 			service.spawn_task(authority_discovery);

client/service/src/builder.rs

@@ -1005,7 +1005,6 @@ ServiceBuilder<
 			network_status_sinks.clone(),
 			system_rpc_rx,
 			has_bootnodes,
-			dht_event_tx,
 		), exit.clone()).map(drop)));
 
 		let telemetry_connection_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<()>>>> = Default::default();

client/service/src/lib.rs

@@ -370,14 +370,10 @@ fn build_network_future<
 	status_sinks: Arc<Mutex<status_sinks::StatusSinks<(NetworkStatus<B>, NetworkState)>>>,
 	mut rpc_rx: mpsc::UnboundedReceiver<sc_rpc::system::Request<B>>,
 	should_have_peers: bool,
-	dht_event_tx: Option<mpsc::Sender<DhtEvent>>,
 ) -> impl Future<Output = ()> {
 	let mut imported_blocks_stream = client.import_notification_stream().fuse();
 	let mut finality_notification_stream = client.finality_notification_stream().fuse();
 
-	// Initializing a stream in order to obtain DHT events from the network.
-	let mut event_stream = network.service().event_stream().compat();
-
 	futures::future::poll_fn(move |cx| {
 		let before_polling = Instant::now();
 
@@ -470,26 +466,6 @@ fn build_network_future<
 			(status, state)
 		});
 
-		// Processing DHT events.
-		while let Poll::Ready(Some(Ok(event))) = Pin::new(&mut event_stream).poll_next(cx) {
-			match event {
-				Event::Dht(event) => {
-					// Given that client/authority-discovery is the only upper stack consumer of Dht events at the moment, all Dht
-					// events are being passed on to the authority-discovery module. In the future there might be multiple
-					// consumers of these events. In that case this would need to be refactored to properly dispatch the events,
-					// e.g. via a subscriber model.
-					if let Some(Err(e)) = dht_event_tx.as_ref().map(|c| c.clone().try_send(event)) {
-						if e.is_full() {
-							warn!(target: "service", "Dht event channel to authority discovery is full, dropping event.");
-						} else if e.is_disconnected() {
-							warn!(target: "service", "Dht event channel to authority discovery is disconnected, dropping event.");
-						}
-					}
-				}
-				_ => {}
-			}
-		}
-
 		// Main network polling.
 		if let Poll::Ready(Ok(())) = Pin::new(&mut (&mut network).compat()).poll(cx).map_err(|err| {
 			warn!(target: "service", "Error in network: {:?}", err);