Unverified Commit 490cbd72 authored by Robert Klotzner's avatar Robert Klotzner Committed by GitHub
Browse files

Some refactoring in network-bridge in the course of dealing with #2177 (#2263)

* More doc fixes.

* Minor refactorings in the process of #2177

By having everything peer set related depend directly on the enum the
code becomes more clear and it is also straight forward to add more
peersets/protocols as the compiler will complain if you forget to
implement parts of it.

* Add peer set infos on startup properly

For feature real_overseer.

+ Fixes from review. Thanks @coriolinus and @ordian

!

* More structure in network-bridge

Some changes, which would have helped me in groking the code faster.

Entry points/public types more to the top. Factored out implementation
in their own files, to clear up the top-level view.

* Get rid of local ProtocolName type definition.

Does not add much at this level.

* Fix tests + import cleanup.

* Make spaces tabs.

* Clarify what correct parameters to send_message are

* Be more less vague in docs of send_message.

* Apply suggestions from code review

Extend copyright on new files to 2021 as well.

Co-authored-by: Andronik Ordian's avatarAndronik Ordian <write@reusable.software>

Co-authored-by: Andronik Ordian's avatarAndronik Ordian <write@reusable.software>
parent 034c08b4
Pipeline #120199 canceled with stages
in 12 minutes and 51 seconds
......@@ -5247,6 +5247,7 @@ dependencies = [
"polkadot-node-primitives",
"polkadot-primitives",
"sc-network",
"strum 0.20.0",
]
[[package]]
......@@ -8415,7 +8416,7 @@ dependencies = [
"lazy_static",
"sp-core",
"sp-runtime",
"strum",
"strum 0.16.0",
]
[[package]]
......@@ -8805,7 +8806,16 @@ version = "0.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6138f8f88a16d90134763314e3fc76fa3ed6a7db4725d6acf9a3ef95a3188d22"
dependencies = [
"strum_macros",
"strum_macros 0.16.0",
]
[[package]]
name = "strum"
version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7318c509b5ba57f18533982607f24070a55d353e90d4cae30c467cdb2ad5ac5c"
dependencies = [
"strum_macros 0.20.1",
]
[[package]]
......@@ -8820,6 +8830,18 @@ dependencies = [
"syn 1.0.58",
]
[[package]]
name = "strum_macros"
version = "0.20.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ee8bc6b87a5112aeeab1f4a9f7ab634fe6cbefc4850006df31267f4cfb9e3149"
dependencies = [
"heck",
"proc-macro2 1.0.24",
"quote 1.0.7",
"syn 1.0.58",
]
[[package]]
name = "substrate-bip39"
version = "0.4.2"
......
// Copyright 2020-2021 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//
use futures::channel::mpsc;
use parity_scale_codec::Decode;
use polkadot_node_network_protocol::{
peer_set::PeerSet, v1 as protocol_v1, PeerId, ReputationChange,
};
use polkadot_primitives::v1::{AuthorityDiscoveryId, BlockNumber};
use polkadot_subsystem::messages::NetworkBridgeMessage;
use polkadot_subsystem::{ActiveLeavesUpdate, FromOverseer, OverseerSignal};
use sc_network::Event as NetworkEvent;
use polkadot_node_network_protocol::ObservedRole;
use super::{WireMessage, LOG_TARGET, MALFORMED_MESSAGE_COST};
/// Internal type combining all actions a `NetworkBridge` might perform.
///
/// Both messages coming from the network (`NetworkEvent`) and messages coming from other
/// subsystems (`FromOverseer`) will be converted to `Action` in `run_network` before being
/// processed.
#[derive(Debug)]
pub(crate) enum Action {
/// Ask network to send a validation message.
SendValidationMessages(Vec<(Vec<PeerId>, protocol_v1::ValidationProtocol)>),
/// Ask network to send a collation message.
SendCollationMessages(Vec<(Vec<PeerId>, protocol_v1::CollationProtocol)>),
/// Ask network to connect to validators.
ConnectToValidators {
validator_ids: Vec<AuthorityDiscoveryId>,
connected: mpsc::Sender<(AuthorityDiscoveryId, PeerId)>,
},
/// Report a peer to the network implementation (decreasing/increasing its reputation).
ReportPeer(PeerId, ReputationChange),
/// A subsystem updates us on the relay chain leaves we consider active.
///
/// Implementation will send `WireMessage::ViewUpdate` message to peers as appropriate to the
/// change.
ActiveLeaves(ActiveLeavesUpdate),
/// A subsystem updates our view on the latest finalized block.
///
/// This information is used for view updates, see also `ActiveLeaves`.
BlockFinalized(BlockNumber),
/// Network tells us about a new peer.
PeerConnected(PeerSet, PeerId, ObservedRole),
/// Network tells us about a peer that left.
PeerDisconnected(PeerSet, PeerId),
/// Messages from the network targeted to other subsystems.
PeerMessages(
PeerId,
Vec<WireMessage<protocol_v1::ValidationProtocol>>,
Vec<WireMessage<protocol_v1::CollationProtocol>>,
),
Abort,
Nop,
}
impl From<polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>>> for Action {
#[tracing::instrument(level = "trace", fields(subsystem = LOG_TARGET))]
fn from(res: polkadot_subsystem::SubsystemResult<FromOverseer<NetworkBridgeMessage>>) -> Self {
match res {
Ok(FromOverseer::Signal(OverseerSignal::ActiveLeaves(active_leaves))) => {
Action::ActiveLeaves(active_leaves)
}
Ok(FromOverseer::Signal(OverseerSignal::BlockFinalized(_hash, number))) => {
Action::BlockFinalized(number)
}
Ok(FromOverseer::Signal(OverseerSignal::Conclude)) => Action::Abort,
Ok(FromOverseer::Communication { msg }) => match msg {
NetworkBridgeMessage::ReportPeer(peer, rep) => Action::ReportPeer(peer, rep),
NetworkBridgeMessage::SendValidationMessage(peers, msg) => {
Action::SendValidationMessages(vec![(peers, msg)])
}
NetworkBridgeMessage::SendCollationMessage(peers, msg) => {
Action::SendCollationMessages(vec![(peers, msg)])
}
NetworkBridgeMessage::SendValidationMessages(msgs) => {
Action::SendValidationMessages(msgs)
}
NetworkBridgeMessage::SendCollationMessages(msgs) => {
Action::SendCollationMessages(msgs)
}
NetworkBridgeMessage::ConnectToValidators {
validator_ids,
connected,
} => Action::ConnectToValidators {
validator_ids,
connected,
},
},
Err(e) => {
tracing::warn!(target: LOG_TARGET, err = ?e, "Shutting down Network Bridge due to error");
Action::Abort
}
}
}
}
impl From<Option<NetworkEvent>> for Action {
#[tracing::instrument(level = "trace", fields(subsystem = LOG_TARGET))]
fn from(event: Option<NetworkEvent>) -> Action {
match event {
None => {
tracing::info!(
target: LOG_TARGET,
"Shutting down Network Bridge: underlying event stream concluded"
);
Action::Abort
}
Some(NetworkEvent::Dht(_))
| Some(NetworkEvent::SyncConnected { .. })
| Some(NetworkEvent::SyncDisconnected { .. }) => Action::Nop,
Some(NetworkEvent::NotificationStreamOpened {
remote,
protocol,
role,
}) => {
let role = role.into();
PeerSet::try_from_protocol_name(&protocol).map_or(Action::Nop, |peer_set| {
Action::PeerConnected(peer_set, remote, role)
})
}
Some(NetworkEvent::NotificationStreamClosed { remote, protocol }) => {
PeerSet::try_from_protocol_name(&protocol).map_or(Action::Nop, |peer_set| {
Action::PeerDisconnected(peer_set, remote)
})
}
Some(NetworkEvent::NotificationsReceived { remote, messages }) => {
let v_messages: Result<Vec<_>, _> = messages
.iter()
.filter(|(protocol, _)| protocol == &PeerSet::Validation.into_protocol_name())
.map(|(_, msg_bytes)| WireMessage::decode(&mut msg_bytes.as_ref()))
.collect();
let v_messages = match v_messages {
Err(_) => return Action::ReportPeer(remote, MALFORMED_MESSAGE_COST),
Ok(v) => v,
};
let c_messages: Result<Vec<_>, _> = messages
.iter()
.filter(|(protocol, _)| protocol == &PeerSet::Collation.into_protocol_name())
.map(|(_, msg_bytes)| WireMessage::decode(&mut msg_bytes.as_ref()))
.collect();
match c_messages {
Err(_) => Action::ReportPeer(remote, MALFORMED_MESSAGE_COST),
Ok(c_messages) => {
if v_messages.is_empty() && c_messages.is_empty() {
Action::Nop
} else {
Action::PeerMessages(remote, v_messages, c_messages)
}
}
}
}
}
}
}
This diff is collapsed.
// Copyright 2020-2021 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
use std::pin::Pin;
use std::sync::Arc;
use futures::future::BoxFuture;
use futures::prelude::*;
use futures::stream::BoxStream;
use parity_scale_codec::Encode;
use sc_network::Event as NetworkEvent;
use super::LOG_TARGET;
use polkadot_node_network_protocol::{peer_set::PeerSet, PeerId, ReputationChange};
use polkadot_primitives::v1::{Block, Hash};
use polkadot_subsystem::{SubsystemError, SubsystemResult};
/// Send a message to the network.
///
/// This function is only used internally by the network-bridge, which is responsible to only send
/// messages that are compatible with the passed peer set, as that is currently not enforced by
/// this function. These are messages of type `WireMessage` parameterized on the matching type.
pub(crate) async fn send_message<M, I>(
net: &mut impl Network,
peers: I,
peer_set: PeerSet,
message: M,
) -> SubsystemResult<()>
where
M: Encode + Clone,
I: IntoIterator<Item = PeerId>,
I::IntoIter: ExactSizeIterator,
{
let mut message_producer = stream::iter({
let peers = peers.into_iter();
let n_peers = peers.len();
let mut message = Some(message.encode());
peers.enumerate().map(move |(i, peer)| {
// optimization: avoid cloning the message for the last peer in the
// list. The message payload can be quite large. If the underlying
// network used `Bytes` this would not be necessary.
let message = if i == n_peers - 1 {
message
.take()
.expect("Only taken in last iteration of loop, never afterwards; qed")
} else {
message
.as_ref()
.expect("Only taken in last iteration of loop, we are not there yet; qed")
.clone()
};
Ok(NetworkAction::WriteNotification(peer, peer_set, message))
})
});
net.action_sink().send_all(&mut message_producer).await
}
/// An action to be carried out by the network.
///
/// This type is used for implementing `Sink` in order to cummunicate asynchronously with the
/// underlying network implementation in the `Network` trait.
#[derive(Debug, PartialEq)]
pub enum NetworkAction {
/// Note a change in reputation for a peer.
ReputationChange(PeerId, ReputationChange),
/// Write a notification to a given peer on the given peer-set.
WriteNotification(PeerId, PeerSet, Vec<u8>),
}
/// An abstraction over networking for the purposes of this subsystem.
///
pub trait Network: Send + 'static {
/// Get a stream of all events occurring on the network. This may include events unrelated
/// to the Polkadot protocol - the user of this function should filter only for events related
/// to the [`VALIDATION_PROTOCOL_NAME`](VALIDATION_PROTOCOL_NAME)
/// or [`COLLATION_PROTOCOL_NAME`](COLLATION_PROTOCOL_NAME)
fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent>;
/// Get access to an underlying sink for all network actions.
fn action_sink<'a>(
&'a mut self,
) -> Pin<Box<dyn Sink<NetworkAction, Error = SubsystemError> + Send + 'a>>;
/// Report a given peer as either beneficial (+) or costly (-) according to the given scalar.
fn report_peer(
&mut self,
who: PeerId,
cost_benefit: ReputationChange,
) -> BoxFuture<SubsystemResult<()>> {
async move {
self.action_sink()
.send(NetworkAction::ReputationChange(who, cost_benefit))
.await
}
.boxed()
}
/// Write a notification to a peer on the given peer-set's protocol.
fn write_notification(
&mut self,
who: PeerId,
peer_set: PeerSet,
message: Vec<u8>,
) -> BoxFuture<SubsystemResult<()>> {
async move {
self.action_sink()
.send(NetworkAction::WriteNotification(who, peer_set, message))
.await
}
.boxed()
}
}
impl Network for Arc<sc_network::NetworkService<Block, Hash>> {
fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent> {
sc_network::NetworkService::event_stream(self, "polkadot-network-bridge").boxed()
}
#[tracing::instrument(level = "trace", skip(self), fields(subsystem = LOG_TARGET))]
fn action_sink<'a>(
&'a mut self,
) -> Pin<Box<dyn Sink<NetworkAction, Error = SubsystemError> + Send + 'a>> {
use futures::task::{Context, Poll};
// wrapper around a NetworkService to make it act like a sink.
struct ActionSink<'b>(&'b sc_network::NetworkService<Block, Hash>);
impl<'b> Sink<NetworkAction> for ActionSink<'b> {
type Error = SubsystemError;
fn poll_ready(self: Pin<&mut Self>, _: &mut Context) -> Poll<SubsystemResult<()>> {
Poll::Ready(Ok(()))
}
fn start_send(self: Pin<&mut Self>, action: NetworkAction) -> SubsystemResult<()> {
match action {
NetworkAction::ReputationChange(peer, cost_benefit) => {
tracing::debug!(
target: LOG_TARGET,
"Changing reputation: {:?} for {}",
cost_benefit,
peer
);
self.0.report_peer(peer, cost_benefit)
}
NetworkAction::WriteNotification(peer, peer_set, message) => self
.0
.write_notification(peer, peer_set.into_protocol_name(), message),
}
Ok(())
}
fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<SubsystemResult<()>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<SubsystemResult<()>> {
Poll::Ready(Ok(()))
}
}
Box::pin(ActionSink(&**self))
}
}
......@@ -28,6 +28,7 @@ use sc_network::multiaddr::{Multiaddr, Protocol};
use sc_authority_discovery::Service as AuthorityDiscoveryService;
use polkadot_node_network_protocol::PeerId;
use polkadot_primitives::v1::{AuthorityDiscoveryId, Block, Hash};
use polkadot_node_network_protocol::peer_set::PeerSet;
const LOG_TARGET: &str = "validator_discovery";
......@@ -276,24 +277,24 @@ impl<N: Network, AD: AuthorityDiscovery> Service<N, AD> {
// ask the network to connect to these nodes and not disconnect
// from them until removed from the set
if let Err(e) = network_service.add_peers_to_reserved_set(
super::COLLATION_PROTOCOL_NAME.into(),
PeerSet::Collation.into_protocol_name(),
multiaddr_to_add.clone(),
).await {
tracing::warn!(target: LOG_TARGET, err = ?e, "AuthorityDiscoveryService returned an invalid multiaddress");
}
if let Err(e) = network_service.add_peers_to_reserved_set(
super::VALIDATION_PROTOCOL_NAME.into(),
PeerSet::Validation.into_protocol_name(),
multiaddr_to_add,
).await {
tracing::warn!(target: LOG_TARGET, err = ?e, "AuthorityDiscoveryService returned an invalid multiaddress");
}
// the addresses are known to be valid
let _ = network_service.remove_peers_from_reserved_set(
super::COLLATION_PROTOCOL_NAME.into(),
PeerSet::Collation.into_protocol_name(),
multiaddr_to_remove.clone()
).await;
let _ = network_service.remove_peers_from_reserved_set(
super::VALIDATION_PROTOCOL_NAME.into(),
PeerSet::Validation.into_protocol_name(),
multiaddr_to_remove
).await;
......
......@@ -11,3 +11,4 @@ polkadot-node-primitives = { path = "../../primitives" }
polkadot-node-jaeger = { path = "../../jaeger" }
parity-scale-codec = { version = "1.3.6", default-features = false, features = ["derive"] }
sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
strum = { version = "0.20", features = ["derive"] }
......@@ -29,6 +29,10 @@ pub use polkadot_node_jaeger::JaegerSpan;
#[doc(hidden)]
pub use std::sync::Arc;
/// Peer-sets and protocols used for parachains.
pub mod peer_set;
/// A unique identifier of a request.
pub type RequestId = u64;
......@@ -47,15 +51,6 @@ impl fmt::Display for WrongVariant {
impl std::error::Error for WrongVariant {}
/// The peer-sets that the network manages. Different subsystems will use different peer-sets.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum PeerSet {
/// The validation peer-set is responsible for all messages related to candidate validation and communication among validators.
Validation,
/// The collation peer-set is used for validator<>collator communication.
Collation,
}
/// The advertised role of a node.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum ObservedRole {
......
// Copyright 2021 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
//! All peersets and protocols used for parachains.
use sc_network::config::{NonDefaultSetConfig, SetConfig};
use std::borrow::Cow;
use strum::{EnumIter, IntoEnumIterator};
/// The peer-sets and thus the protocols which are used for the network.
#[derive(Debug, Clone, Copy, PartialEq, EnumIter)]
pub enum PeerSet {
/// The validation peer-set is responsible for all messages related to candidate validation and communication among validators.
Validation,
/// The collation peer-set is used for validator<>collator communication.
Collation,
}
impl PeerSet {
/// Get `sc_network` peer set configurations for each peerset.
///
/// Those should be used in the network configuration to register the protocols with the
/// network service.
pub fn get_info(self) -> NonDefaultSetConfig {
let protocol = self.into_protocol_name();
match self {
PeerSet::Validation => NonDefaultSetConfig {
notifications_protocol: protocol,
set_config: sc_network::config::SetConfig {
in_peers: 25,
out_peers: 0,
reserved_nodes: Vec::new(),
non_reserved_mode: sc_network::config::NonReservedPeerMode::Accept,
},
},
PeerSet::Collation => NonDefaultSetConfig {
notifications_protocol: protocol,
set_config: SetConfig {
in_peers: 25,
out_peers: 0,
reserved_nodes: Vec::new(),
non_reserved_mode: sc_network::config::NonReservedPeerMode::Accept,
},
},
}
}
/// Get the protocol name associated with each peer set as static str.
pub const fn get_protocol_name_static(self) -> &'static str {
match self {
PeerSet::Validation => "/polkadot/validation/1",
PeerSet::Collation => "/polkadot/collation/1",
}
}
/// Convert a peer set into a protocol name as understood by Substrate.
pub fn into_protocol_name(self) -> Cow<'static, str> {
self.get_protocol_name_static().into()
}
/// Try parsing a protocol name into a peer set.
pub fn try_from_protocol_name(name: &Cow<'static, str>) -> Option<PeerSet> {
match name {
n if n == &PeerSet::Validation.into_protocol_name() => Some(PeerSet::Validation),
n if n == &PeerSet::Collation.into_protocol_name() => Some(PeerSet::Collation),
_ => None,
}
}
}
/// Get `NonDefaultSetConfig`s for all available peer sets.
///
/// Should be used during network configuration (added to [`NetworkConfiguration::extra_sets`])
/// or shortly after startup to register the protocols with the network service.
pub fn peer_sets_info() -> Vec<sc_network::config::NonDefaultSetConfig> {
PeerSet::iter().map(PeerSet::get_info).collect()
}
......@@ -567,7 +567,7 @@ pub fn new_full<RuntimeApi, Executor>(
// Substrate nodes.
config.network.extra_sets.push(grandpa::grandpa_peers_set_config());
#[cfg(feature = "real-overseer")]
config.network.extra_sets.extend(polkadot_network_bridge::peers_sets_info());
config.network.extra_sets.extend(polkadot_network_bridge::peer_sets_info());
let (network, network_status_sinks, system_rpc_tx, network_starter) =
service::build_network(service::BuildNetworkParams {
......
......@@ -32,7 +32,7 @@ Output:
This network bridge sends messages of these types over the network.
```rust
enum ProtocolMessage<M> {
enum WireMessage<M> {
ProtocolMessage(M),
ViewUpdate(View),
}
......@@ -41,8 +41,8 @@ enum ProtocolMessage<M> {
and instantiates this type twice, once using the [`ValidationProtocolV1`][VP1] message type, and once with the [`CollationProtocolV1`][CP1] message type.