gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-20 21:34:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge current master and fix ssz-fuzzing

Browse Source
This commit is contained in:
Kirk Baird
2019-03-26 14:53:50 +11:00
parent 1bdd737a11 209614e373
commit 116d995960
190 changed files with 7696 additions and 3882 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
beacon_node/Cargo.toml
View File

@@ -1,24 +1,19 @@
[package]
name = "beacon_node"
version = "0.1.0"
authors = ["Paul Hauner <paul@paulhauner.com>"]
authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com"]
edition = "2018"
[dependencies]
bls = { path = "../eth2/utils/bls" }
beacon_chain = { path = "beacon_chain" }
grpcio = { version = "0.4", default-features = false, features = ["protobuf-codec"] }
protobuf = "2.0.2"
protos = { path = "../protos" }
types = { path = "../eth2/types" }
client = { path = "client" }
version = { path = "version" }
clap = "2.32.0"
db = { path = "db" }
dirs = "1.0.3"
futures = "0.1.23"
fork_choice = { path = "../eth2/fork_choice" }
slog = "^2.2.3"
slot_clock = { path = "../eth2/utils/slot_clock" }
slog-term = "^2.4.0"
slog-async = "^2.3.0"
types = { path = "../eth2/types" }
ssz = { path = "../eth2/utils/ssz" }
tokio = "0.1"
ctrlc = { version = "3.1.1", features = ["termination"] }
tokio = "0.1.15"
futures = "0.1.25"
exit-future = "0.1.3"
state_processing = { path = "../eth2/state_processing" }

2
beacon_node/beacon_chain/Cargo.toml
View File

@@ -1,7 +1,7 @@
[package]
name = "beacon_chain"
version = "0.1.0"
authors = ["Paul Hauner <paul@paulhauner.com>"]
authors = ["Paul Hauner <paul@paulhauner.com>", "Age Manning <Age@AgeManning.com>"]
edition = "2018"
[dependencies]

53
beacon_node/beacon_chain/src/attestation_aggregator.rs
View File

