pezkuwichain/pezkuwi-subxt
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-subxt.git synced 2026-06-13 19:51:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove necessity to pass ConsensusEngineId when registering notifications protocol (#7549)

Browse Source 
* Remove necessity to pass ConsensusEngineId when registering notifications protocol

* Line width

* Fix tests protocol name

* Other renames

* Doc update

* Change issue in TODO
This commit is contained in:
Pierre Krieger
2020-11-18 16:05:35 +01:00
committed by GitHub
parent 22a02d3e7a
commit 1eae9f5792
18 changed files with 228 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files
substrate/client/network/src/service.rs
+46 -81
View File
@@ -53,10 +53,7 @@ use metrics::{Metrics, MetricSources, Histogram, HistogramVec};
use parking_lot::Mutex;
use sc_peerset::PeersetHandle;
use sp_consensus::import_queue::{BlockImportError, BlockImportResult, ImportQueue, Link};
use sp_runtime::{
traits::{Block as BlockT, NumberFor},
ConsensusEngineId,
};
use sp_runtime::traits::{Block as BlockT, NumberFor};
use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
use std::{
borrow::Cow,
@@ -100,9 +97,7 @@ pub struct NetworkService<B: BlockT + 'static, H: ExHashT> {
to_worker: TracingUnboundedSender<ServiceToWorkerMsg<B, H>>,
/// For each peer and protocol combination, an object that allows sending notifications to
/// that peer. Updated by the [`NetworkWorker`].
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ConsensusEngineId), NotificationsSink>>>,
/// For each legacy gossiping engine ID, the corresponding new protocol name.
protocol_name_by_engine: Mutex<HashMap<ConsensusEngineId, Cow<'static, str>>>,
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, Cow<'static, str>), NotificationsSink>>>,
/// Field extracted from the [`Metrics`] struct and necessary to report the
/// notifications-related metrics.
notifications_sizes_metric: Option<HistogramVec>,
@@ -331,8 +326,8 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
}
};
for (engine_id, protocol_name) in &params.network_config.notifications_protocols {
behaviour.register_notifications_protocol(*engine_id, protocol_name.clone());
for protocol in &params.network_config.notifications_protocols {
behaviour.register_notifications_protocol(protocol.clone());
}
let (transport, bandwidth) = {
let (config_mem, config_wasm) = match params.network_config.transport {
@@ -384,9 +379,6 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
let external_addresses = Arc::new(Mutex::new(Vec::new()));
let peers_notifications_sinks = Arc::new(Mutex::new(HashMap::new()));
let protocol_name_by_engine = Mutex::new({
params.network_config.notifications_protocols.iter().cloned().collect()
});
let service = Arc::new(NetworkService {
bandwidth,
@@ -397,7 +389,6 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkWorker<B, H> {
local_peer_id,
to_worker,
peers_notifications_sinks: peers_notifications_sinks.clone(),
protocol_name_by_engine,
notifications_sizes_metric:
metrics.as_ref().map(|metrics| metrics.notifications_sizes.clone()),
_marker: PhantomData,
@@ -640,40 +631,32 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
/// The protocol must have been registered with `register_notifications_protocol` or
/// [`NetworkConfiguration::notifications_protocols`](crate::config::NetworkConfiguration::notifications_protocols).
///
pub fn write_notification(&self, target: PeerId, engine_id: ConsensusEngineId, message: Vec<u8>) {
pub fn write_notification(&self, target: PeerId, protocol: Cow<'static, str>, message: Vec<u8>) {
// We clone the `NotificationsSink` in order to be able to unlock the network-wide
// `peers_notifications_sinks` mutex as soon as possible.
let sink = {
let peers_notifications_sinks = self.peers_notifications_sinks.lock();
if let Some(sink) = peers_notifications_sinks.get(&(target, engine_id)) {
if let Some(sink) = peers_notifications_sinks.get(&(target, protocol.clone())) {
sink.clone()
} else {
// Notification silently discarded, as documented.
log::error!(
target: "sub-libp2p",
"Attempted to send notification on unknown protocol: {:?}",
protocol,
);
return;
}
};
// Used later for the metrics report.
let message_len = message.len();
// Determine the wire protocol name corresponding to this `engine_id`.
let protocol_name = self.protocol_name_by_engine.lock().get(&engine_id).cloned();
if let Some(protocol_name) = protocol_name {
sink.send_sync_notification(protocol_name, message);
} else {
log::error!(
target: "sub-libp2p",
"Attempted to send notification on unknown protocol: {:?}",
engine_id,
);
return;
}
if let Some(notifications_sizes_metric) = self.notifications_sizes_metric.as_ref() {
notifications_sizes_metric
.with_label_values(&["out", &maybe_utf8_bytes_to_string(&engine_id)])
.observe(message_len as f64);
.with_label_values(&["out", &protocol])
.observe(message.len() as f64);
}
// Sending is communicated to the `NotificationsSink`.
sink.send_sync_notification(protocol, message);
}
/// Obtains a [`NotificationSender`] for a connected peer, if it exists.
@@ -746,31 +729,27 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
pub fn notification_sender(
&self,
target: PeerId,
engine_id: ConsensusEngineId,
protocol: Cow<'static, str>,
) -> Result<NotificationSender, NotificationSenderError> {
// We clone the `NotificationsSink` in order to be able to unlock the network-wide
// `peers_notifications_sinks` mutex as soon as possible.
let sink = {
let peers_notifications_sinks = self.peers_notifications_sinks.lock();
if let Some(sink) = peers_notifications_sinks.get(&(target, engine_id)) {
if let Some(sink) = peers_notifications_sinks.get(&(target, protocol.clone())) {
sink.clone()
} else {
return Err(NotificationSenderError::Closed);
}
};
// Determine the wire protocol name corresponding to this `engine_id`.
let protocol_name = match self.protocol_name_by_engine.lock().get(&engine_id).cloned() {
Some(p) => p,
None => return Err(NotificationSenderError::BadProtocol),
};
let notification_size_metric = self.notifications_sizes_metric.as_ref().map(|histogram| {
histogram.with_label_values(&["out", &protocol])
});
Ok(NotificationSender {
sink,
protocol_name,
notification_size_metric: self.notifications_sizes_metric.as_ref().map(|histogram| {
histogram.with_label_values(&["out", &maybe_utf8_bytes_to_string(&engine_id)])
}),
protocol_name: protocol,
notification_size_metric,
})
}
@@ -841,17 +820,13 @@ impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
///
/// Please call `event_stream` before registering a protocol, otherwise you may miss events
/// about the protocol that you have registered.
// TODO: remove this method after https://github.com/paritytech/substrate/issues/4587
// TODO: remove this method after https://github.com/paritytech/substrate/issues/6827
pub fn register_notifications_protocol(
&self,
engine_id: ConsensusEngineId,
protocol_name: impl Into<Cow<'static, str>>,
) {
let protocol_name = protocol_name.into();
self.protocol_name_by_engine.lock().insert(engine_id, protocol_name.clone());
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::RegisterNotifProtocol {
engine_id,
protocol_name,
protocol_name: protocol_name.into(),
});
}
@@ -1209,7 +1184,6 @@ enum ServiceToWorkerMsg<B: BlockT, H: ExHashT> {
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
},
RegisterNotifProtocol {
engine_id: ConsensusEngineId,
protocol_name: Cow<'static, str>,
},
DisconnectPeer(PeerId),
@@ -1253,7 +1227,7 @@ pub struct NetworkWorker<B: BlockT + 'static, H: ExHashT> {
>,
/// For each peer and protocol combination, an object that allows sending notifications to
/// that peer. Shared with the [`NetworkService`].
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ConsensusEngineId), NotificationsSink>>>,
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, Cow<'static, str>), NotificationsSink>>>,
}
impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
@@ -1347,10 +1321,8 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
},
}
},
ServiceToWorkerMsg::RegisterNotifProtocol { engine_id, protocol_name } => {
this.network_service
.register_notifications_protocol(engine_id, protocol_name);
},
ServiceToWorkerMsg::RegisterNotifProtocol { protocol_name } =>
this.network_service.register_notifications_protocol(protocol_name),
ServiceToWorkerMsg::DisconnectPeer(who) =>
this.network_service.user_protocol_mut().disconnect_peer(&who),
ServiceToWorkerMsg::UpdateChain =>
@@ -1474,24 +1446,28 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
.inc();
}
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamOpened { remote, engine_id, notifications_sink, role })) => {
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamOpened {
remote, protocol, notifications_sink, role
})) => {
if let Some(metrics) = this.metrics.as_ref() {
metrics.notifications_streams_opened_total
.with_label_values(&[&maybe_utf8_bytes_to_string(&engine_id)]).inc();
.with_label_values(&[&protocol]).inc();
}
{
let mut peers_notifications_sinks = this.peers_notifications_sinks.lock();
peers_notifications_sinks.insert((remote.clone(), engine_id), notifications_sink);
peers_notifications_sinks.insert((remote.clone(), protocol.clone()), notifications_sink);
}
this.event_streams.send(Event::NotificationStreamOpened {
remote,
engine_id,
protocol,
role,
});
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamReplaced { remote, engine_id, notifications_sink })) => {
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamReplaced {
remote, protocol, notifications_sink
})) => {
let mut peers_notifications_sinks = this.peers_notifications_sinks.lock();
if let Some(s) = peers_notifications_sinks.get_mut(&(remote, engine_id)) {
if let Some(s) = peers_notifications_sinks.get_mut(&(remote, protocol)) {
*s = notifications_sink;
} else {
log::error!(
@@ -1513,33 +1489,33 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
// https://github.com/paritytech/substrate/issues/6403.
/*this.event_streams.send(Event::NotificationStreamClosed {
remote,
engine_id,
protocol,
});
this.event_streams.send(Event::NotificationStreamOpened {
remote,
engine_id,
protocol,
role,
});*/
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamClosed { remote, engine_id })) => {
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamClosed { remote, protocol })) => {
if let Some(metrics) = this.metrics.as_ref() {
metrics.notifications_streams_closed_total
.with_label_values(&[&maybe_utf8_bytes_to_string(&engine_id[..])]).inc();
.with_label_values(&[&protocol[..]]).inc();
}
this.event_streams.send(Event::NotificationStreamClosed {
remote: remote.clone(),
engine_id,
protocol: protocol.clone(),
});
{
let mut peers_notifications_sinks = this.peers_notifications_sinks.lock();
peers_notifications_sinks.remove(&(remote.clone(), engine_id));
peers_notifications_sinks.remove(&(remote.clone(), protocol));
}
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationsReceived { remote, messages })) => {
if let Some(metrics) = this.metrics.as_ref() {
for (engine_id, message) in &messages {
for (protocol, message) in &messages {
metrics.notifications_sizes
.with_label_values(&["in", &maybe_utf8_bytes_to_string(engine_id)])
.with_label_values(&["in", protocol])
.observe(message.len() as f64);
}
}
@@ -1748,17 +1724,6 @@ impl<B: BlockT + 'static, H: ExHashT> Future for NetworkWorker<B, H> {
impl<B: BlockT + 'static, H: ExHashT> Unpin for NetworkWorker<B, H> {
}
/// Turns bytes that are potentially UTF-8 into a reasonable representable string.
///
/// Meant to be used only for debugging or metrics-reporting purposes.
pub(crate) fn maybe_utf8_bytes_to_string(id: &[u8]) -> Cow<str> {
if let Ok(s) = std::str::from_utf8(&id[..]) {
Cow::Borrowed(s)
} else {
Cow::Owned(format!("{:?}", id))
}
}
/// The libp2p swarm, customized for our needs.
type Swarm<B, H> = libp2p::swarm::Swarm<Behaviour<B, H>>;