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 'michael/state-root-summary' into tree-states

Browse Source
This commit is contained in:
Michael Sproul
2022-02-15 12:05:54 +11:00
parent 5ed951d84c 7245161fc2
commit 5ff4868280
11 changed files with 160 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -43,6 +43,7 @@ pub enum Error {
expected: Hash256,
computed: Hash256,
},
MissingStateRoot(Slot),
BlockReplayError(BlockReplayError),
#[cfg(feature = "milhouse")]
MilhouseError(milhouse::Error),

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

@@ -3,6 +3,7 @@ use crate::chunked_vector::{
};
use crate::config::{OnDiskStoreConfig, StoreConfig};
use crate::forwards_iter::{HybridForwardsBlockRootsIterator, HybridForwardsStateRootsIterator};
use crate::hot_state_iter::ForwardsHotStateRootIter;
use crate::impls::beacon_state::{get_full_state, store_full_state};
use crate::iter::{ParentRootBlockIterator, StateRootsIterator};
use crate::leveldb_store::BytesKey;
@@ -22,13 +23,12 @@ use crate::{
use leveldb::iterator::LevelDBIterator;
use lru::LruCache;
use parking_lot::{Mutex, RwLock};
use safe_arith::SafeArith;
use serde_derive::{Deserialize, Serialize};
use slog::{debug, error, info, trace, warn, Logger};
use slog::{debug, error, info, trace, Logger};
use ssz::{Decode, Encode};
use ssz_derive::{Decode, Encode};
use state_processing::{
BlockProcessingError, BlockReplayer, SlotProcessingError, StateRootStrategy,
};
use state_processing::{BlockProcessingError, BlockReplayer, SlotProcessingError};
use std::cmp::min;
use std::convert::TryInto;
use std::marker::PhantomData;
@@ -383,10 +383,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.get_hot_state(state_root, StateRootStrategy::Accurate)
self.get_hot_state(state_root)
}
} else {
match self.get_hot_state(state_root, StateRootStrategy::Accurate)? {
match self.get_hot_state(state_root)? {
Some(state) => Ok(Some(state)),
None => self.load_cold_state(state_root),
}
@@ -425,6 +425,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
///
/// - `state.state_roots`
/// - `state.block_roots`
// FIXME(sproul): delete this whole function
pub fn get_inconsistent_state_for_attestation_verification_only(
&self,
state_root: &Hash256,
@@ -442,7 +443,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
}
.into())
} else {
self.get_hot_state(state_root, StateRootStrategy::Inconsistent)
self.get_hot_state(state_root)
}
}
@@ -534,11 +535,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
//
// `StateRootStrategy` should be irrelevant here since we never replay blocks for an epoch
// boundary state in the hot DB.
let state = self
.get_hot_state(&epoch_boundary_state_root, StateRootStrategy::Accurate)?
.ok_or(HotColdDBError::MissingEpochBoundaryState(
epoch_boundary_state_root,
))?;
let state = self.get_hot_state(&epoch_boundary_state_root)?.ok_or(
HotColdDBError::MissingEpochBoundaryState(epoch_boundary_state_root),
)?;
Ok(Some(state))
} else {
// Try the cold DB
@@ -716,8 +715,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
pub fn load_hot_state(
&self,
state_root: &Hash256,
state_root_strategy: StateRootStrategy,
) -> Result<Option<(BeaconState<E>, Hash256)>, Error> {
let _timer = metrics::start_timer(&metrics::BEACON_HOT_STATE_READ_TIMES);
metrics::inc_counter(&metrics::BEACON_STATE_HOT_GET_COUNT);
// If the state is marked as temporary, do not return it. It will become visible
@@ -730,6 +729,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
slot,
latest_block_root,
epoch_boundary_state_root,
prev_state_root,
}) = self.load_hot_state_summary(state_root)?
{
let boundary_state =
@@ -744,13 +744,15 @@ 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,
let state_root_iter = ForwardsHotStateRootIter::new(
self,
boundary_state.slot(),
slot,
no_state_root_iter(),
state_root_strategy,
)?
*state_root,
prev_state_root,
)?;
self.replay_blocks(boundary_state, blocks, slot, state_root_iter)?
};
Ok(Some((state, latest_block_root)))
@@ -888,13 +890,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
&self.spec,
)?;
self.replay_blocks(
low_restore_point,
blocks,
slot,
Some(state_root_iter),
StateRootStrategy::Accurate,
)
self.replay_blocks(low_restore_point, blocks, slot, state_root_iter)
}
/// Get the restore point with the given index, or if it is out of bounds, the split state.
@@ -980,31 +976,19 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
state: BeaconState<E>,
blocks: Vec<SignedBeaconBlock<E>>,
target_slot: Slot,
state_root_iter: Option<impl Iterator<Item = Result<(Hash256, Slot), Error>>>,
state_root_strategy: StateRootStrategy,
state_root_iter: impl Iterator<Item = Result<(Hash256, Slot), Error>>,
) -> Result<BeaconState<E>, Error> {
let mut block_replayer = BlockReplayer::new(state, &self.spec)
.state_root_strategy(state_root_strategy)
BlockReplayer::new(state, &self.spec)
.no_signature_verification()
.minimal_block_root_verification();
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);
}
block_replayer
.minimal_block_root_verification()
.state_root_iter(state_root_iter)
.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,
);
.and_then(|block_replayer| {
if block_replayer.state_root_miss() {
Err(Error::MissingStateRoot(target_slot))
} else {
Ok(block_replayer.into_state())
}
block_replayer.into_state()
})
}
@@ -1496,19 +1480,17 @@ 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.
#[derive(Debug, Clone, Copy, Default, Encode, Decode)]
pub struct HotStateSummary {
slot: Slot,
latest_block_root: Hash256,
epoch_boundary_state_root: Hash256,
pub(crate) slot: Slot,
pub(crate) latest_block_root: Hash256,
pub(crate) epoch_boundary_state_root: Hash256,
/// The state root of the state at the prior slot.
// FIXME(sproul): migrate
pub(crate) prev_state_root: Hash256,
}
impl StoreItem for HotStateSummary {
@@ -1530,20 +1512,29 @@ impl HotStateSummary {
pub fn new<E: EthSpec>(state_root: &Hash256, state: &BeaconState<E>) -> Result<Self, Error> {
// Fill in the state root on the latest block header if necessary (this happens on all
// slots where there isn't a skip).
let slot = state.slot();
let latest_block_root = state.get_latest_block_root(*state_root);
let epoch_boundary_slot = state.slot() / E::slots_per_epoch() * E::slots_per_epoch();
let epoch_boundary_state_root = if epoch_boundary_slot == state.slot() {
let epoch_boundary_slot = slot / E::slots_per_epoch() * E::slots_per_epoch();
let epoch_boundary_state_root = if epoch_boundary_slot == slot {
*state_root
} else {
*state
.get_state_root(epoch_boundary_slot)
.map_err(HotColdDBError::HotStateSummaryError)?
};
let prev_state_root = if let Ok(prev_slot) = slot.safe_sub(1) {
*state
.get_state_root(prev_slot)
.map_err(HotColdDBError::HotStateSummaryError)?
} else {
Hash256::zero()
};
Ok(HotStateSummary {
slot: state.slot(),
latest_block_root,
epoch_boundary_state_root,
prev_state_root,
})
}
}

90
beacon_node/store/src/hot_state_iter.rs Normal file
View File

@@ -0,0 +1,90 @@
use crate::{hot_cold_store::HotColdDBError, Error, HotColdDB, ItemStore};
use itertools::process_results;
use std::iter;
use take_until::TakeUntilExt;
use types::{EthSpec, Hash256, Slot};
pub struct HotStateRootIter<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> {
store: &'a HotColdDB<E, Hot, Cold>,
next_slot: Slot,
next_state_root: Hash256,
}
impl<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotStateRootIter<'a, E, Hot, Cold> {
pub fn new(
store: &'a HotColdDB<E, Hot, Cold>,
next_slot: Slot,
next_state_root: Hash256,
) -> Self {
Self {
store,
next_slot,
next_state_root,
}
}
fn do_next(&mut self) -> Result<Option<(Hash256, Slot)>, Error> {
if self.next_state_root.is_zero() {
return Ok(None);
}
let summary = self
.store
.load_hot_state_summary(&self.next_state_root)?
.ok_or_else(|| HotColdDBError::MissingHotStateSummary(self.next_state_root))?;
let slot = self.next_slot;
let state_root = self.next_state_root;
self.next_state_root = summary.prev_state_root;
self.next_slot -= 1;
Ok(Some((state_root, slot)))
}
}
impl<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> Iterator
for HotStateRootIter<'a, E, Hot, Cold>
{
type Item = Result<(Hash256, Slot), Error>;
fn next(&mut self) -> Option<Self::Item> {
self.do_next().transpose()
}
}
pub struct ForwardsHotStateRootIter {
// Values from the backwards iterator (in slot descending order)
values: Vec<(Hash256, Slot)>,
}
impl ForwardsHotStateRootIter {
pub fn new<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>(
store: &HotColdDB<E, Hot, Cold>,
start_slot: Slot,
end_slot: Slot,
last_state_root: Hash256,
second_last_state_root: Hash256,
) -> Result<Self, Error> {
process_results(
iter::once(Ok((last_state_root, end_slot))).chain(HotStateRootIter::new(
store,
end_slot - 1,
second_last_state_root,
)),
|iter| {
let values = iter.take_until(|(_, slot)| *slot == start_slot).collect();
Self { values }
},
)
}
}
impl Iterator for ForwardsHotStateRootIter {
type Item = Result<(Hash256, Slot), Error>;
fn next(&mut self) -> Option<Self::Item> {
// Pop from the end of the vector to get the state roots in slot-ascending order.
Ok(self.values.pop()).transpose()
}
}

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

@@ -18,6 +18,7 @@ pub mod errors;
mod forwards_iter;
mod garbage_collection;
pub mod hot_cold_store;
mod hot_state_iter;
mod impls;
mod leveldb_store;
mod memory_store;

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(8);
pub const CURRENT_SCHEMA_VERSION: SchemaVersion = SchemaVersion(9000);
// All the keys that get stored under the `BeaconMeta` column.
//

14
beacon_node/store/src/metrics.rs
View File

@@ -54,17 +54,13 @@ lazy_static! {
"store_beacon_state_hot_get_total",
"Total number of hot beacon states requested from the store (cache or DB)"
);
pub static ref BEACON_STATE_CACHE_HIT_COUNT: Result<IntCounter> = try_create_int_counter(
"store_beacon_state_cache_hit_total",
"Number of hits to the store's state cache"
);
pub static ref BEACON_STATE_CACHE_CLONE_TIME: Result<Histogram> = try_create_histogram(
"store_beacon_state_cache_clone_time",
"Time to load a beacon block from the block cache"
);
pub static ref BEACON_STATE_READ_TIMES: Result<Histogram> = try_create_histogram(
"store_beacon_state_read_seconds",
"Total time required to read a BeaconState from the database"
"Total time required to read a full BeaconState from the database"
);
pub static ref BEACON_HOT_STATE_READ_TIMES: Result<Histogram> = try_create_histogram(
"store_beacon_hot_state_read_seconds",
"Total time required to read a hot BeaconState from the database"
);
pub static ref BEACON_STATE_READ_OVERHEAD_TIMES: Result<Histogram> = try_create_histogram(
"store_beacon_state_read_overhead_seconds",