Skip to content
This repository has been archived by the owner on Nov 15, 2023. It is now read-only.

authority-discovery: futures 03 Future #3848

Merged
merged 10 commits into from
Nov 1, 2019
Merged
5 changes: 2 additions & 3 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

5 changes: 2 additions & 3 deletions core/authority-discovery/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
2 changes: 0 additions & 2 deletions core/authority-discovery/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
117 changes: 53 additions & 64 deletions core/authority-discovery/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -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();

Expand Down Expand Up @@ -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) {
Expand All @@ -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
),
}
}

Expand Down Expand Up @@ -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) {}
mxinden marked this conversation as resolved.
Show resolved Hide resolved

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) {
mxinden marked this conversation as resolved.
Show resolved Hide resolved
// 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) {}
mxinden marked this conversation as resolved.
Show resolved Hide resolved

self.request_addresses_of_others()?;
}
Expand All @@ -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
}
}

Expand Down Expand Up @@ -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 {}
Expand Down Expand Up @@ -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());

Expand All @@ -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());

Expand All @@ -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());

Expand Down Expand Up @@ -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);
Expand All @@ -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));
}
}