authority-discovery: futures 03 Future

core/authority-discovery/Cargo.toml

@@ -14,18 +14,17 @@ bytes = "0.4.12"
 client = { package = "substrate-client", path = "../../core/client" }
 codec = { package = "parity-scale-codec", default-features = false, version = "1.0.3" }
 derive_more = "0.15.0"
-futures = "0.1.29"
+futures-preview = "0.3.0-alpha.19"
 libp2p = { version = "0.12.0", default-features = false, features = ["secp256k1", "libp2p-websocket"] }
 log = "0.4.8"
 network = { package = "substrate-network", path = "../../core/network" }
 primitives = { package = "substrate-primitives", path = "../primitives" }
 prost = "0.5.0"
 serde_json = "1.0.41"
 sr-primitives = { path = "../../core/sr-primitives" }
-tokio-timer = "0.2.11"
+futures-timer = "0.4"
 
 [dev-dependencies]
 parking_lot = "0.9.0"
 peerset = { package = "substrate-peerset", path = "../../core/peerset" }
 test-client = { package = "substrate-test-runtime-client", path = "../../core/test-runtime/client" }
-tokio = "0.1.22"

core/authority-discovery/src/error.rs

@@ -42,6 +42,4 @@ pub enum Error {
 	Decoding(prost::DecodeError),
 	/// Failed to parse a libp2p multi address.
 	ParsingMultiaddress(libp2p::core::multiaddr::Error),
-	/// Tokio timer error.
-	PollingTokioTimer(tokio_timer::Error)
 }

core/authority-discovery/src/lib.rs

@@ -42,23 +42,29 @@
 //!    3. Validates the signatures of the retrieved key value pairs.
 //!
 //!    4. Adds the retrieved external addresses as priority nodes to the peerset.
+use std::collections::{HashMap, HashSet};
+use std::convert::TryInto;
+use std::iter::FromIterator;
+use std::marker::PhantomData;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures::channel::mpsc::Receiver;
+use futures::stream::StreamExt;
+use futures::task::{Context, Poll};
+use futures::Future;
+use futures_timer::Interval;
 
 use authority_discovery_primitives::{AuthorityDiscoveryApi, AuthorityId, Signature};
 use client::blockchain::HeaderBackend;
 use error::{Error, Result};
-use futures::{prelude::*, sync::mpsc::Receiver};
 use log::{debug, error, log_enabled, warn};
 use network::specialization::NetworkSpecialization;
 use network::{DhtEvent, ExHashT};
 use prost::Message;
 use sr_primitives::generic::BlockId;
 use sr_primitives::traits::{Block as BlockT, ProvideRuntimeApi};
-use std::collections::{HashMap, HashSet};
-use std::convert::TryInto;
-use std::iter::FromIterator;
-use std::marker::PhantomData;
-use std::sync::Arc;
-use std::time::{Duration, Instant};
 
 mod error;
 /// Dht payload schemas generated from Protobuf definitions via Prost crate in build.rs.
@@ -81,9 +87,9 @@ where
 	dht_event_rx: Receiver<DhtEvent>,
 
 	/// Interval to be proactive, publishing own addresses.
-	publish_interval: tokio_timer::Interval,
+	publish_interval: Interval,
 	/// Interval on which to query for addresses of other authorities.
-	query_interval: tokio_timer::Interval,
+	query_interval: Interval,
 
 	/// The network peerset interface for priority groups lets us only set an entire group, but we retrieve the
 	/// addresses of other authorities one by one from the network. To use the peerset interface we need to cache the
@@ -96,27 +102,26 @@ where
 
 impl<Client, Network, Block> AuthorityDiscovery<Client, Network, Block>
 where
-	Block: BlockT + 'static,
+	Block: BlockT + Unpin + 'static,
 	Network: NetworkProvider,
 	Client: ProvideRuntimeApi + Send + Sync + 'static + HeaderBackend<Block>,
 	<Client as ProvideRuntimeApi>::Api: AuthorityDiscoveryApi<Block>,
