Network crate cleanups (#3049)

* Remove useless internal messages * Remove NetworkService::disconnect_peer * Remove NetworkMsg altogether * Rename ProtocolMsg ServerToWorkerMsg * Remove useless code * Add example for parse_str_addr * Move parse_str_addr and ProtocolId to config * Don't reexport the content of config * Rework the imports * More reexports rework * Add documentation * Move finalization report to network future * Move on_block_imported to worker * get_value/put_value no longer locking * local_peer_id() no longer locks * Remove FetchFuture * Service imports cleanup * Produce the network state in the network task * Merge network task and RPC network task * Move network methods to NetworkWorker * Remove Arc peers system from network * add_reserved_peer now goes through the channel * Remove Mutex around network swarm * Remove the FnOnce alias traits * Replace is_offline with num_connected * Improve style of poll() * Fix network tests * Some doc in service module * Remove macro export * Minor doc changes * Remove the synchronized() method of the import queue * Line width * Line widths * Fix import queue tests * Fix CLI tests
2026-06-12 17:01:09 +00:00 · 2019-07-08 15:33:29 +02:00
parent 7df8e52cfe
commit 1e126eab2f
24 changed files with 598 additions and 814 deletions
@@ -21,8 +21,7 @@ use transaction_pool;
 use crate::chain_spec::ChainSpec;
 pub use client::ExecutionStrategies;
 pub use client_db::PruningMode;
-pub use network::ExtTransport;
-pub use network::config::{NetworkConfiguration, Roles};
+pub use network::config::{ExtTransport, NetworkConfiguration, Roles};
 use runtime_primitives::BuildStorage;
 use serde::{Serialize, de::DeserializeOwned};
 use target_info::Target;
@@ -51,8 +50,8 @@ pub struct Configuration<C, G: Serialize + DeserializeOwned + BuildStorage> {
 	pub database_cache_size: Option<u32>,
 	/// Size of internal state cache in Bytes
 	pub state_cache_size: usize,
-	/// Size in percent of cache size dedicated to child tries 
-	pub state_cache_child_ratio: Option<usize>, 
+	/// Size in percent of cache size dedicated to child tries
+	pub state_cache_child_ratio: Option<usize>,
 	/// Pruning settings.
 	pub pruning: PruningMode,
 	/// Additional key seeds.
@@ -36,6 +36,7 @@ use client::{BlockchainEvents, backend::Backend, runtime_api::BlockT};
 use exit_future::Signal;
 use futures::prelude::*;
 use keystore::Store as Keystore;
+use network::NetworkState;
 use log::{info, warn, debug, error};
 use parity_codec::{Encode, Decode};
 use primitives::{Pair, Public, crypto::TypedKey, ed25519};
@@ -76,7 +77,9 @@ pub struct Service<Components: components::Components> {
 	select_chain: Option<<Components as components::Components>::SelectChain>,
 	network: Arc<components::NetworkService<Components>>,
 	/// Sinks to propagate network status updates.
-	network_status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<NetworkStatus<ComponentBlock<Components>>>>>>,
+	network_status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<(
+		NetworkStatus<ComponentBlock<Components>>, NetworkState
+	)>>>>,
 	transaction_pool: Arc<TransactionPool<Components::TransactionPoolApi>>,
 	keystore: Option<Keystore>,
 	exit: ::exit_future::Exit,
@@ -236,7 +239,7 @@ impl<Components: components::Components> Service<Components> {
 					DEFAULT_PROTOCOL_ID
 				}
 			}.as_bytes();
-			network::ProtocolId::from(protocol_id_full)
+			network::config::ProtocolId::from(protocol_id_full)
 		};

 		let network_params = network::config::Params {
@@ -256,11 +259,6 @@ impl<Components: components::Components> Service<Components> {
 		let network = network_mut.service().clone();
 		let network_status_sinks = Arc::new(Mutex::new(Vec::new()));

-		let _ = to_spawn_tx.unbounded_send(Box::new(build_network_future(network_mut, network_status_sinks.clone())
-			.map_err(|_| ())
-			.select(exit.clone())
-			.then(|_| Ok(()))));
-
 		#[allow(deprecated)]
 		let offchain_storage = client.backend().offchain_storage();
 		let offchain_workers = match (config.offchain_worker, offchain_storage) {
@@ -279,7 +277,6 @@ impl<Components: components::Components> Service<Components> {

 		{
 			// block notifications
-			let network = Arc::downgrade(&network);
 			let txpool = Arc::downgrade(&transaction_pool);
 			let wclient = Arc::downgrade(&client);
 			let offchain = offchain_workers.as_ref().map(Arc::downgrade);
@@ -289,10 +286,6 @@ impl<Components: components::Components> Service<Components> {
 				.for_each(move |notification| {
 					let number = *notification.header.number();

-					if let Some(network) = network.upgrade() {
-						network.on_block_imported(notification.hash, notification.header);
-					}
-
 					if let (Some(txpool), Some(client)) = (txpool.upgrade(), wclient.upgrade()) {
 						Components::RuntimeServices::maintain_transaction_pool(
 							&BlockId::hash(notification.hash),
@@ -317,52 +310,6 @@ impl<Components: components::Components> Service<Components> {
 			let _ = to_spawn_tx.unbounded_send(Box::new(events));
 		}

-		{
-			// finality notifications
-			let network = Arc::downgrade(&network);
-
-			// A utility stream that drops all ready items and only returns the last one.
-			// This is used to only keep the last finality notification and avoid
-			// overloading the sync module with notifications.
-			struct MostRecentNotification<B: BlockT>(futures::stream::Fuse<FinalityNotifications<B>>);
-
-			impl<B: BlockT> Stream for MostRecentNotification<B> {
-				type Item = <FinalityNotifications<B> as Stream>::Item;
-				type Error = <FinalityNotifications<B> as Stream>::Error;
-
-				fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-					let mut last = None;
-					let last = loop {
-						match self.0.poll()? {
-							Async::Ready(Some(item)) => { last = Some(item) }
-							Async::Ready(None) => match last {
-								None => return Ok(Async::Ready(None)),
-								Some(last) => break last,
-							},
-							Async::NotReady => match last {
-								None => return Ok(Async::NotReady),
-								Some(last) => break last,
-							},
-						}
-					};
-
-					Ok(Async::Ready(Some(last)))
-				}
-			}
-
-			let events = MostRecentNotification(client.finality_notification_stream().fuse())
-				.for_each(move |notification| {
-					if let Some(network) = network.upgrade() {
-						network.on_block_finalized(notification.hash, notification.header);
-					}
-					Ok(())
-				})
-				.select(exit.clone())
-				.then(|_| Ok(()));
-
-			let _ = to_spawn_tx.unbounded_send(Box::new(events));
-		}
-
 		{
 			// extrinsic notifications
 			let network = Arc::downgrade(&network);
@@ -388,12 +335,11 @@ impl<Components: components::Components> Service<Components> {
 		// Periodically notify the telemetry.
 		let transaction_pool_ = transaction_pool.clone();
 		let client_ = client.clone();
-		let network_ = network.clone();
 		let mut sys = System::new();
 		let self_pid = get_current_pid().ok();
-		let (netstat_tx, netstat_rx) = mpsc::unbounded();
+		let (netstat_tx, netstat_rx) = mpsc::unbounded::<(NetworkStatus<ComponentBlock<Components>>, NetworkState)>();
 		network_status_sinks.lock().push(netstat_tx);
-		let tel_task = netstat_rx.for_each(move |net_status| {
+		let tel_task = netstat_rx.for_each(move |(net_status, network_state)| {
 			let info = client_.info();
 			let best_number = info.chain.best_number.saturated_into::<u64>();
 			let best_hash = info.chain.best_hash;
@@ -419,8 +365,6 @@ impl<Components: components::Components> Service<Components> {
 				} else { (0.0, 0) }
 			} else { (0.0, 0) };

-			let network_state = network_.network_state();
-
 			telemetry!(
 				SUBSTRATE_INFO;
 				"system.interval";
@@ -461,11 +405,17 @@ impl<Components: components::Components> Service<Components> {
 		};
 		let rpc_handlers = gen_handler();
 		let rpc = start_rpc_servers::<Components::Factory, _>(&config, gen_handler)?;
-		let _ = to_spawn_tx.unbounded_send(Box::new(build_system_rpc_handler::<Components>(
-			network.clone(),
+
+		let _ = to_spawn_tx.unbounded_send(Box::new(build_network_future::<Components, _, _>(
+			network_mut,
+			client.clone(),
+			network_status_sinks.clone(),
 			system_rpc_rx,
 			has_bootnodes
-		)));
+		)
+			.map_err(|_| ())
+			.select(exit.clone())
+			.then(|_| Ok(()))));

 		let telemetry_connection_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<()>>>> = Default::default();

@@ -597,7 +547,7 @@ impl<Components: components::Components> Service<Components> {
 	}

 	/// Returns a receiver that periodically receives a status of the network.
-	pub fn network_status(&self) -> mpsc::UnboundedReceiver<NetworkStatus<ComponentBlock<Components>>> {
+	pub fn network_status(&self) -> mpsc::UnboundedReceiver<(NetworkStatus<ComponentBlock<Components>>, NetworkState)> {
 		let (sink, stream) = mpsc::unbounded();
 		self.network_status_sinks.lock().push(sink);
 		stream
@@ -660,15 +610,67 @@ impl<Components> Executor<Box<dyn Future<Item = (), Error = ()> + Send>>
 /// Builds a never-ending future that continuously polls the network.
 ///
 /// The `status_sink` contain a list of senders to send a periodic network status to.
-fn build_network_future<B: BlockT, S: network::specialization::NetworkSpecialization<B>, H: network::ExHashT>(
-	mut network: network::NetworkWorker<B, S, H>,
-	status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<NetworkStatus<B>>>>>,
+fn build_network_future<
+	Components: components::Components,
+	S: network::specialization::NetworkSpecialization<ComponentBlock<Components>>,
+	H: network::ExHashT
+> (
+	mut network: network::NetworkWorker<ComponentBlock<Components>, S, H>,
+	client: Arc<ComponentClient<Components>>,
+	status_sinks: Arc<Mutex<Vec<mpsc::UnboundedSender<(NetworkStatus<ComponentBlock<Components>>, NetworkState)>>>>,
+	mut rpc_rx: mpsc::UnboundedReceiver<rpc::apis::system::Request<ComponentBlock<Components>>>,
+	should_have_peers: bool,
 ) -> impl Future<Item = (), Error = ()> {
 	// Interval at which we send status updates on the status stream.
 	const STATUS_INTERVAL: Duration = Duration::from_millis(5000);
 	let mut status_interval = tokio_timer::Interval::new_interval(STATUS_INTERVAL);

+	let mut imported_blocks_stream = client.import_notification_stream().fuse();
+	let mut finality_notification_stream = client.finality_notification_stream().fuse();
+
 	futures::future::poll_fn(move || {
+		// We poll `imported_blocks_stream`.
+		while let Ok(Async::Ready(Some(notification))) = imported_blocks_stream.poll() {
+			network.on_block_imported(notification.hash, notification.header);
+		}
+
+		// We poll `finality_notification_stream`, but we only take the last event.
+		let mut last = None;
+		while let Ok(Async::Ready(Some(item))) = finality_notification_stream.poll() {
+			last = Some(item);
+		}
+		if let Some(notification) = last {
+			network.on_block_finalized(notification.hash, notification.header);
+		}
+
+		// Poll the RPC requests and answer them.
+		while let Ok(Async::Ready(Some(request))) = rpc_rx.poll() {
+			match request {
+				rpc::apis::system::Request::Health(sender) => {
+					let _ = sender.send(rpc::apis::system::Health {
+						peers: network.peers_debug_info().len(),
+						is_syncing: network.service().is_major_syncing(),
+						should_have_peers,
+					});
+				},
+				rpc::apis::system::Request::Peers(sender) => {
+					let _ = sender.send(network.peers_debug_info().into_iter().map(|(peer_id, p)|
+						rpc::apis::system::PeerInfo {
+							peer_id: peer_id.to_base58(),
+							roles: format!("{:?}", p.roles),
+							protocol_version: p.protocol_version,
+							best_hash: p.best_hash,
+							best_number: p.best_number,
+						}
+					).collect());
+				}
+				rpc::apis::system::Request::NetworkState(sender) => {
+					let _ = sender.send(network.network_state());
+				}
+			};
+		}
+
+		// Interval report for the external API.
 		while let Ok(Async::Ready(_)) = status_interval.poll() {
 			let status = NetworkStatus {
 				sync_state: network.sync_state(),
@@ -679,10 +681,12 @@ fn build_network_future<B: BlockT, S: network::specialization::NetworkSpecializa
 				average_download_per_sec: network.average_download_per_sec(),
 				average_upload_per_sec: network.average_upload_per_sec(),
 			};
+			let state = network.network_state();

-			status_sinks.lock().retain(|sink| sink.unbounded_send(status.clone()).is_ok());
+			status_sinks.lock().retain(|sink| sink.unbounded_send((status.clone(), state.clone())).is_ok());
 		}

+		// Main network polling.
 		network.poll()
 			.map_err(|err| {
 				warn!(target: "service", "Error in network: {:?}", err);
@@ -866,39 +870,6 @@ impl<C: Components> network::TransactionPool<ComponentExHash<C>, ComponentBlock<
 	}
 }

-/// Builds a never-ending `Future` that answers the RPC requests coming on the receiver.
-fn build_system_rpc_handler<Components: components::Components>(
-	network: Arc<NetworkService<Components>>,
-	rx: mpsc::UnboundedReceiver<rpc::apis::system::Request<ComponentBlock<Components>>>,
-	should_have_peers: bool,
-) -> impl Future<Item = (), Error = ()> {
-	rx.for_each(move |request| {
-		match request {
-			rpc::apis::system::Request::Health(sender) => {
-				let _ = sender.send(rpc::apis::system::Health {
-					peers: network.peers_debug_info().len(),
-					is_syncing: network.is_major_syncing(),
-					should_have_peers,
-				});
-			},
-			rpc::apis::system::Request::Peers(sender) => {
-				let _ = sender.send(network.peers_debug_info().into_iter().map(|(peer_id, p)| rpc::apis::system::PeerInfo {
-					peer_id: peer_id.to_base58(),
-					roles: format!("{:?}", p.roles),
-					protocol_version: p.protocol_version,
-					best_hash: p.best_hash,
-					best_number: p.best_number,
-				}).collect());
-			}
-			rpc::apis::system::Request::NetworkState(sender) => {
-				let _ = sender.send(network.network_state());
-			}
-		};
-
-		Ok(())
-	})
-}
-
 /// Constructs a service factory with the given name that implements the `ServiceFactory` trait.
 /// The required parameters are required to be given in the exact order. Some parameters are followed
 /// by `{}` blocks. These blocks are required and used to initialize the given parameter.