Various small improvements to service construction. (#6738)

* Remove service components and add build_network, build_offchain_workers etc * Improve transaction pool api * Remove commented out line * Add PartialComponents * Add BuildNetworkParams, documentation * Remove unused imports in tests * Apply suggestions from code review Co-authored-by: Nikolay Volf <nikvolf@gmail.com> * Remove unused imports in node-bench Co-authored-by: Nikolay Volf <nikvolf@gmail.com>
2026-04-26 16:57:58 +00:00 · 2020-07-28 15:21:33 +02:00
parent 97f400f0f0
commit 9220b646d2
17 changed files with 401 additions and 326 deletions
@@ -18,37 +18,35 @@

 use crate::{
 	NetworkStatus, NetworkState, error::Error, DEFAULT_PROTOCOL_ID, MallocSizeOfWasm,
+	TelemetryConnectionSinks, RpcHandlers, NetworkStatusSinks,
 	start_rpc_servers, build_network_future, TransactionPoolAdapter, TaskManager, SpawnTaskHandle,
 	status_sinks, metrics::MetricsService,
 	client::{light, Client, ClientConfig},
-	config::{Configuration, KeystoreConfig, PrometheusConfig, OffchainWorkerConfig},
+	config::{Configuration, KeystoreConfig, PrometheusConfig},
 };
 use sc_client_api::{
 	light::RemoteBlockchain, ForkBlocks, BadBlocks, UsageProvider, ExecutorProvider,
 };
-use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender, TracingUnboundedReceiver};
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender};
 use sc_chain_spec::get_extension;
 use sp_consensus::{
 	block_validation::{BlockAnnounceValidator, DefaultBlockAnnounceValidator, Chain},
 	import_queue::ImportQueue,
 };
-use futures::{
-	Future, FutureExt, StreamExt,
-	future::ready,
-};
+use futures::{FutureExt, StreamExt, future::ready};
 use jsonrpc_pubsub::manager::SubscriptionManager;
 use sc_keystore::Store as Keystore;
 use log::{info, warn, error};
 use sc_network::config::{Role, FinalityProofProvider, OnDemand, BoxFinalityProofRequestBuilder};
 use sc_network::NetworkService;
-use parking_lot::{Mutex, RwLock};
+use parking_lot::RwLock;
 use sp_runtime::generic::BlockId;
 use sp_runtime::traits::{
 	Block as BlockT, SaturatedConversion, HashFor, Zero, BlockIdTo,
 };
 use sp_api::{ProvideRuntimeApi, CallApiAt};
 use sc_executor::{NativeExecutor, NativeExecutionDispatch, RuntimeInfo};
-use std::{collections::HashMap, sync::Arc, pin::Pin};
+use std::{collections::HashMap, sync::Arc};
 use wasm_timer::SystemTime;
 use sc_telemetry::{telemetry, SUBSTRATE_INFO};
 use sp_transaction_pool::MaintainedTransactionPool;
@@ -63,7 +61,6 @@ use sc_client_api::{
 	execution_extensions::ExecutionExtensions
 };
 use sp_blockchain::{HeaderMetadata, HeaderBackend};
-use crate::{ServiceComponents, TelemetryOnConnectSinks, RpcHandlers, NetworkStatusSinks};

 /// A utility trait for building an RPC extension given a `DenyUnsafe` instance.
 /// This is useful since at service definition time we don't know whether the
@@ -371,7 +368,7 @@ pub fn new_client<E, Block, RA>(
 }

 /// Parameters to pass into `build`.
