Update the service to std futures (#4447)

* Switch service to futures03

* Fix tests

* Fix service test and cli

* Re-add Executor trait to SpawnTaskHandle

* Fix node-service

* Update babe

* Fix browser node

* Update aura

* Revert back to tokio-executor to fix runtime panic

* Add todo item

* Fix service tests again

* Timeout test futures

* Fix tests

* nits

* Fix service test

* Remove zstd patch

* Re-add futures01 to aura and babe tests as a dev-dep

* Change failing test to tee

* Fix node

* Upgrade tokio

* fix society

* Start switching grandpa to stable futures

* Revert "Start switching grandpa to stable futures"

This reverts commit 9c19763.

* Fix utils

* Revert substrate service test

* Revert gitlab

Co-authored-by: thiolliere <gui.thiolliere@gmail.com>

bin/node-template/Cargo.toml

@@ -11,10 +11,9 @@ path = "src/main.rs"
 
 [dependencies]
 futures = "0.3.1"
-futures01 = { package = "futures", version = "0.1.29" }
 ctrlc = { version = "3.1.3", features = ["termination"] }
 log = "0.4.8"
-tokio = "0.1.22"
+tokio = { version = "0.2", features = ["rt-threaded"] }
 parking_lot = "0.9.0"
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 trie-root = "0.15.2"

bin/node-template/src/cli.rs

@@ -1,5 +1,5 @@
 use crate::service;
-use futures::{future::{select, Map}, FutureExt, TryFutureExt, channel::oneshot, compat::Future01CompatExt};
+use futures::{future::{select, Map, Either}, FutureExt, channel::oneshot};
 use std::cell::RefCell;
 use tokio::runtime::Runtime;
 pub use sc_cli::{VersionInfo, IntoExit, error};
@@ -75,36 +75,23 @@ where
 
 	let informant = informant::build(&service);
 
-	let future = select(exit, informant)
-		.map(|_| Ok(()))
-		.compat();
-
-	runtime.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 exit = e.into_exit();
-		let service = service
-			.map_err(|err| error::Error::Service(err))
-			.compat();
-		let select = select(service, exit)
-			.map(|_| Ok(()))
-			.compat();
-		runtime.block_on(select)
-	};
+	let exit = e.into_exit();
+	let service_res = runtime.block_on(select(service, exit));
 
 	let _ = exit_send.send(());
 
-	// TODO [andre]: timeout this future #1318
-
-	use futures01::Future;
+	runtime.block_on(handle);
 
-	let _ = runtime.shutdown_on_idle().wait();
-
-	service_res
+	match service_res {
+		Either::Left((res, _)) => res.map_err(error::Error::Service),
+		Either::Right((_, _)) => Ok(())
+	}
 }
 
 // handles ctrl-c

bin/node-template/src/service.rs

@@ -12,6 +12,7 @@ pub use sc_executor::NativeExecutor;
 use sp_consensus_aura::sr25519::{AuthorityPair as AuraPair};
 use grandpa::{self, FinalityProofProvider as GrandpaFinalityProofProvider};
 use sc_basic_authority;
