Traffic statistics (#4017)

* Network stats

* Fixed tests

substrate/core/network/src/legacy_proto/behaviour.rs

@@ -17,7 +17,7 @@
 use crate::{DiscoveryNetBehaviour, config::ProtocolId};
 use crate::legacy_proto::handler::{CustomProtoHandlerProto, CustomProtoHandlerOut, CustomProtoHandlerIn};
 use crate::legacy_proto::upgrade::RegisteredProtocol;
-use crate::protocol::message::Message;
+use bytes::BytesMut;
 use fnv::FnvHashMap;
 use futures::prelude::*;
 use futures03::{compat::Compat, TryFutureExt as _, StreamExt as _, TryStreamExt as _};
@@ -25,7 +25,6 @@ use libp2p::core::{ConnectedPoint, Multiaddr, PeerId};
 use libp2p::swarm::{NetworkBehaviour, NetworkBehaviourAction, PollParameters};
 use log::{debug, error, trace, warn};
 use rand::distributions::{Distribution as _, Uniform};
-use sr_primitives::traits::Block as BlockT;
 use smallvec::SmallVec;
 use std::{borrow::Cow, collections::hash_map::Entry, cmp, error, marker::PhantomData, mem, pin::Pin};
 use std::time::{Duration, Instant};
@@ -61,9 +60,9 @@ use tokio_io::{AsyncRead, AsyncWrite};
 /// Note that this "banning" system is not an actual ban. If a "banned" node tries to connect to
 /// us, we accept the connection. The "banning" system is only about delaying dialing attempts.
 ///
-pub struct LegacyProto<B: BlockT, TSubstream> {
+pub struct LegacyProto< TSubstream> {
 	/// List of protocols to open with peers. Never modified.
-	protocol: RegisteredProtocol<B>,
+	protocol: RegisteredProtocol,
 
 	/// Receiver for instructions about who to connect to or disconnect from.
 	peerset: peerset::Peerset,
@@ -80,7 +79,7 @@ pub struct LegacyProto<B: BlockT, TSubstream> {
 	next_incoming_index: peerset::IncomingIndex,
 
 	/// Events to produce from `poll()`.
-	events: SmallVec<[NetworkBehaviourAction<CustomProtoHandlerIn<B>, LegacyProtoOut<B>>; 4]>,
+	events: SmallVec<[NetworkBehaviourAction<CustomProtoHandlerIn, LegacyProtoOut>; 4]>,
 
 	/// Marker to pin the generics.
 	marker: PhantomData<TSubstream>,
@@ -189,7 +188,7 @@ struct IncomingPeer {
 
 /// Event that can be emitted by the `LegacyProto`.
 #[derive(Debug)]
-pub enum LegacyProtoOut<B: BlockT> {
+pub enum LegacyProtoOut {
 	/// Opened a custom protocol with the remote.
 	CustomProtocolOpen {
 		/// Version of the protocol that has been opened.
@@ -213,7 +212,7 @@ pub enum LegacyProtoOut<B: BlockT> {
 		/// Id of the peer the message came from.
 		peer_id: PeerId,
 		/// Message that has been received.
-		message: Message<B>,
+		message: BytesMut,
 	},
 
 	/// The substream used by the protocol is pretty large. We should print avoid sending more
@@ -222,11 +221,11 @@ pub enum LegacyProtoOut<B: BlockT> {
 		/// Id of the peer which is clogged.
 		peer_id: PeerId,
 		/// Copy of the messages that are within the buffer, for further diagnostic.
-		messages: Vec<Message<B>>,
+		messages: Vec<Vec<u8>>,
 	},
 }
 
-impl<B: BlockT, TSubstream> LegacyProto<B, TSubstream> {
+impl<TSubstream> LegacyProto<TSubstream> {
 	/// Creates a `CustomProtos`.
 	pub fn new(
 		protocol: impl Into<ProtocolId>,
@@ -350,8 +349,7 @@ impl<B: BlockT, TSubstream> LegacyProto<B, TSubstream> {
 	///
 	/// Also note that even we have a valid open substream, it may in fact be already closed
 	/// without us knowing, in which case the packet will not be received.
-	pub fn send_packet(&mut self, target: &PeerId, message: Message<B>)
-	where B: BlockT {
+	pub fn send_packet(&mut self, target: &PeerId, message: Vec<u8>) {
 		if !self.is_open(target) {
 			return;
 		}
@@ -607,7 +605,7 @@ impl<B: BlockT, TSubstream> LegacyProto<B, TSubstream> {
 	}
 }
 
-impl<B: BlockT, TSubstream> DiscoveryNetBehaviour for LegacyProto<B, TSubstream> {
+impl<TSubstream> DiscoveryNetBehaviour for LegacyProto<TSubstream> {
 	fn add_discovered_nodes(&mut self, peer_ids: impl Iterator<Item = PeerId>) {
 		self.peerset.discovered(peer_ids.into_iter().map(|peer_id| {
 			debug!(target: "sub-libp2p", "PSM <= Discovered({:?})", peer_id);
@@ -616,13 +614,12 @@ impl<B: BlockT, TSubstream> DiscoveryNetBehaviour for LegacyProto<B, TSubstream>
 	}
 }
 
-impl<B, TSubstream> NetworkBehaviour for LegacyProto<B, TSubstream>
+impl<TSubstream> NetworkBehaviour for LegacyProto<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
-	type ProtocolsHandler = CustomProtoHandlerProto<B, TSubstream>;
-	type OutEvent = LegacyProtoOut<B>;
+	type ProtocolsHandler = CustomProtoHandlerProto<TSubstream>;
+	type OutEvent = LegacyProtoOut;
 
 	fn new_handler(&mut self) -> Self::ProtocolsHandler {
 		CustomProtoHandlerProto::new(self.protocol.clone())
@@ -825,7 +822,7 @@ where
 	fn inject_node_event(
 		&mut self,
 		source: PeerId,
-		event: CustomProtoHandlerOut<B>,
+		event: CustomProtoHandlerOut,
 	) {
 		match event {
 			CustomProtoHandlerOut::CustomProtocolClosed { reason } => {
@@ -954,7 +951,7 @@ where
 		_params: &mut impl PollParameters,
 	) -> Async<
 		NetworkBehaviourAction<
-			CustomProtoHandlerIn<B>,
+			CustomProtoHandlerIn,
 			Self::OutEvent,
 		>,
 	> {

substrate/core/network/src/legacy_proto/handler.rs

@@ -15,7 +15,7 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::legacy_proto::upgrade::{RegisteredProtocol, RegisteredProtocolEvent, RegisteredProtocolSubstream};
-use crate::protocol::message::Message;
+use bytes::BytesMut;
 use futures::prelude::*;
 use futures03::{compat::Compat, TryFutureExt as _};
 use futures_timer::Delay;
@@ -29,7 +29,6 @@ use libp2p::swarm::{
 	SubstreamProtocol,
 };
 use log::{debug, error};
-use sr_primitives::traits::Block as BlockT;
 use smallvec::{smallvec, SmallVec};
 use std::{borrow::Cow, error, fmt, io, marker::PhantomData, mem, time::Duration};
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -88,21 +87,20 @@ use tokio_io::{AsyncRead, AsyncWrite};
 /// We consider that we are now "closed" if the remote closes all the existing substreams.
 /// Re-opening it can then be performed by closing all active substream and re-opening one.
 ///
-pub struct CustomProtoHandlerProto<B, TSubstream> {
+pub struct CustomProtoHandlerProto<TSubstream> {
 	/// Configuration for the protocol upgrade to negotiate.
-	protocol: RegisteredProtocol<B>,
+	protocol: RegisteredProtocol,
 
 	/// Marker to pin the generic type.
 	marker: PhantomData<TSubstream>,
 }
 
-impl<B, TSubstream> CustomProtoHandlerProto<B, TSubstream>
+impl<TSubstream> CustomProtoHandlerProto<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
 	/// Builds a new `CustomProtoHandlerProto`.
-	pub fn new(protocol: RegisteredProtocol<B>) -> Self {
+	pub fn new(protocol: RegisteredProtocol) -> Self {
 		CustomProtoHandlerProto {
 			protocol,
 			marker: PhantomData,
@@ -110,14 +108,13 @@ where
 	}
 }
 
-impl<B, TSubstream> IntoProtocolsHandler for CustomProtoHandlerProto<B, TSubstream>
+impl<TSubstream> IntoProtocolsHandler for CustomProtoHandlerProto<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
-	type Handler = CustomProtoHandler<B, TSubstream>;
+	type Handler = CustomProtoHandler<TSubstream>;
 
-	fn inbound_protocol(&self) -> RegisteredProtocol<B> {
+	fn inbound_protocol(&self) -> RegisteredProtocol {
 		self.protocol.clone()
 	}
 
@@ -136,12 +133,12 @@ where
 }
 
 /// The actual handler once the connection has been established.
-pub struct CustomProtoHandler<B: BlockT, TSubstream> {
+pub struct CustomProtoHandler<TSubstream> {
 	/// Configuration for the protocol upgrade to negotiate.
-	protocol: RegisteredProtocol<B>,
+	protocol: RegisteredProtocol,
 
 	/// State of the communications with the remote.
-	state: ProtocolState<B, TSubstream>,
+	state: ProtocolState<TSubstream>,
 
 	/// Identifier of the node we're talking to. Used only for logging purposes and shouldn't have
 	/// any influence on the behaviour.
@@ -155,15 +152,15 @@ pub struct CustomProtoHandler<B: BlockT, TSubstream> {
 	///
 	/// This queue must only ever be modified to insert elements at the back, or remove the first
 	/// element.
-	events_queue: SmallVec<[ProtocolsHandlerEvent<RegisteredProtocol<B>, (), CustomProtoHandlerOut<B>>; 16]>,
+	events_queue: SmallVec<[ProtocolsHandlerEvent<RegisteredProtocol, (), CustomProtoHandlerOut>; 16]>,
 }
 
 /// State of the handler.
-enum ProtocolState<B, TSubstream> {
+enum ProtocolState<TSubstream> {
 	/// Waiting for the behaviour to tell the handler whether it is enabled or disabled.
 	Init {
 		/// List of substreams opened by the remote but that haven't been processed yet.
-		substreams: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 6]>,
+		substreams: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 6]>,
 		/// Deadline after which the initialization is abnormally long.
 		init_deadline: Compat<Delay>,
 	},
@@ -179,9 +176,9 @@ enum ProtocolState<B, TSubstream> {
 	/// If we are in this state, we have sent a `CustomProtocolOpen` message to the outside.
 	Normal {
 		/// The substreams where bidirectional communications happen.
-		substreams: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 4]>,
+		substreams: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 4]>,
 		/// Contains substreams which are being shut down.
-		shutdown: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 4]>,
+		shutdown: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 4]>,
 	},
 
 	/// We are disabled. Contains substreams that are being closed.
@@ -189,7 +186,7 @@ enum ProtocolState<B, TSubstream> {
 	/// outside or we have never sent any `CustomProtocolOpen` in the first place.
 	Disabled {
 		/// List of substreams to shut down.
-		shutdown: SmallVec<[RegisteredProtocolSubstream<B, TSubstream>; 6]>,
+		shutdown: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 6]>,
 
 		/// If true, we should reactivate the handler after all the substreams in `shutdown` have
 		/// been closed.
@@ -210,7 +207,7 @@ enum ProtocolState<B, TSubstream> {
 
 /// Event that can be received by a `CustomProtoHandler`.
 #[derive(Debug)]
-pub enum CustomProtoHandlerIn<B: BlockT> {
+pub enum CustomProtoHandlerIn {
 	/// The node should start using custom protocols.
 	Enable,
 
@@ -220,13 +217,13 @@ pub enum CustomProtoHandlerIn<B: BlockT> {
 	/// Sends a message through a custom protocol substream.
 	SendCustomMessage {
 		/// The message to send.
-		message: Message<B>,
+		message: Vec<u8>,
 	},
 }
 
 /// Event that can be emitted by a `CustomProtoHandler`.
 #[derive(Debug)]
-pub enum CustomProtoHandlerOut<B: BlockT> {
+pub enum CustomProtoHandlerOut {
 	/// Opened a custom protocol with the remote.
 	CustomProtocolOpen {
 		/// Version of the protocol that has been opened.
@@ -242,14 +239,14 @@ pub enum CustomProtoHandlerOut<B: BlockT> {
 	/// Receives a message on a custom protocol substream.
 	CustomMessage {
 		/// Message that has been received.
-		message: Message<B>,
+		message: BytesMut,
 	},
 
 	/// A substream to the remote is clogged. The send buffer is very large, and we should print
 	/// a diagnostic message and/or avoid sending more data.
 	Clogged {
 		/// Copy of the messages that are within the buffer, for further diagnostic.
-		messages: Vec<Message<B>>,
+		messages: Vec<Vec<u8>>,
 	},
 
 	/// An error has happened on the protocol level with this node.
@@ -261,10 +258,9 @@ pub enum CustomProtoHandlerOut<B: BlockT> {
 	},
 }
 
-impl<B, TSubstream> CustomProtoHandler<B, TSubstream>
+impl<TSubstream> CustomProtoHandler<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
-	B: BlockT,
 {
 	/// Enables the handler.
 	fn enable(&mut self) {
@@ -342,7 +338,7 @@ where
 	/// Polls the state for events. Optionally returns an event to produce.
 	#[must_use]
 	fn poll_state(&mut self)
-		-> Option<ProtocolsHandlerEvent<RegisteredProtocol<B>, (), CustomProtoHandlerOut<B>>> {
+		-> Option<ProtocolsHandlerEvent<RegisteredProtocol, (), CustomProtoHandlerOut>> {
 		match mem::replace(&mut self.state, ProtocolState::Poisoned) {
 			ProtocolState::Poisoned => {
 				error!(target: "sub-libp2p", "Handler with {:?} is in poisoned state",
@@ -471,7 +467,7 @@ where
 	/// Called by `inject_fully_negotiated_inbound` and `inject_fully_negotiated_outbound`.
 	fn inject_fully_negotiated(
 		&mut self,
-		mut substream: RegisteredProtocolSubstream<B, TSubstream>
+		mut substream: RegisteredProtocolSubstream<TSubstream>
 	) {
 		self.state = match mem::replace(&mut self.state, ProtocolState::Poisoned) {
 			ProtocolState::Poisoned => {
@@ -516,7 +512,7 @@ where
 	}
 
 	/// Sends a message to the remote.
-	fn send_message(&mut self, message: Message<B>) {
+	fn send_message(&mut self, message: Vec<u8>) {
 		match self.state {
 			ProtocolState::Normal { ref mut substreams, .. } =>
 				substreams[0].send_message(message),
@@ -527,14 +523,14 @@ where
 	}
 }
 
-impl<B, TSubstream> ProtocolsHandler for CustomProtoHandler<B, TSubstream>
-where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
-	type InEvent = CustomProtoHandlerIn<B>;
-	type OutEvent = CustomProtoHandlerOut<B>;
+impl<TSubstream> ProtocolsHandler for CustomProtoHandler<TSubstream>
+where TSubstream: AsyncRead + AsyncWrite {
+	type InEvent = CustomProtoHandlerIn;
+	type OutEvent = CustomProtoHandlerOut;
 	type Substream = TSubstream;
 	type Error = ConnectionKillError;
-	type InboundProtocol = RegisteredProtocol<B>;
-	type OutboundProtocol = RegisteredProtocol<B>;
+	type InboundProtocol = RegisteredProtocol;
+	type OutboundProtocol = RegisteredProtocol;
 	type OutboundOpenInfo = ();
 
 	fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol> {
@@ -556,7 +552,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 		self.inject_fully_negotiated(proto);
 	}
 
-	fn inject_event(&mut self, message: CustomProtoHandlerIn<B>) {
+	fn inject_event(&mut self, message: CustomProtoHandlerIn) {
 		match message {
 			CustomProtoHandlerIn::Disable => self.disable(),
 			CustomProtoHandlerIn::Enable => self.enable(),
@@ -613,7 +609,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 	}
 }
 
-impl<B: BlockT, TSubstream> fmt::Debug for CustomProtoHandler<B, TSubstream>
+impl<TSubstream> fmt::Debug for CustomProtoHandler<TSubstream>
 where
 	TSubstream: AsyncRead + AsyncWrite,
 {
@@ -625,9 +621,9 @@ where
 
 /// Given a list of substreams, tries to shut them down. The substreams that have been successfully
 /// shut down are removed from the list.
-fn shutdown_list<B, TSubstream>
-	(list: &mut SmallVec<impl smallvec::Array<Item = RegisteredProtocolSubstream<B, TSubstream>>>)
-where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
+fn shutdown_list<TSubstream>
+	(list: &mut SmallVec<impl smallvec::Array<Item = RegisteredProtocolSubstream<TSubstream>>>)
+where TSubstream: AsyncRead + AsyncWrite {
 	'outer: for n in (0..list.len()).rev() {
 		let mut substream = list.swap_remove(n);
 		loop {

substrate/core/network/src/legacy_proto/tests.rs

@@ -17,6 +17,7 @@
 #![cfg(test)]
 
 use futures::{future, prelude::*, try_ready};
+use codec::{Encode, Decode};
 use libp2p::core::nodes::Substream;
 use libp2p::core::{ConnectedPoint, transport::boxed::Boxed, muxing::StreamMuxerBox};
 use libp2p::swarm::{Swarm, ProtocolsHandler, IntoProtocolsHandler};
@@ -24,9 +25,9 @@ use libp2p::swarm::{PollParameters, NetworkBehaviour, NetworkBehaviourAction};
 use libp2p::{PeerId, Multiaddr, Transport};
 use rand::seq::SliceRandom;
 use std::{io, time::Duration, time::Instant};
-use test_client::runtime::Block;
-use crate::message::generic::Message;
+use crate::message::Message;
 use crate::legacy_proto::{LegacyProto, LegacyProtoOut};
+use test_client::runtime::Block;
 
 /// Builds two nodes that have each other as bootstrap nodes.
 /// This is to be used only for testing, and a panic will happen if something goes wrong.
@@ -101,12 +102,12 @@ fn build_nodes()
 
 /// Wraps around the `CustomBehaviour` network behaviour, and adds hardcoded node addresses to it.
 struct CustomProtoWithAddr {
-	inner: LegacyProto<Block, Substream<StreamMuxerBox>>,
+	inner: LegacyProto<Substream<StreamMuxerBox>>,
 	addrs: Vec<(PeerId, Multiaddr)>,
 }
 
 impl std::ops::Deref for CustomProtoWithAddr {
-	type Target = LegacyProto<Block, Substream<StreamMuxerBox>>;
+	type Target = LegacyProto<Substream<StreamMuxerBox>>;
 
 	fn deref(&self) -> &Self::Target {
 		&self.inner
@@ -121,8 +122,8 @@ impl std::ops::DerefMut for CustomProtoWithAddr {
 
 impl NetworkBehaviour for CustomProtoWithAddr {
 	type ProtocolsHandler =
-		<LegacyProto<Block, Substream<StreamMuxerBox>> as NetworkBehaviour>::ProtocolsHandler;
-	type OutEvent = <LegacyProto<Block, Substream<StreamMuxerBox>> as NetworkBehaviour>::OutEvent;
+		<LegacyProto<Substream<StreamMuxerBox>> as NetworkBehaviour>::ProtocolsHandler;
+	type OutEvent = <LegacyProto<Substream<StreamMuxerBox>> as NetworkBehaviour>::OutEvent;
 
 	fn new_handler(&mut self) -> Self::ProtocolsHandler {
 		self.inner.new_handler()
@@ -209,7 +210,7 @@ fn two_nodes_transfer_lots_of_packets() {
 					for n in 0 .. NUM_PACKETS {
 						service1.send_packet(
 							&peer_id,
-							Message::ChainSpecific(vec![(n % 256) as u8])
+							Message::<Block>::ChainSpecific(vec![(n % 256) as u8]).encode()
 						);
 					}
 				},
@@ -223,11 +224,16 @@ fn two_nodes_transfer_lots_of_packets() {
 		loop {
 			match try_ready!(service2.poll()) {
 				Some(LegacyProtoOut::CustomProtocolOpen { .. }) => {},
-				Some(LegacyProtoOut::CustomMessage { message: Message::ChainSpecific(message), .. }) => {
-					assert_eq!(message.len(), 1);
-					packet_counter += 1;
-					if packet_counter == NUM_PACKETS {
-						return Ok(Async::Ready(()))
+				Some(LegacyProtoOut::CustomMessage { message, .. }) => {
+					match Message::<Block>::decode(&mut &message[..]).unwrap() {
+						Message::<Block>::ChainSpecific(message) => {
+							assert_eq!(message.len(), 1);
+							packet_counter += 1;
+							if packet_counter == NUM_PACKETS {
+								return Ok(Async::Ready(()))
+							}
+						},
+						_ => panic!(),
 					}
 				}
 				_ => panic!(),
@@ -248,7 +254,7 @@ fn basic_two_nodes_requests_in_parallel() {
 		let mut to_send = Vec::new();
 		for _ in 0..200 { // Note: don't make that number too high or the CPU usage will explode.
 			let msg = (0..10).map(|_| rand::random::<u8>()).collect::<Vec<_>>();
-			to_send.push(Message::ChainSpecific(msg));
+			to_send.push(Message::<Block>::ChainSpecific(msg));
 		}
 		to_send
 	};
@@ -263,7 +269,7 @@ fn basic_two_nodes_requests_in_parallel() {
 			match try_ready!(service1.poll()) {
 				Some(LegacyProtoOut::CustomProtocolOpen { peer_id, .. }) => {
 					for msg in to_send.drain(..) {
-						service1.send_packet(&peer_id, msg);
+						service1.send_packet(&peer_id, msg.encode());
 					}
 				},
 				_ => panic!(),
@@ -276,7 +282,7 @@ fn basic_two_nodes_requests_in_parallel() {
 			match try_ready!(service2.poll()) {
 				Some(LegacyProtoOut::CustomProtocolOpen { .. }) => {},
 				Some(LegacyProtoOut::CustomMessage { message, .. }) => {
-					let pos = to_receive.iter().position(|m| *m == message).unwrap();
+					let pos = to_receive.iter().position(|m| m.encode() == message).unwrap();
 					to_receive.remove(pos);
 					if to_receive.is_empty() {
 						return Ok(Async::Ready(()))

substrate/core/network/src/legacy_proto/upgrade.rs

@@ -15,15 +15,11 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::config::ProtocolId;
-use crate::protocol::message::Message;
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use libp2p::core::{Negotiated, Endpoint, UpgradeInfo, InboundUpgrade, OutboundUpgrade, upgrade::ProtocolName};
 use libp2p::tokio_codec::Framed;
-use log::debug;
-use std::{collections::VecDeque, io, marker::PhantomData, vec::IntoIter as VecIntoIter};
+use std::{collections::VecDeque, io, vec::IntoIter as VecIntoIter};
 use futures::{prelude::*, future, stream};
-use codec::{Decode, Encode};
-use sr_primitives::traits::Block as BlockT;
 use tokio_io::{AsyncRead, AsyncWrite};
 use unsigned_varint::codec::UviBytes;
 
@@ -31,7 +27,7 @@ use unsigned_varint::codec::UviBytes;
 ///
 /// Note that "a single protocol" here refers to `par` for example. However
 /// each protocol can have multiple different versions for networking purposes.
-pub struct RegisteredProtocol<B> {
+pub struct RegisteredProtocol {
 	/// Id of the protocol for API purposes.
 	id: ProtocolId,
 	/// Base name of the protocol as advertised on the network.
@@ -40,11 +36,9 @@ pub struct RegisteredProtocol<B> {
 	/// List of protocol versions that we support.
 	/// Ordered in descending order so that the best comes first.
 	supported_versions: Vec<u8>,
-	/// Marker to pin the generic.
-	marker: PhantomData<B>,
 }
 
-impl<B> RegisteredProtocol<B> {
+impl RegisteredProtocol {
 	/// Creates a new `RegisteredProtocol`. The `custom_data` parameter will be
 	/// passed inside the `RegisteredProtocolOutput`.
 	pub fn new(protocol: impl Into<ProtocolId>, versions: &[u8])
@@ -62,24 +56,22 @@ impl<B> RegisteredProtocol<B> {
 				tmp.sort_unstable_by(|a, b| b.cmp(&a));
 				tmp
 			},
-			marker: PhantomData,
 		}
 	}
 }
 
-impl<B> Clone for RegisteredProtocol<B> {
+impl Clone for RegisteredProtocol {
 	fn clone(&self) -> Self {
 		RegisteredProtocol {
 			id: self.id.clone(),
 			base_name: self.base_name.clone(),
 			supported_versions: self.supported_versions.clone(),
-			marker: PhantomData,
 		}
 	}
 }
 
 /// Output of a `RegisteredProtocol` upgrade.
-pub struct RegisteredProtocolSubstream<B, TSubstream> {
+pub struct RegisteredProtocolSubstream<TSubstream> {
 	/// If true, we are in the process of closing the sink.
 	is_closing: bool,
 	/// Whether the local node opened this substream (dialer), or we received this substream from
@@ -96,11 +88,9 @@ pub struct RegisteredProtocolSubstream<B, TSubstream> {
 	/// If true, we have sent a "remote is clogged" event recently and shouldn't send another one
 	/// unless the buffer empties then fills itself again.
 	clogged_fuse: bool,
-	/// Marker to pin the generic.
-	marker: PhantomData<B>,
 }
 
-impl<B, TSubstream> RegisteredProtocolSubstream<B, TSubstream> {
+impl<TSubstream> RegisteredProtocolSubstream<TSubstream> {
 	/// Returns the version of the protocol that was negotiated.
 	pub fn protocol_version(&self) -> u8 {
 		self.protocol_version
@@ -124,33 +114,32 @@ impl<B, TSubstream> RegisteredProtocolSubstream<B, TSubstream> {
 	}
 
 	/// Sends a message to the substream.
-	pub fn send_message(&mut self, data: Message<B>)
-	where B: BlockT {
+	pub fn send_message(&mut self, data: Vec<u8>) {
 		if self.is_closing {
 			return
 		}
 
-		self.send_queue.push_back(data.encode());
+		self.send_queue.push_back(data);
 	}
 }
 
 /// Event produced by the `RegisteredProtocolSubstream`.
 #[derive(Debug, Clone)]
-pub enum RegisteredProtocolEvent<B: BlockT> {
+pub enum RegisteredProtocolEvent {
 	/// Received a message from the remote.
-	Message(Message<B>),
+	Message(BytesMut),
 
 	/// Diagnostic event indicating that the connection is clogged and we should avoid sending too
 	/// many messages to it.
 	Clogged {
 		/// Copy of the messages that are within the buffer, for further diagnostic.
-		messages: Vec<Message<B>>,
+		messages: Vec<Vec<u8>>,
 	},
 }
 
-impl<B, TSubstream> Stream for RegisteredProtocolSubstream<B, TSubstream>
-where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
-	type Item = RegisteredProtocolEvent<B>;
+impl<TSubstream> Stream for RegisteredProtocolSubstream<TSubstream>
+where TSubstream: AsyncRead + AsyncWrite {
+	type Item = RegisteredProtocolEvent;
 	type Error = io::Error;
 
 	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
@@ -179,8 +168,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 				self.clogged_fuse = true;
 				return Ok(Async::Ready(Some(RegisteredProtocolEvent::Clogged {
 					messages: self.send_queue.iter()
-						.map(|m| Decode::decode(&mut &m[..]))
-						.filter_map(Result::ok)
+						.map(|m| m.clone())
 						.collect(),
 				})))
 			}
@@ -199,15 +187,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 		// Note that `inner` is wrapped in a `Fuse`, therefore we can poll it forever.
 		match self.inner.poll()? {
 			Async::Ready(Some(data)) => {
-				let message = <Message<B> as Decode>::decode(&mut &data[..])
-					.map_err(|err| {
-						debug!(
-							target: "sub-libp2p",
-							"Couldn't decode packet sent by the remote: {:?}: {}", data, err.what(),
-						);
-						io::ErrorKind::InvalidData
-					})?;
-				Ok(Async::Ready(Some(RegisteredProtocolEvent::Message(message))))
+				Ok(Async::Ready(Some(RegisteredProtocolEvent::Message(data))))
 			}
 			Async::Ready(None) =>
 				if !self.requires_poll_complete && self.send_queue.is_empty() {
@@ -220,7 +200,7 @@ where TSubstream: AsyncRead + AsyncWrite, B: BlockT {
 	}
 }
 
-impl<B> UpgradeInfo for RegisteredProtocol<B> {
+impl UpgradeInfo for RegisteredProtocol {
 	type Info = RegisteredProtocolName;
 	type InfoIter = VecIntoIter<Self::Info>;
 
@@ -255,10 +235,10 @@ impl ProtocolName for RegisteredProtocolName {
 	}
 }
 
-impl<B, TSubstream> InboundUpgrade<TSubstream> for RegisteredProtocol<B>
+impl<TSubstream> InboundUpgrade<TSubstream> for RegisteredProtocol
 where TSubstream: AsyncRead + AsyncWrite,
 {
-	type Output = RegisteredProtocolSubstream<B, TSubstream>;
+	type Output = RegisteredProtocolSubstream<TSubstream>;
 	type Future = future::FutureResult<Self::Output, io::Error>;
 	type Error = io::Error;
 
@@ -281,12 +261,11 @@ where TSubstream: AsyncRead + AsyncWrite,
 			inner: framed.fuse(),
 			protocol_version: info.version,
 			clogged_fuse: false,
-			marker: PhantomData,
 		})
 	}
 }
 
-impl<B, TSubstream> OutboundUpgrade<TSubstream> for RegisteredProtocol<B>
+impl<TSubstream> OutboundUpgrade<TSubstream> for RegisteredProtocol
 where TSubstream: AsyncRead + AsyncWrite,
 {
 	type Output = <Self as InboundUpgrade<TSubstream>>::Output;
@@ -308,7 +287,6 @@ where TSubstream: AsyncRead + AsyncWrite,
 			inner: framed.fuse(),
 			protocol_version: info.version,
 			clogged_fuse: false,
-			marker: PhantomData,
 		})
 	}
 }

substrate/core/network/src/protocol.rs

@@ -16,6 +16,7 @@
 
 use crate::{DiscoveryNetBehaviour, config::ProtocolId};
 use crate::legacy_proto::{LegacyProto, LegacyProtoOut};
+use bytes::BytesMut;
 use futures::prelude::*;
 use futures03::{StreamExt as _, TryStreamExt as _};
 use libp2p::{Multiaddr, PeerId};
@@ -28,6 +29,7 @@ use consensus::{
 	block_validation::BlockAnnounceValidator,
 	import_queue::{BlockImportResult, BlockImportError, IncomingBlock, Origin}
 };
+use codec::{Decode, Encode};
 use sr_primitives::{generic::BlockId, ConsensusEngineId, Justification};
 use sr_primitives::traits::{
 	Block as BlockT, Header as HeaderT, NumberFor, One, Zero,
@@ -44,6 +46,7 @@ use crate::config::{BoxFinalityProofRequestBuilder, Roles};
 use rustc_hex::ToHex;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
+use std::fmt::Write;
 use std::{cmp, num::NonZeroUsize, time};
 use log::{trace, debug, warn, error};
 use crate::chain::{Client, FinalityProofProvider};
@@ -90,6 +93,8 @@ const PEER_BEHIND_US_LIGHT_REPUTATION_CHANGE: i32 = -(1 << 8);
 const NEW_EXTRINSIC_REPUTATION_CHANGE: i32 = 1 << 7;
 /// We sent an RPC query to the given node, but it failed.
 const RPC_FAILED_REPUTATION_CHANGE: i32 = -(1 << 12);
+/// We received a message that failed to decode.
+const BAD_MESSAGE_REPUTATION_CHANGE: i32 = -(1 << 12);
 
 // Lock must always be taken in order declared here.
 pub struct Protocol<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
@@ -113,7 +118,15 @@ pub struct Protocol<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> {
 	/// When asked for a proof of finality, we use this struct to build one.
 	finality_proof_provider: Option<Arc<dyn FinalityProofProvider<B>>>,
 	/// Handles opening the unique substream and sending and receiving raw messages.
-	behaviour: LegacyProto<B, Substream<StreamMuxerBox>>,
+	behaviour: LegacyProto<Substream<StreamMuxerBox>>,
+}
+
+#[derive(Default)]
+struct PacketStats {
+	bytes_in: u64,
+	bytes_out: u64,
+	count_in: u64,
+	count_out: u64,
 }
 
 /// A peer that we are connected to
@@ -151,12 +164,12 @@ pub struct PeerInfo<B: BlockT> {
 	pub best_number: <B::Header as HeaderT>::Number,
 }
 
-struct LightDispatchIn<'a, B: BlockT> {
-	behaviour: &'a mut LegacyProto<B, Substream<StreamMuxerBox>>,
+struct LightDispatchIn<'a> {
+	behaviour: &'a mut LegacyProto<Substream<StreamMuxerBox>>,
 	peerset: peerset::PeersetHandle,
 }
 
-impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
+impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a> {
 	fn report_peer(&mut self, who: &PeerId, reputation: i32) {
 		self.peerset.report_peer(who.clone(), reputation)
 	}
@@ -166,12 +179,12 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 	}
 
 	fn send_header_request(&mut self, who: &PeerId, id: RequestId, block: <<B as BlockT>::Header as HeaderT>::Number) {
-		let message = message::generic::Message::RemoteHeaderRequest(message::RemoteHeaderRequest {
+		let message: Message<B> = message::generic::Message::RemoteHeaderRequest(message::RemoteHeaderRequest {
 			id,
 			block,
 		});
 
-		self.behaviour.send_packet(who, message)
+		self.behaviour.send_packet(who, message.encode())
 	}
 
 	fn send_read_request(
@@ -181,13 +194,13 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 		block: <B as BlockT>::Hash,
 		keys: Vec<Vec<u8>>,
 	) {
-		let message = message::generic::Message::RemoteReadRequest(message::RemoteReadRequest {
+		let message: Message<B> = message::generic::Message::RemoteReadRequest(message::RemoteReadRequest {
 			id,
 			block,
 			keys,
 		});
 
-		self.behaviour.send_packet(who, message)
+		self.behaviour.send_packet(who, message.encode())
 	}
 
 	fn send_read_child_request(
@@ -198,14 +211,14 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 		storage_key: Vec<u8>,
 		keys: Vec<Vec<u8>>,
 	) {
-		let message = message::generic::Message::RemoteReadChildRequest(message::RemoteReadChildRequest {
+		let message: Message<B> = message::generic::Message::RemoteReadChildRequest(message::RemoteReadChildRequest {
 			id,
 			block,
 			storage_key,
 			keys,
 		});
 
-		self.behaviour.send_packet(who, message)
+		self.behaviour.send_packet(who, message.encode())
 	}
 
 	fn send_call_request(
@@ -216,14 +229,14 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 		method: String,
 		data: Vec<u8>
 	) {
-		let message = message::generic::Message::RemoteCallRequest(message::RemoteCallRequest {
+		let message: Message<B> = message::generic::Message::RemoteCallRequest(message::RemoteCallRequest {
 			id,
 			block,
 			method,
 			data,
 		});
 
-		self.behaviour.send_packet(who, message)
+		self.behaviour.send_packet(who, message.encode())
 	}
 
 	fn send_changes_request(
@@ -237,7 +250,7 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 		storage_key: Option<Vec<u8>>,
 		key: Vec<u8>,
 	) {
-		let message = message::generic::Message::RemoteChangesRequest(message::RemoteChangesRequest {
+		let message: Message<B> = message::generic::Message::RemoteChangesRequest(message::RemoteChangesRequest {
 			id,
 			first,
 			last,
@@ -247,7 +260,7 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 			key,
 		});
 
-		self.behaviour.send_packet(who, message)
+		self.behaviour.send_packet(who, message.encode())
 	}
 
 	fn send_body_request(
@@ -260,7 +273,7 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 		direction: Direction,
 		max: Option<u32>
 	) {
-		let message = message::generic::Message::BlockRequest(message::BlockRequest::<B> {
+		let message: Message<B> = message::generic::Message::BlockRequest(message::BlockRequest::<B> {
 			id,
 			fields,
 			from,
@@ -269,7 +282,7 @@ impl<'a, B: BlockT> LightDispatchNetwork<B> for LightDispatchIn<'a, B> {
 			max,
 		});
 
-		self.behaviour.send_packet(who, message)
+		self.behaviour.send_packet(who, message.encode())
 	}
 }
 
@@ -291,7 +304,7 @@ pub trait Context<B: BlockT> {
 
 /// Protocol context.
 struct ProtocolContext<'a, B: 'a + BlockT, H: 'a + ExHashT> {
-	behaviour: &'a mut LegacyProto<B, Substream<StreamMuxerBox>>,
+	behaviour: &'a mut LegacyProto<Substream<StreamMuxerBox>>,
 	context_data: &'a mut ContextData<B, H>,
 	peerset_handle: &'a peerset::PeersetHandle,
 }
@@ -299,7 +312,7 @@ struct ProtocolContext<'a, B: 'a + BlockT, H: 'a + ExHashT> {
 impl<'a, B: BlockT + 'a, H: 'a + ExHashT> ProtocolContext<'a, B, H> {
 	fn new(
 		context_data: &'a mut ContextData<B, H>,
-		behaviour: &'a mut LegacyProto<B, Substream<StreamMuxerBox>>,
+		behaviour: &'a mut LegacyProto<Substream<StreamMuxerBox>>,
 		peerset_handle: &'a peerset::PeersetHandle,
 	) -> Self {
 		ProtocolContext { context_data, peerset_handle, behaviour }
@@ -316,19 +329,19 @@ impl<'a, B: BlockT + 'a, H: ExHashT + 'a> Context<B> for ProtocolContext<'a, B,
 	}
 
 	fn send_consensus(&mut self, who: PeerId, consensus: ConsensusMessage) {
-		send_message(
+		send_message::<B> (
 			self.behaviour,
-			&mut self.context_data.peers,
-			who,
+			&mut self.context_data.stats,
+			&who,
 			GenericMessage::Consensus(consensus)
 		)
 	}
 
 	fn send_chain_specific(&mut self, who: PeerId, message: Vec<u8>) {
-		send_message(
+		send_message::<B> (
 			self.behaviour,
-			&mut self.context_data.peers,
-			who,
+			&mut self.context_data.stats,
+			&who,
 			GenericMessage::ChainSpecific(message)
 		)
 	}
@@ -338,6 +351,7 @@ impl<'a, B: BlockT + 'a, H: ExHashT + 'a> Context<B> for ProtocolContext<'a, B,
 struct ContextData<B: BlockT, H: ExHashT> {
 	// All connected peers
 	peers: HashMap<PeerId, Peer<B, H>>,
+	stats: HashMap<&'static str, PacketStats>,
 	pub chain: Arc<dyn Client<B>>,
 }
 
@@ -388,6 +402,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			config,
 			context_data: ContextData {
 				peers: HashMap::new(),
+				stats: HashMap::new(),
 				chain,
 			},
 			light_dispatch: LightDispatch::new(checker),
@@ -517,8 +532,22 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 	pub fn on_custom_message(
 		&mut self,
 		who: PeerId,
-		message: Message<B>,
+		data: BytesMut,
 	) -> CustomMessageOutcome<B> {
+
+		let message = match <Message<B> as Decode>::decode(&mut &data[..]) {
+			Ok(message) => message,
+			Err(err) => {
+				debug!(target: "sync", "Couldn't decode packet sent by {}: {:?}: {}", who, data, err.what());
+				self.peerset_handle.report_peer(who.clone(), BAD_MESSAGE_REPUTATION_CHANGE);
+				return CustomMessageOutcome::None;
+			}
+		};
+
+		let mut stats = self.context_data.stats.entry(message.id()).or_default();
+		stats.bytes_in += data.len() as u64;
+		stats.count_in += 1;
+
 		match message {
 			GenericMessage::Status(s) => self.on_status_message(who, s),
 			GenericMessage::BlockRequest(r) => self.on_block_request(who, r),
@@ -581,15 +610,25 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		CustomMessageOutcome::None
 	}
 
-	fn send_message(&mut self, who: PeerId, message: Message<B>) {
-		send_message::<B, H>(
+	fn send_request(&mut self, who: &PeerId, message: Message<B>) {
+		send_request::<B, H>(
 			&mut self.behaviour,
+			&mut self.context_data.stats,
 			&mut self.context_data.peers,
 			who,
 			message,
 		);
 	}
 
+	fn send_message(&mut self, who: &PeerId, message: Message<B>) {
+		send_message::<B>(
+			&mut self.behaviour,
+			&mut self.context_data.stats,
+			who,
+			message,
+		);
+	}
+
 	/// Locks `self` and returns a context plus the `ConsensusGossip` struct.
 	pub fn consensus_gossip_lock<'a>(
 		&'a mut self,
@@ -622,7 +661,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			GossipMessageRecipient::BroadcastNew =>
 				self.consensus_gossip.multicast(&mut context, topic, message, false),
 			GossipMessageRecipient::Peer(who) =>
-				self.send_message(who, GenericMessage::Consensus(message)),
+				self.send_message(&who, GenericMessage::Consensus(message)),
 		}
 	}
 
@@ -745,7 +784,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			blocks: blocks,
 		};
 		trace!(target: "sync", "Sending BlockResponse with {} blocks", response.blocks.len());
-		self.send_message(peer, GenericMessage::BlockResponse(response))
+		self.send_message(&peer, GenericMessage::BlockResponse(response))
 	}
 
 	/// Adjusts the reputation of a node.
@@ -792,7 +831,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 				Ok(sync::OnBlockData::Import(origin, blocks)) =>
 					CustomMessageOutcome::BlockImport(origin, blocks),
 				Ok(sync::OnBlockData::Request(peer, req)) => {
-					self.send_message(peer, GenericMessage::BlockRequest(req));
+					self.send_request(&peer, GenericMessage::BlockRequest(req));
 					CustomMessageOutcome::None
 				}
 				Err(sync::BadPeer(id, repu)) => {
@@ -939,7 +978,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		}, who.clone(), status.roles, status.best_number);
 		match self.sync.new_peer(who.clone(), info) {
 			Ok(None) => (),
-			Ok(Some(req)) => self.send_message(who.clone(), GenericMessage::BlockRequest(req)),
+			Ok(Some(req)) => self.send_request(&who, GenericMessage::BlockRequest(req)),
 			Err(sync::BadPeer(id, repu)) => {
 				self.behaviour.disconnect_peer(&id);
 				self.peerset_handle.report_peer(id, repu)
@@ -1020,7 +1059,12 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 						.push(who.to_base58());
 				}
 				trace!(target: "sync", "Sending {} transactions to {}", to_send.len(), who);
-				self.behaviour.send_packet(who, GenericMessage::Transactions(to_send))
+				send_message::<B> (
+					&mut self.behaviour,
+					&mut self.context_data.stats,
+					&who,
+					GenericMessage::Transactions(to_send)
+				)
 			}
 		}
 
@@ -1061,7 +1105,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			trace!(target: "sync", "Announcing block {:?} to {}", hash, who);
 			let inserted = peer.known_blocks.insert(hash);
 			if inserted || force {
-				let message = GenericMessage::BlockAnnounce(message::BlockAnnounce {
+				let message: Message<B> = GenericMessage::BlockAnnounce(message::BlockAnnounce {
 					header: header.clone(),
 					state: if peer.info.protocol_version >= 4  {
 						if is_best {
@@ -1079,7 +1123,12 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 					},
 				});
 
-				self.behaviour.send_packet(who, message)
+				send_message::<B> (
+					&mut self.behaviour,
+					&mut self.context_data.stats,
+					&who,
+					message,
+				)
 			}
 		}
 	}
@@ -1097,7 +1146,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			chain_status: self.specialization.status(),
 		};
 
-		self.send_message(who, GenericMessage::Status(status))
+		self.send_message(&who, GenericMessage::Status(status))
 	}
 
 	fn on_block_announce(&mut self, who: PeerId, announce: BlockAnnounce<B::Header>) -> CustomMessageOutcome<B> {
@@ -1157,7 +1206,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		match blocks_to_import {
 			Ok(sync::OnBlockData::Import(origin, blocks)) => CustomMessageOutcome::BlockImport(origin, blocks),
 			Ok(sync::OnBlockData::Request(peer, req)) => {
-				self.send_message(peer, GenericMessage::BlockRequest(req));
+				self.send_request(&peer, GenericMessage::BlockRequest(req));
 				CustomMessageOutcome::None
 			}
 			Err(sync::BadPeer(id, repu)) => {
@@ -1226,7 +1275,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 		};
 
 		self.send_message(
-			who,
+			&who,
 			GenericMessage::RemoteCallResponse(message::RemoteCallResponse {
 				id: request.id,
 				proof,
@@ -1269,7 +1318,13 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			match result {
 				Ok((id, req)) => {
 					let msg = GenericMessage::BlockRequest(req);
-					send_message(&mut self.behaviour, &mut self.context_data.peers, id, msg)
+					send_request(
+						&mut self.behaviour,
+						&mut self.context_data.stats,
+						&mut self.context_data.peers,
+						&id,
+						msg
+					)
 				}
 				Err(sync::BadPeer(id, repu)) => {
 					self.behaviour.disconnect_peer(&id);
@@ -1342,7 +1397,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			}
 		};
 		self.send_message(
-			who,
+			&who,
 			GenericMessage::RemoteReadResponse(message::RemoteReadResponse {
 				id: request.id,
 				proof,
@@ -1385,7 +1440,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			}
 		};
 		self.send_message(
-			who,
+			&who,
 			GenericMessage::RemoteReadResponse(message::RemoteReadResponse {
 				id: request.id,
 				proof,
@@ -1425,7 +1480,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			}
 		};
 		self.send_message(
-			who,
+			&who,
 			GenericMessage::RemoteHeaderResponse(message::RemoteHeaderResponse {
 				id: request.id,
 				header,
@@ -1495,7 +1550,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			}
 		};
 		self.send_message(
-			who,
+			&who,
 			GenericMessage::RemoteChangesResponse(message::RemoteChangesResponse {
 				id: request.id,
 				max: proof.max_block,
@@ -1545,7 +1600,7 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			},
 		};
  		self.send_message(
-			who,
+			&who,
 			GenericMessage::FinalityProofResponse(message::FinalityProofResponse {
 				id: 0,
 				block: request.block,
@@ -1582,6 +1637,22 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Protocol<B, S, H> {
 			peerset: self.peerset_handle.clone(),
 		}, peer, response);
 	}
+
+	fn format_stats(&self) -> String {
+		let mut out = String::new();
+		for (id, stats) in &self.context_data.stats {
+			let _ = writeln!(
+				&mut out,
+				"{}: In: {} bytes ({}), Out: {} bytes ({})",
+				id,
+				stats.bytes_in,
+				stats.count_in,
+				stats.bytes_out,
+				stats.count_out,
+			);
+		}
+		out
+	}
 }
 
 /// Outcome of an incoming custom message.
@@ -1593,14 +1664,15 @@ pub enum CustomMessageOutcome<B: BlockT> {
 	None,
 }
 
-fn send_message<B: BlockT, H: ExHashT>(
-	behaviour: &mut LegacyProto<B, Substream<StreamMuxerBox>>,
+fn send_request<B: BlockT, H: ExHashT>(
+	behaviour: &mut LegacyProto<Substream<StreamMuxerBox>>,
+	stats: &mut HashMap<&'static str, PacketStats>,
 	peers: &mut HashMap<PeerId, Peer<B, H>>,
-	who: PeerId,
+	who: &PeerId,
 	mut message: Message<B>,
 ) {
 	if let GenericMessage::BlockRequest(ref mut r) = message {
-		if let Some(ref mut peer) = peers.get_mut(&who) {
+		if let Some(ref mut peer) = peers.get_mut(who) {
 			r.id = peer.next_request_id;
 			peer.next_request_id = peer.next_request_id + 1;
 			if let Some((timestamp, request)) = peer.block_request.take() {
@@ -1610,12 +1682,25 @@ fn send_message<B: BlockT, H: ExHashT>(
 			peer.block_request = Some((time::Instant::now(), r.clone()));
 		}
 	}
-	behaviour.send_packet(&who, message);
+	send_message::<B>(behaviour, stats, who, message)
+}
+
+fn send_message<B: BlockT>(
+	behaviour: &mut LegacyProto<Substream<StreamMuxerBox>>,
+	stats: &mut HashMap<&'static str, PacketStats>,
+	who: &PeerId,
+	message: Message<B>,
+) {
+	let encoded = message.encode();
+	let mut stats = stats.entry(message.id()).or_default();
+	stats.bytes_out += encoded.len() as u64;
+	stats.count_out += 1;
+	behaviour.send_packet(who, encoded);
 }
 
 impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> NetworkBehaviour for
 Protocol<B, S, H> {
-	type ProtocolsHandler = <LegacyProto<B, Substream<StreamMuxerBox>> as NetworkBehaviour>::ProtocolsHandler;
+	type ProtocolsHandler = <LegacyProto<Substream<StreamMuxerBox>> as NetworkBehaviour>::ProtocolsHandler;
 	type OutEvent = CustomMessageOutcome<B>;
 
 	fn new_handler(&mut self) -> Self::ProtocolsHandler {
@@ -1660,13 +1745,30 @@ Protocol<B, S, H> {
 		}
 
 		for (id, r) in self.sync.block_requests() {
-			send_message(&mut self.behaviour, &mut self.context_data.peers, id, GenericMessage::BlockRequest(r))
+			send_request(
+				&mut self.behaviour,
+				&mut self.context_data.stats,
+				&mut self.context_data.peers,
+				&id,
+				GenericMessage::BlockRequest(r)
+			)
 		}
 		for (id, r) in self.sync.justification_requests() {
-			send_message(&mut self.behaviour, &mut self.context_data.peers, id, GenericMessage::BlockRequest(r))
+			send_request(
+				&mut self.behaviour,
+				&mut self.context_data.stats,
+				&mut self.context_data.peers,
+				&id,
+				GenericMessage::BlockRequest(r)
+			)
 		}
 		for (id, r) in self.sync.finality_proof_requests() {
-			send_message(&mut self.behaviour, &mut self.context_data.peers, id, GenericMessage::FinalityProofRequest(r))
+			send_request(
+				&mut self.behaviour,
+				&mut self.context_data.stats,
+				&mut self.context_data.peers,
+				&id,
+				GenericMessage::FinalityProofRequest(r))
 		}
 
 		let event = match self.behaviour.poll(params) {
@@ -1700,8 +1802,9 @@ Protocol<B, S, H> {
 			LegacyProtoOut::Clogged { peer_id, messages } => {
 				debug!(target: "sync", "{} clogging messages:", messages.len());
 				for msg in messages.into_iter().take(5) {
-					debug!(target: "sync", "{:?}", msg);
-					self.on_clogged_peer(peer_id.clone(), Some(msg));
+					let message: Option<Message<B>> = Decode::decode(&mut &msg[..]).ok();
+					debug!(target: "sync", "{:?}", message);
+					self.on_clogged_peer(peer_id.clone(), message);
 				}
 				CustomMessageOutcome::None
 			}
@@ -1749,3 +1852,9 @@ impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> DiscoveryNetBehaviour f
 		self.behaviour.add_discovered_nodes(peer_ids)
 	}
 }
+
+impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Drop for Protocol<B, S, H> {
+	fn drop(&mut self) {
+		debug!(target: "sync", "Network stats:\n{}", self.format_stats());
+	}
+}

substrate/core/network/src/protocol/message.rs

@@ -222,6 +222,32 @@ pub mod generic {
 		ChainSpecific(Vec<u8>),
 	}
 
+	impl<Header, Hash, Number, Extrinsic> Message<Header, Hash, Number, Extrinsic> {
+		/// Message id useful for logging.
+		pub fn id(&self) -> &'static str {
+			match self {
+				Message::Status(_) => "Status",
+				Message::BlockRequest(_) => "BlockRequest",
+				Message::BlockResponse(_) => "BlockResponse",
+				Message::BlockAnnounce(_) => "BlockAnnounce",
+				Message::Transactions(_) => "Transactions",
+				Message::Consensus(_) => "Consensus",
+				Message::RemoteCallRequest(_) => "RemoteCallRequest",
+				Message::RemoteCallResponse(_) => "RemoteCallResponse",
+				Message::RemoteReadRequest(_) => "RemoteReadRequest",
+				Message::RemoteReadResponse(_) => "RemoteReadResponse",
+				Message::RemoteHeaderRequest(_) => "RemoteHeaderRequest",
+				Message::RemoteHeaderResponse(_) => "RemoteHeaderResponse",
+				Message::RemoteChangesRequest(_) => "RemoteChangesRequest",
+				Message::RemoteChangesResponse(_) => "RemoteChangesResponse",
+				Message::RemoteReadChildRequest(_) => "RemoteReadChildRequest",
+				Message::FinalityProofRequest(_) => "FinalityProofRequest",
+				Message::FinalityProofResponse(_) => "FinalityProofResponse",
+				Message::ChainSpecific(_) => "ChainSpecific",
+			}
+		}
+	}
+
 	/// Status sent on connection.
 	#[derive(Debug, PartialEq, Eq, Clone, Encode, Decode)]
 	pub struct Status<Hash, Number> {