Start removing last few instances of futures01

client/network-gossip/Cargo.toml

@@ -8,7 +8,6 @@ edition = "2018"
 
 [dependencies]
 log = "0.4.8"
-futures01 = { package = "futures", version = "0.1.29" }
 futures = { version = "0.3.1", features = ["compat"] }
 futures-timer = "0.4.0"
 lru = "0.1.2"

client/network-gossip/src/bridge.rs

@@ -20,7 +20,7 @@ use crate::state_machine::{ConsensusGossip, TopicNotification};
 use sc_network::message::generic::ConsensusMessage;
 use sc_network::{Event, ReputationChange};
 
-use futures::{prelude::*, channel::mpsc, compat::Compat01As03, task::SpawnExt as _};
+use futures::{prelude::*, channel::mpsc, task::SpawnExt as _};
 use libp2p::PeerId;
 use parking_lot::Mutex;
 use sp_runtime::{traits::Block as BlockT, ConsensusEngineId};
@@ -50,7 +50,7 @@ impl<B: BlockT> GossipEngine<B> {
 
 		// We grab the event stream before registering the notifications protocol, otherwise we
 		// might miss events.
-		let event_stream = network.event_stream();
+		let mut event_stream = network.event_stream();
 
 		network.register_notifications_protocol(engine_id);
 		state_machine.register_validator(&mut network, engine_id, validator);
@@ -89,8 +89,7 @@ impl<B: BlockT> GossipEngine<B> {
 		}
 
 		let res = executor.spawn(async move {
-			let mut stream = Compat01As03::new(event_stream);
-			while let Some(Ok(event)) = stream.next().await {
+			while let Some(event) = event_stream.next().await {
 				match event {
 					Event::NotificationStreamOpened { remote, engine_id: msg_engine_id, roles } => {
 						if msg_engine_id != engine_id {

client/network-gossip/src/lib.rs

@@ -62,6 +62,7 @@ use futures::prelude::*;
 use sc_network::{specialization::NetworkSpecialization, Event, ExHashT, NetworkService, PeerId, ReputationChange};
 use sp_runtime::{traits::Block as BlockT, ConsensusEngineId};
 use std::sync::Arc;
+use std::pin::Pin;
 
 mod bridge;
 mod state_machine;
@@ -70,7 +71,7 @@ mod validator;
 /// Abstraction over a network.
 pub trait Network<B: BlockT> {
 	/// Returns a stream of events representing what happens on the network.
-	fn event_stream(&self) -> Box<dyn futures01::Stream<Item = Event, Error = ()> + Send>;
+	fn event_stream(&self) -> Pin<Box<dyn Stream<Item = Event> + Send>>;
 
 	/// Adjust the reputation of a node.
 	fn report_peer(&self, peer_id: PeerId, reputation: ReputationChange);
@@ -97,8 +98,8 @@ pub trait Network<B: BlockT> {
 }
 
 impl<B: BlockT, S: NetworkSpecialization<B>, H: ExHashT> Network<B> for Arc<NetworkService<B, S, H>> {
-	fn event_stream(&self) -> Box<dyn futures01::Stream<Item = Event, Error = ()> + Send> {
-		Box::new(NetworkService::event_stream(self).map(|v| Ok::<_, ()>(v)).compat())
+	fn event_stream(&self) -> Pin<Box<dyn Stream<Item = Event> + Send>> {
+		Box::pin(NetworkService::event_stream(self))
 	}
 
 	fn report_peer(&self, peer_id: PeerId, reputation: ReputationChange) {

client/service/Cargo.toml

@@ -21,7 +21,7 @@ parking_lot = "0.9.0"
 lazy_static = "1.4.0"
 log = "0.4.8"
 slog = { version = "2.5.2", features = ["nested-values"] }
-tokio-executor = "0.1.8"
+tokio = { version = "0.2", features = ["rt-core"] }
 futures-timer = "2"
 exit-future = "0.2.0"
 serde = "1.0.101"

client/service/src/lib.rs

@@ -38,7 +38,7 @@ use parking_lot::Mutex;
 use sc_client::Client;
 use exit_future::Signal;
 use futures::{
-	Future, FutureExt, Stream, StreamExt, TryFutureExt,
+	Future, FutureExt, Stream, StreamExt,
 	future::select, channel::mpsc,
 	compat::*,
 	sink::SinkExt,
@@ -108,6 +108,8 @@ pub struct Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {
 	marker: PhantomData<TBl>,
 }
 
+impl<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Unpin for Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> {}
+
 /// Alias for a an implementation of `futures::future::Executor`.
 pub type TaskExecutor = Arc<dyn Spawn + Send + Sync>;
 
@@ -210,11 +212,11 @@ impl<TBl, TBackend, TExec, TRtApi, TSc, TNetSpec, TExPool, TOc> AbstractService
 	Service<TBl, Client<TBackend, TExec, TBl, TRtApi>, TSc, NetworkStatus<TBl>,
 		NetworkService<TBl, TNetSpec, TBl::Hash>, TExPool, TOc>
 where
-	TBl: BlockT + Unpin,
+	TBl: BlockT,
 	TBackend: 'static + sc_client_api::backend::Backend<TBl>,
 	TExec: 'static + sc_client::CallExecutor<TBl> + Send + Sync + Clone,
 	TRtApi: 'static + Send + Sync,
-	TSc: sp_consensus::SelectChain<TBl> + 'static + Clone + Send + Unpin,
+	TSc: sp_consensus::SelectChain<TBl> + 'static + Clone + Send,
 	TExPool: 'static + TransactionPool<Block = TBl>
 		+ TransactionPoolMaintainer<Block = TBl>,
 	TOc: 'static + Send + Sync,
@@ -304,7 +306,7 @@ where
 	}
 }
 
-impl<TBl: Unpin, TCl, TSc: Unpin, TNetStatus, TNet, TTxPool, TOc> Future for
+impl<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc> Future for
 	Service<TBl, TCl, TSc, TNetStatus, TNet, TTxPool, TOc>
 {
 	type Output = Result<(), Error>;
@@ -322,17 +324,7 @@ impl<TBl: Unpin, TCl, TSc: Unpin, TNetStatus, TNet, TTxPool, TOc> Future for
 		}
 
 		while let Poll::Ready(Some(task_to_spawn)) = Pin::new(&mut this.to_spawn_rx).poll_next(cx) {
-			// TODO: Update to tokio 0.2 when libp2p get switched to std futures (#4383)
-			let executor = tokio_executor::DefaultExecutor::current();
-			use futures01::future::Executor;
-			if let Err(err) = executor.execute(task_to_spawn.unit_error().compat()) {
-				debug!(
-					target: "service",
-					"Failed to spawn background task: {:?}; falling back to manual polling",
-					err
-				);
-				this.to_poll.push(Box::pin(err.into_future().compat().map(drop)));
-			}
+			tokio::spawn(task_to_spawn);
 		}
 
 		// Polling all the `to_poll` futures.