Use async/await in the NetworkWorker and in build_network_future

client/network/src/lib.rs

@@ -16,6 +16,7 @@
 
 #![warn(unused_extern_crates)]
 #![warn(missing_docs)]
+#![recursion_limit = "2048"]
 
 //! Substrate-specific P2P networking.
 //!

client/network/src/service/import_queue.rs

@@ -0,0 +1,181 @@
+// Copyright 2017-2020 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// 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/>.
+
+//! Wraps around a `Box<dyn ImportQueue<_>>` and gives it a better future-oriented API.
+
+use crate::PeerId;
+
+use futures::prelude::*;
+use sp_consensus::BlockOrigin;
+use sp_consensus::import_queue::{BlockImportError, BlockImportResult, ImportQueue, Link, Origin, IncomingBlock};
+use sp_runtime::{Justification, traits::{Block as BlockT, NumberFor}};
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
+use std::task::Poll;
+
+/// Wraps around a `Box<dyn ImportQueue<B>>`.
+pub struct SmartImportQueue<B: BlockT> {
+	queue: Box<dyn ImportQueue<B>>,
+	pending_actions_tx: TracingUnboundedSender<ImportQueueAction<B>>,
+	pending_actions_rx: TracingUnboundedReceiver<ImportQueueAction<B>>,
+}
+
+impl<B: BlockT> SmartImportQueue<B> {
+	/// Import bunch of blocks.
+	pub fn import_blocks(&mut self, origin: BlockOrigin, blocks: Vec<IncomingBlock<B>>) {
+		self.queue.import_blocks(origin, blocks);
+	}
+
+	/// Import a block justification.
+	pub fn import_justification(
+		&mut self,
+		who: Origin,
+		hash: B::Hash,
+		number: NumberFor<B>,
+		justification: Justification
+	) {
+		self.queue.import_justification(who, hash, number, justification);
+	}
+
+	/// Import block finality proof.
+	pub fn import_finality_proof(
+		&mut self,
+		who: Origin,
+		hash: B::Hash,
+		number: NumberFor<B>,
+		finality_proof: Vec<u8>
+	) {
+		self.queue.import_finality_proof(who, hash, number, finality_proof);
+	}
+
+	/// Returns the next action reported by the import queue.
+	pub fn next_action<'a>(&'a mut self) -> impl Future<Output = ImportQueueAction<B>> + 'a {
+		future::poll_fn(move |cx| {
+			// Try to empty the receiver first, so that the unbounded queue doesn't get filled up
+			// if `next_action` isn't called quickly enough.
+			if let Poll::Ready(Some(action)) = self.pending_actions_rx.poll_next_unpin(cx) {
+				return Poll::Ready(action)
+			}
+
+			// If the receiver is empty, ask the import queue to push things on it.
+			self.queue.poll_actions(cx, &mut NetworkLink {
+				out: &mut self.pending_actions_tx,
+			});
+
+			if let Poll::Ready(Some(action)) = self.pending_actions_rx.poll_next_unpin(cx) {
+				Poll::Ready(action)
+			} else {
+				Poll::Pending
+			}
+		})
+	}
+}
+
+impl<B: BlockT> From<Box<dyn ImportQueue<B>>> for SmartImportQueue<B> {
+	fn from(queue: Box<dyn ImportQueue<B>>) -> Self {
+		let (pending_actions_tx, pending_actions_rx) = tracing_unbounded("smart-import-queue");
+		SmartImportQueue {
+			queue,
+			pending_actions_tx,
+			pending_actions_rx,
+		}
+	}
+}
+
+/// Action that the import queue has performed.
+///
+/// This enum mimics the functions of the `Link` trait.
+pub enum ImportQueueAction<B: BlockT> {
+	BlocksProcessed {
+		imported: usize,
+		count: usize,
+		results: Vec<(Result<BlockImportResult<NumberFor<B>>, BlockImportError>, B::Hash)>,
+	},
+	JustificationImported {
+		who: PeerId,
+		hash: B::Hash,
+		number: NumberFor<B>,
+		success: bool,
+	},
+	RequestJustification {
+		hash: B::Hash,
+		number: NumberFor<B>,
+	},
+	RequestFinalityProof {
+		hash: B::Hash,
+		number: NumberFor<B>,
+	},
+	FinalityProofImported {
+		who: PeerId,
+		request_block: (B::Hash, NumberFor<B>),
+		finalization_result: Result<(B::Hash, NumberFor<B>), ()>,
+	},
+}
+
+// Implementation of `import_queue::Link` trait that sends actions on the sender inside of it.
+struct NetworkLink<'a, B: BlockT> {
+	out: &'a mut TracingUnboundedSender<ImportQueueAction<B>>,
+}
+
+impl<'a, B: BlockT> Link<B> for NetworkLink<'a, B> {
+	fn blocks_processed(
+		&mut self,
+		imported: usize,
+		count: usize,
+		results: Vec<(Result<BlockImportResult<NumberFor<B>>, BlockImportError>, B::Hash)>
+	) {
+		let _ = self.out.unbounded_send(ImportQueueAction::BlocksProcessed {
+			imported,
+			count,
+			results,
+		});
+	}
+
+	fn justification_imported(&mut self, who: PeerId, hash: &B::Hash, number: NumberFor<B>, success: bool) {
+		let _ = self.out.unbounded_send(ImportQueueAction::JustificationImported {
+			who,
+			hash: hash.clone(),
+			number,
+			success,
+		});
+	}
+
+	fn request_justification(&mut self, hash: &B::Hash, number: NumberFor<B>) {
+		let _ = self.out.unbounded_send(ImportQueueAction::RequestJustification {
+			hash: hash.clone(),
+			number,
+		});
+	}
+
+	fn request_finality_proof(&mut self, hash: &B::Hash, number: NumberFor<B>) {
+		let _ = self.out.unbounded_send(ImportQueueAction::RequestFinalityProof {
+			hash: hash.clone(),
+			number,
+		});
+	}
+
+	fn finality_proof_imported(
+		&mut self,
+		who: PeerId,
+		request_block: (B::Hash, NumberFor<B>),
+		finalization_result: Result<(B::Hash, NumberFor<B>), ()>,
+	) {
+		let _ = self.out.unbounded_send(ImportQueueAction::FinalityProofImported {
+			who,
+			request_block,
+			finalization_result,
+		});
+	}
+}

