Instantiate environment with asynchronous API (paritytech#768)

* point to in-progress Substrate branch

* instantiate environment async

* Fix futures

* Bump runtime

* Fix collation tests

* point to polkadot-master again

* point to polkadot-master again

* update deps

Co-authored-by: Ashley <[email protected]>

cli/Cargo.toml

@@ -23,7 +23,7 @@ sc-client-db = { git = "https://github.com/paritytech/substrate", branch = "polk
 sc-executor = { git = "https://github.com/paritytech/substrate", branch = "polkadot-master" }
 service = { package = "polkadot-service", path = "../service", default-features = false }
 
-tokio = { version = "0.1.22", optional = true }
+tokio = { version = "0.2", features = ["rt-threaded"], optional = true }
 
 wasm-bindgen = { version = "0.2.57", optional = true }
 wasm-bindgen-futures = { version = "0.4.7", optional = true }

cli/src/lib.rs

@@ -24,9 +24,7 @@ mod chain_spec;
 mod browser;
 
 use chain_spec::ChainSpec;
-use futures::{
-	Future, FutureExt, TryFutureExt, future::select, channel::oneshot, compat::Future01CompatExt,
-};
+use futures::{Future, future::{select, Either}, channel::oneshot};
 #[cfg(feature = "cli")]
 use tokio::runtime::Runtime;
 use log::info;
@@ -217,33 +215,22 @@ pub fn run_until_exit(
 ) -> error::Result<()> {
 	let (exit_send, exit) = oneshot::channel();
 
-	let executor = runtime.executor();
 	let informant = sc_cli::informant::build(&service);
-	let future = select(exit, informant)
-		.map(|_| Ok(()))
-		.compat();
 
-	executor.spawn(future);
+	let handle = runtime.spawn(select(exit, informant));
 
 	// we eagerly drop the service so that the internal exit future is fired,
 	// but we need to keep holding a reference to the global telemetry guard
 	let _telemetry = service.telemetry();
 
-	let service_res = {
-		let service = service
-			.map_err(|err| error::Error::Service(err))
-			.compat();
-		let select = select(service, e)
-			.map(|_| Ok(()))
-			.compat();
-		runtime.block_on(select)
-	};
+	let service_res = runtime.block_on(select(service, e));
 
 	let _ = exit_send.send(());
 
-	use futures01::Future;
-	// TODO [andre]: timeout this future substrate/#1318
-	let _ = runtime.shutdown_on_idle().wait();
+	runtime.block_on(handle);
 
-	service_res
+	match service_res {
+		Either::Left((res, _)) => res.map_err(error::Error::Service),
+		Either::Right((_, _)) => Ok(())
+	}
 }

collator/Cargo.toml

@@ -21,7 +21,7 @@ polkadot-network = { path = "../network" }
 polkadot-validation = { path = "../validation" }
 polkadot-service = { path = "../service" }
 log = "0.4.8"
-tokio = "0.1.22"
+tokio = "0.2"
 futures-timer = "1.0"
 codec = { package = "parity-scale-codec", version = "1.1.0" }
 

collator/src/lib.rs

@@ -425,7 +425,7 @@ fn run_collator_node<S, E, P, Extrinsic>(
 						);
 
 						let exit = inner_exit_2.clone();
-						tokio::spawn(future::select(res.boxed(), exit).map(drop).map(|_| Ok(())).compat());
+						tokio::spawn(future::select(res.boxed(), exit).map(drop));
 					})
 				});
 
@@ -449,11 +449,11 @@ fn run_collator_node<S, E, P, Extrinsic>(
 					inner_exit.clone()
 				).map(drop);
 
-			tokio::spawn(future.map(|_| Ok(())).compat());
+			tokio::spawn(future);
 			future::ready(())
 		});
 
-	service.spawn_essential_task(work.map(|_| Ok::<_, ()>(())).compat());
+	service.spawn_essential_task(work);
 
 	polkadot_cli::run_until_exit(runtime, service, exit)
 }

runtime/kusama/src/lib.rs

@@ -77,7 +77,7 @@ pub const VERSION: RuntimeVersion = RuntimeVersion {
 	spec_name: create_runtime_str!("kusama"),
 	impl_name: create_runtime_str!("parity-kusama"),
 	authoring_version: 2,
-	spec_version: 1040,
+	spec_version: 1041,
 	impl_version: 2,
 	apis: RUNTIME_API_VERSIONS,
 };

service/src/lib.rs

@@ -18,7 +18,11 @@
 
 pub mod chain_spec;
 
-use futures::{FutureExt, TryFutureExt, task::{Spawn, SpawnError, FutureObj}};
+use futures::{
+	FutureExt, TryFutureExt,
+	task::{Spawn, SpawnError, FutureObj},
+	compat::Future01CompatExt,
+};
 use sc_client::LongestChain;
 use std::sync::Arc;
 use std::time::Duration;
@@ -455,9 +459,7 @@ pub fn new_full<Runtime, Dispatch, Extrinsic>(config: Configuration)
 				service.keystore(),
 				dht_event_stream,
 			);
-			let future01_authority_discovery = authority_discovery.map(|x| Ok(x)).compat();
-
-			service.spawn_task(future01_authority_discovery);
+			service.spawn_task(authority_discovery);
 		}
 	}
 
@@ -498,7 +500,9 @@ pub fn new_full<Runtime, Dispatch, Extrinsic>(config: Configuration)
 			executor: service.spawn_task_handle(),
 		};
 
-		service.spawn_essential_task(grandpa::run_grandpa_voter(grandpa_config)?);
+		service.spawn_essential_task(
+			grandpa::run_grandpa_voter(grandpa_config)?.compat().map(drop)
+		);
 	} else {
 		grandpa::setup_disabled_grandpa(
 			service.client(),

validation/src/lib.rs

@@ -569,23 +569,24 @@ impl<C, N, P, SC, TxPool, B> consensus::Environment<Block> for ProposerFactory<C
 	// Rust bug: https://github.com/rust-lang/rust/issues/24159
 	sp_api::StateBackendFor<P, Block>: sp_api::StateBackend<HasherFor<Block>> + Send,
 {
+	type CreateProposer = Pin<Box<
+		dyn Future<Output = Result<Self::Proposer, Self::Error>> + Send + Unpin + 'static
+	>>;
 	type Proposer = Proposer<P, TxPool, B>;
 	type Error = Error;
 
 	fn init(
 		&mut self,
 		parent_header: &Header,
-	) -> Result<Self::Proposer, Error> {
+	) -> Self::CreateProposer {
 		let parent_hash = parent_header.hash();
 		let parent_id = BlockId::hash(parent_hash);
 
-		let tracker = self.parachain_validation.get_or_instantiate(
+		let maybe_proposer = self.parachain_validation.get_or_instantiate(
 			parent_hash,
 			&self.keystore,
 			self.max_block_data_size,
-		)?;
-
-		Ok(Proposer {
+		).map(|tracker| Proposer {
 			client: self.parachain_validation.client.clone(),
 			tracker,
 			parent_hash,
@@ -594,7 +595,9 @@ impl<C, N, P, SC, TxPool, B> consensus::Environment<Block> for ProposerFactory<C
 			transaction_pool: self.transaction_pool.clone(),
 			slot_duration: self.babe_slot_duration,
 			backend: self.backend.clone(),
-		})
+		});
+
+		Box::pin(future::ready(maybe_proposer))
 	}
 }