Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

seed command #83

Merged
merged 15 commits into from
Nov 13, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 53 additions & 6 deletions zebra-network/src/config.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,8 @@
use std::{net::SocketAddr, time::Duration};
use std::{
net::{SocketAddr, ToSocketAddrs},
string::String,
time::Duration,
};

use crate::network::Network;

Expand All @@ -8,38 +12,81 @@ use crate::network::Network;
pub struct Config {
/// The address on which this node should listen for connections.
pub listen_addr: SocketAddr,

/// The network to connect to.
pub network: Network,

/// The user-agent to advertise.
pub user_agent: String,
/// A list of initial peers for the peerset.
///
/// XXX this should be replaced with DNS names, not SocketAddrs
pub initial_peers: Vec<SocketAddr>,

/// A list of initial peers for the peerset when operating on
/// mainnet.
pub initial_mainnet_peers: Vec<String>,

/// A list of initial peers for the peerset when operating on
/// testnet.
pub initial_testnet_peers: Vec<String>,

/// The outgoing request buffer size for the peer set.
pub peerset_request_buffer_size: usize,

// Note: due to the way this is rendered by the toml
// serializer, the Duration fields should come last.
/// The default RTT estimate for peer responses, used in load-balancing.
pub ewma_default_rtt: Duration,

/// The decay time for the exponentially-weighted moving average response time.
pub ewma_decay_time: Duration,

/// The timeout for peer handshakes.
pub handshake_timeout: Duration,

/// How frequently we attempt to connect to a new peer.
pub new_peer_interval: Duration,
}

