Struct sc_network::NetworkWorker

impl<B, H, Client> NetworkWorker<B, H, Client>where
 B: BlockT + 'static,
 H: ExHashT,
 Client: HeaderBackend + 'static,

pub fn new(params: Params<B, Client>) -> Result<Self, Error>

Creates the network service.

Returns a NetworkWorker that implements Future and must be regularly polled in order for the network processing to advance. From it, you can extract a NetworkService using worker.service(). The NetworkService can be shared through the codebase.

pub fn status(&self) -> NetworkStatus

High-level network status information.

Examples found in repository ?

src/service.rs (line 1381)

	fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context) -> Poll<Self::Output> {
		let this = &mut *self;

		// At the time of writing of this comment, due to a high volume of messages, the network
		// worker sometimes takes a long time to process the loop below. When that happens, the
		// rest of the polling is frozen. In order to avoid negative side-effects caused by this
		// freeze, a limit to the number of iterations is enforced below. If the limit is reached,
		// the task is interrupted then scheduled again.
		//
		// This allows for a more even distribution in the time taken by each sub-part of the
		// polling.
		let mut num_iterations = 0;
		loop {
			num_iterations += 1;
			if num_iterations >= 100 {
				cx.waker().wake_by_ref();
				break
			}

			// Process the next message coming from the `NetworkService`.
			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),
			}
		}

		// `num_iterations` serves the same purpose as in the previous loop.
		// See the previous loop for explanations.
		let mut num_iterations = 0;
		loop {
			num_iterations += 1;
			if num_iterations >= 1000 {
				cx.waker().wake_by_ref();
				break
			}

			// Process the next action coming from the network.
			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,
									) =>
									// `UnsupportedProtocols` is reported for every single
									// inbound request whenever a request with an unsupported
									// protocol is received. This is not reported in order to
									// avoid confusions.
										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);
					}

					// TODO: Notifications might have been lost as a result of the previous
					// connection being dropped, and as a result it would be preferable to notify
					// the users of this fact by simulating the substream being closed then
					// reopened.
					// The code below doesn't compile because `role` is unknown. Propagating the
					// handshake of the secondary connections is quite an invasive change and
					// would conflict with https://github.com/paritytech/substrate/issues/6403.
					// Considering that dropping notifications is generally regarded as
					// acceptable, this bug is at the moment intentionally left there and is
					// intended to be fixed at the same time as
					// https://github.com/paritytech/substrate/issues/6403.
					// this.event_streams.send(Event::NotificationStreamClosed {
					// remote,
					// protocol,
					// });
					// this.event_streams.send(Event::NotificationStreamOpened {
					// remote,
					// protocol,
					// role,
					// });
				},
				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)) => {
					// Ignored event from lower layers.
				},
				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();

						// `num_established` represents the number of *remaining* connections.
						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, // ignore them
						};
						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, // ignore it
						};

						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();

		// Update the variables shared with the `NetworkService`.
		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
	}

pub fn total_bytes_inbound(&self) -> u64

Returns the total number of bytes received so far.

Examples found in repository ?

src/service.rs (line 470)

	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_outbound(&self) -> u64

Returns the total number of bytes sent so far.

Examples found in repository ?

src/service.rs (line 471)

	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 num_connected_peers(&self) -> usize

Returns the number of peers we’re connected to.

Examples found in repository ?

src/service.rs (line 468)

	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 num_active_peers(&self) -> usize

Returns the number of peers we’re connected to and that are being queried.

Examples found in repository ?

src/service.rs (line 469)

	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 sync_state(&self) -> SyncStatus

Current global sync state.

Examples found in repository ?

src/service.rs (line 463)

	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 best_seen_block(&self) -> Option<NumberFor>

Target sync block number.

Examples found in repository ?

src/service.rs (line 466)

	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 num_sync_peers(&self) -> u32

Number of peers participating in syncing.

Examples found in repository ?

src/service.rs (line 467)

	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 num_queued_blocks(&self) -> u32

Number of blocks in the import queue.

pub fn num_downloaded_blocks(&self) -> usize

Returns the number of downloaded blocks.

pub fn num_sync_requests(&self) -> usize

Number of active sync requests.

pub fn add_known_address(&mut self, peer_id: PeerId, addr: Multiaddr)

Adds an address for a node.

pub fn service(&self) -> &Arc<NetworkService<B, H>>

Return a NetworkService that can be shared through the code base and can be used to manipulate the worker.

pub fn on_block_finalized(&mut self, hash: B::Hash, header: B::Header)

You must call this when a new block is finalized by the client.

pub fn new_best_block_imported(&mut self, hash: B::Hash, number: NumberFor)

Inform the network service about new best imported block.

pub fn local_peer_id(&self) -> &PeerId

