Companion for #6726 (#1469)

* Companion for #6726

* Spaces

* 'Update substrate'

Co-authored-by: parity-processbot <>

polkadot/network/src/protocol/tests.rs

@@ -30,25 +30,12 @@ use av_store::{Store as AvailabilityStore, ErasureNetworking};
 use sc_network_gossip::TopicNotification;
 use sp_api::{ApiRef, ProvideRuntimeApi};
 use sp_runtime::traits::Block as BlockT;
-use sp_core::{crypto::Pair, traits::SpawnNamed};
+use sp_core::{crypto::Pair, testing::TaskExecutor};
 use sp_keyring::Sr25519Keyring;
 
-use futures::executor::{LocalPool, LocalSpawner};
+use futures::executor::LocalPool;
 use futures::task::{LocalSpawnExt, SpawnExt};
 
-#[derive(Clone)]
-struct Executor(LocalSpawner);
-
-impl SpawnNamed for Executor {
-	fn spawn(&self, _: &'static str, future: futures::future::BoxFuture<'static, ()>) {
-		self.0.spawn_local(future).unwrap();
-	}
-
-	fn spawn_blocking(&self, name: &'static str, future: futures::future::BoxFuture<'static, ()>) {
-		self.spawn(name, future);
-	}
-}
-
 #[derive(Default)]
 pub struct MockNetworkOps {
 	recorded: Mutex<Recorded>,
@@ -256,7 +243,7 @@ fn test_setup(config: Config) -> (
 		mock_gossip.clone(),
 		api.clone(),
 		worker_rx,
-		Executor(pool.spawner()),
+		TaskExecutor::new(),
 	);
 
 	let service = Service {

polkadot/network/test/src/block_import.rs

@@ -88,7 +88,7 @@ fn import_single_good_block_without_header_fails() {
 
 #[test]
 fn async_import_queue_drops() {
-	let executor = sp_core::testing::SpawnBlockingExecutor::new();
+	let executor = sp_core::testing::TaskExecutor::new();
 	// Perform this test multiple times since it exhibits non-deterministic behavior.
 	for _ in 0..100 {
 		let verifier = PassThroughVerifier::new(true);

polkadot/network/test/src/lib.rs

@@ -573,7 +573,7 @@ pub trait TestNetFactory: Sized {
 			Box::new(block_import.clone()),
 			justification_import,
 			finality_proof_import,
-			&sp_core::testing::SpawnBlockingExecutor::new(),
+			&sp_core::testing::TaskExecutor::new(),
 			None,
 		));
 
@@ -650,7 +650,7 @@ pub trait TestNetFactory: Sized {
 			Box::new(block_import.clone()),
 			justification_import,
 			finality_proof_import,
-			&sp_core::testing::SpawnBlockingExecutor::new(),
+			&sp_core::testing::TaskExecutor::new(),
 			None,
 		));
 

polkadot/node/core/backing/src/lib.rs

@@ -903,7 +903,7 @@ mod tests {
 	}
 
 	fn test_harness<T: Future<Output=()>>(keystore: KeyStorePtr, test: impl FnOnce(TestHarness) -> T) {
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 
 		let (context, virtual_overseer) = polkadot_subsystem::test_helpers::make_subsystem_context(pool.clone());
 

polkadot/node/network/bitfield-distribution/src/lib.rs

@@ -706,7 +706,7 @@ mod test {
 			signed_availability: signed.clone(),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) =
 			make_subsystem_context::<BitfieldDistributionMessage, _>(pool);
 
@@ -766,7 +766,7 @@ mod test {
 			signed_availability: signed.clone(),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) =
 			make_subsystem_context::<BitfieldDistributionMessage, _>(pool);
 
@@ -818,7 +818,7 @@ mod test {
 			signed_availability: signed_bitfield.clone(),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) =
 			make_subsystem_context::<BitfieldDistributionMessage, _>(pool);
 
@@ -915,7 +915,7 @@ mod test {
 			signed_availability: signed_bitfield.clone(),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) =
 			make_subsystem_context::<BitfieldDistributionMessage, _>(pool);
 
@@ -1052,7 +1052,7 @@ mod test {
 		// validator 0 key pair
 		let (mut state, _signing_context, _validator_pair) = state_with_view(view![], hash_a.clone());
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) =
 			make_subsystem_context::<BitfieldDistributionMessage, _>(pool);
 

polkadot/node/network/bridge/src/lib.rs

@@ -635,7 +635,7 @@ mod tests {
 	}
 
 	fn test_harness<T: Future<Output=()>>(test: impl FnOnce(TestHarness) -> T) {
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (network, network_handle) = new_test_network();
 		let (context, virtual_overseer) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 

polkadot/node/network/pov-distribution/src/lib.rs

@@ -619,7 +619,7 @@ mod tests {
 			our_view: View(vec![hash_a, hash_b]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 		let mut descriptor = CandidateDescriptor::default();
 		descriptor.pov_hash = pov_hash;
@@ -699,7 +699,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 		let mut descriptor = CandidateDescriptor::default();
 		descriptor.pov_hash = pov_hash;
@@ -777,7 +777,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -849,7 +849,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -937,7 +937,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1000,7 +1000,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1061,7 +1061,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1119,7 +1119,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1204,7 +1204,7 @@ mod tests {
 			our_view: View(vec![hash_a, hash_b]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1266,7 +1266,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1343,7 +1343,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {
@@ -1426,7 +1426,7 @@ mod tests {
 			our_view: View(vec![hash_a]),
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {

polkadot/node/network/statement-distribution/src/lib.rs

@@ -1212,7 +1212,7 @@ mod tests {
 			},
 		};
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 		let peer = PeerId::random();
 
@@ -1304,7 +1304,7 @@ mod tests {
 			(peer_c.clone(), peer_data_from_view(peer_c_view)),
 		].into_iter().collect();
 
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (mut ctx, mut handle) = polkadot_subsystem::test_helpers::make_subsystem_context(pool);
 
 		executor::block_on(async move {

polkadot/node/overseer/examples/minimal-example.rs

@@ -135,7 +135,7 @@ impl<C> Subsystem<C> for Subsystem2
 
 fn main() {
 	femme::with_level(femme::LevelFilter::Trace);
-	let spawner = sp_core::testing::SpawnBlockingExecutor::new();
+	let spawner = sp_core::testing::TaskExecutor::new();
 	futures::executor::block_on(async {
 		let timer_stream = stream::repeat(()).then(|_| async {
 			Delay::new(Duration::from_secs(1)).await;

polkadot/node/overseer/src/lib.rs

@@ -485,7 +485,7 @@ where
 	/// }
 	///
 	/// # fn main() { executor::block_on(async move {
-	/// let spawner = sp_core::testing::SpawnBlockingExecutor::new();
+	/// let spawner = sp_core::testing::TaskExecutor::new();
 	/// let all_subsystems = AllSubsystems {
 	///     candidate_validation: ValidationSubsystem,
 	///     candidate_backing: DummySubsystem,
@@ -1058,7 +1058,7 @@ mod tests {
 	// Checks that a minimal configuration of two jobs can run and exchange messages.
 	#[test]
 	fn overseer_works() {
-		let spawner = sp_core::testing::SpawnBlockingExecutor::new();
+		let spawner = sp_core::testing::TaskExecutor::new();
 
 		executor::block_on(async move {
 			let (s1_tx, mut s1_rx) = mpsc::channel(64);
@@ -1123,7 +1123,7 @@ mod tests {
 	// Should immediately conclude the overseer itself with an error.
 	#[test]
 	fn overseer_panics_on_subsystem_exit() {
-		let spawner = sp_core::testing::SpawnBlockingExecutor::new();
+		let spawner = sp_core::testing::TaskExecutor::new();
 
 		executor::block_on(async move {
 			let (s1_tx, _) = mpsc::channel(64);
@@ -1218,7 +1218,7 @@ mod tests {
 	// notifications on imported blocks triggers expected `StartWork` and `StopWork` heartbeats.
 	#[test]
 	fn overseer_start_stop_works() {
-		let spawner = sp_core::testing::SpawnBlockingExecutor::new();
+		let spawner = sp_core::testing::TaskExecutor::new();
 
 		executor::block_on(async move {
 			let first_block_hash = [1; 32].into();
@@ -1314,7 +1314,7 @@ mod tests {
 	// notifications on imported blocks triggers expected `StartWork` and `StopWork` heartbeats.
 	#[test]
 	fn overseer_finalize_works() {
-		let spawner = sp_core::testing::SpawnBlockingExecutor::new();
+		let spawner = sp_core::testing::TaskExecutor::new();
 
 		executor::block_on(async move {
 			let first_block_hash = [1; 32].into();

polkadot/node/subsystem/src/util.rs

@@ -891,7 +891,7 @@ mod tests {
 	type OverseerHandle = test_helpers::TestSubsystemContextHandle<CandidateSelectionMessage>;
 
 	fn test_harness<T: Future<Output=()>>(run_args: HashMap<Hash, Vec<FromJob>>, test: impl FnOnce(OverseerHandle, mpsc::Receiver<(Option<Hash>, JobsError<Error>)>) -> T) {
-		let pool = sp_core::testing::SpawnBlockingExecutor::new();
+		let pool = sp_core::testing::TaskExecutor::new();
 		let (context, overseer_handle) = make_subsystem_context(pool.clone());
 		let (err_tx, err_rx) = mpsc::channel(16);
 

polkadot/parachain/Cargo.toml

@@ -23,6 +23,7 @@ sc-executor = { git = "https://github.com/paritytech/substrate", branch = "maste
 sp-io = { git = "https://github.com/paritytech/substrate", branch = "master", optional = true }
 parking_lot = { version = "0.10.0", optional = true }
 log = { version = "0.4.8", optional = true }
+futures = { version = "0.3.4", optional = true }
 
 [target.'cfg(not(any(target_os = "android", target_os = "unknown")))'.dependencies]
 shared_memory = { version = "0.10.0", optional = true }
@@ -43,4 +44,5 @@ std = [
 	"sc-executor",
 	"sp-io",
 	"polkadot-core-primitives/std",
+	"futures",
 ]

polkadot/parachain/src/wasm_executor/mod.rs

@@ -23,8 +23,7 @@
 use std::any::{TypeId, Any};
 use crate::primitives::{ValidationParams, ValidationResult};
 use codec::{Decode, Encode};
-use sp_core::storage::ChildInfo;
-use sp_core::traits::CallInWasm;
+use sp_core::{storage::ChildInfo, traits::{CallInWasm, SpawnNamed}};
 use sp_externalities::Extensions;
 use sp_wasm_interface::HostFunctions as _;
 
@@ -119,10 +118,11 @@ pub fn validate_candidate(
 	validation_code: &[u8],
 	params: ValidationParams,
 	options: ExecutionMode<'_>,
+	spawner: impl SpawnNamed + 'static,
 ) -> Result<ValidationResult, Error> {
 	match options {
 		ExecutionMode::Local => {
-			validate_candidate_internal(validation_code, &params.encode())
+			validate_candidate_internal(validation_code, &params.encode(), spawner)
 		},
 		#[cfg(not(any(target_os = "android", target_os = "unknown")))]
 		ExecutionMode::Remote(pool) => {
@@ -154,9 +154,10 @@ type HostFunctions = sp_io::SubstrateHostFunctions;
 pub fn validate_candidate_internal(
 	validation_code: &[u8],
 	encoded_call_data: &[u8],
+	spawner: impl SpawnNamed + 'static,
 ) -> Result<ValidationResult, Error> {
 	let mut extensions = Extensions::new();
-	extensions.register(sp_core::traits::TaskExecutorExt(sp_core::tasks::executor()));
+	extensions.register(sp_core::traits::TaskExecutorExt::new(spawner));
 
 	let mut ext = ValidationExternalities(extensions);
 

polkadot/parachain/src/wasm_executor/validation_host.rs

@@ -19,11 +19,12 @@
 use std::{process, env, sync::Arc, sync::atomic};
 use codec::{Decode, Encode};
 use crate::primitives::{ValidationParams, ValidationResult};
-use super::{validate_candidate_internal, Error};
-use super::{MAX_CODE_MEM, MAX_RUNTIME_MEM};
+use super::{validate_candidate_internal, Error, MAX_CODE_MEM, MAX_RUNTIME_MEM};
 use shared_memory::{SharedMem, SharedMemConf, EventState, WriteLockable, EventWait, EventSet};
 use parking_lot::Mutex;
 use log::{debug, trace};
+use futures::executor::ThreadPool;
+use sp_core::traits::SpawnNamed;
 
 const WORKER_ARGS_TEST: &[&'static str] = &["--nocapture", "validation_worker"];
 /// CLI Argument to start in validation worker mode.
@@ -43,6 +44,25 @@ enum Event {
 	WorkerReady = 2,
 }
 
+#[derive(Clone)]
+struct TaskExecutor(ThreadPool);
+
+impl TaskExecutor {
+	fn new() -> Result<Self, String> {
+		ThreadPool::new().map_err(|e| e.to_string()).map(Self)
+	}
+}
+
+impl SpawnNamed for TaskExecutor {
+	fn spawn_blocking(&self, _: &'static str, future: futures::future::BoxFuture<'static, ()>) {
+		self.0.spawn_ok(future);
+	}
+
+	fn spawn(&self, _: &'static str, future: futures::future::BoxFuture<'static, ()>) {
+		self.0.spawn_ok(future);
+	}
+}
+
 /// A pool of hosts.
 #[derive(Clone)]
 pub struct ValidationPool {
@@ -92,6 +112,7 @@ pub fn run_worker(mem_id: &str) -> Result<(), String> {
 	};
 
 	let exit = Arc::new(atomic::AtomicBool::new(false));
+    let task_executor = TaskExecutor::new()?;
 	// spawn parent monitor thread
 	let watch_exit = exit.clone();
 	std::thread::spawn(move || {
@@ -139,7 +160,7 @@ pub fn run_worker(mem_id: &str) -> Result<(), String> {
 				let (call_data, _) = rest.split_at_mut(MAX_RUNTIME_MEM);
 				let (call_data, _) = call_data.split_at_mut(header.params_size as usize);
 
-				let result = validate_candidate_internal(code, call_data);
+				let result = validate_candidate_internal(code, call_data, task_executor.clone());
 				debug!("{} Candidate validated: {:?}", process::id(), result);
 
 				match result {

polkadot/parachain/test-parachains/Cargo.toml

@@ -14,6 +14,9 @@ adder = { package = "test-parachain-adder", path = "adder" }
 halt = { package = "test-parachain-halt", path = "halt" }
 code-upgrader = { package = "test-parachain-code-upgrader", path = "code-upgrader" }
 
+[dev-dependencies]
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
+
 [features]
 default = [ "std" ]
 std = [

polkadot/parachain/test-parachains/tests/adder/mod.rs

@@ -78,6 +78,7 @@ pub fn execute_good_on_parent() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	).unwrap();
 
 	let new_head = HeadData::decode(&mut &ret.head_data.0[..]).unwrap();
@@ -117,6 +118,7 @@ fn execute_good_chain_on_parent() {
 				code_upgrade_allowed: None,
 			},
 			parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+			sp_core::testing::TaskExecutor::new(),
 		).unwrap();
 
 		let new_head = HeadData::decode(&mut &ret.head_data.0[..]).unwrap();
@@ -157,5 +159,6 @@ fn execute_bad_on_parent() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	).unwrap_err();
 }

polkadot/parachain/test-parachains/tests/code_upgrader/mod.rs

@@ -50,6 +50,7 @@ pub fn execute_good_no_upgrade() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	).unwrap();
 
 	let new_head = HeadData::decode(&mut &ret.head_data.0[..]).unwrap();
@@ -86,6 +87,7 @@ pub fn execute_good_with_upgrade() {
 			code_upgrade_allowed: Some(20),
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	).unwrap();
 
 	let new_head = HeadData::decode(&mut &ret.head_data.0[..]).unwrap();
@@ -129,6 +131,7 @@ pub fn code_upgrade_not_allowed() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	).unwrap();
 }
 
@@ -159,6 +162,7 @@ pub fn applies_code_upgrade_after_delay() {
 				code_upgrade_allowed: Some(2),
 			},
 			parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+			sp_core::testing::TaskExecutor::new(),
 		).unwrap();
 
 		let new_head = HeadData::decode(&mut &ret.head_data.0[..]).unwrap();
@@ -194,6 +198,7 @@ pub fn applies_code_upgrade_after_delay() {
 				code_upgrade_allowed: None,
 			},
 			parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+			sp_core::testing::TaskExecutor::new(),
 		).unwrap();
 
 		let new_head = HeadData::decode(&mut &ret.head_data.0[..]).unwrap();

polkadot/parachain/test-parachains/tests/wasm_executor/mod.rs

@@ -37,6 +37,7 @@ fn terminates_on_timeout() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	);
 	match result {
 		Err(parachain::wasm_executor::Error::Timeout) => {},
@@ -66,6 +67,7 @@ fn parallel_execution() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool2),
+		sp_core::testing::TaskExecutor::new(),
 	).ok());
 	let _ = parachain::wasm_executor::validate_candidate(
 		halt::wasm_binary_unwrap(),
@@ -78,6 +80,7 @@ fn parallel_execution() {
 			code_upgrade_allowed: None,
 		},
 		parachain::wasm_executor::ExecutionMode::RemoteTest(&pool),
+		sp_core::testing::TaskExecutor::new(),
 	);
 	thread.join().unwrap();
 	// total time should be < 2 x EXECUTION_TIMEOUT_SEC

polkadot/runtime/test-runtime/client/src/lib.rs

@@ -295,7 +295,7 @@ pub fn new_light() -> (
 	let local_call_executor = LocalCallExecutor::new(
 		backend.clone(),
 		executor,
-		sp_core::tasks::executor(),
+		Box::new(sp_core::testing::TaskExecutor::new()),
 		Default::default()
 	);
 	let call_executor = LightExecutor::new(

polkadot/validation/src/collation.rs

@@ -29,6 +29,7 @@ use polkadot_erasure_coding as erasure;
 use sp_api::ProvideRuntimeApi;
 use futures::prelude::*;
 use log::debug;
+use primitives::traits::SpawnNamed;
 
 /// Encapsulates connections to collators and allows collation on any parachain.
 ///
@@ -64,6 +65,7 @@ pub async fn collation_fetch<C: Collators, P>(
 	client: Arc<P>,
 	max_block_data_size: Option<u64>,
 	n_validators: usize,
+	spawner: impl SpawnNamed + Clone + 'static,
 ) -> Result<(CollationInfo, crate::pipeline::FullOutput), C::Error>
 	where
 		P::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
@@ -82,6 +84,7 @@ pub async fn collation_fetch<C: Collators, P>(
 			&relay_parent,
 			max_block_data_size,
 			n_validators,
+			spawner.clone(),
 		);
 
 		match res {

polkadot/validation/src/pipeline.rs

@@ -32,6 +32,7 @@ use parachain::{
 use runtime_primitives::traits::{BlakeTwo256, Hash as HashT};
 use sp_api::ProvideRuntimeApi;
 use crate::Error;
+use primitives::traits::SpawnNamed;
 
 pub use parachain::wasm_executor::ValidationPool;
 
@@ -191,6 +192,7 @@ pub fn validate<'a>(
 	local_validation: &'a LocalValidationData,
 	global_validation: &'a GlobalValidationData,
 	validation_code: &ValidationCode,
+	spawner: impl SpawnNamed + 'static,
 ) -> Result<ValidatedCandidate<'a>, Error> {
 	if collation.head_data.0.len() > global_validation.max_head_data_size as _ {
 		return Err(Error::HeadDataTooLarge(
@@ -222,6 +224,7 @@ pub fn validate<'a>(
 		&validation_code.0,
 		params,
 		execution_mode,
+		spawner,
 	) {
 		Ok(result) => {
 			if result.head_data == collation.head_data {
@@ -277,6 +280,7 @@ pub fn full_output_validation_with_api<P>(
 	expected_relay_parent: &Hash,
 	max_block_data_size: Option<u64>,
 	n_validators: usize,
+	spawner: impl SpawnNamed + 'static,
 ) -> Result<FullOutput, Error> where
 	P: ProvideRuntimeApi<Block>,
 	P::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
@@ -302,6 +306,7 @@ pub fn full_output_validation_with_api<P>(
 				&local_validation,
 				&global_validation,
 				&validation_code,
+				spawner,
 			);
 
 			match res {

polkadot/validation/src/shared_table/mod.rs

@@ -312,6 +312,7 @@ impl<Fetch: Future + Unpin> ParachainWork<Fetch> {
 				&expected_relay_parent,
 				max_block_data_size,
 				n_validators,
+				primitives::testing::TaskExecutor::new(),
 			)?;
 
 			full_output.check_consistency(candidate)?;

polkadot/validation/src/validation_service/mod.rs

@@ -145,7 +145,7 @@ impl<C, N, P, SC, SP> ServiceBuilder<C, N, P, SC, SP> where
 	N::BuildTableRouter: Send + Unpin + 'static,
 	<N::TableRouter as TableRouter>::SendLocalCollation: Send,
 	SC: SelectChain<Block> + 'static,
-	SP: SpawnNamed + Send + 'static,
+	SP: SpawnNamed + Clone + 'static,
 	// Rust bug: https://github.com/rust-lang/rust/issues/24159
 	sp_api::StateBackendFor<P, Block>: sp_api::StateBackend<HashFor<Block>>,
 {
@@ -167,11 +167,15 @@ impl<C, N, P, SC, SP> ServiceBuilder<C, N, P, SC, SP> where
 		let mut parachain_validation = ParachainValidationInstances {
 			client: self.client.clone(),
 			network: self.network,
-			spawner: self.spawner,
+			spawner: self.spawner.clone(),
 			availability_store: self.availability_store,
 			live_instances: HashMap::new(),
 			validation_pool: validation_pool.clone(),
-			collation_fetch: DefaultCollationFetch(self.collators, validation_pool),
+			collation_fetch: DefaultCollationFetch {
+				collators: self.collators,
+				validation_pool,
+				spawner: self.spawner,
+			},
 		};
 
 		let client = self.client;
@@ -253,11 +257,17 @@ pub(crate) trait CollationFetch {
 }
 
 #[derive(Clone)]
-struct DefaultCollationFetch<C>(C, Option<ValidationPool>);
-impl<C> CollationFetch for DefaultCollationFetch<C>
+struct DefaultCollationFetch<C, S> {
+	collators: C,
+	validation_pool: Option<ValidationPool>,
+	spawner: S,
+}
+
+impl<C, S> CollationFetch for DefaultCollationFetch<C, S>
 	where
 		C: Collators + Send + Sync + Unpin + 'static,
 		C::Collation: Send + Unpin + 'static,
+		S: SpawnNamed + Clone + 'static,
 {
 	type Error = C::Error;
 
@@ -273,15 +283,15 @@ impl<C> CollationFetch for DefaultCollationFetch<C>
 			P::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
 			P: ProvideRuntimeApi<Block> + Send + Sync + 'static,
 	{
-		let DefaultCollationFetch(collators, validation_pool) = self;
 		crate::collation::collation_fetch(
-			validation_pool,
+			self.validation_pool,
 			parachain,
 			relay_parent,
-			collators,
+			self.collators,
 			client,
 			max_block_data_size,
 			n_validators,
+			self.spawner,
 		).boxed()
 	}
 }
@@ -553,7 +563,7 @@ mod tests {
 	use runtime_primitives::traits::Block as BlockT;
 	use std::pin::Pin;
 	use sp_keyring::sr25519::Keyring;
-	use primitives::testing::SpawnBlockingExecutor;
+	use primitives::testing::TaskExecutor;
 
 	/// Events fired while running mock implementations to follow execution.
 	enum Events {
@@ -714,7 +724,7 @@ mod tests {
 
 	#[test]
 	fn launch_work_is_executed_properly() {
-		let executor = SpawnBlockingExecutor::new();
+		let executor = TaskExecutor::new();
 		let keystore = keystore::Store::new_in_memory();
 
 		// Make sure `Bob` key is in the keystore, so this mocked node will be a parachain validator.
@@ -754,7 +764,7 @@ mod tests {
 
 	#[test]
 	fn router_is_built_on_relay_chain_validator() {
-		let executor = SpawnBlockingExecutor::new();
+		let executor = TaskExecutor::new();
 		let keystore = keystore::Store::new_in_memory();
 
 		// Make sure `Alice` key is in the keystore, so this mocked node will be a relay-chain validator.