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

Merge remote-tracking branch 'origin/unstable' into tree-states

Browse Source
This commit is contained in:
Michael Sproul
2022-01-25 15:54:02 +11:00
parent fca92c37ad ca29b580a2
commit 6714edf95b
350 changed files with 16324 additions and 4500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
beacon_node/store/src/chunked_iter.rs
View File

@@ -1,27 +1,26 @@
use crate::chunked_vector::{chunk_key, Chunk, Field};
use crate::{HotColdDB, ItemStore};
use slog::error;
use std::sync::Arc;
use types::{ChainSpec, EthSpec, Slot};
/// Iterator over the values of a `BeaconState` vector field (like `block_roots`).
///
/// Uses the freezer DB's separate table to load the values.
pub struct ChunkedVectorIter<F, E, Hot, Cold>
pub struct ChunkedVectorIter<'a, F, E, Hot, Cold>
where
F: Field<E>,
E: EthSpec,
Hot: ItemStore<E>,
Cold: ItemStore<E>,
{
pub(crate) store: Arc<HotColdDB<E, Hot, Cold>>,
pub(crate) store: &'a HotColdDB<E, Hot, Cold>,
current_vindex: usize,
pub(crate) end_vindex: usize,
next_cindex: usize,
current_chunk: Chunk<F::Value>,
}
impl<F, E, Hot, Cold> ChunkedVectorIter<F, E, Hot, Cold>
impl<'a, F, E, Hot, Cold> ChunkedVectorIter<'a, F, E, Hot, Cold>
where
F: Field<E>,
E: EthSpec,
@@ -35,7 +34,7 @@ where
/// `HotColdDB::get_latest_restore_point_slot`. We pass it as a parameter so that the caller can
/// maintain a stable view of the database (see `HybridForwardsBlockRootsIterator`).
pub fn new(
store: Arc<HotColdDB<E, Hot, Cold>>,
store: &'a HotColdDB<E, Hot, Cold>,
start_vindex: usize,
last_restore_point_slot: Slot,
spec: &ChainSpec,
@@ -57,7 +56,7 @@ where
}
}
impl<F, E, Hot, Cold> Iterator for ChunkedVectorIter<F, E, Hot, Cold>
impl<'a, F, E, Hot, Cold> Iterator for ChunkedVectorIter<'a, F, E, Hot, Cold>
where
F: Field<E>,
E: EthSpec,

8
beacon_node/store/src/errors.rs
View File