+	Self: Future<Output = ()>,
 {
 	/// Return a new authority discovery.
 	pub fn new(
 		client: Arc<Client>,
 		network: Arc<Network>,
-		dht_event_rx: futures::sync::mpsc::Receiver<DhtEvent>,
-	) -> AuthorityDiscovery<Client, Network, Block> {
+		dht_event_rx: Receiver<DhtEvent>,
+	) -> Self {
 		// Kademlia's default time-to-live for Dht records is 36h, republishing records every 24h. Given that a node
 		// could restart at any point in time, one can not depend on the republishing process, thus publishing own
 		// external addresses should happen on an interval < 36h.
-		let publish_interval =
-			tokio_timer::Interval::new(Instant::now(), Duration::from_secs(12 * 60 * 60));
+		let publish_interval = Interval::new(Duration::from_secs(12 * 60 * 60));
 
 		// External addresses of other authorities can change at any given point in time. The interval on which to query
 		// for external addresses of other authorities is a trade off between efficiency and performance.
-		let query_interval =
-			tokio_timer::Interval::new(Instant::now(), Duration::from_secs(10 * 60));
+		let query_interval = Interval::new(Duration::from_secs(10 * 60));
 
 		let address_cache = HashMap::new();
 
@@ -191,8 +196,8 @@ where
 		Ok(())
 	}
 
-	fn handle_dht_events(&mut self) -> Result<()> {
-		while let Ok(Async::Ready(Some(event))) = self.dht_event_rx.poll() {
+	fn handle_dht_events(&mut self, cx: &mut Context) -> Result<()> {
+		while let Poll::Ready(Some(event)) = self.dht_event_rx.poll_next_unpin(cx) {
 			match event {
 				DhtEvent::ValueFound(v) => {
 					if log_enabled!(log::Level::Debug) {
@@ -202,15 +207,17 @@ where
 
 					self.handle_dht_value_found_event(v)?;
 				}
-				DhtEvent::ValueNotFound(hash) => {
-					warn!(target: "sub-authority-discovery", "Value for hash '{:?}' not found on Dht.", hash)
-				}
-				DhtEvent::ValuePut(hash) => {
-					debug!(target: "sub-authority-discovery", "Successfully put hash '{:?}' on Dht.", hash)
-				}
-				DhtEvent::ValuePutFailed(hash) => {
-					warn!(target: "sub-authority-discovery", "Failed to put hash '{:?}' on Dht.", hash)
-				}
+				DhtEvent::ValueNotFound(hash) => warn!(
+					target: "sub-authority-discovery",
+					"Value for hash '{:?}' not found on Dht.", hash
+				),
+				DhtEvent::ValuePut(hash) => debug!(
+					target: "sub-authority-discovery",
+					"Successfully put hash '{:?}' on Dht.", hash),
+				DhtEvent::ValuePutFailed(hash) => warn!(
+					target: "sub-authority-discovery",
+					"Failed to put hash '{:?}' on Dht.", hash
+				),
 			}
 		}
 
@@ -291,53 +298,36 @@ where
 	}
 }
 
-impl<Client, Network, Block> futures::Future for AuthorityDiscovery<Client, Network, Block>
+impl<Client, Network, Block> Future for AuthorityDiscovery<Client, Network, Block>
 where
-	Block: BlockT + 'static,
+	Block: BlockT + Unpin + 'static,
 	Network: NetworkProvider,
 	Client: ProvideRuntimeApi + Send + Sync + 'static + HeaderBackend<Block>,
 	<Client as ProvideRuntimeApi>::Api: AuthorityDiscoveryApi<Block>,
 {
-	type Item = ();
-	type Error = ();
+	type Output = ();
 
-	fn poll(&mut self) -> futures::Poll<Self::Item, Self::Error> {
+	fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
 		let mut inner = || -> Result<()> {
 			// Process incoming events before triggering new ones.
-			self.handle_dht_events()?;
+			self.handle_dht_events(cx)?;
 
-			if let Async::Ready(_) = self
-				.publish_interval
-				.poll()
-				.map_err(Error::PollingTokioTimer)?
-			{
+			if let Poll::Ready(_) = self.publish_interval.poll_next_unpin(cx) {
 				// Make sure to call interval.poll until it returns Async::NotReady once. Otherwise, in case one of the
 				// function calls within this block do a `return`, we don't call `interval.poll` again and thereby the
 				// underlying Tokio task is never registered with Tokio's Reactor to be woken up on the next interval
 				// tick.
-				while let Async::Ready(_) = self
-					.publish_interval
-					.poll()
-					.map_err(Error::PollingTokioTimer)?
-				{}
+				while let Poll::Ready(_) = self.publish_interval.poll_next_unpin(cx) {}
 
 				self.publish_own_ext_addresses()?;
 			}
 
-			if let Async::Ready(_) = self
-				.query_interval
-				.poll()
-				.map_err(Error::PollingTokioTimer)?
-			{
+			if let Poll::Ready(_) = self.query_interval.poll_next_unpin(cx) {
 				// Make sure to call interval.poll until it returns Async::NotReady once. Otherwise, in case one of the
 				// function calls within this block do a `return`, we don't call `interval.poll` again and thereby the
 				// underlying Tokio task is never registered with Tokio's Reactor to be woken up on the next interval
 				// tick.
-				while let Async::Ready(_) = self
-					.query_interval
-					.poll()
-					.map_err(Error::PollingTokioTimer)?
-				{}
+				while let Poll::Ready(_) = self.query_interval.poll_next_unpin(cx) {}
 
 				self.request_addresses_of_others()?;
 			}
@@ -351,7 +341,7 @@ where
 		};
 
 		// Make sure to always return NotReady as this is a long running task with the same lifetime as the node itself.
-		Ok(futures::Async::NotReady)
+		Poll::Pending
 	}
 }
 
@@ -415,13 +405,14 @@ fn hash_authority_id(id: &[u8]) -> Result<libp2p::kad::record::Key> {
 mod tests {
 	use super::*;
 	use client::runtime_api::{ApiExt, Core, RuntimeVersion};
+	use futures::channel::mpsc::channel;
+	use futures::executor::block_on;
 	use futures::future::poll_fn;
 	use primitives::{ExecutionContext, NativeOrEncoded};
 	use sr_primitives::traits::Zero;
 	use sr_primitives::traits::{ApiRef, Block as BlockT, NumberFor, ProvideRuntimeApi};
 	use std::sync::{Arc, Mutex};
 	use test_client::runtime::Block;
-	use tokio::runtime::current_thread;
 
 	#[derive(Clone)]
 	struct TestApi {}
@@ -611,7 +602,7 @@ mod tests {
 
 	#[test]
 	fn publish_own_ext_addresses_puts_record_on_dht() {
-		let (_dht_event_tx, dht_event_rx) = futures::sync::mpsc::channel(1000);
+		let (_dht_event_tx, dht_event_rx) = channel(1000);
 		let test_api = Arc::new(TestApi {});
 		let network: Arc<TestNetwork> = Arc::new(Default::default());
 
@@ -626,7 +617,7 @@ mod tests {
 
 	#[test]
 	fn request_addresses_of_others_triggers_dht_get_query() {
-		let (_dht_event_tx, dht_event_rx) = futures::sync::mpsc::channel(1000);
+		let (_dht_event_tx, dht_event_rx) = channel(1000);
 		let test_api = Arc::new(TestApi {});
 		let network: Arc<TestNetwork> = Arc::new(Default::default());
 
@@ -643,7 +634,7 @@ mod tests {
 	fn handle_dht_events_with_value_found_should_call_set_priority_group() {
 		// Create authority discovery.
 
-		let (mut dht_event_tx, dht_event_rx) = futures::sync::mpsc::channel(1000);
+		let (mut dht_event_tx, dht_event_rx) = channel(1000);
 		let test_api = Arc::new(TestApi {});
 		let network: Arc<TestNetwork> = Arc::new(Default::default());
 
@@ -674,9 +665,8 @@ mod tests {
 		dht_event_tx.try_send(dht_event).unwrap();
 
 		// Make authority discovery handle the event.
-
-		let f = || {
-			authority_discovery.handle_dht_events().unwrap();
+		let f = |cx: &mut Context<'_>| -> Poll<()> {
+			authority_discovery.handle_dht_events(cx).unwrap();
 
 			// Expect authority discovery to set the priority set.
 			assert_eq!(network.set_priority_group_call.lock().unwrap().len(), 1);
@@ -689,10 +679,9 @@ mod tests {
 				)
 			);
 
-			Ok(Async::Ready(()))
+			Poll::Ready(())
 		};
 
-		let mut runtime = current_thread::Runtime::new().unwrap();
-		runtime.block_on(poll_fn::<(), (), _>(f)).unwrap();
+		let _ = block_on(poll_fn(f));
 	}
 }