client/network/src/service/tests.rs

@@ -87,7 +87,7 @@ fn build_test_full_node(config: config::NetworkConfiguration)
 		None,
 	));
 
-	let worker = NetworkWorker::new(config::Params {
+	let mut worker = NetworkWorker::new(config::Params {
 		role: config::Role::Full,
 		executor: None,
 		network_config: config,
@@ -109,8 +109,9 @@ fn build_test_full_node(config: config::NetworkConfiguration)
 	let event_stream = service.event_stream("test");
 
 	async_std::task::spawn(async move {
-		futures::pin_mut!(worker);
-		let _ = worker.await;
+		loop {
+			worker.next_action().await;
+		}
 	});
 
 	(service, event_stream)

client/network/test/src/lib.rs

@@ -787,8 +787,12 @@ pub trait TestNetFactory: Sized {
 		self.mut_peers(|peers| {
 			for peer in peers {
 				trace!(target: "sync", "-- Polling {}", peer.id());
-				if let Poll::Ready(res) = Pin::new(&mut peer.network).poll(cx) {
-					res.unwrap();
+				loop {
+					let net_poll_future = peer.network.next_action();
+					futures::pin_mut!(net_poll_future);
+					if let Poll::Pending = net_poll_future.poll(cx) {
+						break;
+					}
 				}
 				trace!(target: "sync", "-- Polling complete {}", peer.id());
 

client/service/src/builder.rs

@@ -831,7 +831,7 @@ ServiceBuilder<
 		let has_bootnodes = !network_params.network_config.boot_nodes.is_empty();
 		let network_mut = sc_network::NetworkWorker::new(network_params)?;
 		let network = network_mut.service().clone();
-		let network_status_sinks = Arc::new(Mutex::new(status_sinks::StatusSinks::new()));
+		let network_status_sinks = Arc::new(status_sinks::StatusSinks::new());
 
 		let offchain_storage = backend.offchain_storage();
 		let offchain_workers = match (config.offchain_worker, offchain_storage.clone()) {
@@ -968,7 +968,7 @@ ServiceBuilder<
 		let transaction_pool_ = transaction_pool.clone();
 		let client_ = client.clone();
 		let (state_tx, state_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat1");
-		network_status_sinks.lock().push(std::time::Duration::from_millis(5000), state_tx);
+		network_status_sinks.push(std::time::Duration::from_millis(5000), state_tx);
 		let tel_task = state_rx.for_each(move |(net_status, _)| {
 			let info = client_.usage_info();
 			metrics_service.tick(
@@ -986,7 +986,7 @@ ServiceBuilder<
 
 		// Periodically send the network state to the telemetry.
 		let (netstat_tx, netstat_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat2");
-		network_status_sinks.lock().push(std::time::Duration::from_secs(30), netstat_tx);
+		network_status_sinks.push(std::time::Duration::from_secs(30), netstat_tx);
 		let tel_task_2 = netstat_rx.for_each(move |(_, network_state)| {
 			telemetry!(
 				SUBSTRATE_INFO;