-pub struct ServiceParams<TBl: BlockT, TCl, TImpQu, TExPool, TRpc, Backend> {
+pub struct SpawnTasksParams<'a, TBl: BlockT, TCl, TExPool, TRpc, Backend> {
 	/// The service configuration.
 	pub config: Configuration,
 	/// A shared client returned by `new_full_parts`/`new_light_parts`.
@@ -379,17 +376,11 @@ pub struct ServiceParams<TBl: BlockT, TCl, TImpQu, TExPool, TRpc, Backend> {
 	/// A shared backend returned by `new_full_parts`/`new_light_parts`.
 	pub backend: Arc<Backend>,
 	/// A task manager returned by `new_full_parts`/`new_light_parts`.
-	pub task_manager: TaskManager,
+	pub task_manager: &'a mut TaskManager,
 	/// A shared keystore returned by `new_full_parts`/`new_light_parts`.
 	pub keystore: Arc<RwLock<Keystore>>,
 	/// An optional, shared data fetcher for light clients.
 	pub on_demand: Option<Arc<OnDemand<TBl>>>,
-	/// An import queue.
-	pub import_queue: TImpQu,
-	/// An optional finality proof request builder.
-	pub finality_proof_request_builder: Option<BoxFinalityProofRequestBuilder<TBl>>,
-	/// An optional, shared finality proof request provider.
-	pub finality_proof_provider: Option<Arc<dyn FinalityProofProvider<TBl>>>,
 	/// A shared transaction pool.
 	pub transaction_pool: Arc<TExPool>,
 	/// A RPC extension builder. Use `NoopRpcExtensionBuilder` if you just want to pass in the
@@ -397,15 +388,61 @@ pub struct ServiceParams<TBl: BlockT, TCl, TImpQu, TExPool, TRpc, Backend> {
 	pub rpc_extensions_builder: Box<dyn RpcExtensionBuilder<Output = TRpc> + Send>,
 	/// An optional, shared remote blockchain instance. Used for light clients.
 	pub remote_blockchain: Option<Arc<dyn RemoteBlockchain<TBl>>>,
-	/// A block annouce validator builder.
-	pub block_announce_validator_builder:
-		Option<Box<dyn FnOnce(Arc<TCl>) -> Box<dyn BlockAnnounceValidator<TBl> + Send> + Send>>,
+	/// A shared network instance.
+	pub network: Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
+	/// Sinks to propagate network status updates.
+	pub network_status_sinks: NetworkStatusSinks<TBl>,
+	/// A Sender for RPC requests.
+	pub system_rpc_tx: TracingUnboundedSender<sc_rpc::system::Request<TBl>>,
+	/// Shared Telemetry connection sinks,
+	pub telemetry_connection_sinks: TelemetryConnectionSinks,
 }

-/// Put together the components of a service from the parameters.
-pub fn build<TBl, TBackend, TImpQu, TExPool, TRpc, TCl>(
-	builder: ServiceParams<TBl, TCl, TImpQu, TExPool, TRpc, TBackend>,
-) -> Result<ServiceComponents<TBl, TBackend, TCl>, Error>
+/// Build a shared offchain workers instance.
+pub fn build_offchain_workers<TBl, TBackend, TCl>(
+	config: &Configuration,
+	backend: Arc<TBackend>,
+	spawn_handle: SpawnTaskHandle,
+	client: Arc<TCl>,
+	network: Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
+) -> Option<Arc<sc_offchain::OffchainWorkers<TCl, TBackend::OffchainStorage, TBl>>>
+	where
+		TBl: BlockT, TBackend: sc_client_api::Backend<TBl>,
+		<TBackend as sc_client_api::Backend<TBl>>::OffchainStorage: 'static,
+		TCl: Send + Sync + ProvideRuntimeApi<TBl> + BlockchainEvents<TBl> + 'static,
+		<TCl as ProvideRuntimeApi<TBl>>::Api: sc_offchain::OffchainWorkerApi<TBl>,
+{
+	let offchain_workers = match backend.offchain_storage() {
+		Some(db) => {
+			Some(Arc::new(sc_offchain::OffchainWorkers::new(client.clone(), db)))
+		},
+		None => {
+			warn!("Offchain workers disabled, due to lack of offchain storage support in backend.");
+			None
+		},
+	};
+
+	// Inform the offchain worker about new imported blocks
+	if let Some(offchain) = offchain_workers.clone() {
+		spawn_handle.spawn(
+			"offchain-notifications",
+			sc_offchain::notification_future(
+				config.role.is_authority(),
+				client.clone(),
+				offchain,
+				Clone::clone(&spawn_handle),
+				network.clone()
+			)
+		);
+	}
+
+	offchain_workers
+}
+
+/// Spawn the tasks that are required to run a node.
+pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
+	params: SpawnTasksParams<TBl, TCl, TExPool, TRpc, TBackend>,
+) -> Result<Arc<RpcHandlers>, Error>
 	where
 		TCl: ProvideRuntimeApi<TBl> + HeaderMetadata<TBl, Error=sp_blockchain::Error> + Chain<TBl> +
 		BlockBackend<TBl> + BlockIdTo<TBl, Error=sp_blockchain::Error> + ProofProvider<TBl> +
@@ -421,26 +458,23 @@ pub fn build<TBl, TBackend, TImpQu, TExPool, TRpc, TCl>(
 			sp_api::ApiExt<TBl, StateBackend = TBackend::State>,
 		TBl: BlockT,
 		TBackend: 'static + sc_client_api::backend::Backend<TBl> + Send,
-		TImpQu: 'static + ImportQueue<TBl>,
 		TExPool: MaintainedTransactionPool<Block=TBl, Hash = <TBl as BlockT>::Hash> +
 			MallocSizeOfWasm + 'static,
 		TRpc: sc_rpc::RpcExtension<sc_rpc::Metadata>
 {
-	let ServiceParams {
+	let SpawnTasksParams {
 		mut config,
-		mut task_manager,
+		task_manager,
 		client,
 		on_demand,
 		backend,
 		keystore,
-		import_queue,
-		finality_proof_request_builder,
-		finality_proof_provider,
 		transaction_pool,
 		rpc_extensions_builder,
 		remote_blockchain,
-		block_announce_validator_builder,
-	} = builder;
+		network, network_status_sinks, system_rpc_tx,
+		telemetry_connection_sinks,
+	} = params;

 	let chain_info = client.usage_info().chain;

@@ -458,57 +492,14 @@ pub fn build<TBl, TBackend, TImpQu, TExPool, TRpc, TCl>(
 		"best" => ?chain_info.best_hash
 	);

-	let (system_rpc_tx, system_rpc_rx) = tracing_unbounded("mpsc_system_rpc");
-
-	let (network, network_status_sinks, network_future) = build_network(
-		&config, client.clone(), transaction_pool.clone(), task_manager.spawn_handle(),
-		on_demand.clone(), block_announce_validator_builder, finality_proof_request_builder,
-		finality_proof_provider, system_rpc_rx, import_queue
-	)?;
-
 	let spawn_handle = task_manager.spawn_handle();

-	// The network worker is responsible for gathering all network messages and processing
-	// them. This is quite a heavy task, and at the time of the writing of this comment it
-	// frequently happens that this future takes several seconds or in some situations
-	// even more than a minute until it has processed its entire queue. This is clearly an
-	// issue, and ideally we would like to fix the network future to take as little time as
-	// possible, but we also take the extra harm-prevention measure to execute the networking
-	// future using `spawn_blocking`.
-	spawn_handle.spawn_blocking("network-worker", network_future);
-
-	let offchain_storage = backend.offchain_storage();
-	let offchain_workers = match (config.offchain_worker.clone(), offchain_storage.clone()) {
-		(OffchainWorkerConfig {enabled: true, .. }, Some(db)) => {
-			Some(Arc::new(sc_offchain::OffchainWorkers::new(client.clone(), db)))
-		},
-		(OffchainWorkerConfig {enabled: true, .. }, None) => {
-			warn!("Offchain workers disabled, due to lack of offchain storage support in backend.");
-			None
-		},
-		_ => None,
-	};
-
 	// Inform the tx pool about imported and finalized blocks.
 	spawn_handle.spawn(
 		"txpool-notifications",
 		sc_transaction_pool::notification_future(client.clone(), transaction_pool.clone()),
 	);

-	// Inform the offchain worker about new imported blocks
-	if let Some(offchain) = offchain_workers.clone() {
-		spawn_handle.spawn(
-			"offchain-notifications",
-			sc_offchain::notification_future(
-				config.role.is_authority(),
-				client.clone(),
-				offchain,
-				task_manager.spawn_handle(),
-				network.clone()
-			)
-		);
-	}
-
 	spawn_handle.spawn(
 		"on-transaction-imported",
 		transaction_notifications(transaction_pool.clone(), network.clone()),
@@ -545,14 +536,12 @@ pub fn build<TBl, TBackend, TImpQu, TExPool, TRpc, TCl>(
 	let gen_handler = |deny_unsafe: sc_rpc::DenyUnsafe| gen_handler(
 		deny_unsafe, &config, task_manager.spawn_handle(), client.clone(), transaction_pool.clone(),
 		keystore.clone(), on_demand.clone(), remote_blockchain.clone(), &*rpc_extensions_builder,
-		offchain_storage.clone(), system_rpc_tx.clone()
+		backend.offchain_storage(), system_rpc_tx.clone()
 	);
 	let rpc = start_rpc_servers(&config, gen_handler)?;
 	// This is used internally, so don't restrict access to unsafe RPC
 	let rpc_handlers = Arc::new(RpcHandlers(gen_handler(sc_rpc::DenyUnsafe::No)));

-	let telemetry_connection_sinks: Arc<Mutex<Vec<TracingUnboundedSender<()>>>> = Default::default();
-
 	// Telemetry
 	let telemetry = config.telemetry_endpoints.clone().and_then(|endpoints| {
 		if endpoints.is_empty() {
@@ -585,18 +574,14 @@ pub fn build<TBl, TBackend, TImpQu, TExPool, TRpc, TCl>(
 	// Spawn informant task
 	spawn_handle.spawn("informant", sc_informant::build(
 		client.clone(),
-		network_status_sinks.clone(),
+		network_status_sinks.clone().0,
 		transaction_pool.clone(),
 		config.informant_output_format,
 	));

 	task_manager.keep_alive((telemetry, config.base_path, rpc, rpc_handlers.clone()));

-	Ok(ServiceComponents {
-		task_manager, network, rpc_handlers, offchain_workers,
-		telemetry_on_connect_sinks: TelemetryOnConnectSinks(telemetry_connection_sinks),
-		network_status_sinks: NetworkStatusSinks::new(network_status_sinks),
-	})
+	Ok(rpc_handlers)
 }

 async fn transaction_notifications<TBl, TExPool>(
@@ -626,7 +611,7 @@ async fn telemetry_periodic_send<TBl, TExPool, TCl>(
 	client: Arc<TCl>,
 	transaction_pool: Arc<TExPool>,
 	mut metrics_service: MetricsService,
-	network_status_sinks: Arc<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>
+	network_status_sinks: NetworkStatusSinks<TBl>,
 )
 	where
 		TBl: BlockT,
@@ -634,7 +619,7 @@ async fn telemetry_periodic_send<TBl, TExPool, TCl>(
 		TExPool: MaintainedTransactionPool<Block=TBl, Hash = <TBl as BlockT>::Hash>,
 {
 	let (state_tx, state_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat1");
-	network_status_sinks.push(std::time::Duration::from_millis(5000), state_tx);
+	network_status_sinks.0.push(std::time::Duration::from_millis(5000), state_tx);
 	state_rx.for_each(move |(net_status, _)| {
 		let info = client.usage_info();
 		metrics_service.tick(
@@ -647,11 +632,11 @@ async fn telemetry_periodic_send<TBl, TExPool, TCl>(
 }

 async fn telemetry_periodic_network_state<TBl: BlockT>(
-	network_status_sinks: Arc<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>
+	network_status_sinks: NetworkStatusSinks<TBl>,
 ) {
 	// Periodically send the network state to the telemetry.
 	let (netstat_tx, netstat_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat2");
-	network_status_sinks.push(std::time::Duration::from_secs(30), netstat_tx);
+	network_status_sinks.0.push(std::time::Duration::from_secs(30), netstat_tx);
 	netstat_rx.for_each(move |(_, network_state)| {
 		telemetry!(
 			SUBSTRATE_INFO;
@@ -665,7 +650,7 @@ async fn telemetry_periodic_network_state<TBl: BlockT>(
 fn build_telemetry<TBl: BlockT>(
 	config: &mut Configuration,
 	endpoints: sc_telemetry::TelemetryEndpoints,
-	telemetry_connection_sinks: Arc<Mutex<Vec<TracingUnboundedSender<()>>>>,
+	telemetry_connection_sinks: TelemetryConnectionSinks,
 	network: Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
 	spawn_handle: SpawnTaskHandle,
 	genesis_hash: <TBl as BlockT>::Hash,
@@ -703,7 +688,7 @@ fn build_telemetry<TBl: BlockT>(
 					"network_id" => network_id.clone()
 				);

-				telemetry_connection_sinks.lock().retain(|sink| {
+				telemetry_connection_sinks.0.lock().retain(|sink| {
 					sink.unbounded_send(()).is_ok()
 				});
 				ready(())
@@ -805,24 +790,38 @@ fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 	))
 }

-fn build_network<TBl, TExPool, TImpQu, TCl>(
-	config: &Configuration,
-	client: Arc<TCl>,
-	transaction_pool: Arc<TExPool>,
-	spawn_handle: SpawnTaskHandle,
-	on_demand: Option<Arc<OnDemand<TBl>>>,
-	block_announce_validator_builder: Option<Box<
+/// Parameters to pass into `build_network`.
+pub struct BuildNetworkParams<'a, TBl: BlockT, TExPool, TImpQu, TCl> {
+	/// The service configuration.
+	pub config: &'a Configuration,
+	/// A shared client returned by `new_full_parts`/`new_light_parts`.
+	pub client: Arc<TCl>,
+	/// A shared transaction pool.
+	pub transaction_pool: Arc<TExPool>,
+	/// A handle for spawning tasks.
+	pub spawn_handle: SpawnTaskHandle,
+	/// An import queue.
+	pub import_queue: TImpQu,
+	/// An optional, shared data fetcher for light clients.
+	pub on_demand: Option<Arc<OnDemand<TBl>>>,
+	/// A block annouce validator builder.
+	pub block_announce_validator_builder: Option<Box<
 		dyn FnOnce(Arc<TCl>) -> Box<dyn BlockAnnounceValidator<TBl> + Send> + Send
 	>>,
-	finality_proof_request_builder: Option<BoxFinalityProofRequestBuilder<TBl>>,
-	finality_proof_provider: Option<Arc<dyn FinalityProofProvider<TBl>>>,
-	system_rpc_rx: TracingUnboundedReceiver<sc_rpc::system::Request<TBl>>,
-	import_queue: TImpQu
+	/// An optional finality proof request builder.
+	pub finality_proof_request_builder: Option<BoxFinalityProofRequestBuilder<TBl>>,
+	/// An optional, shared finality proof request provider.
+	pub finality_proof_provider: Option<Arc<dyn FinalityProofProvider<TBl>>>,
+}
+
+/// Build the network service, the network status sinks and an RPC sender.
+pub fn build_network<TBl, TExPool, TImpQu, TCl>(
+	params: BuildNetworkParams<TBl, TExPool, TImpQu, TCl>
 ) -> Result<
 	(
 		Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
-		Arc<status_sinks::StatusSinks<(NetworkStatus<TBl>, NetworkState)>>,
-		Pin<Box<dyn Future<Output = ()> + Send>>
+		NetworkStatusSinks<TBl>,
+		TracingUnboundedSender<sc_rpc::system::Request<TBl>>,
 	),
 	Error
 >
@@ -834,6 +833,11 @@ fn build_network<TBl, TExPool, TImpQu, TCl>(
 		TExPool: MaintainedTransactionPool<Block=TBl, Hash = <TBl as BlockT>::Hash> + 'static,
 		TImpQu: ImportQueue<TBl> + 'static,
 {
+	let BuildNetworkParams {
+		config, client, transaction_pool, spawn_handle, import_queue, on_demand,
+		block_announce_validator_builder, finality_proof_request_builder, finality_proof_provider,
+	} = params;
+
 	let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
 		imports_external_transactions: !matches!(config.role, Role::Light),
 		pool: transaction_pool,
@@ -862,6 +866,7 @@ fn build_network<TBl, TExPool, TImpQu, TCl>(
 	let network_params = sc_network::config::Params {
 		role: config.role.clone(),
 		executor: {
+			let spawn_handle = Clone::clone(&spawn_handle);
 			Some(Box::new(move |fut| {
 				spawn_handle.spawn("libp2p-node", fut);
 			}))
@@ -881,7 +886,9 @@ fn build_network<TBl, TExPool, TImpQu, TCl>(
 	let has_bootnodes = !network_params.network_config.boot_nodes.is_empty();
 	let network_mut = sc_network::NetworkWorker::new(network_params)?;
 	let network = network_mut.service().clone();
-	let network_status_sinks = Arc::new(status_sinks::StatusSinks::new());
+	let network_status_sinks = NetworkStatusSinks::new(Arc::new(status_sinks::StatusSinks::new()));
+
+	let (system_rpc_tx, system_rpc_rx) = tracing_unbounded("mpsc_system_rpc");

 	let future = build_network_future(
 		config.role.clone(),
@@ -891,7 +898,16 @@ fn build_network<TBl, TExPool, TImpQu, TCl>(
 		system_rpc_rx,
 		has_bootnodes,
 		config.announce_block,
-	).boxed();
+	);

-	Ok((network, network_status_sinks, future))
+	// The network worker is responsible for gathering all network messages and processing
+	// them. This is quite a heavy task, and at the time of the writing of this comment it
+	// frequently happens that this future takes several seconds or in some situations
+	// even more than a minute until it has processed its entire queue. This is clearly an
+	// issue, and ideally we would like to fix the network future to take as little time as
+	// possible, but we also take the extra harm-prevention measure to execute the networking
+	// future using `spawn_blocking`.
+	spawn_handle.spawn_blocking("network-worker", future);
+
+	Ok((network, network_status_sinks, system_rpc_tx))
 }