Revert "Instantiate environment with asynchronous API (#768)"

This reverts commit 989db4b.

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.2", features = ["rt-threaded"], optional = true }
+tokio = { version = "0.1.22", optional = true }
 
 wasm-bindgen = { version = "0.2.57", optional = true }
 wasm-bindgen-futures = { version = "0.4.7", optional = true }

cli/src/lib.rs

@@ -24,7 +24,9 @@ mod chain_spec;
 mod browser;
 
 use chain_spec::ChainSpec;
-use futures::{Future, future::{select, Either}, channel::oneshot};
+use futures::{
+	Future, FutureExt, TryFutureExt, future::select, channel::oneshot,
+};
 #[cfg(feature = "cli")]
 use tokio::runtime::Runtime;
 use log::info;
@@ -215,22 +217,34 @@ 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();
 
-	let handle = runtime.spawn(select(exit, informant));
+	executor.spawn(future);
 
 	// 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 = runtime.block_on(select(service, e));
+	let service_res = {
+		let service = service
+			.map_err(|err| error::Error::Service(err));
+
+		let select = select(service, e)
+			.map(|_| Ok(()))
+			.compat();
+
+		runtime.block_on(select)
+	};
 
 	let _ = exit_send.send(());
 
-	runtime.block_on(handle);
+	use futures01::Future;
+	// TODO [andre]: timeout this future substrate/#1318
+	let _ = runtime.shutdown_on_idle().wait();
 
-	match service_res {
-		Either::Left((res, _)) => res.map_err(error::Error::Service),
-		Either::Right((_, _)) => Ok(())
-	}
+	service_res
 }

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

service/src/lib.rs

@@ -467,6 +467,7 @@ pub fn new_full<Runtime, Dispatch, Extrinsic>(config: Configuration)
 				service.keystore(),
 				dht_event_stream,
 			);
+
 			service.spawn_task(authority_discovery);
 		}
 	}
@@ -509,8 +510,8 @@ pub fn new_full<Runtime, Dispatch, Extrinsic>(config: Configuration)
 		};
 
 		service.spawn_essential_task(
-			grandpa::run_grandpa_voter(grandpa_config)?.compat().map(drop)
-		);
+			grandpa::run_grandpa_voter(grandpa_config)?.compat().map(|_| ())
+		)
 	} else {
 		grandpa::setup_disabled_grandpa(
 			service.client(),