impl Config {
fn parse_peers<S: ToSocketAddrs>(peers: Vec<S>) -> Vec<SocketAddr> {
peers
.iter()
.flat_map(|s| s.to_socket_addrs())
.flatten()
.collect::<Vec<SocketAddr>>()
}

/// Get the initial seed peers based on the configured network.
pub fn initial_peers(&self) -> Vec<SocketAddr> {
match self.network {
Network::Mainnet => Config::parse_peers(self.initial_mainnet_peers.clone()),
Network::Testnet => Config::parse_peers(self.initial_testnet_peers.clone()),
}
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

take-or-leave-it: since parse_peers is never used outside of initial_peers, it could be folded in as match self.network { ... } .iter().flat_map(...).flatten().collect(), and this saves an allocation by only doing one collect instead of two.

}

impl Default for Config {
fn default() -> Config {
let mainnet_peers = [
"dnsseed.z.cash:8233",
"dnsseed.str4d.xyz:8233",
"dnsseed.znodes.org:8233",
]
.iter()
.map(|&s| String::from(s))
.collect();

let testnet_peers = ["dnsseed.testnet.z.cash:18233"]
.iter()
.map(|&s| String::from(s))
.collect();

Config {
listen_addr: "127.0.0.1:28233"
.parse()
.expect("Hardcoded address should be parseable"),
user_agent: crate::constants::USER_AGENT.to_owned(),
network: Network::Mainnet,
initial_peers: Vec::new(),
initial_mainnet_peers: mainnet_peers,
initial_testnet_peers: testnet_peers,
hdevalence marked this conversation as resolved.
Show resolved Hide resolved
ewma_default_rtt: Duration::from_secs(1),
ewma_decay_time: Duration::from_secs(60),
peerset_request_buffer_size: 1,
Expand Down
1 change: 1 addition & 0 deletions zebra-network/src/peer/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -304,6 +304,7 @@ where
// and try to construct an appropriate request object.
let req = match msg {
Message::Addr(addrs) => Some(Request::PushPeers(addrs)),
Message::GetAddr => Some(Request::GetPeers),
_ => {
debug!("unhandled message type");
None
Expand Down
2 changes: 1 addition & 1 deletion zebra-network/src/peer_set.rs
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ where

// 1. Initial peers, specified in the config.
tokio::spawn(add_initial_peers(
config.initial_peers.clone(),
config.initial_peers(),
connector.clone(),
peerset_tx.clone(),
));
Expand Down
19 changes: 8 additions & 11 deletions zebrad/src/commands.rs
Original file line number Diff line number Diff line change
@@ -1,21 +1,14 @@
//! Zebrad Subcommands
//!
//! This is where you specify the subcommands of your application.
//!
//! The default application comes with two subcommands:
//!
//! - `start`: launches the application
//! - `version`: print application version
//!
//! See the `impl Configurable` below for how to specify the path to the
//! application's configuration file.

mod config;
mod connect;
mod seed;
mod start;
mod version;

use self::{config::ConfigCmd, connect::ConnectCmd, start::StartCmd, version::VersionCmd};
use self::{
config::ConfigCmd, connect::ConnectCmd, seed::SeedCmd, start::StartCmd, version::VersionCmd,
};
use crate::config::ZebradConfig;
use abscissa_core::{
config::Override, Command, Configurable, FrameworkError, Help, Options, Runnable,
Expand Down Expand Up @@ -47,6 +40,10 @@ pub enum ZebradCmd {
/// The `connect` subcommand
#[options(help = "testing stub for dumping network messages")]
Connect(ConnectCmd),

/// The `seed` subcommand
#[options(help = "dns seeder")]
Seed(SeedCmd),
}

/// This trait allows you to define how application configuration is loaded.
Expand Down
2 changes: 1 addition & 1 deletion zebrad/src/commands/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ impl Runnable for ConfigCmd {
# can be found in Rustdoc. XXX add link to rendered docs.

"
.to_owned();
.to_owned(); // The default name and location of the config file is defined in ../commands.rs
hdevalence marked this conversation as resolved.
Show resolved Hide resolved
output += &toml::to_string_pretty(&default_config)
.expect("default config should be serializable");
match self.output_file {
Expand Down
4 changes: 1 addition & 3 deletions zebrad/src/commands/connect.rs
Original file line number Diff line number Diff line change
Expand Up @@ -66,9 +66,7 @@ impl ConnectCmd {
1,
);

let mut config = app_config().network.clone();

config.initial_peers = vec![self.addr];
let config = app_config().network.clone();

let (mut peer_set, address_book) = zebra_network::init(config, node).await;

Expand Down
155 changes: 155 additions & 0 deletions zebrad/src/commands/seed.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,155 @@
//! `seed` subcommand - runs a dns seeder

use std::{
future::Future,
pin::Pin,
sync::{Arc, Mutex},
task::{Context, Poll},
};

use abscissa_core::{config, Command, FrameworkError, Options, Runnable};
use futures::channel::oneshot;
use tower::{buffer::Buffer, Service, ServiceExt};
use zebra_network::{AddressBook, BoxedStdError, Request, Response};

use crate::{config::ZebradConfig, prelude::*};

/// Whether our `SeedService` is poll_ready or not.
#[derive(Debug)]
enum SeederState {
/// Waiting for the address book to be shared with us via the oneshot channel.
AwaitingAddressBook(oneshot::Receiver<Arc<Mutex<AddressBook>>>),
/// Address book received, ready to service requests.
Ready(Arc<Mutex<AddressBook>>),
}

#[derive(Debug)]
struct SeedService {
state: SeederState,
}

impl Service<Request> for SeedService {
type Response = Response;
type Error = BoxedStdError;
type Future =
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;

fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
debug!("SeedService.state: {:?}", self.state);

match self.state {
SeederState::Ready(_) => return Poll::Ready(Ok(())),
SeederState::AwaitingAddressBook(ref mut rx) => match rx.try_recv() {
Err(e) => {
error!("SeedService oneshot sender dropped: {:?}", e);
return Poll::Ready(Err(e.into()));
}
Ok(None) => {
trace!("SeedService hasn't received a message via the oneshot yet.");
return Poll::Pending;
}
Ok(Some(address_book)) => {
info!(
"SeedService received address_book via oneshot {:?}",
address_book
);
self.state = SeederState::Ready(address_book);
return Poll::Ready(Ok(()));
}
},
}
}

fn call(&mut self, req: Request) -> Self::Future {
info!("SeedService handling a request: {:?}", req);

let address_book = if let SeederState::Ready(address_book) = &self.state {
address_book
} else {
panic!("SeedService::call without SeedService::poll_ready");
};

let response = match req {
Request::GetPeers => {
debug!(address_book.len = address_book.lock().unwrap().len());
info!("SeedService responding to GetPeers");
Ok::<Response, Self::Error>(Response::Peers(
address_book.lock().unwrap().peers().collect(),
))
}
_ => Ok::<Response, Self::Error>(Response::Ok),
hdevalence marked this conversation as resolved.
Show resolved Hide resolved
};

info!("SeedService response: {:?}", response);
return Box::pin(futures::future::ready(response));
}
}

/// `seed` subcommand
///
/// A DNS seeder command to spider and collect as many valid peer
/// addresses as we can.
// This is not a unit-like struct because it makes Command and Options sad.
#[derive(Command, Debug, Default, Options)]
pub struct SeedCmd {}

impl Runnable for SeedCmd {
/// Start the application.
fn run(&self) {
use crate::components::tokio::TokioComponent;

let _ = app_reader()
.state()
.components
.get_downcast_ref::<TokioComponent>()
.expect("TokioComponent should be available")
.rt
.block_on(self.seed());
}
}

impl SeedCmd {
async fn seed(&self) -> Result<(), failure::Error> {
use failure::Error;

info!("begin tower-based peer handling test stub");

let (addressbook_tx, addressbook_rx) = oneshot::channel();
let seed_service = SeedService {
state: SeederState::AwaitingAddressBook(addressbook_rx),
};
let node = Buffer::new(seed_service, 1);

let config = app_config().network.clone();

let (mut peer_set, address_book) = zebra_network::init(config, node).await;

let _ = addressbook_tx.send(address_book);

info!("waiting for peer_set ready");
peer_set.ready().await.map_err(Error::from_boxed_compat)?;

info!("peer_set became ready");

#[cfg(dos)]
use std::time::Duration;
use tokio::timer::Interval;

#[cfg(dos)]
// Fire GetPeers requests at ourselves, for testing.
tokio::spawn(async move {
let mut interval_stream = Interval::new_interval(Duration::from_secs(1));

loop {
interval_stream.next().await;

let _ = seed_service.call(Request::GetPeers);
}
});

let eternity = tokio::future::pending::<()>();
eternity.await;

Ok(())
}
}