Remove more instances of futures01

bin/node-template/src/service.rs

@@ -11,7 +11,6 @@ use sc_executor::native_executor_instance;
 pub use sc_executor::NativeExecutor;
 use sp_consensus_aura::sr25519::{AuthorityPair as AuraPair};
 use grandpa::{self, FinalityProofProvider as GrandpaFinalityProofProvider};
-use futures::{FutureExt, compat::Future01CompatExt};
 
 // Our native executor instance.
 native_executor_instance!(

client/cli/Cargo.toml

@@ -16,7 +16,7 @@ time = "0.1.42"
 ansi_term = "0.12.1"
 lazy_static = "1.4.0"
 app_dirs = "1.2.1"
-tokio = "0.2"
+tokio = "0.2.10"
 futures = "0.3.1"
 fdlimit = "0.1.1"
 serde_json = "1.0.41"

client/consensus/aura/Cargo.toml

@@ -37,7 +37,5 @@ sc-network = { version = "0.8", path = "../../network" }
 sc-network-test = { version = "0.8.0", path = "../../network/test" }
 sc-service = { version = "0.8", path = "../../service" }
 substrate-test-runtime-client = { version = "2.0.0", path = "../../../test-utils/runtime/client" }
-tokio = "0.1.22"
 env_logger = "0.7.0"
 tempfile = "3.1.0"
-futures01 = { package = "futures", version = "0.1" }

client/consensus/aura/src/lib.rs

@@ -863,10 +863,10 @@ mod tests {
 	use sp_runtime::traits::{Block as BlockT, DigestFor};
 	use sc_network::config::ProtocolConfig;
 	use parking_lot::Mutex;
-	use tokio::runtime::current_thread;
 	use sp_keyring::sr25519::Keyring;
 	use sc_client::BlockchainEvents;
 	use sp_consensus_aura::sr25519::AuthorityPair;
+	use std::task::Poll;
 
 	type Error = sp_blockchain::Error;
 
@@ -987,8 +987,8 @@ mod tests {
 
 		let net = Arc::new(Mutex::new(net));
 		let mut import_notifications = Vec::new();
+		let mut aura_futures = Vec::new();
 
-		let mut runtime = current_thread::Runtime::new().unwrap();
 		let mut keystore_paths = Vec::new();
 		for (peer_id, key) in peers {
 			let mut net = net.lock();
@@ -1017,7 +1017,7 @@ mod tests {
 				&inherent_data_providers, slot_duration.get()
 			).expect("Registers aura inherent data provider");
 
-			let aura = start_aura::<_, _, _, _, _, AuthorityPair, _, _, _>(
+			aura_futures.push(start_aura::<_, _, _, _, _, AuthorityPair, _, _, _>(
 				slot_duration,
 				client.clone(),
 				select_chain,
@@ -1028,21 +1028,19 @@ mod tests {
 				false,
 				keystore,
 				sp_consensus::AlwaysCanAuthor,
-			)
-				.expect("Starts aura")
-				.unit_error()
-				.compat();
-
-			runtime.spawn(aura);
+			).expect("Starts aura"));
 		}
 
-		runtime.spawn(futures01::future::poll_fn(move || {
-			net.lock().poll();
-			Ok::<_, ()>(futures01::Async::NotReady::<()>)
-		}));
-
-		runtime.block_on(future::join_all(import_notifications)
-			.unit_error().compat()).unwrap();
+		futures::executor::block_on(future::select(
+			future::poll_fn(move |cx| {
+				net.lock().poll(cx);
+				Poll::<()>::Pending
+			}),
+			future::select(
+				future::join_all(aura_futures),
+				future::join_all(import_notifications)
+			)
+		));
 	}
 
 	#[test]

client/consensus/babe/Cargo.toml

@@ -47,10 +47,8 @@ sc-network-test = { version = "0.8.0", path = "../../network/test" }
 sc-service = { version = "0.8", path = "../../service" }
 substrate-test-runtime-client = { version = "2.0.0", path = "../../../test-utils/runtime/client" }
 sc-block-builder = { version = "0.8", path = "../../block-builder" }
-tokio = "0.1.22"
 env_logger = "0.7.0"
 tempfile = "3.1.0"
-futures01 = { package = "futures", version = "0.1" }
 
 [features]
 test-helpers = []

client/consensus/babe/src/lib.rs

@@ -276,7 +276,7 @@ pub struct BabeParams<B: BlockT, C, E, I, SO, SC, CAW> {
 	pub can_author_with: CAW,
 }
 
-/// Start the babe worker. The returned future should be run in a tokio runtime.
+/// Start the babe worker.
 pub fn start_babe<B, C, SC, E, I, SO, CAW, Error>(BabeParams {
 	keystore,
 	client,

client/consensus/babe/src/tests.rs

@@ -32,10 +32,9 @@ use sc_network_test::*;
 use sc_network_test::{Block as TestBlock, PeersClient};
 use sc_network::config::{BoxFinalityProofRequestBuilder, ProtocolConfig};
 use sp_runtime::{generic::DigestItem, traits::{Block as BlockT, DigestFor}};
-use tokio::runtime::current_thread;
 use sc_client_api::{BlockchainEvents, backend::TransactionFor};
 use log::debug;
-use std::{time::Duration, cell::RefCell};
+use std::{time::Duration, cell::RefCell, task::Poll};
 
 type Item = DigestItem<Hash>;
 
@@ -364,7 +363,7 @@ fn run_one_test(
 
 	let net = Arc::new(Mutex::new(net));
 	let mut import_notifications = Vec::new();
-	let mut runtime = current_thread::Runtime::new().unwrap();
+	let mut babe_futures = Vec::new();
 	let mut keystore_paths = Vec::new();
 
 	for (peer_id, seed) in peers {
@@ -409,7 +408,7 @@ fn run_one_test(
 		);
 
 
-		runtime.spawn(start_babe(BabeParams {
+		babe_futures.push(start_babe(BabeParams {
 			block_import: data.block_import.lock().take().expect("import set up during init"),
 			select_chain,
 			client,
@@ -420,16 +419,16 @@ fn run_one_test(
 			babe_link: data.link.clone(),
 			keystore,
 			can_author_with: sp_consensus::AlwaysCanAuthor,
-		}).expect("Starts babe").unit_error().compat());
+		}).expect("Starts babe"));
 	}
 
-	runtime.spawn(futures01::future::poll_fn(move || {
-		net.lock().poll();
-		Ok::<_, ()>(futures01::Async::NotReady::<()>)
-	}));
-
-	runtime.block_on(future::join_all(import_notifications)
-		.unit_error().compat()).unwrap();
+	futures::executor::block_on(future::select(
+		futures::future::poll_fn(move |cx| {
+			net.lock().poll(cx);
+			Poll::<()>::Pending
+		}),
+		future::select(future::join_all(import_notifications), future::join_all(babe_futures))
+	));
 }
 
 #[test]

client/finality-grandpa/Cargo.toml

@@ -39,7 +39,6 @@ substrate-test-runtime-client = { version = "2.0.0",  path = "../../test-utils/r
 sp-consensus-babe = { version = "0.8", path = "../../primitives/consensus/babe" }
 sp-state-machine = { version = "0.8", path = "../../primitives/state-machine" }
 env_logger = "0.7.0"
-tokio = "0.1.22"
+tokio = { version = "0.2", features = ["rt-core"] }
 tempfile = "3.1.0"
 sp-api = { version = "2.0.0", path = "../../primitives/api" }
-futures01 = { package = "futures", version = "0.1.29" }

client/finality-grandpa/src/communication/tests.rs

@@ -43,20 +43,12 @@ struct TestNetwork {
 	sender: mpsc::UnboundedSender<Event>,
 }
 
-impl TestNetwork {
-	fn event_stream_03(&self) -> Pin<Box<dyn futures::Stream<Item = NetworkEvent> + Send>> {
+impl sc_network_gossip::Network<Block> for TestNetwork {
+	fn event_stream(&self) -> Pin<Box<dyn Stream<Item = NetworkEvent> + Send>> {
 		let (tx, rx) = mpsc::unbounded();
 		let _ = self.sender.unbounded_send(Event::EventStream(tx));
 		Box::pin(rx)
 	}
-}
-
-impl sc_network_gossip::Network<Block> for TestNetwork {
-	fn event_stream(&self) -> Box<dyn futures01::Stream<Item = NetworkEvent, Error = ()> + Send> {
-		Box::new(
-			self.event_stream_03().map(Ok::<_, ()>).compat()
-		)
-	}
 
 	fn report_peer(&self, who: sc_network::PeerId, cost_benefit: sc_network::ReputationChange) {
 		let _ = self.sender.unbounded_send(Event::Report(who, cost_benefit));
@@ -155,7 +147,6 @@ fn voter_set_state() -> SharedVoterSetState<Block> {
 	set_state.into()
 }
 
-// needs to run in a tokio runtime.
 fn make_test_network(executor: &impl futures::task::Spawn) -> (
 	impl Future<Output = Tester>,
 	TestNetwork,

client/finality-grandpa/src/lib.rs

@@ -171,15 +171,6 @@ type CommunicationInH<Block, H> = finality_grandpa::voter::CommunicationIn<
 	AuthorityId,
 >;
 
-/// A global communication sink for commits. Not exposed publicly, used
-/// internally to simplify types in the communication layer.
-type CommunicationOut<Block> = finality_grandpa::voter::CommunicationOut<
-	<Block as BlockT>::Hash,
-	NumberFor<Block>,
-	AuthoritySignature,
-	AuthorityId,
->;
-
 /// Global communication sink for commits with the hash type not being derived
 /// from the block, useful for forcing the hash to some type (e.g. `H256`) when
 /// the compiler can't do the inference.