@@ -1,4 +1,3 @@
use log::trace;
use ssz::TreeHash;
use state_processing::per_block_processing::validate_attestation_without_signature;
use std::collections::{HashMap, HashSet};
@@ -86,34 +85,22 @@ impl AttestationAggregator {
free_attestation: &FreeAttestation,
spec: &ChainSpec,
) -> Result<Outcome, BeaconStateError> {
let attestation_duties = match state.attestation_slot_and_shard_for_validator(
free_attestation.validator_index as usize,
spec,
) {
Err(BeaconStateError::EpochCacheUninitialized(e)) => {
panic!("Attempted to access unbuilt cache {:?}.", e)
}
Err(BeaconStateError::EpochOutOfBounds) => invalid_outcome!(Message::TooOld),
Err(BeaconStateError::ShardOutOfBounds) => invalid_outcome!(Message::BadShard),
Err(e) => return Err(e),
Ok(None) => invalid_outcome!(Message::BadValidatorIndex),
Ok(Some(attestation_duties)) => attestation_duties,
};
let duties =
match state.get_attestation_duties(free_attestation.validator_index as usize, spec) {
Err(BeaconStateError::EpochCacheUninitialized(e)) => {
panic!("Attempted to access unbuilt cache {:?}.", e)
}
Err(BeaconStateError::EpochOutOfBounds) => invalid_outcome!(Message::TooOld),
Err(BeaconStateError::ShardOutOfBounds) => invalid_outcome!(Message::BadShard),
Err(e) => return Err(e),
Ok(None) => invalid_outcome!(Message::BadValidatorIndex),
Ok(Some(attestation_duties)) => attestation_duties,
};
let (slot, shard, committee_index) = attestation_duties;
trace!(
"slot: {}, shard: {}, committee_index: {}, val_index: {}",
slot,
shard,
committee_index,
free_attestation.validator_index
);
if free_attestation.data.slot != slot {
if free_attestation.data.slot != duties.slot {
invalid_outcome!(Message::BadSlot);
}
if free_attestation.data.shard != shard {
if free_attestation.data.shard != duties.shard {
invalid_outcome!(Message::BadShard);
}
@@ -143,7 +130,7 @@ impl AttestationAggregator {
if let Some(updated_attestation) = aggregate_attestation(
existing_attestation,
&free_attestation.signature,
committee_index as usize,
duties.committee_index as usize,
) {
self.store.insert(signable_message, updated_attestation);
valid_outcome!(Message::Aggregated);
@@ -154,7 +141,7 @@ impl AttestationAggregator {
let mut aggregate_signature = AggregateSignature::new();
aggregate_signature.add(&free_attestation.signature);
let mut aggregation_bitfield = Bitfield::new();
aggregation_bitfield.set(committee_index as usize, true);
aggregation_bitfield.set(duties.committee_index as usize, true);
let new_attestation = Attestation {
data: free_attestation.data.clone(),
aggregation_bitfield,
@@ -177,9 +164,13 @@ impl AttestationAggregator {
) -> Vec<Attestation> {
let mut known_attestation_data: HashSet<AttestationData> = HashSet::new();
state.latest_attestations.iter().for_each(|attestation| {
known_attestation_data.insert(attestation.data.clone());
});
state
.previous_epoch_attestations
.iter()
.chain(state.current_epoch_attestations.iter())
.for_each(|attestation| {
known_attestation_data.insert(attestation.data.clone());
});
self.store
.values()

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

@@ -15,10 +15,7 @@ use state_processing::{
per_slot_processing, BlockProcessingError, SlotProcessingError,
};
use std::sync::Arc;
use types::{
readers::{BeaconBlockReader, BeaconStateReader},
*,
};
use types::*;
#[derive(Debug, PartialEq)]
pub enum ValidBlock {
@@ -85,20 +82,18 @@ where
let state_root = genesis_state.canonical_root();
state_store.put(&state_root, &ssz_encode(&genesis_state)[..])?;
let block_root = genesis_block.canonical_root();
let block_root = genesis_block.block_header().canonical_root();
block_store.put(&block_root, &ssz_encode(&genesis_block)[..])?;
let finalized_head = RwLock::new(CheckPoint::new(
genesis_block.clone(),
block_root,
// TODO: this is a memory waste; remove full clone.
genesis_state.clone(),
state_root,
));
let canonical_head = RwLock::new(CheckPoint::new(
genesis_block.clone(),
block_root,
// TODO: this is a memory waste; remove full clone.
genesis_state.clone(),
state_root,
));
@@ -106,7 +101,8 @@ where
genesis_state.build_epoch_cache(RelativeEpoch::Previous, &spec)?;
genesis_state.build_epoch_cache(RelativeEpoch::Current, &spec)?;
genesis_state.build_epoch_cache(RelativeEpoch::Next, &spec)?;
genesis_state.build_epoch_cache(RelativeEpoch::NextWithoutRegistryChange, &spec)?;
genesis_state.build_epoch_cache(RelativeEpoch::NextWithRegistryChange, &spec)?;
Ok(Self {
block_store,
@@ -192,10 +188,13 @@ where
/// processing applied to it.
pub fn advance_state(&self, slot: Slot) -> Result<(), SlotProcessingError> {
let state_slot = self.state.read().slot;
let head_block_root = self.head().beacon_block_root;
let latest_block_header = self.head().beacon_block.block_header();
for _ in state_slot.as_u64()..slot.as_u64() {
per_slot_processing(&mut *self.state.write(), head_block_root, &self.spec)?;
per_slot_processing(&mut *self.state.write(), &latest_block_header, &self.spec)?;
}
Ok(())
}
@@ -248,19 +247,15 @@ where
/// present and prior epoch is available.
pub fn block_proposer(&self, slot: Slot) -> Result<usize, BeaconStateError> {
trace!("BeaconChain::block_proposer: slot: {}", slot);
let index = self
.state
.read()
.get_beacon_proposer_index(slot, &self.spec)?;
let index = self.state.read().get_beacon_proposer_index(
slot,
RelativeEpoch::Current,
&self.spec,
)?;
Ok(index)
}
/// Returns the justified slot for the present state.
pub fn justified_epoch(&self) -> Epoch {
self.state.read().justified_epoch
}
/// Returns the attestation slot and shard for a given validator index.
///
/// Information is read from the current state, so only information from the present and prior
@@ -273,12 +268,12 @@ where
"BeaconChain::validator_attestion_slot_and_shard: validator_index: {}",
validator_index
);
if let Some((slot, shard, _committee)) = self
if let Some(attestation_duty) = self
.state
.read()
.attestation_slot_and_shard_for_validator(validator_index, &self.spec)?
.get_attestation_duties(validator_index, &self.spec)?
{
Ok(Some((slot, shard)))
Ok(Some((attestation_duty.slot, attestation_duty.shard)))
} else {
Ok(None)
}
@@ -287,37 +282,33 @@ where
/// Produce an `AttestationData` that is valid for the present `slot` and given `shard`.
pub fn produce_attestation_data(&self, shard: u64) -> Result<AttestationData, Error> {
trace!("BeaconChain::produce_attestation_data: shard: {}", shard);
let justified_epoch = self.justified_epoch();
let justified_block_root = *self
.state
.read()
.get_block_root(
justified_epoch.start_slot(self.spec.slots_per_epoch),
&self.spec,
)
.ok_or_else(|| Error::BadRecentBlockRoots)?;
let source_epoch = self.state.read().current_justified_epoch;
let source_root = *self.state.read().get_block_root(
source_epoch.start_slot(self.spec.slots_per_epoch),
&self.spec,
)?;
let epoch_boundary_root = *self
.state
.read()
.get_block_root(
self.state.read().current_epoch_start_slot(&self.spec),
&self.spec,
)
.ok_or_else(|| Error::BadRecentBlockRoots)?;
let target_root = *self.state.read().get_block_root(
self.state
.read()
.slot
.epoch(self.spec.slots_per_epoch)
.start_slot(self.spec.slots_per_epoch),
&self.spec,
)?;
Ok(AttestationData {
slot: self.state.read().slot,
shard,
beacon_block_root: self.head().beacon_block_root,
epoch_boundary_root,
target_root,
crosslink_data_root: Hash256::zero(),
latest_crosslink: Crosslink {
previous_crosslink: Crosslink {
epoch: self.state.read().slot.epoch(self.spec.slots_per_epoch),
crosslink_data_root: Hash256::zero(),
},
justified_epoch,
justified_block_root,
source_epoch,
source_root,
})
}
@@ -564,66 +555,13 @@ where
}
}
/// Dumps the entire canonical chain, from the head to genesis to a vector for analysis.
///
/// This could be a very expensive operation and should only be done in testing/analysis
/// activities.
pub fn chain_dump(&self) -> Result<Vec<CheckPoint>, Error> {
let mut dump = vec![];
let mut last_slot = CheckPoint {
beacon_block: self.head().beacon_block.clone(),
beacon_block_root: self.head().beacon_block_root,
beacon_state: self.head().beacon_state.clone(),
beacon_state_root: self.head().beacon_state_root,
};
dump.push(last_slot.clone());
loop {
let beacon_block_root = last_slot.beacon_block.parent_root;
if beacon_block_root == self.spec.zero_hash {
break; // Genesis has been reached.
}
let beacon_block = self
.block_store
.get_deserialized(&beacon_block_root)?
.ok_or_else(|| {
Error::DBInconsistent(format!("Missing block {}", beacon_block_root))
})?;
let beacon_state_root = beacon_block.state_root;
let beacon_state = self
.state_store
.get_deserialized(&beacon_state_root)?
.ok_or_else(|| {
Error::DBInconsistent(format!("Missing state {}", beacon_state_root))
})?;
let slot = CheckPoint {
beacon_block,
beacon_block_root,
beacon_state,
beacon_state_root,
};
dump.push(slot.clone());
last_slot = slot;
}
dump.reverse();
Ok(dump)
}
/// Accept some block and attempt to add it to block DAG.
///
/// Will accept blocks from prior slots, however it will reject any block from a future slot.
pub fn process_block(&self, block: BeaconBlock) -> Result<BlockProcessingOutcome, Error> {
debug!("Processing block with slot {}...", block.slot());
debug!("Processing block with slot {}...", block.slot);
let block_root = block.canonical_root();
let block_root = block.block_header().canonical_root();
let present_slot = self.present_slot();
@@ -635,9 +573,9 @@ where
// Load the blocks parent block from the database, returning invalid if that block is not
// found.
let parent_block_root = block.parent_root;
let parent_block = match self.block_store.get_reader(&parent_block_root)? {
Some(parent_root) => parent_root,
let parent_block_root = block.previous_block_root;
let parent_block = match self.block_store.get_deserialized(&parent_block_root)? {
Some(previous_block_root) => previous_block_root,
None => {
return Ok(BlockProcessingOutcome::InvalidBlock(
InvalidBlock::ParentUnknown,
@@ -647,23 +585,20 @@ where
// Load the parent blocks state from the database, returning an error if it is not found.
// It is an error because if know the parent block we should also know the parent state.
let parent_state_root = parent_block.state_root();
let parent_state_root = parent_block.state_root;
let parent_state = self
.state_store
.get_reader(&parent_state_root)?
.ok_or_else(|| Error::DBInconsistent(format!("Missing state {}", parent_state_root)))?
.into_beacon_state()
.ok_or_else(|| {
Error::DBInconsistent(format!("State SSZ invalid {}", parent_state_root))
})?;
.get_deserialized(&parent_state_root)?
.ok_or_else(|| Error::DBInconsistent(format!("Missing state {}", parent_state_root)))?;
// TODO: check the block proposer signature BEFORE doing a state transition. This will
// significantly lower exposure surface to DoS attacks.
// Transition the parent state to the present slot.
let mut state = parent_state;
let previous_block_header = parent_block.block_header();
for _ in state.slot.as_u64()..present_slot.as_u64() {
if let Err(e) = per_slot_processing(&mut state, parent_block_root, &self.spec) {
if let Err(e) = per_slot_processing(&mut state, &previous_block_header, &self.spec) {
return Ok(BlockProcessingOutcome::InvalidBlock(
InvalidBlock::SlotProcessingError(e),
));
@@ -739,22 +674,22 @@ where
attestations.len()
);
let parent_root = *state
.get_block_root(state.slot.saturating_sub(1_u64), &self.spec)
.ok_or_else(|| BlockProductionError::UnableToGetBlockRootFromState)?;
let previous_block_root = *state
.get_block_root(state.slot - 1, &self.spec)
.map_err(|_| BlockProductionError::UnableToGetBlockRootFromState)?;
let mut block = BeaconBlock {
slot: state.slot,
parent_root,
previous_block_root,
state_root: Hash256::zero(), // Updated after the state is calculated.
randao_reveal,
eth1_data: Eth1Data {
// TODO: replace with real data
deposit_root: Hash256::zero(),
block_hash: Hash256::zero(),
},
signature: self.spec.empty_signature.clone(), // To be completed by a validator.
body: BeaconBlockBody {
randao_reveal,
eth1_data: Eth1Data {
// TODO: replace with real data
deposit_root: Hash256::zero(),
block_hash: Hash256::zero(),
},
proposer_slashings: self.get_proposer_slashings_for_block(),
attester_slashings: self.get_attester_slashings_for_block(),
attestations,
@@ -802,6 +737,59 @@ where
Ok(())
}
/// Dumps the entire canonical chain, from the head to genesis to a vector for analysis.
///
/// This could be a very expensive operation and should only be done in testing/analysis
/// activities.
pub fn chain_dump(&self) -> Result<Vec<CheckPoint>, Error> {
let mut dump = vec![];
let mut last_slot = CheckPoint {
beacon_block: self.head().beacon_block.clone(),
beacon_block_root: self.head().beacon_block_root,
beacon_state: self.head().beacon_state.clone(),
beacon_state_root: self.head().beacon_state_root,
};
dump.push(last_slot.clone());
loop {
let beacon_block_root = last_slot.beacon_block.previous_block_root;
if beacon_block_root == self.spec.zero_hash {
break; // Genesis has been reached.
}
let beacon_block = self
.block_store
.get_deserialized(&beacon_block_root)?
.ok_or_else(|| {
Error::DBInconsistent(format!("Missing block {}", beacon_block_root))
})?;
let beacon_state_root = beacon_block.state_root;
let beacon_state = self
.state_store
.get_deserialized(&beacon_state_root)?
.ok_or_else(|| {
Error::DBInconsistent(format!("Missing state {}", beacon_state_root))
})?;
let slot = CheckPoint {
beacon_block,
beacon_block_root,
beacon_state,
beacon_state_root,
};
dump.push(slot.clone());
last_slot = slot;
}
dump.reverse();
Ok(dump)
}
}
impl From<DBError> for Error {

94
beacon_node/beacon_chain/src/initialise.rs Normal file
View File

@@ -0,0 +1,94 @@
// Initialisation functions to generate a new BeaconChain.
// Note: A new version of ClientTypes may need to be implemented for the lighthouse
// testnet. These are examples. Also. there is code duplication which can/should be cleaned up.
use crate::BeaconChain;
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 types::test_utils::TestingBeaconStateBuilder;
use types::{BeaconBlock, ChainSpec, Hash256};
//TODO: Correct this for prod
//TODO: Account for historical db
pub fn initialise_beacon_chain(
spec: &ChainSpec,
db_name: Option<&PathBuf>,
) -> Arc<BeaconChain<DiskDB, SystemTimeSlotClock, BitwiseLMDGhost<DiskDB>>> {
// set up the db
let db = Arc::new(DiskDB::open(
db_name.expect("Database directory must be included"),
None,
));
let block_store = Arc::new(BeaconBlockStore::new(db.clone()));
let state_store = Arc::new(BeaconStateStore::new(db.clone()));
let state_builder = TestingBeaconStateBuilder::from_deterministic_keypairs(8, &spec);
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());
// Slot clock
let slot_clock = SystemTimeSlotClock::new(genesis_state.genesis_time, spec.seconds_per_slot)
.expect("Unable to load SystemTimeSlotClock");
// Choose the fork choice
let fork_choice = BitwiseLMDGhost::new(block_store.clone(), state_store.clone());
// Genesis chain
//TODO: Handle error correctly
Arc::new(
BeaconChain::from_genesis(
state_store.clone(),
block_store.clone(),
slot_clock,
genesis_state,
genesis_block,
spec.clone(),
fork_choice,
)
.expect("Terminate if beacon chain generation fails"),
)
}
/// Initialisation of a test beacon chain, uses an in memory db with fixed genesis time.
pub fn initialise_test_beacon_chain(
spec: &ChainSpec,
_db_name: Option<&PathBuf>,
) -> Arc<BeaconChain<MemoryDB, SystemTimeSlotClock, BitwiseLMDGhost<MemoryDB>>> {
let db = Arc::new(MemoryDB::open());
let block_store = Arc::new(BeaconBlockStore::new(db.clone()));
let state_store = Arc::new(BeaconStateStore::new(db.clone()));
let state_builder = TestingBeaconStateBuilder::from_deterministic_keypairs(8, spec);
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());
// Slot clock
let slot_clock = SystemTimeSlotClock::new(genesis_state.genesis_time, spec.seconds_per_slot)
.expect("Unable to load SystemTimeSlotClock");
// Choose the fork choice
let fork_choice = BitwiseLMDGhost::new(block_store.clone(), state_store.clone());
// Genesis chain
//TODO: Handle error correctly
Arc::new(
BeaconChain::from_genesis(
state_store.clone(),
block_store.clone(),
slot_clock,
genesis_state,
genesis_block,
spec.clone(),
fork_choice,
)
.expect("Terminate if beacon chain generation fails"),
)
}

7
beacon_node/beacon_chain/src/lib.rs
View File

@@ -2,8 +2,13 @@ mod attestation_aggregator;
mod beacon_chain;
mod checkpoint;
mod errors;
pub mod initialise;
pub use self::beacon_chain::{BeaconChain, BlockProcessingOutcome, InvalidBlock, ValidBlock};
pub use self::checkpoint::CheckPoint;
pub use self::errors::BeaconChainError;
pub use fork_choice::{ForkChoice, ForkChoiceAlgorithm, ForkChoiceError};
pub use db;
pub use fork_choice;
pub use parking_lot;
pub use slot_clock;
pub use types;

1
beacon_node/beacon_chain/test_harness/specs/validator_registry.yaml
View File

@@ -9,6 +9,7 @@ test_cases:
deposits_for_chain_start: 1000
num_slots: 64
skip_slots: [2, 3]
persistent_committee_period: 0
deposits:
# At slot 1, create a new validator deposit of 5 ETH.
- slot: 1

29
beacon_node/beacon_chain/test_harness/src/beacon_chain_harness.rs
View File

@@ -46,8 +46,8 @@ impl BeaconChainHarness {
TestingBeaconStateBuilder::from_default_keypairs_file_if_exists(validator_count, &spec);
let (genesis_state, keypairs) = state_builder.build();
let state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
let genesis_block = BeaconBlock::genesis(state_root, &spec);
let mut genesis_block = BeaconBlock::empty(&spec);
genesis_block.state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
// Create the Beacon Chain
let beacon_chain = Arc::new(
@@ -127,8 +127,8 @@ impl BeaconChainHarness {
.get_crosslink_committees_at_slot(present_slot, &self.spec)
.unwrap()
.iter()
.fold(vec![], |mut acc, (committee, _slot)| {
acc.append(&mut committee.clone());
.fold(vec![], |mut acc, c| {
acc.append(&mut c.committee.clone());
acc
});
let attesting_validators: HashSet<usize> =
@@ -233,6 +233,27 @@ impl BeaconChainHarness {
Some(Signature::new(message, domain, &validator.keypair.sk))
}
/// Returns the current `Fork` of the `beacon_chain`.
pub fn fork(&self) -> Fork {
self.beacon_chain.state.read().fork.clone()
}
/// Returns the current `epoch` of the `beacon_chain`.
pub fn epoch(&self) -> Epoch {
self.beacon_chain
.state
.read()
.slot
.epoch(self.spec.slots_per_epoch)
}
/// Returns the keypair for some validator index.
pub fn validator_keypair(&self, validator_index: usize) -> Option<&Keypair> {
self.validators
.get(validator_index)
.and_then(|v| Some(&v.keypair))
}
/// Submit a deposit to the `BeaconChain` and, if given a keypair, create a new
/// `ValidatorHarness` instance for this validator.
///

71
beacon_node/beacon_chain/test_harness/src/test_case.rs
View File

@@ -3,12 +3,11 @@
use crate::beacon_chain_harness::BeaconChainHarness;
use beacon_chain::CheckPoint;
use bls::get_withdrawal_credentials;
use log::{info, warn};
use ssz::SignedRoot;
use types::*;
use types::test_utils::{TestingAttesterSlashingBuilder, TestingProposerSlashingBuilder};
use types::test_utils::*;
use yaml_rust::Yaml;
mod config;
@@ -63,6 +62,10 @@ impl TestCase {
spec.slots_per_epoch = n;
}
if let Some(n) = self.config.persistent_committee_period {
spec.persistent_committee_period = n;
}
spec
}
@@ -222,27 +225,20 @@ impl TestCase {
}
/// Builds a `Deposit` this is valid for the given `BeaconChainHarness` at its next slot.
fn build_transfer(harness: &BeaconChainHarness, from: u64, to: u64, amount: u64) -> Transfer {
fn build_transfer(
harness: &BeaconChainHarness,
sender: u64,
recipient: u64,
amount: u64,
) -> Transfer {
let slot = harness.beacon_chain.state.read().slot + 1;
let mut transfer = Transfer {
from,
to,
amount,
fee: 0,
slot,
pubkey: harness.validators[from as usize].keypair.pk.clone(),
signature: Signature::empty_signature(),
};
let mut builder = TestingTransferBuilder::new(sender, recipient, amount, slot);
let message = transfer.signed_root();
let epoch = slot.epoch(harness.spec.slots_per_epoch);
let keypair = harness.validator_keypair(sender as usize).unwrap();
builder.sign(keypair.clone(), &harness.fork(), &harness.spec);
transfer.signature = harness
.validator_sign(from as usize, &message[..], epoch, Domain::Transfer)
.expect("Unable to sign Transfer");
transfer
builder.build()
}
/// Builds a `Deposit` this is valid for the given `BeaconChainHarness`.
@@ -255,41 +251,12 @@ fn build_deposit(
index_offset: u64,
) -> (Deposit, Keypair) {
let keypair = Keypair::random();
let withdrawal_credentials = Hash256::from_slice(
&get_withdrawal_credentials(&keypair.pk, harness.spec.bls_withdrawal_prefix_byte)[..],
);
let proof_of_possession = DepositInput::create_proof_of_possession(
&keypair,
&withdrawal_credentials,
harness.spec.get_domain(
harness
.beacon_chain
.state
.read()
.current_epoch(&harness.spec),
Domain::Deposit,
&harness.beacon_chain.state.read().fork,
),
);
let index = harness.beacon_chain.state.read().deposit_index + index_offset;
let deposit = Deposit {
// Note: `branch` and `index` will need to be updated once the spec defines their
// validity.
branch: vec![],
index,
deposit_data: DepositData {
amount,
timestamp: 1,
deposit_input: DepositInput {
pubkey: keypair.pk.clone(),
withdrawal_credentials,
proof_of_possession,
},
},
};
let mut builder = TestingDepositBuilder::new(keypair.pk.clone(), amount);
builder.set_index(harness.beacon_chain.state.read().deposit_index + index_offset);
builder.sign(&keypair, harness.epoch(), &harness.fork(), &harness.spec);
(deposit, keypair)
(builder.build(), keypair)
}
/// Builds a `VoluntaryExit` this is valid for the given `BeaconChainHarness`.

3
beacon_node/beacon_chain/test_harness/src/test_case/config.rs
View File

@@ -20,6 +20,8 @@ pub struct Config {
pub deposits_for_chain_start: usize,
/// Number of slots in an epoch.
pub slots_per_epoch: Option<u64>,
/// Affects the number of epochs a validator must be active before they can withdraw.
pub persistent_committee_period: Option<u64>,
/// Number of slots to build before ending execution.
pub num_slots: u64,
/// Number of slots that should be skipped due to inactive validator.
@@ -45,6 +47,7 @@ impl Config {
deposits_for_chain_start: as_usize(&yaml, "deposits_for_chain_start")
.expect("Must specify validator count"),
slots_per_epoch: as_u64(&yaml, "slots_per_epoch"),
persistent_committee_period: as_u64(&yaml, "persistent_committee_period"),
num_slots: as_u64(&yaml, "num_slots").expect("Must specify `config.num_slots`"),
skip_slots: as_vec_u64(yaml, "skip_slots"),
deposits: parse_deposits(&yaml),

21
beacon_node/client/Cargo.toml Normal file
View File

@@ -0,0 +1,21 @@
[package]
name = "client"
version = "0.1.0"
authors = ["Age Manning <Age@AgeManning.com>"]
edition = "2018"
[dependencies]
beacon_chain = { path = "../beacon_chain" }
network = { path = "../network" }
db = { path = "../db" }
rpc = { path = "../rpc" }
fork_choice = { path = "../../eth2/fork_choice" }
types = { path = "../../eth2/types" }
slot_clock = { path = "../../eth2/utils/slot_clock" }
error-chain = "0.12.0"
slog = "^2.2.3"
tokio = "0.1.15"
clap = "2.32.0"
dirs = "1.0.3"
exit-future = "0.1.3"
futures = "0.1.25"

124
beacon_node/client/src/client_config.rs Normal file
View File

@@ -0,0 +1,124 @@
use clap::ArgMatches;
use db::DBType;
use fork_choice::ForkChoiceAlgorithm;
use network::NetworkConfig;
use slog::error;
use std::fs;
use std::net::SocketAddr;
use std::net::{IpAddr, Ipv4Addr};
use std::path::PathBuf;
use types::multiaddr::Protocol;
use types::multiaddr::ToMultiaddr;
use types::ChainSpec;
/// Stores the client configuration for this Lighthouse instance.
#[derive(Debug, Clone)]
pub struct ClientConfig {
pub data_dir: PathBuf,
pub spec: ChainSpec,
pub net_conf: network::NetworkConfig,
pub fork_choice: ForkChoiceAlgorithm,
pub db_type: DBType,
pub db_name: PathBuf,
pub rpc_conf: rpc::RPCConfig,
//pub ipc_conf:
}
impl Default for ClientConfig {
/// Build a new lighthouse configuration from defaults.
fn default() -> Self {
let data_dir = {
let home = dirs::home_dir().expect("Unable to determine home dir.");
home.join(".lighthouse/")
};
fs::create_dir_all(&data_dir)
.unwrap_or_else(|_| panic!("Unable to create {:?}", &data_dir));
let default_spec = ChainSpec::lighthouse_testnet();
let default_net_conf = NetworkConfig::new(default_spec.boot_nodes.clone());
Self {
data_dir: data_dir.clone(),
// default to foundation for chain specs
spec: default_spec,
net_conf: default_net_conf,
// default to bitwise LMD Ghost
fork_choice: ForkChoiceAlgorithm::BitwiseLMDGhost,
// default to memory db for now
db_type: DBType::Memory,
// default db name for disk-based dbs
db_name: data_dir.join("chain.db"),
rpc_conf: rpc::RPCConfig::default(),
}
}
}
impl ClientConfig {
/// Parses the CLI arguments into a `Config` struct.
pub fn parse_args(args: ArgMatches, log: &slog::Logger) -> Result<Self, &'static str> {
let mut config = ClientConfig::default();
/* Network related arguments */
// Custom p2p listen port
if let Some(port_str) = args.value_of("port") {
if let Ok(port) = port_str.parse::<u16>() {
config.net_conf.listen_port = port;
// update the listening multiaddrs
for address in &mut config.net_conf.listen_addresses {
address.pop();
address.append(Protocol::Tcp(port));
}
} else {
error!(log, "Invalid port"; "port" => port_str);
return Err("Invalid port");
}
}
// Custom listening address ipv4/ipv6
// TODO: Handle list of addresses
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()
.expect("Invalid listen address format");
config.net_conf.listen_addresses = vec![multiaddr];
} else {
error!(log, "Invalid IP Address"; "Address" => listen_address_str);
return Err("Invalid IP Address");
}
}
/* Filesystem related arguments */
// Custom datadir
if let Some(dir) = args.value_of("datadir") {
config.data_dir = PathBuf::from(dir.to_string());
};
/* RPC related arguments */
if args.is_present("rpc") {
config.rpc_conf.enabled = true;
}
if let Some(rpc_address) = args.value_of("rpc-address") {
if let Ok(listen_address) = rpc_address.parse::<Ipv4Addr>() {
config.rpc_conf.listen_address = listen_address;
} else {
error!(log, "Invalid RPC listen address"; "Address" => rpc_address);
return Err("Invalid RPC listen address");
}
}
if let Some(rpc_port) = args.value_of("rpc-port") {
if let Ok(port) = rpc_port.parse::<u16>() {
config.rpc_conf.port = port;
} else {
error!(log, "Invalid RPC port"; "port" => rpc_port);
return Err("Invalid RPC port");
}
}
Ok(config)
}
}

49
beacon_node/client/src/client_types.rs Normal file
View File

@@ -0,0 +1,49 @@
use crate::ClientConfig;
use beacon_chain::{
db::{ClientDB, DiskDB, MemoryDB},
fork_choice::BitwiseLMDGhost,
initialise,
slot_clock::{SlotClock, SystemTimeSlotClock},
BeaconChain,
};
use fork_choice::ForkChoice;
use std::sync::Arc;
pub trait ClientTypes {
type DB: ClientDB + 'static;
type SlotClock: SlotClock + 'static;
type ForkChoice: ForkChoice + 'static;
fn initialise_beacon_chain(
config: &ClientConfig,
) -> Arc<BeaconChain<Self::DB, Self::SlotClock, Self::ForkChoice>>;
}
pub struct StandardClientType;
impl ClientTypes for StandardClientType {
type DB = DiskDB;
type SlotClock = SystemTimeSlotClock;
type ForkChoice = BitwiseLMDGhost<DiskDB>;
fn initialise_beacon_chain(
config: &ClientConfig,
) -> Arc<BeaconChain<Self::DB, Self::SlotClock, Self::ForkChoice>> {
initialise::initialise_beacon_chain(&config.spec, Some(&config.db_name))
}
}
pub struct TestingClientType;
impl ClientTypes for TestingClientType {
type DB = MemoryDB;
type SlotClock = SystemTimeSlotClock;
type ForkChoice = BitwiseLMDGhost<MemoryDB>;
fn initialise_beacon_chain(
config: &ClientConfig,
) -> Arc<BeaconChain<Self::DB, Self::SlotClock, Self::ForkChoice>> {
initialise::initialise_test_beacon_chain(&config.spec, None)
}
}

14
beacon_node/client/src/error.rs Normal file
View File

@@ -0,0 +1,14 @@
// generates error types
use network;
use error_chain::{
error_chain, error_chain_processing, impl_error_chain_kind, impl_error_chain_processed,
impl_extract_backtrace,
};
error_chain! {
links {
Network(network::error::Error, network::error::ErrorKind);
}
}

77
beacon_node/client/src/lib.rs Normal file
View File

@@ -0,0 +1,77 @@
extern crate slog;
mod client_config;
pub mod client_types;
pub mod error;
pub mod notifier;
use beacon_chain::BeaconChain;
pub use client_config::ClientConfig;
pub use client_types::ClientTypes;
use exit_future::Signal;
use network::Service as NetworkService;
use slog::o;
use std::marker::PhantomData;
use std::sync::Arc;
use tokio::runtime::TaskExecutor;
/// 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,
/// The beacon chain for the running client.
beacon_chain: Arc<BeaconChain<T::DB, T::SlotClock, T::ForkChoice>>,
/// Reference to the network service.
pub network: Arc<NetworkService>,
/// Future to stop and begin shutdown of the Client.
//TODO: Decide best way to handle shutdown
pub exit: exit_future::Exit,
/// The sending future to call to terminate the Client.
//TODO: Decide best way to handle shutdown
pub exit_signal: Signal,
/// The clients logger.
log: slog::Logger,
/// Marker to pin the beacon chain generics.
phantom: PhantomData<T>,
}
impl<TClientType: ClientTypes> Client<TClientType> {
/// Generate an instance of the client. Spawn and link all internal sub-processes.
pub fn new(
config: ClientConfig,
log: slog::Logger,
executor: &TaskExecutor,
) -> error::Result<Self> {
let (exit_signal, exit) = exit_future::signal();
// generate a beacon chain
let beacon_chain = TClientType::initialise_beacon_chain(&config);
// Start the network service, libp2p and syncing threads
// TODO: Add beacon_chain reference to network parameters
let network_config = &config.net_conf;
let network_logger = log.new(o!("Service" => "Network"));
let (network, _network_send) = NetworkService::new(
beacon_chain.clone(),
network_config,
executor,
network_logger,
)?;
// spawn the RPC server
if config.rpc_conf.enabled {
rpc::start_server(&config.rpc_conf, &log);
}
Ok(Client {
config,
beacon_chain,
exit,
exit_signal,
log,
network,
phantom: PhantomData,
})
}
}

45
beacon_node/client/src/notifier.rs Normal file
View File

@@ -0,0 +1,45 @@
use crate::Client;
use crate::ClientTypes;
use exit_future::Exit;
use futures::{Future, Stream};
use slog::{debug, info, 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.
pub fn run<T: ClientTypes>(client: &Client<T>, executor: TaskExecutor, exit: Exit) {
// notification heartbeat
let interval = Interval::new(Instant::now(), Duration::from_secs(5));
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 |_| {
info!(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();
}
Ok(())
};
// map error and spawn
let log = client.log.clone();
let heartbeat_interval = interval
.map_err(move |e| debug!(log, "Timer error {}", e))
.for_each(heartbeat);
executor.spawn(exit.until(heartbeat_interval).map(|_| ()));
}

7
beacon_node/db/src/lib.rs
View File

@@ -12,3 +12,10 @@ use self::stores::COLUMNS;
pub use self::disk_db::DiskDB;
pub use self::memory_db::MemoryDB;
pub use self::traits::{ClientDB, DBError, DBValue};
/// Currently available database options
#[derive(Debug, Clone)]
pub enum DBType {
Memory,
RocksDB,
}

37
beacon_node/db/src/stores/beacon_block_store.rs
View File

@@ -2,7 +2,7 @@ use super::BLOCKS_DB_COLUMN as DB_COLUMN;
use super::{ClientDB, DBError};
use ssz::decode;
use std::sync::Arc;
use types::{readers::BeaconBlockReader, BeaconBlock, Hash256, Slot};
use types::{BeaconBlock, Hash256, Slot};
#[derive(Clone, Debug, PartialEq)]
pub enum BeaconBlockAtSlotError {
@@ -38,23 +38,6 @@ impl<T: ClientDB> BeaconBlockStore<T> {
}
}
/// Retuns an object implementing `BeaconBlockReader`, or `None` (if hash not known).
///
/// Note: Presently, this function fully deserializes a `BeaconBlock` and returns that. In the
/// future, it would be ideal to return an object capable of reading directly from serialized
/// SSZ bytes.
pub fn get_reader(&self, hash: &Hash256) -> Result<Option<impl BeaconBlockReader>, DBError> {
match self.get(&hash)? {
None => Ok(None),
Some(ssz) => {
let block = decode::<BeaconBlock>(&ssz).map_err(|_| DBError {
message: "Bad BeaconBlock SSZ.".to_string(),
})?;
Ok(Some(block))
}
}
}
/// Retrieve the block at a slot given a "head_hash" and a slot.
///
/// A "head_hash" must be a block hash with a slot number greater than or equal to the desired
@@ -72,17 +55,17 @@ impl<T: ClientDB> BeaconBlockStore<T> {
&self,
head_hash: &Hash256,
slot: Slot,
) -> Result<Option<(Hash256, impl BeaconBlockReader)>, BeaconBlockAtSlotError> {
) -> Result<Option<(Hash256, BeaconBlock)>, BeaconBlockAtSlotError> {
let mut current_hash = *head_hash;
loop {
if let Some(block_reader) = self.get_reader(&current_hash)? {
if block_reader.slot() == slot {
break Ok(Some((current_hash, block_reader)));
} else if block_reader.slot() < slot {
if let Some(block) = self.get_deserialized(&current_hash)? {
if block.slot == slot {
break Ok(Some((current_hash, block)));
} else if block.slot < slot {
break Ok(None);
} else {
current_hash = block_reader.parent_root();
current_hash = block.previous_block_root;
}
} else {
break Err(BeaconBlockAtSlotError::UnknownBeaconBlock(current_hash));
@@ -228,7 +211,7 @@ mod tests {
for i in 0..block_count {
let mut block = BeaconBlock::random_for_test(&mut rng);
block.parent_root = parent_hashes[i];
block.previous_block_root = parent_hashes[i];
block.slot = slots[i];
let ssz = ssz_encode(&block);
@@ -240,12 +223,12 @@ mod tests {
// Test that certain slots can be reached from certain hashes.
let test_cases = vec![(4, 4), (4, 3), (4, 2), (4, 1), (4, 0)];
for (hashes_index, slot_index) in test_cases {
let (matched_block_hash, reader) = bs
let (matched_block_hash, block) = bs
.block_at_slot(&hashes[hashes_index], slots[slot_index])
.unwrap()
.unwrap();
assert_eq!(matched_block_hash, hashes[slot_index]);
assert_eq!(reader.slot(), slots[slot_index]);
assert_eq!(block.slot, slots[slot_index]);
}
let ssz = bs.block_at_slot(&hashes[4], Slot::new(2)).unwrap();

22
beacon_node/db/src/stores/beacon_state_store.rs
View File

@@ -2,7 +2,7 @@ use super::STATES_DB_COLUMN as DB_COLUMN;
use super::{ClientDB, DBError};
use ssz::decode;
use std::sync::Arc;
use types::{readers::BeaconStateReader, BeaconState, Hash256};
use types::{BeaconState, Hash256};
pub struct BeaconStateStore<T>
where
@@ -30,23 +30,6 @@ impl<T: ClientDB> BeaconStateStore<T> {
}
}
}
/// Retuns an object implementing `BeaconStateReader`, or `None` (if hash not known).
///
/// Note: Presently, this function fully deserializes a `BeaconState` and returns that. In the
/// future, it would be ideal to return an object capable of reading directly from serialized
/// SSZ bytes.
pub fn get_reader(&self, hash: &Hash256) -> Result<Option<impl BeaconStateReader>, DBError> {
match self.get(&hash)? {
None => Ok(None),
Some(ssz) => {
let state = decode::<BeaconState>(&ssz).map_err(|_| DBError {
message: "Bad State SSZ.".to_string(),
})?;
Ok(Some(state))
}
}
}
}
#[cfg(test)]
@@ -72,8 +55,7 @@ mod tests {
store.put(&state_root, &ssz_encode(&state)).unwrap();
let reader = store.get_reader(&state_root).unwrap().unwrap();
let decoded = reader.into_beacon_state().unwrap();
let decoded = store.get_deserialized(&state_root).unwrap().unwrap();
assert_eq!(state, decoded);
}

17
beacon_node/eth2-libp2p/Cargo.toml Normal file
View File

@@ -0,0 +1,17 @@
[package]
name = "eth2-libp2p"
version = "0.1.0"
authors = ["Age Manning <Age@AgeManning.com>"]
edition = "2018"
[dependencies]
# SigP repository until PR is merged
libp2p = { git = "https://github.com/SigP/rust-libp2p", branch = "gossipsub" }
types = { path = "../../eth2/types" }
ssz = { path = "../../eth2/utils/ssz" }
ssz_derive = { path = "../../eth2/utils/ssz_derive" }
slog = "2.4.1"
version = { path = "../version" }
tokio = "0.1.16"
futures = "0.1.25"
error-chain = "0.12.0"

156
beacon_node/eth2-libp2p/src/behaviour.rs Normal file
View File

@@ -0,0 +1,156 @@
use crate::rpc::{RPCEvent, RPCMessage, Rpc};
use crate::NetworkConfig;
use futures::prelude::*;
use libp2p::{
core::{
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess},
PublicKey,
},
gossipsub::{Gossipsub, GossipsubEvent},
identify::{protocol::IdentifyInfo, Identify, IdentifyEvent},
ping::{Ping, PingEvent},
tokio_io::{AsyncRead, AsyncWrite},
NetworkBehaviour, PeerId,
};
use slog::{debug, o};
use types::Topic;
/// Builds the network behaviour for the libp2p Swarm.
/// Implements gossipsub message routing.
#[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.
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>,
#[behaviour(ignore)]
events: Vec<BehaviourEvent>,
/// Logger for behaviour actions.
#[behaviour(ignore)]
log: slog::Logger,
}
// Implement the NetworkBehaviourEventProcess trait so that we can derive NetworkBehaviour for Behaviour
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<GossipsubEvent>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, event: GossipsubEvent) {
match event {
GossipsubEvent::Message(message) => {
let gs_message = String::from_utf8_lossy(&message.data);
// TODO: Remove this type - debug only
self.events
.push(BehaviourEvent::Message(gs_message.to_string()))
}
_ => {}
}
}
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<RPCMessage>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, event: RPCMessage) {
match event {
RPCMessage::PeerDialed(peer_id) => {
self.events.push(BehaviourEvent::PeerDialed(peer_id))
}
RPCMessage::RPC(peer_id, rpc_event) => {
self.events.push(BehaviourEvent::RPC(peer_id, rpc_event))
}
}
}
}
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, info));
}
IdentifyEvent::Error { .. } => {}
IdentifyEvent::SendBack { .. } => {}
}
}
}
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<PingEvent>
for Behaviour<TSubstream>
{
fn inject_event(&mut self, _event: PingEvent) {
// not interested in ping responses at the moment.
}
}
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,
) -> Async<NetworkBehaviourAction<TBehaviourIn, BehaviourEvent>> {
if !self.events.is_empty() {
return Async::Ready(NetworkBehaviourAction::GenerateEvent(self.events.remove(0)));
}
Async::NotReady
}
}
/// Implements the combined behaviour for the libp2p service.
impl<TSubstream: AsyncRead + AsyncWrite> Behaviour<TSubstream> {
/// 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);
}
}
/// The types of events than can be obtained from polling the behaviour.
pub enum BehaviourEvent {
RPC(PeerId, RPCEvent),
PeerDialed(PeerId),
Identified(PeerId, IdentifyInfo),
// TODO: This is a stub at the moment
Message(String),
}

66
beacon_node/eth2-libp2p/src/config.rs Normal file
View File

@@ -0,0 +1,66 @@
use crate::Multiaddr;
use libp2p::gossipsub::{GossipsubConfig, GossipsubConfigBuilder};
#[derive(Clone, Debug)]
/// Network configuration for lighthouse.
pub struct Config {
//TODO: stubbing networking initial params, change in the future
/// IP address to listen on.
pub listen_addresses: Vec<Multiaddr>,
/// Listen port UDP/TCP.
pub listen_port: u16,
/// Gossipsub configuration parameters.
pub gs_config: GossipsubConfig,
/// Configuration parameters for node identification protocol.
pub identify_config: IdentifyConfig,
/// List of nodes to initially connect to.
pub boot_nodes: Vec<Multiaddr>,
/// Client version
pub client_version: String,
/// List of topics to subscribe to as strings
pub topics: Vec<String>,
}
impl Default for Config {
/// Generate a default network configuration.
fn default() -> Self {
Config {
listen_addresses: vec!["/ip4/127.0.0.1/tcp/9000"
.parse()
.expect("is a correct multi-address")],
listen_port: 9000,
gs_config: GossipsubConfigBuilder::new().build(),
identify_config: IdentifyConfig::default(),
boot_nodes: Vec::new(),
client_version: version::version(),
topics: vec![String::from("beacon_chain")],
}
}
}
impl Config {
pub fn new(boot_nodes: Vec<Multiaddr>) -> Self {
let mut conf = Config::default();
conf.boot_nodes = boot_nodes;
conf
}
}
/// 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(),
}
}
}

8
beacon_node/eth2-libp2p/src/error.rs Normal file
View File

@@ -0,0 +1,8 @@
// generates error types
use error_chain::{
error_chain, error_chain_processing, impl_error_chain_kind, impl_error_chain_processed,
impl_extract_backtrace,
};
error_chain! {}

20
beacon_node/eth2-libp2p/src/lib.rs Normal file
View File

@@ -0,0 +1,20 @@
/// This crate contains the main link for lighthouse to rust-libp2p. It therefore re-exports
/// all required libp2p functionality.
///
/// This crate builds and manages the libp2p services required by the beacon node.
pub mod behaviour;
mod config;
pub mod error;
pub mod rpc;
mod service;
pub use config::Config as NetworkConfig;
pub use libp2p::{
gossipsub::{GossipsubConfig, GossipsubConfigBuilder},
PeerId,
};
pub use rpc::{HelloMessage, RPCEvent};
pub use service::Libp2pEvent;
pub use service::Service;
pub use types::multiaddr;
pub use types::Multiaddr;

161
beacon_node/eth2-libp2p/src/rpc/methods.rs Normal file
View File

@@ -0,0 +1,161 @@
/// Available RPC methods types and ids.
use ssz_derive::{Decode, Encode};
use types::{BeaconBlockBody, BeaconBlockHeader, Epoch, Hash256, Slot};
#[derive(Debug)]
/// Available Serenity Libp2p RPC methods
pub enum RPCMethod {
/// Initialise handshake between connecting peers.
Hello,
/// Terminate a connection providing a reason.
Goodbye,
/// Requests a number of beacon block roots.
BeaconBlockRoots,
/// Requests a number of beacon block headers.
BeaconBlockHeaders,
/// Requests a number of beacon block bodies.
BeaconBlockBodies,
/// Requests values for a merkle proof for the current blocks state root.
BeaconChainState, // Note: experimental, not complete.
/// Unknown method received.
Unknown,
}
impl From<u16> for RPCMethod {
fn from(method_id: u16) -> Self {
match method_id {
0 => RPCMethod::Hello,
1 => RPCMethod::Goodbye,
10 => RPCMethod::BeaconBlockRoots,
11 => RPCMethod::BeaconBlockHeaders,
12 => RPCMethod::BeaconBlockBodies,
13 => RPCMethod::BeaconChainState,
_ => RPCMethod::Unknown,
}
}
}
impl Into<u16> for RPCMethod {
fn into(self) -> u16 {
match self {
RPCMethod::Hello => 0,
RPCMethod::Goodbye => 1,
RPCMethod::BeaconBlockRoots => 10,
RPCMethod::BeaconBlockHeaders => 11,
RPCMethod::BeaconBlockBodies => 12,
RPCMethod::BeaconChainState => 13,
_ => 0,
}
}
}
#[derive(Debug, Clone)]
pub enum RPCRequest {
Hello(HelloMessage),
Goodbye(u64),
BeaconBlockRoots(BeaconBlockRootsRequest),
BeaconBlockHeaders(BeaconBlockHeadersRequest),
BeaconBlockBodies(BeaconBlockBodiesRequest),
BeaconChainState(BeaconChainStateRequest),
}
#[derive(Debug, Clone)]
pub enum RPCResponse {
Hello(HelloMessage),
BeaconBlockRoots(BeaconBlockRootsResponse),
BeaconBlockHeaders(BeaconBlockHeadersResponse),
BeaconBlockBodies(BeaconBlockBodiesResponse),
BeaconChainState(BeaconChainStateResponse),
}
/* Request/Response data structures for RPC methods */
/// The HELLO request/response handshake message.
#[derive(Encode, Decode, Clone, Debug)]
pub struct HelloMessage {
/// The network ID of the peer.
pub network_id: u8,
/// The peers last finalized root.
pub latest_finalized_root: Hash256,
/// The peers last finalized epoch.
pub latest_finalized_epoch: Epoch,
/// The peers last block root.
pub best_root: Hash256,
/// The peers last slot.
pub best_slot: Slot,
}
/// Request a number of beacon block roots from a peer.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconBlockRootsRequest {
/// The starting slot of the requested blocks.
start_slot: Slot,
/// The number of blocks from the start slot.
count: u64, // this must be less than 32768. //TODO: Enforce this in the lower layers
}
/// Response containing a number of beacon block roots from a peer.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconBlockRootsResponse {
/// List of requested blocks and associated slots.
roots: Vec<BlockRootSlot>,
}
/// Contains a block root and associated slot.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BlockRootSlot {
/// The block root.
block_root: Hash256,
/// The block slot.
slot: Slot,
}
/// Request a number of beacon block headers from a peer.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconBlockHeadersRequest {
/// The starting header hash of the requested headers.
start_root: Hash256,
/// The starting slot of the requested headers.
start_slot: Slot,
/// The maximum number of headers than can be returned.
max_headers: u64,
/// The maximum number of slots to skip between blocks.
skip_slots: u64,
}
/// Response containing requested block headers.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconBlockHeadersResponse {
/// The list of requested beacon block headers.
headers: Vec<BeaconBlockHeader>,
}
/// Request a number of beacon block bodies from a peer.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconBlockBodiesRequest {
/// The list of beacon block bodies being requested.
block_roots: Hash256,
}
/// Response containing the list of requested beacon block bodies.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconBlockBodiesResponse {
/// The list of beacon block bodies being requested.
block_bodies: Vec<BeaconBlockBody>,
}
/// Request values for tree hashes which yield a blocks `state_root`.
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconChainStateRequest {
/// The tree hashes that a value is requested for.
hashes: Vec<Hash256>,
}
/// Request values for tree hashes which yield a blocks `state_root`.
// Note: TBD
#[derive(Encode, Decode, Clone, Debug)]
pub struct BeaconChainStateResponse {
/// The values corresponding the to the requested tree hashes.
values: bool, //TBD - stubbed with encodeable bool
}

138
beacon_node/eth2-libp2p/src/rpc/mod.rs Normal file
View File

@@ -0,0 +1,138 @@
/// RPC Protocol over libp2p.
///
/// This is purpose built for Ethereum 2.0 serenity and the protocol listens on
/// `/eth/serenity/rpc/1.0.0`
mod methods;
mod protocol;
use futures::prelude::*;
use libp2p::core::protocols_handler::{OneShotHandler, ProtocolsHandler};
use libp2p::core::swarm::{
ConnectedPoint, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
};
use libp2p::{Multiaddr, PeerId};
pub use methods::{HelloMessage, RPCMethod, RPCRequest, RPCResponse};
pub use protocol::{RPCEvent, RPCProtocol};
use slog::o;
use std::marker::PhantomData;
use tokio::io::{AsyncRead, AsyncWrite};
/// The network behaviour handles RPC requests/responses as specified in the Eth 2.0 phase 0
/// specification.
pub struct Rpc<TSubstream> {
/// Queue of events to processed.
events: Vec<NetworkBehaviourAction<RPCEvent, RPCMessage>>,
/// Pins the generic substream.
marker: PhantomData<TSubstream>,
/// Slog logger for RPC behaviour.
log: slog::Logger,
}
impl<TSubstream> Rpc<TSubstream> {
pub fn new(log: &slog::Logger) -> Self {
let log = log.new(o!("Service" => "Libp2p-RPC"));
Rpc {
events: Vec::new(),
marker: PhantomData,
log,
}
}
/// Submits and RPC request.
pub fn send_rpc(&mut self, peer_id: PeerId, rpc_event: RPCEvent) {
self.events.push(NetworkBehaviourAction::SendEvent {
peer_id,
event: rpc_event,
});
}
}
impl<TSubstream> NetworkBehaviour for Rpc<TSubstream>
where
TSubstream: AsyncRead + AsyncWrite,
{
type ProtocolsHandler = OneShotHandler<TSubstream, RPCProtocol, RPCEvent, OneShotEvent>;
type OutEvent = RPCMessage;
fn new_handler(&mut self) -> Self::ProtocolsHandler {
Default::default()
}
fn addresses_of_peer(&mut self, _peer_id: &PeerId) -> Vec<Multiaddr> {
Vec::new()
}
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 {
self.events.push(NetworkBehaviourAction::GenerateEvent(
RPCMessage::PeerDialed(peer_id),
));
}
}
fn inject_disconnected(&mut self, _: &PeerId, _: ConnectedPoint) {}
fn inject_node_event(
&mut self,
source: PeerId,
event: <Self::ProtocolsHandler as ProtocolsHandler>::OutEvent,
) {
// ignore successful send events
let event = match event {
OneShotEvent::Rx(event) => event,
OneShotEvent::Sent => return,
};
// send the event to the user
self.events
.push(NetworkBehaviourAction::GenerateEvent(RPCMessage::RPC(
source, event,
)));
}
fn poll(
&mut self,
_: &mut PollParameters<'_>,
) -> Async<
NetworkBehaviourAction<
<Self::ProtocolsHandler as ProtocolsHandler>::InEvent,
Self::OutEvent,
>,
> {
if !self.events.is_empty() {
return Async::Ready(self.events.remove(0));
}
Async::NotReady
}
}
/// Messages sent to the user from the RPC protocol.
pub enum RPCMessage {
RPC(PeerId, RPCEvent),
PeerDialed(PeerId),
}
/// Transmission between the `OneShotHandler` and the `RPCEvent`.
#[derive(Debug)]
pub enum OneShotEvent {
/// We received an RPC from a remote.
Rx(RPCEvent),
/// We successfully sent an RPC request.
Sent,
}
impl From<RPCEvent> for OneShotEvent {
#[inline]
fn from(rpc: RPCEvent) -> OneShotEvent {
OneShotEvent::Rx(rpc)
}
}
impl From<()> for OneShotEvent {
#[inline]
fn from(_: ()) -> OneShotEvent {
OneShotEvent::Sent
}
}

