diff --git a/Cargo.lock b/Cargo.lock index 2dadb19b69103..2a7e1b2e85450 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4804,7 +4804,8 @@ version = "2.0.0" dependencies = [ "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", "derive_more 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)", - "futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-preview 0.3.0-alpha.19 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-timer 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)", "libp2p 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", "parity-scale-codec 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", @@ -4819,8 +4820,6 @@ dependencies = [ "substrate-peerset 2.0.0", "substrate-primitives 2.0.0", "substrate-test-runtime-client 2.0.0", - "tokio 0.1.22 (registry+https://github.com/rust-lang/crates.io-index)", - "tokio-timer 0.2.11 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] diff --git a/core/authority-discovery/Cargo.toml b/core/authority-discovery/Cargo.toml index 7283e07f89ca7..c4cab438f4007 100644 --- a/core/authority-discovery/Cargo.toml +++ b/core/authority-discovery/Cargo.toml @@ -14,7 +14,7 @@ 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" } @@ -22,10 +22,9 @@ 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" diff --git a/core/authority-discovery/src/error.rs b/core/authority-discovery/src/error.rs index e8c1ad9705f0c..dca50cc0beb9e 100644 --- a/core/authority-discovery/src/error.rs +++ b/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) } diff --git a/core/authority-discovery/src/lib.rs b/core/authority-discovery/src/lib.rs index 987169ead90b1..25d17cb1e903f 100644 --- a/core/authority-discovery/src/lib.rs +++ b/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, /// 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 AuthorityDiscovery where - Block: BlockT + 'static, + Block: BlockT + Unpin + 'static, Network: NetworkProvider, Client: ProvideRuntimeApi + Send + Sync + 'static + HeaderBackend, ::Api: AuthorityDiscoveryApi, + Self: Future, { /// Return a new authority discovery. pub fn new( client: Arc, network: Arc, - dht_event_rx: futures::sync::mpsc::Receiver, - ) -> AuthorityDiscovery { + dht_event_rx: Receiver, + ) -> 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 futures::Future for AuthorityDiscovery +impl Future for AuthorityDiscovery where - Block: BlockT + 'static, + Block: BlockT + Unpin + 'static, Network: NetworkProvider, Client: ProvideRuntimeApi + Send + Sync + 'static + HeaderBackend, ::Api: AuthorityDiscoveryApi, { - type Item = (); - type Error = (); + type Output = (); - fn poll(&mut self) -> futures::Poll { + fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { 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 { 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 = 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 = 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 = 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)); } }