Skip to content
This repository was archived by the owner on Nov 15, 2023. It is now read-only.

Some gossip improvements #1892

Merged
gavofyork merged 5 commits into from
Feb 28, 2019
Merged

Some gossip improvements #1892

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
b176b65
queue messages in future
rphmeier Feb 27, 2019
776f79b
use new gossip API in GRANDPA
rphmeier Feb 27, 2019
e2967e3
implement message_expired for grandpa
rphmeier Feb 27, 2019
d3856a9
fix indent
rphmeier Feb 27, 2019
8fa7557
Merge branch 'master' into rh-gossip-improvements
rphmeier Feb 28, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions core/finality-grandpa/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,7 +455,7 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>,> Network<B
self.validator.note_round(round, set_id);
let (tx, rx) = sync::oneshot::channel();
self.service.with_gossip(move |gossip, _| {
let inner_rx = gossip.messages_for(message_topic::<B>(round, set_id));
let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, message_topic::<B>(round, set_id));
let _ = tx.send(inner_rx);
});
NetworkStream { outer: rx, inner: None }
Expand All @@ -480,7 +480,7 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>,> Network<B
self.validator.note_set(set_id);
let (tx, rx) = sync::oneshot::channel();
self.service.with_gossip(move |gossip, _| {
let inner_rx = gossip.messages_for(commit_topic::<B>(set_id));
let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, commit_topic::<B>(set_id));
let _ = tx.send(inner_rx);
});
NetworkStream { outer: rx, inner: None }
Expand Down
10 changes: 8 additions & 2 deletions core/finality-grandpa/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,7 +182,10 @@ impl Network<Block> for MessageRouting {
self.validator.note_round(round, set_id);
let inner = self.inner.lock();
let peer = inner.peer(self.peer_id);
let messages = peer.consensus_gossip_messages_for(make_topic(round, set_id));
let messages = peer.consensus_gossip_messages_for(
GRANDPA_ENGINE_ID,
make_topic(round, set_id),
);

let messages = messages.map_err(
move |_| panic!("Messages for round {} dropped too early", round)
Expand Down Expand Up @@ -212,7 +215,10 @@ impl Network<Block> for MessageRouting {
self.validator.note_set(set_id);
let inner = self.inner.lock();
let peer = inner.peer(self.peer_id);
let messages = peer.consensus_gossip_messages_for(make_commit_topic(set_id));
let messages = peer.consensus_gossip_messages_for(
GRANDPA_ENGINE_ID,
make_commit_topic(set_id),
);

let messages = messages.map_err(
move |_| panic!("Commit messages for set {} dropped too early", set_id)
Expand Down
173 changes: 134 additions & 39 deletions core/network/src/consensus_gossip.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,17 +40,26 @@ struct PeerConsensus<H> {
is_authority: bool,
}

#[derive(Clone, Copy)]
enum Status {
Live,
Future,
}

struct MessageEntry<B: BlockT> {
message_hash: B::Hash,
topic: B::Hash,
message: ConsensusMessage,
timestamp: Instant,
status: Status,
}

/// Message validation result.
pub enum ValidationResult<H> {
/// Message is valid with this topic.
Valid(H),
/// Message is future with this topic.
Future(H),

@andresilva andresilva Feb 28, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not separate this into ValidationResult { Valid, Invalid } and ExpirationResult { Live, Expired, Future }? This would make sure that the interface we provide always starts by checking expiration, and only calls validate once which presumably might do heavier checks.

@andresilva andresilva Feb 28, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With the current usage of validate in GRANDPA it doesn't make a difference since it always starts by doing the expiration checks, but I think it would make the interface cleaner.

@rphmeier rphmeier Feb 28, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(I agree, but could we file a follow-up for that?)

/// Invalid message.
Invalid,
/// Obsolete message.
Expand All @@ -61,12 +70,20 @@ pub enum ValidationResult<H> {
pub trait Validator<H> {
/// Validate consensus message.
fn validate(&self, data: &[u8]) -> ValidationResult<H>;

/// Produce a closure for validating messages on a given topic.
fn message_expired<'a>(&'a self) -> Box<FnMut(H, &[u8]) -> bool + 'a> {

@andresilva andresilva Feb 28, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why return a closure instead of just calling the method directly like we do for validate?

@rphmeier rphmeier Feb 28, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

my intuition was that this can get called with a lot of topic messages at once, so the closure lets us hold locks

Box::new(move |_topic, data| match self.validate(data) {
ValidationResult::Valid(_) | ValidationResult::Future(_) => false,
ValidationResult::Invalid | ValidationResult::Expired => true,
})
}
}

/// Consensus network protocol handler. Manages statements and candidate requests.
pub struct ConsensusGossip<B: BlockT> {
peers: HashMap<NodeIndex, PeerConsensus<B::Hash>>,
live_message_sinks: HashMap<B::Hash, Vec<mpsc::UnboundedSender<Vec<u8>>>>,
live_message_sinks: HashMap<(ConsensusEngineId, B::Hash), Vec<mpsc::UnboundedSender<Vec<u8>>>>,
messages: Vec<MessageEntry<B>>,
known_messages: LruCache<B::Hash, ()>,
validators: HashMap<ConsensusEngineId, Arc<Validator<B::Hash>>>,
Expand Down Expand Up @@ -102,7 +119,9 @@ impl<B: BlockT> ConsensusGossip<B> {
// Send out all known messages to authorities.
let mut known_messages = HashSet::new();
for entry in self.messages.iter() {
if entry.timestamp + MESSAGE_LIFETIME < now { continue };
if entry.timestamp + MESSAGE_LIFETIME < now { continue }
if let Status::Future = entry.status { continue }

known_messages.insert(entry.message_hash);
protocol.send_message(who, Message::Consensus(entry.message.clone()));
}
Expand Down Expand Up @@ -161,18 +180,23 @@ impl<B: BlockT> ConsensusGossip<B> {
}
}

fn register_message<F>(&mut self, message_hash: B::Hash, topic: B::Hash, get_message: F)
fn register_message<F>(
&mut self,
message_hash: B::Hash,
topic: B::Hash,
status: Status,
get_message: F,
)
where F: Fn() -> ConsensusMessage
{
if self.known_messages.insert(message_hash, ()).is_none()
{
if self.known_messages.insert(message_hash, ()).is_none() {
self.messages.push(MessageEntry {
topic,
message_hash,
message: get_message(),
timestamp: Instant::now(),
status,
});

}
}

Expand All @@ -184,6 +208,8 @@ impl<B: BlockT> ConsensusGossip<B> {
/// Prune old or no longer relevant consensus messages. Provide a predicate
/// for pruning, which returns `false` when the items with a given topic should be pruned.
pub fn collect_garbage(&mut self) {
use std::collections::hash_map::Entry;

self.live_message_sinks.retain(|_, sinks| {
sinks.retain(|sink| !sink.is_closed());
!sinks.is_empty()
Expand All @@ -194,15 +220,23 @@ impl<B: BlockT> ConsensusGossip<B> {
let validators = &self.validators;
let now = Instant::now();

self.messages.retain(|entry| {
entry.timestamp + MESSAGE_LIFETIME >= now
&& match validators.get(&entry.message.engine_id)
.map(|v| v.validate(&entry.message.data))
{
Some(ValidationResult::Valid(_)) => true,
_ => false,
}
});
let mut check_fns = HashMap::new();
let mut message_expired = move |entry: &MessageEntry<B>| {
let engine_id = entry.message.engine_id;
let check_fn = match check_fns.entry(engine_id) {
Entry::Occupied(entry) => entry.into_mut(),
Entry::Vacant(vacant) => match validators.get(&engine_id) {
None => return true, // treat all messages with no validator as expired
Some(validator) => vacant.insert(validator.message_expired()),
}
};

(check_fn)(entry.topic, &entry.message.data)
};

self.messages.retain(|entry|
entry.timestamp + MESSAGE_LIFETIME >= now && !message_expired(entry)
);

trace!(target: "gossip", "Cleaned up {} stale messages, {} left ({} known)",
before - self.messages.len(),
Expand All @@ -216,12 +250,46 @@ impl<B: BlockT> ConsensusGossip<B> {
}

/// Get data of valid, incoming messages for a topic (but might have expired meanwhile)
pub fn messages_for(&mut self, topic: B::Hash) -> mpsc::UnboundedReceiver<Vec<u8>> {
pub fn messages_for(&mut self, engine_id: ConsensusEngineId, topic: B::Hash)
-> mpsc::UnboundedReceiver<Vec<u8>>
{
let (tx, rx) = mpsc::unbounded();
for entry in self.messages.iter().filter(|e| e.topic == topic) {
tx.unbounded_send(entry.message.data.clone()).expect("receiver known to be live; qed");

let validator = match self.validators.get(&engine_id) {
None => {
self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);
return rx;
}
Some(v) => v,
};

for entry in self.messages.iter_mut()
.filter(|e| e.topic == topic && e.message.engine_id == engine_id)
{
let live = match entry.status {
Status::Live => true,
Status::Future => match validator.validate(&entry.message.data) {
Comment thread
arkpar marked this conversation as resolved.
ValidationResult::Valid(_) => {
entry.status = Status::Live;
true
}
_ => {
// don't send messages considered to be future still.
// if messages are considered expired they'll be cleaned up when we
// collect garbage.
false
}
}
};

if live {
entry.status = Status::Live;
tx.unbounded_send(entry.message.data.clone())
.expect("receiver known to be live; qed");
}
}
self.live_message_sinks.entry(topic).or_default().push(tx);

self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);

rx
}
Expand All @@ -247,11 +315,13 @@ impl<B: BlockT> ConsensusGossip<B> {
if let Some(ref mut peer) = self.peers.get_mut(&who) {
use std::collections::hash_map::Entry;

let engine_id = message.engine_id;
//validate the message
let topic = match self.validators.get(&message.engine_id)
let (topic, status) = match self.validators.get(&engine_id)
.map(|v| v.validate(&message.data))
{
Some(ValidationResult::Valid(topic)) => topic,
Some(ValidationResult::Valid(topic)) => (topic, Status::Live),
Some(ValidationResult::Future(topic)) => (topic, Status::Future),
Some(ValidationResult::Invalid) => {
trace!(target:"gossip", "Invalid message from {}", who);
protocol.report_peer(
Expand All @@ -275,13 +345,14 @@ impl<B: BlockT> ConsensusGossip<B> {
who,
Severity::Useless(format!("Sent unknown consensus engine id")),
);
trace!(target:"gossip", "Unknown message engine id {:?} from {}", message.engine_id, who);
trace!(target:"gossip", "Unknown message engine id {:?} from {}",
engine_id, who);
return None;
}
};

peer.known_messages.insert(message_hash);
if let Entry::Occupied(mut entry) = self.live_message_sinks.entry(topic) {
if let Entry::Occupied(mut entry) = self.live_message_sinks.entry((engine_id, topic)) {
debug!(target: "gossip", "Pushing consensus message to sinks for {}.", topic);
entry.get_mut().retain(|sink| {
if let Err(e) = sink.unbounded_send(message.data.clone()) {
Expand All @@ -293,7 +364,7 @@ impl<B: BlockT> ConsensusGossip<B> {
entry.remove_entry();
}
}
self.multicast_inner(protocol, message_hash, topic, || message.clone());
self.multicast_inner(protocol, message_hash, topic, status, || message.clone());
Some((topic, message))
} else {
trace!(target:"gossip", "Ignored statement from unregistered peer {}", who);
Expand All @@ -309,20 +380,23 @@ impl<B: BlockT> ConsensusGossip<B> {
message: ConsensusMessage,
) {
let message_hash = HashFor::<B>::hash(&message.data);
self.multicast_inner(protocol, message_hash, topic, || message.clone());
self.multicast_inner(protocol, message_hash, topic, Status::Live, || message.clone());
}

fn multicast_inner<F>(
&mut self,
protocol: &mut Context<B>,
message_hash: B::Hash,
topic: B::Hash,
status: Status,
get_message: F,
)
where F: Fn() -> ConsensusMessage
{
self.register_message(message_hash, topic, &get_message);
self.propagate(protocol, message_hash, get_message);
self.register_message(message_hash, topic, status, &get_message);
if let Status::Live = status {
self.propagate(protocol, message_hash, get_message);
}
}

/// Note new consensus session.
Expand All @@ -335,6 +409,8 @@ impl<B: BlockT> ConsensusGossip<B> {
mod tests {
use runtime_primitives::testing::{H256, Block as RawBlock, ExtrinsicWrapper};
use std::time::Instant;
use futures::Stream;

use super::*;

type Block = RawBlock<ExtrinsicWrapper<u64>>;
Expand All @@ -347,21 +423,21 @@ mod tests {
message_hash: $hash,
message: ConsensusMessage { data: $m, engine_id: [0, 0, 0, 0]},
timestamp: $now,
status: Status::Live,
});
}
}
}

#[test]
fn collects_garbage() {

struct AllowAll;

@arkpar arkpar Feb 27, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Indentation is off now

impl Validator<H256> for AllowAll {
fn validate(&self, _data: &[u8]) -> ValidationResult<H256> {
ValidationResult::Valid(H256::default())
}
}

#[test]
fn collects_garbage() {
struct AllowOne;
impl Validator<H256> for AllowOne {
fn validate(&self, data: &[u8]) -> ValidationResult<H256> {
Expand Down Expand Up @@ -417,14 +493,15 @@ mod tests {
use futures::Stream;

let mut consensus = ConsensusGossip::<Block>::new();
consensus.register_validator([0, 0, 0, 0], Arc::new(AllowAll));

let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };

let message_hash = HashFor::<Block>::hash(&message.data);
let topic = HashFor::<Block>::hash(&[1,2,3]);

consensus.register_message(message_hash, topic, || message.clone());
let stream = consensus.messages_for(topic);
consensus.register_message(message_hash, topic, Status::Live, || message.clone());
let stream = consensus.messages_for([0, 0, 0, 0], topic);

assert_eq!(stream.wait().next(), Some(Ok(message.data)));
}
Expand All @@ -437,29 +514,47 @@ mod tests {
let msg_a = ConsensusMessage { data: vec![1, 2, 3], engine_id: [0, 0, 0, 0] };
let msg_b = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };

consensus.register_message(HashFor::<Block>::hash(&msg_a.data), topic, || msg_a.clone());
consensus.register_message(HashFor::<Block>::hash(&msg_b.data), topic, || msg_b.clone());
consensus.register_message(HashFor::<Block>::hash(&msg_a.data), topic, Status::Live, || msg_a.clone());
consensus.register_message(HashFor::<Block>::hash(&msg_b.data), topic, Status::Live, || msg_b.clone());

assert_eq!(consensus.messages.len(), 2);
}

#[test]
fn can_keep_multiple_subscribers_per_topic() {
use futures::Stream;

let mut consensus = ConsensusGossip::<Block>::new();
consensus.register_validator([0, 0, 0, 0], Arc::new(AllowAll));

let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };

let message_hash = HashFor::<Block>::hash(&message.data);
let topic = HashFor::<Block>::hash(&[1,2,3]);

consensus.register_message(message_hash, topic, || message.clone());
consensus.register_message(message_hash, topic, Status::Live, || message.clone());

let stream1 = consensus.messages_for(topic);
let stream2 = consensus.messages_for(topic);
let stream1 = consensus.messages_for([0, 0, 0, 0], topic);
let stream2 = consensus.messages_for([0, 0, 0, 0], topic);

assert_eq!(stream1.wait().next(), Some(Ok(message.data.clone())));
assert_eq!(stream2.wait().next(), Some(Ok(message.data)));
}

#[test]
fn topics_are_localized_to_engine_id() {
let mut consensus = ConsensusGossip::<Block>::new();
consensus.register_validator([0, 0, 0, 0], Arc::new(AllowAll));

let topic = [1; 32].into();
let msg_a = ConsensusMessage { data: vec![1, 2, 3], engine_id: [0, 0, 0, 0] };
let msg_b = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 1] };

consensus.register_message(HashFor::<Block>::hash(&msg_a.data), topic, Status::Live, || msg_a.clone());
consensus.register_message(HashFor::<Block>::hash(&msg_b.data), topic, Status::Live, || msg_b.clone());

let mut stream = consensus.messages_for([0, 0, 0, 0], topic).wait();

assert_eq!(stream.next(), Some(Ok(vec![1, 2, 3])));
let _ = consensus.live_message_sinks.remove(&([0, 0, 0, 0], topic));
assert_eq!(stream.next(), None);
}
}
Loading