Returns the local PeerId.

pub fn listen_addresses(&self) -> impl Iterator<Item = &Multiaddr>

Returns the list of addresses we are listening on.

Does NOT include a trailing /p2p/ with our PeerId.

pub fn network_state(&mut self) -> NetworkState

Get network state.

Note: Use this only for debugging. This API is unstable. There are warnings literally everywhere about this. Please don’t use this function to retrieve actual information.

Examples found in repository ?

src/service.rs (line 1384)

	fn poll(mut self: Pin<&mut Self>, cx: &mut std::task::Context) -> Poll<Self::Output> {
		let this = &mut *self;

		// At the time of writing of this comment, due to a high volume of messages, the network
		// worker sometimes takes a long time to process the loop below. When that happens, the
		// rest of the polling is frozen. In order to avoid negative side-effects caused by this
		// freeze, a limit to the number of iterations is enforced below. If the limit is reached,
		// the task is interrupted then scheduled again.
		//
		// This allows for a more even distribution in the time taken by each sub-part of the
		// polling.
		let mut num_iterations = 0;
		loop {
			num_iterations += 1;
			if num_iterations >= 100 {
				cx.waker().wake_by_ref();
				break
			}

			// Process the next message coming from the `NetworkService`.
			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),
			}
		}

		// `num_iterations` serves the same purpose as in the previous loop.
		// See the previous loop for explanations.
		let mut num_iterations = 0;
		loop {
			num_iterations += 1;
			if num_iterations >= 1000 {
				cx.waker().wake_by_ref();
				break
			}

			// Process the next action coming from the network.
			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,
									) =>
									// `UnsupportedProtocols` is reported for every single
									// inbound request whenever a request with an unsupported
									// protocol is received. This is not reported in order to
									// avoid confusions.
										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);
					}

					// TODO: Notifications might have been lost as a result of the previous
					// connection being dropped, and as a result it would be preferable to notify
					// the users of this fact by simulating the substream being closed then
					// reopened.
					// The code below doesn't compile because `role` is unknown. Propagating the
					// handshake of the secondary connections is quite an invasive change and
					// would conflict with https://github.com/paritytech/substrate/issues/6403.
					// Considering that dropping notifications is generally regarded as
					// acceptable, this bug is at the moment intentionally left there and is
					// intended to be fixed at the same time as
					// https://github.com/paritytech/substrate/issues/6403.
					// this.event_streams.send(Event::NotificationStreamClosed {
					// remote,
					// protocol,
					// });
					// this.event_streams.send(Event::NotificationStreamOpened {
					// remote,
					// protocol,
					// role,
					// });
				},
				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)) => {
					// Ignored event from lower layers.
				},
				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();

						// `num_established` represents the number of *remaining* connections.
						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, // ignore them
						};
						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, // ignore it
						};

						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();

		// Update the variables shared with the `NetworkService`.
		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
	}

pub fn peers_debug_info(&mut self) -> Vec<(PeerId, PeerInfo)> ⓘ

Get currently connected peers.

pub fn remove_reserved_peer(&self, peer: PeerId)

Removes a PeerId from the list of reserved peers.

pub fn add_reserved_peer(&self, peer: MultiaddrWithPeerId) -> Result<(), String>

Adds a PeerId and its Multiaddr as reserved.

pub fn reserved_peers(&self) -> impl Iterator<Item = &PeerId>

Returns the list of reserved peers.

Trait Implementations§

impl<B, H, Client> Future for NetworkWorker<B, H, Client>where
 B: BlockT + 'static,
 H: ExHashT,
 Client: HeaderBackend + 'static,

type Output = ()

The type of value produced on completion.

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>

Attempt to resolve the future to a final value, registering the current task for wakeup if the value is not yet available. Read more

impl<B, H, Client> Unpin for NetworkWorker<B, H, Client>where
 B: BlockT + 'static,
 H: ExHashT,
 Client: HeaderBackend + 'static,

Auto Trait Implementations§

impl<B, H, Client> !UnwindSafe for NetworkWorker<B, H, Client>

Blanket Implementations§

impl<T> Any for Twhere
T: 'static + ?Sized,

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more

impl<T> Any for Twhere
T: Any,

fn into_any(self: Box<T, Global>) -> Box<dyn Any + 'static, Global>

fn into_any_rc(self: Rc<T>) -> Rc<dyn Any + 'static>

fn type_name(&self) -> &'static str

impl<T> AnySync for Twhere
T: Any + Send + Sync,

fn into_any_arc(self: Arc<T>) -> Arc<dyn Any + Send + Sync + 'static>

impl<T> Borrow<T> for Twhere
T: ?Sized,

const: unstable · source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more