247
beacon_node/eth2-libp2p/src/rpc/protocol.rs Normal file
View File

@@ -0,0 +1,247 @@
use super::methods::*;
use libp2p::core::{upgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
use ssz::{ssz_encode, Decodable, Encodable, SszStream};
use std::io;
use std::iter;
use tokio::io::{AsyncRead, AsyncWrite};
/// The maximum bytes that can be sent across the RPC.
const MAX_READ_SIZE: usize = 4_194_304; // 4M
/// Implementation of the `ConnectionUpgrade` for the rpc protocol.
#[derive(Debug, Clone)]
pub struct RPCProtocol;
impl UpgradeInfo for RPCProtocol {
type Info = &'static [u8];
type InfoIter = iter::Once<Self::Info>;
#[inline]
fn protocol_info(&self) -> Self::InfoIter {
iter::once(b"/eth/serenity/rpc/1.0.0")
}
}
impl Default for RPCProtocol {
fn default() -> Self {
RPCProtocol
}
}
/// The RPC types which are sent/received in this protocol.
#[derive(Debug, Clone)]
pub enum RPCEvent {
Request {
id: u64,
method_id: u16,
body: RPCRequest,
},
Response {
id: u64,
method_id: u16, //TODO: Remove and process decoding upstream
result: RPCResponse,
},
}
impl UpgradeInfo for RPCEvent {
type Info = &'static [u8];
type InfoIter = iter::Once<Self::Info>;
#[inline]
fn protocol_info(&self) -> Self::InfoIter {
iter::once(b"/eth/serenity/rpc/1.0.0")
}
}
impl<TSocket> InboundUpgrade<TSocket> for RPCProtocol
where
TSocket: AsyncRead + AsyncWrite,
{
type Output = RPCEvent;
type Error = DecodeError;
type Future =
upgrade::ReadOneThen<TSocket, (), fn(Vec<u8>, ()) -> Result<RPCEvent, DecodeError>>;
fn upgrade_inbound(self, socket: TSocket, _: Self::Info) -> Self::Future {
upgrade::read_one_then(socket, MAX_READ_SIZE, (), |packet, ()| Ok(decode(packet)?))
}
}
fn decode(packet: Vec<u8>) -> Result<RPCEvent, DecodeError> {
// decode the header of the rpc
// request/response
let (request, index) = bool::ssz_decode(&packet, 0)?;
let (id, index) = u64::ssz_decode(&packet, index)?;
let (method_id, index) = u16::ssz_decode(&packet, index)?;
if request {
let body = match RPCMethod::from(method_id) {
RPCMethod::Hello => {
let (hello_body, _index) = HelloMessage::ssz_decode(&packet, index)?;
RPCRequest::Hello(hello_body)
}
RPCMethod::Goodbye => {
let (goodbye_code, _index) = u64::ssz_decode(&packet, index)?;
RPCRequest::Goodbye(goodbye_code)
}
RPCMethod::BeaconBlockRoots => {
let (block_roots_request, _index) =
BeaconBlockRootsRequest::ssz_decode(&packet, index)?;
RPCRequest::BeaconBlockRoots(block_roots_request)
}
RPCMethod::BeaconBlockHeaders => {
let (block_headers_request, _index) =
BeaconBlockHeadersRequest::ssz_decode(&packet, index)?;
RPCRequest::BeaconBlockHeaders(block_headers_request)
}
RPCMethod::BeaconBlockBodies => {
let (block_bodies_request, _index) =
BeaconBlockBodiesRequest::ssz_decode(&packet, index)?;
RPCRequest::BeaconBlockBodies(block_bodies_request)
}
RPCMethod::BeaconChainState => {
let (chain_state_request, _index) =
BeaconChainStateRequest::ssz_decode(&packet, index)?;
RPCRequest::BeaconChainState(chain_state_request)
}
RPCMethod::Unknown => return Err(DecodeError::UnknownRPCMethod),
};
Ok(RPCEvent::Request {
id,
method_id,
body,
})
}
// we have received a response
else {
let result = match RPCMethod::from(method_id) {
RPCMethod::Hello => {
let (body, _index) = HelloMessage::ssz_decode(&packet, index)?;
RPCResponse::Hello(body)
}
RPCMethod::Goodbye => unreachable!("Should never receive a goodbye response"),
RPCMethod::BeaconBlockRoots => {
let (body, _index) = BeaconBlockRootsResponse::ssz_decode(&packet, index)?;
RPCResponse::BeaconBlockRoots(body)
}
RPCMethod::BeaconBlockHeaders => {
let (body, _index) = BeaconBlockHeadersResponse::ssz_decode(&packet, index)?;
RPCResponse::BeaconBlockHeaders(body)
}
RPCMethod::BeaconBlockBodies => {
let (body, _index) = BeaconBlockBodiesResponse::ssz_decode(&packet, index)?;
RPCResponse::BeaconBlockBodies(body)
}
RPCMethod::BeaconChainState => {
let (body, _index) = BeaconChainStateResponse::ssz_decode(&packet, index)?;
RPCResponse::BeaconChainState(body)
}
RPCMethod::Unknown => return Err(DecodeError::UnknownRPCMethod),
};
Ok(RPCEvent::Response {
id,
method_id,
result,
})
}
}
impl<TSocket> OutboundUpgrade<TSocket> for RPCEvent
where
TSocket: AsyncWrite,
{
type Output = ();
type Error = io::Error;
type Future = upgrade::WriteOne<TSocket>;
#[inline]
fn upgrade_outbound(self, socket: TSocket, _: Self::Info) -> Self::Future {
let bytes = ssz_encode(&self);
upgrade::write_one(socket, bytes)
}
}
impl Encodable for RPCEvent {
fn ssz_append(&self, s: &mut SszStream) {
match self {
RPCEvent::Request {
id,
method_id,
body,
} => {
s.append(&true);
s.append(id);
s.append(method_id);
match body {
RPCRequest::Hello(body) => {
s.append(body);
}
RPCRequest::Goodbye(body) => {
s.append(body);
}
RPCRequest::BeaconBlockRoots(body) => {
s.append(body);
}
RPCRequest::BeaconBlockHeaders(body) => {
s.append(body);
}
RPCRequest::BeaconBlockBodies(body) => {
s.append(body);
}
RPCRequest::BeaconChainState(body) => {
s.append(body);
}
}
}
RPCEvent::Response {
id,
method_id,
result,
} => {
s.append(&false);
s.append(id);
s.append(method_id);
match result {
RPCResponse::Hello(response) => {
s.append(response);
}
RPCResponse::BeaconBlockRoots(response) => {
s.append(response);
}
RPCResponse::BeaconBlockHeaders(response) => {
s.append(response);
}
RPCResponse::BeaconBlockBodies(response) => {
s.append(response);
}
RPCResponse::BeaconChainState(response) => {
s.append(response);
}
}
}
}
}
}
#[derive(Debug)]
pub enum DecodeError {
ReadError(upgrade::ReadOneError),
SSZDecodeError(ssz::DecodeError),
UnknownRPCMethod,
}
impl From<upgrade::ReadOneError> for DecodeError {
#[inline]
fn from(err: upgrade::ReadOneError) -> Self {
DecodeError::ReadError(err)
}
}
impl From<ssz::DecodeError> for DecodeError {
#[inline]
fn from(err: ssz::DecodeError) -> Self {
DecodeError::SSZDecodeError(err)
}
}

178
beacon_node/eth2-libp2p/src/service.rs Normal file
View File

@@ -0,0 +1,178 @@
use crate::behaviour::{Behaviour, BehaviourEvent};
use crate::error;
use crate::multiaddr::Protocol;
use crate::rpc::RPCEvent;
use crate::NetworkConfig;
use futures::prelude::*;
use futures::Stream;
use libp2p::core::{
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::io::{Error, ErrorKind};
use std::time::Duration;
use types::TopicBuilder;
/// 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>>>,
/// This node's PeerId.
local_peer_id: PeerId,
/// The libp2p logger handle.
pub log: slog::Logger,
}
impl Service {
pub fn new(config: NetworkConfig, log: slog::Logger) -> error::Result<Self> {
debug!(log, "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 = secio::SecioKeyPair::secp256k1_generated().unwrap();
let local_public_key = local_private_key.to_public_key();
let local_peer_id = local_private_key.to_peer_id();
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)
};
// listen on all addresses
for address in &config.listen_addresses {
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 {
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
),
};
}
// subscribe to default gossipsub topics
let mut subscribed_topics = vec![];
for topic in config.topics {
let t = TopicBuilder::new(topic.to_string()).build();
if swarm.subscribe(t) {
trace!(log, "Subscribed to topic: {:?}", topic);
subscribed_topics.push(topic);
} else {
warn!(log, "Could not subscribe to topic: {:?}", topic)
}
}
info!(log, "Subscribed to topics: {:?}", subscribed_topics);
Ok(Service {
local_peer_id,
swarm,
log,
})
}
}
impl Stream for Service {
type Item = Libp2pEvent;
type Error = crate::error::Error;
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 {
// TODO: Stub here for debugging
BehaviourEvent::Message(m) => {
debug!(self.log, "Message received: {}", m);
return Ok(Async::Ready(Some(Libp2pEvent::Message(m))));
}
BehaviourEvent::RPC(peer_id, event) => {
return Ok(Async::Ready(Some(Libp2pEvent::RPC(peer_id, event))));
}
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,
_ => break,
}
}
Ok(Async::NotReady)
}
}
/// 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: secio::SecioKeyPair,
) -> 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();
let transport = libp2p::dns::DnsConfig::new(transport);
#[cfg(feature = "libp2p-websocket")]
let transport = {
let trans_clone = transport.clone();
transport.or_transport(websocket::WsConfig::new(trans_clone))
};
transport
.with_upgrade(secio::SecioConfig::new(local_private_key))
.and_then(move |out, endpoint| {
let peer_id = out.remote_key.into_peer_id();
let peer_id2 = peer_id.clone();
let upgrade = core::upgrade::SelectUpgrade::new(
libp2p::yamux::Config::default(),
libp2p::mplex::MplexConfig::new(),
)
// TODO: use a single `.map` instead of two maps
.map_inbound(move |muxer| (peer_id, muxer))
.map_outbound(move |muxer| (peer_id2, muxer));
core::upgrade::apply(out.stream, upgrade, endpoint)
.map(|(id, muxer)| (id, core::muxing::StreamMuxerBox::new(muxer)))
})
.with_timeout(Duration::from_secs(20))
.map_err(|err| Error::new(ErrorKind::Other, err))
.boxed()
}
/// Events that can be obtained from polling the Libp2p Service.
pub enum Libp2pEvent {
/// An RPC response request has been received on the swarm.
RPC(PeerId, RPCEvent),
/// Initiated the connection to a new peer.
PeerDialed(PeerId),
/// Received information about a peer on the network.
Identified(PeerId, IdentifyInfo),
// TODO: Pub-sub testing only.
Message(String),
}

