route consensus statements to all peers

polkadot/network/src/consensus.rs

@@ -176,6 +176,7 @@ impl<P: LocalPolkadotApi + Send + Sync + 'static> MessageProcessTask<P> {
 				}
 			}
 			ConsensusMessage::ChainSpecific(msg, _) => {
+				debug!(target: "consensus", "Processing consensus statement for live consensus");
 				if let Some(Message::Statement(parent_hash, statement)) = Decode::decode(&mut msg.as_slice()) {
 					if ::polkadot_consensus::check_statement(&statement.statement, &statement.signature, statement.sender, &parent_hash) {
 						self.table_router.import_statement(statement);

polkadot/network/src/lib.rs

@@ -307,8 +307,8 @@ impl PolkadotProtocol {
 		}
 	}
 
-	/// Send a statement to a validator.
-	fn send_statement(&mut self, ctx: &mut Context<Block>, _val: SessionKey, parent_hash: Hash, statement: SignedStatement) {
+	/// Gossip a consensus statement.
+	fn gossip_statement(&mut self, ctx: &mut Context<Block>, parent_hash: Hash, statement: SignedStatement) {
 		// TODO: something more targeted than gossip.
 		let raw = Message::Statement(parent_hash, statement).encode();
 		self.consensus_gossip.multicast_chain_specific(ctx, raw, parent_hash);

polkadot/network/src/router.rs

@@ -25,7 +25,7 @@
 use polkadot_api::{PolkadotApi, LocalPolkadotApi};
 use polkadot_consensus::{SharedTable, TableRouter, SignedStatement, GenericStatement, StatementProducer};
 use polkadot_primitives::{Hash, BlockId, SessionKey};
-use polkadot_primitives::parachain::{BlockData, Extrinsic, CandidateReceipt, Id as ParaId};
+use polkadot_primitives::parachain::{BlockData, Extrinsic, CandidateReceipt};
 
 use futures::prelude::*;
 use tokio::runtime::TaskExecutor;
@@ -89,14 +89,16 @@ impl<P: PolkadotApi> Clone for Router<P> {
 impl<P: LocalPolkadotApi + Send + Sync + 'static> Router<P> {
 	/// Import a statement whose signature has been checked already.
 	pub(crate) fn import_statement(&self, statement: SignedStatement) {
+		trace!(target: "p_net", "importing consensus statement {:?}", statement.statement);
+
 		// defer any statements for which we haven't imported the candidate yet
-		let (c_hash, parachain_index) = {
+		let c_hash = {
 			let candidate_data = match statement.statement {
-				GenericStatement::Candidate(ref c) => Some((c.hash(), c.parachain_index)),
+				GenericStatement::Candidate(ref c) => Some(c.hash()),
 				GenericStatement::Valid(ref hash)
 					| GenericStatement::Invalid(ref hash)
 					| GenericStatement::Available(ref hash)
-					=> self.table.with_candidate(hash, |c| c.map(|c| (*hash, c.parachain_index))),
+					=> self.table.with_candidate(hash, |c| c.map(|_| *hash)),
 			};
 			match candidate_data {
 				Some(x) => x,
@@ -128,11 +130,11 @@ impl<P: LocalPolkadotApi + Send + Sync + 'static> Router<P> {
 			};
 
 			self.knowledge.lock().note_statement(statement.sender, &statement.statement);
-			self.dispatch_work(c_hash, producer, parachain_index);
+			self.dispatch_work(c_hash, producer);
 		}
 	}
 
-	fn dispatch_work<D, E>(&self, candidate_hash: Hash, producer: StatementProducer<D, E>, parachain: ParaId) where
+	fn dispatch_work<D, E>(&self, candidate_hash: Hash, producer: StatementProducer<D, E>) where
 		D: Future<Item=BlockData,Error=()> + Send + 'static,
 		E: Future<Item=Extrinsic,Error=()> + Send + 'static,
 	{
@@ -161,12 +163,12 @@ impl<P: LocalPolkadotApi + Send + Sync + 'static> Router<P> {
 			// propagate the statements
 			if let Some(validity) = produced.validity {
 				let signed = table.sign_and_import(validity.clone());
-				route_statement(&*network, &*table, parachain, parent_hash, signed);
+				network.with_spec(|spec, ctx| spec.gossip_statement(ctx, parent_hash, signed));
 			}
 
 			if let Some(availability) = produced.availability {
 				let signed = table.sign_and_import(availability);
-				route_statement(&*network, &*table, parachain, parent_hash, signed);
+				network.with_spec(|spec, ctx| spec.gossip_statement(ctx, parent_hash, signed));
 			}
 		});
 
@@ -182,11 +184,10 @@ impl<P: LocalPolkadotApi + Send> TableRouter for Router<P> {
 	fn local_candidate(&self, receipt: CandidateReceipt, block_data: BlockData, extrinsic: Extrinsic) {
 		// give to network to make available.
 		let hash = receipt.hash();
-		let para_id = receipt.parachain_index;
 		let signed = self.table.sign_and_import(GenericStatement::Candidate(receipt));
 
 		self.knowledge.lock().note_candidate(hash, Some(block_data), Some(extrinsic));
-		route_statement(&*self.network, &*self.table, para_id, self.parent_hash, signed);
+		self.network.with_spec(|spec, ctx| spec.gossip_statement(ctx, self.parent_hash, signed));
 	}
 
 	fn fetch_block_data(&self, candidate: &CandidateReceipt) -> BlockDataReceiver {
@@ -217,32 +218,6 @@ impl Future for BlockDataReceiver {
 	}
 }
 
-// get statement to relevant validators.
-fn route_statement(network: &NetworkService, table: &SharedTable, para_id: ParaId, parent_hash: Hash, statement: SignedStatement) {
-	let broadcast = |i: &mut Iterator<Item=&SessionKey>| {
-		let local_key = table.session_key();
-		network.with_spec(|spec, ctx| {
-			for val in i.filter(|&x| x != &local_key) {
-				spec.send_statement(ctx, *val, parent_hash, statement.clone());
-			}
-		});
-	};
-
-	let g_info = table
-		.group_info()
-		.get(&para_id)
-		.expect("statements only produced about groups which exist");
-
-	match statement.statement {
-		GenericStatement::Candidate(_) =>
-			broadcast(&mut g_info.validity_guarantors.iter().chain(g_info.availability_guarantors.iter())),
-		GenericStatement::Valid(_) | GenericStatement::Invalid(_) =>
-			broadcast(&mut g_info.validity_guarantors.iter()),
-		GenericStatement::Available(_) =>
-			broadcast(&mut g_info.availability_guarantors.iter()),
-	}
-}
-
 // A unique trace for valid statements issued by a validator.
 #[derive(Hash, PartialEq, Eq, Clone, Debug)]
 enum StatementTrace {

substrate/network/src/consensus_gossip.rs

@@ -124,7 +124,9 @@ impl<B: BlockT> ConsensusGossip<B> where B::Header: HeaderT<Number=u64> {
 
 	/// Handles incoming chain-specific message and repropagates
 	pub fn on_chain_specific(&mut self, protocol: &mut Context<B>, peer_id: PeerId, message: Vec<u8>, parent_hash: B::Hash) {
+		debug!(target: "gossip", "received chain-specific gossip message");
 		if let Some((hash, message)) = self.handle_incoming(protocol, peer_id, ConsensusMessage::ChainSpecific(message, parent_hash)) {
+			debug!(target: "gossip", "handled incoming chain-specific message");
 			// propagate to other peers.
 			self.multicast(protocol, message, Some(hash));
 		}
@@ -245,6 +247,7 @@ impl<B: BlockT> ConsensusGossip<B> where B::Header: HeaderT<Number=u64> {
 			peer.known_messages.insert(hash);
 			if let Some((sink, parent_hash)) = self.message_sink.take() {
 				if parent == parent_hash {
+					debug!(target: "gossip", "Pushing relevant consensus message to sink.");
 					if let Err(e) = sink.unbounded_send(message.clone()) {
 						trace!(target:"gossip", "Error broadcasting message notification: {:?}", e);
 					}