gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-07-01 03:44:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into process-free-attestation

Browse Source
This commit is contained in:
Grant Wuerker
2019-07-18 12:33:22 +02:00
parent 40b166edcd bbde06eb9f
commit b90edaf7f6
60 changed files with 3857 additions and 715 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
beacon_node/Cargo.toml
View File

@@ -13,7 +13,7 @@ client = { path = "client" }
version = { path = "version" }
clap = "2.32.0"
serde = "1.0"
slog = { version = "^2.2.3" , features = ["max_level_trace", "release_max_level_debug"] }
slog = { version = "^2.2.3" , features = ["max_level_trace"] }
slog-term = "^2.4.0"
slog-async = "^2.3.0"
ctrlc = { version = "3.1.1", features = ["termination"] }
@@ -22,3 +22,5 @@ tokio-timer = "0.2.10"
futures = "0.1.25"
exit-future = "0.1.3"
state_processing = { path = "../eth2/state_processing" }
env_logger = "0.6.1"
dirs = "2.0.1"

15
beacon_node/beacon_chain/src/beacon_chain.rs
View File

@@ -18,7 +18,7 @@ use state_processing::{
per_slot_processing, BlockProcessingError, common
};
use std::sync::Arc;
use store::iter::{BlockIterator, BlockRootsIterator, StateRootsIterator};
use store::iter::{BestBlockRootsIterator, BlockIterator, BlockRootsIterator, StateRootsIterator};
use store::{Error as DBError, Store};
use tree_hash::TreeHash;
use types::*;
@@ -226,6 +226,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
BlockRootsIterator::owned(self.store.clone(), self.state.read().clone(), slot)
}
/// Iterates in reverse (highest to lowest slot) through all block roots from largest
/// `slot <= beacon_state.slot` through to genesis.
///
/// Returns `None` for roots prior to genesis or when there is an error reading from `Store`.
///
/// Contains duplicate roots when skip slots are encountered.
pub fn rev_iter_best_block_roots(
&self,
slot: Slot,
) -> BestBlockRootsIterator<T::EthSpec, T::Store> {
BestBlockRootsIterator::owned(self.store.clone(), self.state.read().clone(), slot)
}
/// Iterates in reverse (highest to lowest slot) through all state roots from `slot` through to
/// genesis.
///

2
beacon_node/beacon_chain/src/test_utils.rs
View File