16
beacon_node/network/Cargo.toml Normal file
View File

@@ -0,0 +1,16 @@
[package]
name = "network"
version = "0.1.0"
authors = ["Age Manning <Age@AgeManning.com>"]
edition = "2018"
[dependencies]
beacon_chain = { path = "../beacon_chain" }
eth2-libp2p = { path = "../eth2-libp2p" }
version = { path = "../version" }
types = { path = "../../eth2/types" }
slog = "2.4.1"
futures = "0.1.25"
error-chain = "0.12.0"
crossbeam-channel = "0.3.8"
tokio = "0.1.16"

43
beacon_node/network/src/beacon_chain.rs Normal file
View File

@@ -0,0 +1,43 @@
use beacon_chain::BeaconChain as RawBeaconChain;
use beacon_chain::{
db::ClientDB,
fork_choice::ForkChoice,
parking_lot::RwLockReadGuard,
slot_clock::SlotClock,
types::{BeaconState, ChainSpec},
CheckPoint,
};
/// The network's API to the beacon chain.
pub trait BeaconChain: Send + Sync {
fn get_spec(&self) -> &ChainSpec;
fn get_state(&self) -> RwLockReadGuard<BeaconState>;
fn head(&self) -> RwLockReadGuard<CheckPoint>;
fn finalized_head(&self) -> RwLockReadGuard<CheckPoint>;
}
impl<T, U, F> BeaconChain for RawBeaconChain<T, U, F>
where
T: ClientDB + Sized,
U: SlotClock,
F: ForkChoice,
{
fn get_spec(&self) -> &ChainSpec {
&self.spec
}
fn get_state(&self) -> RwLockReadGuard<BeaconState> {
self.state.read()
}
fn head(&self) -> RwLockReadGuard<CheckPoint> {
self.head()
}
fn finalized_head(&self) -> RwLockReadGuard<CheckPoint> {
self.finalized_head()
}
}