+use futures::{FutureExt, compat::Future01CompatExt};
 
 // Our native executor instance.
 native_executor_instance!(
@@ -163,7 +164,7 @@ pub fn new_full<C: Send + Default + 'static>(config: Configuration<C, GenesisCon
 				service.network(),
 				service.on_exit(),
 				service.spawn_task_handle(),
-			)?);
+			)?.compat().map(drop));
 		},
 		(true, false) => {
 			// start the full GRANDPA voter
@@ -180,7 +181,7 @@ pub fn new_full<C: Send + Default + 'static>(config: Configuration<C, GenesisCon
 
 			// the GRANDPA voter task is considered infallible, i.e.
 			// if it fails we take down the service with it.
-			service.spawn_essential_task(grandpa::run_grandpa_voter(voter_config)?);
+			service.spawn_essential_task(grandpa::run_grandpa_voter(voter_config)?.compat().map(drop));
 		},
 		(_, true) => {
 			grandpa::setup_disabled_grandpa(

bin/node/cli/Cargo.toml

@@ -25,7 +25,6 @@ crate-type = ["cdylib", "rlib"]
 # third-party dependencies
 codec = { package = "parity-scale-codec", version = "1.0.6" }
 serde = { version = "1.0.102", features = ["derive"] }
-futures01 = { package = "futures", version = "0.1.29" }
 futures = { version = "0.3.1", features = ["compat"] }
 hex-literal = "0.2.1"
 jsonrpc-core = "14.0.3"
@@ -81,7 +80,7 @@ node-primitives = { version = "2.0.0", path = "../primitives" }
 node-executor = { version = "2.0.0", path = "../executor" }
 
 # CLI-specific dependencies
-tokio = { version = "0.1.22", optional = true }
+tokio = { version = "0.2", features = ["rt-threaded"], optional = true }
 sc-cli = { version = "2.0.0", optional = true, path = "../../../client/cli" }
 ctrlc = { version = "3.1.3", features = ["termination"], optional = true }
 node-transaction-factory = { version = "2.0.0", optional = true, path = "../transaction-factory" }

bin/node/cli/src/cli.rs

@@ -15,7 +15,6 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 pub use sc_cli::VersionInfo;
-use tokio::prelude::Future;
 use tokio::runtime::{Builder as RuntimeBuilder, Runtime};
 use sc_cli::{IntoExit, NoCustom, SharedParams, ImportParams, error};
 use sc_service::{AbstractService, Roles as ServiceRoles, Configuration};
@@ -25,6 +24,7 @@ use sc_cli::{display_role, parse_and_prepare, GetSharedParams, ParseAndPrepare};
 use crate::{service, ChainSpec, load_spec};
 use crate::factory_impl::FactoryState;
 use node_transaction_factory::RuntimeAdapter;
+use futures::{channel::oneshot, future::{select, Either}};
 
 /// Custom subcommands.
 #[derive(Clone, Debug, StructOpt)]
@@ -105,7 +105,10 @@ pub fn run<I, T, E>(args: I, exit: E, version: sc_cli::VersionInfo) -> error::Re
 			info!("Chain specification: {}", config.chain_spec.name());
 			info!("Node name: {}", config.name);
 			info!("Roles: {}", display_role(&config));
-			let runtime = RuntimeBuilder::new().name_prefix("main-tokio-").build()
+			let runtime = RuntimeBuilder::new()
+				.thread_name("main-tokio-")
+				.threaded_scheduler()
+				.build()
 				.map_err(|e| format!("{:?}", e))?;
 			match config.roles {
 				ServiceRoles::LIGHT => run_until_exit(
@@ -172,37 +175,25 @@ where
 	T: AbstractService,
 	E: IntoExit,
 {
-	use futures::{FutureExt, TryFutureExt, channel::oneshot, future::select, compat::Future01CompatExt};
-
 	let (exit_send, exit) = oneshot::channel();
 
 	let informant = sc_cli::informant::build(&service);
 
-	let future = select(informant, exit)
-		.map(|_| Ok(()))
-		.compat();
-
-	runtime.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 exit = e.into_exit();
-		let service = service
-			.map_err(|err| error::Error::Service(err))
-			.compat();
-		let select = select(service, exit)
-			.map(|_| Ok(()))
-			.compat();
-		runtime.block_on(select)
-	};
+	let exit = e.into_exit();
+	let service_res = runtime.block_on(select(service, exit));
 
 	let _ = exit_send.send(());
 
-	// TODO [andre]: timeout this future #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(())
+	}
 }

bin/node/cli/src/service.rs

@@ -113,8 +113,8 @@ macro_rules! new_full_start {
 macro_rules! new_full {
 	($config:expr, $with_startup_data: expr) => {{
 		use futures::{
-			stream::StreamExt,
-			future::{FutureExt, TryFutureExt},
+			prelude::*,
+			compat::Future01CompatExt
 		};
 		use sc_network::Event;
 
@@ -191,9 +191,8 @@ macro_rules! new_full {
 				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);
 		}
 
 		// if the node isn't actively participating in consensus then it doesn't
@@ -223,7 +222,7 @@ macro_rules! new_full {
 					service.network(),
 					service.on_exit(),
 					service.spawn_task_handle(),
-				)?);
+				)?.compat().map(drop));
 			},
 			(true, false) => {
 				// start the full GRANDPA voter
@@ -239,7 +238,9 @@ macro_rules! new_full {
 				};
 				// the GRANDPA voter task is considered infallible, i.e.
 				// if it fails we take down the service with it.
-				service.spawn_essential_task(grandpa::run_grandpa_voter(grandpa_config)?);
+				service.spawn_essential_task(
+					grandpa::run_grandpa_voter(grandpa_config)?.compat().map(drop)
+				);
 			},
 			(_, true) => {
 				grandpa::setup_disabled_grandpa(

client/cli/Cargo.toml

@@ -16,8 +16,8 @@ time = "0.1.42"
 ansi_term = "0.12.1"
 lazy_static = "1.4.0"
 app_dirs = "1.2.1"
-tokio = "0.2.1"
-futures = { version = "0.3.1", features = ["compat"] }
+tokio = "0.2"
+futures = "0.3.1"
 fdlimit = "0.1.1"
 serde_json = "1.0.41"
 sp-panic-handler = { version = "2.0.0", path = "../../primitives/panic-handler" }

client/cli/src/informant.rs

@@ -17,7 +17,7 @@
 //! Console informant. Prints sync progress and block events. Runs on the calling thread.
 
 use sc_client_api::BlockchainEvents;
-use futures::{StreamExt, TryStreamExt, FutureExt, future, compat::Stream01CompatExt};
+use futures::prelude::*;
 use log::{info, warn, trace};
 use sp_runtime::traits::Header;
 use sc_service::AbstractService;
@@ -33,16 +33,15 @@ pub fn build(service: &impl AbstractService) -> impl futures::Future<Output = ()
 
 	let display_notifications = service
 		.network_status(Duration::from_millis(5000))
-		.compat()
-		.try_for_each(move |(net_status, _)| {
+		.for_each(move |(net_status, _)| {
 			let info = client.usage_info();
 			if let Some(ref usage) = info.usage {
 				trace!(target: "usage", "Usage statistics: {}", usage);
 			} else {
 				trace!(target: "usage", "Usage statistics not displayed as backend does not provide it")
 			}
 			display.display(&info, net_status);
-			future::ok(())
+			future::ready(())
 		});
 
 	let client = service.client();

client/cli/src/lib.rs

@@ -61,7 +61,7 @@ pub use traits::GetSharedParams;
 use app_dirs::{AppInfo, AppDataType};
 use log::info;
 use lazy_static::lazy_static;
-use futures::{Future, compat::Future01CompatExt, executor::block_on};
+use futures::{Future, executor::block_on};
 use sc_telemetry::TelemetryEndpoints;
 use sp_runtime::generic::BlockId;
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
@@ -426,8 +426,7 @@ impl<'a> ParseAndPrepareExport<'a> {
 		});
 
 		let mut export_fut = builder(config)?
-			.export_blocks(file, from.into(), to, json)
-			.compat();
+			.export_blocks(file, from.into(), to, json);
 		let fut = futures::future::poll_fn(|cx| {
 			if exit_recv.try_recv().is_ok() {
 				return Poll::Ready(Ok(()));
@@ -485,8 +484,7 @@ impl<'a> ParseAndPrepareImport<'a> {
 		});
 
 		let mut import_fut = builder(config)?
-			.import_blocks(file, false)
-			.compat();
+			.import_blocks(file, false);
 		let fut = futures::future::poll_fn(|cx| {
 			if exit_recv.try_recv().is_ok() {
 				return Poll::Ready(Ok(()));
@@ -537,8 +535,7 @@ impl<'a> CheckBlock<'a> {
 
 		let start = std::time::Instant::now();
 		let check = builder(config)?
-			.check_block(block_id)
-			.compat();
+			.check_block(block_id);
 		let mut runtime = tokio::runtime::Runtime::new().unwrap();
 		runtime.block_on(check)?;
 		println!("Completed in {} ms.", start.elapsed().as_millis());

client/consensus/aura/Cargo.toml

@@ -14,8 +14,7 @@ sc-client-api = { version = "2.0.0", path = "../../api" }
 codec = { package = "parity-scale-codec", version = "1.0.0" }
 sp-consensus = { version = "0.8", path = "../../../primitives/consensus/common" }
 derive_more = "0.99.2"
-futures = { version = "0.3.1", features = ["compat"] }
-futures01 = { package = "futures", version = "0.1" }
+futures = "0.3.1"
 futures-timer = "0.4.0"
 sp-inherents = { version = "2.0.0", path = "../../../primitives/inherents" }
 sc-keystore = { version = "2.0.0", path = "../../keystore" }
@@ -41,3 +40,4 @@ substrate-test-runtime-client = { version = "2.0.0", path = "../../../test-utils
 tokio = "0.1.22"
 env_logger = "0.7.0"
 tempfile = "3.1.0"
+futures01 = { package = "futures", version = "0.1" }