cerc-io/lighthouse
11
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into disk-db

Browse Source
This commit is contained in:
Paul Hauner
2019-04-30 16:02:23 +10:00
parent d6664cb4ac f68dff0e52
commit 05df7702d3
205 changed files with 19474 additions and 2246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
beacon_node/beacon_chain/Cargo.toml
+1
View File
@@ -23,4 +23,5 @@ serde_json = "1.0"
slot_clock = { path = "../../eth2/utils/slot_clock" }
ssz = { path = "../../eth2/utils/ssz" }
state_processing = { path = "../../eth2/state_processing" }
tree_hash = { path = "../../eth2/utils/tree_hash" }
types = { path = "../../eth2/types" }
beacon_node/beacon_chain/src/beacon_chain.rs
+7 -16
View File
@@ -303,8 +303,6 @@ where
/// then having it iteratively updated -- in such a case it's possible for another thread to
/// find the state at an old slot.
pub fn update_state(&self, mut state: BeaconState) -> Result<(), Error> {
let latest_block_header = self.head().beacon_block.block_header();
let present_slot = match self.slot_clock.present_slot() {
Ok(Some(slot)) => slot,
_ => return Err(Error::UnableToReadSlot),
@@ -312,7 +310,7 @@ where
// If required, transition the new state to the present slot.
for _ in state.slot.as_u64()..present_slot.as_u64() {
per_slot_processing(&mut state, &latest_block_header, &self.spec)?;
per_slot_processing(&mut state, &self.spec)?;
}
state.build_all_caches(&self.spec)?;
@@ -324,8 +322,6 @@ where
/// Ensures the current canonical `BeaconState` has been transitioned to match the `slot_clock`.
pub fn catchup_state(&self) -> Result<(), Error> {
let latest_block_header = self.head().beacon_block.block_header();
let present_slot = match self.slot_clock.present_slot() {
Ok(Some(slot)) => slot,
_ => return Err(Error::UnableToReadSlot),
@@ -339,7 +335,7 @@ where
state.build_epoch_cache(RelativeEpoch::NextWithoutRegistryChange, &self.spec)?;
state.build_epoch_cache(RelativeEpoch::NextWithRegistryChange, &self.spec)?;
per_slot_processing(&mut *state, &latest_block_header, &self.spec)?;
per_slot_processing(&mut *state, &self.spec)?;
}
state.build_all_caches(&self.spec)?;
@@ -497,21 +493,17 @@ where
} else {
// If the current head block is not from this slot, use the slot from the previous
// epoch.
let root = *self.state.read().get_block_root(
*self.state.read().get_block_root(
current_epoch_start_slot - self.spec.slots_per_epoch,
&self.spec,
)?;
root
)?
}
} else {
// If we're not on the first slot of the epoch.
let root = *self
*self
.state
.read()
.get_block_root(current_epoch_start_slot, &self.spec)?;
root
.get_block_root(current_epoch_start_slot, &self.spec)?
};
Ok(AttestationData {
@@ -621,9 +613,8 @@ where
// Transition the parent state to the block slot.
let mut state = parent_state;
let previous_block_header = parent_block.block_header();
for _ in state.slot.as_u64()..block.slot.as_u64() {
if let Err(e) = per_slot_processing(&mut state, &previous_block_header, &self.spec) {
if let Err(e) = per_slot_processing(&mut state, &self.spec) {
return Ok(BlockProcessingOutcome::InvalidBlock(
InvalidBlock::SlotProcessingError(e),
));
beacon_node/beacon_chain/src/initialise.rs
+3 -3
View File
@@ -7,9 +7,9 @@ use db::stores::{BeaconBlockStore, BeaconStateStore};
use db::{DiskDB, MemoryDB};
use fork_choice::BitwiseLMDGhost;
use slot_clock::SystemTimeSlotClock;
use ssz::TreeHash;
use std::path::PathBuf;
use std::sync::Arc;
use tree_hash::TreeHash;
use types::test_utils::TestingBeaconStateBuilder;
use types::{BeaconBlock, ChainSpec, Hash256};
@@ -32,7 +32,7 @@ pub fn initialise_beacon_chain(
let (genesis_state, _keypairs) = state_builder.build();
let mut genesis_block = BeaconBlock::empty(&spec);
genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
genesis_block.state_root = Hash256::from_slice(&genesis_state.tree_hash_root());
// Slot clock
let slot_clock = SystemTimeSlotClock::new(
@@ -73,7 +73,7 @@ pub fn initialise_test_beacon_chain_with_memory_db(
let (genesis_state, _keypairs) = state_builder.build();
let mut genesis_block = BeaconBlock::empty(spec);
genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
genesis_block.state_root = Hash256::from_slice(&genesis_state.tree_hash_root());
// Slot clock
let slot_clock = SystemTimeSlotClock::new(
beacon_node/beacon_chain/src/test_utils/testing_beacon_chain_builder.rs
+2 -2
View File
@@ -5,8 +5,8 @@ use db::{
};
use fork_choice::BitwiseLMDGhost;
use slot_clock::TestingSlotClock;
use ssz::TreeHash;
use std::sync::Arc;
use tree_hash::TreeHash;
use types::test_utils::TestingBeaconStateBuilder;
use types::*;
@@ -27,7 +27,7 @@ impl TestingBeaconChainBuilder {
let (genesis_state, _keypairs) = self.state_builder.build();
let mut genesis_block = BeaconBlock::empty(&spec);
genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
genesis_block.state_root = Hash256::from_slice(&genesis_state.tree_hash_root());
// Create the Beacon Chain
BeaconChain::from_genesis(
beacon_node/beacon_chain/test_harness/Cargo.toml
+1
View File
@@ -38,5 +38,6 @@ serde_json = "1.0"
serde_yaml = "0.8"
slot_clock = { path = "../../../eth2/utils/slot_clock" }
ssz = { path = "../../../eth2/utils/ssz" }
tree_hash = { path = "../../../eth2/utils/tree_hash" }
types = { path = "../../../eth2/types" }
yaml-rust = "0.4.2"
beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs
+5 -6
View File
@@ -9,8 +9,8 @@ use fork_choice::BitwiseLMDGhost;
use log::debug;
use rayon::prelude::*;
use slot_clock::TestingSlotClock;
use ssz::TreeHash;
use std::sync::Arc;
use tree_hash::TreeHash;
use types::{test_utils::TestingBeaconStateBuilder, *};
type TestingBeaconChain = BeaconChain<MemoryDB, TestingSlotClock, BitwiseLMDGhost<MemoryDB>>;
@@ -54,7 +54,7 @@ impl BeaconChainHarness {
let (mut genesis_state, keypairs) = state_builder.build();
let mut genesis_block = BeaconBlock::empty(&spec);
genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
genesis_block.state_root = Hash256::from_slice(&genesis_state.tree_hash_root());
genesis_state
.build_epoch_cache(RelativeEpoch::Previous, &spec)
@@ -163,7 +163,7 @@ impl BeaconChainHarness {
data: data.clone(),
custody_bit: false,
}
.hash_tree_root();
.tree_hash_root();
let domain = self.spec.get_domain(
state.slot.epoch(self.spec.slots_per_epoch),
Domain::Attestation,
@@ -211,9 +211,8 @@ impl BeaconChainHarness {
// Ensure the validators slot clock is accurate.
self.validators[proposer].set_slot(present_slot);
let block = self.validators[proposer].produce_block().unwrap();
block
self.validators[proposer].produce_block().unwrap()
}
/// Advances the chain with a BeaconBlock and attestations from all validators.
@@ -245,7 +244,7 @@ impl BeaconChainHarness {
.for_each(|(i, attestation)| {
self.beacon_chain
.process_attestation(attestation.clone())
.expect(&format!("Attestation {} invalid: {:?}", i, attestation));
.unwrap_or_else(|_| panic!("Attestation {} invalid: {:?}", i, attestation));
});
debug!("Attestations processed.");
beacon_node/beacon_chain/test_harness/src/lib.rs
+1 -1
View File
@@ -8,7 +8,7 @@
//! producing blocks and attestations.
//!
//! Example:
//! ```
//! ```rust,no_run
//! use test_harness::BeaconChainHarness;
//! use types::ChainSpec;
//!
beacon_node/beacon_chain/test_harness/src/test_case.rs
+1 -1
View File
@@ -4,7 +4,7 @@
use crate::beacon_chain_harness::BeaconChainHarness;
use beacon_chain::CheckPoint;
use log::{info, warn};
use ssz::SignedRoot;
use tree_hash::SignedRoot;
use types::*;
use types::test_utils::*;
beacon_node/beacon_chain/test_harness/src/test_case/state_check.rs
+1
View File
@@ -52,6 +52,7 @@ impl StateCheck {
/// # Panics
///
/// Panics with an error message if any test fails.
#[allow(clippy::cyclomatic_complexity)]
pub fn assert_valid(&self, state: &BeaconState, spec: &ChainSpec) {
let state_epoch = state.slot.epoch(spec.slots_per_epoch);
beacon_node/beacon_chain/test_harness/src/validator_harness/direct_beacon_node.rs
-8
View File
@@ -14,9 +14,6 @@ use slot_clock::SlotClock;
use std::sync::Arc;
use types::{AttestationData, BeaconBlock, FreeAttestation, Signature, Slot};
// mod attester;
// mod producer;
/// Connect directly to a borrowed `BeaconChain` instance so an attester/producer can request/submit
/// blocks/attestations.
///
@@ -42,11 +39,6 @@ impl<T: ClientDB, U: SlotClock, F: ForkChoice> DirectBeaconNode<T, U, F> {
pub fn last_published_block(&self) -> Option<BeaconBlock> {
Some(self.published_blocks.read().last()?.clone())
}
/// Get the last published attestation (if any).
pub fn last_published_free_attestation(&self) -> Option<FreeAttestation> {
Some(self.published_attestations.read().last()?.clone())
}
}
impl<T: ClientDB, U: SlotClock, F: ForkChoice> AttesterBeaconNode for DirectBeaconNode<T, U, F> {
beacon_node/beacon_chain/test_harness/src/validator_harness/mod.rs
+2 -24
View File
@@ -2,8 +2,7 @@ mod direct_beacon_node;
mod direct_duties;
mod local_signer;
use attester::PollOutcome as AttestationPollOutcome;
use attester::{Attester, Error as AttestationPollError};
use attester::Attester;
use beacon_chain::BeaconChain;
use block_proposer::PollOutcome as BlockPollOutcome;
use block_proposer::{BlockProducer, Error as BlockPollError};
@@ -14,7 +13,7 @@ use fork_choice::BitwiseLMDGhost;
use local_signer::LocalSigner;
use slot_clock::TestingSlotClock;
use std::sync::Arc;
use types::{BeaconBlock, ChainSpec, FreeAttestation, Keypair, Slot};
use types::{BeaconBlock, ChainSpec, Keypair, Slot};
#[derive(Debug, PartialEq)]
pub enum BlockProduceError {
@@ -22,12 +21,6 @@ pub enum BlockProduceError {
PollError(BlockPollError),
}
#[derive(Debug, PartialEq)]
pub enum AttestationProduceError {
DidNotProduce(AttestationPollOutcome),
PollError(AttestationPollError),
}
type TestingBlockProducer = BlockProducer<
TestingSlotClock,
DirectBeaconNode<MemoryDB, TestingSlotClock, BitwiseLMDGhost<MemoryDB>>,
@@ -117,21 +110,6 @@ impl ValidatorHarness {
.expect("Unable to obtain produced block."))
}
/// Run the `poll` function on the `Attester` and produce a `FreeAttestation`.
///
/// An error is returned if the attester refuses to attest.
pub fn produce_free_attestation(&mut self) -> Result<FreeAttestation, AttestationProduceError> {
match self.attester.poll() {
Ok(AttestationPollOutcome::AttestationProduced(_)) => {}
Ok(outcome) => return Err(AttestationProduceError::DidNotProduce(outcome)),
Err(error) => return Err(AttestationProduceError::PollError(error)),
};
Ok(self
.beacon_node
.last_published_free_attestation()
.expect("Unable to obtain produced attestation."))
}
/// Set the validators slot clock to the specified slot.
///
/// The validators slot clock will always read this value until it is set to something else.
beacon_node/beacon_chain/test_harness/tests/chain.rs
+2
View File
@@ -1,3 +1,5 @@
#![cfg(not(debug_assertions))]
use env_logger::{Builder, Env};
use log::debug;
use test_harness::BeaconChainHarness;
beacon_node/client/src/client_config.rs
+13 -1
View File
@@ -10,6 +10,7 @@ use std::path::PathBuf;
use types::multiaddr::Protocol;
use types::multiaddr::ToMultiaddr;
use types::ChainSpec;
use types::Multiaddr;
/// Stores the client configuration for this Lighthouse instance.
#[derive(Debug, Clone)]
@@ -76,7 +77,7 @@ impl ClientConfig {
}
// Custom listening address ipv4/ipv6
// TODO: Handle list of addresses
if let Some(listen_address_str) = args.value_of("listen_address") {
if let Some(listen_address_str) = args.value_of("listen-address") {
if let Ok(listen_address) = listen_address_str.parse::<IpAddr>() {
let multiaddr = SocketAddr::new(listen_address, config.net_conf.listen_port)
.to_multiaddr()
@@ -88,6 +89,17 @@ impl ClientConfig {
}
}
// Custom bootnodes
// TODO: Handle list of addresses
if let Some(boot_addresses_str) = args.value_of("boot-nodes") {
if let Ok(boot_address) = boot_addresses_str.parse::<Multiaddr>() {
config.net_conf.boot_nodes.append(&mut vec![boot_address]);
} else {
error!(log, "Invalid Bootnode multiaddress"; "Multiaddr" => boot_addresses_str);
return Err("Invalid IP Address");
}
}
/* Filesystem related arguments */
// Custom datadir
beacon_node/client/src/lib.rs
+10 -11
View File
@@ -15,21 +15,19 @@ use futures::{future::Future, Stream};
use network::Service as NetworkService;
use slog::{error, info, o};
use slot_clock::SlotClock;
use ssz::TreeHash;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::runtime::TaskExecutor;
use tokio::timer::Interval;
use types::Hash256;
/// Main beacon node client service. This provides the connection and initialisation of the clients
/// sub-services in multiple threads.
pub struct Client<T: ClientTypes> {
/// Configuration for the lighthouse client.
config: ClientConfig,
_config: ClientConfig,
/// The beacon chain for the running client.
beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>,
_beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>,
/// Reference to the network service.
pub network: Arc<NetworkService>,
/// Signal to terminate the RPC server.
@@ -92,17 +90,18 @@ impl<TClientType: ClientTypes> Client<TClientType> {
network_logger,
)?;
let mut rpc_exit_signal = None;
// spawn the RPC server
if config.rpc_conf.enabled {
rpc_exit_signal = Some(rpc::start_server(
let rpc_exit_signal = if config.rpc_conf.enabled {
Some(rpc::start_server(
&config.rpc_conf,
executor,
network_send,
beacon_chain.clone(),
&log,
));
}
))
} else {
None
};
let (slot_timer_exit_signal, exit) = exit_future::signal();
if let Ok(Some(duration_to_next_slot)) = beacon_chain.slot_clock.duration_to_next_slot() {
@@ -131,8 +130,8 @@ impl<TClientType: ClientTypes> Client<TClientType> {
}
Ok(Client {
config,
beacon_chain,
_config: config,
_beacon_chain: beacon_chain,
rpc_exit_signal,
slot_timer_exit_signal: Some(slot_timer_exit_signal),
log,
beacon_node/client/src/notifier.rs
+1 -1
View File
@@ -14,7 +14,7 @@ pub fn run<T: ClientTypes>(client: &Client<T>, executor: TaskExecutor, exit: Exi
// notification heartbeat
let interval = Interval::new(Instant::now(), Duration::from_secs(5));
let log = client.log.new(o!("Service" => "Notifier"));
let _log = client.log.new(o!("Service" => "Notifier"));
// TODO: Debugging only
let counter = Arc::new(Mutex::new(0));
beacon_node/db/src/stores/beacon_block_store.rs
+2 -2
View File
@@ -1,6 +1,6 @@
use super::BLOCKS_DB_COLUMN as DB_COLUMN;
use super::{ClientDB, DBError};
use ssz::Decodable;
use ssz::decode;
use std::sync::Arc;
use types::{BeaconBlock, Hash256, Slot};
@@ -30,7 +30,7 @@ impl<T: ClientDB> BeaconBlockStore<T> {
match self.get(&hash)? {
None => Ok(None),
Some(ssz) => {
let (block, _) = BeaconBlock::ssz_decode(&ssz, 0).map_err(|_| DBError {
let block = decode::<BeaconBlock>(&ssz).map_err(|_| DBError {
message: "Bad BeaconBlock SSZ.".to_string(),
})?;
Ok(Some(block))
beacon_node/db/src/stores/beacon_state_store.rs
+2 -2
View File
@@ -1,6 +1,6 @@
use super::STATES_DB_COLUMN as DB_COLUMN;
use super::{ClientDB, DBError};
use ssz::Decodable;
use ssz::decode;
use std::sync::Arc;
use types::{BeaconState, Hash256};
@@ -23,7 +23,7 @@ impl<T: ClientDB> BeaconStateStore<T> {
match self.get(&hash)? {
None => Ok(None),
Some(ssz) => {
let (state, _) = BeaconState::ssz_decode(&ssz, 0).map_err(|_| DBError {
let state = decode::<BeaconState>(&ssz).map_err(|_| DBError {
message: "Bad State SSZ.".to_string(),
})?;
Ok(Some(state))
beacon_node/db/src/stores/validator_store.rs
+3 -3
View File
@@ -4,7 +4,7 @@ use self::bytes::{BufMut, BytesMut};
use super::VALIDATOR_DB_COLUMN as DB_COLUMN;
use super::{ClientDB, DBError};
use bls::PublicKey;
use ssz::{ssz_encode, Decodable};
use ssz::{decode, ssz_encode};
use std::sync::Arc;
#[derive(Debug, PartialEq)]
@@ -69,8 +69,8 @@ impl<T: ClientDB> ValidatorStore<T> {
let val = self.db.get(DB_COLUMN, &key[..])?;
match val {
None => Ok(None),
Some(val) => match PublicKey::ssz_decode(&val, 0) {
Ok((key, _)) => Ok(Some(key)),
Some(val) => match decode::<PublicKey>(&val) {
Ok(key) => Ok(Some(key)),
Err(_) => Err(ValidatorStoreError::DecodeError),
},
}
beacon_node/eth2-libp2p/src/behaviour.rs
+7 -12
View File
@@ -65,17 +65,11 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubE
self.events.push(BehaviourEvent::GossipMessage {
source: gs_msg.source,
topics: gs_msg.topics,
message: pubsub_message,
message: Box::new(pubsub_message),
});
}
GossipsubEvent::Subscribed {
peer_id: _,
topic: _,
}
| GossipsubEvent::Unsubscribed {
peer_id: _,
topic: _,
} => {}
GossipsubEvent::Subscribed { .. } => {}
GossipsubEvent::Unsubscribed { .. } => {}
}
}
}
@@ -110,7 +104,8 @@ impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<IdentifyEv
);
info.listen_addrs.truncate(20);
}
self.events.push(BehaviourEvent::Identified(peer_id, info));
self.events
.push(BehaviourEvent::Identified(peer_id, Box::new(info)));
}
IdentifyEvent::Error { .. } => {}
IdentifyEvent::SendBack { .. } => {}
@@ -183,12 +178,12 @@ impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
pub enum BehaviourEvent {
RPC(PeerId, RPCEvent),
PeerDialed(PeerId),
Identified(PeerId, IdentifyInfo),
Identified(PeerId, Box<IdentifyInfo>),
// TODO: This is a stub at the moment
GossipMessage {
source: PeerId,
topics: Vec<TopicHash>,
message: PubsubMessage,
message: Box<PubsubMessage>,
},
}
beacon_node/eth2-libp2p/src/rpc/methods.rs
+1 -1
View File
@@ -1,7 +1,7 @@
use ssz::{Decodable, DecodeError, Encodable, SszStream};
/// Available RPC methods types and ids.
use ssz_derive::{Decode, Encode};
use types::{Attestation, BeaconBlockBody, BeaconBlockHeader, Epoch, Hash256, Slot};
use types::{BeaconBlockBody, BeaconBlockHeader, Epoch, Hash256, Slot};
#[derive(Debug)]
/// Available Serenity Libp2p RPC methods
beacon_node/eth2-libp2p/src/rpc/mod.rs
+3 -3
View File
@@ -26,7 +26,7 @@ pub struct Rpc<TSubstream> {
/// Pins the generic substream.
marker: PhantomData<TSubstream>,
/// Slog logger for RPC behaviour.
log: slog::Logger,
_log: slog::Logger,
}
impl<TSubstream> Rpc<TSubstream> {
@@ -35,7 +35,7 @@ impl<TSubstream> Rpc<TSubstream> {
Rpc {
events: Vec::new(),
marker: PhantomData,
log,
_log: log,
}
}
@@ -65,7 +65,7 @@ where
fn inject_connected(&mut self, peer_id: PeerId, connected_point: ConnectedPoint) {
// if initialised the connection, report this upwards to send the HELLO request
if let ConnectedPoint::Dialer { address: _ } = connected_point {
if let ConnectedPoint::Dialer { .. } = connected_point {
self.events.push(NetworkBehaviourAction::GenerateEvent(
RPCMessage::PeerDialed(peer_id),
));
beacon_node/eth2-libp2p/src/rpc/protocol.rs
+10 -6
View File
@@ -31,7 +31,7 @@ impl Default for RPCProtocol {
}
/// A monotonic counter for ordering `RPCRequest`s.
#[derive(Debug, Clone, PartialEq, Default)]
#[derive(Debug, Clone, Default)]
pub struct RequestId(u64);
impl RequestId {
@@ -48,6 +48,12 @@ impl RequestId {
impl Eq for RequestId {}
impl PartialEq for RequestId {
fn eq(&self, other: &RequestId) -> bool {
self.0 == other.0
}
}
impl Hash for RequestId {
fn hash<H: Hasher>(&self, state: &mut H) {
self.0.hash(state);
@@ -104,17 +110,15 @@ impl UpgradeInfo for RPCEvent {
}
}
type FnDecodeRPCEvent = fn(Vec<u8>, ()) -> Result<RPCEvent, DecodeError>;
impl<TSocket> InboundUpgrade<TSocket> for RPCProtocol
where
TSocket: AsyncRead + AsyncWrite,
{
type Output = RPCEvent;
type Error = DecodeError;
type Future = upgrade::ReadOneThen<
upgrade::Negotiated<TSocket>,
(),
fn(Vec<u8>, ()) -> Result<RPCEvent, DecodeError>,
>;
type Future = upgrade::ReadOneThen<upgrade::Negotiated<TSocket>, (), FnDecodeRPCEvent>;
fn upgrade_inbound(self, socket: upgrade::Negotiated<TSocket>, _: Self::Info) -> Self::Future {
upgrade::read_one_then(socket, MAX_READ_SIZE, (), |packet, ()| Ok(decode(packet)?))
beacon_node/eth2-libp2p/src/service.rs
+8 -5
View File
@@ -19,13 +19,16 @@ use std::io::{Error, ErrorKind};
use std::time::Duration;
use types::{TopicBuilder, TopicHash};
type Libp2pStream = Boxed<(PeerId, StreamMuxerBox), Error>;
type Libp2pBehaviour = Behaviour<Substream<StreamMuxerBox>>;
/// The configuration and state of the libp2p components for the beacon node.
pub struct Service {
/// The libp2p Swarm handler.
//TODO: Make this private
pub swarm: Swarm<Boxed<(PeerId, StreamMuxerBox), Error>, Behaviour<Substream<StreamMuxerBox>>>,
pub swarm: Swarm<Libp2pStream, Libp2pBehaviour>,
/// This node's PeerId.
local_peer_id: PeerId,
_local_peer_id: PeerId,
/// The libp2p logger handle.
pub log: slog::Logger,
}
@@ -89,7 +92,7 @@ impl Service {
info!(log, "Subscribed to topics: {:?}", subscribed_topics);
Ok(Service {
local_peer_id,
_local_peer_id: local_peer_id,
swarm,
log,
})
@@ -179,11 +182,11 @@ pub enum Libp2pEvent {
/// Initiated the connection to a new peer.
PeerDialed(PeerId),
/// Received information about a peer on the network.
Identified(PeerId, IdentifyInfo),
Identified(PeerId, Box<IdentifyInfo>),
/// Received pubsub message.
PubsubMessage {
source: PeerId,
topics: Vec<TopicHash>,
message: PubsubMessage,
message: Box<PubsubMessage>,
},
}
beacon_node/network/Cargo.toml
+1
View File
@@ -15,6 +15,7 @@ version = { path = "../version" }
types = { path = "../../eth2/types" }
slog = { version = "^2.2.3" , features = ["max_level_trace", "release_max_level_debug"] }
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"
beacon_node/network/src/message_handler.rs
+2 -2
View File
@@ -41,7 +41,7 @@ pub enum HandlerMessage {
/// An RPC response/request has been received.
RPC(PeerId, RPCEvent),
/// A gossip message has been received.
PubsubMessage(PeerId, PubsubMessage),
PubsubMessage(PeerId, Box<PubsubMessage>),
}
impl MessageHandler {
@@ -93,7 +93,7 @@ impl MessageHandler {
}
// we have received an RPC message request/response
HandlerMessage::PubsubMessage(peer_id, gossip) => {
self.handle_gossip(peer_id, gossip);
self.handle_gossip(peer_id, *gossip);
}
//TODO: Handle all messages
_ => {}
beacon_node/network/src/service.rs
+5 -7
View File
@@ -17,7 +17,7 @@ use types::Topic;
/// Service that handles communication between internal services and the eth2_libp2p network service.
pub struct Service {
//libp2p_service: Arc<Mutex<LibP2PService>>,
libp2p_exit: oneshot::Sender<()>,
_libp2p_exit: oneshot::Sender<()>,
network_send: crossbeam_channel::Sender<NetworkMessage>,
//message_handler: MessageHandler,
//message_handler_send: Sender<HandlerMessage>,
@@ -54,7 +54,7 @@ impl Service {
log,
)?;
let network_service = Service {
libp2p_exit,
_libp2p_exit: libp2p_exit,
network_send: network_send.clone(),
};
@@ -131,9 +131,7 @@ fn network_service(
);
}
Libp2pEvent::PubsubMessage {
source,
topics: _,
message,
source, message, ..
} => {
//TODO: Decide if we need to propagate the topic upwards. (Potentially for
//attestations)
@@ -167,7 +165,7 @@ fn network_service(
}
Ok(NetworkMessage::Publish { topics, message }) => {
debug!(log, "Sending pubsub message on topics {:?}", topics);
libp2p_service.swarm.publish(topics, message);
libp2p_service.swarm.publish(topics, *message);
}
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
@@ -190,7 +188,7 @@ pub enum NetworkMessage {
/// Publish a message to pubsub mechanism.
Publish {
topics: Vec<Topic>,
message: PubsubMessage,
message: Box<PubsubMessage>,
},
}
beacon_node/network/src/sync/import_queue.rs
+5 -5
View File
@@ -2,9 +2,9 @@ use crate::beacon_chain::BeaconChain;
use eth2_libp2p::rpc::methods::*;
use eth2_libp2p::PeerId;
use slog::{debug, error};
use ssz::TreeHash;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tree_hash::TreeHash;
use types::{BeaconBlock, BeaconBlockBody, BeaconBlockHeader, Hash256, Slot};
/// Provides a queue for fully and partially built `BeaconBlock`s.
@@ -15,7 +15,7 @@ use types::{BeaconBlock, BeaconBlockBody, BeaconBlockHeader, Hash256, Slot};
///
/// - When we receive a `BeaconBlockBody`, the only way we can find it's matching
/// `BeaconBlockHeader` is to find a header such that `header.beacon_block_body ==
/// hash_tree_root(body)`. Therefore, if we used a `HashMap` we would need to use the root of
/// tree_hash_root(body)`. Therefore, if we used a `HashMap` we would need to use the root of
/// `BeaconBlockBody` as the key.
/// - It is possible for multiple distinct blocks to have identical `BeaconBlockBodies`. Therefore
/// we cannot use a `HashMap` keyed by the root of `BeaconBlockBody`.
@@ -166,7 +166,7 @@ impl ImportQueue {
let mut required_bodies: Vec<Hash256> = vec![];
for header in headers {
let block_root = Hash256::from_slice(&header.hash_tree_root()[..]);
let block_root = Hash256::from_slice(&header.tree_hash_root()[..]);
if self.chain_has_not_seen_block(&block_root) {
self.insert_header(block_root, header, sender.clone());
@@ -230,7 +230,7 @@ impl ImportQueue {
///
/// If the body already existed, the `inserted` time is set to `now`.
fn insert_body(&mut self, body: BeaconBlockBody, sender: PeerId) {
let body_root = Hash256::from_slice(&body.hash_tree_root()[..]);
let body_root = Hash256::from_slice(&body.tree_hash_root()[..]);
self.partials.iter_mut().for_each(|mut p| {
if let Some(header) = &mut p.header {
@@ -250,7 +250,7 @@ impl ImportQueue {
///
/// If the partial already existed, the `inserted` time is set to `now`.
fn insert_full_block(&mut self, block: BeaconBlock, sender: PeerId) {
let block_root = Hash256::from_slice(&block.hash_tree_root()[..]);
let block_root = Hash256::from_slice(&block.tree_hash_root()[..]);
let partial = PartialBeaconBlock {
slot: block.slot,
beacon_node/network/src/sync/simple_sync.rs
+33 -5
View File
@@ -5,17 +5,17 @@ use eth2_libp2p::rpc::methods::*;
use eth2_libp2p::rpc::{RPCRequest, RPCResponse, RequestId};
use eth2_libp2p::PeerId;
use slog::{debug, error, info, o, warn};
use ssz::TreeHash;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tree_hash::TreeHash;
use types::{Attestation, BeaconBlock, Epoch, Hash256, Slot};
/// The number of slots that we can import blocks ahead of us, before going into full Sync mode.
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 = 60;
const QUEUE_STALE_SECS: u64 = 600;
/// If a block is more than `FUTURE_SLOT_TOLERANCE` slots ahead of our slot clock, we drop it.
/// Otherwise we queue it.
@@ -65,7 +65,7 @@ pub enum PeerStatus {
}
impl PeerStatus {
pub fn should_handshake(&self) -> bool {
pub fn should_handshake(self) -> bool {
match self {
PeerStatus::DifferentNetworkId => false,
PeerStatus::FinalizedEpochNotInChain => false,
@@ -565,7 +565,7 @@ impl SimpleSync {
return false;
}
let block_root = Hash256::from_slice(&block.hash_tree_root());
let block_root = Hash256::from_slice(&block.tree_hash_root());
// Ignore any block that the chain already knows about.
if self.chain_has_seen_block(&block_root) {
@@ -582,7 +582,26 @@ impl SimpleSync {
);
match self.chain.process_block(block.clone()) {
Ok(BlockProcessingOutcome::InvalidBlock(InvalidBlock::ParentUnknown)) => {
// get the parent.
// The block was valid and we processed it successfully.
debug!(
self.log, "NewGossipBlock";
"msg" => "parent block unknown",
"parent_root" => format!("{}", block.previous_block_root),
"peer" => format!("{:?}", peer_id),
);
// Queue the block for later processing.
self.import_queue
.enqueue_full_blocks(vec![block], peer_id.clone());
// 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(self.chain.hello_message()),
);
// Forward the block onto our peers.
//
// Note: this may need to be changed if we decide to only forward blocks if we have
// all required info.
true
}
Ok(BlockProcessingOutcome::InvalidBlock(InvalidBlock::FutureSlot {
@@ -710,12 +729,21 @@ impl SimpleSync {
"reason" => format!("{:?}", outcome),
);
network.disconnect(sender, GoodbyeReason::Fault);
break;
}
// If this results to true, the item will be removed from the queue.
if outcome.sucessfully_processed() {
successful += 1;
self.import_queue.remove(block_root);
} else {
debug!(
self.log,
"ProcessImportQueue";
"msg" => "Block not imported",
"outcome" => format!("{:?}", outcome),
"peer" => format!("{:?}", sender),
);
}
}
Err(e) => {
beacon_node/rpc/src/attestation.rs
+4 -6
View File
@@ -43,9 +43,9 @@ impl AttestationService for AttestationServiceInstance {
let f = sink
.fail(RpcStatus::new(
RpcStatusCode::OutOfRange,
Some(format!(
"AttestationData request for a slot that is in the future."
)),
Some(
"AttestationData request for a slot that is in the future.".to_string(),
),
))
.map_err(move |e| {
error!(log_clone, "Failed to reply with failure {:?}: {:?}", req, e)
@@ -58,9 +58,7 @@ impl AttestationService for AttestationServiceInstance {
let f = sink
.fail(RpcStatus::new(
RpcStatusCode::InvalidArgument,
Some(format!(
"AttestationData request for a slot that is in the past."
)),
Some("AttestationData request for a slot that is in the past.".to_string()),
))
.map_err(move |e| {
error!(log_clone, "Failed to reply with failure {:?}: {:?}", req, e)
beacon_node/rpc/src/beacon_block.rs
+2 -2
View File
@@ -43,7 +43,7 @@ impl BeaconBlockService for BeaconBlockServiceInstance {
let f = sink
.fail(RpcStatus::new(
RpcStatusCode::InvalidArgument,
Some(format!("Invalid randao reveal signature")),
Some("Invalid randao reveal signature".to_string()),
))
.map_err(move |e| warn!(log_clone, "failed to reply {:?}: {:?}", req, e));
return ctx.spawn(f);
@@ -114,7 +114,7 @@ impl BeaconBlockService for BeaconBlockServiceInstance {
self.network_chan
.send(NetworkMessage::Publish {
topics: vec![topic],
message,
message: Box::new(message),
})
.unwrap_or_else(|e| {
error!(
beacon_node/rpc/src/beacon_node.rs
+2 -2
View File
@@ -24,7 +24,7 @@ impl BeaconNodeService for BeaconNodeServiceInstance {
// get the chain state
let state = self.chain.get_state();
let state_fork = state.fork.clone();
let genesis_time = state.genesis_time.clone();
let genesis_time = state.genesis_time;
// build the rpc fork struct
let mut fork = Fork::new();
@@ -35,7 +35,7 @@ impl BeaconNodeService for BeaconNodeServiceInstance {
node_info.set_fork(fork);
node_info.set_genesis_time(genesis_time);
node_info.set_genesis_slot(self.chain.get_spec().genesis_slot.as_u64());
node_info.set_chain_id(self.chain.get_spec().chain_id as u32);
node_info.set_chain_id(u32::from(self.chain.get_spec().chain_id));
// send the node_info the requester
let error_log = self.log.clone();
beacon_node/rpc/src/validator.rs
+4 -4
View File
@@ -5,7 +5,7 @@ use grpcio::{RpcContext, RpcStatus, RpcStatusCode, UnarySink};
use protos::services::{ActiveValidator, GetDutiesRequest, GetDutiesResponse, ValidatorDuty};
use protos::services_grpc::ValidatorService;
use slog::{trace, warn};
use ssz::Decodable;
use ssz::decode;
use std::sync::Arc;
use types::{Epoch, RelativeEpoch};
@@ -74,8 +74,8 @@ impl ValidatorService for ValidatorServiceInstance {
for validator_pk in validators.get_public_keys() {
let mut active_validator = ActiveValidator::new();
let public_key = match PublicKey::ssz_decode(validator_pk, 0) {
Ok((v, _index)) => v,
let public_key = match decode::<PublicKey>(validator_pk) {
Ok(v) => v,
Err(_) => {
let log_clone = self.log.clone();
let f = sink
@@ -83,7 +83,7 @@ impl ValidatorService for ValidatorServiceInstance {
RpcStatusCode::InvalidArgument,
Some("Invalid public_key".to_string()),
))
.map_err(move |e| warn!(log_clone, "failed to reply {:?}", req));
.map_err(move |_| warn!(log_clone, "failed to reply {:?}", req));
return ctx.spawn(f);
}
};
beacon_node/src/main.rs
+11 -1
View File
@@ -16,6 +16,7 @@ fn main() {
.version(version::version().as_str())
.author("Sigma Prime <contact@sigmaprime.io>")
.about("Eth 2.0 Client")
// file system related arguments
.arg(
Arg::with_name("datadir")
.long("datadir")
@@ -23,8 +24,9 @@ fn main() {
.help("Data directory for keys and databases.")
.takes_value(true),
)
// network related arguments
.arg(
Arg::with_name("listen_address")
Arg::with_name("listen-address")
.long("listen-address")
.value_name("Listen Address")
.help("The Network address to listen for p2p connections.")
@@ -37,6 +39,14 @@ fn main() {
.help("Network listen port for p2p connections.")
.takes_value(true),
)
.arg(
Arg::with_name("boot-nodes")
.long("boot-nodes")
.value_name("BOOTNODES")
.help("A list of comma separated multi addresses representing bootnodes to connect to.")
.takes_value(true),
)
// rpc related arguments
.arg(
Arg::with_name("rpc")
.long("rpc")