13
beacon_node/network/src/error.rs Normal file
View File

@@ -0,0 +1,13 @@
// generates error types
use eth2_libp2p;
use error_chain::{
error_chain, error_chain_processing, impl_error_chain_kind, impl_error_chain_processed,
impl_extract_backtrace,
};
error_chain! {
links {
Libp2p(eth2_libp2p::error::Error, eth2_libp2p::error::ErrorKind);
}
}

9
beacon_node/network/src/lib.rs Normal file
View File

@@ -0,0 +1,9 @@
/// This crate provides the network server for Lighthouse.
pub mod beacon_chain;
pub mod error;
mod message_handler;
mod service;
pub mod sync;
pub use eth2_libp2p::NetworkConfig;
pub use service::Service;

225
beacon_node/network/src/message_handler.rs Normal file
View File

@@ -0,0 +1,225 @@
use crate::beacon_chain::BeaconChain;
use crate::error;
use crate::service::{NetworkMessage, OutgoingMessage};
use crate::sync::SimpleSync;
use crossbeam_channel::{unbounded as channel, Sender};
use eth2_libp2p::{
rpc::{RPCMethod, RPCRequest, RPCResponse},
HelloMessage, PeerId, RPCEvent,
};
use futures::future;
use slog::warn;
use slog::{debug, trace};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
/// Timeout for RPC requests.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// Timeout before banning a peer for non-identification.
const HELLO_TIMEOUT: Duration = Duration::from_secs(30);
/// Handles messages received from the network and client and organises syncing.
pub struct MessageHandler {
/// Currently loaded and initialised beacon chain.
chain: Arc<BeaconChain>,
/// The syncing framework.
sync: SimpleSync,
/// The network channel to relay messages to the Network service.
network_send: crossbeam_channel::Sender<NetworkMessage>,
/// A mapping of peers and the RPC id we have sent an RPC request to.
requests: HashMap<(PeerId, u64), Instant>,
/// A counter of request id for each peer.
request_ids: HashMap<PeerId, u64>,
/// The `MessageHandler` logger.
log: slog::Logger,
}
/// Types of messages the handler can receive.
#[derive(Debug, Clone)]
pub enum HandlerMessage {
/// We have initiated a connection to a new peer.
PeerDialed(PeerId),
/// Peer has disconnected,
PeerDisconnected(PeerId),
/// An RPC response/request has been received.
RPC(PeerId, RPCEvent),
/// A block has been imported.
BlockImported(), //TODO: This comes from pub-sub - decide its contents
}
impl MessageHandler {
/// Initializes and runs the MessageHandler.
pub fn spawn(
beacon_chain: Arc<BeaconChain>,
network_send: crossbeam_channel::Sender<NetworkMessage>,
executor: &tokio::runtime::TaskExecutor,
log: slog::Logger,
) -> error::Result<Sender<HandlerMessage>> {
debug!(log, "Service starting");
let (handler_send, handler_recv) = channel();
// Initialise sync and begin processing in thread
// generate the Message handler
let sync = SimpleSync::new(beacon_chain.clone(), &log);
let mut handler = MessageHandler {
// TODO: The handler may not need a chain, perhaps only sync?
chain: beacon_chain.clone(),
sync,
network_send,
requests: HashMap::new(),
request_ids: HashMap::new(),
log: log.clone(),
};
// 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(|_| {
debug!(log, "Network message handler terminated.");
})?);
}
}));
Ok(handler_send)
}
/// Handle all messages incoming from the network service.
fn handle_message(&mut self, message: HandlerMessage) {
match message {
// we have initiated a connection to a peer
HandlerMessage::PeerDialed(peer_id) => {
let id = self.generate_request_id(&peer_id);
self.send_hello(peer_id, id, true);
}
// we have received an RPC message request/response
HandlerMessage::RPC(peer_id, rpc_event) => {
self.handle_rpc_message(peer_id, rpc_event);
}
//TODO: Handle all messages
_ => {}
}
}
/* RPC - Related functionality */
/// Handle RPC messages
fn handle_rpc_message(&mut self, peer_id: PeerId, rpc_message: RPCEvent) {
match rpc_message {
RPCEvent::Request { id, body, .. // TODO: Clean up RPC Message types, have a cleaner type by this point.
} => self.handle_rpc_request(peer_id, id, body),
RPCEvent::Response { id, result, .. } => self.handle_rpc_response(peer_id, id, result),
}
}
/// A new RPC request has been received from the network.
fn handle_rpc_request(&mut self, peer_id: PeerId, id: u64, request: RPCRequest) {
match request {
RPCRequest::Hello(hello_message) => {
self.handle_hello_request(peer_id, id, hello_message)
}
// TODO: Handle all requests
_ => {}
}
}
/// An RPC response has been received from the network.
// we match on id and ignore responses past the timeout.
fn handle_rpc_response(&mut self, peer_id: PeerId, id: u64, response: RPCResponse) {
// if response id is related to a request, ignore (likely RPC timeout)
if self.requests.remove(&(peer_id.clone(), id)).is_none() {
debug!(self.log, "Unrecognized response from peer: {:?}", peer_id);
return;
}
match response {
RPCResponse::Hello(hello_message) => {
debug!(self.log, "Hello response received from peer: {:?}", peer_id);
self.validate_hello(peer_id, hello_message);
}
// TODO: Handle all responses
_ => {}
}
}
/// Handle a HELLO RPC request message.
fn handle_hello_request(&mut self, peer_id: PeerId, id: u64, hello_message: HelloMessage) {
// send back a HELLO message
self.send_hello(peer_id.clone(), id, false);
// validate the peer
self.validate_hello(peer_id, hello_message);
}
/// Validate a HELLO RPC message.
fn validate_hello(&mut self, peer_id: PeerId, message: HelloMessage) {
// validate the peer
if !self.sync.validate_peer(peer_id.clone(), message) {
debug!(
self.log,
"Peer dropped due to mismatching HELLO messages: {:?}", peer_id
);
//TODO: block/ban the peer
}
}
/* General RPC helper functions */
/// Generates a new request id for a peer.
fn generate_request_id(&mut self, peer_id: &PeerId) -> u64 {
// generate a unique id for the peer
let id = {
let borrowed_id = self.request_ids.entry(peer_id.clone()).or_insert_with(|| 0);
let id = borrowed_id.clone();
//increment the counter
*borrowed_id += 1;
id
};
// register RPC request
self.requests.insert((peer_id.clone(), id), Instant::now());
debug!(
self.log,
"Hello request registered with peer: {:?}", peer_id
);
id
}
/// Sends a HELLO RPC request or response to a newly connected peer.
//TODO: The boolean determines if sending request/respond, will be cleaner in the RPC re-write
fn send_hello(&mut self, peer_id: PeerId, id: u64, is_request: bool) {
let rpc_event = if is_request {
RPCEvent::Request {
id,
method_id: RPCMethod::Hello.into(),
body: RPCRequest::Hello(self.sync.generate_hello()),
}
} else {
RPCEvent::Response {
id,
method_id: RPCMethod::Hello.into(),
result: RPCResponse::Hello(self.sync.generate_hello()),
}
};
// send the hello request to the network
trace!(self.log, "Sending HELLO message to peer {:?}", peer_id);
self.send_rpc(peer_id, rpc_event);
}
/// Sends an RPC request/response to the network server.
fn send_rpc(&self, peer_id: PeerId, rpc_event: RPCEvent) {
self.network_send
.send(NetworkMessage::Send(
peer_id,
OutgoingMessage::RPC(rpc_event),
))
.unwrap_or_else(|_| {
warn!(
self.log,
"Could not send RPC message to the network service"
)
});
}
}

