use crate::{
behaviour::{self, Behaviour, BehaviourOut},
config::Params,
discovery::DiscoveryConfig,
network_state::{
NetworkState, NotConnectedPeer as NetworkStateNotConnectedPeer, Peer as NetworkStatePeer,
},
protocol::{self, NotificationsSink, NotifsHandlerError, PeerInfo, Protocol, Ready},
transport, ChainSyncInterface, ReputationChange,
};
use futures::{channel::oneshot, prelude::*};
use libp2p::{
core::{either::EitherError, upgrade, ConnectedPoint},
identify::Info as IdentifyInfo,
kad::record::Key as KademliaKey,
multiaddr,
ping::Failure as PingFailure,
swarm::{
AddressScore, ConnectionError, ConnectionLimits, DialError, Executor, NetworkBehaviour,
PendingConnectionError, Swarm, SwarmBuilder, SwarmEvent,
},
Multiaddr, PeerId,
};
use log::{debug, error, info, trace, warn};
use metrics::{Histogram, HistogramVec, MetricSources, Metrics};
use parking_lot::Mutex;
use sc_network_common::{
config::{MultiaddrWithPeerId, TransportConfig},
error::Error,
protocol::{
event::{DhtEvent, Event},
ProtocolName,
},
request_responses::{IfDisconnected, RequestFailure},
service::{
NetworkDHTProvider, NetworkEventStream, NetworkNotification, NetworkPeers, NetworkSigner,
NetworkStateInfo, NetworkStatus, NetworkStatusProvider, NetworkSyncForkRequest,
NotificationSender as NotificationSenderT, NotificationSenderError,
NotificationSenderReady as NotificationSenderReadyT, Signature, SigningError,
},
sync::SyncStatus,
ExHashT,
};
use sc_peerset::PeersetHandle;
use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
use sp_blockchain::HeaderBackend;
use sp_runtime::traits::{Block as BlockT, NumberFor, Zero};
use std::{
cmp,
collections::{HashMap, HashSet},
fs, iter,
marker::PhantomData,
num::NonZeroUsize,
pin::Pin,
str,
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
Arc,
},
task::Poll,
};
pub use behaviour::{InboundFailure, OutboundFailure, ResponseFailure};
mod metrics;
mod out_events;
#[cfg(test)]
mod tests;
pub use libp2p::identity::{error::DecodingError, Keypair, PublicKey};
use sc_network_common::service::{NetworkBlock, NetworkRequest};
pub struct NetworkService<B: BlockT + 'static, H: ExHashT> {
num_connected: Arc<AtomicUsize>,
external_addresses: Arc<Mutex<Vec<Multiaddr>>>,
is_major_syncing: Arc<AtomicBool>,
local_peer_id: PeerId,
local_identity: Keypair,
bandwidth: Arc<transport::BandwidthSinks>,
peerset: PeersetHandle,
to_worker: TracingUnboundedSender<ServiceToWorkerMsg<B>>,
chain_sync_service: Box<dyn ChainSyncInterface<B>>,
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ProtocolName), NotificationsSink>>>,
notifications_sizes_metric: Option<HistogramVec>,
_marker: PhantomData<H>,
}
impl<B, H, Client> NetworkWorker<B, H, Client>
where
B: BlockT + 'static,
H: ExHashT,
Client: HeaderBackend<B> + 'static,
{
pub fn new(mut params: Params<B, Client>) -> Result<Self, Error> {
let local_identity = params.network_config.node_key.clone().into_keypair()?;
let local_public = local_identity.public();
let local_peer_id = local_public.to_peer_id();
params
.network_config
.request_response_protocols
.extend(params.request_response_protocol_configs);
params.network_config.boot_nodes = params
.network_config
.boot_nodes
.into_iter()
.filter(|boot_node| boot_node.peer_id != local_peer_id)
.collect();
params.network_config.default_peers_set.reserved_nodes = params
.network_config
.default_peers_set
.reserved_nodes
.into_iter()
.filter(|reserved_node| {
if reserved_node.peer_id == local_peer_id {
warn!(
target: "sub-libp2p",
"Local peer ID used in reserved node, ignoring: {}",
reserved_node,
);
false
} else {
true
}
})
.collect();
ensure_addresses_consistent_with_transport(
params.network_config.listen_addresses.iter(),
¶ms.network_config.transport,
)?;
ensure_addresses_consistent_with_transport(
params.network_config.boot_nodes.iter().map(|x| &x.multiaddr),
¶ms.network_config.transport,
)?;
ensure_addresses_consistent_with_transport(
params
.network_config
.default_peers_set
.reserved_nodes
.iter()
.map(|x| &x.multiaddr),
¶ms.network_config.transport,
)?;
for extra_set in ¶ms.network_config.extra_sets {
ensure_addresses_consistent_with_transport(
extra_set.set_config.reserved_nodes.iter().map(|x| &x.multiaddr),
¶ms.network_config.transport,
)?;
}
ensure_addresses_consistent_with_transport(
params.network_config.public_addresses.iter(),
¶ms.network_config.transport,
)?;
let (to_worker, from_service) = tracing_unbounded("mpsc_network_worker", 100_000);
if let Some(path) = ¶ms.network_config.net_config_path {
fs::create_dir_all(path)?;
}
info!(
target: "sub-libp2p",
"🏷 Local node identity is: {}",
local_peer_id.to_base58(),
);
let (protocol, peerset_handle, mut known_addresses) = Protocol::new(
From::from(¶ms.role),
params.chain.clone(),
¶ms.network_config,
params.metrics_registry.as_ref(),
params.chain_sync,
params.block_announce_config,
)?;
let mut boot_node_ids = HashSet::new();
for bootnode in params.network_config.boot_nodes.iter() {
boot_node_ids.insert(bootnode.peer_id);
known_addresses.push((bootnode.peer_id, bootnode.multiaddr.clone()));
}
let boot_node_ids = Arc::new(boot_node_ids);
params.network_config.boot_nodes.iter().try_for_each(|bootnode| {
if let Some(other) = params
.network_config
.boot_nodes
.iter()
.filter(|o| o.multiaddr == bootnode.multiaddr)
.find(|o| o.peer_id != bootnode.peer_id)
{
Err(Error::DuplicateBootnode {
address: bootnode.multiaddr.clone(),
first_id: bootnode.peer_id,
second_id: other.peer_id,
})
} else {
Ok(())
}
})?;
let num_connected = Arc::new(AtomicUsize::new(0));
let is_major_syncing = Arc::new(AtomicBool::new(false));
let (mut swarm, bandwidth): (Swarm<Behaviour<B, Client>>, _) = {
let user_agent = format!(
"{} ({})",
params.network_config.client_version, params.network_config.node_name
);
let discovery_config = {
let mut config = DiscoveryConfig::new(local_public.clone());
config.with_permanent_addresses(known_addresses);
config.discovery_limit(
u64::from(params.network_config.default_peers_set.out_peers) + 15,
);
let genesis_hash = params
.chain
.hash(Zero::zero())
.ok()
.flatten()
.expect("Genesis block exists; qed");
config.with_kademlia(genesis_hash, params.fork_id.as_deref(), ¶ms.protocol_id);
config.with_dht_random_walk(params.network_config.enable_dht_random_walk);
config.allow_non_globals_in_dht(params.network_config.allow_non_globals_in_dht);
config.use_kademlia_disjoint_query_paths(
params.network_config.kademlia_disjoint_query_paths,
);
match params.network_config.transport {
TransportConfig::MemoryOnly => {
config.with_mdns(false);
config.allow_private_ipv4(false);
},
TransportConfig::Normal { enable_mdns, allow_private_ipv4, .. } => {
config.with_mdns(enable_mdns);
config.allow_private_ipv4(allow_private_ipv4);
},
}
config
};
let (transport, bandwidth) = {
let config_mem = match params.network_config.transport {
TransportConfig::MemoryOnly => true,
TransportConfig::Normal { .. } => false,
};
let yamux_maximum_buffer_size = {
let requests_max = params
.network_config
.request_response_protocols
.iter()
.map(|cfg| usize::try_from(cfg.max_request_size).unwrap_or(usize::MAX));
let responses_max =
params.network_config.request_response_protocols.iter().map(|cfg| {
usize::try_from(cfg.max_response_size).unwrap_or(usize::MAX)
});
let notifs_max = params.network_config.extra_sets.iter().map(|cfg| {
usize::try_from(cfg.max_notification_size).unwrap_or(usize::MAX)
});
let default_max = cmp::max(
1024 * 1024,
usize::try_from(protocol::BLOCK_ANNOUNCES_TRANSACTIONS_SUBSTREAM_SIZE)
.unwrap_or(usize::MAX),
);
iter::once(default_max)
.chain(requests_max)
.chain(responses_max)
.chain(notifs_max)
.max()
.expect("iterator known to always yield at least one element; qed")
.saturating_add(10)
};
transport::build_transport(
local_identity.clone(),
config_mem,
params.network_config.yamux_window_size,
yamux_maximum_buffer_size,
)
};
let behaviour = {
let result = Behaviour::new(
protocol,
user_agent,
local_public,
discovery_config,
params.network_config.request_response_protocols,
peerset_handle.clone(),
);
match result {
Ok(b) => b,
Err(crate::request_responses::RegisterError::DuplicateProtocol(proto)) =>
return Err(Error::DuplicateRequestResponseProtocol { protocol: proto }),
}
};
let builder = {
struct SpawnImpl<F>(F);
impl<F: Fn(Pin<Box<dyn Future<Output = ()> + Send>>)> Executor for SpawnImpl<F> {
fn exec(&self, f: Pin<Box<dyn Future<Output = ()> + Send>>) {
(self.0)(f)
}
}
SwarmBuilder::with_executor(
transport,
behaviour,
local_peer_id,
SpawnImpl(params.executor),
)
};
let builder = builder
.connection_limits(
ConnectionLimits::default()
.with_max_established_per_peer(Some(crate::MAX_CONNECTIONS_PER_PEER as u32))
.with_max_established_incoming(Some(
crate::MAX_CONNECTIONS_ESTABLISHED_INCOMING,
)),
)
.substream_upgrade_protocol_override(upgrade::Version::V1Lazy)
.notify_handler_buffer_size(NonZeroUsize::new(32).expect("32 != 0; qed"))
.connection_event_buffer_size(1024)
.max_negotiating_inbound_streams(2048);
(builder.build(), bandwidth)
};
let metrics = match ¶ms.metrics_registry {
Some(registry) => Some(metrics::register(
registry,
MetricSources {
bandwidth: bandwidth.clone(),
major_syncing: is_major_syncing.clone(),
connected_peers: num_connected.clone(),
},
)?),
None => None,
};
for addr in ¶ms.network_config.listen_addresses {
if let Err(err) = Swarm::<Behaviour<B, Client>>::listen_on(&mut swarm, addr.clone()) {
warn!(target: "sub-libp2p", "Can't listen on {} because: {:?}", addr, err)
}
}
for addr in ¶ms.network_config.public_addresses {
Swarm::<Behaviour<B, Client>>::add_external_address(
&mut swarm,
addr.clone(),
AddressScore::Infinite,
);
}
let external_addresses = Arc::new(Mutex::new(Vec::new()));
let peers_notifications_sinks = Arc::new(Mutex::new(HashMap::new()));
let service = Arc::new(NetworkService {
bandwidth,
external_addresses: external_addresses.clone(),
num_connected: num_connected.clone(),
is_major_syncing: is_major_syncing.clone(),
peerset: peerset_handle,
local_peer_id,
local_identity,
to_worker,
chain_sync_service: params.chain_sync_service,
peers_notifications_sinks: peers_notifications_sinks.clone(),
notifications_sizes_metric: metrics
.as_ref()
.map(|metrics| metrics.notifications_sizes.clone()),
_marker: PhantomData,
});
Ok(NetworkWorker {
external_addresses,
num_connected,
is_major_syncing,
network_service: swarm,
service,
from_service,
event_streams: out_events::OutChannels::new(params.metrics_registry.as_ref())?,
peers_notifications_sinks,
metrics,
boot_node_ids,
_marker: Default::default(),
})
}
pub fn status(&self) -> NetworkStatus<B> {
let status = self.sync_state();
NetworkStatus {
sync_state: status.state,
best_seen_block: self.best_seen_block(),
num_sync_peers: self.num_sync_peers(),
num_connected_peers: self.num_connected_peers(),
num_active_peers: self.num_active_peers(),
total_bytes_inbound: self.total_bytes_inbound(),
total_bytes_outbound: self.total_bytes_outbound(),
state_sync: status.state_sync,
warp_sync: status.warp_sync,
}
}
pub fn total_bytes_inbound(&self) -> u64 {
self.service.bandwidth.total_inbound()
}
pub fn total_bytes_outbound(&self) -> u64 {
self.service.bandwidth.total_outbound()
}
pub fn num_connected_peers(&self) -> usize {
self.network_service.behaviour().user_protocol().num_connected_peers()
}
pub fn num_active_peers(&self) -> usize {
self.network_service.behaviour().user_protocol().num_active_peers()
}
pub fn sync_state(&self) -> SyncStatus<B> {
self.network_service.behaviour().user_protocol().sync_state()
}
pub fn best_seen_block(&self) -> Option<NumberFor<B>> {
self.network_service.behaviour().user_protocol().best_seen_block()
}
pub fn num_sync_peers(&self) -> u32 {
self.network_service.behaviour().user_protocol().num_sync_peers()
}
pub fn num_queued_blocks(&self) -> u32 {
self.network_service.behaviour().user_protocol().num_queued_blocks()
}
pub fn num_downloaded_blocks(&self) -> usize {
self.network_service.behaviour().user_protocol().num_downloaded_blocks()
}
pub fn num_sync_requests(&self) -> usize {
self.network_service.behaviour().user_protocol().num_sync_requests()
}
pub fn add_known_address(&mut self, peer_id: PeerId, addr: Multiaddr) {
self.network_service.behaviour_mut().add_known_address(peer_id, addr);
}
pub fn service(&self) -> &Arc<NetworkService<B, H>> {
&self.service
}
pub fn on_block_finalized(&mut self, hash: B::Hash, header: B::Header) {
self.network_service
.behaviour_mut()
.user_protocol_mut()
.on_block_finalized(hash, &header);
}
pub fn new_best_block_imported(&mut self, hash: B::Hash, number: NumberFor<B>) {
self.network_service
.behaviour_mut()
.user_protocol_mut()
.new_best_block_imported(hash, number);
}
pub fn local_peer_id(&self) -> &PeerId {
Swarm::<Behaviour<B, Client>>::local_peer_id(&self.network_service)
}
pub fn listen_addresses(&self) -> impl Iterator<Item = &Multiaddr> {
Swarm::<Behaviour<B, Client>>::listeners(&self.network_service)
}
pub fn network_state(&mut self) -> NetworkState {
let swarm = &mut self.network_service;
let open = swarm.behaviour_mut().user_protocol().open_peers().cloned().collect::<Vec<_>>();
let connected_peers = {
let swarm = &mut *swarm;
open.iter()
.filter_map(move |peer_id| {
let known_addresses =
NetworkBehaviour::addresses_of_peer(swarm.behaviour_mut(), peer_id)
.into_iter()
.collect();
let endpoint = if let Some(e) =
swarm.behaviour_mut().node(peer_id).and_then(|i| i.endpoint())
{
e.clone().into()
} else {
error!(target: "sub-libp2p", "Found state inconsistency between custom protocol \
and debug information about {:?}", peer_id);
return None
};
Some((
peer_id.to_base58(),
NetworkStatePeer {
endpoint,
version_string: swarm
.behaviour_mut()
.node(peer_id)
.and_then(|i| i.client_version().map(|s| s.to_owned())),
latest_ping_time: swarm
.behaviour_mut()
.node(peer_id)
.and_then(|i| i.latest_ping()),
known_addresses,
},
))
})
.collect()
};
let not_connected_peers = {
let swarm = &mut *swarm;
swarm
.behaviour_mut()
.known_peers()
.into_iter()
.filter(|p| open.iter().all(|n| n != p))
.map(move |peer_id| {
(
peer_id.to_base58(),
NetworkStateNotConnectedPeer {
version_string: swarm
.behaviour_mut()
.node(&peer_id)
.and_then(|i| i.client_version().map(|s| s.to_owned())),
latest_ping_time: swarm
.behaviour_mut()
.node(&peer_id)
.and_then(|i| i.latest_ping()),
known_addresses: NetworkBehaviour::addresses_of_peer(
swarm.behaviour_mut(),
&peer_id,
)
.into_iter()
.collect(),
},
)
})
.collect()
};
let peer_id = Swarm::<Behaviour<B, Client>>::local_peer_id(swarm).to_base58();
let listened_addresses = swarm.listeners().cloned().collect();
let external_addresses = swarm.external_addresses().map(|r| &r.addr).cloned().collect();
NetworkState {
peer_id,
listened_addresses,
external_addresses,
connected_peers,
not_connected_peers,
peerset: swarm.behaviour_mut().user_protocol_mut().peerset_debug_info(),
}
}
pub fn peers_debug_info(&mut self) -> Vec<(PeerId, PeerInfo<B>)> {
self.network_service
.behaviour_mut()
.user_protocol_mut()
.peers_info()
.map(|(id, info)| (*id, info.clone()))
.collect()
}
pub fn remove_reserved_peer(&self, peer: PeerId) {
self.service.remove_reserved_peer(peer);
}
pub fn add_reserved_peer(&self, peer: MultiaddrWithPeerId) -> Result<(), String> {
self.service.add_reserved_peer(peer)
}
pub fn reserved_peers(&self) -> impl Iterator<Item = &PeerId> {
self.network_service.behaviour().user_protocol().reserved_peers()
}
}
impl<B: BlockT + 'static, H: ExHashT> NetworkService<B, H> {
pub async fn network_state(&self) -> Result<NetworkState, ()> {
let (tx, rx) = oneshot::channel();
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::NetworkState { pending_response: tx });
match rx.await {
Ok(v) => v.map_err(|_| ()),
Err(_) => Err(()),
}
}
fn split_multiaddr_and_peer_id(
&self,
peers: HashSet<Multiaddr>,
) -> Result<Vec<(PeerId, Multiaddr)>, String> {
peers
.into_iter()
.map(|mut addr| {
let peer = match addr.pop() {
Some(multiaddr::Protocol::P2p(key)) => PeerId::from_multihash(key)
.map_err(|_| "Invalid PeerId format".to_string())?,
_ => return Err("Missing PeerId from address".to_string()),
};
if peer == self.local_peer_id {
Err("Local peer ID in peer set.".to_string())
} else {
Ok((peer, addr))
}
})
.collect::<Result<Vec<(PeerId, Multiaddr)>, String>>()
}
}
impl<B: BlockT + 'static, H: ExHashT> sp_consensus::SyncOracle for NetworkService<B, H> {
fn is_major_syncing(&self) -> bool {
self.is_major_syncing.load(Ordering::Relaxed)
}
fn is_offline(&self) -> bool {
self.num_connected.load(Ordering::Relaxed) == 0
}
}
impl<B: BlockT, H: ExHashT> sc_consensus::JustificationSyncLink<B> for NetworkService<B, H> {
fn request_justification(&self, hash: &B::Hash, number: NumberFor<B>) {
let _ = self.chain_sync_service.request_justification(hash, number);
}
fn clear_justification_requests(&self) {
let _ = self.chain_sync_service.clear_justification_requests();
}
}
impl<B, H> NetworkStateInfo for NetworkService<B, H>
where
B: sp_runtime::traits::Block,
H: ExHashT,
{
fn external_addresses(&self) -> Vec<Multiaddr> {
self.external_addresses.lock().clone()
}
fn local_peer_id(&self) -> PeerId {
self.local_peer_id
}
}
impl<B, H> NetworkSigner for NetworkService<B, H>
where
B: sp_runtime::traits::Block,
H: ExHashT,
{
fn sign_with_local_identity(&self, msg: impl AsRef<[u8]>) -> Result<Signature, SigningError> {
Signature::sign_message(msg.as_ref(), &self.local_identity)
}
}
impl<B, H> NetworkDHTProvider for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
fn get_value(&self, key: &KademliaKey) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::GetValue(key.clone()));
}
fn put_value(&self, key: KademliaKey, value: Vec<u8>) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::PutValue(key, value));
}
}
impl<B, H> NetworkSyncForkRequest<B::Hash, NumberFor<B>> for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
fn set_sync_fork_request(&self, peers: Vec<PeerId>, hash: B::Hash, number: NumberFor<B>) {
self.chain_sync_service.set_sync_fork_request(peers, hash, number);
}
}
#[async_trait::async_trait]
impl<B, H> NetworkStatusProvider<B> for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
async fn status(&self) -> Result<NetworkStatus<B>, ()> {
let (tx, rx) = oneshot::channel();
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::NetworkStatus { pending_response: tx });
match rx.await {
Ok(v) => v.map_err(|_| ()),
Err(_) => Err(()),
}
}
}
impl<B, H> NetworkPeers for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
fn set_authorized_peers(&self, peers: HashSet<PeerId>) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::SetReserved(peers));
}
fn set_authorized_only(&self, reserved_only: bool) {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::SetReservedOnly(reserved_only));
}
fn add_known_address(&self, peer_id: PeerId, addr: Multiaddr) {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddKnownAddress(peer_id, addr));
}
fn report_peer(&self, who: PeerId, cost_benefit: ReputationChange) {
self.peerset.report_peer(who, cost_benefit);
}
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::DisconnectPeer(who, protocol));
}
fn accept_unreserved_peers(&self) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::SetReservedOnly(false));
}
fn deny_unreserved_peers(&self) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::SetReservedOnly(true));
}
fn add_reserved_peer(&self, peer: MultiaddrWithPeerId) -> Result<(), String> {
if peer.peer_id == self.local_peer_id {
return Err("Local peer ID cannot be added as a reserved peer.".to_string())
}
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddKnownAddress(peer.peer_id, peer.multiaddr));
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::AddReserved(peer.peer_id));
Ok(())
}
fn remove_reserved_peer(&self, peer_id: PeerId) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::RemoveReserved(peer_id));
}
fn set_reserved_peers(
&self,
protocol: ProtocolName,
peers: HashSet<Multiaddr>,
) -> Result<(), String> {
let peers_addrs = self.split_multiaddr_and_peer_id(peers)?;
let mut peers: HashSet<PeerId> = HashSet::with_capacity(peers_addrs.len());
for (peer_id, addr) in peers_addrs.into_iter() {
if peer_id == self.local_peer_id {
return Err("Local peer ID cannot be added as a reserved peer.".to_string())
}
peers.insert(peer_id);
if !addr.is_empty() {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddKnownAddress(peer_id, addr));
}
}
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::SetPeersetReserved(protocol, peers));
Ok(())
}
fn add_peers_to_reserved_set(
&self,
protocol: ProtocolName,
peers: HashSet<Multiaddr>,
) -> Result<(), String> {
let peers = self.split_multiaddr_and_peer_id(peers)?;
for (peer_id, addr) in peers.into_iter() {
if peer_id == self.local_peer_id {
return Err("Local peer ID cannot be added as a reserved peer.".to_string())
}
if !addr.is_empty() {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddKnownAddress(peer_id, addr));
}
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddSetReserved(protocol.clone(), peer_id));
}
Ok(())
}
fn remove_peers_from_reserved_set(&self, protocol: ProtocolName, peers: Vec<PeerId>) {
for peer_id in peers.into_iter() {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::RemoveSetReserved(protocol.clone(), peer_id));
}
}
fn add_to_peers_set(
&self,
protocol: ProtocolName,
peers: HashSet<Multiaddr>,
) -> Result<(), String> {
let peers = self.split_multiaddr_and_peer_id(peers)?;
for (peer_id, addr) in peers.into_iter() {
if peer_id == self.local_peer_id {
return Err("Local peer ID cannot be added as a reserved peer.".to_string())
}
if !addr.is_empty() {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddKnownAddress(peer_id, addr));
}
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::AddToPeersSet(protocol.clone(), peer_id));
}
Ok(())
}
fn remove_from_peers_set(&self, protocol: ProtocolName, peers: Vec<PeerId>) {
for peer_id in peers.into_iter() {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::RemoveFromPeersSet(protocol.clone(), peer_id));
}
}
fn sync_num_connected(&self) -> usize {
self.num_connected.load(Ordering::Relaxed)
}
}
impl<B, H> NetworkEventStream for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
fn event_stream(&self, name: &'static str) -> Pin<Box<dyn Stream<Item = Event> + Send>> {
let (tx, rx) = out_events::channel(name, 100_000);
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::EventStream(tx));
Box::pin(rx)
}
}
impl<B, H> NetworkNotification for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
fn write_notification(&self, target: PeerId, protocol: ProtocolName, message: Vec<u8>) {
let sink = {
let peers_notifications_sinks = self.peers_notifications_sinks.lock();
if let Some(sink) = peers_notifications_sinks.get(&(target, protocol.clone())) {
sink.clone()
} else {
debug!(
target: "sub-libp2p",
"Attempted to send notification on missing or closed substream: {}, {:?}",
target, protocol,
);
return
}
};
if let Some(notifications_sizes_metric) = self.notifications_sizes_metric.as_ref() {
notifications_sizes_metric
.with_label_values(&["out", &protocol])
.observe(message.len() as f64);
}
trace!(
target: "sub-libp2p",
"External API => Notification({:?}, {:?}, {} bytes)",
target, protocol, message.len()
);
trace!(target: "sub-libp2p", "Handler({:?}) <= Sync notification", target);
sink.send_sync_notification(message);
}
fn notification_sender(
&self,
target: PeerId,
protocol: ProtocolName,
) -> Result<Box<dyn NotificationSenderT>, NotificationSenderError> {
let sink = {
let peers_notifications_sinks = self.peers_notifications_sinks.lock();
if let Some(sink) = peers_notifications_sinks.get(&(target, protocol.clone())) {
sink.clone()
} else {
return Err(NotificationSenderError::Closed)
}
};
let notification_size_metric = self
.notifications_sizes_metric
.as_ref()
.map(|histogram| histogram.with_label_values(&["out", &protocol]));
Ok(Box::new(NotificationSender { sink, protocol_name: protocol, notification_size_metric }))
}
}
#[async_trait::async_trait]
impl<B, H> NetworkRequest for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
async fn request(
&self,
target: PeerId,
protocol: ProtocolName,
request: Vec<u8>,
connect: IfDisconnected,
) -> Result<Vec<u8>, RequestFailure> {
let (tx, rx) = oneshot::channel();
self.start_request(target, protocol, request, tx, connect);
match rx.await {
Ok(v) => v,
Err(_) => Err(RequestFailure::Network(OutboundFailure::ConnectionClosed)),
}
}
fn start_request(
&self,
target: PeerId,
protocol: ProtocolName,
request: Vec<u8>,
tx: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
connect: IfDisconnected,
) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::Request {
target,
protocol: protocol.into(),
request,
pending_response: tx,
connect,
});
}
}
impl<B, H> NetworkBlock<B::Hash, NumberFor<B>> for NetworkService<B, H>
where
B: BlockT + 'static,
H: ExHashT,
{
fn announce_block(&self, hash: B::Hash, data: Option<Vec<u8>>) {
let _ = self.to_worker.unbounded_send(ServiceToWorkerMsg::AnnounceBlock(hash, data));
}
fn new_best_block_imported(&self, hash: B::Hash, number: NumberFor<B>) {
let _ = self
.to_worker
.unbounded_send(ServiceToWorkerMsg::NewBestBlockImported(hash, number));
}
}
#[must_use]
pub struct NotificationSender {
sink: NotificationsSink,
protocol_name: ProtocolName,
notification_size_metric: Option<Histogram>,
}
#[async_trait::async_trait]
impl NotificationSenderT for NotificationSender {
async fn ready(
&self,
) -> Result<Box<dyn NotificationSenderReadyT + '_>, NotificationSenderError> {
Ok(Box::new(NotificationSenderReady {
ready: match self.sink.reserve_notification().await {
Ok(r) => Some(r),
Err(()) => return Err(NotificationSenderError::Closed),
},
peer_id: self.sink.peer_id(),
protocol_name: &self.protocol_name,
notification_size_metric: self.notification_size_metric.clone(),
}))
}
}
#[must_use]
pub struct NotificationSenderReady<'a> {
ready: Option<Ready<'a>>,
peer_id: &'a PeerId,
protocol_name: &'a ProtocolName,
notification_size_metric: Option<Histogram>,
}
impl<'a> NotificationSenderReadyT for NotificationSenderReady<'a> {
fn send(&mut self, notification: Vec<u8>) -> Result<(), NotificationSenderError> {
if let Some(notification_size_metric) = &self.notification_size_metric {
notification_size_metric.observe(notification.len() as f64);
}
trace!(
target: "sub-libp2p",
"External API => Notification({:?}, {}, {} bytes)",
self.peer_id, self.protocol_name, notification.len(),
);
trace!(target: "sub-libp2p", "Handler({:?}) <= Async notification", self.peer_id);
self.ready
.take()
.ok_or(NotificationSenderError::Closed)?
.send(notification)
.map_err(|()| NotificationSenderError::Closed)
}
}
enum ServiceToWorkerMsg<B: BlockT> {
AnnounceBlock(B::Hash, Option<Vec<u8>>),
GetValue(KademliaKey),
PutValue(KademliaKey, Vec<u8>),
AddKnownAddress(PeerId, Multiaddr),
SetReservedOnly(bool),
AddReserved(PeerId),
RemoveReserved(PeerId),
SetReserved(HashSet<PeerId>),
SetPeersetReserved(ProtocolName, HashSet<PeerId>),
AddSetReserved(ProtocolName, PeerId),
RemoveSetReserved(ProtocolName, PeerId),
AddToPeersSet(ProtocolName, PeerId),
RemoveFromPeersSet(ProtocolName, PeerId),
EventStream(out_events::Sender),
Request {
target: PeerId,
protocol: ProtocolName,
request: Vec<u8>,
pending_response: oneshot::Sender<Result<Vec<u8>, RequestFailure>>,
connect: IfDisconnected,
},
NetworkStatus {
pending_response: oneshot::Sender<Result<NetworkStatus<B>, RequestFailure>>,
},
NetworkState {
pending_response: oneshot::Sender<Result<NetworkState, RequestFailure>>,
},
DisconnectPeer(PeerId, ProtocolName),
NewBestBlockImported(B::Hash, NumberFor<B>),
}
#[must_use = "The NetworkWorker must be polled in order for the network to advance"]
pub struct NetworkWorker<B, H, Client>
where
B: BlockT + 'static,
H: ExHashT,
Client: HeaderBackend<B> + 'static,
{
external_addresses: Arc<Mutex<Vec<Multiaddr>>>,
num_connected: Arc<AtomicUsize>,
is_major_syncing: Arc<AtomicBool>,
service: Arc<NetworkService<B, H>>,
network_service: Swarm<Behaviour<B, Client>>,
from_service: TracingUnboundedReceiver<ServiceToWorkerMsg<B>>,
event_streams: out_events::OutChannels,
metrics: Option<Metrics>,
boot_node_ids: Arc<HashSet<PeerId>>,
peers_notifications_sinks: Arc<Mutex<HashMap<(PeerId, ProtocolName), NotificationsSink>>>,
_marker: PhantomData<H>,
}
impl<B, H, Client> Future for NetworkWorker<B, H, Client>
where
B: BlockT + 'static,
H: ExHashT,
Client: HeaderBackend<B> + 'static,
{
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context) -> Poll<Self::Output> {
let this = &mut *self;
let mut num_iterations = 0;
loop {
num_iterations += 1;
if num_iterations >= 100 {
cx.waker().wake_by_ref();
break
}
let msg = match this.from_service.poll_next_unpin(cx) {
Poll::Ready(Some(msg)) => msg,
Poll::Ready(None) => return Poll::Ready(()),
Poll::Pending => break,
};
match msg {
ServiceToWorkerMsg::AnnounceBlock(hash, data) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.announce_block(hash, data),
ServiceToWorkerMsg::GetValue(key) =>
this.network_service.behaviour_mut().get_value(key),
ServiceToWorkerMsg::PutValue(key, value) =>
this.network_service.behaviour_mut().put_value(key, value),
ServiceToWorkerMsg::SetReservedOnly(reserved_only) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.set_reserved_only(reserved_only),
ServiceToWorkerMsg::SetReserved(peers) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.set_reserved_peers(peers),
ServiceToWorkerMsg::SetPeersetReserved(protocol, peers) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.set_reserved_peerset_peers(protocol, peers),
ServiceToWorkerMsg::AddReserved(peer_id) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.add_reserved_peer(peer_id),
ServiceToWorkerMsg::RemoveReserved(peer_id) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.remove_reserved_peer(peer_id),
ServiceToWorkerMsg::AddSetReserved(protocol, peer_id) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.add_set_reserved_peer(protocol, peer_id),
ServiceToWorkerMsg::RemoveSetReserved(protocol, peer_id) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.remove_set_reserved_peer(protocol, peer_id),
ServiceToWorkerMsg::AddKnownAddress(peer_id, addr) =>
this.network_service.behaviour_mut().add_known_address(peer_id, addr),
ServiceToWorkerMsg::AddToPeersSet(protocol, peer_id) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.add_to_peers_set(protocol, peer_id),
ServiceToWorkerMsg::RemoveFromPeersSet(protocol, peer_id) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.remove_from_peers_set(protocol, peer_id),
ServiceToWorkerMsg::EventStream(sender) => this.event_streams.push(sender),
ServiceToWorkerMsg::Request {
target,
protocol,
request,
pending_response,
connect,
} => {
this.network_service.behaviour_mut().send_request(
&target,
&protocol,
request,
pending_response,
connect,
);
},
ServiceToWorkerMsg::NetworkStatus { pending_response } => {
let _ = pending_response.send(Ok(this.status()));
},
ServiceToWorkerMsg::NetworkState { pending_response } => {
let _ = pending_response.send(Ok(this.network_state()));
},
ServiceToWorkerMsg::DisconnectPeer(who, protocol_name) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.disconnect_peer(&who, protocol_name),
ServiceToWorkerMsg::NewBestBlockImported(hash, number) => this
.network_service
.behaviour_mut()
.user_protocol_mut()
.new_best_block_imported(hash, number),
}
}
let mut num_iterations = 0;
loop {
num_iterations += 1;
if num_iterations >= 1000 {
cx.waker().wake_by_ref();
break
}
let next_event = this.network_service.select_next_some();
futures::pin_mut!(next_event);
let poll_value = next_event.poll_unpin(cx);
match poll_value {
Poll::Pending => break,
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::InboundRequest {
protocol,
result,
..
})) => {
if let Some(metrics) = this.metrics.as_ref() {
match result {
Ok(serve_time) => {
metrics
.requests_in_success_total
.with_label_values(&[&protocol])
.observe(serve_time.as_secs_f64());
},
Err(err) => {
let reason = match err {
ResponseFailure::Network(InboundFailure::Timeout) => "timeout",
ResponseFailure::Network(
InboundFailure::UnsupportedProtocols,
) =>
continue,
ResponseFailure::Network(InboundFailure::ResponseOmission) =>
"busy-omitted",
ResponseFailure::Network(InboundFailure::ConnectionClosed) =>
"connection-closed",
};
metrics
.requests_in_failure_total
.with_label_values(&[&protocol, reason])
.inc();
},
}
}
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::RequestFinished {
protocol,
duration,
result,
..
})) =>
if let Some(metrics) = this.metrics.as_ref() {
match result {
Ok(_) => {
metrics
.requests_out_success_total
.with_label_values(&[&protocol])
.observe(duration.as_secs_f64());
},
Err(err) => {
let reason = match err {
RequestFailure::NotConnected => "not-connected",
RequestFailure::UnknownProtocol => "unknown-protocol",
RequestFailure::Refused => "refused",
RequestFailure::Obsolete => "obsolete",
RequestFailure::Network(OutboundFailure::DialFailure) =>
"dial-failure",
RequestFailure::Network(OutboundFailure::Timeout) => "timeout",
RequestFailure::Network(OutboundFailure::ConnectionClosed) =>
"connection-closed",
RequestFailure::Network(
OutboundFailure::UnsupportedProtocols,
) => "unsupported",
};
metrics
.requests_out_failure_total
.with_label_values(&[&protocol, reason])
.inc();
},
}
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::ReputationChanges {
peer,
changes,
})) =>
for change in changes {
this.network_service.behaviour().user_protocol().report_peer(peer, change);
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::PeerIdentify {
peer_id,
info:
IdentifyInfo {
protocol_version,
agent_version,
mut listen_addrs,
protocols,
..
},
})) => {
if listen_addrs.len() > 30 {
debug!(
target: "sub-libp2p",
"Node {:?} has reported more than 30 addresses; it is identified by {:?} and {:?}",
peer_id, protocol_version, agent_version
);
listen_addrs.truncate(30);
}
for addr in listen_addrs {
this.network_service
.behaviour_mut()
.add_self_reported_address_to_dht(&peer_id, &protocols, addr);
}
this.network_service
.behaviour_mut()
.user_protocol_mut()
.add_default_set_discovered_nodes(iter::once(peer_id));
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::Discovered(peer_id))) => {
this.network_service
.behaviour_mut()
.user_protocol_mut()
.add_default_set_discovered_nodes(iter::once(peer_id));
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::RandomKademliaStarted)) =>
if let Some(metrics) = this.metrics.as_ref() {
metrics.kademlia_random_queries_total.inc();
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamOpened {
remote,
protocol,
negotiated_fallback,
notifications_sink,
role,
})) => {
if let Some(metrics) = this.metrics.as_ref() {
metrics
.notifications_streams_opened_total
.with_label_values(&[&protocol])
.inc();
}
{
let mut peers_notifications_sinks = this.peers_notifications_sinks.lock();
let _previous_value = peers_notifications_sinks
.insert((remote, protocol.clone()), notifications_sink);
debug_assert!(_previous_value.is_none());
}
this.event_streams.send(Event::NotificationStreamOpened {
remote,
protocol,
negotiated_fallback,
role,
});
},
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, protocol)) {
*s = notifications_sink;
} else {
error!(
target: "sub-libp2p",
"NotificationStreamReplaced for non-existing substream"
);
debug_assert!(false);
}
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationStreamClosed {
remote,
protocol,
})) => {
if let Some(metrics) = this.metrics.as_ref() {
metrics
.notifications_streams_closed_total
.with_label_values(&[&protocol[..]])
.inc();
}
this.event_streams.send(Event::NotificationStreamClosed {
remote,
protocol: protocol.clone(),
});
{
let mut peers_notifications_sinks = this.peers_notifications_sinks.lock();
let _previous_value = peers_notifications_sinks.remove(&(remote, protocol));
debug_assert!(_previous_value.is_some());
}
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::NotificationsReceived {
remote,
messages,
})) => {
if let Some(metrics) = this.metrics.as_ref() {
for (protocol, message) in &messages {
metrics
.notifications_sizes
.with_label_values(&["in", protocol])
.observe(message.len() as f64);
}
}
this.event_streams.send(Event::NotificationsReceived { remote, messages });
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::SyncConnected(remote))) => {
this.event_streams.send(Event::SyncConnected { remote });
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::SyncDisconnected(remote))) => {
this.event_streams.send(Event::SyncDisconnected { remote });
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::Dht(event, duration))) => {
if let Some(metrics) = this.metrics.as_ref() {
let query_type = match event {
DhtEvent::ValueFound(_) => "value-found",
DhtEvent::ValueNotFound(_) => "value-not-found",
DhtEvent::ValuePut(_) => "value-put",
DhtEvent::ValuePutFailed(_) => "value-put-failed",
};
metrics
.kademlia_query_duration
.with_label_values(&[query_type])
.observe(duration.as_secs_f64());
}
this.event_streams.send(Event::Dht(event));
},
Poll::Ready(SwarmEvent::Behaviour(BehaviourOut::None)) => {
},
Poll::Ready(SwarmEvent::ConnectionEstablished {
peer_id,
endpoint,
num_established,
concurrent_dial_errors,
}) => {
if let Some(errors) = concurrent_dial_errors {
debug!(target: "sub-libp2p", "Libp2p => Connected({:?}) with errors: {:?}", peer_id, errors);
} else {
debug!(target: "sub-libp2p", "Libp2p => Connected({:?})", peer_id);
}
if let Some(metrics) = this.metrics.as_ref() {
let direction = match endpoint {
ConnectedPoint::Dialer { .. } => "out",
ConnectedPoint::Listener { .. } => "in",
};
metrics.connections_opened_total.with_label_values(&[direction]).inc();
if num_established.get() == 1 {
metrics.distinct_peers_connections_opened_total.inc();
}
}
},
Poll::Ready(SwarmEvent::ConnectionClosed {
peer_id,
cause,
endpoint,
num_established,
}) => {
debug!(target: "sub-libp2p", "Libp2p => Disconnected({:?}, {:?})", peer_id, cause);
if let Some(metrics) = this.metrics.as_ref() {
let direction = match endpoint {
ConnectedPoint::Dialer { .. } => "out",
ConnectedPoint::Listener { .. } => "in",
};
let reason = match cause {
Some(ConnectionError::IO(_)) => "transport-error",
Some(ConnectionError::Handler(EitherError::A(EitherError::A(
EitherError::B(EitherError::A(PingFailure::Timeout)),
)))) => "ping-timeout",
Some(ConnectionError::Handler(EitherError::A(EitherError::A(
EitherError::A(NotifsHandlerError::SyncNotificationsClogged),
)))) => "sync-notifications-clogged",
Some(ConnectionError::Handler(_)) => "protocol-error",
Some(ConnectionError::KeepAliveTimeout) => "keep-alive-timeout",
None => "actively-closed",
};
metrics
.connections_closed_total
.with_label_values(&[direction, reason])
.inc();
if num_established == 0 {
metrics.distinct_peers_connections_closed_total.inc();
}
}
},
Poll::Ready(SwarmEvent::NewListenAddr { address, .. }) => {
trace!(target: "sub-libp2p", "Libp2p => NewListenAddr({})", address);
if let Some(metrics) = this.metrics.as_ref() {
metrics.listeners_local_addresses.inc();
}
},
Poll::Ready(SwarmEvent::ExpiredListenAddr { address, .. }) => {
info!(target: "sub-libp2p", "📪 No longer listening on {}", address);
if let Some(metrics) = this.metrics.as_ref() {
metrics.listeners_local_addresses.dec();
}
},
Poll::Ready(SwarmEvent::OutgoingConnectionError { peer_id, error }) => {
if let Some(peer_id) = peer_id {
trace!(
target: "sub-libp2p",
"Libp2p => Failed to reach {:?}: {}",
peer_id, error,
);
if this.boot_node_ids.contains(&peer_id) {
if let DialError::WrongPeerId { obtained, endpoint } = &error {
if let ConnectedPoint::Dialer { address, role_override: _ } =
endpoint
{
warn!(
"💔 The bootnode you want to connect to at `{}` provided a different peer ID `{}` than the one you expect `{}`.",
address,
obtained,
peer_id,
);
}
}
}
}
if let Some(metrics) = this.metrics.as_ref() {
let reason = match error {
DialError::ConnectionLimit(_) => Some("limit-reached"),
DialError::InvalidPeerId(_) => Some("invalid-peer-id"),
DialError::Transport(_) | DialError::ConnectionIo(_) =>
Some("transport-error"),
DialError::Banned |
DialError::LocalPeerId |
DialError::NoAddresses |
DialError::DialPeerConditionFalse(_) |
DialError::WrongPeerId { .. } |
DialError::Aborted => None, };
if let Some(reason) = reason {
metrics
.pending_connections_errors_total
.with_label_values(&[reason])
.inc();
}
}
},
Poll::Ready(SwarmEvent::Dialing(peer_id)) => {
trace!(target: "sub-libp2p", "Libp2p => Dialing({:?})", peer_id)
},
Poll::Ready(SwarmEvent::IncomingConnection { local_addr, send_back_addr }) => {
trace!(target: "sub-libp2p", "Libp2p => IncomingConnection({},{}))",
local_addr, send_back_addr);
if let Some(metrics) = this.metrics.as_ref() {
metrics.incoming_connections_total.inc();
}
},
Poll::Ready(SwarmEvent::IncomingConnectionError {
local_addr,
send_back_addr,
error,
}) => {
debug!(
target: "sub-libp2p",
"Libp2p => IncomingConnectionError({},{}): {}",
local_addr, send_back_addr, error,
);
if let Some(metrics) = this.metrics.as_ref() {
let reason = match error {
PendingConnectionError::ConnectionLimit(_) => Some("limit-reached"),
PendingConnectionError::WrongPeerId { .. } => Some("invalid-peer-id"),
PendingConnectionError::Transport(_) |
PendingConnectionError::IO(_) => Some("transport-error"),
PendingConnectionError::Aborted => None, };
if let Some(reason) = reason {
metrics
.incoming_connections_errors_total
.with_label_values(&[reason])
.inc();
}
}
},
Poll::Ready(SwarmEvent::BannedPeer { peer_id, endpoint }) => {
debug!(
target: "sub-libp2p",
"Libp2p => BannedPeer({}). Connected via {:?}.",
peer_id, endpoint,
);
if let Some(metrics) = this.metrics.as_ref() {
metrics
.incoming_connections_errors_total
.with_label_values(&["banned"])
.inc();
}
},
Poll::Ready(SwarmEvent::ListenerClosed { reason, addresses, .. }) => {
if let Some(metrics) = this.metrics.as_ref() {
metrics.listeners_local_addresses.sub(addresses.len() as u64);
}
let addrs =
addresses.into_iter().map(|a| a.to_string()).collect::<Vec<_>>().join(", ");
match reason {
Ok(()) => error!(
target: "sub-libp2p",
"📪 Libp2p listener ({}) closed gracefully",
addrs
),
Err(e) => error!(
target: "sub-libp2p",
"📪 Libp2p listener ({}) closed: {}",
addrs, e
),
}
},
Poll::Ready(SwarmEvent::ListenerError { error, .. }) => {
debug!(target: "sub-libp2p", "Libp2p => ListenerError: {}", error);
if let Some(metrics) = this.metrics.as_ref() {
metrics.listeners_errors_total.inc();
}
},
};
}
let num_connected_peers =
this.network_service.behaviour_mut().user_protocol_mut().num_connected_peers();
this.num_connected.store(num_connected_peers, Ordering::Relaxed);
{
let external_addresses =
Swarm::<Behaviour<B, Client>>::external_addresses(&this.network_service)
.map(|r| &r.addr)
.cloned()
.collect();
*this.external_addresses.lock() = external_addresses;
}
let is_major_syncing = this
.network_service
.behaviour_mut()
.user_protocol_mut()
.sync_state()
.state
.is_major_syncing();
this.is_major_syncing.store(is_major_syncing, Ordering::Relaxed);
if let Some(metrics) = this.metrics.as_ref() {
if let Some(buckets) = this.network_service.behaviour_mut().num_entries_per_kbucket() {
for (lower_ilog2_bucket_bound, num_entries) in buckets {
metrics
.kbuckets_num_nodes
.with_label_values(&[&lower_ilog2_bucket_bound.to_string()])
.set(num_entries as u64);
}
}
if let Some(num_entries) = this.network_service.behaviour_mut().num_kademlia_records() {
metrics.kademlia_records_count.set(num_entries as u64);
}
if let Some(num_entries) =
this.network_service.behaviour_mut().kademlia_records_total_size()
{
metrics.kademlia_records_sizes_total.set(num_entries as u64);
}
metrics
.peerset_num_discovered
.set(this.network_service.behaviour_mut().user_protocol().num_discovered_peers()
as u64);
metrics.pending_connections.set(
Swarm::network_info(&this.network_service).connection_counters().num_pending()
as u64,
);
}
Poll::Pending
}
}
impl<B, H, Client> Unpin for NetworkWorker<B, H, Client>
where
B: BlockT + 'static,
H: ExHashT,
Client: HeaderBackend<B> + 'static,
{
}
fn ensure_addresses_consistent_with_transport<'a>(
addresses: impl Iterator<Item = &'a Multiaddr>,
transport: &TransportConfig,
) -> Result<(), Error> {
if matches!(transport, TransportConfig::MemoryOnly) {
let addresses: Vec<_> = addresses
.filter(|x| {
x.iter().any(|y| !matches!(y, libp2p::core::multiaddr::Protocol::Memory(_)))
})
.cloned()
.collect();
if !addresses.is_empty() {
return Err(Error::AddressesForAnotherTransport {
transport: transport.clone(),
addresses,
})
}
} else {
let addresses: Vec<_> = addresses
.filter(|x| x.iter().any(|y| matches!(y, libp2p::core::multiaddr::Protocol::Memory(_))))
.cloned()
.collect();
if !addresses.is_empty() {
return Err(Error::AddressesForAnotherTransport {
transport: transport.clone(),
addresses,
})
}
}
Ok(())
}