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

Add proto_array fork choice (#804)

Browse Source 
* Start implementing proto_array

* Add progress

* Add unfinished progress

* Add further progress

* Add progress

* Add tree filtering

* Add half-finished modifications

* Add refactored version

* Tidy, add incomplete LmdGhost impl

* Move impls in LmdGhost trait def

* Remove old reduced_tree fork choice

* Combine two functions in to `compute_deltas`

* Start testing

* Add more compute_deltas tests

* Add fork choice testing

* Add more fork choice testing

* Add more fork choice tests

* Add more testing to proto-array

* Remove old tests

* Modify tests

* Add more tests

* Add more testing

* Add comments and fixes

* Re-organise crate

* Tidy, finish pruning tests

* Add ssz encoding, other pub fns

* Rename lmd_ghost > proto_array_fork_choice

* Integrate proto_array into lighthouse

* Add first pass at fixing filter

* Clean out old comments

* Add more comments

* Attempt to fix prune error

* Adjust TODO

* Fix test compile errors

* Add extra justification change check

* Update cargo.lock

* Fix fork choice test compile errors

* Most remove ffg_update_required

* Fix bug with epoch of attestation votes

* Start adding new test format

* Make fork choice tests declarative

* Create test def concept

* Move test defs into crate

* Add binary, re-org crate

* Shuffle files

* Start adding ffg tests

* Add more fork choice tests

* Add fork choice JSON dumping

* Add more detail to best node error

* Ensure fin+just checkpoints from from same block

* Rename JustificationManager

* Move checkpoint manager into own file

* Tidy

* Add targetted logging for sneaky sync bug

* Fix justified balances bug

* Add cache metrics

* Add metrics for log levels

* Fix bug in checkpoint manager

* Fix compile error in fork choice tests

* Ignore duplicate blocks in fork choice

* Add block to fock choice before db

* Rename on_new_block fn

* Fix spec inconsistency in `CheckpointManager`

* Remove BlockRootTree

* Remove old reduced_tree code fragment

* Add API endpoint for fork choice

* Add more ffg tests

* Remove block_root_tree reminents

* Ensure effective balances are used

* Remove old debugging code, fix API fault

* Add check to ensure parent block is in fork choice

* Update readme dates

* Fix readme

* Tidy checkpoint manager

* Remove fork choice yaml files from repo

* Remove fork choice yaml from repo

* General tidy

* Address majority of Michael's comments

* Tidy bin/lib business

* Remove dangling file

* Undo changes for rpc/handler from master

* Revert "Undo changes for rpc/handler from master"

This reverts commit 876edff0e4a501aafbb47113454852826dcc24e8.

Co-authored-by: Age Manning <Age@AgeManning.com>
This commit is contained in:
Paul Hauner
2020-01-29 15:05:00 +11:00
committed by GitHub
co-authored by Age Manning
parent cd401147ea
commit b771bbb60c
48 changed files with 3469 additions and 2332 deletions
Download Patch File Download Diff File Expand all files Collapse all files
beacon_node/beacon_chain/Cargo.toml
+1 -1
View File
@@ -33,7 +33,6 @@ eth2_ssz_derive = "0.1.0"
state_processing = { path = "../../eth2/state_processing" }
tree_hash = "0.1.0"
types = { path = "../../eth2/types" }
lmd_ghost = { path = "../../eth2/lmd_ghost" }
eth1 = { path = "../eth1" }
websocket_server = { path = "../websocket_server" }
futures = "0.1.25"
@@ -41,6 +40,7 @@ exit-future = "0.1.3"
genesis = { path = "../genesis" }
integer-sqrt = "0.1"
rand = "0.7.2"
proto_array_fork_choice = { path = "../../eth2/proto_array_fork_choice" }
[dev-dependencies]
tempfile = "3.1.0"
beacon_node/beacon_chain/src/beacon_chain.rs
+46 -45
View File
@@ -8,7 +8,6 @@ use crate::head_tracker::HeadTracker;
use crate::metrics;
use crate::persisted_beacon_chain::{PersistedBeaconChain, BEACON_CHAIN_DB_KEY};
use crate::timeout_rw_lock::TimeoutRwLock;
use lmd_ghost::LmdGhost;
use operation_pool::{OperationPool, PersistedOperationPool};
use slog::{debug, error, info, trace, warn, Logger};
use slot_clock::SlotClock;
@@ -32,7 +31,7 @@ use std::time::{Duration, Instant};
use store::iter::{
BlockRootsIterator, ReverseBlockRootIterator, ReverseStateRootIterator, StateRootsIterator,
};
use store::{BlockRootTree, Error as DBError, Migrate, Store};
use store::{Error as DBError, Migrate, Store};
use tree_hash::TreeHash;
use types::*;
@@ -59,8 +58,11 @@ const HEAD_LOCK_TIMEOUT: Duration = Duration::from_secs(1);
pub enum BlockProcessingOutcome {
/// Block was valid and imported into the block graph.
Processed { block_root: Hash256 },
/// The blocks parent_root is unknown.
ParentUnknown { parent: Hash256 },
/// The parent block was unknown.
ParentUnknown {
parent: Hash256,
reference_location: &'static str,
},
/// The block slot is greater than the present slot.
FutureSlot {
present_slot: Slot,
@@ -116,7 +118,6 @@ pub trait BeaconChainTypes: Send + Sync + 'static {
type Store: store::Store<Self::EthSpec>;
type StoreMigrator: store::Migrate<Self::Store, Self::EthSpec>;
type SlotClock: slot_clock::SlotClock;
type LmdGhost: LmdGhost<Self::Store, Self::EthSpec>;
type Eth1Chain: Eth1ChainBackend<Self::EthSpec, Self::Store>;
type EthSpec: types::EthSpec;
type EventHandler: EventHandler<Self::EthSpec>;
@@ -150,8 +151,6 @@ pub struct BeaconChain<T: BeaconChainTypes> {
pub(crate) head_tracker: HeadTracker,
/// Provides a small cache of `BeaconState` and `BeaconBlock`.
pub(crate) checkpoint_cache: CheckPointCache<T::EthSpec>,
/// Cache of block roots for all known forks post-finalization.
pub block_root_tree: Arc<BlockRootTree>,
/// Logging to CLI, etc.
pub(crate) log: Logger,
}
@@ -192,7 +191,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
ssz_head_tracker: self.head_tracker.to_ssz_container(),
fork_choice: self.fork_choice.as_ssz_container(),
eth1_cache: self.eth1_chain.as_ref().map(|x| x.as_ssz_container()),
block_root_tree: self.block_root_tree.as_ssz_container(),
};
let key = Hash256::from_slice(&BEACON_CHAIN_DB_KEY.as_bytes());
@@ -1063,14 +1061,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
{
// Provide the attestation to fork choice, updating the validator latest messages but
// _without_ finding and updating the head.
if let Err(e) = self
.fork_choice
.process_attestation(&state, &attestation, block)
{
if let Err(e) = self.fork_choice.process_attestation(&state, &attestation) {
error!(
self.log,
"Add attestation to fork choice failed";
"fork_choice_integrity" => format!("{:?}", self.fork_choice.verify_integrity()),
"beacon_block_root" => format!("{}", attestation.data.beacon_block_root),
"error" => format!("{:?}", e)
);
@@ -1232,6 +1226,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
});
}
// Reject any block if its parent is not known to fork choice.
//
// A block that is not in fork choice is either:
//
// - Not yet imported: we should reject this block because we should only import a child
// after its parent has been fully imported.
// - Pre-finalized: if the parent block is _prior_ to finalization, we should ignore it
// because it will revert finalization. Note that the finalized block is stored in fork
// choice, so we will not reject any child of the finalized block (this is relevant during
// genesis).
if !self.fork_choice.contains_block(&block.parent_root) {
return Ok(BlockProcessingOutcome::ParentUnknown {
parent: block.parent_root,
reference_location: "fork_choice",
});
}
let block_root_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_BLOCK_ROOT);
let block_root = block.canonical_root();
@@ -1252,8 +1263,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
}
// Check if the block is already known. We know it is post-finalization, so it is
// sufficient to check the block root tree.
if self.block_root_tree.is_known_block_root(&block_root) {
// sufficient to check the fork choice.
if self.fork_choice.contains_block(&block_root) {
return Ok(BlockProcessingOutcome::BlockIsAlreadyKnown);
}
@@ -1269,6 +1280,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
None => {
return Ok(BlockProcessingOutcome::ParentUnknown {
parent: block.parent_root,
reference_location: "database",
});
}
};
@@ -1363,6 +1375,24 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
});
}
let fork_choice_register_timer =
metrics::start_timer(&metrics::BLOCK_PROCESSING_FORK_CHOICE_REGISTER);
// Register the new block with the fork choice service.
if let Err(e) = self
.fork_choice
.process_block(self, &state, &block, block_root)
{
error!(
self.log,
"Add block to fork choice failed";
"block_root" => format!("{}", block_root),
"error" => format!("{:?}", e),
)
}
metrics::stop_timer(fork_choice_register_timer);
let db_write_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_DB_WRITE);
// Store all the states between the parent block state and this blocks slot before storing
@@ -1392,30 +1422,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
metrics::stop_timer(db_write_timer);
self.block_root_tree
.add_block_root(block_root, block.parent_root, block.slot)?;
self.head_tracker.register_block(block_root, &block);
let fork_choice_register_timer =
metrics::start_timer(&metrics::BLOCK_PROCESSING_FORK_CHOICE_REGISTER);
// Register the new block with the fork choice service.
if let Err(e) = self
.fork_choice
.process_block(self, &state, &block, block_root)
{
error!(
self.log,
"Add block to fork choice failed";
"fork_choice_integrity" => format!("{:?}", self.fork_choice.verify_integrity()),
"block_root" => format!("{}", block_root),
"error" => format!("{:?}", e),
)
}
metrics::stop_timer(fork_choice_register_timer);
metrics::inc_counter(&metrics::BLOCK_PROCESSING_SUCCESSES);
metrics::observe(
&metrics::OPERATIONS_PER_BLOCK_ATTESTATION,
@@ -1706,8 +1714,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
new_epoch: new_finalized_epoch,
})
} else {
self.fork_choice
.process_finalization(&finalized_block, finalized_block_root)?;
self.fork_choice.prune()?;
let finalized_state = self
.get_state_caching_only_with_committee_caches(
@@ -1726,12 +1733,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
max_finality_distance,
);
// Prune in-memory block root tree.
self.block_root_tree.prune_to(
finalized_block_root,
self.heads().into_iter().map(|(block_root, _)| block_root),
);
let _ = self.event_handler.register(EventKind::BeaconFinalization {
epoch: new_finalized_epoch,
root: finalized_block_root,
beacon_node/beacon_chain/src/builder.rs
+30 -113
View File
@@ -9,54 +9,35 @@ use crate::{
ForkChoice,
};
use eth1::Config as Eth1Config;
use lmd_ghost::{LmdGhost, ThreadSafeReducedTree};
use operation_pool::OperationPool;
use proto_array_fork_choice::ProtoArrayForkChoice;
use slog::{info, Logger};
use slot_clock::{SlotClock, TestingSlotClock};
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Duration;
use store::{BlockRootTree, Store};
use store::Store;
use types::{BeaconBlock, BeaconState, ChainSpec, EthSpec, Hash256, Slot};
/// An empty struct used to "witness" all the `BeaconChainTypes` traits. It has no user-facing
/// functionality and only exists to satisfy the type system.
pub struct Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>(
pub struct Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>(
PhantomData<(
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
)>,
);
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
BeaconChainTypes
for Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler> BeaconChainTypes
for Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -64,7 +45,6 @@ where
type Store = TStore;
type StoreMigrator = TStoreMigrator;
type SlotClock = TSlotClock;
type LmdGhost = TLmdGhost;
type Eth1Chain = TEth1Backend;
type EthSpec = TEthSpec;
type EventHandler = TEventHandler;
@@ -92,28 +72,18 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
slot_clock: Option<T::SlotClock>,
persisted_beacon_chain: Option<PersistedBeaconChain<T>>,
head_tracker: Option<HeadTracker>,
block_root_tree: Option<Arc<BlockRootTree>>,
spec: ChainSpec,
log: Option<Logger>,
}
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
BeaconChainBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>
where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -135,7 +105,6 @@ where
slot_clock: None,
persisted_beacon_chain: None,
head_tracker: None,
block_root_tree: None,
spec: TEthSpec::default_spec(),
log: None,
}
@@ -194,15 +163,7 @@ where
let key = Hash256::from_slice(&BEACON_CHAIN_DB_KEY.as_bytes());
let p: PersistedBeaconChain<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
> = match store.get(&key) {
Err(e) => {
return Err(format!(
@@ -230,7 +191,6 @@ where
Some(cache) => Some(Eth1Chain::from_ssz_container(cache, config, store, log)?),
None => None,
};
self.block_root_tree = Some(Arc::new(p.block_root_tree.clone().into()));
self.persisted_beacon_chain = Some(p);
Ok(self)
@@ -273,11 +233,6 @@ where
)
})?;
self.block_root_tree = Some(Arc::new(BlockRootTree::new(
beacon_block_root,
beacon_block.slot,
)));
self.finalized_checkpoint = Some(CheckPoint {
beacon_block_root,
beacon_block,
@@ -327,15 +282,7 @@ where
self,
) -> Result<
BeaconChain<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>,
String,
> {
@@ -387,9 +334,6 @@ where
.event_handler
.ok_or_else(|| "Cannot build without an event handler".to_string())?,
head_tracker: self.head_tracker.unwrap_or_default(),
block_root_tree: self
.block_root_tree
.ok_or_else(|| "Cannot build without a block root tree".to_string())?,
checkpoint_cache: CheckPointCache::default(),
log: log.clone(),
};
@@ -412,15 +356,7 @@ where
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
BeaconChainBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
ThreadSafeReducedTree<TStore, TEthSpec>,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>
where
TStore: Store<TEthSpec> + 'static,
@@ -435,23 +371,9 @@ where
/// If this builder is being "resumed" from disk, then rebuild the last fork choice stored to
/// the database. Otherwise, create a new, empty fork choice.
pub fn reduced_tree_fork_choice(mut self) -> Result<Self, String> {
let store = self
.store
.clone()
.ok_or_else(|| "reduced_tree_fork_choice requires a store")?;
let block_root_tree = self
.block_root_tree
.clone()
.ok_or_else(|| "reduced_tree_fork_choice requires a block root tree")?;
let fork_choice = if let Some(persisted_beacon_chain) = &self.persisted_beacon_chain {
ForkChoice::from_ssz_container(
persisted_beacon_chain.fork_choice.clone(),
store,
block_root_tree,
)
.map_err(|e| format!("Unable to decode fork choice from db: {:?}", e))?
ForkChoice::from_ssz_container(persisted_beacon_chain.fork_choice.clone())
.map_err(|e| format!("Unable to decode fork choice from db: {:?}", e))?
} else {
let finalized_checkpoint = &self
.finalized_checkpoint
@@ -461,14 +383,22 @@ where
.genesis_block_root
.ok_or_else(|| "fork_choice_backend requires a genesis_block_root")?;
let backend = ThreadSafeReducedTree::new(
store,
block_root_tree,
&finalized_checkpoint.beacon_block,
let backend = ProtoArrayForkChoice::new(
finalized_checkpoint.beacon_block.slot,
// Note: here we set the `justified_epoch` to be the same as the epoch of the
// finalized checkpoint. Whilst this finalized checkpoint may actually point to
// a _later_ justified checkpoint, that checkpoint won't yet exist in the fork
// choice.
finalized_checkpoint.beacon_state.current_epoch(),
finalized_checkpoint.beacon_state.current_epoch(),
finalized_checkpoint.beacon_block_root,
);
)?;
ForkChoice::new(backend, genesis_block_root, self.spec.genesis_slot)
ForkChoice::new(
backend,
genesis_block_root,
&finalized_checkpoint.beacon_state,
)
};
self.fork_choice = Some(fork_choice);
@@ -477,13 +407,12 @@ where
}
}
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEthSpec, TEventHandler>
impl<TStore, TStoreMigrator, TSlotClock, TEthSpec, TEventHandler>
BeaconChainBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
CachingEth1Backend<TEthSpec, TStore>,
TEthSpec,
TEventHandler,
@@ -493,7 +422,6 @@ where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
{
@@ -529,22 +457,13 @@ where
}
}
impl<TStore, TStoreMigrator, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TStore, TStoreMigrator, TEth1Backend, TEthSpec, TEventHandler>
BeaconChainBuilder<
Witness<
TStore,
TStoreMigrator,
TestingSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TestingSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>
where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -570,13 +489,12 @@ where
}
}
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec>
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec>
BeaconChainBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
NullEventHandler<TEthSpec>,
@@ -586,7 +504,6 @@ where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec> + 'static,
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
{
beacon_node/beacon_chain/src/errors.rs
-3
View File
@@ -6,7 +6,6 @@ use state_processing::per_block_processing::errors::AttestationValidationError;
use state_processing::BlockProcessingError;
use state_processing::SlotProcessingError;
use std::time::Duration;
use store::block_root_tree::BlockRootTreeError;
use types::*;
macro_rules! easy_from_to {
@@ -51,13 +50,11 @@ pub enum BeaconChainError {
InvariantViolated(String),
SszTypesError(SszTypesError),
CanonicalHeadLockTimeout,
BlockRootTreeError(BlockRootTreeError),
}
easy_from_to!(SlotProcessingError, BeaconChainError);
easy_from_to!(AttestationValidationError, BeaconChainError);
easy_from_to!(SszTypesError, BeaconChainError);
easy_from_to!(BlockRootTreeError, BeaconChainError);
#[derive(Debug, PartialEq)]
pub enum BlockProductionError {
beacon_node/beacon_chain/src/fork_choice.rs
+88 -167
View File
@@ -1,13 +1,14 @@
mod checkpoint_manager;
use crate::{errors::BeaconChainError, metrics, BeaconChain, BeaconChainTypes};
use lmd_ghost::LmdGhost;
use parking_lot::RwLock;
use checkpoint_manager::{get_effective_balances, CheckpointManager, CheckpointWithBalances};
use parking_lot::{RwLock, RwLockReadGuard};
use proto_array_fork_choice::{core::ProtoArray, ProtoArrayForkChoice};
use ssz_derive::{Decode, Encode};
use state_processing::{common::get_attesting_indices, per_slot_processing};
use std::sync::Arc;
use store::{BlockRootTree, Error as StoreError, Store};
use types::{
Attestation, BeaconBlock, BeaconState, BeaconStateError, Checkpoint, EthSpec, Hash256, Slot,
};
use state_processing::common::get_attesting_indices;
use std::marker::PhantomData;
use store::Error as StoreError;
use types::{Attestation, BeaconBlock, BeaconState, BeaconStateError, Epoch, Hash256};
type Result<T> = std::result::Result<T, Error>;
@@ -19,27 +20,29 @@ pub enum Error {
BeaconStateError(BeaconStateError),
StoreError(StoreError),
BeaconChainError(Box<BeaconChainError>),
UnknownBlockSlot(Hash256),
UnknownJustifiedBlock(Hash256),
UnknownJustifiedState(Hash256),
UnableToJsonEncode(String),
}
pub struct ForkChoice<T: BeaconChainTypes> {
backend: T::LmdGhost,
backend: ProtoArrayForkChoice,
/// Used for resolving the `0x00..00` alias back to genesis.
///
/// Does not necessarily need to be the _actual_ genesis, it suffices to be the finalized root
/// whenever the struct was instantiated.
genesis_block_root: Hash256,
/// The fork choice rule's current view of the justified checkpoint.
justified_checkpoint: RwLock<Checkpoint>,
/// The best justified checkpoint we've seen, which may be ahead of `justified_checkpoint`.
best_justified_checkpoint: RwLock<Checkpoint>,
checkpoint_manager: RwLock<CheckpointManager>,
_phantom: PhantomData<T>,
}
impl<T: BeaconChainTypes> PartialEq for ForkChoice<T> {
/// This implementation ignores the `store`.
fn eq(&self, other: &Self) -> bool {
self.backend == other.backend
&& self.genesis_block_root == other.genesis_block_root
&& *self.justified_checkpoint.read() == *other.justified_checkpoint.read()
&& *self.best_justified_checkpoint.read() == *other.best_justified_checkpoint.read()
&& *self.checkpoint_manager.read() == *other.checkpoint_manager.read()
}
}
@@ -48,122 +51,48 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
///
/// "Genesis" does not necessarily need to be the absolute genesis, it can be some finalized
/// block.
pub fn new(backend: T::LmdGhost, genesis_block_root: Hash256, genesis_slot: Slot) -> Self {
let justified_checkpoint = Checkpoint {
epoch: genesis_slot.epoch(T::EthSpec::slots_per_epoch()),
pub fn new(
backend: ProtoArrayForkChoice,
genesis_block_root: Hash256,
genesis_state: &BeaconState<T::EthSpec>,
) -> Self {
let genesis_checkpoint = CheckpointWithBalances {
epoch: genesis_state.current_epoch(),
root: genesis_block_root,
balances: get_effective_balances(genesis_state),
};
Self {
backend,
genesis_block_root,
justified_checkpoint: RwLock::new(justified_checkpoint.clone()),
best_justified_checkpoint: RwLock::new(justified_checkpoint),
checkpoint_manager: RwLock::new(CheckpointManager::new(genesis_checkpoint)),
_phantom: PhantomData,
}
}
/// Determine whether the fork choice's view of the justified checkpoint should be updated.
///
/// To prevent the bouncing attack, an update is allowed only in these conditions:
///
/// * We're in the first SAFE_SLOTS_TO_UPDATE_JUSTIFIED slots of the epoch, or
/// * The new justified checkpoint is a descendant of the current justified checkpoint
fn should_update_justified_checkpoint(
&self,
chain: &BeaconChain<T>,
new_justified_checkpoint: &Checkpoint,
) -> Result<bool> {
if Self::compute_slots_since_epoch_start(chain.slot()?)
< chain.spec.safe_slots_to_update_justified
{
return Ok(true);
}
let justified_checkpoint = self.justified_checkpoint.read().clone();
let current_justified_block = chain
.get_block(&justified_checkpoint.root)?
.ok_or_else(|| Error::MissingBlock(justified_checkpoint.root))?;
let new_justified_block = chain
.get_block(&new_justified_checkpoint.root)?
.ok_or_else(|| Error::MissingBlock(new_justified_checkpoint.root))?;
let slots_per_epoch = T::EthSpec::slots_per_epoch();
Ok(
new_justified_block.slot > justified_checkpoint.epoch.start_slot(slots_per_epoch)
&& chain.get_ancestor_block_root(
new_justified_checkpoint.root,
current_justified_block.slot,
)? == Some(justified_checkpoint.root),
)
}
/// Calculate how far `slot` lies from the start of its epoch.
fn compute_slots_since_epoch_start(slot: Slot) -> u64 {
let slots_per_epoch = T::EthSpec::slots_per_epoch();
(slot - slot.epoch(slots_per_epoch).start_slot(slots_per_epoch)).as_u64()
}
/// Run the fork choice rule to determine the head.
pub fn find_head(&self, chain: &BeaconChain<T>) -> Result<Hash256> {
let timer = metrics::start_timer(&metrics::FORK_CHOICE_FIND_HEAD_TIMES);
let (start_state, start_block_root, start_block_slot) = {
// Check if we should update our view of the justified checkpoint.
// Doing this check here should be quasi-equivalent to the update in the `on_tick`
// function of the spec, so long as `find_head` is called at least once during the first
// SAFE_SLOTS_TO_UPDATE_JUSTIFIED slots.
let best_justified_checkpoint = self.best_justified_checkpoint.read();
if self.should_update_justified_checkpoint(chain, &best_justified_checkpoint)? {
*self.justified_checkpoint.write() = best_justified_checkpoint.clone();
}
let current_justified_checkpoint = self.justified_checkpoint.read().clone();
let (block_root, block_justified_slot) = (
current_justified_checkpoint.root,
current_justified_checkpoint
.epoch
.start_slot(T::EthSpec::slots_per_epoch()),
);
let block = chain
.store
.get::<BeaconBlock<T::EthSpec>>(&block_root)?
.ok_or_else(|| Error::MissingBlock(block_root))?;
// Resolve the `0x00.. 00` alias back to genesis
let block_root = if block_root == Hash256::zero() {
let remove_alias = |root| {
if root == Hash256::zero() {
self.genesis_block_root
} else {
block_root
};
let mut state: BeaconState<T::EthSpec> = chain
.get_state_caching_only_with_committee_caches(&block.state_root, Some(block.slot))?
.ok_or_else(|| Error::MissingState(block.state_root))?;
// Fast-forward the state to the start slot of the epoch where it was justified.
for _ in block.slot.as_u64()..block_justified_slot.as_u64() {
per_slot_processing(&mut state, None, &chain.spec)
.map_err(BeaconChainError::SlotProcessingError)?
root
}
(state, block_root, block_justified_slot)
};
// A function that returns the weight for some validator index.
let weight = |validator_index: usize| -> Option<u64> {
start_state
.validators
.get(validator_index)
.map(|v| v.effective_balance)
};
let mut manager = self.checkpoint_manager.write();
manager.maybe_update(chain.slot()?, chain)?;
let result = self
.backend
.find_head(start_block_slot, start_block_root, weight)
.find_head(
manager.current.justified.epoch,
remove_alias(manager.current.justified.root),
manager.current.finalized.epoch,
&manager.current.justified.balances,
)
.map_err(Into::into);
metrics::stop_timer(timer);
@@ -171,6 +100,11 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
result
}
/// Returns true if the given block is known to fork choice.
pub fn contains_block(&self, block_root: &Hash256) -> bool {
self.backend.contains_block(block_root)
}
/// Process all attestations in the given `block`.
///
/// Assumes the block (and therefore its attestations) are valid. It is a logic error to
@@ -183,36 +117,35 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
block_root: Hash256,
) -> Result<()> {
let timer = metrics::start_timer(&metrics::FORK_CHOICE_PROCESS_BLOCK_TIMES);
// Note: we never count the block as a latest message, only attestations.
//
// I (Paul H) do not have an explicit reference to this, but I derive it from this
// document:
//
// https://github.com/ethereum/eth2.0-specs/blob/v0.7.0/specs/core/0_fork-choice.md
for attestation in &block.body.attestations {
// If the `data.beacon_block_root` block is not known to us, simply ignore the latest
// vote.
if let Some(block) = chain.get_block_caching(&attestation.data.beacon_block_root)? {
self.process_attestation(state, attestation, &block)?;
}
}
// Check if we should update our view of the justified checkpoint
if state.current_justified_checkpoint.epoch > self.justified_checkpoint.read().epoch {
*self.best_justified_checkpoint.write() = state.current_justified_checkpoint.clone();
self.checkpoint_manager
.write()
.process_state(block_root, state, chain, &self.backend)?;
self.checkpoint_manager
.write()
.maybe_update(chain.slot()?, chain)?;
// Note: we never count the block as a latest message, only attestations.
for attestation in &block.body.attestations {
// If the `data.beacon_block_root` block is not known to the fork choice, simply ignore
// the vote.
if self
.should_update_justified_checkpoint(chain, &state.current_justified_checkpoint)?
.backend
.contains_block(&attestation.data.beacon_block_root)
{
*self.justified_checkpoint.write() = state.current_justified_checkpoint.clone();
self.process_attestation(state, attestation)?;
}
}
// This does not apply a vote to the block, it just makes fork choice aware of the block so
// it can still be identified as the head even if it doesn't have any votes.
//
// A case where a block without any votes can be the head is where it is the only child of
// a block that has the majority of votes applied to it.
self.backend.process_block(block, block_root)?;
self.backend.process_block(
block.slot,
block_root,
block.parent_root,
state.current_justified_checkpoint.epoch,
state.finalized_checkpoint.epoch,
)?;
metrics::stop_timer(timer);
@@ -226,7 +159,6 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
&self,
state: &BeaconState<T::EthSpec>,
attestation: &Attestation<T::EthSpec>,
block: &BeaconBlock<T::EthSpec>,
) -> Result<()> {
let timer = metrics::start_timer(&metrics::FORK_CHOICE_PROCESS_ATTESTATION_TIMES);
@@ -252,8 +184,11 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
get_attesting_indices(state, &attestation.data, &attestation.aggregation_bits)?;
for validator_index in validator_indices {
self.backend
.process_attestation(validator_index, block_hash, block.slot)?;
self.backend.process_attestation(
validator_index,
block_hash,
attestation.data.target.epoch,
)?;
}
}
@@ -265,38 +200,29 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
/// Returns the latest message for a given validator, if any.
///
/// Returns `(block_root, block_slot)`.
pub fn latest_message(&self, validator_index: usize) -> Option<(Hash256, Slot)> {
pub fn latest_message(&self, validator_index: usize) -> Option<(Hash256, Epoch)> {
self.backend.latest_message(validator_index)
}
/// Runs an integrity verification function on the underlying fork choice algorithm.
///
/// Returns `Ok(())` if the underlying fork choice has maintained it's integrity,
/// `Err(description)` otherwise.
pub fn verify_integrity(&self) -> core::result::Result<(), String> {
self.backend.verify_integrity()
/// Trigger a prune on the underlying fork choice backend.
pub fn prune(&self) -> Result<()> {
let finalized_root = self.checkpoint_manager.read().current.finalized.root;
self.backend.maybe_prune(finalized_root).map_err(Into::into)
}
/// Inform the fork choice that the given block (and corresponding root) have been finalized so
/// it may prune it's storage.
/// Returns a read-lock to the core `ProtoArray` struct.
///
/// `finalized_block_root` must be the root of `finalized_block`.
pub fn process_finalization(
&self,
finalized_block: &BeaconBlock<T::EthSpec>,
finalized_block_root: Hash256,
) -> Result<()> {
self.backend
.update_finalized_root(finalized_block, finalized_block_root)
.map_err(Into::into)
/// Should only be used when encoding/decoding during troubleshooting.
pub fn core_proto_array(&self) -> RwLockReadGuard<ProtoArray> {
self.backend.core_proto_array()
}
/// Returns a `SszForkChoice` which contains the current state of `Self`.
pub fn as_ssz_container(&self) -> SszForkChoice {
SszForkChoice {
genesis_block_root: self.genesis_block_root,
justified_checkpoint: self.justified_checkpoint.read().clone(),
best_justified_checkpoint: self.best_justified_checkpoint.read().clone(),
genesis_block_root: self.genesis_block_root.clone(),
checkpoint_manager: self.checkpoint_manager.read().clone(),
backend_bytes: self.backend.as_bytes(),
}
}
@@ -304,18 +230,14 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
/// Instantiates `Self` from a prior `SszForkChoice`.
///
/// The created `Self` will have the same state as the `Self` that created the `SszForkChoice`.
pub fn from_ssz_container(
ssz_container: SszForkChoice,
store: Arc<T::Store>,
block_root_tree: Arc<BlockRootTree>,
) -> Result<Self> {
let backend = LmdGhost::from_bytes(&ssz_container.backend_bytes, store, block_root_tree)?;
pub fn from_ssz_container(ssz_container: SszForkChoice) -> Result<Self> {
let backend = ProtoArrayForkChoice::from_bytes(&ssz_container.backend_bytes)?;
Ok(Self {
backend,
genesis_block_root: ssz_container.genesis_block_root,
justified_checkpoint: RwLock::new(ssz_container.justified_checkpoint),
best_justified_checkpoint: RwLock::new(ssz_container.best_justified_checkpoint),
checkpoint_manager: RwLock::new(ssz_container.checkpoint_manager),
_phantom: PhantomData,
})
}
}
@@ -326,8 +248,7 @@ impl<T: BeaconChainTypes> ForkChoice<T> {
#[derive(Encode, Decode, Clone)]
pub struct SszForkChoice {
genesis_block_root: Hash256,
justified_checkpoint: Checkpoint,
best_justified_checkpoint: Checkpoint,
checkpoint_manager: CheckpointManager,
backend_bytes: Vec<u8>,
}
beacon_node/beacon_chain/src/fork_choice/checkpoint_manager.rs
+340
View File
@@ -0,0 +1,340 @@
use super::Error;
use crate::{metrics, BeaconChain, BeaconChainTypes};
use proto_array_fork_choice::ProtoArrayForkChoice;
use ssz_derive::{Decode, Encode};
use types::{BeaconState, Checkpoint, Epoch, EthSpec, Hash256, Slot};
const MAX_BALANCE_CACHE_SIZE: usize = 4;
/// An item that is stored in the `BalancesCache`.
#[derive(PartialEq, Clone, Encode, Decode)]
struct CacheItem {
/// The block root at which `self.balances` are valid.
block_root: Hash256,
/// The `state.balances` list.
balances: Vec<u64>,
}
/// Provides a cache to avoid reading `BeaconState` from disk when updating the current justified
/// checkpoint.
///
/// It should store a mapping of `epoch_boundary_block_root -> state.balances`.
#[derive(PartialEq, Clone, Default, Encode, Decode)]
struct BalancesCache {
items: Vec<CacheItem>,
}
impl BalancesCache {
/// Inspect the given `state` and determine the root of the block at the first slot of
/// `state.current_epoch`. If there is not already some entry for the given block root, then
/// add `state.balances` to the cache.
pub fn process_state<E: EthSpec>(
&mut self,
block_root: Hash256,
state: &BeaconState<E>,
) -> Result<(), Error> {
// We are only interested in balances from states that are at the start of an epoch,
// because this is where the `current_justified_checkpoint.root` will point.
if !Self::is_first_block_in_epoch(block_root, state)? {
return Ok(());
}
let epoch_boundary_slot = state.current_epoch().start_slot(E::slots_per_epoch());
let epoch_boundary_root = if epoch_boundary_slot == state.slot {
block_root
} else {
// This call remains sensible as long as `state.block_roots` is larger than a single
// epoch.
*state.get_block_root(epoch_boundary_slot)?
};
if self.position(epoch_boundary_root).is_none() {
let item = CacheItem {
block_root: epoch_boundary_root,
balances: get_effective_balances(state),
};
if self.items.len() == MAX_BALANCE_CACHE_SIZE {
self.items.remove(0);
}
self.items.push(item);
}
Ok(())
}
/// Returns `true` if the given `block_root` is the first/only block to have been processed in
/// the epoch of the given `state`.
///
/// We can determine if it is the first block by looking back through `state.block_roots` to
/// see if there is a block in the current epoch with a different root.
fn is_first_block_in_epoch<E: EthSpec>(
block_root: Hash256,
state: &BeaconState<E>,
) -> Result<bool, Error> {
let mut prior_block_found = false;
for slot in state.current_epoch().slot_iter(E::slots_per_epoch()) {
if slot < state.slot {
if *state.get_block_root(slot)? != block_root {
prior_block_found = true;
break;
}
} else {
break;
}
}
Ok(!prior_block_found)
}
fn position(&self, block_root: Hash256) -> Option<usize> {
self.items
.iter()
.position(|item| item.block_root == block_root)
}
/// Get the balances for the given `block_root`, if any.
///
/// If some balances are found, they are removed from the cache.
pub fn get(&mut self, block_root: Hash256) -> Option<Vec<u64>> {
let i = self.position(block_root)?;
Some(self.items.remove(i).balances)
}
}
/// Returns the effective balances for every validator in the given `state`.
///
/// Any validator who is not active in the epoch of the given `state` is assigned a balance of
/// zero.
pub fn get_effective_balances<T: EthSpec>(state: &BeaconState<T>) -> Vec<u64> {
state
.validators
.iter()
.map(|validator| {
if validator.is_active_at(state.current_epoch()) {
validator.effective_balance
} else {
0
}
})
.collect()
}
/// A `types::Checkpoint` that also stores the validator balances from a `BeaconState`.
///
/// Useful because we need to track the justified checkpoint balances.
#[derive(PartialEq, Clone, Encode, Decode)]
pub struct CheckpointWithBalances {
pub epoch: Epoch,
pub root: Hash256,
/// These are the balances of the state with `self.root`.
///
/// Importantly, these are _not_ the balances of the first state that we saw that has
/// `self.epoch` and `self.root` as `state.current_justified_checkpoint`. These are the
/// balances of the state from the block with `state.current_justified_checkpoint.root`.
pub balances: Vec<u64>,
}
impl Into<Checkpoint> for CheckpointWithBalances {
fn into(self) -> Checkpoint {
Checkpoint {
epoch: self.epoch,
root: self.root,
}
}
}
/// A pair of checkpoints, representing `state.current_justified_checkpoint` and
/// `state.finalized_checkpoint` for some `BeaconState`.
#[derive(PartialEq, Clone, Encode, Decode)]
pub struct FFGCheckpoints {
pub justified: CheckpointWithBalances,
pub finalized: Checkpoint,
}
/// A struct to manage the justified and finalized checkpoints to be used for `ForkChoice`.
///
/// This struct exists to manage the `should_update_justified_checkpoint` logic in the fork choice
/// section of the spec:
///
/// https://github.com/ethereum/eth2.0-specs/blob/dev/specs/phase0/fork-choice.md#should_update_justified_checkpoint
#[derive(PartialEq, Clone, Encode, Decode)]
pub struct CheckpointManager {
/// The current FFG checkpoints that should be used for finding the head.
pub current: FFGCheckpoints,
/// The best-known checkpoints that should be moved to `self.current` when the time is right.
best: FFGCheckpoints,
/// The epoch at which `self.current` should become `self.best`, if any.
update_at: Option<Epoch>,
/// A cached used to try and avoid DB reads when updating `self.current` and `self.best`.
balances_cache: BalancesCache,
}
impl CheckpointManager {
/// Create a new checkpoint cache from `genesis_checkpoint` derived from the genesis block.
pub fn new(genesis_checkpoint: CheckpointWithBalances) -> Self {
let ffg_checkpoint = FFGCheckpoints {
justified: genesis_checkpoint.clone(),
finalized: genesis_checkpoint.into(),
};
Self {
current: ffg_checkpoint.clone(),
best: ffg_checkpoint,
update_at: None,
balances_cache: BalancesCache::default(),
}
}
/// Potentially updates `self.current`, if the conditions are correct.
///
/// Should be called before running the fork choice `find_head` function to ensure
/// `self.current` is up-to-date.
pub fn maybe_update<T: BeaconChainTypes>(
&mut self,
current_slot: Slot,
chain: &BeaconChain<T>,
) -> Result<(), Error> {
if self.best.justified.epoch > self.current.justified.epoch {
let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
match self.update_at {
None => {
if self.best.justified.epoch > self.current.justified.epoch {
if Self::compute_slots_since_epoch_start::<T>(current_slot)
< chain.spec.safe_slots_to_update_justified
{
self.current = self.best.clone();
} else {
self.update_at = Some(current_epoch + 1)
}
}
}
Some(epoch) if epoch <= current_epoch => {
self.current = self.best.clone();
self.update_at = None
}
_ => {}
}
}
Ok(())
}
/// Checks the given `state` (must correspond to the given `block_root`) to see if it contains
/// a `current_justified_checkpoint` that is better than `self.best_justified_checkpoint`. If
/// so, the value is updated.
///
/// Note: this does not update `self.justified_checkpoint`.
pub fn process_state<T: BeaconChainTypes>(
&mut self,
block_root: Hash256,
state: &BeaconState<T::EthSpec>,
chain: &BeaconChain<T>,
proto_array: &ProtoArrayForkChoice,
) -> Result<(), Error> {
// Only proceed if the new checkpoint is better than our current checkpoint.
if state.current_justified_checkpoint.epoch > self.current.justified.epoch
&& state.finalized_checkpoint.epoch >= self.current.finalized.epoch
{
let candidate = FFGCheckpoints {
justified: CheckpointWithBalances {
epoch: state.current_justified_checkpoint.epoch,
root: state.current_justified_checkpoint.root,
balances: self
.get_balances_for_block(state.current_justified_checkpoint.root, chain)?,
},
finalized: state.finalized_checkpoint.clone(),
};
// Using the given `state`, determine its ancestor at the slot of our current justified
// epoch. Later, this will be compared to the root of the current justified checkpoint
// to determine if this state is descendant of our current justified state.
let new_checkpoint_ancestor = Self::get_block_root_at_slot(
state,
chain,
candidate.justified.root,
self.current
.justified
.epoch
.start_slot(T::EthSpec::slots_per_epoch()),
)?;
let candidate_justified_block_slot = proto_array
.block_slot(&candidate.justified.root)
.ok_or_else(|| Error::UnknownBlockSlot(candidate.justified.root))?;
// If the new justified checkpoint is an ancestor of the current justified checkpoint,
// it is always safe to change it.
if new_checkpoint_ancestor == Some(self.current.justified.root)
&& candidate_justified_block_slot
>= candidate
.justified
.epoch
.start_slot(T::EthSpec::slots_per_epoch())
{
self.current = candidate.clone()
}
if candidate.justified.epoch > self.best.justified.epoch {
// Always update the best checkpoint, if it's better.
self.best = candidate;
}
// Add the state's balances to the balances cache to avoid a state read later.
self.balances_cache.process_state(block_root, state)?;
}
Ok(())
}
fn get_balances_for_block<T: BeaconChainTypes>(
&mut self,
block_root: Hash256,
chain: &BeaconChain<T>,
) -> Result<Vec<u64>, Error> {
if let Some(balances) = self.balances_cache.get(block_root) {
metrics::inc_counter(&metrics::BALANCES_CACHE_HITS);
Ok(balances)
} else {
metrics::inc_counter(&metrics::BALANCES_CACHE_MISSES);
let block = chain
.get_block_caching(&block_root)?
.ok_or_else(|| Error::UnknownJustifiedBlock(block_root))?;
let state = chain
.get_state_caching_only_with_committee_caches(&block.state_root, Some(block.slot))?
.ok_or_else(|| Error::UnknownJustifiedState(block.state_root))?;
Ok(get_effective_balances(&state))
}
}
/// Attempts to get the block root for the given `slot`.
///
/// First, the `state` is used to see if the slot is within the distance of its historical
/// lists. Then, the `chain` is used which will anchor the search at the given
/// `justified_root`.
fn get_block_root_at_slot<T: BeaconChainTypes>(
state: &BeaconState<T::EthSpec>,
chain: &BeaconChain<T>,
justified_root: Hash256,
slot: Slot,
) -> Result<Option<Hash256>, Error> {
match state.get_block_root(slot) {
Ok(root) => Ok(Some(*root)),
Err(_) => chain
.get_ancestor_block_root(justified_root, slot)
.map_err(Into::into),
}
}
/// Calculate how far `slot` lies from the start of its epoch.
fn compute_slots_since_epoch_start<T: BeaconChainTypes>(slot: Slot) -> u64 {
let slots_per_epoch = T::EthSpec::slots_per_epoch();
(slot - slot.epoch(slots_per_epoch).start_slot(slots_per_epoch)).as_u64()
}
}
beacon_node/beacon_chain/src/lib.rs
-1
View File
@@ -24,7 +24,6 @@ pub use self::errors::{BeaconChainError, BlockProductionError};
pub use eth1_chain::{Eth1Chain, Eth1ChainBackend};
pub use events::EventHandler;
pub use fork_choice::ForkChoice;
pub use lmd_ghost;
pub use metrics::scrape_for_metrics;
pub use parking_lot;
pub use slot_clock;
beacon_node/beacon_chain/src/metrics.rs
+4
View File
@@ -138,6 +138,10 @@ lazy_static! {
"beacon_fork_choice_process_attestation_seconds",
"Time taken to add an attestation to fork choice"
);
pub static ref BALANCES_CACHE_HITS: Result<IntCounter> =
try_create_int_counter("beacon_balances_cache_hits_total", "Count of times balances cache fulfils request");
pub static ref BALANCES_CACHE_MISSES: Result<IntCounter> =
try_create_int_counter("beacon_balances_cache_misses_total", "Count of times balances cache fulfils request");
/*
* Persisting BeaconChain to disk
beacon_node/beacon_chain/src/persisted_beacon_chain.rs
+1 -2
View File
@@ -5,7 +5,7 @@ use crate::{BeaconChainTypes, CheckPoint};
use operation_pool::PersistedOperationPool;
use ssz::{Decode, Encode};
use ssz_derive::{Decode, Encode};
use store::{DBColumn, Error as StoreError, SimpleStoreItem, SszBlockRootTree};
use store::{DBColumn, Error as StoreError, SimpleStoreItem};
use types::Hash256;
/// 32-byte key for accessing the `PersistedBeaconChain`.
@@ -20,7 +20,6 @@ pub struct PersistedBeaconChain<T: BeaconChainTypes> {
pub ssz_head_tracker: SszHeadTracker,
pub fork_choice: SszForkChoice,
pub eth1_cache: Option<SszEth1>,
pub block_root_tree: SszBlockRootTree,
}
impl<T: BeaconChainTypes> SimpleStoreItem for PersistedBeaconChain<T> {
beacon_node/beacon_chain/src/test_utils.rs
-2
View File
@@ -6,7 +6,6 @@ use crate::{
};
use eth1::Config as Eth1Config;
use genesis::interop_genesis_state;
use lmd_ghost::ThreadSafeReducedTree;
use rayon::prelude::*;
use sloggers::{terminal::TerminalLoggerBuilder, types::Severity, Build};
use slot_clock::TestingSlotClock;
@@ -35,7 +34,6 @@ pub type BaseHarnessType<TStore, TStoreMigrator, TEthSpec> = Witness<
TStore,
TStoreMigrator,
TestingSlotClock,
ThreadSafeReducedTree<TStore, TEthSpec>,
CachingEth1Backend<TEthSpec, TStore>,
TEthSpec,
NullEventHandler<TEthSpec>,
beacon_node/beacon_chain/tests/tests.rs
+2 -2
View File
@@ -391,7 +391,7 @@ fn free_attestations_added_to_fork_choice_some_none() {
if slot <= num_blocks_produced && slot != 0 {
assert_eq!(
latest_message.unwrap().1,
slot,
slot.epoch(MinimalEthSpec::slots_per_epoch()),
"Latest message slot for {} should be equal to slot {}.",
validator,
slot
@@ -483,7 +483,7 @@ fn free_attestations_added_to_fork_choice_all_updated() {
assert_eq!(
latest_message.unwrap().1,
slot,
slot.epoch(MinimalEthSpec::slots_per_epoch()),
"Latest message slot should be equal to attester duty."
);
beacon_node/client/Cargo.toml
-1
View File
@@ -33,7 +33,6 @@ exit-future = "0.1.4"
futures = "0.1.29"
reqwest = "0.9.22"
url = "2.1.0"
lmd_ghost = { path = "../../eth2/lmd_ghost" }
eth1 = { path = "../eth1" }
genesis = { path = "../genesis" }
environment = { path = "../../lighthouse/environment" }
beacon_node/client/src/builder.rs
+13 -62
View File
@@ -4,7 +4,6 @@ use crate::Client;
use beacon_chain::{
builder::{BeaconChainBuilder, Witness},
eth1_chain::CachingEth1Backend,
lmd_ghost::ThreadSafeReducedTree,
slot_clock::{SlotClock, SystemTimeSlotClock},
store::{
migrate::{BackgroundMigrator, Migrate, NullMigrator},
@@ -21,7 +20,6 @@ use genesis::{
generate_deterministic_keypairs, interop_genesis_state, state_from_ssz_file, Eth1GenesisService,
};
use lighthouse_bootstrap::Bootstrapper;
use lmd_ghost::LmdGhost;
use network::{NetworkConfig, NetworkMessage, Service as NetworkService};
use slog::info;
use ssz::Decode;
@@ -67,23 +65,14 @@ pub struct ClientBuilder<T: BeaconChainTypes> {
eth_spec_instance: T::EthSpec,
}
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>
where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec>,
TSlotClock: SlotClock + Clone + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -367,17 +356,8 @@ where
/// If type inference errors are being raised, see the comment on the definition of `Self`.
pub fn build(
self,
) -> Client<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
> {
) -> Client<Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>>
{
Client {
beacon_chain: self.beacon_chain,
libp2p_network: self.libp2p_network,
@@ -390,15 +370,7 @@ where
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
ThreadSafeReducedTree<TStore, TEthSpec>,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>
where
TStore: Store<TEthSpec> + 'static,
@@ -435,13 +407,12 @@ where
}
}
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec>
impl<TStore, TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec>
ClientBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
WebSocketSender<TEthSpec>,
@@ -451,7 +422,6 @@ where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec>,
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
{
@@ -485,13 +455,12 @@ where
}
}
impl<TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
DiskStore<TEthSpec>,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
@@ -500,7 +469,6 @@ impl<TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandle
where
TSlotClock: SlotClock + 'static,
TStoreMigrator: store::Migrate<DiskStore<TEthSpec>, TEthSpec> + 'static,
TLmdGhost: LmdGhost<DiskStore<TEthSpec>, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, DiskStore<TEthSpec>> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -535,13 +503,12 @@ where
}
}
impl<TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TStoreMigrator, TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
SimpleDiskStore<TEthSpec>,
TStoreMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
@@ -550,7 +517,6 @@ impl<TStoreMigrator, TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandle
where
TSlotClock: SlotClock + 'static,
TStoreMigrator: store::Migrate<SimpleDiskStore<TEthSpec>, TEthSpec> + 'static,
TLmdGhost: LmdGhost<SimpleDiskStore<TEthSpec>, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, SimpleDiskStore<TEthSpec>> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -564,13 +530,12 @@ where
}
}
impl<TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
MemoryStore<TEthSpec>,
NullMigrator,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
@@ -578,7 +543,6 @@ impl<TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
>
where
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<MemoryStore<TEthSpec>, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, MemoryStore<TEthSpec>> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -594,13 +558,12 @@ where
}
}
impl<TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TSlotClock, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
DiskStore<TEthSpec>,
BackgroundMigrator<TEthSpec>,
TSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
@@ -608,7 +571,6 @@ impl<TSlotClock, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
>
where
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<DiskStore<TEthSpec>, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, DiskStore<TEthSpec>> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
@@ -622,13 +584,12 @@ where
}
}
impl<TStore, TStoreMigrator, TSlotClock, TLmdGhost, TEthSpec, TEventHandler>
impl<TStore, TStoreMigrator, TSlotClock, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
TStore,
TStoreMigrator,
TSlotClock,
TLmdGhost,
CachingEth1Backend<TEthSpec, TStore>,
TEthSpec,
TEventHandler,
@@ -638,7 +599,6 @@ where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec>,
TSlotClock: SlotClock + 'static,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
{
@@ -724,22 +684,13 @@ where
}
}
impl<TStore, TStoreMigrator, TLmdGhost, TEth1Backend, TEthSpec, TEventHandler>
impl<TStore, TStoreMigrator, TEth1Backend, TEthSpec, TEventHandler>
ClientBuilder<
Witness<
TStore,
TStoreMigrator,
SystemTimeSlotClock,
TLmdGhost,
TEth1Backend,
TEthSpec,
TEventHandler,
>,
Witness<TStore, TStoreMigrator, SystemTimeSlotClock, TEth1Backend, TEthSpec, TEventHandler>,
>
where
TStore: Store<TEthSpec> + 'static,
TStoreMigrator: store::Migrate<TStore, TEthSpec>,
TLmdGhost: LmdGhost<TStore, TEthSpec> + 'static,
TEth1Backend: Eth1ChainBackend<TEthSpec, TStore> + 'static,
TEthSpec: EthSpec + 'static,
TEventHandler: EventHandler<TEthSpec> + 'static,
beacon_node/eth2-libp2p/src/rpc/handler.rs
+7 -3
View File
@@ -12,7 +12,7 @@ use libp2p::core::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeError};
use libp2p::swarm::protocols_handler::{
KeepAlive, ProtocolsHandler, ProtocolsHandlerEvent, ProtocolsHandlerUpgrErr, SubstreamProtocol,
};
use slog::{crit, debug, error};
use slog::{crit, debug, error, warn};
use smallvec::SmallVec;
use std::collections::hash_map::Entry;
use std::time::{Duration, Instant};
@@ -319,8 +319,12 @@ where
substream: out,
request,
};
self.outbound_substreams
.insert(id, (awaiting_stream, delay_key));
if let Some(_) = self
.outbound_substreams
.insert(id, (awaiting_stream, delay_key))
{
warn!(self.log, "Duplicate outbound substream id"; "id" => format!("{:?}", id));
}
}
_ => { // a response is not expected, drop the stream for all other requests
}
beacon_node/network/src/sync/range_sync/batch_processing.rs
+1 -1
View File
@@ -70,7 +70,7 @@ fn process_batch<T: BeaconChainTypes>(
);
successful_block_import = true;
}
BlockProcessingOutcome::ParentUnknown { parent } => {
BlockProcessingOutcome::ParentUnknown { parent, .. } => {
// blocks should be sequential and all parents should exist
warn!(
log, "Parent block is unknown";
beacon_node/network/src/sync/range_sync/chain_collection.rs
+4 -4
View File
@@ -307,8 +307,8 @@ impl<T: BeaconChainTypes> ChainCollection<T> {
self.finalized_chains.retain(|chain| {
if chain.target_head_slot <= local_finalized_slot
|| beacon_chain
.block_root_tree
.is_known_block_root(&chain.target_head_root)
.fork_choice
.contains_block(&chain.target_head_root)
{
debug!(log, "Purging out of finalized chain"; "start_slot" => chain.start_slot, "end_slot" => chain.target_head_slot);
chain.status_peers(network);
@@ -320,8 +320,8 @@ impl<T: BeaconChainTypes> ChainCollection<T> {
self.head_chains.retain(|chain| {
if chain.target_head_slot <= local_finalized_slot
|| beacon_chain
.block_root_tree
.is_known_block_root(&chain.target_head_root)
.fork_choice
.contains_block(&chain.target_head_root)
{
debug!(log, "Purging out of date head chain"; "start_slot" => chain.start_slot, "end_slot" => chain.target_head_slot);
chain.status_peers(network);
beacon_node/network/src/sync/range_sync/range.rs
+1 -3
View File
@@ -150,9 +150,7 @@ impl<T: BeaconChainTypes> RangeSync<T> {
self.chains.purge_outdated_chains(network, &self.log);
if remote_finalized_slot > local_info.head_slot
&& !chain
.block_root_tree
.is_known_block_root(&remote.finalized_root)
&& !chain.fork_choice.contains_block(&remote.finalized_root)
{
debug!(self.log, "Finalization sync peer joined"; "peer_id" => format!("{:?}", peer_id));
// Finalized chain search
beacon_node/rest_api/src/advanced.rs
+15
View File
@@ -0,0 +1,15 @@
use crate::response_builder::ResponseBuilder;
use crate::ApiResult;
use beacon_chain::{BeaconChain, BeaconChainTypes};
use hyper::{Body, Request};
use std::sync::Arc;
/// Returns the `proto_array` fork choice struct, encoded as JSON.
///
/// Useful for debugging or advanced inspection of the chain.
pub fn get_fork_choice<T: BeaconChainTypes>(
req: Request<Body>,
beacon_chain: Arc<BeaconChain<T>>,
) -> ApiResult {
ResponseBuilder::new(&req)?.body_no_ssz(&*beacon_chain.fork_choice.core_proto_array())
}
beacon_node/rest_api/src/lib.rs
+1
View File
@@ -4,6 +4,7 @@ mod macros;
extern crate lazy_static;
extern crate network as client_network;
mod advanced;
mod beacon;
pub mod config;
mod consensus;
beacon_node/rest_api/src/router.rs
+7 -2
View File
@@ -1,6 +1,6 @@
use crate::{
beacon, consensus, error::ApiError, helpers, metrics, network, node, spec, validator, BoxFut,
NetworkChannel,
advanced, beacon, consensus, error::ApiError, helpers, metrics, network, node, spec, validator,
BoxFut, NetworkChannel,
};
use beacon_chain::{BeaconChain, BeaconChainTypes};
use client_network::Service as NetworkService;
@@ -147,6 +147,11 @@ pub fn route<T: BeaconChainTypes>(
into_boxfut(spec::get_eth2_config::<T>(req, eth2_config))
}
// Methods for advanced parameters
(&Method::GET, "/advanced/fork_choice") => {
into_boxfut(advanced::get_fork_choice::<T>(req, beacon_chain))
}
(&Method::GET, "/metrics") => into_boxfut(metrics::get_prometheus::<T>(
req,
beacon_chain,
beacon_node/rest_api/tests/test.rs
+24
View File
@@ -792,6 +792,30 @@ fn get_committees() {
assert_eq!(result, expected, "result should be as expected");
}
#[test]
fn get_fork_choice() {
let mut env = build_env();
let node = build_node(&mut env, testing_client_config());
let remote_node = node.remote_node().expect("should produce remote node");
let fork_choice = env
.runtime()
.block_on(remote_node.http.advanced().get_fork_choice())
.expect("should not error when getting fork choice");
assert_eq!(
fork_choice,
*node
.client
.beacon_chain()
.expect("node should have beacon chain")
.fork_choice
.core_proto_array(),
"result should be as expected"
);
}
fn compare_validator_response<T: EthSpec>(
state: &BeaconState<T>,
response: &ValidatorResponse,
beacon_node/src/lib.rs
+1 -2
View File
@@ -11,7 +11,7 @@ pub use eth2_config::Eth2Config;
use beacon_chain::{
builder::Witness, eth1_chain::CachingEth1Backend, events::WebSocketSender,
lmd_ghost::ThreadSafeReducedTree, slot_clock::SystemTimeSlotClock,
slot_clock::SystemTimeSlotClock,
};
use clap::ArgMatches;
use config::get_configs;
@@ -28,7 +28,6 @@ pub type ProductionClient<E> = Client<
DiskStore<E>,
BackgroundMigrator<E>,
SystemTimeSlotClock,
ThreadSafeReducedTree<DiskStore<E>, E>,
CachingEth1Backend<E, DiskStore<E>>,
E,
WebSocketSender<E>,
beacon_node/store/src/block_root_tree.rs
-363
View File
@@ -1,363 +0,0 @@
use itertools::Itertools;
use parking_lot::RwLock;
use ssz_derive::{Decode, Encode};
use std::collections::{HashMap, HashSet};
use std::iter::{self, FromIterator};
use types::{Hash256, Slot};
/// In-memory cache of all block roots post-finalization. Includes short-lived forks.
///
/// Used by fork choice to avoid reconstructing hot states just for their block roots.
// NOTE: could possibly be streamlined by combining with the head tracker and/or fork choice
#[derive(Debug)]
pub struct BlockRootTree {
nodes: RwLock<HashMap<Hash256, Node>>,
}
impl Clone for BlockRootTree {
fn clone(&self) -> Self {
Self {
nodes: RwLock::new(self.nodes.read().clone()),
}
}
}
#[derive(Debug, PartialEq)]
pub enum BlockRootTreeError {
PrevUnknown(Hash256),
}
/// Data for a single `block_root` in the tree.
#[derive(Debug, Clone, Encode, Decode)]
struct Node {
/// Hash of the preceding block (should be the parent block).
///
/// A `previous` of `Hash256::zero` indicates the root of the tree.
previous: Hash256,
/// Slot of this node's block.
slot: Slot,
}
impl BlockRootTree {
/// Create a new block root tree where `(root_hash, root_slot)` is considered finalized.
///
/// All subsequent blocks added should descend from the root block.
pub fn new(root_hash: Hash256, root_slot: Slot) -> Self {
Self {
nodes: RwLock::new(HashMap::from_iter(iter::once((
root_hash,
Node {
previous: Hash256::zero(),
slot: root_slot,
},
)))),
}
}
/// Check if `block_root` exists in the tree.
pub fn is_known_block_root(&self, block_root: &Hash256) -> bool {
self.nodes.read().contains_key(block_root)
}
/// Add a new `block_root` to the tree.
///
/// Will return an error if `prev_block_root` doesn't exist in the tree.
pub fn add_block_root(
&self,
block_root: Hash256,
prev_block_root: Hash256,
block_slot: Slot,
) -> Result<(), BlockRootTreeError> {
let mut nodes = self.nodes.write();
if nodes.contains_key(&prev_block_root) {
nodes.insert(
block_root,
Node {
previous: prev_block_root,
slot: block_slot,
},
);
Ok(())
} else {
Err(BlockRootTreeError::PrevUnknown(prev_block_root))
}
}
/// Create a reverse iterator from `block_root` (inclusive).
///
/// Will skip slots, see `every_slot_iter_from` for a non-skipping variant.
pub fn iter_from(&self, block_root: Hash256) -> BlockRootTreeIter {
BlockRootTreeIter {
tree: self,
current_block_root: block_root,
}
}
/// Create a reverse iterator that yields a block root for every slot.
///
/// E.g. if slot 6 is skipped, this iterator will return the block root from slot 5 at slot 6.
pub fn every_slot_iter_from<'a>(
&'a self,
block_root: Hash256,
) -> impl Iterator<Item = (Hash256, Slot)> + 'a {
let mut block_roots = self.iter_from(block_root).peekable();
// Include the value for the first `block_root` if any, then fill in the skipped slots
// between each pair of previous block roots by duplicating the older root.
block_roots
.peek()
.cloned()
.into_iter()
.chain(block_roots.tuple_windows().flat_map(
|((_, high_slot), (low_hash, low_slot))| {
(low_slot.as_u64()..high_slot.as_u64())
.rev()
.map(move |slot| (low_hash, Slot::new(slot)))
},
))
}
/// Prune the tree.
///
/// Only keep block roots descended from `finalized_root`, which lie on a chain leading
/// to one of the heads contained in `heads`.
pub fn prune_to(&self, finalized_root: Hash256, heads: impl IntoIterator<Item = Hash256>) {
let mut keep = HashSet::new();
keep.insert(finalized_root);
for head_block_root in heads.into_iter() {
// Iterate backwards until we reach a portion of the chain that we've already decided
// to keep. This also discards the pre-finalization block roots.
let mut keep_head = false;
let head_blocks = self
.iter_from(head_block_root)
.map(|(block_root, _)| block_root)
.inspect(|block_root| {
if block_root == &finalized_root {
keep_head = true;
}
})
.take_while(|block_root| !keep.contains(&block_root))
.collect::<HashSet<_>>();
// If the head descends from the finalized root, keep it. Else throw it out.
if keep_head {
keep.extend(head_blocks);
}
}
self.nodes
.write()
.retain(|block_root, _| keep.contains(block_root));
}
pub fn as_ssz_container(&self) -> SszBlockRootTree {
SszBlockRootTree {
nodes: Vec::from_iter(self.nodes.read().clone()),
}
}
}
/// Simple (skipping) iterator for `BlockRootTree`.
#[derive(Debug)]
pub struct BlockRootTreeIter<'a> {
tree: &'a BlockRootTree,
current_block_root: Hash256,
}
impl<'a> Iterator for BlockRootTreeIter<'a> {
type Item = (Hash256, Slot);
fn next(&mut self) -> Option<Self::Item> {
// Genesis
if self.current_block_root.is_zero() {
None
} else {
let block_root = self.current_block_root;
self.tree.nodes.read().get(&block_root).map(|node| {
self.current_block_root = node.previous;
(block_root, node.slot)
})
}
}
}
/// Serializable version of `BlockRootTree` that can be persisted to disk.
#[derive(Debug, Clone, Encode, Decode)]
pub struct SszBlockRootTree {
nodes: Vec<(Hash256, Node)>,
}
impl Into<BlockRootTree> for SszBlockRootTree {
fn into(self) -> BlockRootTree {
BlockRootTree {
nodes: RwLock::new(HashMap::from_iter(self.nodes)),
}
}
}
#[cfg(test)]
mod test {
use super::*;
fn int_hash(x: u64) -> Hash256 {
Hash256::from_low_u64_be(x)
}
fn check_iter_from(
block_tree: &BlockRootTree,
start_block_root: Hash256,
expected: &[(Hash256, Slot)],
) {
assert_eq!(
&block_tree.iter_from(start_block_root).collect::<Vec<_>>()[..],
expected
);
}
fn check_every_slot_iter_from(
block_tree: &BlockRootTree,
start_block_root: Hash256,
expected: &[(Hash256, Slot)],
) {
assert_eq!(
&block_tree
.every_slot_iter_from(start_block_root)
.collect::<Vec<_>>()[..],
expected
);
}
#[test]
fn single_chain() {
let block_tree = BlockRootTree::new(int_hash(1), Slot::new(1));
for i in 2..100 {
block_tree
.add_block_root(int_hash(i), int_hash(i - 1), Slot::new(i))
.expect("add_block_root ok");
let expected = (1..=i)
.rev()
.map(|j| (int_hash(j), Slot::new(j)))
.collect::<Vec<_>>();
check_iter_from(&block_tree, int_hash(i), &expected);
check_every_slot_iter_from(&block_tree, int_hash(i), &expected);
// Still OK after pruning.
block_tree.prune_to(int_hash(1), vec![int_hash(i)]);
check_iter_from(&block_tree, int_hash(i), &expected);
check_every_slot_iter_from(&block_tree, int_hash(i), &expected);
}
}
#[test]
fn skips_of_2() {
let block_tree = BlockRootTree::new(int_hash(1), Slot::new(1));
let step_length = 2u64;
for i in (1 + step_length..100).step_by(step_length as usize) {
block_tree
.add_block_root(int_hash(i), int_hash(i - step_length), Slot::new(i))
.expect("add_block_root ok");
let sparse_expected = (1..=i)
.rev()
.step_by(step_length as usize)
.map(|j| (int_hash(j), Slot::new(j)))
.collect_vec();
let every_slot_expected = (1..=i)
.rev()
.map(|j| {
let nearest = 1 + (j - 1) / step_length * step_length;
(int_hash(nearest), Slot::new(j))
})
.collect_vec();
check_iter_from(&block_tree, int_hash(i), &sparse_expected);
check_every_slot_iter_from(&block_tree, int_hash(i), &every_slot_expected);
// Still OK after pruning.
block_tree.prune_to(int_hash(1), vec![int_hash(i)]);
check_iter_from(&block_tree, int_hash(i), &sparse_expected);
check_every_slot_iter_from(&block_tree, int_hash(i), &every_slot_expected);
}
}
#[test]
fn prune_small_fork() {
let tree = BlockRootTree::new(int_hash(1), Slot::new(1));
// Space between fork hash values
let offset = 1000;
let num_blocks = 50;
let fork1_start = 2;
let fork2_start = 2 + offset;
tree.add_block_root(int_hash(fork1_start), int_hash(1), Slot::new(2))
.expect("add first block of left fork");
tree.add_block_root(int_hash(fork2_start), int_hash(1), Slot::new(2))
.expect("add first block of right fork");
for i in 3..num_blocks {
tree.add_block_root(int_hash(i), int_hash(i - 1), Slot::new(i))
.expect("add block to left fork");
tree.add_block_root(int_hash(i + offset), int_hash(i + offset - 1), Slot::new(i))
.expect("add block to right fork");
}
let root = (int_hash(1), Slot::new(1));
let (all_fork1_blocks, all_fork2_blocks): (Vec<_>, Vec<_>) = (2..num_blocks)
.rev()
.map(|i| {
(
(int_hash(i), Slot::new(i)),
(int_hash(i + offset), Slot::new(i)),
)
})
.chain(iter::once((root, root)))
.unzip();
let fork1_head = int_hash(num_blocks - 1);
let fork2_head = int_hash(num_blocks + offset - 1);
// Check that pruning with both heads preserves both chains.
let both_tree = tree.clone();
both_tree.prune_to(root.0, vec![fork1_head, fork2_head]);
check_iter_from(&both_tree, fork1_head, &all_fork1_blocks);
check_iter_from(&both_tree, fork2_head, &all_fork2_blocks);
// Check that pruning to either of the single chains leaves just that chain in the tree.
let fork1_tree = tree.clone();
fork1_tree.prune_to(root.0, vec![fork1_head]);
check_iter_from(&fork1_tree, fork1_head, &all_fork1_blocks);
check_iter_from(&fork1_tree, fork2_head, &[]);
let fork2_tree = tree.clone();
fork2_tree.prune_to(root.0, vec![fork2_head]);
check_iter_from(&fork2_tree, fork1_head, &[]);
check_iter_from(&fork2_tree, fork2_head, &all_fork2_blocks);
// Check that advancing the finalized root onto one side completely removes the other
// side.
let fin_tree = tree;
let prune_point = num_blocks / 2;
let remaining_fork1_blocks = all_fork1_blocks
.into_iter()
.take_while(|(_, slot)| *slot >= prune_point)
.collect_vec();
fin_tree.prune_to(int_hash(prune_point), vec![fork1_head, fork2_head]);
check_iter_from(&fin_tree, fork1_head, &remaining_fork1_blocks);
check_iter_from(&fin_tree, fork2_head, &[]);
}
#[test]
fn iter_zero() {
let block_tree = BlockRootTree::new(int_hash(0), Slot::new(0));
assert_eq!(block_tree.iter_from(int_hash(0)).count(), 0);
assert_eq!(block_tree.every_slot_iter_from(int_hash(0)).count(), 0);
}
}
beacon_node/store/src/lib.rs
-2
View File
@@ -11,7 +11,6 @@
extern crate lazy_static;
mod block_at_slot;
pub mod block_root_tree;
pub mod chunked_iter;
pub mod chunked_vector;
pub mod config;
@@ -29,7 +28,6 @@ pub mod migrate;
use std::sync::Arc;
pub use self::block_root_tree::{BlockRootTree, SszBlockRootTree};
pub use self::config::StoreConfig;
pub use self::hot_cold_store::HotColdDB as DiskStore;
pub use self::leveldb_store::LevelDB as SimpleDiskStore;