@@ -2,6 +2,7 @@ use crate::chunked_vector::ChunkError;
use crate::config::StoreConfigError;
use crate::hot_cold_store::HotColdDBError;
use ssz::DecodeError;
use state_processing::BlockReplayError;
use types::{BeaconStateError, Hash256, Slot};
#[cfg(feature = "milhouse")]
@@ -42,6 +43,7 @@ pub enum Error {
expected: Hash256,
computed: Hash256,
},
BlockReplayError(BlockReplayError),
#[cfg(feature = "milhouse")]
MilhouseError(milhouse::Error),
}
@@ -103,6 +105,12 @@ impl From<milhouse::Error> for Error {
}
}
impl From<BlockReplayError> for Error {
fn from(e: BlockReplayError) -> Error {
Error::BlockReplayError(e)
}
}
#[derive(Debug)]
pub struct DBError {
pub message: String,

351
beacon_node/store/src/forwards_iter.rs
View File

@@ -1,74 +1,33 @@
use crate::chunked_iter::ChunkedVectorIter;
use crate::chunked_vector::{BlockRoots, StateRoots};
use crate::chunked_vector::{BlockRoots, Field, StateRoots};
use crate::errors::{Error, Result};
use crate::iter::{BlockRootsIterator, StateRootsIterator};
use crate::{HotColdDB, ItemStore};
use itertools::process_results;
use std::sync::Arc;
use types::{BeaconState, ChainSpec, EthSpec, Hash256, Slot};
/// Forwards block roots iterator that makes use of the `block_roots` table in the freezer DB.
pub struct FrozenForwardsBlockRootsIterator<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
inner: ChunkedVectorIter<BlockRoots, E, Hot, Cold>,
}
pub type HybridForwardsBlockRootsIterator<'a, E, Hot, Cold> =
HybridForwardsIterator<'a, E, BlockRoots, Hot, Cold>;
pub type HybridForwardsStateRootsIterator<'a, E, Hot, Cold> =
HybridForwardsIterator<'a, E, StateRoots, Hot, Cold>;
/// Forwards block roots iterator that reverses a backwards iterator (only good for short ranges).
pub struct SimpleForwardsBlockRootsIterator {
// Values from the backwards iterator (in slot descending order)
values: Vec<(Hash256, Slot)>,
}
/// Fusion of the above two approaches to forwards iteration. Fast and efficient.
pub enum HybridForwardsBlockRootsIterator<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
PreFinalization {
iter: Box<FrozenForwardsBlockRootsIterator<E, Hot, Cold>>,
/// Data required by the `PostFinalization` iterator when we get to it.
continuation_data: Box<Option<(BeaconState<E>, Hash256)>>,
},
PostFinalization {
iter: SimpleForwardsBlockRootsIterator,
},
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
FrozenForwardsBlockRootsIterator<E, Hot, Cold>
{
pub fn new(
store: Arc<HotColdDB<E, Hot, Cold>>,
/// Trait unifying `BlockRoots` and `StateRoots` for forward iteration.
pub trait Root<E: EthSpec>: Field<E, Value = Hash256> {
fn simple_forwards_iterator<Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
start_slot: Slot,
last_restore_point_slot: Slot,
spec: &ChainSpec,
) -> Self {
Self {
inner: ChunkedVectorIter::new(
store,
start_slot.as_usize(),
last_restore_point_slot,
spec,
),
}
}
end_state: BeaconState<E>,
end_root: Hash256,
) -> Result<SimpleForwardsIterator>;
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for FrozenForwardsBlockRootsIterator<E, Hot, Cold>
{
type Item = (Hash256, Slot);
fn next(&mut self) -> Option<Self::Item> {
self.inner
.next()
.map(|(slot, block_hash)| (block_hash, Slot::from(slot)))
}
}
impl SimpleForwardsBlockRootsIterator {
pub fn new<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: Arc<HotColdDB<E, Hot, Cold>>,
impl<E: EthSpec> Root<E> for BlockRoots {
fn simple_forwards_iterator<Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
start_slot: Slot,
end_state: BeaconState<E>,
end_block_root: Hash256,
) -> Result<Self> {
) -> Result<SimpleForwardsIterator> {
// Iterate backwards from the end state, stopping at the start slot.
let values = process_results(
std::iter::once(Ok((end_block_root, end_state.slot())))
@@ -78,129 +37,41 @@ impl SimpleForwardsBlockRootsIterator {
.collect::<Vec<_>>()
},
)?;
Ok(Self { values })
Ok(SimpleForwardsIterator { values })
}
}
impl Iterator for SimpleForwardsBlockRootsIterator {
type Item = Result<(Hash256, Slot)>;
fn next(&mut self) -> Option<Self::Item> {
// Pop from the end of the vector to get the block roots in slot-ascending order.
Ok(self.values.pop()).transpose()
}
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
HybridForwardsBlockRootsIterator<E, Hot, Cold>
{
pub fn new(
store: Arc<HotColdDB<E, Hot, Cold>>,
impl<E: EthSpec> Root<E> for StateRoots {
fn simple_forwards_iterator<Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
start_slot: Slot,
end_state: BeaconState<E>,
end_block_root: Hash256,
spec: &ChainSpec,
) -> Result<Self> {
use HybridForwardsBlockRootsIterator::*;
let latest_restore_point_slot = store.get_latest_restore_point_slot();
let result = if start_slot < latest_restore_point_slot {
PreFinalization {
iter: Box::new(FrozenForwardsBlockRootsIterator::new(
store,
start_slot,
latest_restore_point_slot,
spec,
)),
continuation_data: Box::new(Some((end_state, end_block_root))),
}
} else {
PostFinalization {
iter: SimpleForwardsBlockRootsIterator::new(
store,
start_slot,
end_state,
end_block_root,
)?,
}
};
Ok(result)
}
fn do_next(&mut self) -> Result<Option<(Hash256, Slot)>> {
use HybridForwardsBlockRootsIterator::*;
match self {
PreFinalization {
iter,
continuation_data,
} => {
match iter.next() {
Some(x) => Ok(Some(x)),
// Once the pre-finalization iterator is consumed, transition
// to a post-finalization iterator beginning from the last slot
// of the pre iterator.
None => {
let (end_state, end_block_root) =
continuation_data.take().ok_or(Error::NoContinuationData)?;
*self = PostFinalization {
iter: SimpleForwardsBlockRootsIterator::new(
iter.inner.store.clone(),
Slot::from(iter.inner.end_vindex),
end_state,
end_block_root,
)?,
};
self.do_next()
}
}
}
PostFinalization { iter } => iter.next().transpose(),
}
end_state_root: Hash256,
) -> Result<SimpleForwardsIterator> {
// Iterate backwards from the end state, stopping at the start slot.
let values = process_results(
std::iter::once(Ok((end_state_root, end_state.slot())))
.chain(StateRootsIterator::owned(store, end_state)),
|iter| {
iter.take_while(|(_, slot)| *slot >= start_slot)
.collect::<Vec<_>>()
},
)?;
Ok(SimpleForwardsIterator { values })
}
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for HybridForwardsBlockRootsIterator<E, Hot, Cold>
/// Forwards root iterator that makes use of a flat field table in the freezer DB.
pub struct FrozenForwardsIterator<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
{
type Item = Result<(Hash256, Slot)>;
fn next(&mut self) -> Option<Self::Item> {
self.do_next().transpose()
}
inner: ChunkedVectorIter<'a, F, E, Hot, Cold>,
}
/// Forwards state roots iterator that makes use of the `state_roots` table in the freezer DB.
pub struct FrozenForwardsStateRootsIterator<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
inner: ChunkedVectorIter<StateRoots, E, Hot, Cold>,
}
/// Forwards state roots iterator that reverses a backwards iterator (only good for short ranges).
pub struct SimpleForwardsStateRootsIterator {
// Values from the backwards iterator (in slot descending order)
values: Vec<(Hash256, Slot)>,
}
/// Fusion of the above two approaches to forwards iteration. Fast and efficient.
pub enum HybridForwardsStateRootsIterator<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
PreFinalization {
iter: Box<FrozenForwardsStateRootsIterator<E, Hot, Cold>>,
/// Data required by the `PostFinalization` iterator when we get to it.
continuation_data: Box<Option<(BeaconState<E>, Hash256)>>,
},
PostFinalization {
iter: SimpleForwardsStateRootsIterator,
},
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
FrozenForwardsStateRootsIterator<E, Hot, Cold>
impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
FrozenForwardsIterator<'a, E, F, Hot, Cold>
{
pub fn new(
store: Arc<HotColdDB<E, Hot, Cold>>,
store: &'a HotColdDB<E, Hot, Cold>,
start_slot: Slot,
last_restore_point_slot: Slot,
spec: &ChainSpec,
@@ -216,39 +87,25 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
}
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for FrozenForwardsStateRootsIterator<E, Hot, Cold>
impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for FrozenForwardsIterator<'a, E, F, Hot, Cold>
{
type Item = (Hash256, Slot);
fn next(&mut self) -> Option<Self::Item> {
self.inner
.next()
.map(|(slot, state_hash)| (state_hash, Slot::from(slot)))
.map(|(slot, root)| (root, Slot::from(slot)))
}
}
impl SimpleForwardsStateRootsIterator {
pub fn new<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: Arc<HotColdDB<E, Hot, Cold>>,
start_slot: Slot,
end_state: BeaconState<E>,
end_state_root: Hash256,
) -> Result<Self> {
// Iterate backwards from the end state, stopping at the start slot.
let values = process_results(
std::iter::once(Ok((end_state_root, end_state.slot())))
.chain(StateRootsIterator::owned(store, end_state)),
|iter| {
iter.take_while(|(_, slot)| *slot >= start_slot)
.collect::<Vec<_>>()
},
)?;
Ok(Self { values })
}
/// Forwards root iterator that reverses a backwards iterator (only good for short ranges).
pub struct SimpleForwardsIterator {
// Values from the backwards iterator (in slot descending order)
values: Vec<(Hash256, Slot)>,
}
impl Iterator for SimpleForwardsStateRootsIterator {
impl Iterator for SimpleForwardsIterator {
type Item = Result<(Hash256, Slot)>;
fn next(&mut self) -> Option<Self::Item> {
@@ -257,38 +114,75 @@ impl Iterator for SimpleForwardsStateRootsIterator {
}
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
HybridForwardsStateRootsIterator<E, Hot, Cold>
{
pub fn new(
store: Arc<HotColdDB<E, Hot, Cold>>,
/// Fusion of the above two approaches to forwards iteration. Fast and efficient.
pub enum HybridForwardsIterator<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>> {
PreFinalization {
iter: Box<FrozenForwardsIterator<'a, E, F, Hot, Cold>>,
/// Data required by the `PostFinalization` iterator when we get to it.
continuation_data: Option<Box<(BeaconState<E>, Hash256)>>,
},
PostFinalizationLazy {
continuation_data: Option<Box<(BeaconState<E>, Hash256)>>,
store: &'a HotColdDB<E, Hot, Cold>,
start_slot: Slot,
end_state: BeaconState<E>,
end_state_root: Hash256,
},
PostFinalization {
iter: SimpleForwardsIterator,
},
}
impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>>
HybridForwardsIterator<'a, E, F, Hot, Cold>
{
/// Construct a new hybrid iterator.
///
/// The `get_state` closure should return a beacon state and final block/state root to backtrack
/// from in the case where the iterated range does not lie entirely within the frozen portion of
/// the database. If an `end_slot` is provided and it is before the database's latest restore
/// point slot then the `get_state` closure will not be called at all.
///
/// It is OK for `get_state` to hold a lock while this function is evaluated, as the returned
/// iterator is as lazy as possible and won't do any work apart from calling `get_state`.
///
/// Conversely, if `get_state` does extensive work (e.g. loading data from disk) then this
/// function may block for some time while `get_state` runs.
pub fn new(
store: &'a HotColdDB<E, Hot, Cold>,
start_slot: Slot,
end_slot: Option<Slot>,
get_state: impl FnOnce() -> (BeaconState<E>, Hash256),
spec: &ChainSpec,
) -> Result<Self> {
use HybridForwardsStateRootsIterator::*;
use HybridForwardsIterator::*;
let latest_restore_point_slot = store.get_latest_restore_point_slot();
let result = if start_slot < latest_restore_point_slot {
let iter = Box::new(FrozenForwardsIterator::new(
store,
start_slot,
latest_restore_point_slot,
spec,
));
// No continuation data is needed if the forwards iterator plans to halt before
// `end_slot`. If it tries to continue further a `NoContinuationData` error will be
// returned.
let continuation_data =
if end_slot.map_or(false, |end_slot| end_slot < latest_restore_point_slot) {
None
} else {
Some(Box::new(get_state()))
};
PreFinalization {
iter: Box::new(FrozenForwardsStateRootsIterator::new(
store,
start_slot,
latest_restore_point_slot,
spec,
)),
continuation_data: Box::new(Some((end_state, end_state_root))),
iter,
continuation_data,
}
} else {
PostFinalization {
iter: SimpleForwardsStateRootsIterator::new(
store,
start_slot,
end_state,
end_state_root,
)?,
PostFinalizationLazy {
continuation_data: Some(Box::new(get_state())),
store,
start_slot,
}
};
@@ -296,7 +190,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
}
fn do_next(&mut self) -> Result<Option<(Hash256, Slot)>> {
use HybridForwardsStateRootsIterator::*;
use HybridForwardsIterator::*;
match self {
PreFinalization {
@@ -309,28 +203,39 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
// to a post-finalization iterator beginning from the last slot
// of the pre iterator.
None => {
let (end_state, end_state_root) =
continuation_data.take().ok_or(Error::NoContinuationData)?;
let continuation_data = continuation_data.take();
let store = iter.inner.store;
let start_slot = Slot::from(iter.inner.end_vindex);
*self = PostFinalization {
iter: SimpleForwardsStateRootsIterator::new(
iter.inner.store.clone(),
Slot::from(iter.inner.end_vindex),
end_state,
end_state_root,
)?,
*self = PostFinalizationLazy {
continuation_data,
store,
start_slot,
};
self.do_next()
}
}
}
PostFinalizationLazy {
continuation_data,
store,
start_slot,
} => {
let (end_state, end_root) =
*continuation_data.take().ok_or(Error::NoContinuationData)?;
*self = PostFinalization {
iter: F::simple_forwards_iterator(store, *start_slot, end_state, end_root)?,
};
self.do_next()
}
PostFinalization { iter } => iter.next().transpose(),
}
}
}
impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for HybridForwardsStateRootsIterator<E, Hot, Cold>
impl<'a, E: EthSpec, F: Root<E>, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for HybridForwardsIterator<'a, E, F, Hot, Cold>
{
type Item = Result<(Hash256, Slot)>;

214
beacon_node/store/src/hot_cold_store.rs
View File

@@ -22,12 +22,11 @@ use leveldb::iterator::LevelDBIterator;
use lru::LruCache;
use parking_lot::{Mutex, RwLock};
use serde_derive::{Deserialize, Serialize};
use slog::{debug, error, info, trace, Logger};
use slog::{debug, error, info, trace, warn, Logger};
use ssz::{Decode, Encode};
use ssz_derive::{Decode, Encode};
use state_processing::{
per_block_processing, per_slot_processing, BlockProcessingError, BlockSignatureStrategy,
SlotProcessingError,
BlockProcessingError, BlockReplayer, SlotProcessingError, StateRootStrategy,
};
use std::cmp::min;
use std::convert::TryInto;
@@ -37,16 +36,6 @@ use std::sync::Arc;
use std::time::Duration;
use types::*;
/// Defines how blocks should be replayed on states.
#[derive(PartialEq)]
pub enum BlockReplay {
/// Perform all transitions faithfully to the specification.
Accurate,
/// Don't compute state roots, eventually computing an invalid beacon state that can only be
/// used for obtaining shuffling.
InconsistentStateRoots,
}
/// On-disk database that stores finalized states efficiently.
///
/// Stores vector fields like the `block_roots` and `state_roots` separately, and only stores
@@ -373,10 +362,10 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
// chain. This way we avoid returning a state that doesn't match `state_root`.
self.load_cold_state(state_root)
} else {
self.load_hot_state(state_root, BlockReplay::Accurate)
self.load_hot_state(state_root, StateRootStrategy::Accurate)
}
} else {
match self.load_hot_state(state_root, BlockReplay::Accurate)? {
match self.load_hot_state(state_root, StateRootStrategy::Accurate)? {
Some(state) => Ok(Some(state)),
None => self.load_cold_state(state_root),
}
@@ -414,7 +403,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
}
.into())
} else {
self.load_hot_state(state_root, BlockReplay::InconsistentStateRoots)
self.load_hot_state(state_root, StateRootStrategy::Inconsistent)
}
}
@@ -439,23 +428,55 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
}
pub fn forwards_block_roots_iterator(
store: Arc<Self>,
&self,
start_slot: Slot,
end_state: BeaconState<E>,
end_block_root: Hash256,
spec: &ChainSpec,
) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
HybridForwardsBlockRootsIterator::new(store, start_slot, end_state, end_block_root, spec)
) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>> + '_, Error> {
HybridForwardsBlockRootsIterator::new(
self,
start_slot,
None,
|| (end_state, end_block_root),
spec,
)
}
pub fn forwards_block_roots_iterator_until(
&self,
start_slot: Slot,
end_slot: Slot,
get_state: impl FnOnce() -> (BeaconState<E>, Hash256),
spec: &ChainSpec,
) -> Result<HybridForwardsBlockRootsIterator<E, Hot, Cold>, Error> {
HybridForwardsBlockRootsIterator::new(self, start_slot, Some(end_slot), get_state, spec)
}
pub fn forwards_state_roots_iterator(
store: Arc<Self>,
&self,
start_slot: Slot,
end_state_root: Hash256,
end_state: BeaconState<E>,
spec: &ChainSpec,
) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>>, Error> {
HybridForwardsStateRootsIterator::new(store, start_slot, end_state, end_state_root, spec)
) -> Result<impl Iterator<Item = Result<(Hash256, Slot), Error>> + '_, Error> {
HybridForwardsStateRootsIterator::new(
self,
start_slot,
None,
|| (end_state, end_state_root),
spec,
)
}
pub fn forwards_state_roots_iterator_until(
&self,
start_slot: Slot,
end_slot: Slot,
get_state: impl FnOnce() -> (BeaconState<E>, Hash256),
spec: &ChainSpec,
) -> Result<HybridForwardsStateRootsIterator<E, Hot, Cold>, Error> {
HybridForwardsStateRootsIterator::new(self, start_slot, Some(end_slot), get_state, spec)
}
/// Load an epoch boundary state by using the hot state summary look-up.
@@ -472,10 +493,10 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
{
// NOTE: minor inefficiency here because we load an unnecessary hot state summary
//
// `BlockReplay` should be irrelevant here since we never replay blocks for an epoch
// `StateRootStrategy` should be irrelevant here since we never replay blocks for an epoch
// boundary state in the hot DB.
let state = self
.load_hot_state(&epoch_boundary_state_root, BlockReplay::Accurate)?
.load_hot_state(&epoch_boundary_state_root, StateRootStrategy::Accurate)?
.ok_or(HotColdDBError::MissingEpochBoundaryState(
epoch_boundary_state_root,
))?;
@@ -620,7 +641,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
pub fn load_hot_state(
&self,
state_root: &Hash256,
block_replay: BlockReplay,
state_root_strategy: StateRootStrategy,
) -> Result<Option<BeaconState<E>>, Error> {
metrics::inc_counter(&metrics::BEACON_STATE_HOT_GET_COUNT);
@@ -648,7 +669,13 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
} else {
let blocks =
self.load_blocks_to_replay(boundary_state.slot(), slot, latest_block_root)?;
self.replay_blocks(boundary_state, blocks, slot, block_replay)?
self.replay_blocks(
boundary_state,
blocks,
slot,
no_state_root_iter(),
state_root_strategy,
)?
};
Ok(Some(state))
@@ -777,7 +804,22 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
)?;
// 3. Replay the blocks on top of the low restore point.
self.replay_blocks(low_restore_point, blocks, slot, BlockReplay::Accurate)
// Use a forwards state root iterator to avoid doing any tree hashing.
// The state root of the high restore point should never be used, so is safely set to 0.
let state_root_iter = self.forwards_state_roots_iterator_until(
low_restore_point.slot(),
slot,
|| (high_restore_point, Hash256::zero()),
&self.spec,
)?;
self.replay_blocks(
low_restore_point,
blocks,
slot,
Some(state_root_iter),
StateRootStrategy::Accurate,
)
}
/// Get the restore point with the given index, or if it is out of bounds, the split state.
@@ -860,85 +902,35 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
/// to have any caches built, beyond those immediately required by block processing.
fn replay_blocks(
&self,
mut state: BeaconState<E>,
mut blocks: Vec<SignedBeaconBlock<E>>,
state: BeaconState<E>,
blocks: Vec<SignedBeaconBlock<E>>,
target_slot: Slot,
block_replay: BlockReplay,
state_root_iter: Option<impl Iterator<Item = Result<(Hash256, Slot), Error>>>,
state_root_strategy: StateRootStrategy,
) -> Result<BeaconState<E>, Error> {
if block_replay == BlockReplay::InconsistentStateRoots {
for i in 0..blocks.len() {
let prev_block_root = if i > 0 {
blocks[i - 1].canonical_root()
} else {
// Not read.
Hash256::zero()
};
let mut block_replayer = BlockReplayer::new(state, &self.spec)
.state_root_strategy(state_root_strategy)
.no_signature_verification()
.minimal_block_root_verification();
let (state_root, parent_root) = match &mut blocks[i] {
SignedBeaconBlock::Base(block) => (
&mut block.message.state_root,
&mut block.message.parent_root,
),
SignedBeaconBlock::Altair(block) => (
&mut block.message.state_root,
&mut block.message.parent_root,
),
};
let have_state_root_iterator = state_root_iter.is_some();
if let Some(state_root_iter) = state_root_iter {
block_replayer = block_replayer.state_root_iter(state_root_iter);
}
*state_root = Hash256::zero();
if i > 0 {
*parent_root = prev_block_root;
block_replayer
.apply_blocks(blocks, Some(target_slot))
.map(|block_replayer| {
if have_state_root_iterator && block_replayer.state_root_miss() {
warn!(
self.log,
"State root iterator miss";
"slot" => target_slot,
);
}
}
}
let state_root_from_prev_block = |i: usize, state: &BeaconState<E>| {
if i > 0 {
let prev_block = blocks[i - 1].message();
if prev_block.slot() == state.slot() {
Some(prev_block.state_root())
} else {
None
}
} else {
None
}
};
for (i, block) in blocks.iter().enumerate() {
if block.slot() <= state.slot() {
continue;
}
while state.slot() < block.slot() {
let state_root = match block_replay {
BlockReplay::Accurate => state_root_from_prev_block(i, &state),
BlockReplay::InconsistentStateRoots => Some(Hash256::zero()),
};
per_slot_processing(&mut state, state_root, &self.spec)
.map_err(HotColdDBError::BlockReplaySlotError)?;
}
per_block_processing(
&mut state,
block,
None,
BlockSignatureStrategy::NoVerification,
&self.spec,
)
.map_err(HotColdDBError::BlockReplayBlockError)?;
}
while state.slot() < target_slot {
let state_root = match block_replay {
BlockReplay::Accurate => state_root_from_prev_block(blocks.len(), &state),
BlockReplay::InconsistentStateRoots => Some(Hash256::zero()),
};
per_slot_processing(&mut state, state_root, &self.spec)
.map_err(HotColdDBError::BlockReplaySlotError)?;
}
Ok(state)
block_replayer.into_state()
})
}
/// Fetch a copy of the current split slot from memory.
@@ -971,6 +963,21 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
self.hot_db.put(&SCHEMA_VERSION_KEY, &schema_version)
}
/// Store the database schema version atomically with additional operations.
pub fn store_schema_version_atomically(
&self,
schema_version: SchemaVersion,
mut ops: Vec<KeyValueStoreOp>,
) -> Result<(), Error> {
let column = SchemaVersion::db_column().into();
let key = SCHEMA_VERSION_KEY.as_bytes();
let db_key = get_key_for_col(column, key);
let op = KeyValueStoreOp::PutKeyValue(db_key, schema_version.as_store_bytes());
ops.push(op);
self.hot_db.do_atomically(ops)
}
/// Initialise the anchor info for checkpoint sync starting from `block`.
pub fn init_anchor_info(&self, block: BeaconBlockRef<'_, E>) -> Result<KeyValueStoreOp, Error> {
let anchor_slot = block.slot();
@@ -1290,7 +1297,7 @@ pub fn migrate_database<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
// 1. Copy all of the states between the head and the split slot, from the hot DB
// to the cold DB.
let state_root_iter = StateRootsIterator::new(store.clone(), frozen_head);
let state_root_iter = StateRootsIterator::new(&store, frozen_head);
for maybe_pair in state_root_iter.take_while(|result| match result {
Ok((_, slot)) => {
slot >= &current_split_slot
@@ -1404,6 +1411,11 @@ impl StoreItem for Split {
}
}
/// Type hint.
fn no_state_root_iter() -> Option<std::iter::Empty<Result<(Hash256, Slot), Error>>> {
None
}
/// Struct for summarising a state in the hot database.
///
/// Allows full reconstruction by replaying blocks.

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

@@ -2,7 +2,6 @@ use crate::errors::HandleUnavailable;
use crate::{Error, HotColdDB, ItemStore};
use std::borrow::Cow;
use std::marker::PhantomData;
use std::sync::Arc;
use types::{
typenum::Unsigned, BeaconState, BeaconStateError, EthSpec, Hash256, SignedBeaconBlock, Slot,
};
@@ -13,19 +12,19 @@ use types::{
///
/// It is assumed that all ancestors for this object are stored in the database. If this is not the
/// case, the iterator will start returning `None` prior to genesis.
pub trait AncestorIter<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>, I: Iterator> {
pub trait AncestorIter<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>, I: Iterator> {
/// Returns an iterator over the roots of the ancestors of `self`.
fn try_iter_ancestor_roots(&self, store: Arc<HotColdDB<E, Hot, Cold>>) -> Option<I>;
fn try_iter_ancestor_roots(&self, store: &'a HotColdDB<E, Hot, Cold>) -> Option<I>;
}
impl<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
AncestorIter<E, Hot, Cold, BlockRootsIterator<'a, E, Hot, Cold>> for SignedBeaconBlock<E>
AncestorIter<'a, E, Hot, Cold, BlockRootsIterator<'a, E, Hot, Cold>> for SignedBeaconBlock<E>
{
/// Iterates across all available prior block roots of `self`, starting at the most recent and ending
/// at genesis.
fn try_iter_ancestor_roots(
&self,
store: Arc<HotColdDB<E, Hot, Cold>>,
store: &'a HotColdDB<E, Hot, Cold>,
) -> Option<BlockRootsIterator<'a, E, Hot, Cold>> {
let state = store
.get_state(&self.message().state_root(), Some(self.slot()))
@@ -36,13 +35,13 @@ impl<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
}
impl<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
AncestorIter<E, Hot, Cold, StateRootsIterator<'a, E, Hot, Cold>> for BeaconState<E>
AncestorIter<'a, E, Hot, Cold, StateRootsIterator<'a, E, Hot, Cold>> for BeaconState<E>
{
/// Iterates across all available prior state roots of `self`, starting at the most recent and ending
/// at genesis.
fn try_iter_ancestor_roots(
&self,
store: Arc<HotColdDB<E, Hot, Cold>>,
store: &'a HotColdDB<E, Hot, Cold>,
) -> Option<StateRootsIterator<'a, E, Hot, Cold>> {
// The `self.clone()` here is wasteful.
Some(StateRootsIterator::owned(store, self.clone()))
@@ -64,13 +63,13 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> Clone
}
impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> StateRootsIterator<'a, T, Hot, Cold> {
pub fn new(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: &'a BeaconState<T>) -> Self {
pub fn new(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: &'a BeaconState<T>) -> Self {
Self {
inner: RootsIterator::new(store, beacon_state),
}
}
pub fn owned(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: BeaconState<T>) -> Self {
pub fn owned(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: BeaconState<T>) -> Self {
Self {
inner: RootsIterator::owned(store, beacon_state),
}
@@ -113,18 +112,27 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> Clone
impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> BlockRootsIterator<'a, T, Hot, Cold> {
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn new(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: &'a BeaconState<T>) -> Self {
pub fn new(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: &'a BeaconState<T>) -> Self {
Self {
inner: RootsIterator::new(store, beacon_state),
}
}
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn owned(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: BeaconState<T>) -> Self {
pub fn owned(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: BeaconState<T>) -> Self {
Self {
inner: RootsIterator::owned(store, beacon_state),
}
}
pub fn from_block(
store: &'a HotColdDB<T, Hot, Cold>,
block_hash: Hash256,
) -> Result<Self, Error> {
Ok(Self {
inner: RootsIterator::from_block(store, block_hash)?,
})
}
}
impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> Iterator
@@ -141,7 +149,7 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> Iterator
/// Iterator over state and block roots that backtracks using the vectors from a `BeaconState`.
pub struct RootsIterator<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> {
store: Arc<HotColdDB<T, Hot, Cold>>,
store: &'a HotColdDB<T, Hot, Cold>,
beacon_state: Cow<'a, BeaconState<T>>,
slot: Slot,
}
@@ -151,7 +159,7 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> Clone
{
fn clone(&self) -> Self {
Self {
store: self.store.clone(),
store: self.store,
beacon_state: self.beacon_state.clone(),
slot: self.slot,
}
@@ -159,7 +167,7 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> Clone
}
impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> RootsIterator<'a, T, Hot, Cold> {
pub fn new(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: &'a BeaconState<T>) -> Self {
pub fn new(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: &'a BeaconState<T>) -> Self {
Self {
store,
slot: beacon_state.slot(),
@@ -167,7 +175,7 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> RootsIterator<'a, T,
}
}
pub fn owned(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: BeaconState<T>) -> Self {
pub fn owned(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: BeaconState<T>) -> Self {
Self {
store,
slot: beacon_state.slot(),
@@ -176,7 +184,7 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> RootsIterator<'a, T,
}
pub fn from_block(
store: Arc<HotColdDB<T, Hot, Cold>>,
store: &'a HotColdDB<T, Hot, Cold>,
block_hash: Hash256,
) -> Result<Self, Error> {
let block = store
@@ -301,14 +309,14 @@ pub struct BlockIterator<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>>
impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> BlockIterator<'a, T, Hot, Cold> {
/// Create a new iterator over all blocks in the given `beacon_state` and prior states.
pub fn new(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: &'a BeaconState<T>) -> Self {
pub fn new(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: &'a BeaconState<T>) -> Self {
Self {
roots: BlockRootsIterator::new(store, beacon_state),
}
}
/// Create a new iterator over all blocks in the given `beacon_state` and prior states.
pub fn owned(store: Arc<HotColdDB<T, Hot, Cold>>, beacon_state: BeaconState<T>) -> Self {
pub fn owned(store: &'a HotColdDB<T, Hot, Cold>, beacon_state: BeaconState<T>) -> Self {
Self {
roots: BlockRootsIterator::owned(store, beacon_state),
}
@@ -388,9 +396,8 @@ mod test {
#[test]
fn block_root_iter() {
let log = NullLoggerBuilder.build().unwrap();
let store = Arc::new(
HotColdDB::open_ephemeral(Config::default(), ChainSpec::minimal(), log).unwrap(),
);
let store =
HotColdDB::open_ephemeral(Config::default(), ChainSpec::minimal(), log).unwrap();
let slots_per_historical_root = MainnetEthSpec::slots_per_historical_root();
let mut state_a: BeaconState<MainnetEthSpec> = get_state();
@@ -413,7 +420,7 @@ mod test {
state_b.state_roots_mut()[0] = state_a_root;
store.put_state(&state_a_root, &state_a).unwrap();
let iter = BlockRootsIterator::new(store, &state_b);
let iter = BlockRootsIterator::new(&store, &state_b);
assert!(
iter.clone()
@@ -436,9 +443,8 @@ mod test {
#[test]
fn state_root_iter() {
let log = NullLoggerBuilder.build().unwrap();
let store = Arc::new(
HotColdDB::open_ephemeral(Config::default(), ChainSpec::minimal(), log).unwrap(),
);
let store =
HotColdDB::open_ephemeral(Config::default(), ChainSpec::minimal(), log).unwrap();
let slots_per_historical_root = MainnetEthSpec::slots_per_historical_root();
let mut state_a: BeaconState<MainnetEthSpec> = get_state();
@@ -466,7 +472,7 @@ mod test {
store.put_state(&state_a_root, &state_a).unwrap();
store.put_state(&state_b_root, &state_b).unwrap();
let iter = StateRootsIterator::new(store, &state_b);
let iter = StateRootsIterator::new(&store, &state_b);
assert!(
iter.clone()

2
beacon_node/store/src/lib.rs
View File

@@ -30,7 +30,7 @@ pub mod iter;
pub use self::chunk_writer::ChunkWriter;
pub use self::config::StoreConfig;
pub use self::hot_cold_store::{BlockReplay, HotColdDB, HotStateSummary, Split};
pub use self::hot_cold_store::{HotColdDB, HotStateSummary, Split};
pub use self::leveldb_store::LevelDB;
pub use self::memory_store::MemoryStore;
pub use self::partial_beacon_state::PartialBeaconState;

2
beacon_node/store/src/metadata.rs
View File

@@ -4,7 +4,7 @@ use ssz::{Decode, Encode};
use ssz_derive::{Decode, Encode};
use types::{Checkpoint, Hash256, Slot};
pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(5);
pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(8);
// All the keys that get stored under the `BeaconMeta` column.
//

45
beacon_node/store/src/partial_beacon_state.rs
View File

@@ -14,8 +14,8 @@ use types::*;
///
/// Utilises lazy-loading from separate storage for its vector fields.
#[superstruct(
variants(Base, Altair),
variant_attributes(derive(Debug, PartialEq, Clone, Encode, Decode),)
variants(Base, Altair, Merge),
variant_attributes(derive(Debug, PartialEq, Clone, Encode, Decode))
)]
#[derive(Debug, PartialEq, Clone, Encode)]
#[ssz(enum_behaviour = "transparent")]
@@ -66,9 +66,9 @@ where
pub current_epoch_attestations: VList<PendingAttestation<T>, T::MaxPendingAttestations>,
// Participation (Altair and later)
#[superstruct(only(Altair))]
#[superstruct(only(Altair, Merge))]
pub previous_epoch_participation: VariableList<ParticipationFlags, T::ValidatorRegistryLimit>,
#[superstruct(only(Altair))]
#[superstruct(only(Altair, Merge))]
pub current_epoch_participation: VariableList<ParticipationFlags, T::ValidatorRegistryLimit>,
// Finality
@@ -78,14 +78,18 @@ where
pub finalized_checkpoint: Checkpoint,
// Inactivity
#[superstruct(only(Altair))]
#[superstruct(only(Altair, Merge))]
pub inactivity_scores: VariableList<u64, T::ValidatorRegistryLimit>,
// Light-client sync committees
#[superstruct(only(Altair))]
#[superstruct(only(Altair, Merge))]
pub current_sync_committee: Arc<SyncCommittee<T>>,
#[superstruct(only(Altair))]
#[superstruct(only(Altair, Merge))]
pub next_sync_committee: Arc<SyncCommittee<T>>,
// Execution
#[superstruct(only(Merge))]
pub latest_execution_payload_header: ExecutionPayloadHeader<T>,
}
/// Implement the conversion function from BeaconState -> PartialBeaconState.
@@ -160,6 +164,20 @@ impl<T: EthSpec> PartialBeaconState<T> {
inactivity_scores
]
),
BeaconState::Merge(s) => impl_from_state_forgetful!(
s,
outer,
Merge,
PartialBeaconStateMerge,
[
previous_epoch_participation,
current_epoch_participation,
current_sync_committee,
next_sync_committee,
inactivity_scores,
latest_execution_payload_header
]
),
}
}
@@ -335,6 +353,19 @@ impl<E: EthSpec> TryInto<BeaconState<E>> for PartialBeaconState<E> {
inactivity_scores
]
),
PartialBeaconState::Merge(inner) => impl_try_into_beacon_state!(
inner,
Merge,
BeaconStateMerge,
[
previous_epoch_participation,
current_epoch_participation,
current_sync_committee,
next_sync_committee,
inactivity_scores,
latest_execution_payload_header
]
),
};
Ok(state)
}

8
beacon_node/store/src/reconstruct.rs
View File

@@ -3,7 +3,9 @@ use crate::hot_cold_store::{HotColdDB, HotColdDBError};
use crate::{Error, ItemStore, KeyValueStore};
use itertools::{process_results, Itertools};
use slog::info;
use state_processing::{per_block_processing, per_slot_processing, BlockSignatureStrategy};
use state_processing::{
per_block_processing, per_slot_processing, BlockSignatureStrategy, VerifyBlockRoot,
};
use std::sync::Arc;
use types::{EthSpec, Hash256};
@@ -48,8 +50,7 @@ where
// Use a dummy root, as we never read the block for the upper limit state.
let upper_limit_block_root = Hash256::repeat_byte(0xff);
let block_root_iter = Self::forwards_block_roots_iterator(
self.clone(),
let block_root_iter = self.forwards_block_roots_iterator(
lower_limit_slot,
upper_limit_state,
upper_limit_block_root,
@@ -91,6 +92,7 @@ where
&block,
Some(block_root),
BlockSignatureStrategy::NoVerification,
VerifyBlockRoot::True,
&self.spec,
)
.map_err(HotColdDBError::BlockReplayBlockError)?;