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

Clean up database abstractions (#1200)

Browse Source 
* Remove redundant method

* Pull out a method out of a struct

* More precise db access abstractions

* Move fake trait method out of it

* cargo fmt

* Fix compilation error after refactoring

* Move another fake method out the Store trait

* Get rid of superfluous method

* Fix refactoring bug

* Rename: SimpleStoreItem -> StoreItem

* Get rid of the confusing DiskStore type alias

* Get rid of SimpleDiskStore type alias

* Correction: A method took both self and a ref to Self
This commit is contained in:
Adam Szkoda
2020-06-01 00:13:49 +02:00
committed by GitHub
parent 08e6b4961d
commit 91cb14ac41
31 changed files with 413 additions and 485 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
use crate::chunked_vector::{chunk_key, Chunk, Field};
use crate::DiskStore;
use crate::HotColdDB;
use slog::error;
use std::sync::Arc;
use types::{ChainSpec, EthSpec, Slot};
@@ -12,7 +12,7 @@ where
F: Field<E>,
E: EthSpec,
{
pub(crate) store: Arc<DiskStore<E>>,
pub(crate) store: Arc<HotColdDB<E>>,
current_vindex: usize,
pub(crate) end_vindex: usize,
next_cindex: usize,
@@ -28,10 +28,10 @@ where
/// index stored by the restore point at `last_restore_point_slot`.
///
/// The `last_restore_point` slot should be the slot of a recent restore point as obtained from
/// `DiskStore::get_latest_restore_point_slot`. We pass it as a parameter so that the caller can
/// `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<DiskStore<E>>,
store: Arc<HotColdDB<E>>,
start_vindex: usize,
last_restore_point_slot: Slot,
spec: &ChainSpec,

18
beacon_node/store/src/chunked_vector.rs
View File

@@ -177,7 +177,7 @@ pub trait Field<E: EthSpec>: Copy {
/// Load the genesis value for a fixed length field from the store.
///
/// This genesis value should be used to fill the initial state of the vector.
fn load_genesis_value<S: Store<E>>(store: &S) -> Result<Self::Value, Error> {
fn load_genesis_value<S: KeyValueStore<E>>(store: &S) -> Result<Self::Value, Error> {
let key = &genesis_value_key()[..];
let chunk =
Chunk::load(store, Self::column(), key)?.ok_or(ChunkError::MissingGenesisValue)?;
@@ -192,7 +192,7 @@ pub trait Field<E: EthSpec>: Copy {
///
/// Check the existing value (if any) for consistency with the value we intend to store, and
/// return an error if they are inconsistent.
fn check_and_store_genesis_value<S: Store<E>>(
fn check_and_store_genesis_value<S: KeyValueStore<E>>(
store: &S,
value: Self::Value,
) -> Result<(), Error> {
@@ -327,7 +327,7 @@ field!(
|state: &BeaconState<_>, index, _| safe_modulo_index(&state.randao_mixes, index)
);
pub fn store_updated_vector<F: Field<E>, E: EthSpec, S: Store<E>>(
pub fn store_updated_vector<F: Field<E>, E: EthSpec, S: KeyValueStore<E>>(
field: F,
store: &S,
state: &BeaconState<E>,
@@ -387,7 +387,7 @@ fn store_range<F, E, S, I>(
where
F: Field<E>,
E: EthSpec,
S: Store<E>,
S: KeyValueStore<E>,
I: Iterator<Item = usize>,
{
for chunk_index in range {
@@ -417,7 +417,7 @@ where
// Chunks at the end index are included.
// TODO: could be more efficient with a real range query (perhaps RocksDB)
fn range_query<S: Store<E>, E: EthSpec, T: Decode + Encode>(
fn range_query<S: KeyValueStore<E>, E: EthSpec, T: Decode + Encode>(
store: &S,
column: DBColumn,
start_index: usize,
@@ -482,7 +482,7 @@ fn stitch<T: Default + Clone>(
Ok(result)
}
pub fn load_vector_from_db<F: FixedLengthField<E>, E: EthSpec, S: Store<E>>(
pub fn load_vector_from_db<F: FixedLengthField<E>, E: EthSpec, S: KeyValueStore<E>>(
store: &S,
slot: Slot,
spec: &ChainSpec,
@@ -514,7 +514,7 @@ pub fn load_vector_from_db<F: FixedLengthField<E>, E: EthSpec, S: Store<E>>(
}
/// The historical roots are stored in vector chunks, despite not actually being a vector.
pub fn load_variable_list_from_db<F: VariableLengthField<E>, E: EthSpec, S: Store<E>>(
pub fn load_variable_list_from_db<F: VariableLengthField<E>, E: EthSpec, S: KeyValueStore<E>>(
store: &S,
slot: Slot,
spec: &ChainSpec,
@@ -574,7 +574,7 @@ where
Chunk { values }
}
pub fn load<S: Store<E>, E: EthSpec>(
pub fn load<S: KeyValueStore<E>, E: EthSpec>(
store: &S,
column: DBColumn,
key: &[u8],
@@ -585,7 +585,7 @@ where
.transpose()
}
pub fn store<S: Store<E>, E: EthSpec>(
pub fn store<S: KeyValueStore<E>, E: EthSpec>(
&self,
store: &S,
column: DBColumn,

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

@@ -1,7 +1,7 @@
use crate::chunked_iter::ChunkedVectorIter;
use crate::chunked_vector::BlockRoots;
use crate::iter::{BlockRootsIterator, ReverseBlockRootIterator};
use crate::{DiskStore, Store};
use crate::{HotColdDB, Store};
use slog::error;
use std::sync::Arc;
use types::{BeaconState, ChainSpec, EthSpec, Hash256, Slot};
@@ -31,7 +31,7 @@ pub enum HybridForwardsBlockRootsIterator<E: EthSpec> {
impl<E: EthSpec> FrozenForwardsBlockRootsIterator<E> {
pub fn new(
store: Arc<DiskStore<E>>,
store: Arc<HotColdDB<E>>,
start_slot: Slot,
last_restore_point_slot: Slot,
spec: &ChainSpec,
@@ -87,7 +87,7 @@ impl Iterator for SimpleForwardsBlockRootsIterator {
impl<E: EthSpec> HybridForwardsBlockRootsIterator<E> {
pub fn new(
store: Arc<DiskStore<E>>,
store: Arc<HotColdDB<E>>,
start_slot: Slot,
end_state: BeaconState<E>,
end_block_root: Hash256,

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

@@ -3,11 +3,12 @@ use crate::chunked_vector::{
};
use crate::config::StoreConfig;
use crate::forwards_iter::HybridForwardsBlockRootsIterator;
use crate::impls::beacon_state::store_full_state;
use crate::impls::beacon_state::{get_full_state, store_full_state};
use crate::iter::{ParentRootBlockIterator, StateRootsIterator};
use crate::metrics;
use crate::{
leveldb_store::LevelDB, DBColumn, Error, PartialBeaconState, SimpleStoreItem, Store, StoreOp,
leveldb_store::LevelDB, DBColumn, Error, ItemStore, KeyValueStore, PartialBeaconState, Store,
StoreItem, StoreOp,
};
use lru::LruCache;
use parking_lot::{Mutex, RwLock};
@@ -86,27 +87,10 @@ pub enum HotColdDBError {
impl<E: EthSpec> Store<E> for HotColdDB<E> {
type ForwardsBlockRootsIterator = HybridForwardsBlockRootsIterator<E>;
// Defer to the hot database for basic operations (including blocks for now)
fn get_bytes(&self, column: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
self.hot_db.get_bytes(column, key)
}
fn put_bytes(&self, column: &str, key: &[u8], value: &[u8]) -> Result<(), Error> {
self.hot_db.put_bytes(column, key, value)
}
fn key_exists(&self, column: &str, key: &[u8]) -> Result<bool, Error> {
self.hot_db.key_exists(column, key)
}
fn key_delete(&self, column: &str, key: &[u8]) -> Result<(), Error> {
self.hot_db.key_delete(column, key)
}
/// Store a block and update the LRU cache.
fn put_block(&self, block_root: &Hash256, block: SignedBeaconBlock<E>) -> Result<(), Error> {
// Store on disk.
self.put(block_root, &block)?;
self.hot_db.put(block_root, &block)?;
// Update cache.
self.block_cache.lock().put(*block_root, block);
@@ -125,7 +109,7 @@ impl<E: EthSpec> Store<E> for HotColdDB<E> {
}
// Fetch from database.
match self.get::<SignedBeaconBlock<E>>(block_root)? {
match self.hot_db.get::<SignedBeaconBlock<E>>(block_root)? {
Some(block) => {
// Add to cache.
self.block_cache.lock().put(*block_root, block.clone());
@@ -138,7 +122,15 @@ impl<E: EthSpec> Store<E> for HotColdDB<E> {
/// Delete a block from the store and the block cache.
fn delete_block(&self, block_root: &Hash256) -> Result<(), Error> {
self.block_cache.lock().pop(block_root);
self.delete::<SignedBeaconBlock<E>>(block_root)
self.hot_db.delete::<SignedBeaconBlock<E>>(block_root)
}
fn put_state_summary(
&self,
state_root: &Hash256,
summary: HotStateSummary,
) -> Result<(), Error> {
self.hot_db.put(state_root, &summary).map_err(Into::into)
}
/// Store a state in the store.
@@ -155,17 +147,6 @@ impl<E: EthSpec> Store<E> for HotColdDB<E> {
&self,
state_root: &Hash256,
slot: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
self.get_state_with(state_root, slot)
}
/// Get a state from the store.
///
/// Fetch a state from the store, controlling which cache fields are cloned.
fn get_state_with(
&self,
state_root: &Hash256,
slot: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
metrics::inc_counter(&metrics::BEACON_STATE_GET_COUNT);
@@ -203,96 +184,6 @@ impl<E: EthSpec> Store<E> for HotColdDB<E> {
Ok(())
}
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error> {
let mut guard = self.block_cache.lock();
self.hot_db.do_atomically(batch)?;
for op in batch {
match op {
StoreOp::DeleteBlock(block_hash) => {
let untyped_hash: Hash256 = (*block_hash).into();
guard.pop(&untyped_hash);
}
StoreOp::DeleteState(_, _) => (),
}
}
Ok(())
}
/// Advance the split point of the store, moving new finalized states to the freezer.
fn process_finalization(
store: Arc<Self>,
frozen_head_root: Hash256,
frozen_head: &BeaconState<E>,
) -> Result<(), Error> {
debug!(
store.log,
"Freezer migration started";
"slot" => frozen_head.slot
);
// 0. Check that the migration is sensible.
// The new frozen head must increase the current split slot, and lie on an epoch
// boundary (in order for the hot state summary scheme to work).
let current_split_slot = store.get_split_slot();
if frozen_head.slot < current_split_slot {
return Err(HotColdDBError::FreezeSlotError {
current_split_slot,
proposed_split_slot: frozen_head.slot,
}
.into());
}
if frozen_head.slot % E::slots_per_epoch() != 0 {
return Err(HotColdDBError::FreezeSlotUnaligned(frozen_head.slot).into());
}
// 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 mut to_delete = vec![];
for (state_root, slot) in
state_root_iter.take_while(|&(_, slot)| slot >= current_split_slot)
{
if slot % store.config.slots_per_restore_point == 0 {
let state: BeaconState<E> = store
.hot_db
.get_state(&state_root, None)?
.ok_or_else(|| HotColdDBError::MissingStateToFreeze(state_root))?;
store.store_cold_state(&state_root, &state)?;
}
// Store a pointer from this state root to its slot, so we can later reconstruct states
// from their state root alone.
store.store_cold_state_slot(&state_root, slot)?;
// Delete the old summary, and the full state if we lie on an epoch boundary.
to_delete.push((state_root, slot));
}
// 2. Update the split slot
*store.split.write() = Split {
slot: frozen_head.slot,
state_root: frozen_head_root,
};
store.store_split()?;
// 3. Delete from the hot DB
for (state_root, slot) in to_delete {
store.delete_state(&state_root, slot)?;
}
debug!(
store.log,
"Freezer migration complete";
"slot" => frozen_head.slot
);
Ok(())
}
fn forwards_block_roots_iterator(
store: Arc<Self>,
start_slot: Slot,
@@ -334,6 +225,33 @@ impl<E: EthSpec> Store<E> for HotColdDB<E> {
}
}
}
fn put_item<I: StoreItem>(&self, key: &Hash256, item: &I) -> Result<(), Error> {
self.hot_db.put(key, item)
}
fn get_item<I: StoreItem>(&self, key: &Hash256) -> Result<Option<I>, Error> {
self.hot_db.get(key)
}
fn item_exists<I: StoreItem>(&self, key: &Hash256) -> Result<bool, Error> {
self.hot_db.exists::<I>(key)
}
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error> {
let mut guard = self.block_cache.lock();
self.hot_db.do_atomically(batch)?;
for op in batch {
match op {
StoreOp::DeleteBlock(block_hash) => {
let untyped_hash: Hash256 = (*block_hash).into();
guard.pop(&untyped_hash);
}
StoreOp::DeleteState(_, _) => (),
}
}
Ok(())
}
}
impl<E: EthSpec> HotColdDB<E> {
@@ -408,9 +326,7 @@ impl<E: EthSpec> HotColdDB<E> {
epoch_boundary_state_root,
}) = self.load_hot_state_summary(state_root)?
{
let boundary_state = self
.hot_db
.get_state(&epoch_boundary_state_root, None)?
let boundary_state = get_full_state(&self.hot_db, &epoch_boundary_state_root)?
.ok_or_else(|| {
HotColdDBError::MissingEpochBoundaryState(epoch_boundary_state_root)
})?;
@@ -758,6 +674,77 @@ impl<E: EthSpec> HotColdDB<E> {
}
}
/// Advance the split point of the store, moving new finalized states to the freezer.
pub fn process_finalization<E: EthSpec>(
store: Arc<HotColdDB<E>>,
frozen_head_root: Hash256,
frozen_head: &BeaconState<E>,
) -> Result<(), Error> {
debug!(
store.log,
"Freezer migration started";
"slot" => frozen_head.slot
);
// 0. Check that the migration is sensible.
// The new frozen head must increase the current split slot, and lie on an epoch
// boundary (in order for the hot state summary scheme to work).
let current_split_slot = store.get_split_slot();
if frozen_head.slot < current_split_slot {
return Err(HotColdDBError::FreezeSlotError {
current_split_slot,
proposed_split_slot: frozen_head.slot,
}
.into());
}
if frozen_head.slot % E::slots_per_epoch() != 0 {
return Err(HotColdDBError::FreezeSlotUnaligned(frozen_head.slot).into());
}
// 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 mut to_delete = vec![];
for (state_root, slot) in state_root_iter.take_while(|&(_, slot)| slot >= current_split_slot) {
if slot % store.config.slots_per_restore_point == 0 {
let state: BeaconState<E> = get_full_state(&store.hot_db, &state_root)?
.ok_or_else(|| HotColdDBError::MissingStateToFreeze(state_root))?;
store.store_cold_state(&state_root, &state)?;
}
// Store a pointer from this state root to its slot, so we can later reconstruct states
// from their state root alone.
store.store_cold_state_slot(&state_root, slot)?;
// Delete the old summary, and the full state if we lie on an epoch boundary.
to_delete.push((state_root, slot));
}
// 2. Update the split slot
*store.split.write() = Split {
slot: frozen_head.slot,
state_root: frozen_head_root,
};
store.store_split()?;
// 3. Delete from the hot DB
for (state_root, slot) in to_delete {
store.delete_state(&state_root, slot)?;
}
debug!(
store.log,
"Freezer migration complete";
"slot" => frozen_head.slot
);
Ok(())
}
/// Struct for storing the split slot and state root in the database.
#[derive(Debug, Clone, Copy, Default, Encode, Decode)]
struct Split {
@@ -765,7 +752,7 @@ struct Split {
state_root: Hash256,
}
impl SimpleStoreItem for Split {
impl StoreItem for Split {
fn db_column() -> DBColumn {
DBColumn::BeaconMeta
}
@@ -789,7 +776,7 @@ pub struct HotStateSummary {
epoch_boundary_state_root: Hash256,
}
impl SimpleStoreItem for HotStateSummary {
impl StoreItem for HotStateSummary {
fn db_column() -> DBColumn {
DBColumn::BeaconStateSummary
}
@@ -832,7 +819,7 @@ struct ColdStateSummary {
slot: Slot,
}
impl SimpleStoreItem for ColdStateSummary {
impl StoreItem for ColdStateSummary {
fn db_column() -> DBColumn {
DBColumn::BeaconStateSummary
}
@@ -852,7 +839,7 @@ struct RestorePointHash {
state_root: Hash256,
}
impl SimpleStoreItem for RestorePointHash {
impl StoreItem for RestorePointHash {
fn db_column() -> DBColumn {
DBColumn::BeaconRestorePoint
}

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

@@ -4,7 +4,7 @@ use ssz::{Decode, Encode};
pub mod beacon_state;
pub mod partial_beacon_state;
impl<T: EthSpec> SimpleStoreItem for SignedBeaconBlock<T> {
impl<T: EthSpec> StoreItem for SignedBeaconBlock<T> {
fn db_column() -> DBColumn {
DBColumn::BeaconBlock
}

10
beacon_node/store/src/impls/beacon_state.rs
View File

@@ -4,8 +4,8 @@ use ssz_derive::{Decode, Encode};
use std::convert::TryInto;
use types::beacon_state::{CloneConfig, CommitteeCache, CACHED_EPOCHS};
pub fn store_full_state<S: Store<E>, E: EthSpec>(
store: &S,
pub fn store_full_state<KV: KeyValueStore<E>, E: EthSpec>(
store: &KV,
state_root: &Hash256,
state: &BeaconState<E>,
) -> Result<(), Error> {
@@ -24,13 +24,13 @@ pub fn store_full_state<S: Store<E>, E: EthSpec>(
result
}
pub fn get_full_state<S: Store<E>, E: EthSpec>(
store: &S,
pub fn get_full_state<KV: KeyValueStore<E>, E: EthSpec>(
db: &KV,
state_root: &Hash256,
) -> Result<Option<BeaconState<E>>, Error> {
let total_timer = metrics::start_timer(&metrics::BEACON_STATE_READ_TIMES);
match store.get_bytes(DBColumn::BeaconState.into(), state_root.as_bytes())? {
match db.get_bytes(DBColumn::BeaconState.into(), state_root.as_bytes())? {
Some(bytes) => {
let overhead_timer = metrics::start_timer(&metrics::BEACON_STATE_READ_OVERHEAD_TIMES);
let container = StorageContainer::from_ssz_bytes(&bytes)?;

2
beacon_node/store/src/impls/partial_beacon_state.rs
View File

@@ -1,7 +1,7 @@
use crate::*;
use ssz::{Decode, Encode};
impl<T: EthSpec> SimpleStoreItem for PartialBeaconState<T> {
impl<T: EthSpec> StoreItem for PartialBeaconState<T> {
fn db_column() -> DBColumn {
DBColumn::BeaconState
}

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

@@ -42,11 +42,11 @@ impl<'a, U: Store<E>, E: EthSpec> AncestorIter<U, E, StateRootsIterator<'a, E, U
}
}
pub struct StateRootsIterator<'a, T: EthSpec, U> {
pub struct StateRootsIterator<'a, T: EthSpec, U: Store<T>> {
inner: RootsIterator<'a, T, U>,
}
impl<'a, T: EthSpec, U> Clone for StateRootsIterator<'a, T, U> {
impl<'a, T: EthSpec, U: Store<T>> Clone for StateRootsIterator<'a, T, U> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
@@ -86,11 +86,11 @@ impl<'a, T: EthSpec, U: Store<T>> Iterator for StateRootsIterator<'a, T, U> {
/// exhausted.
///
/// Returns `None` for roots prior to genesis or when there is an error reading from `Store`.
pub struct BlockRootsIterator<'a, T: EthSpec, U> {
pub struct BlockRootsIterator<'a, T: EthSpec, U: Store<T>> {
inner: RootsIterator<'a, T, U>,
}
impl<'a, T: EthSpec, U> Clone for BlockRootsIterator<'a, T, U> {
impl<'a, T: EthSpec, U: Store<T>> Clone for BlockRootsIterator<'a, T, U> {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
@@ -125,13 +125,13 @@ impl<'a, T: EthSpec, U: Store<T>> Iterator for BlockRootsIterator<'a, T, U> {
}
/// Iterator over state and block roots that backtracks using the vectors from a `BeaconState`.
pub struct RootsIterator<'a, T: EthSpec, U> {
pub struct RootsIterator<'a, T: EthSpec, U: Store<T>> {
store: Arc<U>,
beacon_state: Cow<'a, BeaconState<T>>,
slot: Slot,
}
impl<'a, T: EthSpec, U> Clone for RootsIterator<'a, T, U> {
impl<'a, T: EthSpec, U: Store<T>> Clone for RootsIterator<'a, T, U> {
fn clone(&self) -> Self {
Self {
store: self.store.clone(),
@@ -245,7 +245,7 @@ impl<'a, E: EthSpec, S: Store<E>> Iterator for ParentRootBlockIterator<'a, E, S>
#[derive(Clone)]
/// Extends `BlockRootsIterator`, returning `SignedBeaconBlock` instances, instead of their roots.
pub struct BlockIterator<'a, T: EthSpec, U> {
pub struct BlockIterator<'a, T: EthSpec, U: Store<T>> {
roots: BlockRootsIterator<'a, T, U>,
}

96
beacon_node/store/src/leveldb_store.rs
View File

@@ -1,6 +1,4 @@
use super::*;
use crate::forwards_iter::SimpleForwardsBlockRootsIterator;
use crate::impls::beacon_state::{get_full_state, store_full_state};
use crate::metrics;
use db_key::Key;
use leveldb::database::batch::{Batch, Writebatch};
@@ -39,35 +37,12 @@ impl<E: EthSpec> LevelDB<E> {
fn write_options(&self) -> WriteOptions {
WriteOptions::new()
}
fn get_key_for_col(col: &str, key: &[u8]) -> BytesKey {
let mut col = col.as_bytes().to_vec();
col.append(&mut key.to_vec());
BytesKey { key: col }
}
}
/// Used for keying leveldb.
pub struct BytesKey {
key: Vec<u8>,
}
impl Key for BytesKey {
fn from_u8(key: &[u8]) -> Self {
Self { key: key.to_vec() }
}
fn as_slice<T, F: Fn(&[u8]) -> T>(&self, f: F) -> T {
f(self.key.as_slice())
}
}
impl<E: EthSpec> Store<E> for LevelDB<E> {
type ForwardsBlockRootsIterator = SimpleForwardsBlockRootsIterator;
impl<E: EthSpec> KeyValueStore<E> for LevelDB<E> {
/// Retrieve some bytes in `column` with `key`.
fn get_bytes(&self, col: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
let column_key = Self::get_key_for_col(col, key);
let column_key = get_key_for_col(col, key);
metrics::inc_counter(&metrics::DISK_DB_READ_COUNT);
let timer = metrics::start_timer(&metrics::DISK_DB_READ_TIMES);
@@ -86,7 +61,7 @@ impl<E: EthSpec> Store<E> for LevelDB<E> {
/// Store some `value` in `column`, indexed with `key`.
fn put_bytes(&self, col: &str, key: &[u8], val: &[u8]) -> Result<(), Error> {
let column_key = Self::get_key_for_col(col, key);
let column_key = get_key_for_col(col, key);
metrics::inc_counter(&metrics::DISK_DB_WRITE_COUNT);
metrics::inc_counter_by(&metrics::DISK_DB_WRITE_BYTES, val.len() as i64);
@@ -102,7 +77,7 @@ impl<E: EthSpec> Store<E> for LevelDB<E> {
/// Return `true` if `key` exists in `column`.
fn key_exists(&self, col: &str, key: &[u8]) -> Result<bool, Error> {
let column_key = Self::get_key_for_col(col, key);
let column_key = get_key_for_col(col, key);
metrics::inc_counter(&metrics::DISK_DB_EXISTS_COUNT);
@@ -114,7 +89,7 @@ impl<E: EthSpec> Store<E> for LevelDB<E> {
/// Removes `key` from `column`.
fn key_delete(&self, col: &str, key: &[u8]) -> Result<(), Error> {
let column_key = Self::get_key_for_col(col, key);
let column_key = get_key_for_col(col, key);
metrics::inc_counter(&metrics::DISK_DB_DELETE_COUNT);
@@ -123,56 +98,28 @@ impl<E: EthSpec> Store<E> for LevelDB<E> {
.map_err(Into::into)
}
/// Store a state in the store.
fn put_state(&self, state_root: &Hash256, state: &BeaconState<E>) -> Result<(), Error> {
store_full_state(self, state_root, &state)
}
/// Fetch a state from the store.
fn get_state(
&self,
state_root: &Hash256,
_: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
get_full_state(self, state_root)
}
fn forwards_block_roots_iterator(
store: Arc<Self>,
start_slot: Slot,
end_state: BeaconState<E>,
end_block_root: Hash256,
_: &ChainSpec,
) -> Self::ForwardsBlockRootsIterator {
SimpleForwardsBlockRootsIterator::new(store, start_slot, end_state, end_block_root)
}
fn do_atomically(&self, ops_batch: &[StoreOp]) -> Result<(), Error> {
let mut leveldb_batch = Writebatch::new();
for op in ops_batch {
match op {
StoreOp::DeleteBlock(block_hash) => {
let untyped_hash: Hash256 = (*block_hash).into();
let key = Self::get_key_for_col(
DBColumn::BeaconBlock.into(),
untyped_hash.as_bytes(),
);
let key =
get_key_for_col(DBColumn::BeaconBlock.into(), untyped_hash.as_bytes());
leveldb_batch.delete(key);
}
StoreOp::DeleteState(state_hash, slot) => {
let untyped_hash: Hash256 = (*state_hash).into();
let state_summary_key = Self::get_key_for_col(
let state_summary_key = get_key_for_col(
DBColumn::BeaconStateSummary.into(),
untyped_hash.as_bytes(),
);
leveldb_batch.delete(state_summary_key);
if *slot % E::slots_per_epoch() == 0 {
let state_key = Self::get_key_for_col(
DBColumn::BeaconState.into(),
untyped_hash.as_bytes(),
);
let state_key =
get_key_for_col(DBColumn::BeaconState.into(), untyped_hash.as_bytes());
leveldb_batch.delete(state_key);
}
}
@@ -183,6 +130,29 @@ impl<E: EthSpec> Store<E> for LevelDB<E> {
}
}
impl<E: EthSpec> ItemStore<E> for LevelDB<E> {}
/// Used for keying leveldb.
pub struct BytesKey {
key: Vec<u8>,
}
impl Key for BytesKey {
fn from_u8(key: &[u8]) -> Self {
Self { key: key.to_vec() }
}
fn as_slice<T, F: Fn(&[u8]) -> T>(&self, f: F) -> T {
f(self.key.as_slice())
}
}
fn get_key_for_col(col: &str, key: &[u8]) -> BytesKey {
let mut col = col.as_bytes().to_vec();
col.append(&mut key.to_vec());
BytesKey { key: col }
}
impl From<LevelDBError> for Error {
fn from(e: LevelDBError) -> Error {
Error::DBError {

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

@@ -2,7 +2,7 @@
//!
//! Provides the following stores:
//!
//! - `DiskStore`: an on-disk store backed by leveldb. Used in production.
//! - `HotColdDB`: an on-disk store backed by leveldb. Used in production.
//! - `MemoryStore`: an in-memory store backed by a hash-map. Used for testing.
//!
//! Provides a simple API for storing/retrieving all types that sometimes needs type-hints. See
@@ -28,8 +28,8 @@ pub mod iter;
use std::sync::Arc;
pub use self::config::StoreConfig;
pub use self::hot_cold_store::{HotColdDB as DiskStore, HotStateSummary};
pub use self::leveldb_store::LevelDB as SimpleDiskStore;
pub use self::hot_cold_store::{HotColdDB, HotStateSummary};
pub use self::leveldb_store::LevelDB;
pub use self::memory_store::MemoryStore;
pub use self::partial_beacon_state::PartialBeaconState;
pub use errors::Error;
@@ -38,14 +38,7 @@ pub use metrics::scrape_for_metrics;
pub use state_batch::StateBatch;
pub use types::*;
/// An object capable of storing and retrieving objects implementing `StoreItem`.
///
/// A `Store` is fundamentally backed by a key-value database, however it provides support for
/// columns. A simple column implementation might involve prefixing a key with some bytes unique to
/// each column.
pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
type ForwardsBlockRootsIterator: Iterator<Item = (Hash256, Slot)>;
pub trait KeyValueStore<E: EthSpec>: Sync + Send + Sized + 'static {
/// Retrieve some bytes in `column` with `key`.
fn get_bytes(&self, column: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error>;
@@ -58,8 +51,13 @@ pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
/// Removes `key` from `column`.
fn key_delete(&self, column: &str, key: &[u8]) -> Result<(), Error>;
/// Execute either all of the operations in `batch` or none at all, returning an error.
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error>;
}
pub trait ItemStore<E: EthSpec>: KeyValueStore<E> + Sync + Send + Sized + 'static {
/// Store an item in `Self`.
fn put<I: SimpleStoreItem>(&self, key: &Hash256, item: &I) -> Result<(), Error> {
fn put<I: StoreItem>(&self, key: &Hash256, item: &I) -> Result<(), Error> {
let column = I::db_column().into();
let key = key.as_bytes();
@@ -68,7 +66,7 @@ pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
}
/// Retrieve an item from `Self`.
fn get<I: SimpleStoreItem>(&self, key: &Hash256) -> Result<Option<I>, Error> {
fn get<I: StoreItem>(&self, key: &Hash256) -> Result<Option<I>, Error> {
let column = I::db_column().into();
let key = key.as_bytes();
@@ -79,7 +77,7 @@ pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
}
/// Returns `true` if the given key represents an item in `Self`.
fn exists<I: SimpleStoreItem>(&self, key: &Hash256) -> Result<bool, Error> {
fn exists<I: StoreItem>(&self, key: &Hash256) -> Result<bool, Error> {
let column = I::db_column().into();
let key = key.as_bytes();
@@ -87,44 +85,40 @@ pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
}
/// Remove an item from `Self`.
fn delete<I: SimpleStoreItem>(&self, key: &Hash256) -> Result<(), Error> {
fn delete<I: StoreItem>(&self, key: &Hash256) -> Result<(), Error> {
let column = I::db_column().into();
let key = key.as_bytes();
self.key_delete(column, key)
}
}
/// An object capable of storing and retrieving objects implementing `StoreItem`.
///
/// A `Store` is fundamentally backed by a key-value database, however it provides support for
/// columns. A simple column implementation might involve prefixing a key with some bytes unique to
/// each column.
pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
type ForwardsBlockRootsIterator: Iterator<Item = (Hash256, Slot)>;
/// Store a block in the store.
fn put_block(&self, block_root: &Hash256, block: SignedBeaconBlock<E>) -> Result<(), Error> {
self.put(block_root, &block)
}
fn put_block(&self, block_root: &Hash256, block: SignedBeaconBlock<E>) -> Result<(), Error>;
/// Fetch a block from the store.
fn get_block(&self, block_root: &Hash256) -> Result<Option<SignedBeaconBlock<E>>, Error> {
self.get(block_root)
}
fn get_block(&self, block_root: &Hash256) -> Result<Option<SignedBeaconBlock<E>>, Error>;
/// Delete a block from the store.
fn delete_block(&self, block_root: &Hash256) -> Result<(), Error> {
self.key_delete(DBColumn::BeaconBlock.into(), block_root.as_bytes())
}
fn delete_block(&self, block_root: &Hash256) -> Result<(), Error>;
/// Store a state in the store.
fn put_state(&self, state_root: &Hash256, state: &BeaconState<E>) -> Result<(), Error>;
/// Execute either all of the operations in `batch` or none at all, returning an error.
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error>;
/// Store a state summary in the store.
// NOTE: this is a hack for the HotColdDb, we could consider splitting this
// trait and removing the generic `S: Store` types everywhere?
fn put_state_summary(
&self,
state_root: &Hash256,
summary: HotStateSummary,
) -> Result<(), Error> {
self.put(state_root, &summary).map_err(Into::into)
}
) -> Result<(), Error>;
/// Fetch a state from the store.
fn get_state(
@@ -133,33 +127,12 @@ pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
slot: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error>;
/// Fetch a state from the store, controlling which cache fields are cloned.
fn get_state_with(
&self,
state_root: &Hash256,
slot: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
// Default impl ignores config. Overriden in `HotColdDb`.
self.get_state(state_root, slot)
}
/// Delete a state from the store.
fn delete_state(&self, state_root: &Hash256, _slot: Slot) -> Result<(), Error> {
self.key_delete(DBColumn::BeaconState.into(), state_root.as_bytes())
}
/// (Optionally) Move all data before the frozen slot to the freezer database.
fn process_finalization(
_store: Arc<Self>,
_frozen_head_root: Hash256,
_frozen_head: &BeaconState<E>,
) -> Result<(), Error> {
Ok(())
}
fn delete_state(&self, state_root: &Hash256, _slot: Slot) -> Result<(), Error>;
/// Get a forwards (slot-ascending) iterator over the beacon block roots since `start_slot`.
///
/// Will be efficient for frozen portions of the database if using `DiskStore`.
/// Will be efficient for frozen portions of the database if using `HotColdDB`.
///
/// The `end_state` and `end_block_root` are required for backtracking in the post-finalization
/// part of the chain, and should be usually be set to the current head. Importantly, the
@@ -175,28 +148,18 @@ pub trait Store<E: EthSpec>: Sync + Send + Sized + 'static {
spec: &ChainSpec,
) -> Self::ForwardsBlockRootsIterator;
/// Load the most recent ancestor state of `state_root` which lies on an epoch boundary.
///
/// If `state_root` corresponds to an epoch boundary state, then that state itself should be
/// returned.
fn load_epoch_boundary_state(
&self,
state_root: &Hash256,
) -> Result<Option<BeaconState<E>>, Error> {
// The default implementation is not very efficient, but isn't used in prod.
// See `HotColdDB` for the optimized implementation.
if let Some(state) = self.get_state(state_root, None)? {
let epoch_boundary_slot = state.slot / E::slots_per_epoch() * E::slots_per_epoch();
if state.slot == epoch_boundary_slot {
Ok(Some(state))
} else {
let epoch_boundary_state_root = state.get_state_root(epoch_boundary_slot)?;
self.get_state(epoch_boundary_state_root, Some(epoch_boundary_slot))
}
} else {
Ok(None)
}
}
) -> Result<Option<BeaconState<E>>, Error>;
fn put_item<I: StoreItem>(&self, key: &Hash256, item: &I) -> Result<(), Error>;
fn get_item<I: StoreItem>(&self, key: &Hash256) -> Result<Option<I>, Error>;
fn item_exists<I: StoreItem>(&self, key: &Hash256) -> Result<bool, Error>;
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error>;
}
/// Reified key-value storage operation. Helps in modifying the storage atomically.
@@ -252,7 +215,7 @@ impl Into<&'static str> for DBColumn {
}
/// An item that may stored in a `Store` by serializing and deserializing from bytes.
pub trait SimpleStoreItem: Sized {
pub trait StoreItem: Sized {
/// Identifies which column this item should be placed in.
fn db_column() -> DBColumn;
@@ -278,7 +241,7 @@ mod tests {
b: u64,
}
impl SimpleStoreItem for StorableThing {
impl StoreItem for StorableThing {
fn db_column() -> DBColumn {
DBColumn::BeaconBlock
}
@@ -292,7 +255,7 @@ mod tests {
}
}
fn test_impl(store: impl Store<MinimalEthSpec>) {
fn test_impl(store: impl ItemStore<MinimalEthSpec>) {
let key = Hash256::random();
let item = StorableThing { a: 1, b: 42 };
@@ -312,31 +275,11 @@ mod tests {
assert_eq!(store.get::<StorableThing>(&key).unwrap(), None);
}
#[test]
fn diskdb() {
use sloggers::{null::NullLoggerBuilder, Build};
let hot_dir = tempdir().unwrap();
let cold_dir = tempdir().unwrap();
let spec = MinimalEthSpec::default_spec();
let log = NullLoggerBuilder.build().unwrap();
let store = DiskStore::open(
&hot_dir.path(),
&cold_dir.path(),
StoreConfig::default(),
spec,
log,
)
.unwrap();
test_impl(store);
}
#[test]
fn simplediskdb() {
let dir = tempdir().unwrap();
let path = dir.path();
let store = SimpleDiskStore::open(&path).unwrap();
let store = LevelDB::open(&path).unwrap();
test_impl(store);
}

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

@@ -1,6 +1,8 @@
use super::{DBColumn, Error, Store, StoreOp};
use super::{DBColumn, Error, ItemStore, KeyValueStore, Store, StoreOp};
use crate::forwards_iter::SimpleForwardsBlockRootsIterator;
use crate::hot_cold_store::HotStateSummary;
use crate::impls::beacon_state::{get_full_state, store_full_state};
use crate::StoreItem;
use parking_lot::RwLock;
use std::collections::HashMap;
use std::marker::PhantomData;
@@ -40,9 +42,7 @@ impl<E: EthSpec> MemoryStore<E> {
}
}
impl<E: EthSpec> Store<E> for MemoryStore<E> {
type ForwardsBlockRootsIterator = SimpleForwardsBlockRootsIterator;
impl<E: EthSpec> KeyValueStore<E> for MemoryStore<E> {
/// Get the value of some key from the database. Returns `None` if the key does not exist.
fn get_bytes(&self, col: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
let column_key = Self::get_key_for_col(col, key);
@@ -75,20 +75,6 @@ impl<E: EthSpec> Store<E> for MemoryStore<E> {
Ok(())
}
/// Store a state in the store.
fn put_state(&self, state_root: &Hash256, state: &BeaconState<E>) -> Result<(), Error> {
store_full_state(self, state_root, &state)
}
/// Fetch a state from the store.
fn get_state(
&self,
state_root: &Hash256,
_: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
get_full_state(self, state_root)
}
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error> {
for op in batch {
match op {
@@ -112,6 +98,50 @@ impl<E: EthSpec> Store<E> for MemoryStore<E> {
}
Ok(())
}
}
impl<E: EthSpec> ItemStore<E> for MemoryStore<E> {}
impl<E: EthSpec> Store<E> for MemoryStore<E> {
type ForwardsBlockRootsIterator = SimpleForwardsBlockRootsIterator;
fn put_block(&self, block_root: &Hash256, block: SignedBeaconBlock<E>) -> Result<(), Error> {
self.put(block_root, &block)
}
fn get_block(&self, block_root: &Hash256) -> Result<Option<SignedBeaconBlock<E>>, Error> {
self.get(block_root)
}
fn delete_block(&self, block_root: &Hash256) -> Result<(), Error> {
self.key_delete(DBColumn::BeaconBlock.into(), block_root.as_bytes())
}
fn put_state_summary(
&self,
state_root: &Hash256,
summary: HotStateSummary,
) -> Result<(), Error> {
self.put(state_root, &summary).map_err(Into::into)
}
/// Store a state in the store.
fn put_state(&self, state_root: &Hash256, state: &BeaconState<E>) -> Result<(), Error> {
store_full_state(self, state_root, &state)
}
/// Fetch a state from the store.
fn get_state(
&self,
state_root: &Hash256,
_: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
get_full_state(self, state_root)
}
fn delete_state(&self, state_root: &Hash256, _slot: Slot) -> Result<(), Error> {
self.key_delete(DBColumn::BeaconState.into(), state_root.as_bytes())
}
fn forwards_block_roots_iterator(
store: Arc<Self>,
@@ -122,4 +152,43 @@ impl<E: EthSpec> Store<E> for MemoryStore<E> {
) -> Self::ForwardsBlockRootsIterator {
SimpleForwardsBlockRootsIterator::new(store, start_slot, end_state, end_block_root)
}
/// Load the most recent ancestor state of `state_root` which lies on an epoch boundary.
///
/// If `state_root` corresponds to an epoch boundary state, then that state itself should be
/// returned.
fn load_epoch_boundary_state(
&self,
state_root: &Hash256,
) -> Result<Option<BeaconState<E>>, Error> {
// The default implementation is not very efficient, but isn't used in prod.
// See `HotColdDB` for the optimized implementation.
if let Some(state) = self.get_state(state_root, None)? {
let epoch_boundary_slot = state.slot / E::slots_per_epoch() * E::slots_per_epoch();
if state.slot == epoch_boundary_slot {
Ok(Some(state))
} else {
let epoch_boundary_state_root = state.get_state_root(epoch_boundary_slot)?;
self.get_state(epoch_boundary_state_root, Some(epoch_boundary_slot))
}
} else {
Ok(None)
}
}
fn put_item<I: StoreItem>(&self, key: &Hash256, item: &I) -> Result<(), Error> {
self.put(key, item)
}
fn get_item<I: StoreItem>(&self, key: &Hash256) -> Result<Option<I>, Error> {
self.get(key)
}
fn item_exists<I: StoreItem>(&self, key: &Hash256) -> Result<bool, Error> {
self.exists::<I>(key)
}
fn do_atomically(&self, batch: &[StoreOp]) -> Result<(), Error> {
KeyValueStore::do_atomically(self, batch)
}
}

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

@@ -2,7 +2,7 @@ use crate::chunked_vector::{
load_variable_list_from_db, load_vector_from_db, BlockRoots, HistoricalRoots, RandaoMixes,
StateRoots,
};
use crate::{Error, Store};
use crate::{Error, KeyValueStore};
use ssz_derive::{Decode, Encode};
use std::convert::TryInto;
use types::*;
@@ -113,7 +113,7 @@ impl<T: EthSpec> PartialBeaconState<T> {
}
}
pub fn load_block_roots<S: Store<T>>(
pub fn load_block_roots<S: KeyValueStore<T>>(
&mut self,
store: &S,
spec: &ChainSpec,
@@ -126,7 +126,7 @@ impl<T: EthSpec> PartialBeaconState<T> {
Ok(())
}
pub fn load_state_roots<S: Store<T>>(
pub fn load_state_roots<S: KeyValueStore<T>>(
&mut self,
store: &S,
spec: &ChainSpec,
@@ -139,7 +139,7 @@ impl<T: EthSpec> PartialBeaconState<T> {
Ok(())
}
pub fn load_historical_roots<S: Store<T>>(
pub fn load_historical_roots<S: KeyValueStore<T>>(
&mut self,
store: &S,
spec: &ChainSpec,
@@ -152,7 +152,7 @@ impl<T: EthSpec> PartialBeaconState<T> {
Ok(())
}
pub fn load_randao_mixes<S: Store<T>>(
pub fn load_randao_mixes<S: KeyValueStore<T>>(
&mut self,
store: &S,
spec: &ChainSpec,