Merge the protocol thread with the network thread (#2475)

* Remove the Incoming enum

* Use tokio in the protocol background thread

* Some internal protocol logic simplifications

* Merge the protocol thread with the network thread

* More the status_sinks logic to Service

* Remove FromNetworkMsg

substrate/Cargo.lock

@@ -4165,6 +4165,7 @@ dependencies = [
  "substrate-primitives 1.0.0",
  "substrate-test-client 1.0.0",
  "tokio 0.1.19 (registry+https://github.com/rust-lang/crates.io-index)",
+ "void 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]

substrate/core/network/Cargo.toml

@@ -31,6 +31,7 @@ peerset = { package = "substrate-peerset", path = "../../core/peerset" }
 tokio = "0.1.11"
 keyring = { package = "substrate-keyring", path = "../../core/keyring", optional = true }
 test_client = { package = "substrate-test-client", path = "../../core/test-client", optional = true }
+void = "1.0"
 
 [dev-dependencies]
 env_logger = { version = "0.6" }

substrate/core/network/src/protocol.rs

@@ -14,9 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
-use crossbeam_channel::{self as channel, Receiver, Sender, select};
+use futures::{prelude::*, sync::mpsc};
-use futures::sync::mpsc;
-use parking_lot::Mutex;
 use network_libp2p::PeerId;
 use primitives::storage::StorageKey;
 use runtime_primitives::{generic::BlockId, ConsensusEngineId};
@@ -35,7 +33,7 @@ use rustc_hex::ToHex;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
-use std::{cmp, num::NonZeroUsize, thread, time};
+use std::{cmp, num::NonZeroUsize, time};
 use log::{trace, debug, warn, error};
 use crate::chain::Client;
 use client::light::fetcher::ChangesProof;
@@ -46,8 +44,6 @@ const REQUEST_TIMEOUT_SEC: u64 = 40;
 const TICK_TIMEOUT: time::Duration = time::Duration::from_millis(1100);
 /// Interval at which we propagate exstrinsics;
 const PROPAGATE_TIMEOUT: time::Duration = time::Duration::from_millis(2900);
-/// Interval at which we send status updates on the SyncProvider status stream.
-const STATUS_INTERVAL: time::Duration = time::Duration::from_millis(5000);
 
 /// Current protocol version.
 pub(crate) const CURRENT_VERSION: u32 = 3;
@@ -77,10 +73,12 @@ const RPC_FAILED_REPUTATION_CHANGE: i32 = -(1 << 12);
 
 // Lock must always be taken in order declared here.
 pub struct Protocol<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
-	status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<ProtocolStatus<B>>>>>,
 	network_chan: NetworkChan<B>,
-	port: Receiver<ProtocolMsg<B, S>>,
+	port: mpsc::UnboundedReceiver<ProtocolMsg<B, S>>,
-	from_network_port: Receiver<FromNetworkMsg<B>>,
+	/// Interval at which we call `tick`.
+	tick_timeout: tokio::timer::Interval,
+	/// Interval at which we call `propagate_extrinsics`.
+	propagate_timeout: tokio::timer::Interval,
 	config: ProtocolConfig,
 	on_demand: Option<Arc<OnDemandService<B>>>,
 	genesis_hash: B::Hash,
@@ -256,8 +254,6 @@ pub enum ProtocolMsg<B: BlockT, S: NetworkSpecialization<B>> {
 	BlocksProcessed(Vec<B::Hash>, bool),
 	/// Tell protocol to restart sync.
 	RestartSync,
-	/// Propagate status updates.
-	Status,
 	/// Tell protocol to propagate extrinsics.
 	PropagateExtrinsics,
 	/// Tell protocol that a block was imported (sent by the import-queue).
@@ -284,39 +280,17 @@ pub enum ProtocolMsg<B: BlockT, S: NetworkSpecialization<B>> {
 	/// Only used in tests.
 	#[cfg(any(test, feature = "test-helpers"))]
 	Abort,
-	/// Tell protocol to abort sync and stop.
-	Stop,
 	/// Tell protocol to perform regular maintenance.
+	#[cfg(any(test, feature = "test-helpers"))]
 	Tick,
 	/// Synchronization request.
 	#[cfg(any(test, feature = "test-helpers"))]
 	Synchronize,
 }
 
-/// Messages sent to Protocol from Network-libp2p.
-pub enum FromNetworkMsg<B: BlockT> {
-	/// A peer connected, with debug info.
-	PeerConnected(PeerId, String),
-	/// A peer disconnected, with debug info.
-	PeerDisconnected(PeerId, String),
-	/// A custom message from another peer.
-	CustomMessage(PeerId, Message<B>),
-	/// Let protocol know a peer is currenlty clogged.
-	PeerClogged(PeerId, Option<Message<B>>),
-	/// Synchronization request.
-	#[cfg(any(test, feature = "test-helpers"))]
-	Synchronize,
-}
-
-enum Incoming<B: BlockT, S: NetworkSpecialization<B>> {
-	FromNetwork(FromNetworkMsg<B>),
-	FromClient(ProtocolMsg<B, S>)
-}
-
 impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	/// Create a new instance.
 	pub fn new(
-		status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<ProtocolStatus<B>>>>>,
 		is_offline: Arc<AtomicBool>,
 		is_major_syncing: Arc<AtomicBool>,
 		connected_peers: Arc<RwLock<HashMap<PeerId, ConnectedPeer<B>>>>,
@@ -327,92 +301,81 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		on_demand: Option<Arc<OnDemandService<B>>>,
 		transaction_pool: Arc<TransactionPool<H, B>>,
 		specialization: S,
-	) -> error::Result<(Sender<ProtocolMsg<B, S>>, Sender<FromNetworkMsg<B>>)> {
+	) -> error::Result<(Protocol<B, S, H>, mpsc::UnboundedSender<ProtocolMsg<B, S>>)> {
-		let (protocol_sender, port) = channel::unbounded();
+		let (protocol_sender, port) = mpsc::unbounded();
-		let (from_network_sender, from_network_port) = channel::bounded(4);
 		let info = chain.info()?;
 		let sync = ChainSync::new(is_offline, is_major_syncing, config.roles, &info, import_queue);
-		let _ = thread::Builder::new()
+		let protocol = Protocol {
-			.name("Protocol".into())
+			network_chan,
-			.spawn(move || {
+			port,
-				let mut protocol = Protocol {
+			tick_timeout: tokio::timer::Interval::new_interval(TICK_TIMEOUT),
-					status_sinks,
+			propagate_timeout: tokio::timer::Interval::new_interval(PROPAGATE_TIMEOUT),
-					network_chan,
+			config: config,
-					from_network_port,
+			context_data: ContextData {
-					port,
+				peers: HashMap::new(),
-					config: config,
+				chain,
-					context_data: ContextData {
-						peers: HashMap::new(),
-						chain,
-					},
-					on_demand,
-					genesis_hash: info.chain.genesis_hash,
-					sync,
-					specialization: specialization,
-					consensus_gossip: ConsensusGossip::new(),
-					handshaking_peers: HashMap::new(),
-					connected_peers,
-					transaction_pool: transaction_pool,
-				};
-				let tick_timeout = channel::tick(TICK_TIMEOUT);
-				let propagate_timeout = channel::tick(PROPAGATE_TIMEOUT);
-				let status_interval = channel::tick(STATUS_INTERVAL);
-				while protocol.run(&tick_timeout, &propagate_timeout, &status_interval) {
-					// Running until all senders have been dropped...
-				}
-			})
-			.expect("Protocol thread spawning failed");
-		Ok((protocol_sender, from_network_sender))
-	}
-
-	fn run(
-		&mut self,
-		tick_timeout: &Receiver<time::Instant>,
-		propagate_timeout: &Receiver<time::Instant>,
-		status_interval: &Receiver<time::Instant>,
-	) -> bool {
-		let msg = select! {
-			recv(self.port) -> event => {
-				match event {
-					Ok(msg) => Incoming::FromClient(msg),
-					// Our sender has been dropped, quit.
-					Err(_) => {
-						Incoming::FromClient(ProtocolMsg::Stop)
-					},
-				}
-			},
-			recv(self.from_network_port) -> event => {
-				match event {
-					Ok(msg) => Incoming::FromNetwork(msg),
-					// Our sender has been dropped, quit.
-					Err(_) => {
-						Incoming::FromClient(ProtocolMsg::Stop)
-					},
-				}
-			},
-			recv(tick_timeout) -> _ => {
-				Incoming::FromClient(ProtocolMsg::Tick)
-			},
-			recv(propagate_timeout) -> _ => {
-				Incoming::FromClient(ProtocolMsg::PropagateExtrinsics)
-			},
-			recv(status_interval) -> _ => {
-				Incoming::FromClient(ProtocolMsg::Status)
 			},
+			on_demand,
+			genesis_hash: info.chain.genesis_hash,
+			sync,
+			specialization: specialization,
+			consensus_gossip: ConsensusGossip::new(),
+			handshaking_peers: HashMap::new(),
+			connected_peers,
+			transaction_pool: transaction_pool,
 		};
-		self.handle_msg(msg)
+
+		Ok((protocol, protocol_sender))
 	}
 
-	fn handle_msg(&mut self, msg: Incoming<B, S>) -> bool {
+	/// Returns an object representing the status of the protocol.
-		match msg {
+	pub fn status(&mut self) -> ProtocolStatus<B> {
-			Incoming::FromNetwork(msg) => self.handle_network_msg(msg),
+		ProtocolStatus {
-			Incoming::FromClient(msg) => self.handle_client_msg(msg),
+			sync: self.sync.status(),
+			num_peers: self.context_data.peers.values().count(),
+			num_active_peers: self
+				.context_data
+				.peers
+				.values()
+				.filter(|p| p.block_request.is_some())
+				.count(),
 		}
 	}
+}
 
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Future for Protocol<B, S, H> {
+	type Item = ();
+	type Error = void::Void;
+
+	fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+		while let Ok(Async::Ready(_)) = self.tick_timeout.poll() {
+			self.tick();
+		}
+
+		while let Ok(Async::Ready(_)) = self.propagate_timeout.poll() {
+			self.propagate_extrinsics();
+		}
+
+		loop {
+			match self.port.poll() {
+				Ok(Async::Ready(None)) | Err(_) => {
+					self.stop();
+					return Ok(Async::Ready(()))
+				}
+				Ok(Async::Ready(Some(msg))) => if !self.handle_client_msg(msg) {
+					return Ok(Async::Ready(()))
+				}
+				Ok(Async::NotReady) => break,
+			}
+		}
+
+		Ok(Async::NotReady)
+	}
+}
+
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	fn handle_client_msg(&mut self, msg: ProtocolMsg<B, S>) -> bool {
 		match msg {
-			ProtocolMsg::Status => self.on_status(),
 			ProtocolMsg::BlockImported(hash, header) => self.on_block_imported(hash, &header),
 			ProtocolMsg::BlockFinalized(hash, header) => self.on_block_finalized(hash, &header),
 			ProtocolMsg::ExecuteWithSpec(task) => {
@@ -449,13 +412,10 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			},
 			ProtocolMsg::JustificationImportResult(hash, number, success) => self.sync.justification_import_result(hash, number, success),
 			ProtocolMsg::PropagateExtrinsics => self.propagate_extrinsics(),
+			#[cfg(any(test, feature = "test-helpers"))]
 			ProtocolMsg::Tick => self.tick(),
 			#[cfg(any(test, feature = "test-helpers"))]
 			ProtocolMsg::Abort => self.abort(),
-			ProtocolMsg::Stop => {
-				self.stop();
-				return false;
-			},
 			#[cfg(any(test, feature = "test-helpers"))]
 			ProtocolMsg::Synchronize => {
 				trace!(target: "sync", "handle_client_msg: received Synchronize msg");
@@ -465,20 +425,6 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		true
 	}
 
-	fn handle_network_msg(&mut self, msg: FromNetworkMsg<B>) -> bool {
-		match msg {
-			FromNetworkMsg::PeerDisconnected(who, debug_info) => self.on_peer_disconnected(who, debug_info),
-			FromNetworkMsg::PeerConnected(who, debug_info) => self.on_peer_connected(who, debug_info),
-			FromNetworkMsg::PeerClogged(who, message) => self.on_clogged_peer(who, message),
-			FromNetworkMsg::CustomMessage(who, message) => {
-				self.on_custom_message(who, message)
-			},
-			#[cfg(any(test, feature = "test-helpers"))]
-			FromNetworkMsg::Synchronize => self.network_chan.send(NetworkMsg::Synchronized),
-		}
-		true
-	}
-
 	fn handle_response(&mut self, who: PeerId, response: &message::BlockResponse<B>) -> Option<message::BlockRequest<B>> {
 		if let Some(ref mut peer) = self.context_data.peers.get_mut(&who) {
 			if let Some(_) = peer.obsolete_requests.remove(&response.id) {
@@ -511,22 +457,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		}
 	}
 
-	/// Propagates protocol statuses.
+	pub fn on_custom_message(&mut self, who: PeerId, message: Message<B>) {
-	fn on_status(&mut self) {
-		let status = ProtocolStatus {
-			sync: self.sync.status(),
-			num_peers: self.context_data.peers.values().count(),
-			num_active_peers: self
-				.context_data
-				.peers
-				.values()
-				.filter(|p| p.block_request.is_some())
-				.count(),
-		};
-		self.status_sinks.lock().retain(|sink| sink.unbounded_send(status.clone()).is_ok());
-	}
-
-	fn on_custom_message(&mut self, who: PeerId, message: Message<B>) {
 		match message {
 			GenericMessage::Status(s) => self.on_status_message(who, s),
 			GenericMessage::BlockRequest(r) => self.on_block_request(who, r),
@@ -595,14 +526,14 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	}
 
 	/// Called when a new peer is connected
-	fn on_peer_connected(&mut self, who: PeerId, debug_info: String) {
+	pub fn on_peer_connected(&mut self, who: PeerId, debug_info: String) {
 		trace!(target: "sync", "Connecting {}: {}", who, debug_info);
 		self.handshaking_peers.insert(who.clone(), HandshakingPeer { timestamp: time::Instant::now() });
 		self.send_status(who);
 	}
 
 	/// Called by peer when it is disconnecting
-	fn on_peer_disconnected(&mut self, peer: PeerId, debug_info: String) {
+	pub fn on_peer_disconnected(&mut self, peer: PeerId, debug_info: String) {
 		trace!(target: "sync", "Disconnecting {}: {}", peer, debug_info);
 		// lock all the the peer lists so that add/remove peer events are in order
 		let removed = {
@@ -637,6 +568,12 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		}
 	}
 
+	/// Puts the `Synchronized` message on `network_chan`.
+	#[cfg(any(test, feature = "test-helpers"))]
+	pub fn synchronize(&self) {
+		self.network_chan.send(NetworkMsg::Synchronized);
+	}
+
 	fn on_block_request(&mut self, peer: PeerId, request: message::BlockRequest<B>) {
 		trace!(target: "sync", "BlockRequest {} from {}: from {:?} to {:?} max {:?}",
 			request.id,
@@ -915,7 +852,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	///
 	/// In chain-based consensus, we often need to make sure non-best forks are
 	/// at least temporarily synced.
-	pub fn announce_block(&mut self, hash: B::Hash) {
+	fn announce_block(&mut self, hash: B::Hash) {
 		let header = match self.context_data.chain.header(&BlockId::Hash(hash)) {
 			Ok(Some(header)) => header,
 			Ok(None) => {

substrate/core/network/src/service.rs

@@ -17,7 +17,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::{io, thread};
+use std::{io, thread, time::Duration};
 
 use log::{warn, debug, error, info};
 use futures::{Async, Future, Stream, sync::oneshot, sync::mpsc};
@@ -31,7 +31,7 @@ use runtime_primitives::{traits::{Block as BlockT, NumberFor}, ConsensusEngineId
 
 use crate::consensus_gossip::{ConsensusGossip, MessageRecipient as GossipMessageRecipient};
 use crate::message::Message;
-use crate::protocol::{self, Context, FromNetworkMsg, Protocol, ConnectedPeer, ProtocolMsg, ProtocolStatus, PeerInfo};
+use crate::protocol::{self, Context, Protocol, ConnectedPeer, ProtocolMsg, ProtocolStatus, PeerInfo};
 use crate::config::Params;
 use crate::error::Error;
 use crate::specialization::NetworkSpecialization;
@@ -40,6 +40,9 @@ use crossbeam_channel::{self as channel, Receiver, Sender, TryRecvError};
 use tokio::prelude::task::AtomicTask;
 use tokio::runtime::Builder as RuntimeBuilder;
 
+/// Interval at which we send status updates on the SyncProvider status stream.
+const STATUS_INTERVAL: Duration = Duration::from_millis(5000);
+
 pub use network_libp2p::PeerId;
 
 /// Type that represents fetch completion future.
@@ -81,22 +84,22 @@ pub trait TransactionPool<H: ExHashT, B: BlockT>: Send + Sync {
 #[derive(Clone)]
 pub struct NetworkLink<B: BlockT, S: NetworkSpecialization<B>> {
 	/// The protocol sender
-	pub(crate) protocol_sender: Sender<ProtocolMsg<B, S>>,
+	pub(crate) protocol_sender: mpsc::UnboundedSender<ProtocolMsg<B, S>>,
 	/// The network sender
 	pub(crate) network_sender: NetworkChan<B>,
 }
 
 impl<B: BlockT, S: NetworkSpecialization<B>> Link<B> for NetworkLink<B, S> {
 	fn block_imported(&self, hash: &B::Hash, number: NumberFor<B>) {
-		let _ = self.protocol_sender.send(ProtocolMsg::BlockImportedSync(hash.clone(), number));
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::BlockImportedSync(hash.clone(), number));
 	}
 
 	fn blocks_processed(&self, processed_blocks: Vec<B::Hash>, has_error: bool) {
-		let _ = self.protocol_sender.send(ProtocolMsg::BlocksProcessed(processed_blocks, has_error));
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::BlocksProcessed(processed_blocks, has_error));
 	}
 
 	fn justification_imported(&self, who: PeerId, hash: &B::Hash, number: NumberFor<B>, success: bool) {
-		let _ = self.protocol_sender.send(ProtocolMsg::JustificationImportResult(hash.clone(), number, success));
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::JustificationImportResult(hash.clone(), number, success));
 		if !success {
 			info!("Invalid justification provided by {} for #{}", who, hash);
 			let _ = self.network_sender.send(NetworkMsg::ReportPeer(who.clone(), i32::min_value()));
@@ -105,11 +108,11 @@ impl<B: BlockT, S: NetworkSpecialization<B>> Link<B> for NetworkLink<B, S> {
 	}
 
 	fn clear_justification_requests(&self) {
-		let _ = self.protocol_sender.send(ProtocolMsg::ClearJustificationRequests);
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::ClearJustificationRequests);
 	}
 
 	fn request_justification(&self, hash: &B::Hash, number: NumberFor<B>) {
-		let _ = self.protocol_sender.send(ProtocolMsg::RequestJustification(hash.clone(), number));
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::RequestJustification(hash.clone(), number));
 	}
 
 	fn report_peer(&self, who: PeerId, reputation_change: i32) {
@@ -117,7 +120,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>> Link<B> for NetworkLink<B, S> {
 	}
 
 	fn restart(&self) {
-		let _ = self.protocol_sender.send(ProtocolMsg::RestartSync);
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::RestartSync);
 	}
 }
 
@@ -151,7 +154,7 @@ pub struct Service<B: BlockT + 'static, S: NetworkSpecialization<B>> {
 	/// nodes it should be connected to or not.
 	peerset: PeersetHandle,
 	/// Protocol sender
-	protocol_sender: Sender<ProtocolMsg<B, S>>,
+	protocol_sender: mpsc::UnboundedSender<ProtocolMsg<B, S>>,
 	/// Sender for messages to the background service task, and handle for the background thread.
 	/// Dropping the sender should close the task and the thread.
 	/// This is an `Option` because we need to extract it in the destructor.
@@ -171,8 +174,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 		let is_offline = Arc::new(AtomicBool::new(true));
 		let is_major_syncing = Arc::new(AtomicBool::new(false));
 		let peers: Arc<RwLock<HashMap<PeerId, ConnectedPeer<B>>>> = Arc::new(Default::default());
-		let (protocol_sender, network_to_protocol_sender) = Protocol::new(
+		let (protocol, protocol_sender) = Protocol::new(
-			status_sinks.clone(),
 			is_offline.clone(),
 			is_major_syncing.clone(),
 			peers.clone(),
@@ -187,8 +189,9 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 		let versions: Vec<_> = ((protocol::MIN_VERSION as u8)..=(protocol::CURRENT_VERSION as u8)).collect();
 		let registered = RegisteredProtocol::new(protocol_id, &versions);
 		let (thread, network, peerset) = start_thread(
-			network_to_protocol_sender,
+			protocol,
 			network_port,
+			status_sinks.clone(),
 			params.network_config,
 			registered,
 		)?;
@@ -236,19 +239,19 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 	pub fn on_block_imported(&self, hash: B::Hash, header: B::Header) {
 		let _ = self
 			.protocol_sender
-			.send(ProtocolMsg::BlockImported(hash, header));
+			.unbounded_send(ProtocolMsg::BlockImported(hash, header));
 	}
 
 	/// Called when a new block is finalized by the client.
 	pub fn on_block_finalized(&self, hash: B::Hash, header: B::Header) {
 		let _ = self
 			.protocol_sender
-			.send(ProtocolMsg::BlockFinalized(hash, header));
+			.unbounded_send(ProtocolMsg::BlockFinalized(hash, header));
 	}
 
 	/// Called when new transactons are imported by the client.
 	pub fn trigger_repropagate(&self) {
-		let _ = self.protocol_sender.send(ProtocolMsg::PropagateExtrinsics);
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::PropagateExtrinsics);
 	}
 
 	/// Make sure an important block is propagated to peers.
@@ -256,7 +259,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 	/// In chain-based consensus, we often need to make sure non-best forks are
 	/// at least temporarily synced.
 	pub fn announce_block(&self, hash: B::Hash) {
-		let _ = self.protocol_sender.send(ProtocolMsg::AnnounceBlock(hash));
+		let _ = self.protocol_sender.unbounded_send(ProtocolMsg::AnnounceBlock(hash));
 	}
 
 	/// Send a consensus message through the gossip
@@ -269,7 +272,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 	) {
 		let _ = self
 			.protocol_sender
-			.send(ProtocolMsg::GossipConsensusMessage(
+			.unbounded_send(ProtocolMsg::GossipConsensusMessage(
 				topic, engine_id, message, recipient,
 			));
 	}
@@ -286,7 +289,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 	{
 		let _ = self
 			.protocol_sender
-			.send(ProtocolMsg::ExecuteWithSpec(Box::new(f)));
+			.unbounded_send(ProtocolMsg::ExecuteWithSpec(Box::new(f)));
 	}
 
 	/// Execute a closure with the consensus gossip.
@@ -295,7 +298,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 	{
 		let _ = self
 			.protocol_sender
-			.send(ProtocolMsg::ExecuteWithGossip(Box::new(f)));
+			.unbounded_send(ProtocolMsg::ExecuteWithGossip(Box::new(f)));
 	}
 
 	/// Are we in the process of downloading the chain?
@@ -471,9 +474,10 @@ pub enum NetworkMsg<B: BlockT + 'static> {
 }
 
 /// Starts the background thread that handles the networking.
-fn start_thread<B: BlockT + 'static>(
+fn start_thread<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT>(
-	protocol_sender: Sender<FromNetworkMsg<B>>,
+	protocol: Protocol<B, S, H>,
 	network_port: NetworkPort<B>,
+	status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<ProtocolStatus<B>>>>>,
 	config: NetworkConfiguration,
 	registered: RegisteredProtocol<Message<B>>,
 ) -> Result<((oneshot::Sender<()>, thread::JoinHandle<()>), Arc<Mutex<NetworkService<Message<B>>>>, PeersetHandle), Error> {
@@ -491,7 +495,7 @@ fn start_thread<B: BlockT + 'static>(
 	let mut runtime = RuntimeBuilder::new().name_prefix("libp2p-").build()?;
 	let peerset_clone = peerset.clone();
 	let thread = thread::Builder::new().name("network".to_string()).spawn(move || {
-		let fut = run_thread(protocol_sender, service_clone, network_port, peerset_clone)
+		let fut = run_thread(protocol, service_clone, network_port, status_sinks, peerset_clone)
 			.select(close_rx.then(|_| Ok(())))
 			.map(|(val, _)| val)
 			.map_err(|(err,_ )| err);
@@ -508,14 +512,28 @@ fn start_thread<B: BlockT + 'static>(
 }
 
 /// Runs the background thread that handles the networking.
-fn run_thread<B: BlockT + 'static>(
+fn run_thread<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT>(
-	protocol_sender: Sender<FromNetworkMsg<B>>,
+	mut protocol: Protocol<B, S, H>,
 	network_service: Arc<Mutex<NetworkService<Message<B>>>>,
 	network_port: NetworkPort<B>,
+	status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<ProtocolStatus<B>>>>>,
 	peerset: PeersetHandle,
 ) -> impl Future<Item = (), Error = io::Error> {
+	// Interval at which we send status updates on the `status_sinks`.
+	let mut status_interval = tokio::timer::Interval::new_interval(STATUS_INTERVAL);
 
 	futures::future::poll_fn(move || {
+		while let Ok(Async::Ready(_)) = status_interval.poll() {
+			let status = protocol.status();
+			status_sinks.lock().retain(|sink| sink.unbounded_send(status.clone()).is_ok());
+		}
+
+		match protocol.poll() {
+			Ok(Async::Ready(())) => return Ok(Async::Ready(())),
+			Ok(Async::NotReady) => {}
+			Err(err) => void::unreachable(err),
+		}
+
 		loop {
 			match network_port.take_one_message() {
 				Ok(None) => break,
@@ -540,19 +558,17 @@ fn run_thread<B: BlockT + 'static>(
 						version <= protocol::CURRENT_VERSION as u8
 						&& version >= protocol::MIN_VERSION as u8
 					);
-					let _ = protocol_sender.send(FromNetworkMsg::PeerConnected(peer_id, debug_info));
+					protocol.on_peer_connected(peer_id, debug_info);
-				}
-				Ok(Async::Ready(Some(NetworkServiceEvent::ClosedCustomProtocol { peer_id, debug_info, .. }))) => {
-					let _ = protocol_sender.send(FromNetworkMsg::PeerDisconnected(peer_id, debug_info));
-				}
-				Ok(Async::Ready(Some(NetworkServiceEvent::CustomMessage { peer_id, message, .. }))) => {
-					let _ = protocol_sender.send(FromNetworkMsg::CustomMessage(peer_id, message));
 				}
+				Ok(Async::Ready(Some(NetworkServiceEvent::ClosedCustomProtocol { peer_id, debug_info, .. }))) =>
+					protocol.on_peer_disconnected(peer_id, debug_info),
+				Ok(Async::Ready(Some(NetworkServiceEvent::CustomMessage { peer_id, message, .. }))) =>
+					protocol.on_custom_message(peer_id, message),
 				Ok(Async::Ready(Some(NetworkServiceEvent::Clogged { peer_id, messages, .. }))) => {
 					debug!(target: "sync", "{} clogging messages:", messages.len());
 					for msg in messages.into_iter().take(5) {
 						debug!(target: "sync", "{:?}", msg);
-						let _ = protocol_sender.send(FromNetworkMsg::PeerClogged(peer_id.clone(), Some(msg)));
+						protocol.on_clogged_peer(peer_id.clone(), Some(msg));
 					}
 				}
 				Ok(Async::Ready(None)) => return Ok(Async::Ready(())),

substrate/core/network/src/test/mod.rs

@@ -34,14 +34,13 @@ use consensus::import_queue::{Link, SharedBlockImport, SharedJustificationImport
 use consensus::{Error as ConsensusError, ErrorKind as ConsensusErrorKind};
 use consensus::{BlockOrigin, ForkChoiceStrategy, ImportBlock, JustificationImport};
 use crate::consensus_gossip::{ConsensusGossip, MessageRecipient as GossipMessageRecipient, TopicNotification};
-use crossbeam_channel::{Sender, RecvError};
+use crossbeam_channel::RecvError;
-use futures::Future;
+use futures::{prelude::*, sync::{mpsc, oneshot}};
-use futures::sync::{mpsc, oneshot};
 use crate::message::Message;
 use network_libp2p::PeerId;
 use parking_lot::{Mutex, RwLock};
 use primitives::{H256, sr25519::Public as AuthorityId};
-use crate::protocol::{ConnectedPeer, Context, FromNetworkMsg, Protocol, ProtocolMsg};
+use crate::protocol::{ConnectedPeer, Context, Protocol, ProtocolMsg};
 use runtime_primitives::generic::BlockId;
 use runtime_primitives::traits::{AuthorityIdFor, Block as BlockT, Digest, DigestItem, Header, NumberFor};
 use runtime_primitives::{Justification, ConsensusEngineId};
@@ -120,13 +119,13 @@ pub struct TestLink<S: NetworkSpecialization<Block>> {
 	link: NetworkLink<Block, S>,
 
 	#[cfg(any(test, feature = "test-helpers"))]
-	network_to_protocol_sender: Sender<FromNetworkMsg<Block>>,
+	network_to_protocol_sender: mpsc::UnboundedSender<FromNetworkMsg<Block>>,
 }
 
 impl<S: NetworkSpecialization<Block>> TestLink<S> {
 	fn new(
-		protocol_sender: Sender<ProtocolMsg<Block, S>>,
+		protocol_sender: mpsc::UnboundedSender<ProtocolMsg<Block, S>>,
-		_network_to_protocol_sender: Sender<FromNetworkMsg<Block>>,
+		_network_to_protocol_sender: mpsc::UnboundedSender<FromNetworkMsg<Block>>,
 		network_sender: NetworkChan<Block>
 	) -> TestLink<S> {
 		TestLink {
@@ -172,7 +171,7 @@ impl<S: NetworkSpecialization<Block>> Link<Block> for TestLink<S> {
 	#[cfg(any(test, feature = "test-helpers"))]
 	fn synchronized(&self) {
 		trace!(target: "test_network", "Synchronizing");
-		drop(self.network_to_protocol_sender.send(FromNetworkMsg::Synchronize))
+		drop(self.network_to_protocol_sender.unbounded_send(FromNetworkMsg::Synchronize))
 	}
 }
 
@@ -191,18 +190,29 @@ pub struct Peer<D, S: NetworkSpecialization<Block>> {
 
 type MessageFilter = Fn(&NetworkMsg<Block>) -> bool;
 
+enum FromNetworkMsg<B: BlockT> {
+	/// A peer connected, with debug info.
+	PeerConnected(PeerId, String),
+	/// A peer disconnected, with debug info.
+	PeerDisconnected(PeerId, String),
+	/// A custom message from another peer.
+	CustomMessage(PeerId, Message<B>),
+	/// Synchronization request.
+	Synchronize,
+}
+
 struct ProtocolChannel<S: NetworkSpecialization<Block>> {
 	buffered_messages: Mutex<VecDeque<NetworkMsg<Block>>>,
-	network_to_protocol_sender: Sender<FromNetworkMsg<Block>>,
+	network_to_protocol_sender: mpsc::UnboundedSender<FromNetworkMsg<Block>>,
-	client_to_protocol_sender: Sender<ProtocolMsg<Block, S>>,
+	client_to_protocol_sender: mpsc::UnboundedSender<ProtocolMsg<Block, S>>,
 	protocol_to_network_receiver: NetworkPort<Block>,
 }
 
 impl<S: NetworkSpecialization<Block>> ProtocolChannel<S> {
 	/// Create new buffered network port.
 	pub fn new(
-		network_to_protocol_sender: Sender<FromNetworkMsg<Block>>,
+		network_to_protocol_sender: mpsc::UnboundedSender<FromNetworkMsg<Block>>,
-		client_to_protocol_sender: Sender<ProtocolMsg<Block, S>>,
+		client_to_protocol_sender: mpsc::UnboundedSender<ProtocolMsg<Block, S>>,
 		protocol_to_network_receiver: NetworkPort<Block>,
 	) -> Self {
 		ProtocolChannel {
@@ -215,17 +225,17 @@ impl<S: NetworkSpecialization<Block>> ProtocolChannel<S> {
 
 	/// Send message from network to protocol.
 	pub fn send_from_net(&self, message: FromNetworkMsg<Block>) {
-		let _ = self.network_to_protocol_sender.send(message);
+		let _ = self.network_to_protocol_sender.unbounded_send(message);
 
-		let _ = self.network_to_protocol_sender.send(FromNetworkMsg::Synchronize);
+		let _ = self.network_to_protocol_sender.unbounded_send(FromNetworkMsg::Synchronize);
 		let _ = self.wait_sync();
 	}
 
 	/// Send message from client to protocol.
 	pub fn send_from_client(&self, message: ProtocolMsg<Block, S>) {
-		let _ = self.client_to_protocol_sender.send(message);
+		let _ = self.client_to_protocol_sender.unbounded_send(message);
 
-		let _ = self.client_to_protocol_sender.send(ProtocolMsg::Synchronize);
+		let _ = self.client_to_protocol_sender.unbounded_send(ProtocolMsg::Synchronize);
 		let _ = self.wait_sync();
 	}
 
@@ -290,8 +300,8 @@ impl<D, S: NetworkSpecialization<Block>> Peer<D, S> {
 		peers: Arc<RwLock<HashMap<PeerId, ConnectedPeer<Block>>>>,
 		client: Arc<PeersClient>,
 		import_queue: Box<BasicQueue<Block>>,
-		network_to_protocol_sender: Sender<FromNetworkMsg<Block>>,
+		network_to_protocol_sender: mpsc::UnboundedSender<FromNetworkMsg<Block>>,
-		protocol_sender: Sender<ProtocolMsg<Block, S>>,
+		protocol_sender: mpsc::UnboundedSender<ProtocolMsg<Block, S>>,
 		network_sender: NetworkChan<Block>,
 		network_port: NetworkPort<Block>,
 		data: D,
@@ -642,14 +652,14 @@ pub trait TestNetFactory: Sized {
 		let (network_sender, network_port) = network_channel();
 
 		let import_queue = Box::new(BasicQueue::new(verifier, block_import, justification_import));
-		let status_sinks = Arc::new(Mutex::new(Vec::new()));
 		let is_offline = Arc::new(AtomicBool::new(true));
 		let is_major_syncing = Arc::new(AtomicBool::new(false));
 		let specialization = self::SpecializationFactory::create();
 		let peers: Arc<RwLock<HashMap<PeerId, ConnectedPeer<Block>>>> = Arc::new(Default::default());
 
-		let (protocol_sender, network_to_protocol_sender) = Protocol::new(
+		let (network_to_protocol_sender, mut network_to_protocol_rx) = mpsc::unbounded();
-			status_sinks,
+
+		let (mut protocol, protocol_sender) = Protocol::new(
 			is_offline.clone(),
 			is_major_syncing.clone(),
 			peers.clone(),
@@ -662,6 +672,29 @@ pub trait TestNetFactory: Sized {
 			specialization,
 		).unwrap();
 
+		std::thread::spawn(move || {
+			tokio::run(futures::future::poll_fn(move || {
+				while let Async::Ready(msg) = network_to_protocol_rx.poll().unwrap() {
+					match msg {
+						Some(FromNetworkMsg::PeerConnected(peer_id, debug_msg)) =>
+							protocol.on_peer_connected(peer_id, debug_msg),
+						Some(FromNetworkMsg::PeerDisconnected(peer_id, debug_msg)) =>
+							protocol.on_peer_disconnected(peer_id, debug_msg),
+						Some(FromNetworkMsg::CustomMessage(peer_id, message)) =>
+							protocol.on_custom_message(peer_id, message),
+						Some(FromNetworkMsg::Synchronize) => protocol.synchronize(),
+						None => return Ok(Async::Ready(()))
+					}
+				}
+
+				if let Async::Ready(_) = protocol.poll().unwrap() {
+					return Ok(Async::Ready(()))
+				}
+
+				Ok(Async::NotReady)
+			}));
+		});
+
 		let peer = Arc::new(Peer::new(
 			is_offline,
 			is_major_syncing,