impl<T> BorrowMut<T> for Twhere
T: ?Sized,

const: unstable · source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more

impl<T> CheckedConversion for T

fn checked_from<T>(t: T) -> Option<Self>where
Self: TryFrom<T>,

Convert from a value of T into an equivalent instance of Option<Self>. Read more

fn checked_into<T>(self) -> Option<T>where
Self: TryInto<T>,

Consume self to return Some equivalent value of Option<T>. Read more

impl<T> Downcast for Twhere
T: Any,

fn into_any(self: Box<T, Global>) -> Box<dyn Any + 'static, Global>

Convert Box<dyn Trait> (where Trait: Downcast) to Box<dyn Any>. Box<dyn Any> can then be further downcast into Box<ConcreteType> where ConcreteType implements Trait.

fn into_any_rc(self: Rc<T>) -> Rc<dyn Any + 'static>

Convert Rc<Trait> (where Trait: Downcast) to Rc<Any>. Rc<Any> can then be further downcast into Rc<ConcreteType> where ConcreteType implements Trait.

fn as_any(&self) -> &(dyn Any + 'static)

Convert &Trait (where Trait: Downcast) to &Any. This is needed since Rust cannot generate &Any’s vtable from &Trait’s.

fn as_any_mut(&mut self) -> &mut (dyn Any + 'static)

Convert &mut Trait (where Trait: Downcast) to &Any. This is needed since Rust cannot generate &mut Any’s vtable from &mut Trait’s.

impl<T> DowncastSync for Twhere
T: Any + Send + Sync,

fn into_any_arc(self: Arc<T>) -> Arc<dyn Any + Send + Sync + 'static>

Convert Arc<Trait> (where Trait: Downcast) to Arc<Any>. Arc<Any> can then be further downcast into Arc<ConcreteType> where ConcreteType implements Trait.

impl<T> From<T> for T

const: unstable · source§

fn from(t: T) -> T

Returns the argument unchanged.

impl<T> FutureExt for Twhere
T: Future + ?Sized,

fn map<U, F>(self, f: F) -> Map<Self, F>where
F: FnOnce(Self::Output) -> U,
Self: Sized,

Map this future’s output to a different type, returning a new future of the resulting type. Read more

fn map_into(self) -> MapInto<Self, U>where
Self::Output: Into,
Self: Sized,

Map this future’s output to a different type, returning a new future of the resulting type. Read more

fn then<Fut, F>(self, f: F) -> Then<Self, Fut, F>where
 F: FnOnce(Self::Output) -> Fut,
 Fut: Future,
 Self: Sized,

Chain on a computation for when a future finished, passing the result of the future to the provided closure f. Read more

fn left_future(self) -> Either<Self, B>where
B: Future<Output = Self::Output>,
Self: Sized,

Wrap this future in an Either future, making it the left-hand variant of that Either. Read more

fn right_future<A>(self) -> Either<A, Self>where
A: Future<Output = Self::Output>,
Self: Sized,

Wrap this future in an Either future, making it the right-hand variant of that Either. Read more

fn into_stream(self) -> IntoStream<Self>where
Self: Sized,

Convert this future into a single element stream. Read more

fn flatten(self) -> Flatten<Self>where
Self::Output: Future,
Self: Sized,

Flatten the execution of this future when the output of this future is itself another future. Read more

fn flatten_stream(self) -> FlattenStream<Self>where
Self::Output: Stream,
Self: Sized,

Flatten the execution of this future when the successful result of this future is a stream. Read more

fn fuse(self) -> Fuse<Self>where
Self: Sized,

Fuse a future such that poll will never again be called once it has completed. This method can be used to turn any Future into a FusedFuture. Read more

fn inspect<F>(self, f: F) -> Inspect<Self, F>where
F: FnOnce(&Self::Output),
Self: Sized,

Do something with the output of a future before passing it on. Read more

fn catch_unwind(self) -> CatchUnwind<Self>where
Self: Sized + UnwindSafe,

Catches unwinding panics while polling the future. Read more

fn shared(self) -> Shared<Self>where
Self: Sized,
Self::Output: Clone,

Create a cloneable handle to this future where all handles will resolve to the same result. Read more

fn remote_handle(self) -> (Remote<Self>, RemoteHandle<Self::Output>)where
Self: Sized,

Turn this future into a future that yields () on completion and sends its output to another future on a separate task. Read more

fn boxed<'a>(
self
) -> Pin<Box<dyn Future<Output = Self::Output> + Send + 'a, Global>>where
Self: 'a + Sized + Send,

Wrap the future in a Box, pinning it. Read more

fn boxed_local<'a>(
self
) -> Pin<Box<dyn Future<Output = Self::Output> + 'a, Global>>where
Self: 'a + Sized,