@@ -191,7 +191,7 @@ where
fn get_state_at_slot(&self, state_slot: Slot) -> BeaconState<E> {
let state_root = self
.chain
.rev_iter_state_roots(self.chain.current_state().slot)
.rev_iter_state_roots(self.chain.current_state().slot - 1)
.find(|(_hash, slot)| *slot == state_slot)
.map(|(hash, _slot)| hash)
.expect("could not find state root");

8
beacon_node/client/Cargo.toml
View File

@@ -9,17 +9,21 @@ beacon_chain = { path = "../beacon_chain" }
network = { path = "../network" }
store = { path = "../store" }
http_server = { path = "../http_server" }
eth2-libp2p = { path = "../eth2-libp2p" }
rpc = { path = "../rpc" }
prometheus = "^0.6"
types = { path = "../../eth2/types" }
tree_hash = { path = "../../eth2/utils/tree_hash" }
eth2_config = { path = "../../eth2/utils/eth2_config" }
slot_clock = { path = "../../eth2/utils/slot_clock" }
serde = "1.0"
serde = "1.0.93"
serde_derive = "1.0"
error-chain = "0.12.0"
slog = "^2.2.3"
eth2_ssz = { path = "../../eth2/utils/ssz" }
slog = { version = "^2.2.3" , features = ["max_level_trace"] }
slog-async = "^2.3.0"
slog-json = "^2.3"
slog-term = "^2.4.0"
tokio = "0.1.15"
clap = "2.32.0"
dirs = "1.0.3"

59
beacon_node/client/src/client_config.rs → beacon_node/client/src/config.rs
View File

@@ -2,36 +2,40 @@ use clap::ArgMatches;
use http_server::HttpServerConfig;
use network::NetworkConfig;
use serde_derive::{Deserialize, Serialize};
use std::fs;
use slog::{info, o, Drain};
use std::fs::{self, OpenOptions};
use std::path::PathBuf;
use std::sync::Mutex;
/// The core configuration of a Lighthouse beacon node.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClientConfig {
pub struct Config {
pub data_dir: PathBuf,
pub db_type: String,
db_name: String,
pub log_file: PathBuf,
pub network: network::NetworkConfig,
pub rpc: rpc::RPCConfig,
pub http: HttpServerConfig,
}
impl Default for ClientConfig {
impl Default for Config {
fn default() -> Self {
Self {
data_dir: PathBuf::from(".lighthouse"),
log_file: PathBuf::from(""),
db_type: "disk".to_string(),
db_name: "chain_db".to_string(),
// Note: there are no default bootnodes specified.
// Once bootnodes are established, add them here.
network: NetworkConfig::new(vec![]),
network: NetworkConfig::new(),
rpc: rpc::RPCConfig::default(),
http: HttpServerConfig::default(),
}
}
}
impl ClientConfig {
impl Config {
/// Returns the path to which the client may initialize an on-disk database.
pub fn db_path(&self) -> Option<PathBuf> {
self.data_dir()
@@ -45,23 +49,64 @@ impl ClientConfig {
Some(path)
}
// Update the logger to output in JSON to specified file
fn update_logger(&mut self, log: &mut slog::Logger) -> Result<(), &'static str> {
let file = OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&self.log_file);
if file.is_err() {
return Err("Cannot open log file");
}
let file = file.unwrap();
if let Some(file) = self.log_file.to_str() {
info!(
*log,
"Log file specified, output will now be written to {} in json.", file
);
} else {
info!(
*log,
"Log file specified output will now be written in json"
);
}
let drain = Mutex::new(slog_json::Json::default(file)).fuse();
let drain = slog_async::Async::new(drain).build().fuse();
*log = slog::Logger::root(drain, o!());
Ok(())
}
/// Apply the following arguments to `self`, replacing values if they are specified in `args`.
///
/// Returns an error if arguments are obviously invalid. May succeed even if some values are
/// invalid.
pub fn apply_cli_args(&mut self, args: &ArgMatches) -> Result<(), &'static str> {
pub fn apply_cli_args(
&mut self,
args: &ArgMatches,
log: &mut slog::Logger,
) -> Result<(), String> {
if let Some(dir) = args.value_of("datadir") {
self.data_dir = PathBuf::from(dir);
};
if let Some(dir) = args.value_of("db") {
self.db_type = dir.to_string();
}
};
self.network.apply_cli_args(args)?;
self.rpc.apply_cli_args(args)?;
self.http.apply_cli_args(args)?;
if let Some(log_file) = args.value_of("logfile") {
self.log_file = PathBuf::from(log_file);
self.update_logger(log)?;
};
Ok(())
}
}

4
beacon_node/client/src/lib.rs
View File

@@ -1,7 +1,7 @@
extern crate slog;
mod beacon_chain_types;
mod client_config;
mod config;
pub mod error;
pub mod notifier;
@@ -21,7 +21,7 @@ use tokio::timer::Interval;
pub use beacon_chain::BeaconChainTypes;
pub use beacon_chain_types::ClientType;
pub use beacon_chain_types::InitialiseBeaconChain;
pub use client_config::ClientConfig;
pub use config::Config as ClientConfig;
pub use eth2_config::Eth2Config;
/// Main beacon node client service. This provides the connection and initialisation of the clients

33
beacon_node/client/src/notifier.rs
View File

@@ -3,39 +3,34 @@ use beacon_chain::BeaconChainTypes;
use exit_future::Exit;
use futures::{Future, Stream};
use slog::{debug, o};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::runtime::TaskExecutor;
use tokio::timer::Interval;
/// Thread that monitors the client and reports useful statistics to the user.
/// The interval between heartbeat events.
pub const HEARTBEAT_INTERVAL_SECONDS: u64 = 5;
/// Spawns a thread that can be used to run code periodically, on `HEARTBEAT_INTERVAL_SECONDS`
/// durations.
///
/// Presently unused, but remains for future use.
pub fn run<T: BeaconChainTypes + Send + Sync + 'static>(
client: &Client<T>,
executor: TaskExecutor,
exit: Exit,
) {
// notification heartbeat
let interval = Interval::new(Instant::now(), Duration::from_secs(5));
let interval = Interval::new(
Instant::now(),
Duration::from_secs(HEARTBEAT_INTERVAL_SECONDS),
);
let _log = client.log.new(o!("Service" => "Notifier"));
// TODO: Debugging only
let counter = Arc::new(Mutex::new(0));
let network = client.network.clone();
// build heartbeat logic here
let heartbeat = move |_| {
//debug!(log, "Temp heartbeat output");
//TODO: Remove this logic. Testing only
let mut count = counter.lock().unwrap();
*count += 1;
if *count % 5 == 0 {
// debug!(log, "Sending Message");
network.send_message();
}
let heartbeat = |_| {
// There is not presently any heartbeat logic.
//
// We leave this function empty for future use.
Ok(())
};

9
beacon_node/eth2-libp2p/Cargo.toml
View File

@@ -7,15 +7,18 @@ edition = "2018"
[dependencies]
beacon_chain = { path = "../beacon_chain" }
clap = "2.32.0"
# SigP repository until PR is merged
libp2p = { git = "https://github.com/SigP/rust-libp2p", rev = "b3c32d9a821ae6cc89079499cc6e8a6bab0bffc3" }
#SigP repository
libp2p = { git = "https://github.com/SigP/rust-libp2p", rev = "be5710bbde69d8c5be732c13ba64239e2f370a7b" }
enr = { git = "https://github.com/SigP/rust-libp2p/", rev = "be5710bbde69d8c5be732c13ba64239e2f370a7b", features = ["serde"] }
types = { path = "../../eth2/types" }
serde = "1.0"
serde_derive = "1.0"
eth2_ssz = { path = "../../eth2/utils/ssz" }
eth2_ssz_derive = { path = "../../eth2/utils/ssz_derive" }
slog = "2.4.1"
slog = { version = "^2.4.1" , features = ["max_level_trace"] }
version = { path = "../version" }
tokio = "0.1.16"
futures = "0.1.25"
error-chain = "0.12.0"
tokio-timer = "0.2.10"
dirs = "2.0.1"

122
beacon_node/eth2-libp2p/src/behaviour.rs
View File

@@ -1,45 +1,72 @@
use crate::discovery::Discovery;
use crate::rpc::{RPCEvent, RPCMessage, Rpc};
use crate::NetworkConfig;
use crate::{error, NetworkConfig};
use crate::{Topic, TopicHash};
use futures::prelude::*;
use libp2p::{
core::{
identity::Keypair,
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess},
PublicKey,
},
discv5::Discv5Event,
gossipsub::{Gossipsub, GossipsubEvent},
identify::{protocol::IdentifyInfo, Identify, IdentifyEvent},
ping::{Ping, PingEvent},
ping::{Ping, PingConfig, PingEvent},
tokio_io::{AsyncRead, AsyncWrite},
NetworkBehaviour, PeerId,
};
use slog::{debug, o, trace, warn};
use slog::{o, trace, warn};
use ssz::{ssz_encode, Decode, DecodeError, Encode};
use std::num::NonZeroU32;
use std::time::Duration;
use types::{Attestation, BeaconBlock};
use types::{Topic, TopicHash};
/// Builds the network behaviour for the libp2p Swarm.
/// Implements gossipsub message routing.
/// Builds the network behaviour that manages the core protocols of eth2.
/// This core behaviour is managed by `Behaviour` which adds peer management to all core
/// behaviours.
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "BehaviourEvent", poll_method = "poll")]
pub struct Behaviour<TSubstream: AsyncRead + AsyncWrite> {
/// The routing pub-sub mechanism for eth2.
gossipsub: Gossipsub<TSubstream>,
// TODO: Add Kademlia for peer discovery
/// The events generated by this behaviour to be consumed in the swarm poll.
/// The serenity RPC specified in the wire-0 protocol.
serenity_rpc: Rpc<TSubstream>,
/// Allows discovery of IP addresses for peers on the network.
identify: Identify<TSubstream>,
/// Keep regular connection to peers and disconnect if absent.
// TODO: Keepalive, likely remove this later.
// TODO: Make the ping time customizeable.
ping: Ping<TSubstream>,
/// Kademlia for peer discovery.
discovery: Discovery<TSubstream>,
#[behaviour(ignore)]
/// The events generated by this behaviour to be consumed in the swarm poll.
events: Vec<BehaviourEvent>,
/// Logger for behaviour actions.
#[behaviour(ignore)]
log: slog::Logger,
}
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
pub fn new(
local_key: &Keypair,
net_conf: &NetworkConfig,
log: &slog::Logger,
) -> error::Result<Self> {
let local_peer_id = local_key.public().clone().into_peer_id();
let behaviour_log = log.new(o!());
let ping_config = PingConfig::new()
.with_timeout(Duration::from_secs(30))
.with_interval(Duration::from_secs(20))
.with_max_failures(NonZeroU32::new(2).expect("2 != 0"))
.with_keep_alive(false);
Ok(Behaviour {
serenity_rpc: Rpc::new(log),
gossipsub: Gossipsub::new(local_peer_id.clone(), net_conf.gs_config.clone()),
discovery: Discovery::new(local_key, net_conf, log)?,
ping: Ping::new(ping_config),
events: Vec::new(),
log: behaviour_log,
})
}
}
// Implement the NetworkBehaviourEventProcess trait so that we can derive NetworkBehaviour for Behaviour
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubEvent>
for Behaviour<TSubstream>
@@ -89,30 +116,6 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RPCMessage
}
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<IdentifyEvent>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, event: IdentifyEvent) {
match event {
IdentifyEvent::Identified {
peer_id, mut info, ..
} => {
if info.listen_addrs.len() > 20 {
debug!(
self.log,
"More than 20 peers have been identified, truncating"
);
info.listen_addrs.truncate(20);
}
self.events
.push(BehaviourEvent::Identified(peer_id, Box::new(info)));
}
IdentifyEvent::Error { .. } => {}
IdentifyEvent::SendBack { .. } => {}
}
}
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
for Behaviour<TSubstream>
{
@@ -122,25 +125,6 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
}
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
pub fn new(local_public_key: PublicKey, net_conf: &NetworkConfig, log: &slog::Logger) -> Self {
let local_peer_id = local_public_key.clone().into_peer_id();
let identify_config = net_conf.identify_config.clone();
let behaviour_log = log.new(o!());
Behaviour {
gossipsub: Gossipsub::new(local_peer_id, net_conf.gs_config.clone()),
serenity_rpc: Rpc::new(log),
identify: Identify::new(
identify_config.version,
identify_config.user_agent,
local_public_key,
),
ping: Ping::new(),
events: Vec::new(),
log: behaviour_log,
}
}
/// Consumes the events list when polled.
fn poll<TBehaviourIn>(
&mut self,
@@ -153,18 +137,23 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
}
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<Discv5Event>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, _event: Discv5Event) {
// discv5 has no events to inject
}
}
/// Implements the combined behaviour for the libp2p service.
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
/* Pubsub behaviour functions */
/// Subscribes to a gossipsub topic.
pub fn subscribe(&mut self, topic: Topic) -> bool {
self.gossipsub.subscribe(topic)
}
/// Sends an RPC Request/Response via the RPC protocol.
pub fn send_rpc(&mut self, peer_id: PeerId, rpc_event: RPCEvent) {
self.serenity_rpc.send_rpc(peer_id, rpc_event);
}
/// Publishes a message on the pubsub (gossipsub) behaviour.
pub fn publish(&mut self, topics: Vec<Topic>, message: PubsubMessage) {
let message_bytes = ssz_encode(&message);
@@ -172,14 +161,19 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
self.gossipsub.publish(topic, message_bytes.clone());
}
}
/* Eth2 RPC behaviour functions */
/// Sends an RPC Request/Response via the RPC protocol.
pub fn send_rpc(&mut self, peer_id: PeerId, rpc_event: RPCEvent) {
self.serenity_rpc.send_rpc(peer_id, rpc_event);
}
}
/// The types of events than can be obtained from polling the behaviour.
pub enum BehaviourEvent {
RPC(PeerId, RPCEvent),
PeerDialed(PeerId),
Identified(PeerId, Box<IdentifyInfo>),
// TODO: This is a stub at the moment
GossipMessage {
source: PeerId,
topics: Vec<TopicHash>,

132
beacon_node/eth2-libp2p/src/config.rs
View File

@@ -1,89 +1,129 @@
use clap::ArgMatches;
use enr::Enr;
use libp2p::gossipsub::{GossipsubConfig, GossipsubConfigBuilder};
use serde_derive::{Deserialize, Serialize};
use types::multiaddr::{Error as MultiaddrError, Multiaddr};
use std::path::PathBuf;
use std::time::Duration;
/// The beacon node topic string to subscribe to.
pub const BEACON_PUBSUB_TOPIC: &str = "beacon_block";
pub const BEACON_ATTESTATION_TOPIC: &str = "beacon_attestation";
pub const SHARD_TOPIC_PREFIX: &str = "shard";
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(default)]
/// Network configuration for lighthouse.
pub struct Config {
/// Data directory where node's keyfile is stored
pub network_dir: PathBuf,
/// IP address to listen on.
listen_addresses: Vec<String>,
pub listen_address: std::net::IpAddr,
/// The TCP port that libp2p listens on.
pub libp2p_port: u16,
/// The address to broadcast to peers about which address we are listening on.
pub discovery_address: std::net::IpAddr,
/// UDP port that discovery listens on.
pub discovery_port: u16,
/// Target number of connected peers.
pub max_peers: usize,
/// Gossipsub configuration parameters.
#[serde(skip)]
pub gs_config: GossipsubConfig,
/// Configuration parameters for node identification protocol.
#[serde(skip)]
pub identify_config: IdentifyConfig,
/// List of nodes to initially connect to.
boot_nodes: Vec<String>,
pub boot_nodes: Vec<Enr>,
/// Client version
pub client_version: String,
/// List of topics to subscribe to as strings
/// List of extra topics to initially subscribe to as strings.
pub topics: Vec<String>,
}
impl Default for Config {
/// Generate a default network configuration.
fn default() -> Self {
let mut network_dir = dirs::home_dir().unwrap_or_else(|| PathBuf::from("."));
network_dir.push(".lighthouse");
network_dir.push("network");
Config {
listen_addresses: vec!["/ip4/127.0.0.1/tcp/9000".to_string()],
network_dir,
listen_address: "127.0.0.1".parse().expect("vaild ip address"),
libp2p_port: 9000,
discovery_address: "127.0.0.1".parse().expect("valid ip address"),
discovery_port: 9000,
max_peers: 10,
//TODO: Set realistic values for production
gs_config: GossipsubConfigBuilder::new()
.max_gossip_size(4_000_000)
.inactivity_timeout(Duration::from_secs(90))
.heartbeat_interval(Duration::from_secs(20))
.build(),
identify_config: IdentifyConfig::default(),
boot_nodes: vec![],
client_version: version::version(),
topics: vec![String::from("beacon_chain")],
topics: Vec::new(),
}
}
}
/// Generates a default Config.
impl Config {
pub fn new(boot_nodes: Vec<String>) -> Self {
let mut conf = Config::default();
conf.boot_nodes = boot_nodes;
conf
pub fn new() -> Self {
Config::default()
}
pub fn listen_addresses(&self) -> Result<Vec<Multiaddr>, MultiaddrError> {
self.listen_addresses.iter().map(|s| s.parse()).collect()
}
pub fn apply_cli_args(&mut self, args: &ArgMatches) -> Result<(), String> {
if let Some(dir) = args.value_of("datadir") {
self.network_dir = PathBuf::from(dir).join("network");
};
pub fn boot_nodes(&self) -> Result<Vec<Multiaddr>, MultiaddrError> {
self.boot_nodes.iter().map(|s| s.parse()).collect()
}
pub fn apply_cli_args(&mut self, args: &ArgMatches) -> Result<(), &'static str> {
if let Some(listen_address_str) = args.value_of("listen-address") {
let listen_addresses = listen_address_str.split(',').map(Into::into).collect();
self.listen_addresses = listen_addresses;
let listen_address = listen_address_str
.parse()
.map_err(|_| format!("Invalid listen address: {:?}", listen_address_str))?;
self.listen_address = listen_address;
self.discovery_address = listen_address;
}
if let Some(boot_addresses_str) = args.value_of("boot-nodes") {
let boot_addresses = boot_addresses_str.split(',').map(Into::into).collect();
self.boot_nodes = boot_addresses;
if let Some(max_peers_str) = args.value_of("maxpeers") {
self.max_peers = max_peers_str
.parse::<usize>()
.map_err(|_| format!("Invalid number of max peers: {}", max_peers_str))?;
}
if let Some(port_str) = args.value_of("port") {
let port = port_str
.parse::<u16>()
.map_err(|_| format!("Invalid port: {}", port_str))?;
self.libp2p_port = port;
self.discovery_port = port;
}
if let Some(boot_enr_str) = args.value_of("boot-nodes") {
self.boot_nodes = boot_enr_str
.split(',')
.map(|enr| enr.parse().map_err(|_| format!("Invalid ENR: {}", enr)))
.collect::<Result<Vec<Enr>, _>>()?;
}
if let Some(discovery_address_str) = args.value_of("discovery-address") {
self.discovery_address = discovery_address_str
.parse()
.map_err(|_| format!("Invalid discovery address: {:?}", discovery_address_str))?
}
if let Some(disc_port_str) = args.value_of("disc-port") {
self.discovery_port = disc_port_str
.parse::<u16>()
.map_err(|_| format!("Invalid discovery port: {}", disc_port_str))?;
}
Ok(())
}
}
/// The configuration parameters for the Identify protocol
#[derive(Debug, Clone)]
pub struct IdentifyConfig {
/// The protocol version to listen on.
pub version: String,
/// The client's name and version for identification.
pub user_agent: String,
}
impl Default for IdentifyConfig {
fn default() -> Self {
Self {
version: "/eth/serenity/1.0".to_string(),
user_agent: version::version(),
}
}
}

313
beacon_node/eth2-libp2p/src/discovery.rs Normal file
View File

@@ -0,0 +1,313 @@
use crate::{error, NetworkConfig};
/// This manages the discovery and management of peers.
///
/// Currently using discv5 for peer discovery.
///
use futures::prelude::*;
use libp2p::core::swarm::{
ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
};
use libp2p::core::{identity::Keypair, Multiaddr, PeerId, ProtocolsHandler};
use libp2p::discv5::{Discv5, Discv5Event};
use libp2p::enr::{Enr, EnrBuilder, NodeId};
use libp2p::multiaddr::Protocol;
use slog::{debug, info, o, warn};
use std::collections::HashSet;
use std::fs::File;
use std::io::prelude::*;
use std::str::FromStr;
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_timer::Delay;
/// Maximum seconds before searching for extra peers.
const MAX_TIME_BETWEEN_PEER_SEARCHES: u64 = 60;
/// Initial delay between peer searches.
const INITIAL_SEARCH_DELAY: u64 = 5;
/// Local ENR storage filename.
const ENR_FILENAME: &str = "enr.dat";
/// Lighthouse discovery behaviour. This provides peer management and discovery using the Discv5
/// libp2p protocol.
pub struct Discovery<TSubstream> {
/// The peers currently connected to libp2p streams.
connected_peers: HashSet<PeerId>,
/// The target number of connected peers on the libp2p interface.
max_peers: usize,
/// The delay between peer discovery searches.
peer_discovery_delay: Delay,
/// Tracks the last discovery delay. The delay is doubled each round until the max
/// time is reached.
past_discovery_delay: u64,
/// The TCP port for libp2p. Used to convert an updated IP address to a multiaddr. Note: This
/// assumes that the external TCP port is the same as the internal TCP port if behind a NAT.
//TODO: Improve NAT handling limit the above restriction
tcp_port: u16,
/// The discovery behaviour used to discover new peers.
discovery: Discv5<TSubstream>,
/// Logger for the discovery behaviour.
log: slog::Logger,
}
impl<TSubstream> Discovery<TSubstream> {
pub fn new(
local_key: &Keypair,
config: &NetworkConfig,
log: &slog::Logger,
) -> error::Result<Self> {
let log = log.new(o!("Service" => "Libp2p-Discovery"));
// checks if current ENR matches that found on disk
let local_enr = load_enr(local_key, config, &log)?;
info!(log, "Local ENR: {}", local_enr.to_base64());
debug!(log, "Local Node Id: {}", local_enr.node_id());
let mut discovery = Discv5::new(local_enr, local_key.clone(), config.listen_address)
.map_err(|e| format!("Discv5 service failed: {:?}", e))?;
// Add bootnodes to routing table
for bootnode_enr in config.boot_nodes.clone() {
debug!(
log,
"Adding node to routing table: {}",
bootnode_enr.node_id()
);
discovery.add_enr(bootnode_enr);
}
Ok(Self {
connected_peers: HashSet::new(),
max_peers: config.max_peers,
peer_discovery_delay: Delay::new(Instant::now()),
past_discovery_delay: INITIAL_SEARCH_DELAY,
tcp_port: config.libp2p_port,
discovery,
log,
})
}
/// Manually search for peers. This restarts the discovery round, sparking multiple rapid
/// queries.
pub fn discover_peers(&mut self) {
self.past_discovery_delay = INITIAL_SEARCH_DELAY;
self.find_peers();
}
/// Add an Enr to the routing table of the discovery mechanism.
pub fn add_enr(&mut self, enr: Enr) {
self.discovery.add_enr(enr);
}
/// Search for new peers using the underlying discovery mechanism.
fn find_peers(&mut self) {
// pick a random NodeId
let random_node = NodeId::random();
debug!(self.log, "Searching for peers...");
self.discovery.find_node(random_node);
// update the time until next discovery
let delay = {
if self.past_discovery_delay < MAX_TIME_BETWEEN_PEER_SEARCHES {
self.past_discovery_delay *= 2;
self.past_discovery_delay
} else {
MAX_TIME_BETWEEN_PEER_SEARCHES
}
};
self.peer_discovery_delay
.reset(Instant::now() + Duration::from_secs(delay));
}
}
// Redirect all behaviour events to underlying discovery behaviour.
impl<TSubstream> NetworkBehaviour for Discovery<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
{
type ProtocolsHandler = <Discv5<TSubstream> as NetworkBehaviour>::ProtocolsHandler;
type OutEvent = <Discv5<TSubstream> as NetworkBehaviour>::OutEvent;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
NetworkBehaviour::new_handler(&mut self.discovery)
}
fn addresses_of_peer(&mut self, peer_id: &PeerId) -> Vec<Multiaddr> {
// Let discovery track possible known peers.
self.discovery.addresses_of_peer(peer_id)
}
fn inject_connected(&mut self, peer_id: PeerId, _endpoint: ConnectedPoint) {
self.connected_peers.insert(peer_id);
}
fn inject_disconnected(&mut self, peer_id: &PeerId, _endpoint: ConnectedPoint) {
self.connected_peers.remove(peer_id);
}
fn inject_replaced(
&mut self,
_peer_id: PeerId,
_closed: ConnectedPoint,
_opened: ConnectedPoint,
) {
// discv5 doesn't implement
}
fn inject_node_event(
&mut self,
_peer_id: PeerId,
_event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
// discv5 doesn't implement
}
fn poll(
&mut self,
params: &mut impl PollParameters,
) -> Async<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
// search for peers if it is time
loop {
match self.peer_discovery_delay.poll() {
Ok(Async::Ready(_)) => {
if self.connected_peers.len() < self.max_peers {
self.find_peers();
}
}
Ok(Async::NotReady) => break,
Err(e) => {
warn!(self.log, "Discovery peer search failed: {:?}", e);
}
}
}
// Poll discovery
loop {
match self.discovery.poll(params) {
Async::Ready(NetworkBehaviourAction::GenerateEvent(event)) => {
match event {
Discv5Event::Discovered(_enr) => {
// not concerned about FINDNODE results, rather the result of an entire
// query.
}
Discv5Event::SocketUpdated(socket) => {
info!(self.log, "Address updated"; "IP" => format!("{}",socket.ip()));
let mut address = Multiaddr::from(socket.ip());
address.push(Protocol::Tcp(self.tcp_port));
return Async::Ready(NetworkBehaviourAction::ReportObservedAddr {
address,
});
}
Discv5Event::FindNodeResult { closer_peers, .. } => {
debug!(self.log, "Discv5 query found {} peers", closer_peers.len());
if closer_peers.is_empty() {
debug!(self.log, "Discv5 random query yielded empty results");
}
for peer_id in closer_peers {
// if we need more peers, attempt a connection
if self.connected_peers.len() < self.max_peers
&& self.connected_peers.get(&peer_id).is_none()
{
debug!(self.log, "Discv5: Peer discovered"; "Peer"=> format!("{:?}", peer_id));
return Async::Ready(NetworkBehaviourAction::DialPeer {
peer_id,
});
}
}
}
_ => {}
}
}
// discv5 does not output any other NetworkBehaviourAction
Async::Ready(_) => {}
Async::NotReady => break,
}
}
Async::NotReady
}
}
/// Loads an ENR from file if it exists and matches the current NodeId and sequence number. If none
/// exists, generates a new one.
///
/// If an ENR exists, with the same NodeId and IP address, we use the disk-generated one as its
/// ENR sequence will be equal or higher than a newly generated one.
fn load_enr(
local_key: &Keypair,
config: &NetworkConfig,
log: &slog::Logger,
) -> Result<Enr, String> {
// Build the local ENR.
// Note: Discovery should update the ENR record's IP to the external IP as seen by the
// majority of our peers.
let mut local_enr = EnrBuilder::new()
.ip(config.discovery_address.into())
.tcp(config.libp2p_port)
.udp(config.discovery_port)
.build(&local_key)
.map_err(|e| format!("Could not build Local ENR: {:?}", e))?;
let enr_f = config.network_dir.join(ENR_FILENAME);
if let Ok(mut enr_file) = File::open(enr_f.clone()) {
let mut enr_string = String::new();
match enr_file.read_to_string(&mut enr_string) {
Err(_) => debug!(log, "Could not read ENR from file"),
Ok(_) => {
match Enr::from_str(&enr_string) {
Ok(enr) => {
debug!(log, "ENR found in file: {:?}", enr_f);
if enr.node_id() == local_enr.node_id() {
if enr.ip() == config.discovery_address.into()
&& enr.tcp() == Some(config.libp2p_port)
&& enr.udp() == Some(config.discovery_port)
{
debug!(log, "ENR loaded from file");
// the stored ENR has the same configuration, use it
return Ok(enr);
}
// same node id, different configuration - update the sequence number
let new_seq_no = enr.seq().checked_add(1).ok_or_else(|| "ENR sequence number on file is too large. Remove it to generate a new NodeId")?;
local_enr.set_seq(new_seq_no, local_key).map_err(|e| {
format!("Could not update ENR sequence number: {:?}", e)
})?;
debug!(log, "ENR sequence number increased to: {}", new_seq_no);
}
}
Err(e) => {
warn!(log, "ENR from file could not be decoded: {:?}", e);
}
}
}
}
}
// write ENR to disk
let _ = std::fs::create_dir_all(&config.network_dir);
match File::create(enr_f.clone())
.and_then(|mut f| f.write_all(&local_enr.to_base64().as_bytes()))
{
Ok(_) => {
debug!(log, "ENR written to disk");
}
Err(e) => {
warn!(
log,
"Could not write ENR to file: {:?}. Error: {}", enr_f, e
);
}
}
Ok(local_enr)
}

10
beacon_node/eth2-libp2p/src/lib.rs
View File

@@ -4,12 +4,18 @@
/// This crate builds and manages the libp2p services required by the beacon node.
pub mod behaviour;
mod config;
mod discovery;
pub mod error;
pub mod rpc;
mod service;
pub use behaviour::PubsubMessage;
pub use config::Config as NetworkConfig;
pub use config::{
Config as NetworkConfig, BEACON_ATTESTATION_TOPIC, BEACON_PUBSUB_TOPIC, SHARD_TOPIC_PREFIX,
};
pub use libp2p::floodsub::{Topic, TopicBuilder, TopicHash};
pub use libp2p::multiaddr;
pub use libp2p::Multiaddr;
pub use libp2p::{
gossipsub::{GossipsubConfig, GossipsubConfigBuilder},
PeerId,
@@ -17,5 +23,3 @@ pub use libp2p::{
pub use rpc::RPCEvent;
pub use service::Libp2pEvent;
pub use service::Service;
pub use types::multiaddr;
pub use types::Multiaddr;

2
beacon_node/eth2-libp2p/src/rpc/mod.rs
View File

@@ -94,7 +94,7 @@ where
fn poll(
&mut self,
_: &mut PollParameters<'_>,
_: &mut impl PollParameters,
) -> Async<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,

1
beacon_node/eth2-libp2p/src/rpc/protocol.rs
View File

@@ -11,7 +11,6 @@ use tokio::io::{AsyncRead, AsyncWrite};
const MAX_READ_SIZE: usize = 4_194_304; // 4M
/// Implementation of the `ConnectionUpgrade` for the rpc protocol.
#[derive(Debug, Clone)]
pub struct RPCProtocol;

145
beacon_node/eth2-libp2p/src/service.rs
View File

@@ -3,25 +3,30 @@ use crate::error;
use crate::multiaddr::Protocol;
use crate::rpc::RPCEvent;
use crate::NetworkConfig;
use crate::{TopicBuilder, TopicHash};
use crate::{BEACON_ATTESTATION_TOPIC, BEACON_PUBSUB_TOPIC};
use futures::prelude::*;
use futures::Stream;
use libp2p::core::{
identity,
identity::Keypair,
multiaddr::Multiaddr,
muxing::StreamMuxerBox,
nodes::Substream,
transport::boxed::Boxed,
upgrade::{InboundUpgradeExt, OutboundUpgradeExt},
};
use libp2p::identify::protocol::IdentifyInfo;
use libp2p::{core, secio, PeerId, Swarm, Transport};
use slog::{debug, info, trace, warn};
use std::fs::File;
use std::io::prelude::*;
use std::io::{Error, ErrorKind};
use std::time::Duration;
use types::{TopicBuilder, TopicHash};
type Libp2pStream = Boxed<(PeerId, StreamMuxerBox), Error>;
type Libp2pBehaviour = Behaviour<Substream<StreamMuxerBox>>;
const NETWORK_KEY_FILENAME: &str = "key";
/// The configuration and state of the libp2p components for the beacon node.
pub struct Service {
/// The libp2p Swarm handler.
@@ -35,59 +40,52 @@ pub struct Service {
impl Service {
pub fn new(config: NetworkConfig, log: slog::Logger) -> error::Result<Self> {
debug!(log, "Libp2p Service starting");
debug!(log, "Network-libp2p Service starting");
// TODO: Currently using secp256k1 key pairs. Wire protocol specifies RSA. Waiting for this
// PR to be merged to generate RSA keys: https://github.com/briansmith/ring/pull/733
// TODO: Save and recover node key from disk
let local_private_key = identity::Keypair::generate_secp256k1();
// load the private key from CLI flag, disk or generate a new one
let local_private_key = load_private_key(&config, &log);
let local_public_key = local_private_key.public();
let local_peer_id = PeerId::from(local_private_key.public());
info!(log, "Local peer id: {:?}", local_peer_id);
let mut swarm = {
// Set up the transport
let transport = build_transport(local_private_key);
// Set up gossipsub routing
let behaviour = Behaviour::new(local_public_key.clone(), &config, &log);
// Set up Topology
let topology = local_peer_id.clone();
Swarm::new(transport, behaviour, topology)
// Set up the transport - tcp/ws with secio and mplex/yamux
let transport = build_transport(local_private_key.clone());
// Lighthouse network behaviour
let behaviour = Behaviour::new(&local_private_key, &config, &log)?;
Swarm::new(transport, behaviour, local_peer_id.clone())
};
// listen on all addresses
for address in config
.listen_addresses()
.map_err(|e| format!("Invalid listen multiaddr: {}", e))?
{
match Swarm::listen_on(&mut swarm, address.clone()) {
Ok(mut listen_addr) => {
listen_addr.append(Protocol::P2p(local_peer_id.clone().into()));
info!(log, "Listening on: {}", listen_addr);
}
Err(err) => warn!(log, "Cannot listen on: {} : {:?}", address, err),
};
}
// connect to boot nodes - these are currently stored as multiaddrs
// Once we have discovery, can set to peerId
for bootnode in config
.boot_nodes()
.map_err(|e| format!("Invalid boot node multiaddr: {:?}", e))?
{
match Swarm::dial_addr(&mut swarm, bootnode.clone()) {
Ok(()) => debug!(log, "Dialing bootnode: {}", bootnode),
Err(err) => debug!(
log,
"Could not connect to bootnode: {} error: {:?}", bootnode, err
),
};
}
// listen on the specified address
let listen_multiaddr = {
let mut m = Multiaddr::from(config.listen_address);
m.push(Protocol::Tcp(config.libp2p_port));
m
};
match Swarm::listen_on(&mut swarm, listen_multiaddr.clone()) {
Ok(_) => {
let mut log_address = listen_multiaddr;
log_address.push(Protocol::P2p(local_peer_id.clone().into()));
info!(log, "Listening on: {}", log_address);
}
Err(err) => warn!(
log,
"Cannot listen on: {} because: {:?}", listen_multiaddr, err
),
};
// subscribe to default gossipsub topics
let mut topics = vec![];
//TODO: Handle multiple shard attestations. For now we simply use a separate topic for
//attestations
topics.push(BEACON_ATTESTATION_TOPIC.to_string());
topics.push(BEACON_PUBSUB_TOPIC.to_string());
topics.append(&mut config.topics.clone());
let mut subscribed_topics = vec![];
for topic in config.topics {
let t = TopicBuilder::new(topic.to_string()).build();
for topic in topics {
let t = TopicBuilder::new(topic.clone()).build();
if swarm.subscribe(t) {
trace!(log, "Subscribed to topic: {:?}", topic);
subscribed_topics.push(topic);
@@ -111,8 +109,6 @@ impl Stream for Service {
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
loop {
// TODO: Currently only gossipsub events passed here.
// Build a type for more generic events
match self.swarm.poll() {
//Behaviour events
Ok(Async::Ready(Some(event))) => match event {
@@ -135,9 +131,6 @@ impl Stream for Service {
BehaviourEvent::PeerDialed(peer_id) => {
return Ok(Async::Ready(Some(Libp2pEvent::PeerDialed(peer_id))));
}
BehaviourEvent::Identified(peer_id, info) => {
return Ok(Async::Ready(Some(Libp2pEvent::Identified(peer_id, info))));
}
},
Ok(Async::Ready(None)) => unreachable!("Swarm stream shouldn't end"),
Ok(Async::NotReady) => break,
@@ -150,7 +143,7 @@ impl Stream for Service {
/// The implementation supports TCP/IP, WebSockets over TCP/IP, secio as the encryption layer, and
/// mplex or yamux as the multiplexing layer.
fn build_transport(local_private_key: identity::Keypair) -> Boxed<(PeerId, StreamMuxerBox), Error> {
fn build_transport(local_private_key: Keypair) -> Boxed<(PeerId, StreamMuxerBox), Error> {
// TODO: The Wire protocol currently doesn't specify encryption and this will need to be customised
// in the future.
let transport = libp2p::tcp::TcpConfig::new();
@@ -187,8 +180,6 @@ pub enum Libp2pEvent {
RPC(PeerId, RPCEvent),
/// Initiated the connection to a new peer.
PeerDialed(PeerId),
/// Received information about a peer on the network.
Identified(PeerId, Box<IdentifyInfo>),
/// Received pubsub message.
PubsubMessage {
source: PeerId,
@@ -196,3 +187,51 @@ pub enum Libp2pEvent {
message: Box<PubsubMessage>,
},
}
/// Loads a private key from disk. If this fails, a new key is
/// generated and is then saved to disk.
///
/// Currently only secp256k1 keys are allowed, as these are the only keys supported by discv5.
fn load_private_key(config: &NetworkConfig, log: &slog::Logger) -> Keypair {
// TODO: Currently using secp256k1 keypairs - currently required for discv5
// check for key from disk
let network_key_f = config.network_dir.join(NETWORK_KEY_FILENAME);
if let Ok(mut network_key_file) = File::open(network_key_f.clone()) {
let mut key_bytes: Vec<u8> = Vec::with_capacity(36);
match network_key_file.read_to_end(&mut key_bytes) {
Err(_) => debug!(log, "Could not read network key file"),
Ok(_) => {
// only accept secp256k1 keys for now
if let Ok(secret_key) =
libp2p::core::identity::secp256k1::SecretKey::from_bytes(&mut key_bytes)
{
let kp: libp2p::core::identity::secp256k1::Keypair = secret_key.into();
debug!(log, "Loaded network key from disk.");
return Keypair::Secp256k1(kp);
} else {
debug!(log, "Network key file is not a valid secp256k1 key");
}
}
}
}
// if a key could not be loaded from disk, generate a new one and save it
let local_private_key = Keypair::generate_secp256k1();
if let Keypair::Secp256k1(key) = local_private_key.clone() {
let _ = std::fs::create_dir_all(&config.network_dir);
match File::create(network_key_f.clone())
.and_then(|mut f| f.write_all(&key.secret().to_bytes()))
{
Ok(_) => {
debug!(log, "New network key generated and written to disk");
}
Err(e) => {
warn!(
log,
"Could not write node key to file: {:?}. Error: {}", network_key_f, e
);
}
}
}
local_private_key
}

3
beacon_node/http_server/Cargo.toml
View File

@@ -27,9 +27,8 @@ futures = "0.1.23"
serde = "1.0"
serde_derive = "1.0"
serde_json = "1.0"
slog = "^2.2.3"
slog = { version = "^2.2.3" , features = ["max_level_trace"] }
slog-term = "^2.4.0"
slog-async = "^2.3.0"
tokio = "0.1.17"
exit-future = "0.1.4"
crossbeam-channel = "0.3.8"

3
beacon_node/http_server/src/lib.rs
View File

@@ -14,6 +14,7 @@ use slog::{info, o, warn};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::runtime::TaskExecutor;
use tokio::sync::mpsc;
#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub struct HttpServerConfig {
@@ -75,7 +76,7 @@ pub fn create_iron_http_server<T: BeaconChainTypes + 'static>(
pub fn start_service<T: BeaconChainTypes + 'static>(
config: &HttpServerConfig,
executor: &TaskExecutor,
_network_chan: crossbeam_channel::Sender<NetworkMessage>,
_network_chan: mpsc::UnboundedSender<NetworkMessage>,
beacon_chain: Arc<BeaconChain<T>>,
db_path: PathBuf,
metrics_registry: Registry,

3
beacon_node/network/Cargo.toml
View File

@@ -13,10 +13,9 @@ store = { path = "../store" }
eth2-libp2p = { path = "../eth2-libp2p" }
version = { path = "../version" }
types = { path = "../../eth2/types" }
slog = { version = "^2.2.3" , features = ["max_level_trace", "release_max_level_debug"] }
slog = { version = "^2.2.3" , features = ["max_level_trace"] }
eth2_ssz = { path = "../../eth2/utils/ssz" }
tree_hash = { path = "../../eth2/utils/tree_hash" }
futures = "0.1.25"
error-chain = "0.12.0"
crossbeam-channel = "0.3.8"
tokio = "0.1.16"

33
beacon_node/network/src/message_handler.rs
View File

@@ -2,17 +2,18 @@ use crate::error;
use crate::service::{NetworkMessage, OutgoingMessage};
use crate::sync::SimpleSync;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use crossbeam_channel::{unbounded as channel, Sender};
use eth2_libp2p::{
behaviour::PubsubMessage,
rpc::{methods::GoodbyeReason, RPCRequest, RPCResponse, RequestId},
PeerId, RPCEvent,
};
use futures::future;
use futures::future::Future;
use futures::stream::Stream;
use slog::{debug, warn};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::mpsc;
/// Timeout for RPC requests.
// const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
@@ -48,13 +49,13 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
/// Initializes and runs the MessageHandler.
pub fn spawn(
beacon_chain: Arc<BeaconChain<T>>,
network_send: crossbeam_channel::Sender<NetworkMessage>,
network_send: mpsc::UnboundedSender<NetworkMessage>,
executor: &tokio::runtime::TaskExecutor,
log: slog::Logger,
) -> error::Result<Sender<HandlerMessage>> {
) -> error::Result<mpsc::UnboundedSender<HandlerMessage>> {
debug!(log, "Service starting");
let (handler_send, handler_recv) = channel();
let (handler_send, handler_recv) = mpsc::unbounded_channel();
// Initialise sync and begin processing in thread
// generate the Message handler
@@ -69,13 +70,13 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
// spawn handler task
// TODO: Handle manual termination of thread
executor.spawn(future::poll_fn(move || -> Result<_, _> {
loop {
handler.handle_message(handler_recv.recv().map_err(|_| {
executor.spawn(
handler_recv
.for_each(move |msg| Ok(handler.handle_message(msg)))
.map_err(move |_| {
debug!(log, "Network message handler terminated.");
})?);
}
}));
}),
);
Ok(handler_send)
}
@@ -222,7 +223,7 @@ impl<T: BeaconChainTypes + 'static> MessageHandler<T> {
pub struct NetworkContext {
/// The network channel to relay messages to the Network service.
network_send: crossbeam_channel::Sender<NetworkMessage>,
network_send: mpsc::UnboundedSender<NetworkMessage>,
/// A mapping of peers and the RPC id we have sent an RPC request to.
outstanding_outgoing_request_ids: HashMap<(PeerId, RequestId), Instant>,
/// Stores the next `RequestId` we should include on an outgoing `RPCRequest` to a `PeerId`.
@@ -232,7 +233,7 @@ pub struct NetworkContext {
}
impl NetworkContext {
pub fn new(network_send: crossbeam_channel::Sender<NetworkMessage>, log: slog::Logger) -> Self {
pub fn new(network_send: mpsc::UnboundedSender<NetworkMessage>, log: slog::Logger) -> Self {
Self {
network_send,
outstanding_outgoing_request_ids: HashMap::new(),
@@ -278,13 +279,13 @@ impl NetworkContext {
);
}
fn send_rpc_event(&self, peer_id: PeerId, rpc_event: RPCEvent) {
fn send_rpc_event(&mut self, peer_id: PeerId, rpc_event: RPCEvent) {
self.send(peer_id, OutgoingMessage::RPC(rpc_event))
}
fn send(&self, peer_id: PeerId, outgoing_message: OutgoingMessage) {
fn send(&mut self, peer_id: PeerId, outgoing_message: OutgoingMessage) {
self.network_send
.send(NetworkMessage::Send(peer_id, outgoing_message))
.try_send(NetworkMessage::Send(peer_id, outgoing_message))
.unwrap_or_else(|_| {
warn!(
self.log,

113
beacon_node/network/src/service.rs
View File

@@ -2,24 +2,23 @@ use crate::error;
use crate::message_handler::{HandlerMessage, MessageHandler};
use crate::NetworkConfig;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use crossbeam_channel::{unbounded as channel, Sender, TryRecvError};
use eth2_libp2p::Service as LibP2PService;
use eth2_libp2p::Topic;
use eth2_libp2p::{Libp2pEvent, PeerId};
use eth2_libp2p::{PubsubMessage, RPCEvent};
use futures::prelude::*;
use futures::sync::oneshot;
use futures::Stream;
use slog::{debug, info, o, trace};
use std::marker::PhantomData;
use std::sync::Arc;
use tokio::runtime::TaskExecutor;
use types::Topic;
use tokio::sync::{mpsc, oneshot};
/// Service that handles communication between internal services and the eth2_libp2p network service.
pub struct Service<T: BeaconChainTypes> {
//libp2p_service: Arc<Mutex<LibP2PService>>,
_libp2p_exit: oneshot::Sender<()>,
network_send: crossbeam_channel::Sender<NetworkMessage>,
network_send: mpsc::UnboundedSender<NetworkMessage>,
_phantom: PhantomData<T>, //message_handler: MessageHandler,
//message_handler_send: Sender<HandlerMessage>
}
@@ -30,9 +29,9 @@ impl<T: BeaconChainTypes + 'static> Service<T> {
config: &NetworkConfig,
executor: &TaskExecutor,
log: slog::Logger,
) -> error::Result<(Arc<Self>, Sender<NetworkMessage>)> {
) -> error::Result<(Arc<Self>, mpsc::UnboundedSender<NetworkMessage>)> {
// build the network channel
let (network_send, network_recv) = channel::<NetworkMessage>();
let (network_send, network_recv) = mpsc::unbounded_channel::<NetworkMessage>();
// launch message handler thread
let message_handler_log = log.new(o!("Service" => "MessageHandler"));
let message_handler_send = MessageHandler::spawn(
@@ -64,9 +63,9 @@ impl<T: BeaconChainTypes + 'static> Service<T> {
}
// TODO: Testing only
pub fn send_message(&self) {
pub fn send_message(&mut self) {
self.network_send
.send(NetworkMessage::Send(
.try_send(NetworkMessage::Send(
PeerId::random(),
OutgoingMessage::NotifierTest,
))
@@ -76,12 +75,12 @@ impl<T: BeaconChainTypes + 'static> Service<T> {
fn spawn_service(
libp2p_service: LibP2PService,
network_recv: crossbeam_channel::Receiver<NetworkMessage>,
message_handler_send: crossbeam_channel::Sender<HandlerMessage>,
network_recv: mpsc::UnboundedReceiver<NetworkMessage>,
message_handler_send: mpsc::UnboundedSender<HandlerMessage>,
executor: &TaskExecutor,
log: slog::Logger,
) -> error::Result<oneshot::Sender<()>> {
let (network_exit, exit_rx) = oneshot::channel();
) -> error::Result<tokio::sync::oneshot::Sender<()>> {
let (network_exit, exit_rx) = tokio::sync::oneshot::channel();
// spawn on the current executor
executor.spawn(
@@ -105,76 +104,76 @@ fn spawn_service(
//TODO: Potentially handle channel errors
fn network_service(
mut libp2p_service: LibP2PService,
network_recv: crossbeam_channel::Receiver<NetworkMessage>,
message_handler_send: crossbeam_channel::Sender<HandlerMessage>,
mut network_recv: mpsc::UnboundedReceiver<NetworkMessage>,
mut message_handler_send: mpsc::UnboundedSender<HandlerMessage>,
log: slog::Logger,
) -> impl futures::Future<Item = (), Error = eth2_libp2p::error::Error> {
futures::future::poll_fn(move || -> Result<_, eth2_libp2p::error::Error> {
// poll the swarm
loop {
// only end the loop once both major polls are not ready.
let mut not_ready_count = 0;
while not_ready_count < 2 {
not_ready_count = 0;
// poll the network channel
match network_recv.poll() {
Ok(Async::Ready(Some(message))) => {
match message {
// TODO: Testing message - remove
NetworkMessage::Send(peer_id, outgoing_message) => {
match outgoing_message {
OutgoingMessage::RPC(rpc_event) => {
trace!(log, "Sending RPC Event: {:?}", rpc_event);
//TODO: Make swarm private
//TODO: Implement correct peer id topic message handling
libp2p_service.swarm.send_rpc(peer_id, rpc_event);
}
OutgoingMessage::NotifierTest => {
// debug!(log, "Received message from notifier");
}
};
}
NetworkMessage::Publish { topics, message } => {
debug!(log, "Sending pubsub message"; "topics" => format!("{:?}",topics));
libp2p_service.swarm.publish(topics, *message);
}
}
}
Ok(Async::NotReady) => not_ready_count += 1,
Ok(Async::Ready(None)) => {
return Err(eth2_libp2p::error::Error::from("Network channel closed"));
}
Err(_) => {
return Err(eth2_libp2p::error::Error::from("Network channel error"));
}
}
// poll the swarm
match libp2p_service.poll() {
Ok(Async::Ready(Some(event))) => match event {
Libp2pEvent::RPC(peer_id, rpc_event) => {
trace!(log, "RPC Event: RPC message received: {:?}", rpc_event);
message_handler_send
.send(HandlerMessage::RPC(peer_id, rpc_event))
.try_send(HandlerMessage::RPC(peer_id, rpc_event))
.map_err(|_| "failed to send rpc to handler")?;
}
Libp2pEvent::PeerDialed(peer_id) => {
debug!(log, "Peer Dialed: {:?}", peer_id);
message_handler_send
.send(HandlerMessage::PeerDialed(peer_id))
.try_send(HandlerMessage::PeerDialed(peer_id))
.map_err(|_| "failed to send rpc to handler")?;
}
Libp2pEvent::Identified(peer_id, info) => {
debug!(
log,
"We have identified peer: {:?} with {:?}", peer_id, info
);
}
Libp2pEvent::PubsubMessage {
source, message, ..
} => {
//TODO: Decide if we need to propagate the topic upwards. (Potentially for
//attestations)
message_handler_send
.send(HandlerMessage::PubsubMessage(source, message))
.try_send(HandlerMessage::PubsubMessage(source, message))
.map_err(|_| " failed to send pubsub message to handler")?;
}
},
Ok(Async::Ready(None)) => unreachable!("Stream never ends"),
Ok(Async::NotReady) => break,
Err(_) => break,
}
}
// poll the network channel
// TODO: refactor - combine poll_fn's?
loop {
match network_recv.try_recv() {
// TODO: Testing message - remove
Ok(NetworkMessage::Send(peer_id, outgoing_message)) => {
match outgoing_message {
OutgoingMessage::RPC(rpc_event) => {
trace!(log, "Sending RPC Event: {:?}", rpc_event);
//TODO: Make swarm private
//TODO: Implement correct peer id topic message handling
libp2p_service.swarm.send_rpc(peer_id, rpc_event);
}
OutgoingMessage::NotifierTest => {
// debug!(log, "Received message from notifier");
}
};
}
Ok(NetworkMessage::Publish { topics, message }) => {
debug!(log, "Sending pubsub message on topics {:?}", topics);
libp2p_service.swarm.publish(topics, *message);
}
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
return Err(eth2_libp2p::error::Error::from(
"Network channel disconnected",
));
}
Ok(Async::NotReady) => not_ready_count += 1,
Err(_) => not_ready_count += 1,
}
}
Ok(Async::NotReady)

127
beacon_node/network/src/sync/import_queue.rs
View File

@@ -41,31 +41,23 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
}
}
/// Completes all possible partials into `BeaconBlock` and returns them, sorted by increasing
/// slot number. Does not delete the partials from the queue, this must be done manually.
///
/// Returns `(queue_index, block, sender)`:
///
/// - `block_root`: may be used to remove the entry if it is successfully processed.
/// - `block`: the completed block.
/// - `sender`: the `PeerId` the provided the `BeaconBlockBody` which completed the partial.
pub fn complete_blocks(&self) -> Vec<(Hash256, BeaconBlock, PeerId)> {
let mut complete: Vec<(Hash256, BeaconBlock, PeerId)> = self
.partials
.iter()
.filter_map(|(_, partial)| partial.clone().complete())
.collect();
// Sort the completable partials to be in ascending slot order.
complete.sort_unstable_by(|a, b| a.1.slot.partial_cmp(&b.1.slot).unwrap());
complete
}
/// Returns true of the if the `BlockRoot` is found in the `import_queue`.
pub fn contains_block_root(&self, block_root: Hash256) -> bool {
self.partials.contains_key(&block_root)
}
/// Attempts to complete the `BlockRoot` if it is found in the `import_queue`.
///
/// Returns an Enum with a `PartialBeaconBlockCompletion`.
/// Does not remove the `block_root` from the `import_queue`.
pub fn attempt_complete_block(&self, block_root: Hash256) -> PartialBeaconBlockCompletion {
if let Some(partial) = self.partials.get(&block_root) {
partial.attempt_complete()
} else {
PartialBeaconBlockCompletion::MissingRoot
}
}
/// Removes the first `PartialBeaconBlock` with a matching `block_root`, returning the partial
/// if it exists.
pub fn remove(&mut self, block_root: Hash256) -> Option<PartialBeaconBlock> {
@@ -102,6 +94,8 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
block_roots: &[BlockRootSlot],
sender: PeerId,
) -> Vec<BlockRootSlot> {
// TODO: This will currently not return a `BlockRootSlot` if this root exists but there is no header.
// It would be more robust if it did.
let new_block_root_slots: Vec<BlockRootSlot> = block_roots
.iter()
// Ignore any roots already stored in the queue.
@@ -135,12 +129,8 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
/// the queue and it's block root is included in the output.
///
/// If a `header` is already in the queue, but not yet processed by the chain the block root is
/// included in the output and the `inserted` time for the partial record is set to
/// not included in the output and the `inserted` time for the partial record is set to
/// `Instant::now()`. Updating the `inserted` time stops the partial from becoming stale.
///
/// Presently the queue enforces that a `BeaconBlockHeader` _must_ be received before its
/// `BeaconBlockBody`. This is not a natural requirement and we could enhance the queue to lift
/// this restraint.
pub fn enqueue_headers(
&mut self,
headers: Vec<BeaconBlockHeader>,
@@ -152,8 +142,10 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
let block_root = Hash256::from_slice(&header.canonical_root()[..]);
if self.chain_has_not_seen_block(&block_root) {
self.insert_header(block_root, header, sender.clone());
required_bodies.push(block_root);
if !self.insert_header(block_root, header, sender.clone()) {
// If a body is empty
required_bodies.push(block_root);
}
}
}
@@ -163,10 +155,17 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
/// If there is a matching `header` for this `body`, adds it to the queue.
///
/// If there is no `header` for the `body`, the body is simply discarded.
pub fn enqueue_bodies(&mut self, bodies: Vec<BeaconBlockBody>, sender: PeerId) {
pub fn enqueue_bodies(
&mut self,
bodies: Vec<BeaconBlockBody>,
sender: PeerId,
) -> Option<Hash256> {
let mut last_block_hash = None;
for body in bodies {
self.insert_body(body, sender.clone());
last_block_hash = self.insert_body(body, sender.clone());
}
last_block_hash
}
pub fn enqueue_full_blocks(&mut self, blocks: Vec<BeaconBlock>, sender: PeerId) {
@@ -179,12 +178,22 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
///
/// If the header already exists, the `inserted` time is set to `now` and not other
/// modifications are made.
fn insert_header(&mut self, block_root: Hash256, header: BeaconBlockHeader, sender: PeerId) {
/// Returns true is `body` exists.
fn insert_header(
&mut self,
block_root: Hash256,
header: BeaconBlockHeader,
sender: PeerId,
) -> bool {
let mut exists = false;
self.partials
.entry(block_root)
.and_modify(|partial| {
partial.header = Some(header.clone());
partial.inserted = Instant::now();
if partial.body.is_some() {
exists = true;
}
})
.or_insert_with(|| PartialBeaconBlock {
slot: header.slot,
@@ -194,28 +203,30 @@ impl<T: BeaconChainTypes> ImportQueue<T> {
inserted: Instant::now(),
sender,
});
exists
}
/// Updates an existing partial with the `body`.
///
/// If there is no header for the `body`, the body is simply discarded.
///
/// If the body already existed, the `inserted` time is set to `now`.
fn insert_body(&mut self, body: BeaconBlockBody, sender: PeerId) {
///
/// Returns the block hash of the inserted body
fn insert_body(&mut self, body: BeaconBlockBody, sender: PeerId) -> Option<Hash256> {
let body_root = Hash256::from_slice(&body.tree_hash_root()[..]);
let mut last_root = None;
self.partials.iter_mut().for_each(|(_, mut p)| {
self.partials.iter_mut().for_each(|(root, mut p)| {
if let Some(header) = &mut p.header {
if body_root == header.block_body_root {
p.inserted = Instant::now();
if p.body.is_none() {
p.body = Some(body.clone());
p.sender = sender.clone();
}
p.body = Some(body.clone());
p.sender = sender.clone();
last_root = Some(*root);
}
}
});
last_root
}
/// Updates an existing `partial` with the completed block, or adds a new (complete) partial.
@@ -257,13 +268,33 @@ pub struct PartialBeaconBlock {
}
impl PartialBeaconBlock {
/// Consumes `self` and returns a full built `BeaconBlock`, it's root and the `sender`
/// `PeerId`, if enough information exists to complete the block. Otherwise, returns `None`.
pub fn complete(self) -> Option<(Hash256, BeaconBlock, PeerId)> {
Some((
self.block_root,
self.header?.into_block(self.body?),
self.sender,
))
/// Attempts to build a block.
///
/// Does not comsume the `PartialBeaconBlock`.
pub fn attempt_complete(&self) -> PartialBeaconBlockCompletion {
if self.header.is_none() {
PartialBeaconBlockCompletion::MissingHeader(self.slot)
} else if self.body.is_none() {
PartialBeaconBlockCompletion::MissingBody
} else {
PartialBeaconBlockCompletion::Complete(
self.header
.clone()
.unwrap()
.into_block(self.body.clone().unwrap()),
)
}
}
}
/// The result of trying to convert a `BeaconBlock` into a `PartialBeaconBlock`.
pub enum PartialBeaconBlockCompletion {
/// The partial contains a valid BeaconBlock.
Complete(BeaconBlock),
/// The partial does not exist.
MissingRoot,
/// The partial contains a `BeaconBlockRoot` but no `BeaconBlockHeader`.
MissingHeader(Slot),
/// The partial contains a `BeaconBlockRoot` and `BeaconBlockHeader` but no `BeaconBlockBody`.
MissingBody,
}

238
beacon_node/network/src/sync/simple_sync.rs
View File

@@ -1,10 +1,10 @@
use super::import_queue::ImportQueue;
use super::import_queue::{ImportQueue, PartialBeaconBlockCompletion};
use crate::message_handler::NetworkContext;
use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
use eth2_libp2p::rpc::methods::*;
use eth2_libp2p::rpc::{RPCRequest, RPCResponse, RequestId};
use eth2_libp2p::PeerId;
use slog::{debug, error, info, o, warn};
use slog::{debug, error, info, o, trace, warn};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
@@ -17,7 +17,7 @@ use types::{
const SLOT_IMPORT_TOLERANCE: u64 = 100;
/// The amount of seconds a block (or partial block) may exist in the import queue.
const QUEUE_STALE_SECS: u64 = 6;
const QUEUE_STALE_SECS: u64 = 100;
/// If a block is more than `FUTURE_SLOT_TOLERANCE` slots ahead of our slot clock, we drop it.
/// Otherwise we queue it.
@@ -227,7 +227,12 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
//
// Therefore, there are some blocks between the local finalized epoch and the remote
// head that are worth downloading.
debug!(self.log, "UsefulPeer"; "peer" => format!("{:?}", peer_id));
debug!(
self.log, "UsefulPeer";
"peer" => format!("{:?}", peer_id),
"local_finalized_epoch" => local.latest_finalized_epoch,
"remote_latest_finalized_epoch" => remote.latest_finalized_epoch,
);
let start_slot = local
.latest_finalized_epoch
@@ -238,7 +243,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
peer_id,
BeaconBlockRootsRequest {
start_slot,
count: required_slots.into(),
count: required_slots.as_u64(),
},
network,
);
@@ -247,7 +252,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
fn root_at_slot(&self, target_slot: Slot) -> Option<Hash256> {
self.chain
.rev_iter_block_roots(target_slot)
.rev_iter_best_block_roots(target_slot)
.take(1)
.find(|(_root, slot)| *slot == target_slot)
.map(|(root, _slot)| root)
@@ -271,8 +276,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
let mut roots: Vec<BlockRootSlot> = self
.chain
.rev_iter_block_roots(req.start_slot + req.count)
.skip(1)
.rev_iter_best_block_roots(req.start_slot + req.count)
.take(req.count as usize)
.map(|(block_root, slot)| BlockRootSlot { slot, block_root })
.collect();
@@ -356,7 +360,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
BeaconBlockHeadersRequest {
start_root: first.block_root,
start_slot: first.slot,
max_headers: (last.slot - first.slot).as_u64(),
max_headers: (last.slot - first.slot + 1).as_u64(),
skip_slots: 0,
},
network,
@@ -386,7 +390,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
// unnecessary block deserialization when `req.skip_slots > 0`.
let mut roots: Vec<Hash256> = self
.chain
.rev_iter_block_roots(req.start_slot + (count - 1))
.rev_iter_best_block_roots(req.start_slot + count)
.take(count as usize)
.map(|(root, _slot)| root)
.collect();
@@ -499,14 +503,26 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
"count" => res.block_bodies.len(),
);
self.import_queue
.enqueue_bodies(res.block_bodies, peer_id.clone());
if !res.block_bodies.is_empty() {
// Import all blocks to queue
let last_root = self
.import_queue
.enqueue_bodies(res.block_bodies, peer_id.clone());
// Attempt to process all recieved bodies by recursively processing the latest block
if let Some(root) = last_root {
match self.attempt_process_partial_block(peer_id, root, network, &"rpc") {
Some(BlockProcessingOutcome::Processed { block_root: _ }) => {
// If processing is successful remove from `import_queue`
self.import_queue.remove(root);
}
_ => {}
}
}
}
// Clear out old entries
self.import_queue.remove_stale();
// Import blocks, if possible.
self.process_import_queue(network);
}
/// Process a gossip message declaring a new block.
@@ -526,26 +542,35 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
match outcome {
BlockProcessingOutcome::Processed { .. } => SHOULD_FORWARD_GOSSIP_BLOCK,
BlockProcessingOutcome::ParentUnknown { parent } => {
// Clean the stale entries from the queue.
self.import_queue.remove_stale();
// Add this block to the queue
self.import_queue
.enqueue_full_blocks(vec![block], peer_id.clone());
.enqueue_full_blocks(vec![block.clone()], peer_id.clone());
debug!(
self.log, "RequestParentBlock";
"parent_root" => format!("{}", parent),
"parent_slot" => block.slot - 1,
"peer" => format!("{:?}", peer_id),
);
// Unless the parent is in the queue, request the parent block from the peer.
//
// It is likely that this is duplicate work, given we already send a hello
// request. However, I believe there are some edge-cases where the hello
// message doesn't suffice, so we perform this request as well.
if !self.import_queue.contains_block_root(parent) {
// Send a hello to learn of the clients best slot so we can then sync the required
// parent(s).
network.send_rpc_request(
peer_id.clone(),
RPCRequest::Hello(hello_message(&self.chain)),
);
}
// Request roots between parent and start of finality from peer.
let start_slot = self
.chain
.head()
.beacon_state
.finalized_epoch
.start_slot(T::EthSpec::slots_per_epoch());
self.request_block_roots(
peer_id,
BeaconBlockRootsRequest {
// Request blocks between `latest_finalized_slot` and the `block`
start_slot,
count: block.slot.as_u64() - start_slot.as_u64(),
},
network,
);
// Clean the stale entries from the queue.
self.import_queue.remove_stale();
SHOULD_FORWARD_GOSSIP_BLOCK
}
@@ -587,40 +612,6 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
}
}
/// Iterate through the `import_queue` and process any complete blocks.
///
/// If a block is successfully processed it is removed from the queue, otherwise it remains in
/// the queue.
pub fn process_import_queue(&mut self, network: &mut NetworkContext) {
let mut successful = 0;
// Loop through all of the complete blocks in the queue.
for (block_root, block, sender) in self.import_queue.complete_blocks() {
let processing_result = self.process_block(sender, block.clone(), network, &"gossip");
let should_dequeue = match processing_result {
Some(BlockProcessingOutcome::ParentUnknown { .. }) => false,
Some(BlockProcessingOutcome::FutureSlot {
present_slot,
block_slot,
}) if present_slot + FUTURE_SLOT_TOLERANCE >= block_slot => false,
_ => true,
};
if processing_result == Some(BlockProcessingOutcome::Processed { block_root }) {
successful += 1;
}
if should_dequeue {
self.import_queue.remove(block_root);
}
}
if successful > 0 {
info!(self.log, "Imported {} blocks", successful)
}
}
/// Request some `BeaconBlockRoots` from the remote peer.
fn request_block_roots(
&mut self,
@@ -695,6 +686,89 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
hello_message(&self.chain)
}
/// Helper function to attempt to process a partial block.
///
/// If the block can be completed recursively call `process_block`
/// else request missing parts.
fn attempt_process_partial_block(
&mut self,
peer_id: PeerId,
block_root: Hash256,
network: &mut NetworkContext,
source: &str,
) -> Option<BlockProcessingOutcome> {
match self.import_queue.attempt_complete_block(block_root) {
PartialBeaconBlockCompletion::MissingBody => {
// Unable to complete the block because the block body is missing.
debug!(
self.log, "RequestParentBody";
"source" => source,
"block_root" => format!("{}", block_root),
"peer" => format!("{:?}", peer_id),
);
// Request the block body from the peer.
self.request_block_bodies(
peer_id,
BeaconBlockBodiesRequest {
block_roots: vec![block_root],
},
network,
);
None
}
PartialBeaconBlockCompletion::MissingHeader(slot) => {
// Unable to complete the block because the block header is missing.
debug!(
self.log, "RequestParentHeader";
"source" => source,
"block_root" => format!("{}", block_root),
"peer" => format!("{:?}", peer_id),
);
// Request the block header from the peer.
self.request_block_headers(
peer_id,
BeaconBlockHeadersRequest {
start_root: block_root,
start_slot: slot,
max_headers: 1,
skip_slots: 0,
},
network,
);
None
}
PartialBeaconBlockCompletion::MissingRoot => {
// The `block_root` is not known to the queue.
debug!(
self.log, "MissingParentRoot";
"source" => source,
"block_root" => format!("{}", block_root),
"peer" => format!("{:?}", peer_id),
);
// Do nothing.
None
}
PartialBeaconBlockCompletion::Complete(block) => {
// The block exists in the queue, attempt to process it
trace!(
self.log, "AttemptProcessParent";
"source" => source,
"block_root" => format!("{}", block_root),
"parent_slot" => block.slot,
"peer" => format!("{:?}", peer_id),
);
self.process_block(peer_id.clone(), block, network, source)
}
}
}
/// Processes the `block` that was received from `peer_id`.
///
/// If the block was submitted to the beacon chain without internal error, `Some(outcome)` is
@@ -721,6 +795,7 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
if let Ok(outcome) = processing_result {
match outcome {
BlockProcessingOutcome::Processed { block_root } => {
// The block was valid and we processed it successfully.
debug!(
self.log, "Imported block from network";
"source" => source,
@@ -730,26 +805,29 @@ impl<T: BeaconChainTypes> SimpleSync<T> {
);
}
BlockProcessingOutcome::ParentUnknown { parent } => {
// The block was valid and we processed it successfully.
debug!(
// The parent has not been processed
trace!(
self.log, "ParentBlockUnknown";
"source" => source,
"parent_root" => format!("{}", parent),
"baby_block_slot" => block.slot,
"peer" => format!("{:?}", peer_id),
);
// Unless the parent is in the queue, request the parent block from the peer.
//
// It is likely that this is duplicate work, given we already send a hello
// request. However, I believe there are some edge-cases where the hello
// message doesn't suffice, so we perform this request as well.
if !self.import_queue.contains_block_root(parent) {
// Send a hello to learn of the clients best slot so we can then sync the require
// parent(s).
network.send_rpc_request(
peer_id.clone(),
RPCRequest::Hello(hello_message(&self.chain)),
);
// If the parent is in the `import_queue` attempt to complete it then process it.
match self.attempt_process_partial_block(peer_id, parent, network, source) {
// If processing parent is sucessful, re-process block and remove parent from queue
Some(BlockProcessingOutcome::Processed { block_root: _ }) => {
self.import_queue.remove(parent);
// Attempt to process `block` again
match self.chain.process_block(block) {
Ok(outcome) => return Some(outcome),
Err(_) => return None,
}
}
// All other cases leave `parent` in `import_queue` and return original outcome.
_ => {}
}
}
BlockProcessingOutcome::FutureSlot {

3
beacon_node/rpc/Cargo.toml
View File

@@ -22,9 +22,8 @@ dirs = "1.0.3"
futures = "0.1.23"
serde = "1.0"
serde_derive = "1.0"
slog = "^2.2.3"
slog = { version = "^2.2.3" , features = ["max_level_trace"] }
slog-term = "^2.4.0"
slog-async = "^2.3.0"
tokio = "0.1.17"
exit-future = "0.1.4"
crossbeam-channel = "0.3.8"

14
beacon_node/rpc/src/attestation.rs
View File

@@ -1,5 +1,7 @@
use beacon_chain::{BeaconChain, BeaconChainTypes};
use eth2_libp2p::PubsubMessage;
use eth2_libp2p::TopicBuilder;
use eth2_libp2p::BEACON_ATTESTATION_TOPIC;
use futures::Future;
use grpcio::{RpcContext, RpcStatus, RpcStatusCode, UnarySink};
use network::NetworkMessage;
@@ -11,12 +13,13 @@ use protos::services_grpc::AttestationService;
use slog::{error, info, trace, warn};
use ssz::{ssz_encode, Decode};
use std::sync::Arc;
use tokio::sync::mpsc;
use types::Attestation;
#[derive(Clone)]
pub struct AttestationServiceInstance<T: BeaconChainTypes> {
pub chain: Arc<BeaconChain<T>>,
pub network_chan: crossbeam_channel::Sender<NetworkMessage>,
pub network_chan: mpsc::UnboundedSender<NetworkMessage>,
pub log: slog::Logger,
}
@@ -136,13 +139,12 @@ impl<T: BeaconChainTypes> AttestationService for AttestationServiceInstance<T> {
"type" => "valid_attestation",
);
// TODO: Obtain topics from the network service properly.
let topic = types::TopicBuilder::new("beacon_chain".to_string()).build();
// valid attestation, propagate to the network
let topic = TopicBuilder::new(BEACON_ATTESTATION_TOPIC).build();
let message = PubsubMessage::Attestation(attestation);
// Publish the attestation to the p2p network via gossipsub.
self.network_chan
.send(NetworkMessage::Publish {
.try_send(NetworkMessage::Publish {
topics: vec![topic],
message: Box::new(message),
})
@@ -150,7 +152,7 @@ impl<T: BeaconChainTypes> AttestationService for AttestationServiceInstance<T> {
error!(
self.log,
"PublishAttestation";
"type" => "failed to publish to gossipsub",
"type" => "failed to publish attestation to gossipsub",
"error" => format!("{:?}", e)
);
});

14
beacon_node/rpc/src/beacon_block.rs
View File

@@ -1,6 +1,6 @@
use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessingOutcome};
use crossbeam_channel;
use eth2_libp2p::PubsubMessage;
use eth2_libp2p::BEACON_PUBSUB_TOPIC;
use eth2_libp2p::{PubsubMessage, TopicBuilder};
use futures::Future;
use grpcio::{RpcContext, RpcStatus, RpcStatusCode, UnarySink};
use network::NetworkMessage;
@@ -13,12 +13,13 @@ use slog::Logger;
use slog::{error, info, trace, warn};
use ssz::{ssz_encode, Decode};
use std::sync::Arc;
use tokio::sync::mpsc;
use types::{BeaconBlock, Signature, Slot};
#[derive(Clone)]
pub struct BeaconBlockServiceInstance<T: BeaconChainTypes> {
pub chain: Arc<BeaconChain<T>>,
pub network_chan: crossbeam_channel::Sender<NetworkMessage>,
pub network_chan: mpsc::UnboundedSender<NetworkMessage>,
pub log: Logger,
}
@@ -104,14 +105,13 @@ impl<T: BeaconChainTypes> BeaconBlockService for BeaconBlockServiceInstance<T> {
"block_root" => format!("{}", block_root),
);
// TODO: Obtain topics from the network service properly.
let topic =
types::TopicBuilder::new("beacon_chain".to_string()).build();
// get the network topic to send on
let topic = TopicBuilder::new(BEACON_PUBSUB_TOPIC).build();
let message = PubsubMessage::Block(block);
// Publish the block to the p2p network via gossipsub.
self.network_chan
.send(NetworkMessage::Publish {
.try_send(NetworkMessage::Publish {
topics: vec![topic],
message: Box::new(message),
})

5
beacon_node/rpc/src/lib.rs
View File

@@ -20,11 +20,12 @@ use protos::services_grpc::{
use slog::{info, o, warn};
use std::sync::Arc;
use tokio::runtime::TaskExecutor;
use tokio::sync::mpsc;
pub fn start_server<T: BeaconChainTypes + Clone + 'static>(
config: &RPCConfig,
executor: &TaskExecutor,
network_chan: crossbeam_channel::Sender<NetworkMessage>,
network_chan: mpsc::UnboundedSender<NetworkMessage>,
beacon_chain: Arc<BeaconChain<T>>,
log: &slog::Logger,
) -> exit_future::Signal {
@@ -60,8 +61,8 @@ pub fn start_server<T: BeaconChainTypes + Clone + 'static>(
};
let attestation_service = {
let instance = AttestationServiceInstance {
chain: beacon_chain.clone(),
network_chan,
chain: beacon_chain.clone(),
log: log.clone(),
};
create_attestation_service(instance)

134
beacon_node/src/main.rs
View File

@@ -1,11 +1,11 @@
extern crate slog;
mod run;
use clap::{App, Arg};
use client::{ClientConfig, Eth2Config};
use eth2_config::{get_data_dir, read_from_file, write_to_file};
use slog::{crit, o, Drain};
use env_logger::{Builder, Env};
use eth2_config::{read_from_file, write_to_file};
use slog::{crit, o, Drain, Level};
use std::fs;
use std::path::PathBuf;
pub const DEFAULT_DATA_DIR: &str = ".lighthouse";
@@ -14,10 +14,8 @@ pub const CLIENT_CONFIG_FILENAME: &str = "beacon-node.toml";
pub const ETH2_CONFIG_FILENAME: &str = "eth2-spec.toml";
fn main() {
let decorator = slog_term::TermDecorator::new().build();
let drain = slog_term::CompactFormat::new(decorator).build().fuse();
let drain = slog_async::Async::new(drain).build().fuse();
let logger = slog::Logger::root(drain, o!());
// debugging output for libp2p and external crates
Builder::from_env(Env::default()).init();
let matches = App::new("Lighthouse")
.version(version::version().as_str())
@@ -30,21 +28,55 @@ fn main() {
.value_name("DIR")
.help("Data directory for keys and databases.")
.takes_value(true)
.default_value(DEFAULT_DATA_DIR),
)
.arg(
Arg::with_name("logfile")
.long("logfile")
.value_name("logfile")
.help("File path where output will be written.")
.takes_value(true),
)
// network related arguments
.arg(
Arg::with_name("listen-address")
.long("listen-address")
.value_name("Listen Address")
.help("One or more comma-delimited multi-addresses to listen for p2p connections.")
.value_name("Address")
.help("The address lighthouse will listen for UDP and TCP connections. (default 127.0.0.1).")
.takes_value(true),
)
.arg(
Arg::with_name("maxpeers")
.long("maxpeers")
.help("The maximum number of peers (default 10).")
.takes_value(true),
)
.arg(
Arg::with_name("boot-nodes")
.long("boot-nodes")
.allow_hyphen_values(true)
.value_name("BOOTNODES")
.help("One or more comma-delimited multi-addresses to bootstrap the p2p network.")
.help("One or more comma-delimited base64-encoded ENR's to bootstrap the p2p network.")
.takes_value(true),
)
.arg(
Arg::with_name("port")
.long("port")
.value_name("Lighthouse Port")
.help("The TCP/UDP port to listen on. The UDP port can be modified by the --discovery-port flag.")
.takes_value(true),
)
.arg(
Arg::with_name("discovery-port")
.long("disc-port")
.value_name("DiscoveryPort")
.help("The discovery UDP port.")
.takes_value(true),
)
.arg(
Arg::with_name("discovery-address")
.long("discovery-address")
.value_name("Address")
.help("The IP address to broadcast to other peers on how to reach this node.")
.takes_value(true),
)
// rpc related arguments
@@ -58,14 +90,13 @@ fn main() {
.arg(
Arg::with_name("rpc-address")
.long("rpc-address")
.value_name("RPCADDRESS")
.value_name("Address")
.help("Listen address for RPC endpoint.")
.takes_value(true),
)
.arg(
Arg::with_name("rpc-port")
.long("rpc-port")
.value_name("RPCPORT")
.help("Listen port for RPC endpoint.")
.takes_value(true),
)
@@ -73,21 +104,19 @@ fn main() {
.arg(
Arg::with_name("http")
.long("http")
.value_name("HTTP")
.help("Enable the HTTP server.")
.takes_value(false),
)
.arg(
Arg::with_name("http-address")
.long("http-address")
.value_name("HTTPADDRESS")
.value_name("Address")
.help("Listen address for the HTTP server.")
.takes_value(true),
)
.arg(
Arg::with_name("http-port")
.long("http-port")
.value_name("HTTPPORT")
.help("Listen port for the HTTP server.")
.takes_value(true),
)
@@ -116,19 +145,60 @@ fn main() {
.short("r")
.help("When present, genesis will be within 30 minutes prior. Only for testing"),
)
.arg(
Arg::with_name("verbosity")
.short("v")
.multiple(true)
.help("Sets the verbosity level")
.takes_value(true),
)
.get_matches();
let data_dir = match get_data_dir(&matches, PathBuf::from(DEFAULT_DATA_DIR)) {
Ok(dir) => dir,
Err(e) => {
crit!(logger, "Failed to initialize data dir"; "error" => format!("{:?}", e));
return;
// build the initial logger
let decorator = slog_term::TermDecorator::new().build();
let drain = slog_term::CompactFormat::new(decorator).build().fuse();
let drain = slog_async::Async::new(drain).build();
let drain = match matches.occurrences_of("verbosity") {
0 => drain.filter_level(Level::Info),
1 => drain.filter_level(Level::Debug),
2 => drain.filter_level(Level::Trace),
_ => drain.filter_level(Level::Trace),
};
let mut log = slog::Logger::root(drain.fuse(), o!());
let data_dir = match matches
.value_of("datadir")
.and_then(|v| Some(PathBuf::from(v)))
{
Some(v) => v,
None => {
// use the default
let mut default_dir = match dirs::home_dir() {
Some(v) => v,
None => {
crit!(log, "Failed to find a home directory");
return;
}
};
default_dir.push(DEFAULT_DATA_DIR);
PathBuf::from(default_dir)
}
};
// create the directory if needed
match fs::create_dir_all(&data_dir) {
Ok(_) => {}
Err(e) => {
crit!(log, "Failed to initialize data dir"; "error" => format!("{}", e));
return;
}
}
let client_config_path = data_dir.join(CLIENT_CONFIG_FILENAME);
// Attempt to lead the `ClientConfig` from disk.
// Attempt to load the `ClientConfig` from disk.
//
// If file doesn't exist, create a new, default one.
let mut client_config = match read_from_file::<ClientConfig>(client_config_path.clone()) {
@@ -136,13 +206,13 @@ fn main() {
Ok(None) => {
let default = ClientConfig::default();
if let Err(e) = write_to_file(client_config_path, &default) {
crit!(logger, "Failed to write default ClientConfig to file"; "error" => format!("{:?}", e));
crit!(log, "Failed to write default ClientConfig to file"; "error" => format!("{:?}", e));
return;
}
default
}
Err(e) => {
crit!(logger, "Failed to load a ChainConfig file"; "error" => format!("{:?}", e));
crit!(log, "Failed to load a ChainConfig file"; "error" => format!("{:?}", e));
return;
}
};
@@ -151,10 +221,10 @@ fn main() {
client_config.data_dir = data_dir.clone();
// Update the client config with any CLI args.
match client_config.apply_cli_args(&matches) {
match client_config.apply_cli_args(&matches, &mut log) {
Ok(()) => (),
Err(s) => {
crit!(logger, "Failed to parse ClientConfig CLI arguments"; "error" => s);
crit!(log, "Failed to parse ClientConfig CLI arguments"; "error" => s);
return;
}
};
@@ -173,13 +243,13 @@ fn main() {
_ => unreachable!(), // Guarded by slog.
};
if let Err(e) = write_to_file(eth2_config_path, &default) {
crit!(logger, "Failed to write default Eth2Config to file"; "error" => format!("{:?}", e));
crit!(log, "Failed to write default Eth2Config to file"; "error" => format!("{:?}", e));
return;
}
default
}
Err(e) => {
crit!(logger, "Failed to load/generate an Eth2Config"; "error" => format!("{:?}", e));
crit!(log, "Failed to load/generate an Eth2Config"; "error" => format!("{:?}", e));
return;
}
};
@@ -188,13 +258,13 @@ fn main() {
match eth2_config.apply_cli_args(&matches) {
Ok(()) => (),
Err(s) => {
crit!(logger, "Failed to parse Eth2Config CLI arguments"; "error" => s);
crit!(log, "Failed to parse Eth2Config CLI arguments"; "error" => s);
return;
}
};
match run::run_beacon_node(client_config, eth2_config, &logger) {
match run::run_beacon_node(client_config, eth2_config, &log) {
Ok(_) => {}
Err(e) => crit!(logger, "Beacon node failed to start"; "reason" => format!("{:}", e)),
Err(e) => crit!(log, "Beacon node failed to start"; "reason" => format!("{:}", e)),
}
}

20
beacon_node/src/run.rs
View File

@@ -41,6 +41,15 @@ pub fn run_beacon_node(
"This software is EXPERIMENTAL and provides no guarantees or warranties."
);
info!(
log,
"Starting beacon node";
"p2p_listen_address" => format!("{:?}", &other_client_config.network.listen_address),
"data_dir" => format!("{:?}", other_client_config.data_dir()),
"spec_constants" => &spec_constants,
"db_type" => &other_client_config.db_type,
);
let result = match (db_type.as_str(), spec_constants.as_str()) {
("disk", "minimal") => run::<ClientType<DiskStore, MinimalEthSpec>>(
&db_path,
@@ -80,17 +89,6 @@ pub fn run_beacon_node(
}
};
if result.is_ok() {
info!(
log,
"Started beacon node";
"p2p_listen_addresses" => format!("{:?}", &other_client_config.network.listen_addresses()),
"data_dir" => format!("{:?}", other_client_config.data_dir()),
"spec_constants" => &spec_constants,
"db_type" => &other_client_config.db_type,
);
}
result
}

167
beacon_node/store/src/iter.rs
View File

@@ -15,15 +15,15 @@ impl<'a, T: EthSpec, U: Store> StateRootsIterator<'a, T, U> {
Self {
store,
beacon_state: Cow::Borrowed(beacon_state),
slot: start_slot,
slot: start_slot + 1,
}
}
pub fn owned(store: Arc<U>, beacon_state: BeaconState<T>, start_slot: Slot) -> Self {
Self {
slot: start_slot,
beacon_state: Cow::Owned(beacon_state),
store,
beacon_state: Cow::Owned(beacon_state),
slot: start_slot + 1,
}
}
}
@@ -90,13 +90,19 @@ impl<'a, T: EthSpec, U: Store> Iterator for BlockIterator<'a, T, U> {
}
}
/// Iterates backwards through block roots.
/// Iterates backwards through block roots. If any specified slot is unable to be retrieved, the
/// iterator returns `None` indefinitely.
///
/// Uses the `latest_block_roots` field of `BeaconState` to as the source of block roots and will
/// perform a lookup on the `Store` for a prior `BeaconState` if `latest_block_roots` has been
/// exhausted.
///
/// Returns `None` for roots prior to genesis or when there is an error reading from `Store`.
///
/// ## Notes
///
/// See [`BestBlockRootsIterator`](struct.BestBlockRootsIterator.html), which has different
/// `start_slot` logic.
#[derive(Clone)]
pub struct BlockRootsIterator<'a, T: EthSpec, U> {
store: Arc<U>,
@@ -108,18 +114,18 @@ impl<'a, T: EthSpec, U: Store> BlockRootsIterator<'a, T, U> {
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn new(store: Arc<U>, beacon_state: &'a BeaconState<T>, start_slot: Slot) -> Self {
Self {
slot: start_slot,
beacon_state: Cow::Borrowed(beacon_state),
store,
beacon_state: Cow::Borrowed(beacon_state),
slot: start_slot + 1,
}
}
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn owned(store: Arc<U>, beacon_state: BeaconState<T>, start_slot: Slot) -> Self {
Self {
slot: start_slot,
beacon_state: Cow::Owned(beacon_state),
store,
beacon_state: Cow::Owned(beacon_state),
slot: start_slot + 1,
}
}
}
@@ -156,6 +162,104 @@ impl<'a, T: EthSpec, U: Store> Iterator for BlockRootsIterator<'a, T, U> {
}
}
/// Iterates backwards through block roots with `start_slot` highest possible value
/// `<= beacon_state.slot`.
///
/// The distinction between `BestBlockRootsIterator` and `BlockRootsIterator` is:
///
/// - `BestBlockRootsIterator` uses best-effort slot. When `start_slot` is greater than the latest available block root
/// on `beacon_state`, returns `Some(root, slot)` where `slot` is the latest available block
/// root.
/// - `BlockRootsIterator` is strict about `start_slot`. When `start_slot` is greater than the latest available block root
/// on `beacon_state`, returns `None`.
///
/// This is distinct from `BestBlockRootsIterator`.
///
/// Uses the `latest_block_roots` field of `BeaconState` to as the source of block roots and will
/// perform a lookup on the `Store` for a prior `BeaconState` if `latest_block_roots` has been
/// exhausted.
///
/// Returns `None` for roots prior to genesis or when there is an error reading from `Store`.
#[derive(Clone)]
pub struct BestBlockRootsIterator<'a, T: EthSpec, U> {
store: Arc<U>,
beacon_state: Cow<'a, BeaconState<T>>,
slot: Slot,
}
impl<'a, T: EthSpec, U: Store> BestBlockRootsIterator<'a, T, U> {
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn new(store: Arc<U>, beacon_state: &'a BeaconState<T>, start_slot: Slot) -> Self {
let mut slot = start_slot;
if slot >= beacon_state.slot {
// Slot may be too high.
slot = beacon_state.slot;
if beacon_state.get_block_root(slot).is_err() {
slot -= 1;
}
}
Self {
store,
beacon_state: Cow::Borrowed(beacon_state),
slot: slot + 1,
}
}
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn owned(store: Arc<U>, beacon_state: BeaconState<T>, start_slot: Slot) -> Self {
let mut slot = start_slot;
if slot >= beacon_state.slot {
// Slot may be too high.
slot = beacon_state.slot;
// TODO: Use a function other than `get_block_root` as this will always return `Err()`
// for slot = state.slot.
if beacon_state.get_block_root(slot).is_err() {
slot -= 1;
}
}
Self {
store,
beacon_state: Cow::Owned(beacon_state),
slot: slot + 1,
}
}
}
impl<'a, T: EthSpec, U: Store> Iterator for BestBlockRootsIterator<'a, T, U> {
type Item = (Hash256, Slot);
fn next(&mut self) -> Option<Self::Item> {
if self.slot == 0 {
// End of Iterator
return None;
}
self.slot -= 1;
match self.beacon_state.get_block_root(self.slot) {
Ok(root) => Some((*root, self.slot)),
Err(BeaconStateError::SlotOutOfBounds) => {
// Read a `BeaconState` from the store that has access to prior historical root.
let beacon_state: BeaconState<T> = {
// Load the earliest state from disk.
let new_state_root = self.beacon_state.get_oldest_state_root().ok()?;
self.store.get(&new_state_root).ok()?
}?;
self.beacon_state = Cow::Owned(beacon_state);
let root = self.beacon_state.get_block_root(self.slot).ok()?;
Some((*root, self.slot))
}
_ => None,
}
}
}
#[cfg(test)]
mod test {
use super::*;
@@ -206,7 +310,50 @@ mod test {
let mut collected: Vec<(Hash256, Slot)> = iter.collect();
collected.reverse();
let expected_len = 2 * MainnetEthSpec::slots_per_historical_root() - 1;
let expected_len = 2 * MainnetEthSpec::slots_per_historical_root();
assert_eq!(collected.len(), expected_len);
for i in 0..expected_len {
assert_eq!(collected[i].0, Hash256::from(i as u64));
}
}
#[test]
fn best_block_root_iter() {
let store = Arc::new(MemoryStore::open());
let slots_per_historical_root = MainnetEthSpec::slots_per_historical_root();
let mut state_a: BeaconState<MainnetEthSpec> = get_state();
let mut state_b: BeaconState<MainnetEthSpec> = get_state();
state_a.slot = Slot::from(slots_per_historical_root);
state_b.slot = Slot::from(slots_per_historical_root * 2);
let mut hashes = (0..).into_iter().map(|i| Hash256::from(i));
for root in &mut state_a.latest_block_roots[..] {
*root = hashes.next().unwrap()
}
for root in &mut state_b.latest_block_roots[..] {
*root = hashes.next().unwrap()
}
let state_a_root = hashes.next().unwrap();
state_b.latest_state_roots[0] = state_a_root;
store.put(&state_a_root, &state_a).unwrap();
let iter = BestBlockRootsIterator::new(store.clone(), &state_b, state_b.slot);
assert!(
iter.clone().find(|(_root, slot)| *slot == 0).is_some(),
"iter should contain zero slot"
);
let mut collected: Vec<(Hash256, Slot)> = iter.collect();
collected.reverse();
let expected_len = 2 * MainnetEthSpec::slots_per_historical_root();
assert_eq!(collected.len(), expected_len);
@@ -255,7 +402,7 @@ mod test {
let mut collected: Vec<(Hash256, Slot)> = iter.collect();
collected.reverse();
let expected_len = MainnetEthSpec::slots_per_historical_root() * 2 - 1;
let expected_len = MainnetEthSpec::slots_per_historical_root() * 2;
assert_eq!(collected.len(), expected_len, "collection length incorrect");