pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-04-26 23:57:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move block announcement protocol config out of Protocol (#12441)

Browse Source 
* Move Role(s) to `sc-network-common`

* Introduce `NotificationHandshake` type

* Move block announce protocol config creation to `ChainSync`

* Include block announcement into `notification_protocols`

* Apply review comments

* Remove unneeded include

* Add missing include

* Apply review comments
This commit is contained in:
Aaro Altonen
2022-10-10 10:10:53 +03:00
committed by GitHub
parent 0b77060986
commit ce9ce49bc6
23 changed files with 439 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/client/network/src/protocol.rs
+35 -116
View File
@@ -31,10 +31,7 @@ use libp2p::{
Multiaddr, PeerId,
};
use log::{debug, error, info, log, trace, warn, Level};
use message::{
generic::{Message as GenericMessage, Roles},
Message,
};
use message::{generic::Message as GenericMessage, Message};
use notifications::{Notifications, NotificationsOut};
use prometheus_endpoint::{register, Gauge, GaugeVec, Opts, PrometheusError, Registry, U64};
use sc_client_api::HeaderBackend;
@@ -42,13 +39,14 @@ use sc_consensus::import_queue::{
BlockImportError, BlockImportStatus, IncomingBlock, RuntimeOrigin,
};
use sc_network_common::{
config::{NonReservedPeerMode, ProtocolId},
config::NonReservedPeerMode,
error,
protocol::ProtocolName,
protocol::{role::Roles, ProtocolName},
request_responses::RequestFailure,
sync::{
message::{
BlockAnnounce, BlockAttributes, BlockData, BlockRequest, BlockResponse, BlockState,
BlockAnnounce, BlockAnnouncesHandshake, BlockAttributes, BlockData, BlockRequest,
BlockResponse, BlockState,
},
warp::{EncodedProof, WarpProofRequest},
BadPeer, ChainSync, OnBlockData, OnBlockJustification, OnStateData, OpaqueBlockRequest,
@@ -85,8 +83,6 @@ const TICK_TIMEOUT: time::Duration = time::Duration::from_millis(1100);
/// Maximum number of known block hashes to keep for a peer.
const MAX_KNOWN_BLOCKS: usize = 1024; // ~32kb per peer + LruHashSet overhead
/// Maximum allowed size for a block announce.
const MAX_BLOCK_ANNOUNCE_SIZE: u64 = 1024 * 1024;
/// Maximum size used for notifications in the block announce and transaction protocols.
// Must be equal to `max(MAX_BLOCK_ANNOUNCE_SIZE, MAX_TRANSACTIONS_SIZE)`.
@@ -235,30 +231,6 @@ pub struct PeerInfo<B: BlockT> {
pub best_number: <B::Header as HeaderT>::Number,
}
/// Handshake sent when we open a block announces substream.
#[derive(Debug, PartialEq, Eq, Clone, Encode, Decode)]
struct BlockAnnouncesHandshake<B: BlockT> {
/// Roles of the node.
roles: Roles,
/// Best block number.
best_number: NumberFor<B>,
/// Best block hash.
best_hash: B::Hash,
/// Genesis block hash.
genesis_hash: B::Hash,
}
impl<B: BlockT> BlockAnnouncesHandshake<B> {
fn build(
roles: Roles,
best_number: NumberFor<B>,
best_hash: B::Hash,
genesis_hash: B::Hash,
) -> Self {
Self { genesis_hash, roles, best_number, best_hash }
}
}
impl<B, Client> Protocol<B, Client>
where
B: BlockT,
@@ -268,12 +240,10 @@ where
pub fn new(
roles: Roles,
chain: Arc<Client>,
protocol_id: ProtocolId,
fork_id: &Option<String>,
network_config: &config::NetworkConfiguration,
notifications_protocols_handshakes: Vec<Vec<u8>>,
metrics_registry: Option<&Registry>,
chain_sync: Box<dyn ChainSync<B>>,
block_announces_protocol: sc_network_common::config::NonDefaultSetConfig,
) -> error::Result<(Self, sc_peerset::PeersetHandle, Vec<(PeerId, Multiaddr)>)> {
let info = chain.info();
@@ -365,51 +335,24 @@ where
sc_peerset::Peerset::from_config(sc_peerset::PeersetConfig { sets })
};
let block_announces_protocol = {
let genesis_hash =
chain.hash(0u32.into()).ok().flatten().expect("Genesis block exists; qed");
let genesis_hash = genesis_hash.as_ref();
if let Some(fork_id) = fork_id {
format!(
"/{}/{}/block-announces/1",
array_bytes::bytes2hex("", genesis_hash),
fork_id
)
} else {
format!("/{}/block-announces/1", array_bytes::bytes2hex("", genesis_hash))
}
};
let legacy_ba_protocol_name = format!("/{}/block-announces/1", protocol_id.as_ref());
let behaviour = {
let best_number = info.best_number;
let best_hash = info.best_hash;
let genesis_hash = info.genesis_hash;
let block_announces_handshake =
BlockAnnouncesHandshake::<B>::build(roles, best_number, best_hash, genesis_hash)
.encode();
let sync_protocol_config = notifications::ProtocolConfig {
name: block_announces_protocol.into(),
fallback_names: iter::once(legacy_ba_protocol_name.into()).collect(),
handshake: block_announces_handshake,
max_notification_size: MAX_BLOCK_ANNOUNCE_SIZE,
};
Notifications::new(
peerset,
iter::once(sync_protocol_config).chain(
network_config.extra_sets.iter().zip(notifications_protocols_handshakes).map(
|(s, hs)| notifications::ProtocolConfig {
name: s.notifications_protocol.clone(),
fallback_names: s.fallback_names.clone(),
handshake: hs,
max_notification_size: s.max_notification_size,
},
),
),
// NOTE: Block announcement protocol is still very much hardcoded into `Protocol`.
// This protocol must be the first notification protocol given to
// `Notifications`
iter::once(notifications::ProtocolConfig {
name: block_announces_protocol.notifications_protocol.clone(),
fallback_names: block_announces_protocol.fallback_names.clone(),
handshake: block_announces_protocol.handshake.as_ref().unwrap().to_vec(),
max_notification_size: block_announces_protocol.max_notification_size,
})
.chain(network_config.extra_sets.iter().map(|s| notifications::ProtocolConfig {
name: s.notifications_protocol.clone(),
fallback_names: s.fallback_names.clone(),
handshake: s.handshake.as_ref().map_or(roles.encode(), |h| (*h).to_vec()),
max_notification_size: s.max_notification_size,
})),
)
};
@@ -437,10 +380,8 @@ where
},
peerset_handle: peerset_handle.clone(),
behaviour,
notification_protocols: network_config
.extra_sets
.iter()
.map(|s| s.notifications_protocol.clone())
notification_protocols: iter::once(block_announces_protocol.notifications_protocol)
.chain(network_config.extra_sets.iter().map(|s| s.notifications_protocol.clone()))
.collect(),
bad_handshake_substreams: Default::default(),
metrics: if let Some(r) = metrics_registry {
@@ -469,10 +410,7 @@ where
pub fn disconnect_peer(&mut self, peer_id: &PeerId, protocol_name: ProtocolName) {
if let Some(position) = self.notification_protocols.iter().position(|p| *p == protocol_name)
{
self.behaviour.disconnect_peer(
peer_id,
sc_peerset::SetId::from(position + NUM_HARDCODED_PEERSETS),
);
self.behaviour.disconnect_peer(peer_id, sc_peerset::SetId::from(position));
} else {
warn!(target: "sub-libp2p", "disconnect_peer() with invalid protocol name")
}
@@ -1095,8 +1033,7 @@ where
/// Sets the list of reserved peers for the given protocol/peerset.
pub fn set_reserved_peerset_peers(&self, protocol: ProtocolName, peers: HashSet<PeerId>) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle
.set_reserved_peers(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peers);
self.peerset_handle.set_reserved_peers(sc_peerset::SetId::from(index), peers);
} else {
error!(
target: "sub-libp2p",
@@ -1109,10 +1046,7 @@ where
/// Removes a `PeerId` from the list of reserved peers.
pub fn remove_set_reserved_peer(&self, protocol: ProtocolName, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle.remove_reserved_peer(
sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS),
peer,
);
self.peerset_handle.remove_reserved_peer(sc_peerset::SetId::from(index), peer);
} else {
error!(
target: "sub-libp2p",
@@ -1125,8 +1059,7 @@ where
/// Adds a `PeerId` to the list of reserved peers.
pub fn add_set_reserved_peer(&self, protocol: ProtocolName, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle
.add_reserved_peer(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
self.peerset_handle.add_reserved_peer(sc_peerset::SetId::from(index), peer);
} else {
error!(
target: "sub-libp2p",
@@ -1148,8 +1081,7 @@ where
/// Add a peer to a peers set.
pub fn add_to_peers_set(&self, protocol: ProtocolName, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle
.add_to_peers_set(sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS), peer);
self.peerset_handle.add_to_peers_set(sc_peerset::SetId::from(index), peer);
} else {
error!(
target: "sub-libp2p",
@@ -1162,10 +1094,7 @@ where
/// Remove a peer from a peers set.
pub fn remove_from_peers_set(&self, protocol: ProtocolName, peer: PeerId) {
if let Some(index) = self.notification_protocols.iter().position(|p| *p == protocol) {
self.peerset_handle.remove_from_peers_set(
sc_peerset::SetId::from(index + NUM_HARDCODED_PEERSETS),
peer,
);
self.peerset_handle.remove_from_peers_set(sc_peerset::SetId::from(index), peer);
} else {
error!(
target: "sub-libp2p",
@@ -1627,14 +1556,12 @@ where
}
} else {
match (
message::Roles::decode_all(&mut &received_handshake[..]),
Roles::decode_all(&mut &received_handshake[..]),
self.peers.get(&peer_id),
) {
(Ok(roles), _) => CustomMessageOutcome::NotificationStreamOpened {
remote: peer_id,
protocol: self.notification_protocols
[usize::from(set_id) - NUM_HARDCODED_PEERSETS]
.clone(),
protocol: self.notification_protocols[usize::from(set_id)].clone(),
negotiated_fallback,
roles,
notifications_sink,
@@ -1646,9 +1573,7 @@ where
// TODO: remove this after https://github.com/paritytech/substrate/issues/5685
CustomMessageOutcome::NotificationStreamOpened {
remote: peer_id,
protocol: self.notification_protocols
[usize::from(set_id) - NUM_HARDCODED_PEERSETS]
.clone(),
protocol: self.notification_protocols[usize::from(set_id)].clone(),
negotiated_fallback,
roles: peer.info.roles,
notifications_sink,
@@ -1672,9 +1597,7 @@ where
} else {
CustomMessageOutcome::NotificationStreamReplaced {
remote: peer_id,
protocol: self.notification_protocols
[usize::from(set_id) - NUM_HARDCODED_PEERSETS]
.clone(),
protocol: self.notification_protocols[usize::from(set_id)].clone(),
notifications_sink,
}
},
@@ -1699,9 +1622,7 @@ where
} else {
CustomMessageOutcome::NotificationStreamClosed {
remote: peer_id,
protocol: self.notification_protocols
[usize::from(set_id) - NUM_HARDCODED_PEERSETS]
.clone(),
protocol: self.notification_protocols[usize::from(set_id)].clone(),
}
}
},
@@ -1734,9 +1655,7 @@ where
_ if self.bad_handshake_substreams.contains(&(peer_id, set_id)) =>
CustomMessageOutcome::None,
_ => {
let protocol_name = self.notification_protocols
[usize::from(set_id) - NUM_HARDCODED_PEERSETS]
.clone();
let protocol_name = self.notification_protocols[usize::from(set_id)].clone();
CustomMessageOutcome::NotificationsReceived {
remote: peer_id,
messages: vec![(protocol_name, message.freeze())],