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

Merge branch 'blop-pool' into validator-enhancements

Browse Source
This commit is contained in:
Age Manning
2019-03-30 16:16:30 +11:00
parent 6e254551af 397e104f9b
commit bc305cacc2
24 changed files with 1423 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
beacon_node/beacon_chain/Cargo.toml
View File

@@ -15,6 +15,7 @@ hashing = { path = "../../eth2/utils/hashing" }
fork_choice = { path = "../../eth2/fork_choice" }
parking_lot = "0.7"
log = "0.4"
operation_pool = { path = "../../eth2/operation_pool" }
env_logger = "0.6"
serde = "1.0"
serde_derive = "1.0"

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

@@ -1,4 +1,3 @@
use crate::attestation_aggregator::{AttestationAggregator, Outcome as AggregationOutcome};
use crate::checkpoint::CheckPoint;
use crate::errors::{BeaconChainError as Error, BlockProductionError};
use db::{
@@ -7,9 +6,15 @@ use db::{
};
use fork_choice::{ForkChoice, ForkChoiceError};
use log::{debug, trace};
use operation_pool::DepositInsertStatus;
use operation_pool::OperationPool;
use parking_lot::{RwLock, RwLockReadGuard};
use slot_clock::SlotClock;
use ssz::ssz_encode;
pub use state_processing::per_block_processing::errors::{
AttestationValidationError, AttesterSlashingValidationError, DepositValidationError,
ExitValidationError, ProposerSlashingValidationError, TransferValidationError,
};
use state_processing::{
per_block_processing, per_block_processing_without_verifying_block_signature,
per_slot_processing, BlockProcessingError, SlotProcessingError,
@@ -82,12 +87,7 @@ pub struct BeaconChain<T: ClientDB + Sized, U: SlotClock, F: ForkChoice> {
pub block_store: Arc<BeaconBlockStore<T>>,
pub state_store: Arc<BeaconStateStore<T>>,
pub slot_clock: U,
pub attestation_aggregator: RwLock<AttestationAggregator>,
pub deposits_for_inclusion: RwLock<Vec<Deposit>>,
pub exits_for_inclusion: RwLock<Vec<VoluntaryExit>>,
pub transfers_for_inclusion: RwLock<Vec<Transfer>>,
pub proposer_slashings_for_inclusion: RwLock<Vec<ProposerSlashing>>,
pub attester_slashings_for_inclusion: RwLock<Vec<AttesterSlashing>>,
pub op_pool: OperationPool,
canonical_head: RwLock<CheckPoint>,
finalized_head: RwLock<CheckPoint>,
pub state: RwLock<BeaconState>,
@@ -129,7 +129,6 @@ where
genesis_state.clone(),
state_root,
));
let attestation_aggregator = RwLock::new(AttestationAggregator::new());
genesis_state.build_epoch_cache(RelativeEpoch::Previous, &spec)?;
genesis_state.build_epoch_cache(RelativeEpoch::Current, &spec)?;
@@ -140,12 +139,7 @@ where
block_store,
state_store,
slot_clock,
attestation_aggregator,
deposits_for_inclusion: RwLock::new(vec![]),
exits_for_inclusion: RwLock::new(vec![]),
transfers_for_inclusion: RwLock::new(vec![]),
proposer_slashings_for_inclusion: RwLock::new(vec![]),
attester_slashings_for_inclusion: RwLock::new(vec![]),
op_pool: OperationPool::new(),
state: RwLock::new(genesis_state),
finalized_head,
canonical_head,
@@ -511,252 +505,55 @@ where
})
}
/// Validate a `FreeAttestation` and either:
/// Accept a new attestation from the network.
///
/// - Create a new `Attestation`.
/// - Aggregate it to an existing `Attestation`.
pub fn process_free_attestation(
/// If valid, the attestation is added to the `op_pool` and aggregated with another attestation
/// if possible.
pub fn process_attestation(
&self,
free_attestation: FreeAttestation,
) -> Result<AggregationOutcome, Error> {
let aggregation_outcome = self
.attestation_aggregator
.write()
.process_free_attestation(&self.state.read(), &free_attestation, &self.spec)?;
// return if the attestation is invalid
if !aggregation_outcome.valid {
return Ok(aggregation_outcome);
}
// valid attestation, proceed with fork-choice logic
self.fork_choice.write().add_attestation(
free_attestation.validator_index,
&free_attestation.data.beacon_block_root,
&self.spec,
)?;
Ok(aggregation_outcome)
attestation: Attestation,
) -> Result<(), AttestationValidationError> {
self.op_pool
.insert_attestation(attestation, &*self.state.read(), &self.spec)
}
/// Accept some deposit and queue it for inclusion in an appropriate block.
pub fn receive_deposit_for_inclusion(&self, deposit: Deposit) {
// TODO: deposits are not checked for validity; check them.
//
// https://github.com/sigp/lighthouse/issues/276
self.deposits_for_inclusion.write().push(deposit);
}
/// Return a vec of deposits suitable for inclusion in some block.
pub fn get_deposits_for_block(&self) -> Vec<Deposit> {
// TODO: deposits are indiscriminately included; check them for validity.
//
// https://github.com/sigp/lighthouse/issues/275
self.deposits_for_inclusion.read().clone()
}
/// Takes a list of `Deposits` that were included in recent blocks and removes them from the
/// inclusion queue.
///
/// This ensures that `Deposits` are not included twice in successive blocks.
pub fn set_deposits_as_included(&self, included_deposits: &[Deposit]) {
// TODO: method does not take forks into account; consider this.
//
// https://github.com/sigp/lighthouse/issues/275
let mut indices_to_delete = vec![];
for included in included_deposits {
for (i, for_inclusion) in self.deposits_for_inclusion.read().iter().enumerate() {
if included == for_inclusion {
indices_to_delete.push(i);
}
}
}
let deposits_for_inclusion = &mut self.deposits_for_inclusion.write();
for i in indices_to_delete {
deposits_for_inclusion.remove(i);
}
pub fn process_deposit(
&self,
deposit: Deposit,
) -> Result<DepositInsertStatus, DepositValidationError> {
self.op_pool
.insert_deposit(deposit, &*self.state.read(), &self.spec)
}
/// Accept some exit and queue it for inclusion in an appropriate block.
pub fn receive_exit_for_inclusion(&self, exit: VoluntaryExit) {
// TODO: exits are not checked for validity; check them.
//
// https://github.com/sigp/lighthouse/issues/276
self.exits_for_inclusion.write().push(exit);
}
/// Return a vec of exits suitable for inclusion in some block.
pub fn get_exits_for_block(&self) -> Vec<VoluntaryExit> {
// TODO: exits are indiscriminately included; check them for validity.
//
// https://github.com/sigp/lighthouse/issues/275
self.exits_for_inclusion.read().clone()
}
/// Takes a list of `Deposits` that were included in recent blocks and removes them from the
/// inclusion queue.
///
/// This ensures that `Deposits` are not included twice in successive blocks.
pub fn set_exits_as_included(&self, included_exits: &[VoluntaryExit]) {
// TODO: method does not take forks into account; consider this.
let mut indices_to_delete = vec![];
for included in included_exits {
for (i, for_inclusion) in self.exits_for_inclusion.read().iter().enumerate() {
if included == for_inclusion {
indices_to_delete.push(i);
}
}
}
let exits_for_inclusion = &mut self.exits_for_inclusion.write();
for i in indices_to_delete {
exits_for_inclusion.remove(i);
}
pub fn process_voluntary_exit(&self, exit: VoluntaryExit) -> Result<(), ExitValidationError> {
self.op_pool
.insert_voluntary_exit(exit, &*self.state.read(), &self.spec)
}
/// Accept some transfer and queue it for inclusion in an appropriate block.
pub fn receive_transfer_for_inclusion(&self, transfer: Transfer) {
// TODO: transfers are not checked for validity; check them.
//
// https://github.com/sigp/lighthouse/issues/276
self.transfers_for_inclusion.write().push(transfer);
}
/// Return a vec of transfers suitable for inclusion in some block.
pub fn get_transfers_for_block(&self) -> Vec<Transfer> {
// TODO: transfers are indiscriminately included; check them for validity.
//
// https://github.com/sigp/lighthouse/issues/275
self.transfers_for_inclusion.read().clone()
}
/// Takes a list of `Deposits` that were included in recent blocks and removes them from the
/// inclusion queue.
///
/// This ensures that `Deposits` are not included twice in successive blocks.
pub fn set_transfers_as_included(&self, included_transfers: &[Transfer]) {
// TODO: method does not take forks into account; consider this.
let mut indices_to_delete = vec![];
for included in included_transfers {
for (i, for_inclusion) in self.transfers_for_inclusion.read().iter().enumerate() {
if included == for_inclusion {
indices_to_delete.push(i);
}
}
}
let transfers_for_inclusion = &mut self.transfers_for_inclusion.write();
for i in indices_to_delete {
transfers_for_inclusion.remove(i);
}
pub fn process_transfer(&self, transfer: Transfer) -> Result<(), TransferValidationError> {
self.op_pool
.insert_transfer(transfer, &*self.state.read(), &self.spec)
}
/// Accept some proposer slashing and queue it for inclusion in an appropriate block.
pub fn receive_proposer_slashing_for_inclusion(&self, proposer_slashing: ProposerSlashing) {
// TODO: proposer_slashings are not checked for validity; check them.
//
// https://github.com/sigp/lighthouse/issues/276
self.proposer_slashings_for_inclusion
.write()
.push(proposer_slashing);
}
/// Return a vec of proposer slashings suitable for inclusion in some block.
pub fn get_proposer_slashings_for_block(&self) -> Vec<ProposerSlashing> {
// TODO: proposer_slashings are indiscriminately included; check them for validity.
//
// https://github.com/sigp/lighthouse/issues/275
self.proposer_slashings_for_inclusion.read().clone()
}
/// Takes a list of `ProposerSlashings` that were included in recent blocks and removes them
/// from the inclusion queue.
///
/// This ensures that `ProposerSlashings` are not included twice in successive blocks.
pub fn set_proposer_slashings_as_included(
pub fn process_proposer_slashing(
&self,
included_proposer_slashings: &[ProposerSlashing],
) {
// TODO: method does not take forks into account; consider this.
//
// https://github.com/sigp/lighthouse/issues/275
let mut indices_to_delete = vec![];
for included in included_proposer_slashings {
for (i, for_inclusion) in self
.proposer_slashings_for_inclusion
.read()
.iter()
.enumerate()
{
if included == for_inclusion {
indices_to_delete.push(i);
}
}
}
let proposer_slashings_for_inclusion = &mut self.proposer_slashings_for_inclusion.write();
for i in indices_to_delete {
proposer_slashings_for_inclusion.remove(i);
}
proposer_slashing: ProposerSlashing,
) -> Result<(), ProposerSlashingValidationError> {
self.op_pool
.insert_proposer_slashing(proposer_slashing, &*self.state.read(), &self.spec)
}
/// Accept some attester slashing and queue it for inclusion in an appropriate block.
pub fn receive_attester_slashing_for_inclusion(&self, attester_slashing: AttesterSlashing) {
// TODO: attester_slashings are not checked for validity; check them.
//
// https://github.com/sigp/lighthouse/issues/276
self.attester_slashings_for_inclusion
.write()
.push(attester_slashing);
}
/// Return a vec of attester slashings suitable for inclusion in some block.
pub fn get_attester_slashings_for_block(&self) -> Vec<AttesterSlashing> {
// TODO: attester_slashings are indiscriminately included; check them for validity.
//
// https://github.com/sigp/lighthouse/issues/275
self.attester_slashings_for_inclusion.read().clone()
}
/// Takes a list of `AttesterSlashings` that were included in recent blocks and removes them
/// from the inclusion queue.
///
/// This ensures that `AttesterSlashings` are not included twice in successive blocks.
pub fn set_attester_slashings_as_included(
pub fn process_attester_slashing(
&self,
included_attester_slashings: &[AttesterSlashing],
) {
// TODO: method does not take forks into account; consider this.
//
// https://github.com/sigp/lighthouse/issues/275
let mut indices_to_delete = vec![];
for included in included_attester_slashings {
for (i, for_inclusion) in self
.attester_slashings_for_inclusion
.read()
.iter()
.enumerate()
{
if included == for_inclusion {
indices_to_delete.push(i);
}
}
}
let attester_slashings_for_inclusion = &mut self.attester_slashings_for_inclusion.write();
for i in indices_to_delete {
attester_slashings_for_inclusion.remove(i);
}
}
/// Returns `true` if the given block root has not been processed.
pub fn is_new_block_root(&self, beacon_block_root: &Hash256) -> Result<bool, Error> {
Ok(!self.block_store.exists(beacon_block_root)?)
attester_slashing: AttesterSlashing,
) -> Result<(), AttesterSlashingValidationError> {
self.op_pool
.insert_attester_slashing(attester_slashing, &*self.state.read(), &self.spec)
}
/// Accept some block and attempt to add it to block DAG.
@@ -832,13 +629,6 @@ where
self.block_store.put(&block_root, &ssz_encode(&block)[..])?;
self.state_store.put(&state_root, &ssz_encode(&state)[..])?;
// Update the inclusion queues so they aren't re-submitted.
self.set_deposits_as_included(&block.body.deposits[..]);
self.set_transfers_as_included(&block.body.transfers[..]);
self.set_exits_as_included(&block.body.voluntary_exits[..]);
self.set_proposer_slashings_as_included(&block.body.proposer_slashings[..]);
self.set_attester_slashings_as_included(&block.body.attester_slashings[..]);
// run the fork_choice add_block logic
self.fork_choice
.write()
@@ -874,20 +664,13 @@ where
trace!("Finding attestations for new block...");
let attestations = self
.attestation_aggregator
.read()
.get_attestations_for_state(&state, &self.spec);
trace!(
"Inserting {} attestation(s) into new block.",
attestations.len()
);
let previous_block_root = *state
.get_block_root(state.slot - 1, &self.spec)
.map_err(|_| BlockProductionError::UnableToGetBlockRootFromState)?;
let (proposer_slashings, attester_slashings) =
self.op_pool.get_slashings(&*self.state.read(), &self.spec);
let mut block = BeaconBlock {
slot: state.slot,
previous_block_root,
@@ -900,12 +683,16 @@ where
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,
deposits: self.get_deposits_for_block(),
voluntary_exits: self.get_exits_for_block(),
transfers: self.get_transfers_for_block(),
proposer_slashings,
attester_slashings,
attestations: self
.op_pool
.get_attestations(&*self.state.read(), &self.spec),
deposits: self.op_pool.get_deposits(&*self.state.read(), &self.spec),
voluntary_exits: self
.op_pool
.get_voluntary_exits(&*self.state.read(), &self.spec),
transfers: self.op_pool.get_transfers(&*self.state.read(), &self.spec),
},
};
@@ -951,6 +738,11 @@ where
Ok(())
}
/// Returns `true` if the given block root has not been processed.
pub fn is_new_block_root(&self, beacon_block_root: &Hash256) -> Result<bool, Error> {
Ok(!self.block_store.exists(beacon_block_root)?)
}
/// 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

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

@@ -47,6 +47,8 @@ test_cases:
states:
- slot: 63
num_validators: 1003
num_previous_epoch_attestations: 0
num_current_epoch_attestations: 10
slashed_validators: [11, 12, 13, 14, 42]
exited_validators: []
exit_initiated_validators: [50]

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

@@ -10,8 +10,6 @@ use log::debug;
use rayon::prelude::*;
use slot_clock::TestingSlotClock;
use ssz::TreeHash;
use std::collections::HashSet;
use std::iter::FromIterator;
use std::sync::Arc;
use types::{test_utils::TestingBeaconStateBuilder, *};
@@ -137,51 +135,64 @@ impl BeaconChainHarness {
slot
}
/// Gather the `FreeAttestation`s from the valiators.
///
/// Note: validators will only produce attestations _once per slot_. So, if you call this twice
/// you'll only get attestations on the first run.
pub fn gather_free_attesations(&mut self) -> Vec<FreeAttestation> {
pub fn gather_attesations(&mut self) -> Vec<Attestation> {
let present_slot = self.beacon_chain.present_slot();
let state = self.beacon_chain.state.read();
let attesting_validators = self
.beacon_chain
.state
.read()
let mut attestations = vec![];
for committee in state
.get_crosslink_committees_at_slot(present_slot, &self.spec)
.unwrap()
.iter()
.fold(vec![], |mut acc, c| {
acc.append(&mut c.committee.clone());
acc
});
let attesting_validators: HashSet<usize> =
HashSet::from_iter(attesting_validators.iter().cloned());
{
for &validator in &committee.committee {
let duties = state
.get_attestation_duties(validator, &self.spec)
.unwrap()
.expect("Attesting validators by definition have duties");
let free_attestations: Vec<FreeAttestation> = self
.validators
.par_iter_mut()
.enumerate()
.filter_map(|(i, validator)| {
if attesting_validators.contains(&i) {
// Advance the validator slot.
validator.set_slot(present_slot);
// Obtain `AttestationData` from the beacon chain.
let data = self
.beacon_chain
.produce_attestation_data(duties.shard)
.unwrap();
// Prompt the validator to produce an attestation (if required).
validator.produce_free_attestation().ok()
} else {
None
}
})
.collect();
// Produce an aggregate signature with a single signature.
let aggregate_signature = {
let message = AttestationDataAndCustodyBit {
data: data.clone(),
custody_bit: false,
}
.hash_tree_root();
let domain = self.spec.get_domain(
state.slot.epoch(self.spec.slots_per_epoch),
Domain::Attestation,
&state.fork,
);
let sig =
Signature::new(&message, domain, &self.validators[validator].keypair.sk);
debug!(
"Gathered {} FreeAttestations for slot {}.",
free_attestations.len(),
present_slot
);
let mut agg_sig = AggregateSignature::new();
agg_sig.add(&sig);
free_attestations
agg_sig
};
let mut aggregation_bitfield = Bitfield::with_capacity(committee.committee.len());
let custody_bitfield = Bitfield::with_capacity(committee.committee.len());
aggregation_bitfield.set(duties.committee_index, true);
attestations.push(Attestation {
aggregation_bitfield,
data,
custody_bitfield,
aggregate_signature,
})
}
}
attestations
}
/// Get the block from the proposer for the slot.
@@ -200,7 +211,9 @@ impl BeaconChainHarness {
// Ensure the validators slot clock is accurate.
self.validators[proposer].set_slot(present_slot);
self.validators[proposer].produce_block().unwrap()
let block = self.validators[proposer].produce_block().unwrap();
block
}
/// Advances the chain with a BeaconBlock and attestations from all validators.
@@ -219,20 +232,23 @@ impl BeaconChainHarness {
};
debug!("...block processed by BeaconChain.");
debug!("Producing free attestations...");
debug!("Producing attestations...");
// Produce new attestations.
let free_attestations = self.gather_free_attesations();
let attestations = self.gather_attesations();
debug!("Processing free attestations...");
debug!("Processing {} attestations...", attestations.len());
free_attestations.par_iter().for_each(|free_attestation| {
self.beacon_chain
.process_free_attestation(free_attestation.clone())
.unwrap();
});
attestations
.par_iter()
.enumerate()
.for_each(|(i, attestation)| {
self.beacon_chain
.process_attestation(attestation.clone())
.expect(&format!("Attestation {} invalid: {:?}", i, attestation));
});
debug!("Free attestations processed.");
debug!("Attestations processed.");
block
}
@@ -285,7 +301,7 @@ impl BeaconChainHarness {
/// If a new `ValidatorHarness` was created, the validator should become fully operational as
/// if the validator were created during `BeaconChainHarness` instantiation.
pub fn add_deposit(&mut self, deposit: Deposit, keypair: Option<Keypair>) {
self.beacon_chain.receive_deposit_for_inclusion(deposit);
self.beacon_chain.process_deposit(deposit).unwrap();
// If a keypair is present, add a new `ValidatorHarness` to the rig.
if let Some(keypair) = keypair {
@@ -301,24 +317,26 @@ impl BeaconChainHarness {
/// will stop receiving duties from the beacon chain and just do nothing when prompted to
/// produce/attest.
pub fn add_exit(&mut self, exit: VoluntaryExit) {
self.beacon_chain.receive_exit_for_inclusion(exit);
self.beacon_chain.process_voluntary_exit(exit).unwrap();
}
/// Submit an transfer to the `BeaconChain` for inclusion in some block.
pub fn add_transfer(&mut self, transfer: Transfer) {
self.beacon_chain.receive_transfer_for_inclusion(transfer);
self.beacon_chain.process_transfer(transfer).unwrap();
}
/// Submit a proposer slashing to the `BeaconChain` for inclusion in some block.
pub fn add_proposer_slashing(&mut self, proposer_slashing: ProposerSlashing) {
self.beacon_chain
.receive_proposer_slashing_for_inclusion(proposer_slashing);
.process_proposer_slashing(proposer_slashing)
.unwrap();
}
/// Submit an attester slashing to the `BeaconChain` for inclusion in some block.
pub fn add_attester_slashing(&mut self, attester_slashing: AttesterSlashing) {
self.beacon_chain
.receive_attester_slashing_for_inclusion(attester_slashing);
.process_attester_slashing(attester_slashing)
.unwrap();
}
/// Executes the fork choice rule on the `BeaconChain`, selecting a new canonical head.

27
beacon_node/beacon_chain/test_harness/src/test_case/state_check.rs
View File

@@ -16,6 +16,10 @@ pub struct StateCheck {
pub slot: Slot,
/// Checked against `beacon_state.validator_registry.len()`.
pub num_validators: Option<usize>,
/// The number of pending attestations from the previous epoch that should be in the state.
pub num_previous_epoch_attestations: Option<usize>,
/// The number of pending attestations from the current epoch that should be in the state.
pub num_current_epoch_attestations: Option<usize>,
/// A list of validator indices which have been penalized. Must be in ascending order.
pub slashed_validators: Option<Vec<u64>>,
/// A list of validator indices which have been fully exited. Must be in ascending order.
@@ -34,6 +38,8 @@ impl StateCheck {
Self {
slot: Slot::from(as_u64(&yaml, "slot").expect("State must specify slot")),
num_validators: as_usize(&yaml, "num_validators"),
num_previous_epoch_attestations: as_usize(&yaml, "num_previous_epoch_attestations"),
num_current_epoch_attestations: as_usize(&yaml, "num_current_epoch_attestations"),
slashed_validators: as_vec_u64(&yaml, "slashed_validators"),
exited_validators: as_vec_u64(&yaml, "exited_validators"),
exit_initiated_validators: as_vec_u64(&yaml, "exit_initiated_validators"),
@@ -58,6 +64,7 @@ impl StateCheck {
"State slot is invalid."
);
// Check the validator count
if let Some(num_validators) = self.num_validators {
assert_eq!(
state.validator_registry.len(),
@@ -67,6 +74,26 @@ impl StateCheck {
info!("OK: num_validators = {}.", num_validators);
}
// Check the previous epoch attestations
if let Some(n) = self.num_previous_epoch_attestations {
assert_eq!(
state.previous_epoch_attestations.len(),
n,
"previous epoch attestations count != expected."
);
info!("OK: num_previous_epoch_attestations = {}.", n);
}
// Check the current epoch attestations
if let Some(n) = self.num_current_epoch_attestations {
assert_eq!(
state.current_epoch_attestations.len(),
n,
"current epoch attestations count != expected."
);
info!("OK: num_current_epoch_attestations = {}.", n);
}
// Check for slashed validators.
if let Some(ref slashed_validators) = self.slashed_validators {
let actually_slashed_validators: Vec<u64> = state