diff --git a/client/network/Cargo.toml b/client/network/Cargo.toml index 39598e2a88755..de4f484535e94 100644 --- a/client/network/Cargo.toml +++ b/client/network/Cargo.toml @@ -16,6 +16,7 @@ targets = ["x86_64-unknown-linux-gnu"] prost-build = "0.6.1" [dependencies] +async-std = { version = "1.6.2", features = ["unstable"] } bitflags = "1.2.0" bs58 = "0.3.1" bytes = "0.5.0" @@ -66,7 +67,6 @@ default-features = false features = ["identify", "kad", "mdns", "mplex", "noise", "ping", "tcp-async-std", "websocket", "yamux"] [dev-dependencies] -async-std = "1.6.2" assert_matches = "1.3" env_logger = "0.7.0" libp2p = { version = "0.23.0", default-features = false, features = ["secio"] } diff --git a/client/network/src/gossip.rs b/client/network/src/gossip.rs new file mode 100644 index 0000000000000..0650e7a2f818b --- /dev/null +++ b/client/network/src/gossip.rs @@ -0,0 +1,245 @@ +// This file is part of Substrate. + +// Copyright (C) 2017-2020 Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! Helper for sending rate-limited gossip messages. +//! +//! # Context +//! +//! The [`NetworkService`] struct provides a way to send notifications to a certain peer through +//! the [`NetworkService::notification_sender`] method. This method is quite low level and isn't +//! expected to be used directly. +//! +//! The [`QueuedSender`] struct provided by this module is built on top of +//! [`NetworkService::notification_sender`] and provides a cleaner way to send notifications. +//! +//! # Behaviour +//! +//! An instance of [`QueuedSender`] is specific to a certain combination of `PeerId` and +//! protocol name. It maintains a buffer of messages waiting to be sent out. The user of this API +//! is able to manipulate that queue, adding or removing obsolete messages. +//! +//! Creating a [`QueuedSender`] also returns a opaque `Future` whose responsibility it to +//! drain that queue and actually send the messages. If the substream with the given combination +//! of peer and protocol is closed, the queue is silently discarded. It is the role of the user +//! to track which peers we are connected to. +//! +//! In normal situations, messages sent through a [`QueuedSender`] will arrive in the same +//! order as they have been sent. +//! It is possible, in the situation of disconnects and reconnects, that messages arrive in a +//! different order. See also https://github.com/paritytech/substrate/issues/6756. +//! However, if multiple instances of [`QueuedSender`] exist for the same peer and protocol, or +//! if some other code uses the [`NetworkService`] to send notifications to this combination or +//! peer and protocol, then the notifications will be interleaved in an unpredictable way. +//! + +use crate::{ExHashT, NetworkService}; + +use async_std::sync::{Condvar, Mutex, MutexGuard}; +use futures::prelude::*; +use libp2p::PeerId; +use sp_runtime::{traits::Block as BlockT, ConsensusEngineId}; +use std::{ + collections::VecDeque, + fmt, + sync::{atomic, Arc}, + time::Duration, +}; + +#[cfg(test)] +mod tests; + +/// Notifications sender for a specific combination of network service, peer, and protocol. +pub struct QueuedSender { + /// Shared between the front and the back task. + shared: Arc>, +} + +impl QueuedSender { + /// Returns a new [`QueuedSender`] containing a queue of message for this specific + /// combination of peer and protocol. + /// + /// In addition to the [`QueuedSender`], also returns a `Future` whose role is to drive + /// the messages sending forward. + pub fn new( + service: Arc>, + peer_id: PeerId, + protocol: ConsensusEngineId, + queue_size_limit: usize, + messages_encode: F + ) -> (Self, impl Future + Send + 'static) + where + M: Send + 'static, + B: BlockT + 'static, + H: ExHashT, + F: Fn(M) -> Vec + Send + 'static, + { + let shared = Arc::new(Shared { + stop_task: atomic::AtomicBool::new(false), + condvar: Condvar::new(), + queue_size_limit, + messages_queue: Mutex::new(VecDeque::with_capacity(queue_size_limit)), + }); + + let task = spawn_task( + service, + peer_id, + protocol, + shared.clone(), + messages_encode + ); + + (QueuedSender { shared }, task) + } + + /// Locks the queue of messages towards this peer. + /// + /// The returned `Future` is expected to be ready quite quickly. + pub async fn lock_queue<'a>(&'a self) -> QueueGuard<'a, M> { + QueueGuard { + messages_queue: self.shared.messages_queue.lock().await, + condvar: &self.shared.condvar, + queue_size_limit: self.shared.queue_size_limit, + } + } + + /// Pushes a message to the queue, or discards it if the queue is full. + /// + /// The returned `Future` is expected to be ready quite quickly. + pub async fn queue_or_discard(&self, message: M) + where + M: Send + 'static + { + self.lock_queue().await.push_or_discard(message); + } +} + +impl fmt::Debug for QueuedSender { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.debug_struct("QueuedSender").finish() + } +} + +impl Drop for QueuedSender { + fn drop(&mut self) { + // The "clean" way to notify the `Condvar` here is normally to first lock the `Mutex`, + // then notify the `Condvar` while the `Mutex` is locked. Unfortunately, the `Mutex` + // being asynchronous, it can't reasonably be locked from within a destructor. + // See also the corresponding code in the background task. + self.shared.stop_task.store(true, atomic::Ordering::Release); + self.shared.condvar.notify_all(); + } +} + +/// Locked queue of messages to the given peer. +/// +/// As long as this struct exists, the background task is asleep and the owner of the [`QueueGuard`] +/// is in total control of the buffer. Messages can only ever be sent out after the [`QueueGuard`] +/// is dropped. +#[must_use] +pub struct QueueGuard<'a, M> { + messages_queue: MutexGuard<'a, VecDeque>, + condvar: &'a Condvar, + /// Same as [`Shared::queue_size_limit`]. + queue_size_limit: usize, +} + +impl<'a, M: Send + 'static> QueueGuard<'a, M> { + /// Pushes a message to the queue, or discards it if the queue is full. + /// + /// The message will only start being sent out after the [`QueueGuard`] is dropped. + pub fn push_or_discard(&mut self, message: M) { + if self.messages_queue.len() < self.queue_size_limit { + self.messages_queue.push_back(message); + } + } + + /// Calls `filter` for each message in the queue, and removes the ones for which `false` is + /// returned. + /// + /// > **Note**: The parameter of `filter` is a `&M` and not a `&mut M` (which would be + /// > better) because the underlying implementation relies on `VecDeque::retain`. + pub fn retain(&mut self, filter: impl FnMut(&M) -> bool) { + self.messages_queue.retain(filter); + } +} + +impl<'a, M> Drop for QueueGuard<'a, M> { + fn drop(&mut self) { + // We notify the `Condvar` in the destructor in order to be able to push multiple + // messages and wake up the background task only once afterwards. + self.condvar.notify_one(); + } +} + +#[derive(Debug)] +struct Shared { + /// Read by the background task after locking `locked`. If true, the task stops. + stop_task: atomic::AtomicBool, + /// Queue of messages waiting to be sent out. + messages_queue: Mutex>, + /// Must be notified every time the content of `locked` changes. + condvar: Condvar, + /// Maximum number of elements in `messages_queue`. + queue_size_limit: usize, +} + +async fn spawn_task Vec>( + service: Arc>, + peer_id: PeerId, + protocol: ConsensusEngineId, + shared: Arc>, + messages_encode: F, +) { + loop { + let next_message = 'next_msg: loop { + let mut queue = shared.messages_queue.lock().await; + + loop { + if shared.stop_task.load(atomic::Ordering::Acquire) { + return; + } + + if let Some(msg) = queue.pop_front() { + break 'next_msg msg; + } + + // It is possible that the destructor of `QueuedSender` sets `stop_task` to + // true and notifies the `Condvar` after the background task loads `stop_task` + // and before it calls `Condvar::wait`. + // See also the corresponding comment in `QueuedSender::drop`. + // For this reason, we use `wait_timeout`. In the worst case scenario, + // `stop_task` will always be checked again after the timeout is reached. + queue = shared.condvar.wait_timeout(queue, Duration::from_secs(10)).await.0; + } + }; + + // Starting from below, we try to send the message. If an error happens when sending, + // the only sane option we have is to silently discard the message. + let sender = match service.notification_sender(peer_id.clone(), protocol) { + Ok(s) => s, + Err(_) => continue, + }; + + let ready = match sender.ready().await { + Ok(r) => r, + Err(_) => continue, + }; + + let _ = ready.send(messages_encode(next_message)); + } +} diff --git a/client/network/src/gossip/tests.rs b/client/network/src/gossip/tests.rs new file mode 100644 index 0000000000000..9b16e057461bf --- /dev/null +++ b/client/network/src/gossip/tests.rs @@ -0,0 +1,201 @@ +// This file is part of Substrate. + +// Copyright (C) 2017-2020 Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +use crate::{config, gossip::QueuedSender, Event, NetworkService, NetworkWorker}; + +use futures::prelude::*; +use sp_runtime::traits::{Block as BlockT, Header as _}; +use std::{sync::Arc, time::Duration}; +use substrate_test_runtime_client::{TestClientBuilder, TestClientBuilderExt as _}; + +type TestNetworkService = NetworkService< + substrate_test_runtime_client::runtime::Block, + substrate_test_runtime_client::runtime::Hash, +>; + +/// Builds a full node to be used for testing. Returns the node service and its associated events +/// stream. +/// +/// > **Note**: We return the events stream in order to not possibly lose events between the +/// > construction of the service and the moment the events stream is grabbed. +fn build_test_full_node(config: config::NetworkConfiguration) + -> (Arc, impl Stream) +{ + let client = Arc::new( + TestClientBuilder::with_default_backend() + .build_with_longest_chain() + .0, + ); + + #[derive(Clone)] + struct PassThroughVerifier(bool); + impl sp_consensus::import_queue::Verifier for PassThroughVerifier { + fn verify( + &mut self, + origin: sp_consensus::BlockOrigin, + header: B::Header, + justification: Option, + body: Option>, + ) -> Result< + ( + sp_consensus::BlockImportParams, + Option)>>, + ), + String, + > { + let maybe_keys = header + .digest() + .log(|l| { + l.try_as_raw(sp_runtime::generic::OpaqueDigestItemId::Consensus(b"aura")) + .or_else(|| { + l.try_as_raw(sp_runtime::generic::OpaqueDigestItemId::Consensus(b"babe")) + }) + }) + .map(|blob| { + vec![( + sp_blockchain::well_known_cache_keys::AUTHORITIES, + blob.to_vec(), + )] + }); + + let mut import = sp_consensus::BlockImportParams::new(origin, header); + import.body = body; + import.finalized = self.0; + import.justification = justification; + import.fork_choice = Some(sp_consensus::ForkChoiceStrategy::LongestChain); + Ok((import, maybe_keys)) + } + } + + let import_queue = Box::new(sp_consensus::import_queue::BasicQueue::new( + PassThroughVerifier(false), + Box::new(client.clone()), + None, + None, + &sp_core::testing::TaskExecutor::new(), + None, + )); + + let worker = NetworkWorker::new(config::Params { + role: config::Role::Full, + executor: None, + network_config: config, + chain: client.clone(), + finality_proof_provider: None, + finality_proof_request_builder: None, + on_demand: None, + transaction_pool: Arc::new(crate::config::EmptyTransactionPool), + protocol_id: config::ProtocolId::from(&b"/test-protocol-name"[..]), + import_queue, + block_announce_validator: Box::new( + sp_consensus::block_validation::DefaultBlockAnnounceValidator, + ), + metrics_registry: None, + }) + .unwrap(); + + let service = worker.service().clone(); + let event_stream = service.event_stream("test"); + + async_std::task::spawn(async move { + futures::pin_mut!(worker); + let _ = worker.await; + }); + + (service, event_stream) +} + +const ENGINE_ID: sp_runtime::ConsensusEngineId = *b"foo\0"; + +/// Builds two nodes and their associated events stream. +/// The nodes are connected together and have the `ENGINE_ID` protocol registered. +fn build_nodes_one_proto() + -> (Arc, impl Stream, Arc, impl Stream) +{ + let listen_addr = config::build_multiaddr![Memory(rand::random::())]; + + let (node1, events_stream1) = build_test_full_node(config::NetworkConfiguration { + notifications_protocols: vec![(ENGINE_ID, From::from(&b"/foo"[..]))], + listen_addresses: vec![listen_addr.clone()], + transport: config::TransportConfig::MemoryOnly, + .. config::NetworkConfiguration::new_local() + }); + + let (node2, events_stream2) = build_test_full_node(config::NetworkConfiguration { + notifications_protocols: vec![(ENGINE_ID, From::from(&b"/foo"[..]))], + listen_addresses: vec![], + reserved_nodes: vec![config::MultiaddrWithPeerId { + multiaddr: listen_addr, + peer_id: node1.local_peer_id().clone(), + }], + transport: config::TransportConfig::MemoryOnly, + .. config::NetworkConfiguration::new_local() + }); + + (node1, events_stream1, node2, events_stream2) +} + +#[test] +fn basic_works() { + const NUM_NOTIFS: usize = 256; + + let (node1, mut events_stream1, node2, mut events_stream2) = build_nodes_one_proto(); + let node2_id = node2.local_peer_id().clone(); + + let receiver = async_std::task::spawn(async move { + let mut received_notifications = 0; + + while received_notifications < NUM_NOTIFS { + match events_stream2.next().await.unwrap() { + Event::NotificationStreamClosed { .. } => panic!(), + Event::NotificationsReceived { messages, .. } => { + for message in messages { + assert_eq!(message.0, ENGINE_ID); + assert_eq!(message.1, &b"message"[..]); + received_notifications += 1; + } + } + _ => {} + }; + + if rand::random::() < 2 { + async_std::task::sleep(Duration::from_millis(rand::random::() % 750)).await; + } + } + }); + + async_std::task::block_on(async move { + let (sender, bg_future) = + QueuedSender::new(node1, node2_id, ENGINE_ID, NUM_NOTIFS, |msg| msg); + async_std::task::spawn(bg_future); + + // Wait for the `NotificationStreamOpened`. + loop { + match events_stream1.next().await.unwrap() { + Event::NotificationStreamOpened { .. } => break, + _ => {} + }; + } + + for _ in 0..NUM_NOTIFS { + sender.queue_or_discard(b"message".to_vec()).await; + } + + receiver.await; + }); +} diff --git a/client/network/src/lib.rs b/client/network/src/lib.rs index 69635fce11f7c..e01b260263566 100644 --- a/client/network/src/lib.rs +++ b/client/network/src/lib.rs @@ -15,6 +15,7 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . + #![warn(unused_extern_crates)] #![warn(missing_docs)] @@ -259,6 +260,7 @@ mod utils; pub mod config; pub mod error; +pub mod gossip; pub mod network_state; pub use service::{NetworkService, NetworkWorker}; diff --git a/client/network/src/service.rs b/client/network/src/service.rs index cc3821a455e9f..d42af16f1d22e 100644 --- a/client/network/src/service.rs +++ b/client/network/src/service.rs @@ -701,6 +701,10 @@ impl NetworkService { /// Notifications should be dropped /// if buffer is full /// + /// + /// See also the [`gossip`](crate::gossip) module for a higher-level way to send + /// notifications. + /// pub fn notification_sender( &self, target: PeerId,