Reduce consensus spam

core/finality-grandpa/src/communication.rs

@@ -18,6 +18,8 @@
 //! that sign or re-shape.
 
 use std::collections::HashMap;
+use std::sync::Arc;
+
 use grandpa::VoterSet;
 use futures::prelude::*;
 use futures::sync::mpsc;
@@ -27,8 +29,6 @@ use runtime_primitives::traits::Block as BlockT;
 use tokio::timer::Interval;
 use {Error, Network, Message, SignedMessage, Commit, CompactCommit};
 
-use std::sync::Arc;
-
 fn localized_payload<E: Encode>(round: u64, set_id: u64, message: &E) -> Vec<u8> {
 	(message, round, set_id).encode()
 }
@@ -47,6 +47,8 @@ enum Broadcast<Block: BlockT> {
 	Announcement(Round, SetId, Block::Hash),
 	// round, set id being dropped.
 	DropRound(Round, SetId),
+	// set_id being dropped.
+	DropSet(SetId),
 }
 
 impl<Block: BlockT> Broadcast<Block> {
@@ -56,6 +58,7 @@ impl<Block: BlockT> Broadcast<Block> {
 			Broadcast::Message(_, s, _) => s,
 			Broadcast::Announcement(_, s, _) => s,
 			Broadcast::DropRound(_, s) => s,
+			Broadcast::DropSet(s) => s,
 		}
 	}
 }
@@ -187,7 +190,11 @@ impl<B: BlockT, N: Network<B>> Future for BroadcastWorker<B, N> {
 						Broadcast::DropRound(round, set_id) => {
 							// stop making announcements for any dead rounds.
 							self.announcements.retain(|_, &mut r| r > round);
-							self.network.drop_messages(round.0, set_id.0);
+							self.network.drop_round_messages(round.0, set_id.0);
+						}
+						Broadcast::DropSet(set_id) => {
+							// stop making announcements for any dead rounds.
+							self.network.drop_set_messages(set_id.0);
 						}
 					}
 				}
@@ -207,10 +214,14 @@ impl<B: BlockT, N: Network<B>> Network<B> for BroadcastHandle<B, N> {
 		let _ = self.relay.unbounded_send(Broadcast::Message(Round(round), SetId(set_id), message));
 	}
 
