Rewrite network protocol/service to use channels (#1340)

* rewrite network protocol/service to use channels * remove use of unwrap * re-introduce with_spec * remove unnecessary mut * remove unused param * improve with_spec, add with_gossip * rename job to task * style: re-add comma * remove extra string allocs * rename use of channel * turn TODO into FIXME * remove mut in match * remove Self in new * pass headers by value to network service * remove network sender from service * remove TODO * better expect * rationalize use of network sender in ondemand
2026-04-28 07:27:55 +00:00 · 2019-02-06 19:54:02 +08:00
parent 8aae19e2db
commit a2d2ed69ab
19 changed files with 1314 additions and 903 deletions
@@ -17,30 +17,27 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::{io, thread};
-use std::time::Duration;
-use futures::{self, Future, Stream, stream, sync::oneshot};
-use parking_lot::{Mutex, RwLock};
+use futures::{Async, Future, Stream, stream, sync::oneshot};
+use parking_lot::Mutex;
 use network_libp2p::{ProtocolId, PeerId, NetworkConfiguration, NodeIndex, ErrorKind, Severity};
-use network_libp2p::{start_service, Service as NetworkService, ServiceEvent as NetworkServiceEvent};
-use network_libp2p::{RegisteredProtocol, parse_str_addr, Protocol as Libp2pProtocol};
-use io::NetSyncIo;
+use network_libp2p::{start_service, parse_str_addr, Service as NetworkService, ServiceEvent as NetworkServiceEvent};
+use network_libp2p::{Protocol as Libp2pProtocol, RegisteredProtocol};
 use consensus::import_queue::{ImportQueue, Link};
 use consensus_gossip::ConsensusGossip;
-use protocol::{self, Protocol, ProtocolContext, Context, ProtocolStatus, PeerInfo};
+use protocol::{self, Context, Protocol, ProtocolMsg, ProtocolStatus, PeerInfo};
+use codec::Decode;
 use config::Params;
+use crossbeam_channel::{self as channel, Receiver, Sender, TryRecvError};
 use error::Error;
-use specialization::NetworkSpecialization;
 use runtime_primitives::traits::{Block as BlockT, NumberFor};
-use sync::ChainSync;
-use std::sync::Weak;
-use tokio::{runtime::Runtime, timer::Interval};
+use specialization::NetworkSpecialization;
+
+use tokio::prelude::task::AtomicTask;
+use tokio::runtime::Runtime;

 /// Type that represents fetch completion future.
 pub type FetchFuture = oneshot::Receiver<Vec<u8>>;

-const TICK_TIMEOUT: Duration = Duration::from_millis(1000);
-const PROPAGATE_TIMEOUT: Duration = Duration::from_millis(5000);
-
 /// Sync status
 pub trait SyncProvider<B: BlockT>: Send + Sync {
 	/// Get sync status
@@ -50,8 +47,14 @@ pub trait SyncProvider<B: BlockT>: Send + Sync {
 }

 /// Minimum Requirements for a Hash within Networking
-pub trait ExHashT: ::std::hash::Hash + Eq + ::std::fmt::Debug + Clone + Send + Sync + 'static {}
-impl<T> ExHashT for T where T: ::std::hash::Hash + Eq + ::std::fmt::Debug + Clone + Send + Sync + 'static {}
+pub trait ExHashT:
+	::std::hash::Hash + Eq + ::std::fmt::Debug + Clone + Send + Sync + 'static
+{
+}
+impl<T> ExHashT for T where
+	T: ::std::hash::Hash + Eq + ::std::fmt::Debug + Clone + Send + Sync + 'static
+{
+}

 /// Transaction pool interface
 pub trait TransactionPool<H: ExHashT, B: BlockT>: Send + Sync {
@@ -63,114 +66,98 @@ pub trait TransactionPool<H: ExHashT, B: BlockT>: Send + Sync {
 	fn on_broadcasted(&self, propagations: HashMap<H, Vec<String>>);
 }

-/// Service able to execute closure in the network context.
-pub trait ExecuteInContext<B: BlockT>: Send + Sync {
-	/// Execute closure in network context.
-	fn execute_in_context<F: Fn(&mut Context<B>)>(&self, closure: F);
-}
-
 /// A link implementation that connects to the network.
-pub struct NetworkLink<B: BlockT, E: ExecuteInContext<B>> {
-	/// The chain-sync handle
-	pub(crate) sync: Weak<RwLock<ChainSync<B>>>,
-	/// Network context.
-	pub(crate) context: Weak<E>,
+pub struct NetworkLink<B: BlockT, S: NetworkSpecialization<B>> {
+	/// The protocol sender
+	pub(crate) protocol_sender: Sender<ProtocolMsg<B, S>>,
+	/// The network sender
+	pub(crate) network_sender: NetworkChan,
 }

-impl<B: BlockT, E: ExecuteInContext<B>> NetworkLink<B, E> {
-	/// Execute closure with locked ChainSync.
-	fn with_sync<F: Fn(&mut ChainSync<B>, &mut Context<B>)>(&self, closure: F) {
-		if let (Some(sync), Some(service)) = (self.sync.upgrade(), self.context.upgrade()) {
-			service.execute_in_context(move |protocol| {
-				let mut sync = sync.write();
-				closure(&mut *sync, protocol)
-			});
-		}
-	}
-}
-
-impl<B: BlockT, E: ExecuteInContext<B>> Link<B> for NetworkLink<B, E> {
+impl<B: BlockT, S: NetworkSpecialization<B>> Link<B> for NetworkLink<B, S> {
 	fn block_imported(&self, hash: &B::Hash, number: NumberFor<B>) {
-		self.with_sync(|sync, _| sync.block_imported(&hash, number))
+		let _ = self.protocol_sender.send(ProtocolMsg::BlockImportedSync(hash.clone(), number));
 	}

 	fn request_justification(&self, hash: &B::Hash, number: NumberFor<B>) {
-		self.with_sync(|sync, protocol| sync.request_justification(hash, number, protocol))
+		let _ = self.protocol_sender.send(ProtocolMsg::RequestJustification(hash.clone(), number));
 	}

 	fn maintain_sync(&self) {
-		self.with_sync(|sync, protocol| sync.maintain_sync(protocol))
+		let _ = self.protocol_sender.send(ProtocolMsg::MaintainSync);
 	}

 	fn useless_peer(&self, who: NodeIndex, reason: &str) {
 		trace!(target:"sync", "Useless peer {}, {}", who, reason);
-		self.with_sync(|_, protocol| protocol.report_peer(who, Severity::Useless(reason)))
+		self.network_sender.send(NetworkMsg::ReportPeer(who, Severity::Useless(reason.to_string())));
 	}

 	fn note_useless_and_restart_sync(&self, who: NodeIndex, reason: &str) {
 		trace!(target:"sync", "Bad peer {}, {}", who, reason);
-		self.with_sync(|sync, protocol| {
-			protocol.report_peer(who, Severity::Useless(reason));	// is this actually malign or just useless?
-			sync.restart(protocol);
-		})
+		// is this actually malign or just useless?
+		self.network_sender.send(NetworkMsg::ReportPeer(who, Severity::Useless(reason.to_string())));
+		let _ = self.protocol_sender.send(ProtocolMsg::RestartSync);
 	}

 	fn restart(&self) {
-		self.with_sync(|sync, protocol| sync.restart(protocol))
+		let _ = self.protocol_sender.send(ProtocolMsg::RestartSync);
 	}
 }

 /// Substrate network service. Handles network IO and manages connectivity.
-pub struct Service<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> {
+pub struct Service<B: BlockT + 'static, S: NetworkSpecialization<B>> {
 	/// Network service
 	network: Arc<Mutex<NetworkService>>,
-	/// Protocol handler
-	handler: Arc<Protocol<B, S, H>>,
-	/// Protocol ID.
-	protocol_id: ProtocolId,
+	/// Protocol sender
+	protocol_sender: Sender<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.
 	bg_thread: Option<(oneshot::Sender<()>, thread::JoinHandle<()>)>,
 }

-impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Service<B, S, H> {
+impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Service<B, S> {
 	/// Creates and register protocol with the network service
-	pub fn new<I: 'static + ImportQueue<B>>(
+	pub fn new<I: 'static + ImportQueue<B>, H: ExHashT>(
 		params: Params<B, S, H>,
 		protocol_id: ProtocolId,
 		import_queue: Arc<I>,
-	) -> Result<Arc<Service<B, S, H>>, Error>
-		where I: ImportQueue<B>
-	{
-		let handler = Arc::new(Protocol::new(
+	) -> Result<(Arc<Service<B, S>>, NetworkChan), Error> {
+		let (network_chan, network_port) = network_channel(protocol_id);
+		let protocol_sender = Protocol::new(
+			network_chan.clone(),
 			params.config,
 			params.chain,
 			import_queue.clone(),
 			params.on_demand,
 			params.transaction_pool,
 			params.specialization,
-		)?);
+		)?;
 		let versions = [(protocol::CURRENT_VERSION as u8)];
 		let registered = RegisteredProtocol::new(protocol_id, &versions[..]);
-		let (thread, network) = start_thread(params.network_config, handler.clone(), registered)?;
+		let (thread, network) = start_thread(
+			protocol_sender.clone(),
+			network_port,
+			network_chan.clone(),
+			params.network_config,
+			registered,
+		)?;

 		let service = Arc::new(Service {
 			network,
-			protocol_id,
-			handler,
-			bg_thread: Some(thread)
+			protocol_sender: protocol_sender.clone(),
+			bg_thread: Some(thread),
 		});

 		// connect the import-queue to the network service.
 		let link = NetworkLink {
-			sync: Arc::downgrade(service.handler.sync()),
-			context: Arc::downgrade(&service),
+			protocol_sender,
+			network_sender: network_chan.clone(),
 		};

 		import_queue.start(link)?;

-		Ok(service)
+		Ok((service, network_chan))
 	}

 	/// Returns the downloaded bytes per second averaged over the past few seconds.
@@ -186,18 +173,22 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> Service<B, S,
 	}

 	/// Called when a new block is imported by the client.
-	pub fn on_block_imported(&self, hash: B::Hash, header: &B::Header) {
-		self.handler.on_block_imported(&mut NetSyncIo::new(&self.network, self.protocol_id), hash, header)
+	pub fn on_block_imported(&self, hash: B::Hash, header: B::Header) {
+		let _ = self
+			.protocol_sender
+			.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) {
-		self.handler.on_block_finalized(&mut NetSyncIo::new(&self.network, self.protocol_id), hash, header)
+	pub fn on_block_finalized(&self, hash: B::Hash, header: B::Header) {
+		let _ = self
+			.protocol_sender
+			.send(ProtocolMsg::BlockFinalized(hash, header));
 	}

 	/// Called when new transactons are imported by the client.
 	pub fn trigger_repropagate(&self) {
-		self.handler.propagate_extrinsics(&mut NetSyncIo::new(&self.network, self.protocol_id));
+		let _ = self.protocol_sender.send(ProtocolMsg::PropagateExtrinsics);
 	}

 	/// Make sure an important block is propagated to peers.
@@ -205,43 +196,60 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> 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) {
-		self.handler.announce_block(&mut NetSyncIo::new(&self.network, self.protocol_id), hash);
+		let _ = self.protocol_sender.send(ProtocolMsg::AnnounceBlock(hash));
 	}

 	/// Send a consensus message through the gossip
 	pub fn gossip_consensus_message(&self, topic: B::Hash, message: Vec<u8>, broadcast: bool) {
-		self.handler.gossip_consensus_message(
-			&mut NetSyncIo::new(&self.network, self.protocol_id),
-			topic,
-			message,
-			broadcast,
-		)
-	}
-	/// Execute a closure with the chain-specific network specialization.
-	pub fn with_spec<F, U>(&self, f: F) -> U
-		where F: FnOnce(&mut S, &mut Context<B>) -> U
-	{
-		self.handler.with_spec(&mut NetSyncIo::new(&self.network, self.protocol_id), f)
+		let _ = self
+			.protocol_sender
+			.send(ProtocolMsg::GossipConsensusMessage(
+				topic, message, broadcast,
+			));
 	}

-	/// access the underlying consensus gossip handler
-	pub fn consensus_gossip<'a>(&'a self) -> &'a RwLock<ConsensusGossip<B>> {
-		self.handler.consensus_gossip()
+	/// Execute a closure with the chain-specific network specialization.
+	pub fn with_spec<F>(&self, f: F)
+		where F: FnOnce(&mut S, &mut Context<B>) + Send + 'static
+	{
+		let _ = self
+			.protocol_sender
+			.send(ProtocolMsg::ExecuteWithSpec(Box::new(f)));
+	}
+
+	/// Execute a closure with the consensus gossip.
+	pub fn with_gossip<F>(&self, f: F)
+		where F: FnOnce(&mut ConsensusGossip<B>, &mut Context<B>) + Send + 'static
+	{
+		let _ = self
+			.protocol_sender
+			.send(ProtocolMsg::ExecuteWithGossip(Box::new(f)));
 	}
 }

-impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> ::consensus::SyncOracle for Service<B, S, H> {
+impl<B: BlockT + 'static, S: NetworkSpecialization<B>> ::consensus::SyncOracle for Service<B, S> {
 	fn is_major_syncing(&self) -> bool {
-		self.handler.sync().read().status().is_major_syncing()
+		let (sender, port) = channel::unbounded();
+		let _ = self
+			.protocol_sender
+			.send(ProtocolMsg::IsMajorSyncing(sender));
+		port.recv().expect("1. Protocol keeps handling messages until all senders are dropped,
+			or the ProtocolMsg::Stop message is received,
+			2 Service keeps a sender to protocol, and the ProtocolMsg::Stop is never sent.")
 	}
 	fn is_offline(&self) -> bool {
-		self.handler.sync().read().status().is_offline()
+		let (sender, port) = channel::unbounded();
+		let _ = self
+			.protocol_sender
+			.send(ProtocolMsg::IsOffline(sender));
+		port.recv().expect("1. Protocol keeps handling messages until all senders are dropped,
+			or the ProtocolMsg::Stop message is received,
+			2 Service keeps a sender to protocol, and the ProtocolMsg::Stop is never sent.")
 	}
 }

-impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H:ExHashT> Drop for Service<B, S, H> {
+impl<B: BlockT + 'static, S: NetworkSpecialization<B>> Drop for Service<B, S> {
 	fn drop(&mut self) {
-		self.handler.stop();
 		if let Some((sender, join)) = self.bg_thread.take() {
 			let _ = sender.send(());
 			if let Err(e) = join.join() {
@@ -251,20 +259,22 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H:ExHashT> Drop for Servi
 	}
 }

-impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> ExecuteInContext<B> for Service<B, S, H> {
-	fn execute_in_context<F: Fn(&mut ::protocol::Context<B>)>(&self, closure: F) {
-		closure(&mut ProtocolContext::new(self.handler.context_data(), &mut NetSyncIo::new(&self.network, self.protocol_id)))
-	}
-}
-
-impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> SyncProvider<B> for Service<B, S, H> {
+impl<B: BlockT + 'static, S: NetworkSpecialization<B>> SyncProvider<B> for Service<B, S> {
 	/// Get sync status
 	fn status(&self) -> ProtocolStatus<B> {
-		self.handler.status()
+		let (sender, port) = channel::unbounded();
+		let _ = self.protocol_sender.send(ProtocolMsg::Status(sender));
+		port.recv().expect("1. Protocol keeps handling messages until all senders are dropped,
+			or the ProtocolMsg::Stop message is received,
+			2 Service keeps a sender to protocol, and the ProtocolMsg::Stop is never sent.")
 	}

 	fn peers(&self) -> Vec<(NodeIndex, Option<PeerId>, PeerInfo<B>)> {
-		let peers = self.handler.peers();
+		let (sender, port) = channel::unbounded();
+		let _ = self.protocol_sender.send(ProtocolMsg::Peers(sender));
+		let peers = port.recv().expect("1. Protocol keeps handling messages until all senders are dropped,
+			or the ProtocolMsg::Stop message is received,
+			2 Service keeps a sender to protocol, and the ProtocolMsg::Stop is never sent.");
 		let network = self.network.lock();
 		peers.into_iter().map(|(idx, info)| {
 			(idx, network.peer_id_of_node(idx).map(|p| p.clone()), info)
@@ -273,7 +283,7 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> SyncProvider<
 }

 /// Trait for managing network
-pub trait ManageNetwork: Send + Sync {
+pub trait ManageNetwork {
 	/// Set to allow unreserved peers to connect
 	fn accept_unreserved_peers(&self);
 	/// Set to deny unreserved peers to connect
@@ -286,7 +296,7 @@ pub trait ManageNetwork: Send + Sync {
 	fn node_id(&self) -> Option<String>;
 }

-impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> ManageNetwork for Service<B, S, H> {
+impl<B: BlockT + 'static, S: NetworkSpecialization<B>> ManageNetwork for Service<B, S> {
 	fn accept_unreserved_peers(&self) {
 		self.network.lock().accept_unreserved_peers();
 	}
@@ -319,10 +329,102 @@ impl<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT> ManageNetwork
 	}
 }

+
+/// Create a NetworkPort/Chan pair.
+pub fn network_channel(protocol_id: ProtocolId) -> (NetworkChan, NetworkPort) {
+	let (network_sender, network_receiver) = channel::unbounded();
+	let task_notify = Arc::new(AtomicTask::new());
+	let network_port = NetworkPort::new(network_receiver, protocol_id, task_notify.clone());
+	let network_chan = NetworkChan::new(network_sender, task_notify);
+	(network_chan, network_port)
+}
+
+
+/// A sender of NetworkMsg that notifies a task when a message has been sent.
+#[derive(Clone)]
+pub struct NetworkChan {
+	sender: Sender<NetworkMsg>,
+	task_notify: Arc<AtomicTask>,
+}
+
+impl NetworkChan {
+	/// Create a new network chan.
+	pub fn new(sender: Sender<NetworkMsg>, task_notify: Arc<AtomicTask>) -> Self {
+		 NetworkChan {
+			sender,
+			task_notify,
+		}
+	}
+
+	/// Send a messaging, to be handled on a stream. Notify the task handling the stream.
+	pub fn send(&self, msg: NetworkMsg) {
+		let _ = self.sender.send(msg);
+		self.task_notify.notify();
+	}
+}
+
+impl Drop for NetworkChan {
+	/// Notifying the task when a sender is dropped(when all are dropped, the stream is finished).
+	fn drop(&mut self) {
+		self.task_notify.notify();
+	}
+}
+
+
+/// A receiver of NetworkMsg that makes the protocol-id available with each message.
+pub struct NetworkPort {
+	receiver: Receiver<NetworkMsg>,
+	protocol_id: ProtocolId,
+	task_notify: Arc<AtomicTask>,
+}
+
+impl NetworkPort {
+	/// Create a new network port for a given protocol-id.
+	pub fn new(receiver: Receiver<NetworkMsg>, protocol_id: ProtocolId, task_notify: Arc<AtomicTask>) -> Self {
+		Self {
+			receiver,
+			protocol_id,
+			task_notify,
+		}
+	}
+
+	/// Receive a message, if any is currently-enqueued.
+	/// Register the current tokio task for notification when a new message is available.
+	pub fn take_one_message(&self) -> Result<Option<(ProtocolId, NetworkMsg)>, ()> {
+		self.task_notify.register();
+		match self.receiver.try_recv() {
+			Ok(msg) => Ok(Some((self.protocol_id.clone(), msg))),
+			Err(TryRecvError::Empty) => Ok(None),
+			Err(TryRecvError::Disconnected) => Err(()),
+		}
+	}
+
+	/// Get a reference to the underlying crossbeam receiver.
+	#[cfg(any(test, feature = "test-helpers"))]
+	pub fn receiver(&self) -> &Receiver<NetworkMsg> {
+		&self.receiver
+	}
+}
+
+/// Messages to be handled by NetworkService.
+#[derive(Debug)]
+pub enum NetworkMsg {
+	/// Ask network to convert a list of nodes, to a list of peers.
+	PeerIds(Vec<NodeIndex>, Sender<Vec<(NodeIndex, Option<PeerId>)>>),
+	/// Send an outgoing custom message.
+	Outgoing(NodeIndex, Vec<u8>),
+	/// Report a peer.
+	ReportPeer(NodeIndex, Severity),
+	/// Get a peer id.
+	GetPeerId(NodeIndex, Sender<Option<String>>),
+}
+
 /// Starts the background thread that handles the networking.
-fn start_thread<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT>(
+fn start_thread<B: BlockT + 'static, S: NetworkSpecialization<B>>(
+	protocol_sender: Sender<ProtocolMsg<B, S>>,
+	network_port: NetworkPort,
+	network_sender: NetworkChan,
 	config: NetworkConfiguration,
-	protocol: Arc<Protocol<B, S, H>>,
 	registered: RegisteredProtocol,
 ) -> Result<((oneshot::Sender<()>, thread::JoinHandle<()>), Arc<Mutex<NetworkService>>), Error> {
 	let protocol_id = registered.id();
@@ -344,7 +446,7 @@ fn start_thread<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT>(
 	let service_clone = service.clone();
 	let mut runtime = Runtime::new()?;
 	let thread = thread::Builder::new().name("network".to_string()).spawn(move || {
-		let fut = run_thread(service_clone, protocol, protocol_id)
+		let fut = run_thread(protocol_sender, service_clone, network_sender, network_port, protocol_id)
 			.select(close_rx.then(|_| Ok(())))
 			.map(|(val, _)| val)
 			.map_err(|(err,_ )| err);
@@ -361,82 +463,111 @@ fn start_thread<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT>(
 }

 /// Runs the background thread that handles the networking.
-fn run_thread<B: BlockT + 'static, S: NetworkSpecialization<B>, H: ExHashT>(
+fn run_thread<B: BlockT + 'static, S: NetworkSpecialization<B>>(
+	protocol_sender: Sender<ProtocolMsg<B, S>>,
 	network_service: Arc<Mutex<NetworkService>>,
-	protocol: Arc<Protocol<B, S, H>>,
+	network_sender: NetworkChan,
+	network_port: NetworkPort,
 	protocol_id: ProtocolId,
 ) -> impl Future<Item = (), Error = io::Error> {
-	// Interval for performing maintenance on the protocol handler.
-	let tick = Interval::new_interval(TICK_TIMEOUT)
-		.for_each({
-			let protocol = protocol.clone();
-			let network_service = network_service.clone();
-			move |_| {
-				protocol.tick(&mut NetSyncIo::new(&network_service, protocol_id));
-				Ok(())
-			}
-		})
-		.then(|res| {
-			match res {
-				Ok(()) => (),
-				Err(err) => error!("Error in the propagation timer: {:?}", err),
-			};
-			Ok(())
-		});

-	// Interval at which we gossip extrinsics over the network.
-	let propagate = Interval::new_interval(PROPAGATE_TIMEOUT)
-		.for_each({
-			let protocol = protocol.clone();
-			let network_service = network_service.clone();
-			move |_| {
-				protocol.propagate_extrinsics(&mut NetSyncIo::new(&network_service, protocol_id));
-				Ok(())
+	let network_service_2 = network_service.clone();
+
+	// Protocol produces a stream of messages about what happens in sync.
+	let protocol = stream::poll_fn(move || {
+		match network_port.take_one_message() {
+			Ok(Some(message)) => Ok(Async::Ready(Some(message))),
+			Ok(None) => Ok(Async::NotReady),
+			Err(_) => Err(())
+		}
+	}).for_each(move |(protocol_id, msg)| {
+		// Handle message from Protocol.
+		match msg {
+			NetworkMsg::PeerIds(node_idxs, sender) => {
+				let reply = node_idxs.into_iter().map(|idx| {
+					(idx, network_service_2.lock().peer_id_of_node(idx).map(|p| p.clone()))
+				}).collect::<Vec<_>>();
+				let _ = sender.send(reply);
 			}
-		})
-		.then(|res| {
-			match res {
-				Ok(()) => (),
-				Err(err) => error!("Error in the propagation timer: {:?}", err),
-			};
-			Ok(())
-		});
+			NetworkMsg::Outgoing(who, outgoing_message) => {
+				network_service_2
+					.lock()
+					.send_custom_message(who, protocol_id, outgoing_message);
+			},
+			NetworkMsg::ReportPeer(who, severity) => {
+				match severity {
+					Severity::Bad(_) => network_service_2.lock().ban_node(who),
+					Severity::Useless(_) => network_service_2.lock().drop_node(who),
+					Severity::Timeout => network_service_2.lock().drop_node(who),
+				}
+			},
+			NetworkMsg::GetPeerId(who, sender) => {
+				let node_id = network_service_2
+					.lock()
+					.peer_id_of_node(who)
+					.cloned()
+					.map(|id| id.to_base58());
+				let _ = sender.send(node_id);
+			},
+		}
+		Ok(())
+	})
+	.then(|res| {
+		match res {
+			Ok(()) => (),
+			Err(_) => error!("Protocol disconnected"),
+		};
+		Ok(())
+	});

 	// The network service produces events about what happens on the network. Let's process them.
-	let network_service2 = network_service.clone();
-	let network = stream::poll_fn(move || network_service2.lock().poll()).for_each(move |event| {
-		let mut net_sync = NetSyncIo::new(&network_service, protocol_id);
-
+	let network = stream::poll_fn(move || network_service.lock().poll()).for_each(move |event| {
 		match event {
-			NetworkServiceEvent::ClosedCustomProtocols { node_index, protocols } => {
+			NetworkServiceEvent::ClosedCustomProtocols { node_index, protocols, debug_info } => {
 				if !protocols.is_empty() {
 					debug_assert_eq!(protocols, &[protocol_id]);
-					protocol.on_peer_disconnected(&mut net_sync, node_index);
+					let _ = protocol_sender.send(
+						ProtocolMsg::PeerDisconnected(node_index, debug_info));
 				}
 			}
-			NetworkServiceEvent::OpenedCustomProtocol { node_index, version, .. } => {
+			NetworkServiceEvent::OpenedCustomProtocol { node_index, version, debug_info, .. } => {
 				debug_assert_eq!(version, protocol::CURRENT_VERSION as u8);
-				protocol.on_peer_connected(&mut net_sync, node_index);
+				let _ = protocol_sender.send(ProtocolMsg::PeerConnected(node_index, debug_info));
 			}
-			NetworkServiceEvent::ClosedCustomProtocol { node_index, .. } => {
-				protocol.on_peer_disconnected(&mut net_sync, node_index);
+			NetworkServiceEvent::ClosedCustomProtocol { node_index, debug_info, .. } => {
+				let _ = protocol_sender.send(ProtocolMsg::PeerDisconnected(node_index, debug_info));
 			}
 			NetworkServiceEvent::CustomMessage { node_index, data, .. } => {
-				protocol.handle_packet(&mut net_sync, node_index, &data);
+				if let Some(m) = Decode::decode(&mut (&data as &[u8])) {
+					let _ = protocol_sender.send(ProtocolMsg::CustomMessage(node_index, m));
+					return Ok(())
+				}
+				let _ = network_sender.send(
+					NetworkMsg::ReportPeer(
+						node_index,
+						Severity::Bad("Peer sent us a packet with invalid format".to_string())
+					)
+				);
 			}
 			NetworkServiceEvent::Clogged { node_index, messages, .. } => {
-				protocol.on_clogged_peer(&mut net_sync, node_index,
-					messages.iter().map(|d| d.as_ref()));
+				debug!(target: "sync", "{} clogging messages:", messages.len());
+				for msg_bytes in messages.iter().take(5) {
+					if let Some(msg) = Decode::decode(&mut (&msg_bytes as &[u8])) {
+						debug!(target: "sync", "{:?}", msg);
+						let _ = protocol_sender.send(ProtocolMsg::PeerClogged(node_index, Some(msg)));
+					} else {
+						debug!(target: "sync", "{:?}", msg_bytes);
+						let _ = protocol_sender.send(ProtocolMsg::PeerClogged(node_index, None));
+					}
+				}
 			}
 		};
-
 		Ok(())
 	});

 	// Merge all futures into one.
 	let futures: Vec<Box<Future<Item = (), Error = io::Error> + Send>> = vec![
-		Box::new(tick) as Box<_>,
-		Box::new(propagate) as Box<_>,
+		Box::new(protocol) as Box<_>,
 		Box::new(network) as Box<_>
 	];