186
beacon_node/network/src/service.rs Normal file
View File

@@ -0,0 +1,186 @@
use crate::beacon_chain::BeaconChain;
use crate::error;
use crate::message_handler::{HandlerMessage, MessageHandler};
use crate::NetworkConfig;
use crossbeam_channel::{unbounded as channel, Sender, TryRecvError};
use eth2_libp2p::RPCEvent;
use eth2_libp2p::Service as LibP2PService;
use eth2_libp2p::{Libp2pEvent, PeerId};
use futures::prelude::*;
use futures::sync::oneshot;
use futures::Stream;
use slog::{debug, info, o, trace};
use std::sync::Arc;
use tokio::runtime::TaskExecutor;
/// 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<()>,
network_send: crossbeam_channel::Sender<NetworkMessage>,
//message_handler: MessageHandler,
//message_handler_send: Sender<HandlerMessage>,
}
impl Service {
pub fn new(
beacon_chain: Arc<BeaconChain>,
config: &NetworkConfig,
executor: &TaskExecutor,
log: slog::Logger,
) -> error::Result<(Arc<Self>, Sender<NetworkMessage>)> {
// build the network channel
let (network_send, network_recv) = channel::<NetworkMessage>();
// launch message handler thread
let message_handler_log = log.new(o!("Service" => "MessageHandler"));
let message_handler_send = MessageHandler::spawn(
beacon_chain,
network_send.clone(),
executor,
message_handler_log,
)?;
// launch libp2p service
let libp2p_log = log.new(o!("Service" => "Libp2p"));
let libp2p_service = LibP2PService::new(config.clone(), libp2p_log)?;
// TODO: Spawn thread to handle libp2p messages and pass to message handler thread.
let libp2p_exit = spawn_service(
libp2p_service,
network_recv,
message_handler_send,
executor,
log,
)?;
let network_service = Service {
libp2p_exit,
network_send: network_send.clone(),
};
Ok((Arc::new(network_service), network_send))
}
// TODO: Testing only
pub fn send_message(&self) {
self.network_send
.send(NetworkMessage::Send(
PeerId::random(),
OutgoingMessage::NotifierTest,
))
.unwrap();
}
}
fn spawn_service(
libp2p_service: LibP2PService,
network_recv: crossbeam_channel::Receiver<NetworkMessage>,
message_handler_send: crossbeam_channel::Sender<HandlerMessage>,
executor: &TaskExecutor,
log: slog::Logger,
) -> error::Result<oneshot::Sender<()>> {
let (network_exit, exit_rx) = oneshot::channel();
// spawn on the current executor
executor.spawn(
network_service(
libp2p_service,
network_recv,
message_handler_send,
log.clone(),
)
// allow for manual termination
.select(exit_rx.then(|_| Ok(())))
.then(move |_| {
info!(log.clone(), "Network service shutdown");
Ok(())
}),
);
Ok(network_exit)
}
fn network_service(
mut libp2p_service: LibP2PService,
network_recv: crossbeam_channel::Receiver<NetworkMessage>,
message_handler_send: crossbeam_channel::Sender<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 {
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))
.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))
.map_err(|_| "failed to send rpc to handler")?;
}
Libp2pEvent::Identified(peer_id, info) => {
debug!(
log,
"We have identified peer: {:?} with {:?}", peer_id, info
);
}
Libp2pEvent::Message(m) => debug!(
libp2p_service.log,
"Network Service: Message received: {}", m
),
},
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");
}
};
}
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
return Err(eth2_libp2p::error::Error::from(
"Network channel disconnected",
));
}
}
}
Ok(Async::NotReady)
})
}
/// Types of messages that the network service can receive.
#[derive(Debug, Clone)]
pub enum NetworkMessage {
/// Send a message to libp2p service.
//TODO: Define typing for messages across the wire
Send(PeerId, OutgoingMessage),
}
/// Type of outgoing messages that can be sent through the network service.
#[derive(Debug, Clone)]
pub enum OutgoingMessage {
/// Send an RPC request/response.
RPC(RPCEvent),
//TODO: Remove
NotifierTest,
}