-	fn drop_messages(&self, round: u64, set_id: u64) {
+	fn drop_round_messages(&self, round: u64, set_id: u64) {
 		let _ = self.relay.unbounded_send(Broadcast::DropRound(Round(round), SetId(set_id)));
 	}
 
+	fn drop_set_messages(&self, set_id: u64) {
+		let _ = self.relay.unbounded_send(Broadcast::DropSet(SetId(set_id)));
+	}
+
 	fn commit_messages(&self, set_id: u64) -> Self::In {
 		self.network.commit_messages(set_id)
 	}
@@ -332,7 +343,7 @@ impl<Block: BlockT, N: Network<Block>> Sink for OutgoingMessages<Block, N>
 
 impl<Block: BlockT, N: Network<Block>> Drop for OutgoingMessages<Block, N> {
 	fn drop(&mut self) {
-		self.network.drop_messages(self.round, self.set_id);
+		self.network.drop_round_messages(self.round, self.set_id);
 	}
 }
 
@@ -439,14 +450,14 @@ pub(crate) fn checked_commit_stream<Block: BlockT, S>(
 }
 
 /// An output sink for commit messages.
-pub(crate) struct CommitsOut<Block, N> {
+pub(crate) struct CommitsOut<Block: BlockT, N: Network<Block>> {
 	network: N,
 	set_id: u64,
 	_marker: ::std::marker::PhantomData<Block>,
 	is_voter: bool,
 }
 
-impl<Block, N> CommitsOut<Block, N> {
+impl<Block: BlockT, N: Network<Block>> CommitsOut<Block, N> {
 	/// Create a new commit output stream.
 	pub(crate) fn new(network: N, set_id: u64, is_voter: bool) -> Self {
 		CommitsOut {
@@ -487,3 +498,9 @@ impl<Block: BlockT, N: Network<Block>> Sink for CommitsOut<Block, N> {
 	fn close(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
 	fn poll_complete(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) }
 }
+
+impl<Block: BlockT, N: Network<Block>> Drop for CommitsOut<Block, N> {
+	fn drop(&mut self) {
+		self.network.drop_set_messages(self.set_id);
+	}
+}

core/finality-grandpa/src/lib.rs

@@ -231,7 +231,10 @@ pub trait Network<Block: BlockT>: Clone {
 	fn send_message(&self, round: u64, set_id: u64, message: Vec<u8>);
 
 	/// Clean up messages for a round.
-	fn drop_messages(&self, round: u64, set_id: u64);
+	fn drop_round_messages(&self, round: u64, set_id: u64);
+
+	/// Clean up messages for a given authority set id (e.g. commit messages).
+	fn drop_set_messages(&self, set_id: u64);
 
 	/// Get a stream of commit messages for a specific set-id. This stream
 	/// should never logically conclude.
@@ -283,9 +286,14 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT
 		self.service.gossip_consensus_message(topic, message, false);
 	}
 
-	fn drop_messages(&self, round: u64, set_id: u64) {
+	fn drop_round_messages(&self, round: u64, set_id: u64) {
 		let topic = message_topic::<B>(round, set_id);
-		self.service.consensus_gossip().write().collect_garbage(|t| t == &topic);
+		self.service.consensus_gossip().write().collect_garbage_for_topic(topic);
+	}
+
+	fn drop_set_messages(&self, set_id: u64) {
+		let topic = commit_topic::<B>(set_id);
+		self.service.consensus_gossip().write().collect_garbage_for_topic(topic);
 	}
 
 	fn commit_messages(&self, set_id: u64) -> Self::In {
@@ -294,7 +302,7 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: ExHashT
 
 	fn send_commit(&self, _round: u64, set_id: u64, message: Vec<u8>) {
 		let topic = commit_topic::<B>(set_id);
-		self.service.gossip_consensus_message(topic, message, true);
+		self.service.gossip_consensus_message(topic, message, false);
 	}
 
 	fn announce(&self, round: u64, _set_id: u64, block: B::Hash) {

core/finality-grandpa/src/tests.rs

@@ -145,6 +145,15 @@ impl MessageRouting {
 			peer_id,
 		}
 	}
+
+	fn drop_messages(&self, topic: Hash) {
+		let inner = self.inner.lock();
+		let peer = inner.peer(self.peer_id);
+		let mut gossip = peer.consensus_gossip().write();
+		peer.with_spec(move |_, _| {
+			gossip.collect_garbage_for_topic(topic);
+		});
+	}
 }
 
 fn make_topic(round: u64, set_id: u64) -> Hash {
@@ -199,14 +208,14 @@ impl Network<Block> for MessageRouting {
 		inner.route_until_complete();
 	}
 
-	fn drop_messages(&self, round: u64, set_id: u64) {
+	fn drop_round_messages(&self, round: u64, set_id: u64) {
 		let topic = make_topic(round, set_id);
-		let inner = self.inner.lock();
-		let peer = inner.peer(self.peer_id);
-		let mut gossip = peer.consensus_gossip().write();
-		peer.with_spec(move |_, _| {
-			gossip.collect_garbage(|t| t == &topic)
-		});
+		self.drop_messages(topic);
+	}
+
+	fn drop_set_messages(&self, set_id: u64) {
+		let topic = make_commit_topic(set_id);
+		self.drop_messages(topic);
 	}
 
 	fn commit_messages(&self, set_id: u64) -> Self::In {
@@ -226,7 +235,7 @@ impl Network<Block> for MessageRouting {
 
 	fn send_commit(&self, _round: u64, set_id: u64, message: Vec<u8>) {
 		let mut inner = self.inner.lock();
-		inner.peer(self.peer_id).gossip_message(make_commit_topic(set_id), message, true);
+		inner.peer(self.peer_id).gossip_message(make_commit_topic(set_id), message, false);
 		inner.route_until_complete();
 	}
 

core/network/Cargo.toml

@@ -14,6 +14,7 @@ error-chain = "0.12"
 bitflags = "1.0"
 futures = "0.1.17"
 linked-hash-map = "0.5"
+lru-cache = "0.1.1"
 rustc-hex = "2.0"
 rand = "0.6"
 substrate-primitives = { path = "../../core/primitives" }