Wrap the future in a Box, pinning it. Read more

fn unit_error(self) -> UnitError<Self>where
Self: Sized,

Turns a Future<Output = T> into a TryFuture<Ok = T, Error = ()>.

fn never_error(self) -> NeverError<Self>where
Self: Sized,

Turns a Future<Output = T> into a TryFuture<Ok = T, Error = Never>.

fn poll_unpin(&mut self, cx: &mut Context<'_>) -> Poll<Self::Output>where
Self: Unpin,

A convenience for calling Future::poll on Unpin future types.

fn now_or_never(self) -> Option<Self::Output>where
Self: Sized,

Evaluates and consumes the future, returning the resulting output if the future is ready after the first call to Future::poll. Read more

impl<F> FutureExt for Fwhere
F: Future + ?Sized,

fn poll(&mut self, cx: &mut Context<'_>) -> Poll<Self::Output>where
Self: Unpin,

A convenience for calling Future::poll() on !Unpin types.

fn or<F>(self, other: F) -> Or<Self, F>where
Self: Sized,
F: Future<Output = Self::Output>,

Returns the result of self or other future, preferring self if both are ready. Read more

fn race<F>(self, other: F) -> Race<Self, F>where
Self: Sized,
F: Future<Output = Self::Output>,

Returns the result of self or other future, with no preference if both are ready. Read more

fn catch_unwind(self) -> CatchUnwind<Self>where
Self: Sized + UnwindSafe,

Catches panics while polling the future. Read more

fn boxed<'a>(
self
) -> Pin<Box<dyn Future<Output = Self::Output> + Send + 'a, Global>>where
Self: 'a + Sized + Send,

Boxes the future and changes its type to dyn Future + Send + 'a. Read more

fn boxed_local<'a>(
self
) -> Pin<Box<dyn Future<Output = Self::Output> + 'a, Global>>where
Self: 'a + Sized,

Boxes the future and changes its type to dyn Future + 'a. Read more

impl<T> Instrument for T

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more

impl<T, U> Into for Twhere
U: From<T>,

const: unstable · source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

impl<F> IntoFuture for Fwhere
F: Future,

type Output = <F as Future>::Output

The output that the future will produce on completion.

type IntoFuture = F

Which kind of future are we turning this into?

fn into_future(self) -> <F as IntoFuture>::IntoFuture

Creates a future from a value. Read more

impl<T, Outer> IsWrappedBy<Outer> for Twhere
Outer: AsRef<T> + AsMut<T> + From<T>,
T: From<Outer>,

fn from_ref(outer: &Outer) -> &T

Get a reference to the inner from the outer.

fn from_mut(outer: &mut Outer) -> &mut T

Get a mutable reference to the inner from the outer.

impl<T> Pointable for T

const ALIGN: usize = mem::align_of::<T>()

The alignment of pointer.

type Init = T

The type for initializers.

unsafe fn init(init: <T as Pointable>::Init) -> usize

Initializes a with the given initializer. Read more

unsafe fn deref<'a>(ptr: usize) -> &'a T

Dereferences the given pointer. Read more

unsafe fn deref_mut<'a>(ptr: usize) -> &'a mut T

Mutably dereferences the given pointer. Read more

unsafe fn drop(ptr: usize)

Drops the object pointed to by the given pointer. Read more

impl<T> Same<T> for T

type Output = T

Should always be Self

impl<T> SaturatedConversion for T

fn saturated_from<T>(t: T) -> Selfwhere
Self: UniqueSaturatedFrom<T>,

Convert from a value of T into an equivalent instance of Self. Read more

fn saturated_into<T>(self) -> Twhere
Self: UniqueSaturatedInto<T>,

Consume self to return an equivalent value of T. Read more

impl<T, U> TryFrom for Twhere
U: Into<T>,

type Error = Infallible

The type returned in the event of a conversion error.

const: unstable · source§

fn try_from(value: U) -> Result<T, <T as TryFrom>::Error>

Performs the conversion.

impl<T, U> TryInto for Twhere
U: TryFrom<T>,

type Error = >::Error

The type returned in the event of a conversion error.

const: unstable · source§

fn try_into(self) -> Result<U, >::Error>

Performs the conversion.

impl<S, T> UncheckedInto<T> for Swhere
T: UncheckedFrom<S>,

fn unchecked_into(self) -> T

The counterpart to unchecked_from.

impl<T, S> UniqueSaturatedInto<T> for Swhere
T: Bounded,
S: TryInto<T>,

fn unique_saturated_into(self) -> T

Consume self to return an equivalent value of T.

impl<V, T> VZip<V> for Twhere
V: MultiLane<T>,

fn vzip(self) -> V

impl<T> WithSubscriber for T

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>where
S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more