Update tokio to 0.2 again and remove TaskExecutors (#786)

* upgrade tokio again

* Remove WrappedExecutor

* switch to spawn_blocking

polkadot/validation/Cargo.toml

@@ -8,11 +8,10 @@ edition = "2018"
 futures = "0.3.1"
 futures-timer = "2.0"
 parking_lot = "0.9.0"
-tokio = { version = "0.2.4", features = ["rt-core"] }
+tokio = { version = "0.2.10", features = ["rt-core", "blocking"] }
 derive_more = "0.14.1"
 log = "0.4.8"
 exit-future = "0.2.0"
-tokio-executor = { version = "0.2.0-alpha.6", features = ["blocking"] }
 codec = { package = "parity-scale-codec", version = "1.1.0", default-features = false, features = ["derive"] }
 availability_store = { package = "polkadot-availability-store", path = "../availability-store" }
 parachain = { package = "polkadot-parachain", path = "../parachain" }

polkadot/validation/src/block_production.rs

@@ -222,11 +222,11 @@ impl<Client, TxPool, Backend> consensus::Proposer<Block> for Proposer<Client, Tx
 
 			Delay::new(enough_candidates).await;
 
-			tokio_executor::blocking::run(move || {
+			tokio::task::spawn_blocking(move || {
 				let proposed_candidates = data.table.proposed_set();
 				data.propose_with(proposed_candidates)
 			})
-				.await
+				.await?
 		}.boxed()
 	}
 }

polkadot/validation/src/validation_service/mod.rs

@@ -118,7 +118,7 @@ fn interval(duration: Duration) -> impl Stream<Item=()> + Send + Unpin {
 }
 
 /// A builder for the validation service.
-pub struct ServiceBuilder<C, N, P, SC> {
+pub struct ServiceBuilder<C, N, P, SC, SP> {
 	/// The underlying blockchain client.
 	pub client: Arc<P>,
 	/// A handle to the network object used to communicate.
@@ -126,7 +126,7 @@ pub struct ServiceBuilder<C, N, P, SC> {
 	/// A handle to the collator pool we are using.
 	pub collators: C,
 	/// A handle to a background executor.
-	pub task_executor: TaskExecutor,
+	pub spawner: SP,
 	/// A handle to the availability store.
 	pub availability_store: AvailabilityStore,
 	/// A chain selector for determining active leaves in the block-DAG.
@@ -137,7 +137,7 @@ pub struct ServiceBuilder<C, N, P, SC> {
 	pub max_block_data_size: Option<u64>,
 }
 
-impl<C, N, P, SC> ServiceBuilder<C, N, P, SC> where
+impl<C, N, P, SC, SP> ServiceBuilder<C, N, P, SC, SP> where
 	C: Collators + Send + Sync + Unpin + 'static,
 	C::Collation: Send + Unpin + 'static,
 	P: BlockchainEvents<Block> + BlockBody<Block>,
@@ -150,6 +150,7 @@ impl<C, N, P, SC> ServiceBuilder<C, N, P, SC> where
 	N::TableRouter: Send + 'static,
 	N::BuildTableRouter: Send + Unpin + 'static,
 	SC: SelectChain<Block> + 'static,
+	SP: Spawn + Send + 'static,
 	// Rust bug: https://github.com/rust-lang/rust/issues/24159
 	sp_api::StateBackendFor<P, Block>: sp_api::StateBackend<HasherFor<Block>>,
 {
@@ -171,7 +172,7 @@ impl<C, N, P, SC> ServiceBuilder<C, N, P, SC> where
 			client: self.client.clone(),
 			network: self.network,
 			collators: self.collators,
-			handle: self.task_executor,
+			spawner: self.spawner,
 			availability_store: self.availability_store,
 			live_instances: HashMap::new(),
 		};
@@ -247,15 +248,15 @@ fn signing_key(validators: &[ValidatorId], keystore: &KeyStorePtr) -> Option<Arc
 }
 
 /// Constructs parachain-agreement instances.
-pub(crate) struct ParachainValidationInstances<C, N, P> {
+pub(crate) struct ParachainValidationInstances<C, N, P, SP> {
 	/// The client instance.
 	client: Arc<P>,
 	/// The backing network handle.
 	network: N,
 	/// Parachain collators.
 	collators: C,
-	/// handle to remote task executor
-	handle: TaskExecutor,
+	/// handle to spawner
+	spawner: SP,
 	/// Store for extrinsic data.
 	availability_store: AvailabilityStore,
 	/// Live agreements. Maps relay chain parent hashes to attestation
@@ -263,7 +264,7 @@ pub(crate) struct ParachainValidationInstances<C, N, P> {
 	live_instances: HashMap<Hash, Arc<ValidationInstanceHandle>>,
 }
 
-impl<C, N, P> ParachainValidationInstances<C, N, P> where
+impl<C, N, P, SP> ParachainValidationInstances<C, N, P, SP> where
 	C: Collators + Send + Unpin + 'static,
 	N: Network,
 	P: ProvideRuntimeApi<Block> + HeaderBackend<Block> + BlockBody<Block> + Send + Sync + 'static,
@@ -271,6 +272,7 @@ impl<C, N, P> ParachainValidationInstances<C, N, P> where
 	C::Collation: Send + Unpin + 'static,
 	N::TableRouter: Send + 'static,
 	N::BuildTableRouter: Unpin + Send + 'static,
+	SP: Spawn + Send + 'static,
 	// Rust bug: https://github.com/rust-lang/rust/issues/24159
 	sp_api::StateBackendFor<P, Block>: sp_api::StateBackend<HasherFor<Block>>,
 {
@@ -448,7 +450,7 @@ impl<C, N, P> ParachainValidationInstances<C, N, P> where
 		let cancellable_work = select(exit, router).map(drop);
 
 		// spawn onto thread pool.
-		if self.handle.spawn(cancellable_work).is_err() {
+		if self.spawner.spawn(cancellable_work).is_err() {
 			error!("Failed to spawn cancellable work task");
 		}
 	}