diff --git a/substrate/core/finality-grandpa/primitives/src/lib.rs b/substrate/core/finality-grandpa/primitives/src/lib.rs index 384319a298..a439953899 100644 --- a/substrate/core/finality-grandpa/primitives/src/lib.rs +++ b/substrate/core/finality-grandpa/primitives/src/lib.rs @@ -52,7 +52,7 @@ pub type AuthorityWeight = u64; /// The index of an authority. pub type AuthorityIndex = u64; -/// The identifier of a GRANDPA set. +/// The monotonic identifier of a GRANDPA set of authorities. pub type SetId = u64; /// The round indicator. diff --git a/substrate/core/finality-grandpa/src/communication/gossip.rs b/substrate/core/finality-grandpa/src/communication/gossip.rs index 1f9d78c867..24402c8a02 100644 --- a/substrate/core/finality-grandpa/src/communication/gossip.rs +++ b/substrate/core/finality-grandpa/src/communication/gossip.rs @@ -183,7 +183,13 @@ impl View { const KEEP_RECENT_ROUNDS: usize = 3; -/// Tracks topics we keep messages for. +/// Tracks gossip topics that we are keeping messages for. We keep topics of: +/// +/// - the last `KEEP_RECENT_ROUNDS` complete GRANDPA rounds, +/// +/// - the topic for the current and next round, +/// +/// - and a global topic for commit and catch-up messages. struct KeepTopics { current_set: SetId, rounds: VecDeque<(Round, SetId)>, @@ -256,7 +262,7 @@ fn neighbor_topics(view: &View>) -> Vec { #[derive(Debug, Encode, Decode)] pub(super) enum GossipMessage { /// Grandpa message with round and set info. - VoteOrPrecommit(VoteOrPrecommitMessage), + Vote(VoteMessage), /// Grandpa commit message with round and set info. Commit(FullCommitMessage), /// A neighbor packet. Not repropagated. @@ -273,9 +279,9 @@ impl From>> for GossipMessage { +pub(super) struct VoteMessage { /// The round this message is from. pub(super) round: Round, /// The voter set ID this message is from. @@ -612,7 +618,7 @@ impl Inner { cost::PAST_REJECTION } - fn validate_round_message(&self, who: &PeerId, full: &VoteOrPrecommitMessage) + fn validate_round_message(&self, who: &PeerId, full: &VoteMessage) -> Action { match self.consider_vote(full.round, full.set_id) { @@ -1003,7 +1009,7 @@ impl GossipValidator { let action = { match GossipMessage::::decode(&mut data) { - Ok(GossipMessage::VoteOrPrecommit(ref message)) + Ok(GossipMessage::Vote(ref message)) => self.inner.write().validate_round_message(who, message), Ok(GossipMessage::Commit(ref message)) => self.inner.write().validate_commit_message(who, message), Ok(GossipMessage::Neighbor(update)) => { @@ -1163,7 +1169,7 @@ impl network_gossip::Validator for GossipValidator Ok(GossipMessage::Neighbor(_)) => false, Ok(GossipMessage::CatchUpRequest(_)) => false, Ok(GossipMessage::CatchUp(_)) => false, - Ok(GossipMessage::VoteOrPrecommit(_)) => false, // should not be the case. + Ok(GossipMessage::Vote(_)) => false, // should not be the case. } }) } @@ -1478,7 +1484,7 @@ mod tests { val.note_round(Round(1), |_, _| {}); let inner = val.inner.read(); - let unknown_voter = inner.validate_round_message(&peer, &VoteOrPrecommitMessage { + let unknown_voter = inner.validate_round_message(&peer, &VoteMessage { round: Round(1), set_id: SetId(set_id), message: SignedMessage:: { @@ -1491,7 +1497,7 @@ mod tests { } }); - let bad_sig = inner.validate_round_message(&peer, &VoteOrPrecommitMessage { + let bad_sig = inner.validate_round_message(&peer, &VoteMessage { round: Round(1), set_id: SetId(set_id), message: SignedMessage:: { diff --git a/substrate/core/finality-grandpa/src/communication/mod.rs b/substrate/core/finality-grandpa/src/communication/mod.rs index 652c33c026..ba7bdce336 100644 --- a/substrate/core/finality-grandpa/src/communication/mod.rs +++ b/substrate/core/finality-grandpa/src/communication/mod.rs @@ -48,7 +48,7 @@ use crate::{ }; use crate::environment::HasVoted; use gossip::{ - GossipMessage, FullCatchUpMessage, FullCommitMessage, VoteOrPrecommitMessage, GossipValidator + GossipMessage, FullCatchUpMessage, FullCommitMessage, VoteMessage, GossipValidator }; use fg_primitives::{ AuthorityPair, AuthorityId, AuthoritySignature, SetId as SetIdNumber, RoundNumber, @@ -148,12 +148,21 @@ impl Network for Arc> where type In = NetworkStream; fn messages_for(&self, topic: B::Hash) -> Self::In { + // Given that one can only communicate with the Substrate network via the `NetworkService` via message-passing, + // and given that methods on the network consensus gossip are not exposed but only reachable by passing a + // closure into `with_gossip` on the `NetworkService` this function needs to make use of the `NetworkStream` + // construction. + // + // We create a oneshot channel and pass the sender within a closure to the network. At some point in the future + // the network passes the message channel back through the oneshot channel. But the consumer of this function + // expects a stream, not a stream within a oneshot. This complexity is abstracted within `NetworkStream`, + // waiting for the oneshot to resolve and from there on acting like a normal message channel. let (tx, rx) = oneshot::channel(); self.with_gossip(move |gossip, _| { let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, topic); let _ = tx.send(inner_rx); }); - NetworkStream { outer: rx, inner: None } + NetworkStream::PollingOneshot(rx) } fn register_validator(&self, validator: Arc>) { @@ -202,10 +211,18 @@ impl Network for Arc> where } } -/// A stream used by NetworkBridge in its implementation of Network. -pub struct NetworkStream { - inner: Option>, - outer: oneshot::Receiver> +/// A stream used by NetworkBridge in its implementation of Network. Given a oneshot that eventually returns a channel +/// which eventually returns messages, instead of: +/// +/// 1. polling the oneshot until it returns a message channel +/// +/// 2. polling the message channel for messages +/// +/// `NetworkStream` combines the two steps into one, requiring a consumer to only poll `NetworkStream` to retrieve +/// messages directly. +pub enum NetworkStream { + PollingOneshot(oneshot::Receiver>), + PollingTopicNotifications(mpsc::UnboundedReceiver), } impl Stream for NetworkStream { @@ -213,17 +230,21 @@ impl Stream for NetworkStream { type Error = (); fn poll(&mut self) -> Poll, Self::Error> { - if let Some(ref mut inner) = self.inner { - return inner.poll(); - } - match self.outer.poll() { - Ok(futures::Async::Ready(mut inner)) => { - let poll_result = inner.poll(); - self.inner = Some(inner); - poll_result + match self { + NetworkStream::PollingOneshot(oneshot) => { + match oneshot.poll() { + Ok(futures::Async::Ready(mut stream)) => { + let poll_result = stream.poll(); + *self = NetworkStream::PollingTopicNotifications(stream); + poll_result + }, + Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady), + Err(_) => Err(()) + } + }, + NetworkStream::PollingTopicNotifications(stream) => { + stream.poll() }, - Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady), - Err(_) => Err(()) } } } @@ -275,8 +296,8 @@ impl> NetworkBridge { validator.note_round(Round(round.number), |_, _| {}); for signed in round.votes.iter() { - let message = gossip::GossipMessage::VoteOrPrecommit( - gossip::VoteOrPrecommitMessage:: { + let message = gossip::GossipMessage::Vote( + gossip::VoteMessage:: { message: signed.clone(), round: Round(round.number), set_id: SetId(set_id), @@ -341,7 +362,8 @@ impl> NetworkBridge { ); } - /// Get the round messages for a round in the current set ID. These are signature-checked. + /// Get a stream of signature-checked round messages from the network as well as a sink for round messages to the + /// network all within the current set. pub(crate) fn round_communication( &self, round: Round, @@ -379,7 +401,7 @@ impl> NetworkBridge { }) .and_then(move |msg| { match msg { - GossipMessage::VoteOrPrecommit(msg) => { + GossipMessage::Vote(msg) => { // check signature. if !voters.contains_key(&msg.message.id) { debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id); @@ -707,7 +729,7 @@ impl> Sink for OutgoingMessages id: local_id.clone(), }; - let message = GossipMessage::VoteOrPrecommit(VoteOrPrecommitMessage:: { + let message = GossipMessage::Vote(VoteMessage:: { message: signed.clone(), round: Round(self.round), set_id: SetId(self.set_id),