slot-based-collator: Move spawning of the futures (#6561)

Move spawning of the slot-based collator into the `run` function. Also
the tasks are being spawned as blocking task and not just as normal
tasks.

---------

Co-authored-by: GitHub Action <[email protected]>

cumulus/client/consensus/aura/src/collators/slot_based/mod.rs

@@ -28,6 +28,7 @@
 //! during the relay chain block. After the block is built, the block builder task sends it to
 //! the collation task which compresses it and submits it to the collation-generation subsystem.
 
+use self::{block_builder_task::run_block_builder, collation_task::run_collation_task};
 use codec::Codec;
 use consensus_common::ParachainCandidate;
 use cumulus_client_collator::service::ServiceInterface as CollatorServiceInterface;
@@ -36,32 +37,28 @@ use cumulus_client_consensus_proposer::ProposerInterface;
 use cumulus_primitives_aura::AuraUnincludedSegmentApi;
 use cumulus_primitives_core::GetCoreSelectorApi;
 use cumulus_relay_chain_interface::RelayChainInterface;
+use futures::FutureExt;
 use polkadot_primitives::{
 	CollatorPair, CoreIndex, Hash as RelayHash, Id as ParaId, ValidationCodeHash,
 };
-
 use sc_client_api::{backend::AuxStore, BlockBackend, BlockOf, UsageProvider};
 use sc_consensus::BlockImport;
 use sc_utils::mpsc::tracing_unbounded;
-
 use sp_api::ProvideRuntimeApi;
 use sp_application_crypto::AppPublic;
 use sp_blockchain::HeaderBackend;
 use sp_consensus_aura::AuraApi;
-use sp_core::crypto::Pair;
+use sp_core::{crypto::Pair, traits::SpawnNamed};
 use sp_inherents::CreateInherentDataProviders;
 use sp_keystore::KeystorePtr;
 use sp_runtime::traits::{Block as BlockT, Member};
-
 use std::{sync::Arc, time::Duration};
 
-use self::{block_builder_task::run_block_builder, collation_task::run_collation_task};
-
 mod block_builder_task;
 mod collation_task;
 
 /// Parameters for [`run`].
-pub struct Params<BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS> {
+pub struct Params<BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS, Spawner> {
 	/// Inherent data providers. Only non-consensus inherent data should be provided, i.e.
 	/// the timestamp, slot, and paras inherents should be omitted, as they are set by this
 	/// collator.
@@ -93,13 +90,30 @@ pub struct Params<BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS> {
 	/// Drift slots by a fixed duration. This can be used to create more preferrable authoring
 	/// timings.
 	pub slot_drift: Duration,
+	/// Spawner for spawning futures.
+	pub spawner: Spawner,
 }
 
 /// Run aura-based block building and collation task.
-pub fn run<Block, P, BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS>(
-	params: Params<BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS>,
-) -> (impl futures::Future<Output = ()>, impl futures::Future<Output = ()>)
-where
+pub fn run<Block, P, BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS, Spawner>(
+	Params {
+		create_inherent_data_providers,
+		block_import,
+		para_client,
+		para_backend,
+		relay_client,
+		code_hash_provider,
+		keystore,
+		collator_key,
+		para_id,
+		proposer,
+		collator_service,
+		authoring_duration,
+		reinitialize,
+		slot_drift,
+		spawner,
+	}: Params<BI, CIDP, Client, Backend, RClient, CHP, Proposer, CS, Spawner>,
+) where
 	Block: BlockT,
 	Client: ProvideRuntimeApi<Block>
 		+ BlockOf
@@ -123,39 +137,49 @@ where
 	P: Pair + 'static,
 	P::Public: AppPublic + Member + Codec,
 	P::Signature: TryFrom<Vec<u8>> + Member + Codec,
+	Spawner: SpawnNamed,
 {
 	let (tx, rx) = tracing_unbounded("mpsc_builder_to_collator", 100);
 	let collator_task_params = collation_task::Params {
-		relay_client: params.relay_client.clone(),
-		collator_key: params.collator_key,
-		para_id: params.para_id,
-		reinitialize: params.reinitialize,
-		collator_service: params.collator_service.clone(),
+		relay_client: relay_client.clone(),
+		collator_key,
+		para_id,
+		reinitialize,
+		collator_service: collator_service.clone(),
 		collator_receiver: rx,
 	};
 
 	let collation_task_fut = run_collation_task::<Block, _, _>(collator_task_params);
 
 	let block_builder_params = block_builder_task::BuilderTaskParams {
-		create_inherent_data_providers: params.create_inherent_data_providers,
-		block_import: params.block_import,
-		para_client: params.para_client,
-		para_backend: params.para_backend,
-		relay_client: params.relay_client,
-		code_hash_provider: params.code_hash_provider,
-		keystore: params.keystore,
-		para_id: params.para_id,
-		proposer: params.proposer,
-		collator_service: params.collator_service,
-		authoring_duration: params.authoring_duration,
+		create_inherent_data_providers,
+		block_import,
+		para_client,
+		para_backend,
+		relay_client,
+		code_hash_provider,
+		keystore,
+		para_id,
+		proposer,
+		collator_service,
+		authoring_duration,
 		collator_sender: tx,
-		slot_drift: params.slot_drift,
+		slot_drift,
 	};
 
 	let block_builder_fut =
 		run_block_builder::<Block, P, _, _, _, _, _, _, _, _>(block_builder_params);
 
-	(collation_task_fut, block_builder_fut)
+	spawner.spawn_blocking(
+		"slot-based-block-builder",
+		Some("slot-based-collator"),
+		block_builder_fut.boxed(),
+	);
+	spawner.spawn_blocking(
+		"slot-based-collation",
+		Some("slot-based-collator"),
+		collation_task_fut.boxed(),
+	);
 }
 
 /// Message to be sent from the block builder to the collation task.

cumulus/polkadot-omni-node/lib/src/nodes/aura.rs

@@ -54,6 +54,7 @@ use sc_service::{Configuration, Error, TaskManager};
 use sc_telemetry::TelemetryHandle;
 use sc_transaction_pool::TransactionPoolHandle;
 use sp_api::ProvideRuntimeApi;
+use sp_core::traits::SpawnNamed;
 use sp_inherents::CreateInherentDataProviders;
 use sp_keystore::KeystorePtr;
 use sp_runtime::{
@@ -242,7 +243,7 @@ where
 	AuraId: AuraIdT + Sync,
 {
 	#[docify::export_content]
-	fn launch_slot_based_collator<CIDP, CHP, Proposer, CS>(
+	fn launch_slot_based_collator<CIDP, CHP, Proposer, CS, Spawner>(
 		params: SlotBasedParams<
 			ParachainBlockImport<Block, RuntimeApi>,
 			CIDP,
@@ -252,28 +253,17 @@ where
 			CHP,
 			Proposer,
 			CS,
+			Spawner,
 		>,
-		task_manager: &TaskManager,
 	) where
 		CIDP: CreateInherentDataProviders<Block, ()> + 'static,
 		CIDP::InherentDataProviders: Send,
 		CHP: cumulus_client_consensus_common::ValidationCodeHashProvider<Hash> + Send + 'static,
 		Proposer: ProposerInterface<Block> + Send + Sync + 'static,
 		CS: CollatorServiceInterface<Block> + Send + Sync + Clone + 'static,
+		Spawner: SpawnNamed,
 	{
-		let (collation_future, block_builder_future) =
-			slot_based::run::<Block, <AuraId as AppCrypto>::Pair, _, _, _, _, _, _, _, _>(params);
-
-		task_manager.spawn_essential_handle().spawn(
-			"collation-task",
-			Some("parachain-block-authoring"),
-			collation_future,
-		);
-		task_manager.spawn_essential_handle().spawn(
-			"block-builder-task",
-			Some("parachain-block-authoring"),
-			block_builder_future,
-		);
+		slot_based::run::<Block, <AuraId as AppCrypto>::Pair, _, _, _, _, _, _, _, _, _>(params);
 	}
 }
 
@@ -335,11 +325,12 @@ where
 			authoring_duration: Duration::from_millis(2000),
 			reinitialize: false,
 			slot_drift: Duration::from_secs(1),
+			spawner: task_manager.spawn_handle(),
 		};
 
 		// We have a separate function only to be able to use `docify::export` on this piece of
 		// code.
-		Self::launch_slot_based_collator(params, task_manager);
+		Self::launch_slot_based_collator(params);
 
 		Ok(())
 	}

cumulus/test/service/src/lib.rs

@@ -497,20 +497,10 @@ where
 					authoring_duration: Duration::from_millis(2000),
 					reinitialize: false,
 					slot_drift: Duration::from_secs(1),
+					spawner: task_manager.spawn_handle(),
 				};
 
-				let (collation_future, block_builder_future) =
-					slot_based::run::<Block, AuthorityPair, _, _, _, _, _, _, _, _>(params);
-				task_manager.spawn_essential_handle().spawn(
-					"collation-task",
-					None,
-					collation_future,
-				);
-				task_manager.spawn_essential_handle().spawn(
-					"block-builder-task",
-					None,
-					block_builder_future,
-				);
+				slot_based::run::<Block, AuthorityPair, _, _, _, _, _, _, _, _, _>(params);
 			} else {
 				tracing::info!(target: LOG_TARGET, "Starting block authoring with lookahead collator.");
 				let params = AuraParams {

prdoc/pr_6561.prdoc

@@ -0,0 +1,11 @@
+title: 'slot-based-collator: Move spawning of the futures'
+doc:
+- audience: Node Dev
+  description: "Move spawning of the slot-based collator into the `run` function.\
+    \ Also the tasks are being spawned as blocking task and not just as normal tasks.\r\
+    \n"
+crates:
+- name: cumulus-client-consensus-aura
+  bump: major
+- name: polkadot-omni-node-lib
+  bump: major