11
beacon_node/network/src/sync/mod.rs Normal file
View File

@@ -0,0 +1,11 @@
/// Syncing for lighthouse.
///
/// Stores the various syncing methods for the beacon chain.
mod simple_sync;
pub use simple_sync::SimpleSync;
/// Currently implemented sync methods.
pub enum SyncMethod {
SimpleSync,
}

112
beacon_node/network/src/sync/simple_sync.rs Normal file
View File

@@ -0,0 +1,112 @@
use crate::beacon_chain::BeaconChain;
use eth2_libp2p::rpc::HelloMessage;
use eth2_libp2p::PeerId;
use slog::{debug, o};
use std::collections::HashMap;
use std::sync::Arc;
use types::{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;
/// Keeps track of syncing information for known connected peers.
pub struct PeerSyncInfo {
latest_finalized_root: Hash256,
latest_finalized_epoch: Epoch,
best_root: Hash256,
best_slot: Slot,
}
/// The current syncing state.
#[derive(PartialEq)]
pub enum SyncState {
Idle,
Downloading,
Stopped,
}
/// Simple Syncing protocol.
//TODO: Decide for HELLO messages whether its better to keep current in RAM or build on the fly
//when asked.
pub struct SimpleSync {
/// A reference to the underlying beacon chain.
chain: Arc<BeaconChain>,
/// A mapping of Peers to their respective PeerSyncInfo.
known_peers: HashMap<PeerId, PeerSyncInfo>,
/// The current state of the syncing protocol.
state: SyncState,
/// The network id, for quick HELLO RPC message lookup.
network_id: u8,
/// The latest epoch of the syncing chain.
latest_finalized_epoch: Epoch,
/// The latest block of the syncing chain.
latest_slot: Slot,
/// Sync logger.
log: slog::Logger,
}
impl SimpleSync {
pub fn new(beacon_chain: Arc<BeaconChain>, log: &slog::Logger) -> Self {
let state = beacon_chain.get_state();
let sync_logger = log.new(o!("Service"=> "Sync"));
SimpleSync {
chain: beacon_chain.clone(),
known_peers: HashMap::new(),
state: SyncState::Idle,
network_id: beacon_chain.get_spec().network_id,
latest_finalized_epoch: state.finalized_epoch,
latest_slot: state.slot - 1, //TODO: Build latest block function into Beacon chain and correct this
log: sync_logger,
}
}
/// Generates our current state in the form of a HELLO RPC message.
pub fn generate_hello(&self) -> HelloMessage {
let state = &self.chain.get_state();
//TODO: Paul to verify the logic of these fields.
HelloMessage {
network_id: self.network_id,
latest_finalized_root: state.finalized_root,
latest_finalized_epoch: state.finalized_epoch,
best_root: Hash256::zero(), //TODO: build correct value as a beacon chain function
best_slot: state.slot - 1,
}
}
pub fn validate_peer(&mut self, peer_id: PeerId, hello_message: HelloMessage) -> bool {
// network id must match
if hello_message.network_id != self.network_id {
return false;
}
// compare latest epoch and finalized root to see if they exist in our chain
if hello_message.latest_finalized_epoch <= self.latest_finalized_epoch {
// ensure their finalized root is in our chain
// TODO: Get the finalized root at hello_message.latest_epoch and ensure they match
//if (hello_message.latest_finalized_root == self.chain.get_state() {
// return false;
// }
}
// the client is valid, add it to our list of known_peers and request sync if required
// update peer list if peer already exists
let peer_info = PeerSyncInfo {
latest_finalized_root: hello_message.latest_finalized_root,
latest_finalized_epoch: hello_message.latest_finalized_epoch,
best_root: hello_message.best_root,
best_slot: hello_message.best_slot,
};
debug!(self.log, "Handshake successful. Peer: {:?}", peer_id);
self.known_peers.insert(peer_id, peer_info);
// set state to sync
if self.state == SyncState::Idle
&& hello_message.best_slot > self.latest_slot + SLOT_IMPORT_TOLERANCE
{
self.state = SyncState::Downloading;
//TODO: Start requesting blocks from known peers. Ideally in batches
}
true
}
}

23
beacon_node/rpc/Cargo.toml Normal file
View File

@@ -0,0 +1,23 @@
[package]
name = "rpc"
version = "0.1.0"
authors = ["Age Manning <Age@AgeManning.com>"]
edition = "2018"
[dependencies]
bls = { path = "../../eth2/utils/bls" }
beacon_chain = { path = "../beacon_chain" }
protos = { path = "../../protos" }
grpcio = { version = "0.4", default-features = false, features = ["protobuf-codec"] }
protobuf = "2.0.2"
clap = "2.32.0"
db = { path = "../db" }
dirs = "1.0.3"
futures = "0.1.23"
slog = "^2.2.3"
slot_clock = { path = "../../eth2/utils/slot_clock" }
slog-term = "^2.4.0"
slog-async = "^2.3.0"
types = { path = "../../eth2/types" }
ssz = { path = "../../eth2/utils/ssz" }

0
beacon_node/src/rpc/beacon_block.rs → beacon_node/rpc/src/beacon_block.rs
View File

22
beacon_node/rpc/src/config.rs Normal file
View File

@@ -0,0 +1,22 @@
use std::net::Ipv4Addr;
/// RPC Configuration
#[derive(Debug, Clone)]
pub struct Config {
/// Enable the RPC server.
pub enabled: bool,
/// The IPv4 address the RPC will listen on.
pub listen_address: Ipv4Addr,
/// The port the RPC will listen on.
pub port: u16,
}
impl Default for Config {
fn default() -> Self {
Config {
enabled: false, // rpc disabled by default
listen_address: Ipv4Addr::new(127, 0, 0, 1),
port: 5051,
}
}
}

12
beacon_node/src/rpc/mod.rs → beacon_node/rpc/src/lib.rs
View File

@@ -1,16 +1,18 @@
mod beacon_block;
pub mod config;
mod validator;
use self::beacon_block::BeaconBlockServiceInstance;
use self::validator::ValidatorServiceInstance;
pub use config::Config as RPCConfig;
use grpcio::{Environment, Server, ServerBuilder};
use protos::services_grpc::{create_beacon_block_service, create_validator_service};
use std::sync::Arc;
use slog::{info, Logger};
use slog::{info, o};
pub fn start_server(log: Logger) -> Server {
let log_clone = log.clone();
pub fn start_server(config: &RPCConfig, log: &slog::Logger) -> Server {
let log = log.new(o!("Service"=>"RPC"));
let env = Arc::new(Environment::new(1));
let beacon_block_service = {
@@ -25,12 +27,12 @@ pub fn start_server(log: Logger) -> Server {
let mut server = ServerBuilder::new(env)
.register_service(beacon_block_service)
.register_service(validator_service)
.bind("127.0.0.1", 50_051)
.bind(config.listen_address.to_string(), config.port)
.build()
.unwrap();
server.start();
for &(ref host, port) in server.bind_addrs() {
info!(log_clone, "gRPC listening on {}:{}", host, port);
info!(log, "gRPC listening on {}:{}", host, port);
}
server
}

0
beacon_node/src/rpc/validator.rs → beacon_node/rpc/src/validator.rs
View File

30
beacon_node/src/config/mod.rs
View File

@@ -1,30 +0,0 @@
use std::fs;
use std::path::PathBuf;
/// Stores the core configuration for this Lighthouse instance.
/// This struct is general, other components may implement more
/// specialized config structs.
#[derive(Clone)]
pub struct LighthouseConfig {
pub data_dir: PathBuf,
pub p2p_listen_port: u16,
}
const DEFAULT_LIGHTHOUSE_DIR: &str = ".lighthouse";
impl LighthouseConfig {
/// Build a new lighthouse configuration from defaults.
pub fn default() -> Self {
let data_dir = {
let home = dirs::home_dir().expect("Unable to determine home dir.");
home.join(DEFAULT_LIGHTHOUSE_DIR)
};
fs::create_dir_all(&data_dir)
.unwrap_or_else(|_| panic!("Unable to create {:?}", &data_dir));
let p2p_listen_port = 0;
Self {
data_dir,
p2p_listen_port,
}
}
}

164
beacon_node/src/main.rs
View File

@@ -1,37 +1,20 @@
extern crate slog;
mod config;
mod rpc;
mod run;
use std::path::PathBuf;
use crate::config::LighthouseConfig;
use crate::rpc::start_server;
use beacon_chain::BeaconChain;
use clap::{App, Arg};
use db::{
stores::{BeaconBlockStore, BeaconStateStore},
MemoryDB,
};
use fork_choice::BitwiseLMDGhost;
use slog::{error, info, o, Drain};
use slot_clock::SystemTimeSlotClock;
use ssz::TreeHash;
use std::sync::Arc;
use types::{
beacon_state::BeaconStateBuilder, BeaconBlock, ChainSpec, Deposit, DepositData, DepositInput,
Domain, Eth1Data, Fork, Hash256, Keypair,
};
use client::ClientConfig;
use slog::{error, o, Drain};
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 log = slog::Logger::root(drain, o!());
let logger = slog::Logger::root(drain, o!());
let matches = App::new("Lighthouse")
.version("0.0.1")
.author("Sigma Prime <paul@sigmaprime.io>")
.version(version::version().as_str())
.author("Sigma Prime <contact@sigmaprime.io>")
.about("Eth 2.0 Client")
.arg(
Arg::with_name("datadir")
@@ -40,6 +23,13 @@ fn main() {
.help("Data directory for keys and databases.")
.takes_value(true),
)
.arg(
Arg::with_name("listen_address")
.long("listen-address")
.value_name("Listen Address")
.help("The Network address to listen for p2p connections.")
.takes_value(true),
)
.arg(
Arg::with_name("port")
.long("port")
@@ -47,110 +37,34 @@ fn main() {
.help("Network listen port for p2p connections.")
.takes_value(true),
)
.arg(
Arg::with_name("rpc")
.long("rpc")
.value_name("RPC")
.help("Enable the RPC server.")
.takes_value(false),
)
.arg(
Arg::with_name("rpc-address")
.long("rpc-address")
.value_name("RPCADDRESS")
.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),
)
.get_matches();
let mut config = LighthouseConfig::default();
// invalid arguments, panic
let config = ClientConfig::parse_args(matches, &logger).unwrap();
// Custom datadir
if let Some(dir) = matches.value_of("datadir") {
config.data_dir = PathBuf::from(dir.to_string());
}
// Custom p2p listen port
if let Some(port_str) = matches.value_of("port") {
if let Ok(port) = port_str.parse::<u16>() {
config.p2p_listen_port = port;
} else {
error!(log, "Invalid port"; "port" => port_str);
return;
}
}
// Log configuration
info!(log, "";
"data_dir" => &config.data_dir.to_str(),
"port" => &config.p2p_listen_port);
// Specification (presently fixed to foundation).
let spec = ChainSpec::foundation();
// Database (presently in-memory)
let db = Arc::new(MemoryDB::open());
let block_store = Arc::new(BeaconBlockStore::new(db.clone()));
let state_store = Arc::new(BeaconStateStore::new(db.clone()));
// Slot clock
let genesis_time = 1_549_935_547; // 12th Feb 2018 (arbitrary value in the past).
let slot_clock = SystemTimeSlotClock::new(genesis_time, spec.seconds_per_slot)
.expect("Unable to load SystemTimeSlotClock");
// Choose the fork choice
let fork_choice = BitwiseLMDGhost::new(block_store.clone(), state_store.clone());
/*
* Generate some random data to start a chain with.
*
* This is will need to be replace for production usage.
*/
let latest_eth1_data = Eth1Data {
deposit_root: Hash256::zero(),
block_hash: Hash256::zero(),
};
let keypairs: Vec<Keypair> = (0..10)
.collect::<Vec<usize>>()
.iter()
.map(|_| Keypair::random())
.collect();
let initial_validator_deposits: Vec<Deposit> = keypairs
.iter()
.map(|keypair| Deposit {
branch: vec![], // branch verification is not specified.
index: 0, // index verification is not specified.
deposit_data: DepositData {
amount: 32_000_000_000, // 32 ETH (in Gwei)
timestamp: genesis_time - 1,
deposit_input: DepositInput {
pubkey: keypair.pk.clone(),
withdrawal_credentials: Hash256::zero(), // Withdrawal not possible.
proof_of_possession: DepositInput::create_proof_of_possession(
&keypair,
&Hash256::zero(),
spec.get_domain(
// Get domain from genesis fork_version
spec.genesis_epoch,
Domain::Deposit,
&Fork {
previous_version: spec.genesis_fork_version,
current_version: spec.genesis_fork_version,
epoch: spec.genesis_epoch,
},
),
),
},
},
})
.collect();
let mut state_builder = BeaconStateBuilder::new(genesis_time, latest_eth1_data, &spec);
state_builder.process_initial_deposits(&initial_validator_deposits, &spec);
let genesis_state = state_builder.build(&spec).unwrap();
let state_root = Hash256::from_slice(&genesis_state.hash_tree_root());
let genesis_block = BeaconBlock::genesis(state_root, &spec);
// Genesis chain
let _chain_result = BeaconChain::from_genesis(
state_store.clone(),
block_store.clone(),
slot_clock,
genesis_state,
genesis_block,
spec,
fork_choice,
);
let _server = start_server(log.clone());
loop {
std::thread::sleep(std::time::Duration::from_secs(1));
match run::run_beacon_node(config, &logger) {
Ok(_) => {}
Err(e) => error!(logger, "Beacon node failed because {:?}", e),
}
}

51
beacon_node/src/run.rs Normal file
View File

@@ -0,0 +1,51 @@
use client::client_types::TestingClientType;
use client::error;
use client::{notifier, Client, ClientConfig};
use futures::sync::oneshot;
use futures::Future;
use slog::info;
use std::cell::RefCell;
use tokio::runtime::Builder;
pub fn run_beacon_node(config: ClientConfig, log: &slog::Logger) -> error::Result<()> {
let mut runtime = Builder::new()
.name_prefix("main-")
.build()
.map_err(|e| format!("{:?}", e))?;
// Log configuration
info!(log, "Listening on {:?}", &config.net_conf.listen_addresses;
"data_dir" => &config.data_dir.to_str(),
"port" => &config.net_conf.listen_port);
// run service until ctrl-c
let (ctrlc_send, ctrlc) = oneshot::channel();
let ctrlc_send_c = RefCell::new(Some(ctrlc_send));
ctrlc::set_handler(move || {
if let Some(ctrlc_send) = ctrlc_send_c.try_borrow_mut().unwrap().take() {
ctrlc_send.send(()).expect("Error sending ctrl-c message");
}
})
.map_err(|e| format!("Could not set ctrlc hander: {:?}", e))?;
let (exit_signal, exit) = exit_future::signal();
let executor = runtime.executor();
// currently testing - using TestingClientType
let client: Client<TestingClientType> = Client::new(config, log.clone(), &executor)?;
notifier::run(&client, executor, exit);
runtime
.block_on(ctrlc)
.map_err(|e| format!("Ctrlc oneshot failed: {:?}", e))?;
// perform global shutdown operations.
info!(log, "Shutting down..");
exit_signal.fire();
// shutdown the client
// client.exit_signal.fire();
drop(client);
runtime.shutdown_on_idle().wait().unwrap();
Ok(())
}

8
beacon_node/version/Cargo.toml Normal file
View File

@@ -0,0 +1,8 @@
[package]
name = "version"
version = "0.1.0"
authors = ["Age Manning <Age@AgeManning.com>"]
edition = "2018"
[dependencies]
target_info = "0.1.0"

25
beacon_node/version/src/lib.rs Normal file
View File

@@ -0,0 +1,25 @@
//TODO: Build the version and hash of the built lighthouse binary
/// Version information for the Lighthouse beacon node.
// currently only supports unstable release
extern crate target_info;
use target_info::Target;
const TRACK: &str = "unstable";
/// Provides the current platform
pub fn platform() -> String {
format!("{}-{}", Target::arch(), Target::os())
}
/// Version of the beacon node.
// TODO: Find the sha3 hash, date and rust version used to build the beacon_node binary
pub fn version() -> String {
format!(
"Lighthouse/v{}-{}/{}",
env!("CARGO_PKG_VERSION"),
TRACK,
platform()
)
}