Companion PR for #5647 (#1035)

collator/src/lib.rs

@@ -441,7 +441,7 @@ mod tests {
 		fn check_send<T: Send>(_: T) {}
 
 		let cli = Cli::from_iter(&["-dev"]);
-		let task_executor = Arc::new(|_| unimplemented!());
+		let task_executor = Arc::new(|_, _| unimplemented!());
 		let config = cli.create_configuration(&cli.run.base, task_executor).unwrap();
 
 		check_send(start_collator(

network/test/src/block_import.rs

@@ -91,7 +91,11 @@ fn async_import_queue_drops() {
 	// Perform this test multiple times since it exhibits non-deterministic behavior.
 	for _ in 0..100 {
 		let verifier = PassThroughVerifier(true);
-		let queue = BasicQueue::new(verifier, Box::new(polkadot_test_runtime_client::new()), None, None);
+
+		let threads_pool = futures::executor::ThreadPool::new().unwrap();
+		let spawner = |future| threads_pool.spawn_ok(future);
+
+		let queue = BasicQueue::new(verifier, Box::new(polkadot_test_runtime_client::new()), None, None, spawner);
 		drop(queue);
 	}
 }

network/test/src/lib.rs

@@ -570,11 +570,15 @@ pub trait TestNetFactory: Sized {
 		);
 		let verifier = VerifierAdapter::new(Arc::new(Mutex::new(Box::new(verifier) as Box<_>)));
 
+		let threads_pool = futures::executor::ThreadPool::new().unwrap();
+		let spawner = |future| threads_pool.spawn_ok(future);
+
 		let import_queue = Box::new(BasicQueue::new(
 			verifier.clone(),
 			Box::new(block_import.clone()),
 			justification_import,
 			finality_proof_import,
+			spawner,
 		));
 
 		let listen_addr = build_multiaddr![Memory(rand::random::<u64>())];
@@ -645,11 +649,15 @@ pub trait TestNetFactory: Sized {
 		);
 		let verifier = VerifierAdapter::new(Arc::new(Mutex::new(Box::new(verifier) as Box<_>)));
 
+		let threads_pool = futures::executor::ThreadPool::new().unwrap();
+		let spawner = |future| threads_pool.spawn_ok(future);
+
 		let import_queue = Box::new(BasicQueue::new(
 			verifier.clone(),
 			Box::new(block_import.clone()),
 			justification_import,
 			finality_proof_import,
+			spawner,
 		));
 
 		let listen_addr = build_multiaddr![Memory(rand::random::<u64>())];

service/src/lib.rs

@@ -163,7 +163,7 @@ macro_rules! new_full_start {
 				let pool = sc_transaction_pool::BasicPool::new(config, std::sync::Arc::new(pool_api), prometheus_registry);
 				Ok(pool)
 			})?
-			.with_import_queue(|config, client, mut select_chain, _| {
+			.with_import_queue(|config, client, mut select_chain, _, spawn_task_handle| {
 				let select_chain = select_chain.take()
 					.ok_or_else(|| service::Error::SelectChainRequired)?;
 
@@ -189,13 +189,15 @@ macro_rules! new_full_start {
 					client.clone(),
 				)?;
 
+				let spawner = |future| spawn_task_handle.spawn_blocking("import-queue-worker", future);
 				let import_queue = babe::import_queue(
 					babe_link.clone(),
 					block_import.clone(),
 					Some(Box::new(justification_import)),
 					None,
 					client,
 					inherent_data_providers.clone(),
+					spawner,
 				)?;
 
 				import_setup = Some((block_import, grandpa_link, babe_link));
@@ -508,7 +510,7 @@ macro_rules! new_light {
 				);
 				Ok(pool)
 			})?
-			.with_import_queue_and_fprb(|_config, client, backend, fetcher, _select_chain, _| {
+			.with_import_queue_and_fprb(|_config, client, backend, fetcher, _select_chain, _, spawn_task_handle| {
 				let fetch_checker = fetcher
 					.map(|fetcher| fetcher.checker().clone())
 					.ok_or_else(|| "Trying to start light import queue without active fetch checker")?;
@@ -526,6 +528,7 @@ macro_rules! new_light {
 					client.clone(),
 				)?;
 
+				let spawner = |future| spawn_task_handle.spawn_blocking("importe-queue-worker", future);
 				// FIXME: pruning task isn't started since light client doesn't do `AuthoritySetup`.
 				let import_queue = babe::import_queue(
 					babe_link,
@@ -534,6 +537,7 @@ macro_rules! new_light {
 					Some(Box::new(finality_proof_import)),
 					client,
 					inherent_data_providers.clone(),
+					spawner,
 				)?;
 
 				Ok((import_queue, finality_proof_request_builder))