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

gossip: futures 03 Receiver #3832

Merged
bkchr merged 12 commits into from
Oct 23, 2019
Merged

gossip: futures 03 Receiver #3832

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
3870231
gossip: futures 03 receiver
kigawas Oct 16, 2019
44f3bc7
fix gossip test
kigawas Oct 17, 2019
02233f9
use tokio 01
kigawas Oct 17, 2019
ab1f19a
add comment
kigawas Oct 17, 2019
29fce83
Update core/finality-grandpa/src/communication/mod.rs
kigawas Oct 21, 2019
65f984f
resolve conflicts
kigawas Oct 21, 2019
a8e5039
fix format
kigawas Oct 21, 2019
5ebb09e
rename
kigawas Oct 21, 2019
9fada80
remove tokio 01 runtime
kigawas Oct 23, 2019
98e324e
Merge branch 'master' of https://github.com/paritytech/substrate into…
kigawas Oct 23, 2019
cdb61f6
minor fix
kigawas Oct 23, 2019
57e4a92
make stable happy
kigawas Oct 23, 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
30 changes: 20 additions & 10 deletions core/finality-grandpa/src/communication/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ use std::sync::Arc;

use futures::prelude::*;
use futures::sync::{oneshot, mpsc};
use futures03::stream::{StreamExt, TryStreamExt};
use grandpa::Message::{Prevote, Precommit, PrimaryPropose};
use grandpa::{voter, voter_set::VoterSet};
use log::{debug, trace};
Expand Down Expand Up @@ -100,7 +101,7 @@ mod benefit {
/// Intended to be a lightweight handle such as an `Arc`.
pub trait Network<Block: BlockT>: Clone + Send + 'static {
/// A stream of input messages for a topic.
type In: Stream<Item=network_gossip::TopicNotification,Error=()>;
type In: Stream<Item = network_gossip::TopicNotification, Error = ()>;

/// Get a stream of messages for a specific gossip topic.
fn messages_for(&self, topic: Block::Hash) -> Self::In;
Expand Down Expand Up @@ -145,7 +146,9 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
S: network::specialization::NetworkSpecialization<B>,
H: network::ExHashT,
{
type In = NetworkStream;
type In = NetworkStream<
Box<dyn Stream<Item = network_gossip::TopicNotification, Error = ()> + Send + 'static>,
>;

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,
Expand All @@ -159,7 +162,11 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
// 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 inner_rx: Box<dyn Stream<Item = _, Error = ()> + Send> = Box::new(gossip
.messages_for(GRANDPA_ENGINE_ID, topic)
.map(|x| Ok(x))
.compat()
Copy link
Contributor

@kianenigma kianenigma Oct 21, 2019

Choose a reason for hiding this comment

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

If I have understood correctly, at some point when the call site of gossip.message_for() also returns a compatible future, this can+should be removed. If this is the case, you can mark #[allow(deprecated)] as TODO: #3099 ... or something similar.

Copy link
Member

@bkchr bkchr Oct 21, 2019

Choose a reason for hiding this comment

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

It should not compile anymore, when message for returns a future03 stream. So, we don't need to have any Todo here.

);
let _ = tx.send(inner_rx);
});
NetworkStream::PollingOneshot(rx)
Expand Down Expand Up @@ -220,13 +227,16 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
///
/// `NetworkStream` combines the two steps into one, requiring a consumer to only poll `NetworkStream` to retrieve
/// messages directly.
pub enum NetworkStream {
PollingOneshot(oneshot::Receiver<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>),
PollingTopicNotifications(mpsc::UnboundedReceiver<network_gossip::TopicNotification>),
pub enum NetworkStream<R> {
PollingOneshot(oneshot::Receiver<R>),
PollingTopicNotifications(R),
}

impl Stream for NetworkStream {
type Item = network_gossip::TopicNotification;
impl<R> Stream for NetworkStream<R>
where
R: Stream<Item = network_gossip::TopicNotification, Error = ()>,
{
type Item = R::Item;
type Error = ();

fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
Expand Down Expand Up @@ -266,11 +276,11 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
service: N,
config: crate::Config,
set_state: crate::environment::SharedVoterSetState<B>,
on_exit: impl Future<Item=(),Error=()> + Clone + Send + 'static,
on_exit: impl Future<Item = (), Error = ()> + Clone + Send + 'static,
catch_up_enabled: bool,
) -> (
Self,
impl futures::Future<Item = (), Error = ()> + Send + 'static,
impl Future<Item = (), Error = ()> + Send + 'static,
) {

let (validator, report_stream) = GossipValidator::new(
Expand Down
30 changes: 15 additions & 15 deletions core/network/src/protocol/consensus_gossip.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ use std::sync::Arc;
use std::iter;
use std::time;
use log::{trace, debug};
use futures::sync::mpsc;
use futures03::channel::mpsc;
use lru_cache::LruCache;
use libp2p::PeerId;
use sr_primitives::traits::{Block as BlockT, Hash, HashFor};
Expand Down Expand Up @@ -608,7 +608,7 @@ impl<B: BlockT> Validator<B> for DiscardAll {
#[cfg(test)]
mod tests {
use sr_primitives::testing::{H256, Block as RawBlock, ExtrinsicWrapper};
use futures::Stream;
use futures03::executor::block_on_stream;

use super::*;

Expand Down Expand Up @@ -670,7 +670,7 @@ mod tests {
let m2 = vec![4, 5, 6];

push_msg!(consensus, prev_hash, m1_hash, m1);
push_msg!(consensus, best_hash, m2_hash, m2.clone());
push_msg!(consensus, best_hash, m2_hash, m2);
consensus.known_messages.insert(m1_hash, ());
consensus.known_messages.insert(m2_hash, ());

Expand All @@ -692,18 +692,16 @@ mod tests {

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

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

let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };
let topic = HashFor::<Block>::hash(&[1,2,3]);

consensus.register_message(topic, message.clone());
let stream = consensus.messages_for([0, 0, 0, 0], topic);
let mut stream = block_on_stream(consensus.messages_for([0, 0, 0, 0], topic));

assert_eq!(stream.wait().next(), Some(Ok(TopicNotification { message: message.data, sender: None })));
assert_eq!(stream.next(), Some(TopicNotification { message: message.data, sender: None }));
}

#[test]
Expand All @@ -725,16 +723,17 @@ mod tests {
let mut consensus = ConsensusGossip::<Block>::new();
consensus.register_validator_internal([0, 0, 0, 0], Arc::new(AllowAll));

let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };
let topic = HashFor::<Block>::hash(&[1,2,3]);
let data = vec![4, 5, 6];
let message = ConsensusMessage { data: data.clone(), engine_id: [0, 0, 0, 0] };
let topic = HashFor::<Block>::hash(&[1, 2, 3]);

consensus.register_message(topic, message.clone());

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

assert_eq!(stream1.wait().next(), Some(Ok(TopicNotification { message: message.data.clone(), sender: None })));
assert_eq!(stream2.wait().next(), Some(Ok(TopicNotification { message: message.data, sender: None })));
assert_eq!(stream1.next(), Some(TopicNotification { message: data.clone(), sender: None }));
assert_eq!(stream2.next(), Some(TopicNotification { message: data, sender: None }));
}

#[test]
Expand All @@ -749,9 +748,10 @@ mod tests {
consensus.register_message(topic, msg_a);
consensus.register_message(topic, msg_b);

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

assert_eq!(stream.next(), Some(TopicNotification { message: vec![1, 2, 3], sender: None }));

assert_eq!(stream.next(), Some(Ok(TopicNotification { message: vec![1, 2, 3], sender: None })));
let _ = consensus.live_message_sinks.remove(&([0, 0, 0, 0], topic));
assert_eq!(stream.next(), None);
}
Expand Down