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

Implement freezer database (#508)

Browse Source 
* Implement freezer database for state vectors

* Improve BeaconState safe accessors

And fix a bug in the compact committees accessor.

* Banish dodgy type bounds back to gRPC

* Clean up

* Switch to exclusive end points in chunked vec

* Cleaning up and start of tests

* Randao fix, more tests

* Fix unsightly hack

* Resolve test FIXMEs

* Config file support

* More clean-ups, migrator beginnings

* Finish migrator, integrate into BeaconChain

* Fixups

* Fix store tests

* Fix BeaconChain tests

* Fix LMD GHOST tests

* Address review comments, delete 'static bounds

* Cargo format

* Address review comments

* Fix LMD ghost tests

* Update to spec v0.9.0

* Update to v0.9.1

* Bump spec tags for v0.9.1

* Formatting, fix CI failures

* Resolve accidental KeyPair merge conflict

* Document new BeaconState functions

* Fix incorrect cache drops in `advance_caches`

* Update fork choice for v0.9.1

* Clean up some FIXMEs

* Fix a few docs/logs

* Update for new builder paradigm, spec changes

* Freezer DB integration into BeaconNode

* Cleaning up

* This works, clean it up

* Cleanups

* Fix and improve store tests

* Refine store test

* Delete unused beacon_chain_builder.rs

* Fix CLI

* Store state at split slot in hot database

* Make fork choice lookup fast again

* Store freezer DB split slot in the database

* Handle potential div by 0 in chunked_vector

* Exclude committee caches from freezer DB

* Remove FIXME about long-running test
This commit is contained in:
Michael Sproul
2019-11-27 10:54:46 +11:00
committed by Paul Hauner
parent a514968155
commit bf2eeae3f2
36 changed files with 2486 additions and 580 deletions
Download Patch File Download Diff File Expand all files Collapse all files

791
beacon_node/store/src/chunked_vector.rs Normal file
View File

@@ -0,0 +1,791 @@
//! Space-efficient storage for `BeaconState` vector fields.
//!
//! This module provides logic for splitting the `FixedVector` fields of a `BeaconState` into
//! chunks, and storing those chunks in contiguous ranges in the on-disk database. The motiviation
//! for doing this is avoiding massive duplication in every on-disk state. For example, rather than
//! storing the whole `historical_roots` vector, which is updated once every couple of thousand
//! slots, at every slot, we instead store all the historical values as a chunked vector on-disk,
//! and fetch only the slice we need when reconstructing the `historical_roots` of a state.
//!
//! ## Terminology
//!
//! * **Chunk size**: the number of vector values stored per on-disk chunk.
//! * **Vector index** (vindex): index into all the historical values, identifying a single element
//! of the vector being stored.
//! * **Chunk index** (cindex): index into the keyspace of the on-disk database, identifying a chunk
//! of elements. To find the chunk index of a vector index: `cindex = vindex / chunk_size`.
use self::UpdatePattern::*;
use crate::*;
use ssz::{Decode, Encode};
use typenum::Unsigned;
/// Description of how a `BeaconState` field is updated during state processing.
///
/// When storing a state, this allows us to efficiently store only those entries
/// which are not present in the DB already.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UpdatePattern {
/// The value is updated once per `n` slots.
OncePerNSlots { n: u64 },
/// The value is updated once per epoch, for the epoch `current_epoch - lag`.
OncePerEpoch { lag: u64 },
}
/// Map a chunk index to bytes that can be used to key the NoSQL database.
///
/// We shift chunks up by 1 to make room for a genesis chunk that is handled separately.
fn chunk_key(cindex: u64) -> [u8; 8] {
(cindex + 1).to_be_bytes()
}
/// Return the database key for the genesis value.
fn genesis_value_key() -> [u8; 8] {
0u64.to_be_bytes()
}
/// Trait for types representing fields of the `BeaconState`.
///
/// All of the required methods are type-level, because we do most things with fields at the
/// type-level. We require their value-level witnesses to be `Copy` so that we can avoid the
/// turbofish when calling functions like `store_updated_vector`.
pub trait Field<E: EthSpec>: Copy {
/// The type of value stored in this field: the `T` from `FixedVector<T, N>`.
///
/// The `Default` impl will be used to fill extra vector entries.
type Value: Decode + Encode + Default + Clone + PartialEq + std::fmt::Debug;
/// The length of this field: the `N` from `FixedVector<T, N>`.
type Length: Unsigned;
/// The database column where the integer-indexed chunks for this field should be stored.
///
/// Each field's column **must** be unique.
fn column() -> DBColumn;
/// Update pattern for this field, so that we can do differential updates.
fn update_pattern(spec: &ChainSpec) -> UpdatePattern;
/// The number of values to store per chunk on disk.
///
/// Default is 128 so that we read/write 4K pages when the values are 32 bytes.
// TODO: benchmark and optimise this parameter
fn chunk_size() -> usize {
128
}
/// Get the value of this field at the given vector index, from the state.
fn get_value(
state: &BeaconState<E>,
vindex: u64,
spec: &ChainSpec,
) -> Result<Self::Value, ChunkError>;
/// True if this is a `FixedLengthField`, false otherwise.
fn is_fixed_length() -> bool;
/// Compute the start and end vector indices of the slice of history required at `current_slot`.
///
/// ## Example
///
/// If we have a field that is updated once per epoch, then the end vindex will be
/// `current_epoch + 1`, because we want to include the value for the current epoch, and the
/// start vindex will be `end_vindex - Self::Length`, because that's how far back we can look.
fn start_and_end_vindex(current_slot: Slot, spec: &ChainSpec) -> (usize, usize) {
// We take advantage of saturating subtraction on slots and epochs
match Self::update_pattern(spec) {
OncePerNSlots { n } => {
// Per-slot changes exclude the index for the current slot, because
// it won't be set until the slot completes (think of `state_roots`, `block_roots`).
// This also works for the `historical_roots` because at the `n`th slot, the 0th
// entry of the list is created, and before that the list is empty.
let end_vindex = current_slot / n;
let start_vindex = end_vindex - Self::Length::to_u64();
(start_vindex.as_usize(), end_vindex.as_usize())
}
OncePerEpoch { lag } => {
// Per-epoch changes include the index for the current epoch, because it
// will have been set at the most recent epoch boundary.
let current_epoch = current_slot.epoch(E::slots_per_epoch());
let end_epoch = current_epoch + 1 - lag;
let start_epoch = end_epoch + lag - Self::Length::to_u64();
(start_epoch.as_usize(), end_epoch.as_usize())
}
}
}
/// Given an `existing_chunk` stored in the DB, construct an updated chunk to replace it.
fn get_updated_chunk(
existing_chunk: &Chunk<Self::Value>,
chunk_index: usize,
start_vindex: usize,
end_vindex: usize,
state: &BeaconState<E>,
spec: &ChainSpec,
) -> Result<Chunk<Self::Value>, Error> {
let chunk_size = Self::chunk_size();
let mut new_chunk = Chunk::new(vec![Self::Value::default(); chunk_size]);
for i in 0..chunk_size {
let vindex = chunk_index * chunk_size + i;
if vindex >= start_vindex && vindex < end_vindex {
let vector_value = Self::get_value(state, vindex as u64, spec)?;
if let Some(existing_value) = existing_chunk.values.get(i) {
if *existing_value != vector_value && *existing_value != Self::Value::default()
{
return Err(ChunkError::Inconsistent {
field: Self::column(),
chunk_index,
existing_value: format!("{:?}", existing_value),
new_value: format!("{:?}", vector_value),
}
.into());
}
}
new_chunk.values[i] = vector_value;
} else {
new_chunk.values[i] = existing_chunk
.values
.get(i)
.cloned()
.unwrap_or_else(Self::Value::default);
}
}
Ok(new_chunk)
}
/// Determine whether a state at `slot` possesses (or requires) the genesis value.
fn slot_needs_genesis_value(slot: Slot, spec: &ChainSpec) -> bool {
let (_, end_vindex) = Self::start_and_end_vindex(slot, spec);
match Self::update_pattern(spec) {
// If the end_vindex is less than the length of the vector, then the vector
// has not yet been completely filled with non-genesis values, and so the genesis
// value is still required.
OncePerNSlots { .. } => {
Self::is_fixed_length() && end_vindex < Self::Length::to_usize()
}
// If the field has lag, then it takes an extra `lag` vindices beyond the
// `end_vindex` before the vector has been filled with non-genesis values.
OncePerEpoch { lag } => {
Self::is_fixed_length() && end_vindex + (lag as usize) < Self::Length::to_usize()
}
}
}
/// 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>(store: &S) -> Result<Self::Value, Error> {
let key = &genesis_value_key()[..];
let chunk =
Chunk::load(store, Self::column(), key)?.ok_or(ChunkError::MissingGenesisValue)?;
chunk
.values
.first()
.cloned()
.ok_or(ChunkError::MissingGenesisValue.into())
}
/// Store the given `value` as the genesis value for this field, unless stored already.
///
/// 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>(store: &S, value: Self::Value) -> Result<(), Error> {
let key = &genesis_value_key()[..];
if let Some(existing_chunk) = Chunk::<Self::Value>::load(store, Self::column(), key)? {
if existing_chunk.values.len() != 1 {
Err(ChunkError::InvalidGenesisChunk {
field: Self::column(),
expected_len: 1,
observed_len: existing_chunk.values.len(),
}
.into())
} else if existing_chunk.values[0] != value {
Err(ChunkError::InconsistentGenesisValue {
field: Self::column(),
existing_value: format!("{:?}", existing_chunk.values[0]),
new_value: format!("{:?}", value),
}
.into())
} else {
Ok(())
}
} else {
Chunk::new(vec![value]).store(store, Self::column(), &genesis_value_key()[..])
}
}
/// Extract the genesis value for a fixed length field from an
///
/// Will only return a correct value if `slot_needs_genesis_value(state.slot, spec) == true`.
fn extract_genesis_value(
state: &BeaconState<E>,
spec: &ChainSpec,
) -> Result<Self::Value, Error> {
let (_, end_vindex) = Self::start_and_end_vindex(state.slot, spec);
match Self::update_pattern(spec) {
// Genesis value is guaranteed to exist at `end_vindex`, as it won't yet have been
// updated
OncePerNSlots { .. } => Ok(Self::get_value(state, end_vindex as u64, spec)?),
// If there's lag, the value of the field at the vindex *without the lag*
// should still be set to the genesis value.
OncePerEpoch { lag } => Ok(Self::get_value(state, end_vindex as u64 + lag, spec)?),
}
}
}
/// Marker trait for fixed-length fields (`FixedVector<T, N>`).
pub trait FixedLengthField<E: EthSpec>: Field<E> {}
/// Marker trait for variable-length fields (`VariableList<T, N>`).
pub trait VariableLengthField<E: EthSpec>: Field<E> {}
/// Macro to implement the `Field` trait on a new unit struct type.
macro_rules! field {
($struct_name:ident, $marker_trait:ident, $value_ty:ty, $length_ty:ty, $column:expr,
$update_pattern:expr, $get_value:expr) => {
#[derive(Clone, Copy)]
pub struct $struct_name;
impl<T> Field<T> for $struct_name
where
T: EthSpec,
{
type Value = $value_ty;
type Length = $length_ty;
fn column() -> DBColumn {
$column
}
fn update_pattern(spec: &ChainSpec) -> UpdatePattern {
$update_pattern(spec)
}
fn get_value(
state: &BeaconState<T>,
vindex: u64,
spec: &ChainSpec,
) -> Result<Self::Value, ChunkError> {
$get_value(state, vindex, spec)
}
fn is_fixed_length() -> bool {
stringify!($marker_trait) == "FixedLengthField"
}
}
impl<E: EthSpec> $marker_trait<E> for $struct_name {}
};
}
field!(
BlockRoots,
FixedLengthField,
Hash256,
T::SlotsPerHistoricalRoot,
DBColumn::BeaconBlockRoots,
|_| OncePerNSlots { n: 1 },
|state: &BeaconState<_>, index, _| safe_modulo_index(&state.block_roots, index)
);
field!(
StateRoots,
FixedLengthField,
Hash256,
T::SlotsPerHistoricalRoot,
DBColumn::BeaconStateRoots,
|_| OncePerNSlots { n: 1 },
|state: &BeaconState<_>, index, _| safe_modulo_index(&state.state_roots, index)
);
field!(
HistoricalRoots,
VariableLengthField,
Hash256,
T::HistoricalRootsLimit,
DBColumn::BeaconHistoricalRoots,
|_| OncePerNSlots {
n: T::SlotsPerHistoricalRoot::to_u64()
},
|state: &BeaconState<_>, index, _| safe_modulo_index(&state.historical_roots, index)
);
field!(
RandaoMixes,
FixedLengthField,
Hash256,
T::EpochsPerHistoricalVector,
DBColumn::BeaconRandaoMixes,
|_| OncePerEpoch { lag: 1 },
|state: &BeaconState<_>, index, _| safe_modulo_index(&state.randao_mixes, index)
);
pub fn store_updated_vector<F: Field<E>, E: EthSpec, S: Store>(
field: F,
store: &S,
state: &BeaconState<E>,
spec: &ChainSpec,
) -> Result<(), Error> {
let chunk_size = F::chunk_size();
let (start_vindex, end_vindex) = F::start_and_end_vindex(state.slot, spec);
let start_cindex = start_vindex / chunk_size;
let end_cindex = end_vindex / chunk_size;
// Store the genesis value if we have access to it, and it hasn't been stored already.
if F::slot_needs_genesis_value(state.slot, spec) {
let genesis_value = F::extract_genesis_value(state, spec)?;
F::check_and_store_genesis_value(store, genesis_value)?;
}
// Start by iterating backwards from the last chunk, storing new chunks in the database.
// Stop once a chunk in the database matches what we were about to store, this indicates
// that a previously stored state has already filled-in a portion of the indices covered.
let full_range_checked = store_range(
field,
(start_cindex..=end_cindex).rev(),
start_vindex,
end_vindex,
store,
state,
spec,
)?;
// If the previous `store_range` did not check the entire range, it may be the case that the
// state's vector includes elements at low vector indices that are not yet stored in the
// database, so run another `store_range` to ensure these values are also stored.
if !full_range_checked {
store_range(
field,
start_cindex..end_cindex,
start_vindex,
end_vindex,
store,
state,
spec,
)?;
}
Ok(())
}
fn store_range<F, E, S, I>(
_: F,
range: I,
start_vindex: usize,
end_vindex: usize,
store: &S,
state: &BeaconState<E>,
spec: &ChainSpec,
) -> Result<bool, Error>
where
F: Field<E>,
E: EthSpec,
S: Store,
I: Iterator<Item = usize>,
{
for chunk_index in range {
let chunk_key = &chunk_key(chunk_index as u64)[..];
let existing_chunk =
Chunk::<F::Value>::load(store, F::column(), chunk_key)?.unwrap_or_else(Chunk::default);
let new_chunk = F::get_updated_chunk(
&existing_chunk,
chunk_index,
start_vindex,
end_vindex,
state,
spec,
)?;
if new_chunk == existing_chunk {
return Ok(false);
}
new_chunk.store(store, F::column(), chunk_key)?;
}
Ok(true)
}
// Chunks at the end index are included.
// TODO: could be more efficient with a real range query (perhaps RocksDB)
fn range_query<S: Store, T: Decode + Encode>(
store: &S,
column: DBColumn,
start_index: usize,
end_index: usize,
) -> Result<Vec<Chunk<T>>, Error> {
let mut result = vec![];
for chunk_index in start_index..=end_index {
let key = &chunk_key(chunk_index as u64)[..];
let chunk = Chunk::load(store, column, key)?.ok_or(ChunkError::Missing { chunk_index })?;
result.push(chunk);
}
Ok(result)
}
/// Combine chunks to form a list or vector of all values with vindex in `start_vindex..end_vindex`.
///
/// The `length` parameter is the length of the vec to construct, with entries set to `default` if
/// they lie outside the vindex range.
fn stitch<T: Default + Clone>(
chunks: Vec<Chunk<T>>,
start_vindex: usize,
end_vindex: usize,
chunk_size: usize,
length: usize,
default: T,
) -> Result<Vec<T>, ChunkError> {
if start_vindex + length < end_vindex {
return Err(ChunkError::OversizedRange {
start_vindex,
end_vindex,
length,
});
}
let start_cindex = start_vindex / chunk_size;
let end_cindex = end_vindex / chunk_size;
let mut result = vec![default; length];
for (chunk_index, chunk) in (start_cindex..=end_cindex).zip(chunks.into_iter()) {
// All chunks but the last chunk must be full-sized
if chunk_index != end_cindex && chunk.values.len() != chunk_size {
return Err(ChunkError::InvalidSize {
chunk_index,
expected: chunk_size,
actual: chunk.values.len(),
});
}
// Copy the chunk entries into the result vector
for (i, value) in chunk.values.into_iter().enumerate() {
let vindex = chunk_index * chunk_size + i;
if vindex >= start_vindex && vindex < end_vindex {
result[vindex % length] = value;
}
}
}
Ok(result)
}
pub fn load_vector_from_db<F: FixedLengthField<E>, E: EthSpec, S: Store>(
store: &S,
slot: Slot,
spec: &ChainSpec,
) -> Result<FixedVector<F::Value, F::Length>, Error> {
// Do a range query
let chunk_size = F::chunk_size();
let (start_vindex, end_vindex) = F::start_and_end_vindex(slot, spec);
let start_cindex = start_vindex / chunk_size;
let end_cindex = end_vindex / chunk_size;
let chunks = range_query(store, F::column(), start_cindex, end_cindex)?;
let default = if F::slot_needs_genesis_value(slot, spec) {
F::load_genesis_value(store)?
} else {
F::Value::default()
};
let result = stitch(
chunks,
start_vindex,
end_vindex,
chunk_size,
F::Length::to_usize(),
default,
)?;
Ok(result.into())
}
/// 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>(
store: &S,
slot: Slot,
spec: &ChainSpec,
) -> Result<VariableList<F::Value, F::Length>, Error> {
let chunk_size = F::chunk_size();
let (start_vindex, end_vindex) = F::start_and_end_vindex(slot, spec);
let start_cindex = start_vindex / chunk_size;
let end_cindex = end_vindex / chunk_size;
let chunks: Vec<Chunk<F::Value>> = range_query(store, F::column(), start_cindex, end_cindex)?;
let mut result = Vec::with_capacity(chunk_size * chunks.len());
for (chunk_index, chunk) in chunks.into_iter().enumerate() {
for (i, value) in chunk.values.into_iter().enumerate() {
let vindex = chunk_index * chunk_size + i;
if vindex >= start_vindex && vindex < end_vindex {
result.push(value);
}
}
}
Ok(result.into())
}
/// Index into a field of the state, avoiding out of bounds and division by 0.
fn safe_modulo_index<T: Copy>(values: &[T], index: u64) -> Result<T, ChunkError> {
if values.is_empty() {
Err(ChunkError::ZeroLengthVector)
} else {
Ok(values[index as usize % values.len()])
}
}
/// A chunk of a fixed-size vector from the `BeaconState`, stored in the database.
#[derive(Debug, Clone, PartialEq)]
pub struct Chunk<T> {
/// A vector of up-to `chunk_size` values.
pub values: Vec<T>,
}
impl<T> Default for Chunk<T>
where
T: Decode + Encode,
{
fn default() -> Self {
Chunk { values: vec![] }
}
}
impl<T> Chunk<T>
where
T: Decode + Encode,
{
pub fn new(values: Vec<T>) -> Self {
Chunk { values }
}
pub fn load<S: Store>(store: &S, column: DBColumn, key: &[u8]) -> Result<Option<Self>, Error> {
store
.get_bytes(column.into(), key)?
.map(|bytes| Self::decode(&bytes))
.transpose()
}
pub fn store<S: Store>(&self, store: &S, column: DBColumn, key: &[u8]) -> Result<(), Error> {
store.put_bytes(column.into(), key, &self.encode()?)?;
Ok(())
}
/// Attempt to decode a single chunk.
pub fn decode(bytes: &[u8]) -> Result<Self, Error> {
if !<T as Decode>::is_ssz_fixed_len() {
return Err(Error::from(ChunkError::InvalidType));
}
let value_size = <T as Decode>::ssz_fixed_len();
if value_size == 0 {
return Err(Error::from(ChunkError::InvalidType));
}
let values = bytes
.chunks(value_size)
.map(T::from_ssz_bytes)
.collect::<Result<_, _>>()?;
Ok(Chunk { values })
}
pub fn encoded_size(&self) -> usize {
self.values.len() * <T as Encode>::ssz_fixed_len()
}
/// Encode a single chunk as bytes.
pub fn encode(&self) -> Result<Vec<u8>, Error> {
if !<T as Encode>::is_ssz_fixed_len() {
return Err(Error::from(ChunkError::InvalidType));
}
Ok(self.values.iter().flat_map(T::as_ssz_bytes).collect())
}
}
#[derive(Debug, PartialEq)]
pub enum ChunkError {
ZeroLengthVector,
InvalidSize {
chunk_index: usize,
expected: usize,
actual: usize,
},
Missing {
chunk_index: usize,
},
MissingGenesisValue,
Inconsistent {
field: DBColumn,
chunk_index: usize,
existing_value: String,
new_value: String,
},
InconsistentGenesisValue {
field: DBColumn,
existing_value: String,
new_value: String,
},
InvalidGenesisChunk {
field: DBColumn,
expected_len: usize,
observed_len: usize,
},
InvalidType,
OversizedRange {
start_vindex: usize,
end_vindex: usize,
length: usize,
},
}
#[cfg(test)]
mod test {
use super::*;
use types::MainnetEthSpec as TestSpec;
use types::*;
fn v(i: u64) -> Hash256 {
Hash256::from_low_u64_be(i)
}
#[test]
fn stitch_default() {
let chunk_size = 4;
let chunks = vec![
Chunk::new(vec![0u64, 1, 2, 3]),
Chunk::new(vec![4, 5, 0, 0]),
];
assert_eq!(
stitch(chunks.clone(), 2, 6, chunk_size, 12, 99).unwrap(),
vec![99, 99, 2, 3, 4, 5, 99, 99, 99, 99, 99, 99]
);
}
#[test]
fn stitch_basic() {
let chunk_size = 4;
let default = v(0);
let chunks = vec![
Chunk::new(vec![v(0), v(1), v(2), v(3)]),
Chunk::new(vec![v(4), v(5), v(6), v(7)]),
Chunk::new(vec![v(8), v(9), v(10), v(11)]),
];
assert_eq!(
stitch(chunks.clone(), 0, 12, chunk_size, 12, default).unwrap(),
(0..12).map(v).collect::<Vec<_>>()
);
assert_eq!(
stitch(chunks.clone(), 2, 10, chunk_size, 8, default).unwrap(),
vec![v(8), v(9), v(2), v(3), v(4), v(5), v(6), v(7)]
);
}
#[test]
fn stitch_oversized_range() {
let chunk_size = 4;
let default = 0;
let chunks = vec![Chunk::new(vec![20u64, 21, 22, 23])];
// Args (start_vindex, end_vindex, length)
let args = vec![(0, 21, 20), (0, 2048, 1024), (0, 2, 1)];
for (start_vindex, end_vindex, length) in args {
assert_eq!(
stitch(
chunks.clone(),
start_vindex,
end_vindex,
chunk_size,
length,
default
),
Err(ChunkError::OversizedRange {
start_vindex,
end_vindex,
length,
})
);
}
}
#[test]
fn fixed_length_fields() {
fn test_fixed_length<F: Field<TestSpec>>(_: F, expected: bool) {
assert_eq!(F::is_fixed_length(), expected);
}
test_fixed_length(BlockRoots, true);
test_fixed_length(StateRoots, true);
test_fixed_length(HistoricalRoots, false);
test_fixed_length(RandaoMixes, true);
}
fn needs_genesis_value_once_per_slot<F: Field<TestSpec>>(_: F) {
let spec = &TestSpec::default_spec();
let max = F::Length::to_u64();
for i in 0..max {
assert!(
F::slot_needs_genesis_value(Slot::new(i), spec),
"slot {}",
i
);
}
assert!(!F::slot_needs_genesis_value(Slot::new(max), spec));
}
#[test]
fn needs_genesis_value_block_roots() {
needs_genesis_value_once_per_slot(BlockRoots);
}
#[test]
fn needs_genesis_value_state_roots() {
needs_genesis_value_once_per_slot(StateRoots);
}
#[test]
fn needs_genesis_value_historical_roots() {
let spec = &TestSpec::default_spec();
assert!(
!<HistoricalRoots as Field<TestSpec>>::slot_needs_genesis_value(Slot::new(0), spec)
);
}
fn needs_genesis_value_test_randao<F: Field<TestSpec>>(_: F) {
let spec = &TestSpec::default_spec();
let max = TestSpec::slots_per_epoch() as u64 * (F::Length::to_u64() - 1);
for i in 0..max {
assert!(
F::slot_needs_genesis_value(Slot::new(i), spec),
"slot {}",
i
);
}
assert!(!F::slot_needs_genesis_value(Slot::new(max), spec));
}
#[test]
fn needs_genesis_value_randao() {
needs_genesis_value_test_randao(RandaoMixes);
}
}

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

@@ -1,8 +1,15 @@
use crate::chunked_vector::ChunkError;
use crate::hot_cold_store::HotColdDbError;
use ssz::DecodeError;
use types::BeaconStateError;
#[derive(Debug, PartialEq)]
pub enum Error {
SszDecodeError(DecodeError),
VectorChunkError(ChunkError),
BeaconStateError(BeaconStateError),
PartialBeaconStateError,
HotColdDbError(HotColdDbError),
DBError { message: String },
}
@@ -12,6 +19,24 @@ impl From<DecodeError> for Error {
}
}
impl From<ChunkError> for Error {
fn from(e: ChunkError) -> Error {
Error::VectorChunkError(e)
}
}
impl From<HotColdDbError> for Error {
fn from(e: HotColdDbError) -> Error {
Error::HotColdDbError(e)
}
}
impl From<BeaconStateError> for Error {
fn from(e: BeaconStateError) -> Error {
Error::BeaconStateError(e)
}
}
impl From<DBError> for Error {
fn from(e: DBError) -> Error {
Error::DBError { message: e.message }

260
beacon_node/store/src/hot_cold_store.rs Normal file
View File

@@ -0,0 +1,260 @@
use crate::chunked_vector::{
store_updated_vector, BlockRoots, HistoricalRoots, RandaoMixes, StateRoots,
};
use crate::iter::StateRootsIterator;
use crate::{
leveldb_store::LevelDB, DBColumn, Error, PartialBeaconState, SimpleStoreItem, Store, StoreItem,
};
use parking_lot::RwLock;
use slog::{info, trace, Logger};
use ssz::{Decode, Encode};
use std::convert::TryInto;
use std::path::Path;
use std::sync::Arc;
use types::*;
/// 32-byte key for accessing the `split_slot` of the freezer DB.
pub const SPLIT_SLOT_DB_KEY: &str = "FREEZERDBSPLITSLOTFREEZERDBSPLIT";
pub struct HotColdDB {
/// The slot before which all data is stored in the cold database.
///
/// Data for slots less than `split_slot` is in the cold DB, while data for slots
/// greater than or equal is in the hot DB.
split_slot: RwLock<Slot>,
/// Cold database containing compact historical data.
cold_db: LevelDB,
/// Hot database containing duplicated but quick-to-access recent data.
hot_db: LevelDB,
/// Chain spec.
spec: ChainSpec,
/// Logger.
pub(crate) log: Logger,
}
#[derive(Debug, PartialEq)]
pub enum HotColdDbError {
FreezeSlotError {
current_split_slot: Slot,
proposed_split_slot: Slot,
},
}
impl Store for HotColdDB {
// 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 state in the store.
fn put_state<E: EthSpec>(
&self,
state_root: &Hash256,
state: &BeaconState<E>,
) -> Result<(), Error> {
if state.slot < self.get_split_slot() {
self.store_archive_state(state_root, state)
} else {
self.hot_db.put_state(state_root, state)
}
}
/// Fetch a state from the store.
fn get_state<E: EthSpec>(
&self,
state_root: &Hash256,
slot: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
if let Some(slot) = slot {
if slot < self.get_split_slot() {
self.load_archive_state(state_root)
} else {
self.hot_db.get_state(state_root, None)
}
} else {
match self.hot_db.get_state(state_root, None)? {
Some(state) => Ok(Some(state)),
None => self.load_archive_state(state_root),
}
}
}
fn freeze_to_state<E: EthSpec>(
store: Arc<Self>,
_frozen_head_root: Hash256,
frozen_head: &BeaconState<E>,
) -> Result<(), Error> {
info!(
store.log,
"Freezer migration started";
"slot" => frozen_head.slot
);
// 1. Copy all of the states between the head and the split slot, from the hot DB
// to the cold DB.
let current_split_slot = store.get_split_slot();
if frozen_head.slot < current_split_slot {
Err(HotColdDbError::FreezeSlotError {
current_split_slot,
proposed_split_slot: frozen_head.slot,
})?;
}
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)
{
trace!(store.log, "Freezing";
"slot" => slot,
"state_root" => format!("{}", state_root));
let state: BeaconState<E> = match store.hot_db.get_state(&state_root, None)? {
Some(s) => s,
// If there's no state it could be a skip slot, which is fine, our job is just
// to move everything that was in the hot DB to the cold.
None => continue,
};
to_delete.push(state_root);
store.store_archive_state(&state_root, &state)?;
}
// 2. Update the split slot
*store.split_slot.write() = frozen_head.slot;
store.store_split_slot()?;
// 3. Delete from the hot DB
for state_root in to_delete {
store
.hot_db
.key_delete(DBColumn::BeaconState.into(), state_root.as_bytes())?;
}
info!(
store.log,
"Freezer migration complete";
"slot" => frozen_head.slot
);
Ok(())
}
}
impl HotColdDB {
pub fn open(
hot_path: &Path,
cold_path: &Path,
spec: ChainSpec,
log: Logger,
) -> Result<Self, Error> {
let db = HotColdDB {
split_slot: RwLock::new(Slot::new(0)),
cold_db: LevelDB::open(cold_path)?,
hot_db: LevelDB::open(hot_path)?,
spec,
log,
};
// Load the previous split slot from the database (if any). This ensures we can
// stop and restart correctly.
if let Some(split_slot) = db.load_split_slot()? {
*db.split_slot.write() = split_slot;
}
Ok(db)
}
pub fn store_archive_state<E: EthSpec>(
&self,
state_root: &Hash256,
state: &BeaconState<E>,
) -> Result<(), Error> {
trace!(
self.log,
"Freezing state";
"slot" => state.slot.as_u64(),
"state_root" => format!("{:?}", state_root)
);
// 1. Convert to PartialBeaconState and store that in the DB.
let partial_state = PartialBeaconState::from_state_forgetful(state);
partial_state.db_put(&self.cold_db, state_root)?;
// 2. Store updated vector entries.
let db = &self.cold_db;
store_updated_vector(BlockRoots, db, state, &self.spec)?;
store_updated_vector(StateRoots, db, state, &self.spec)?;
store_updated_vector(HistoricalRoots, db, state, &self.spec)?;
store_updated_vector(RandaoMixes, db, state, &self.spec)?;
Ok(())
}
pub fn load_archive_state<E: EthSpec>(
&self,
state_root: &Hash256,
) -> Result<Option<BeaconState<E>>, Error> {
let mut partial_state = match PartialBeaconState::db_get(&self.cold_db, state_root)? {
Some(s) => s,
None => return Ok(None),
};
// Fill in the fields of the partial state.
partial_state.load_block_roots(&self.cold_db, &self.spec)?;
partial_state.load_state_roots(&self.cold_db, &self.spec)?;
partial_state.load_historical_roots(&self.cold_db, &self.spec)?;
partial_state.load_randao_mixes(&self.cold_db, &self.spec)?;
let state: BeaconState<E> = partial_state.try_into()?;
Ok(Some(state))
}
pub fn get_split_slot(&self) -> Slot {
*self.split_slot.read()
}
fn load_split_slot(&self) -> Result<Option<Slot>, Error> {
let key = Hash256::from_slice(SPLIT_SLOT_DB_KEY.as_bytes());
let split_slot: Option<SplitSlot> = self.hot_db.get(&key)?;
Ok(split_slot.map(|s| Slot::new(s.0)))
}
fn store_split_slot(&self) -> Result<(), Error> {
let key = Hash256::from_slice(SPLIT_SLOT_DB_KEY.as_bytes());
self.hot_db
.put(&key, &SplitSlot(self.get_split_slot().as_u64()))?;
Ok(())
}
}
/// Struct for storing the split slot in the database.
#[derive(Clone, Copy)]
struct SplitSlot(u64);
impl SimpleStoreItem for SplitSlot {
fn db_column() -> DBColumn {
DBColumn::BeaconMeta
}
fn as_store_bytes(&self) -> Vec<u8> {
self.0.as_ssz_bytes()
}
fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
Ok(SplitSlot(u64::from_ssz_bytes(bytes)?))
}
}

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

@@ -1,9 +1,10 @@
use crate::*;
use ssz::{Decode, Encode};
mod beacon_state;
pub mod beacon_state;
pub mod partial_beacon_state;
impl<T: EthSpec> StoreItem for BeaconBlock<T> {
impl<T: EthSpec> SimpleStoreItem for BeaconBlock<T> {
fn db_column() -> DBColumn {
DBColumn::BeaconBlock
}
@@ -19,7 +20,7 @@ impl<T: EthSpec> StoreItem for BeaconBlock<T> {
bytes
}
fn from_store_bytes(bytes: &mut [u8]) -> Result<Self, Error> {
fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
let timer = metrics::start_timer(&metrics::BEACON_BLOCK_READ_TIMES);
let len = bytes.len();

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

@@ -4,6 +4,43 @@ use ssz_derive::{Decode, Encode};
use std::convert::TryInto;
use types::beacon_state::{BeaconTreeHashCache, CommitteeCache, CACHED_EPOCHS};
pub fn store_full_state<S: Store, E: EthSpec>(
store: &S,
state_root: &Hash256,
state: &BeaconState<E>,
) -> Result<(), Error> {
let timer = metrics::start_timer(&metrics::BEACON_STATE_WRITE_TIMES);
let bytes = StorageContainer::new(state).as_ssz_bytes();
let result = store.put_bytes(DBColumn::BeaconState.into(), state_root.as_bytes(), &bytes);
metrics::stop_timer(timer);
metrics::inc_counter(&metrics::BEACON_STATE_WRITE_COUNT);
metrics::inc_counter_by(&metrics::BEACON_STATE_WRITE_BYTES, bytes.len() as i64);
result
}
pub fn get_full_state<S: Store, E: EthSpec>(
store: &S,
state_root: &Hash256,
) -> Result<Option<BeaconState<E>>, Error> {
let timer = metrics::start_timer(&metrics::BEACON_STATE_READ_TIMES);
match store.get_bytes(DBColumn::BeaconState.into(), state_root.as_bytes())? {
Some(bytes) => {
let container = StorageContainer::from_ssz_bytes(&bytes)?;
metrics::stop_timer(timer);
metrics::inc_counter(&metrics::BEACON_STATE_READ_COUNT);
metrics::inc_counter_by(&metrics::BEACON_STATE_READ_BYTES, bytes.len() as i64);
Ok(Some(container.try_into()?))
}
None => Ok(None),
}
}
/// A container for storing `BeaconState` components.
// TODO: would be more space efficient with the caches stored separately and referenced by hash
#[derive(Encode, Decode)]
@@ -53,36 +90,3 @@ impl<T: EthSpec> TryInto<BeaconState<T>> for StorageContainer {
Ok(state)
}
}
impl<T: EthSpec> StoreItem for BeaconState<T> {
fn db_column() -> DBColumn {
DBColumn::BeaconState
}
fn as_store_bytes(&self) -> Vec<u8> {
let timer = metrics::start_timer(&metrics::BEACON_STATE_WRITE_TIMES);
let container = StorageContainer::new(self);
let bytes = container.as_ssz_bytes();
metrics::stop_timer(timer);
metrics::inc_counter(&metrics::BEACON_STATE_WRITE_COUNT);
metrics::inc_counter_by(&metrics::BEACON_STATE_WRITE_BYTES, bytes.len() as i64);
bytes
}
fn from_store_bytes(bytes: &mut [u8]) -> Result<Self, Error> {
let timer = metrics::start_timer(&metrics::BEACON_STATE_READ_TIMES);
let len = bytes.len();
let container = StorageContainer::from_ssz_bytes(bytes)?;
let result = container.try_into();
metrics::stop_timer(timer);
metrics::inc_counter(&metrics::BEACON_STATE_READ_COUNT);
metrics::inc_counter_by(&metrics::BEACON_STATE_READ_BYTES, len as i64);
result
}
}

16
beacon_node/store/src/impls/partial_beacon_state.rs Normal file
View File

@@ -0,0 +1,16 @@
use crate::*;
use ssz::{Decode, Encode};
impl<T: EthSpec> SimpleStoreItem for PartialBeaconState<T> {
fn db_column() -> DBColumn {
DBColumn::BeaconState
}
fn as_store_bytes(&self) -> Vec<u8> {
self.as_ssz_bytes()
}
fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
Ok(Self::from_ssz_bytes(bytes)?)
}
}

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

@@ -18,7 +18,7 @@ impl<'a, U: Store, E: EthSpec> AncestorIter<U, BlockRootsIterator<'a, E, U>> for
/// 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<U>) -> Option<BlockRootsIterator<'a, E, U>> {
let state = store.get::<BeaconState<E>>(&self.state_root).ok()??;
let state = store.get_state(&self.state_root, Some(self.slot)).ok()??;
Some(BlockRootsIterator::owned(store, state))
}
@@ -33,13 +33,22 @@ impl<'a, U: Store, E: EthSpec> AncestorIter<U, StateRootsIterator<'a, E, U>> for
}
}
#[derive(Clone)]
pub struct StateRootsIterator<'a, T: EthSpec, U> {
store: Arc<U>,
beacon_state: Cow<'a, BeaconState<T>>,
slot: Slot,
}
impl<'a, T: EthSpec, U> Clone for StateRootsIterator<'a, T, U> {
fn clone(&self) -> Self {
Self {
store: self.store.clone(),
beacon_state: self.beacon_state.clone(),
slot: self.slot,
}
}
}
impl<'a, T: EthSpec, U: Store> StateRootsIterator<'a, T, U> {
pub fn new(store: Arc<U>, beacon_state: &'a BeaconState<T>) -> Self {
Self {
@@ -62,7 +71,7 @@ impl<'a, T: EthSpec, U: Store> Iterator for StateRootsIterator<'a, T, U> {
type Item = (Hash256, Slot);
fn next(&mut self) -> Option<Self::Item> {
if (self.slot == 0) || (self.slot > self.beacon_state.slot) {
if self.slot == 0 || self.slot > self.beacon_state.slot {
return None;
}
@@ -75,7 +84,7 @@ impl<'a, T: EthSpec, U: Store> Iterator for StateRootsIterator<'a, T, U> {
let beacon_state: BeaconState<T> = {
let new_state_root = self.beacon_state.get_oldest_state_root().ok()?;
self.store.get(&new_state_root).ok()?
self.store.get_state(&new_state_root, None).ok()?
}?;
self.beacon_state = Cow::Owned(beacon_state);
@@ -128,13 +137,22 @@ impl<'a, T: EthSpec, U: Store> Iterator for BlockIterator<'a, T, U> {
/// exhausted.
///
/// Returns `None` for roots prior to genesis or when there is an error reading from `Store`.
#[derive(Clone)]
pub struct BlockRootsIterator<'a, T: EthSpec, U> {
store: Arc<U>,
beacon_state: Cow<'a, BeaconState<T>>,
slot: Slot,
}
impl<'a, T: EthSpec, U> Clone for BlockRootsIterator<'a, T, U> {
fn clone(&self) -> Self {
Self {
store: self.store.clone(),
beacon_state: self.beacon_state.clone(),
slot: self.slot,
}
}
}
impl<'a, T: EthSpec, U: Store> BlockRootsIterator<'a, T, U> {
/// Create a new iterator over all block roots in the given `beacon_state` and prior states.
pub fn new(store: Arc<U>, beacon_state: &'a BeaconState<T>) -> Self {
@@ -173,7 +191,7 @@ impl<'a, T: EthSpec, U: Store> Iterator for BlockRootsIterator<'a, T, U> {
// Load the earliest state from disk.
let new_state_root = self.beacon_state.get_oldest_state_root().ok()?;
self.store.get(&new_state_root).ok()?
self.store.get_state(&new_state_root, None).ok()?
}?;
self.beacon_state = Cow::Owned(beacon_state);
@@ -187,6 +205,52 @@ impl<'a, T: EthSpec, U: Store> Iterator for BlockRootsIterator<'a, T, U> {
}
}
pub type ReverseBlockRootIterator<'a, E, S> =
ReverseHashAndSlotIterator<BlockRootsIterator<'a, E, S>>;
pub type ReverseStateRootIterator<'a, E, S> =
ReverseHashAndSlotIterator<StateRootsIterator<'a, E, S>>;
pub type ReverseHashAndSlotIterator<I> = ReverseChainIterator<(Hash256, Slot), I>;
/// Provides a wrapper for an iterator that returns a given `T` before it starts returning results of
/// the `Iterator`.
pub struct ReverseChainIterator<T, I> {
first_value_used: bool,
first_value: T,
iter: I,
}
impl<T, I> ReverseChainIterator<T, I>
where
T: Sized,
I: Iterator<Item = T> + Sized,
{
pub fn new(first_value: T, iter: I) -> Self {
Self {
first_value_used: false,
first_value,
iter,
}
}
}
impl<T, I> Iterator for ReverseChainIterator<T, I>
where
T: Clone,
I: Iterator<Item = T>,
{
type Item = T;
fn next(&mut self) -> Option<Self::Item> {
if self.first_value_used {
self.iter.next()
} else {
self.first_value_used = true;
Some(self.first_value.clone())
}
}
}
#[cfg(test)]
mod test {
use super::*;
@@ -225,7 +289,7 @@ mod test {
let state_a_root = hashes.next().unwrap();
state_b.state_roots[0] = state_a_root;
store.put(&state_a_root, &state_a).unwrap();
store.put_state(&state_a_root, &state_a).unwrap();
let iter = BlockRootsIterator::new(store.clone(), &state_b);
@@ -273,8 +337,8 @@ mod test {
let state_a_root = Hash256::from_low_u64_be(slots_per_historical_root as u64);
let state_b_root = Hash256::from_low_u64_be(slots_per_historical_root as u64 * 2);
store.put(&state_a_root, &state_a).unwrap();
store.put(&state_b_root, &state_b).unwrap();
store.put_state(&state_a_root, &state_a).unwrap();
store.put_state(&state_b_root, &state_b).unwrap();
let iter = StateRootsIterator::new(store.clone(), &state_b);

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

@@ -1,4 +1,5 @@
use super::*;
use crate::impls::beacon_state::{get_full_state, store_full_state};
use crate::metrics;
use db_key::Key;
use leveldb::database::kv::KV;
@@ -6,14 +7,10 @@ use leveldb::database::Database;
use leveldb::error::Error as LevelDBError;
use leveldb::options::{Options, ReadOptions, WriteOptions};
use std::path::Path;
use std::sync::Arc;
/// A wrapped leveldb database.
#[derive(Clone)]
pub struct LevelDB {
// Note: this `Arc` is only included because of an artificial constraint by gRPC. Hopefully we
// can remove this one day.
db: Arc<Database<BytesKey>>,
db: Database<BytesKey>,
}
impl LevelDB {
@@ -23,7 +20,7 @@ impl LevelDB {
options.create_if_missing = true;
let db = Arc::new(Database::open(path, options)?);
let db = Database::open(path, options)?;
Ok(Self { db })
}
@@ -111,6 +108,24 @@ impl Store for LevelDB {
.delete(self.write_options(), column_key)
.map_err(Into::into)
}
/// Store a state in the store.
fn put_state<E: EthSpec>(
&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<E: EthSpec>(
&self,
state_root: &Hash256,
_: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
get_full_state(self, state_root)
}
}
impl From<LevelDBError> for Error {

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

@@ -11,16 +11,25 @@
extern crate lazy_static;
mod block_at_slot;
pub mod chunked_vector;
mod errors;
mod hot_cold_store;
mod impls;
mod leveldb_store;
mod memory_store;
mod metrics;
mod partial_beacon_state;
pub mod iter;
pub mod migrate;
pub use self::leveldb_store::LevelDB as DiskStore;
use std::sync::Arc;
pub use self::hot_cold_store::HotColdDB as DiskStore;
pub use self::leveldb_store::LevelDB as SimpleDiskStore;
pub use self::memory_store::MemoryStore;
pub use self::migrate::Migrate;
pub use self::partial_beacon_state::PartialBeaconState;
pub use errors::Error;
pub use metrics::scrape_for_metrics;
pub use types::*;
@@ -30,9 +39,21 @@ pub use types::*;
/// 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: Sync + Send + Sized {
pub trait Store: Sync + Send + Sized + 'static {
/// Retrieve some bytes in `column` with `key`.
fn get_bytes(&self, column: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error>;
/// Store some `value` in `column`, indexed with `key`.
fn put_bytes(&self, column: &str, key: &[u8], value: &[u8]) -> Result<(), Error>;
/// Return `true` if `key` exists in `column`.
fn key_exists(&self, column: &str, key: &[u8]) -> Result<bool, Error>;
/// Removes `key` from `column`.
fn key_delete(&self, column: &str, key: &[u8]) -> Result<(), Error>;
/// Store an item in `Self`.
fn put(&self, key: &Hash256, item: &impl StoreItem) -> Result<(), Error> {
fn put<I: StoreItem>(&self, key: &Hash256, item: &I) -> Result<(), Error> {
item.db_put(self, key)
}
@@ -51,6 +72,20 @@ pub trait Store: Sync + Send + Sized {
I::db_delete(self, key)
}
/// Store a state in the store.
fn put_state<E: EthSpec>(
&self,
state_root: &Hash256,
state: &BeaconState<E>,
) -> Result<(), Error>;
/// Fetch a state from the store.
fn get_state<E: EthSpec>(
&self,
state_root: &Hash256,
slot: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error>;
/// Given the root of an existing block in the store (`start_block_root`), return a parent
/// block with the specified `slot`.
///
@@ -64,42 +99,48 @@ pub trait Store: Sync + Send + Sized {
block_at_slot::get_block_at_preceeding_slot::<_, E>(self, slot, start_block_root)
}
/// Retrieve some bytes in `column` with `key`.
fn get_bytes(&self, column: &str, key: &[u8]) -> Result<Option<Vec<u8>>, Error>;
/// Store some `value` in `column`, indexed with `key`.
fn put_bytes(&self, column: &str, key: &[u8], value: &[u8]) -> Result<(), Error>;
/// Return `true` if `key` exists in `column`.
fn key_exists(&self, column: &str, key: &[u8]) -> Result<bool, Error>;
/// Removes `key` from `column`.
fn key_delete(&self, column: &str, key: &[u8]) -> Result<(), Error>;
/// (Optionally) Move all data before the frozen slot to the freezer database.
fn freeze_to_state<E: EthSpec>(
_store: Arc<Self>,
_frozen_head_root: Hash256,
_frozen_head: &BeaconState<E>,
) -> Result<(), Error> {
Ok(())
}
}
/// A unique column identifier.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum DBColumn {
/// For data related to the database itself.
BeaconMeta,
BeaconBlock,
BeaconState,
BeaconChain,
BeaconBlockRoots,
BeaconStateRoots,
BeaconHistoricalRoots,
BeaconRandaoMixes,
}
impl<'a> Into<&'a str> for DBColumn {
impl Into<&'static str> for DBColumn {
/// Returns a `&str` that can be used for keying a key-value data base.
fn into(self) -> &'a str {
fn into(self) -> &'static str {
match self {
DBColumn::BeaconBlock => &"blk",
DBColumn::BeaconState => &"ste",
DBColumn::BeaconChain => &"bch",
DBColumn::BeaconMeta => "bma",
DBColumn::BeaconBlock => "blk",
DBColumn::BeaconState => "ste",
DBColumn::BeaconChain => "bch",
DBColumn::BeaconBlockRoots => "bbr",
DBColumn::BeaconStateRoots => "bsr",
DBColumn::BeaconHistoricalRoots => "bhr",
DBColumn::BeaconRandaoMixes => "brm",
}
}
}
/// An item that may be stored in a `Store`.
///
/// Provides default methods that are suitable for most applications, however when overridden they
/// provide full customizability of `Store` operations.
pub trait StoreItem: Sized {
/// An item that may stored in a `Store` by serializing and deserializing from bytes.
pub trait SimpleStoreItem: Sized {
/// Identifies which column this item should be placed in.
fn db_column() -> DBColumn;
@@ -107,10 +148,32 @@ pub trait StoreItem: Sized {
fn as_store_bytes(&self) -> Vec<u8>;
/// De-serialize `self` from bytes.
fn from_store_bytes(bytes: &mut [u8]) -> Result<Self, Error>;
///
/// Return an instance of the type and the number of bytes that were read.
fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error>;
}
/// An item that may be stored in a `Store`.
pub trait StoreItem: Sized {
/// Store `self`.
fn db_put(&self, store: &impl Store, key: &Hash256) -> Result<(), Error> {
fn db_put<S: Store>(&self, store: &S, key: &Hash256) -> Result<(), Error>;
/// Retrieve an instance of `Self` from `store`.
fn db_get<S: Store>(store: &S, key: &Hash256) -> Result<Option<Self>, Error>;
/// Return `true` if an instance of `Self` exists in `store`.
fn db_exists<S: Store>(store: &S, key: &Hash256) -> Result<bool, Error>;
/// Delete an instance of `Self` from `store`.
fn db_delete<S: Store>(store: &S, key: &Hash256) -> Result<(), Error>;
}
impl<T> StoreItem for T
where
T: SimpleStoreItem,
{
/// Store `self`.
fn db_put<S: Store>(&self, store: &S, key: &Hash256) -> Result<(), Error> {
let column = Self::db_column().into();
let key = key.as_bytes();
@@ -120,18 +183,18 @@ pub trait StoreItem: Sized {
}
/// Retrieve an instance of `Self`.
fn db_get(store: &impl Store, key: &Hash256) -> Result<Option<Self>, Error> {
fn db_get<S: Store>(store: &S, key: &Hash256) -> Result<Option<Self>, Error> {
let column = Self::db_column().into();
let key = key.as_bytes();
match store.get_bytes(column, key)? {
Some(mut bytes) => Ok(Some(Self::from_store_bytes(&mut bytes[..])?)),
Some(bytes) => Ok(Some(Self::from_store_bytes(&bytes[..])?)),
None => Ok(None),
}
}
/// Return `true` if an instance of `Self` exists in `Store`.
fn db_exists(store: &impl Store, key: &Hash256) -> Result<bool, Error> {
fn db_exists<S: Store>(store: &S, key: &Hash256) -> Result<bool, Error> {
let column = Self::db_column().into();
let key = key.as_bytes();
@@ -139,7 +202,7 @@ pub trait StoreItem: Sized {
}
/// Delete `self` from the `Store`.
fn db_delete(store: &impl Store, key: &Hash256) -> Result<(), Error> {
fn db_delete<S: Store>(store: &S, key: &Hash256) -> Result<(), Error> {
let column = Self::db_column().into();
let key = key.as_bytes();
@@ -160,7 +223,7 @@ mod tests {
b: u64,
}
impl StoreItem for StorableThing {
impl SimpleStoreItem for StorableThing {
fn db_column() -> DBColumn {
DBColumn::BeaconBlock
}
@@ -169,7 +232,7 @@ mod tests {
self.as_ssz_bytes()
}
fn from_store_bytes(bytes: &mut [u8]) -> Result<Self, Error> {
fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
Self::from_ssz_bytes(bytes).map_err(Into::into)
}
}
@@ -196,9 +259,22 @@ mod tests {
#[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(), spec, log).unwrap();
test_impl(store);
}
#[test]
fn simplediskdb() {
let dir = tempdir().unwrap();
let path = dir.path();
let store = DiskStore::open(&path).unwrap();
let store = SimpleDiskStore::open(&path).unwrap();
test_impl(store);
}

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

@@ -1,23 +1,29 @@
use super::{Error, Store};
use crate::impls::beacon_state::{get_full_state, store_full_state};
use parking_lot::RwLock;
use std::collections::HashMap;
use std::sync::Arc;
use types::*;
type DBHashMap = HashMap<Vec<u8>, Vec<u8>>;
/// A thread-safe `HashMap` wrapper.
#[derive(Clone)]
pub struct MemoryStore {
// Note: this `Arc` is only included because of an artificial constraint by gRPC. Hopefully we
// can remove this one day.
db: Arc<RwLock<DBHashMap>>,
db: RwLock<DBHashMap>,
}
impl Clone for MemoryStore {
fn clone(&self) -> Self {
Self {
db: RwLock::new(self.db.read().clone()),
}
}
}
impl MemoryStore {
/// Create a new, empty database.
pub fn open() -> Self {
Self {
db: Arc::new(RwLock::new(HashMap::new())),
db: RwLock::new(HashMap::new()),
}
}
@@ -64,4 +70,22 @@ impl Store for MemoryStore {
Ok(())
}
/// Store a state in the store.
fn put_state<E: EthSpec>(
&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<E: EthSpec>(
&self,
state_root: &Hash256,
_: Option<Slot>,
) -> Result<Option<BeaconState<E>>, Error> {
get_full_state(self, state_root)
}
}

142
beacon_node/store/src/migrate.rs Normal file
View File

@@ -0,0 +1,142 @@
use crate::{DiskStore, MemoryStore, SimpleDiskStore, Store};
use parking_lot::Mutex;
use slog::warn;
use std::mem;
use std::sync::mpsc;
use std::sync::Arc;
use std::thread;
use types::{BeaconState, EthSpec, Hash256, Slot};
/// Trait for migration processes that update the database upon finalization.
pub trait Migrate<S, E: EthSpec>: Send + Sync + 'static {
fn new(db: Arc<S>) -> Self;
fn freeze_to_state(
&self,
_state_root: Hash256,
_state: BeaconState<E>,
_max_finality_distance: u64,
) {
}
}
/// Migrator that does nothing, for stores that don't need migration.
pub struct NullMigrator;
impl<E: EthSpec> Migrate<SimpleDiskStore, E> for NullMigrator {
fn new(_: Arc<SimpleDiskStore>) -> Self {
NullMigrator
}
}
impl<E: EthSpec> Migrate<MemoryStore, E> for NullMigrator {
fn new(_: Arc<MemoryStore>) -> Self {
NullMigrator
}
}
/// Migrator that immediately calls the store's migration function, blocking the current execution.
///
/// Mostly useful for tests.
pub struct BlockingMigrator<S>(Arc<S>);
impl<E: EthSpec, S: Store> Migrate<S, E> for BlockingMigrator<S> {
fn new(db: Arc<S>) -> Self {
BlockingMigrator(db)
}
fn freeze_to_state(
&self,
state_root: Hash256,
state: BeaconState<E>,
_max_finality_distance: u64,
) {
if let Err(e) = S::freeze_to_state(self.0.clone(), state_root, &state) {
// This migrator is only used for testing, so we just log to stderr without a logger.
eprintln!("Migration error: {:?}", e);
}
}
}
/// Migrator that runs a background thread to migrate state from the hot to the cold database.
pub struct BackgroundMigrator<E: EthSpec> {
db: Arc<DiskStore>,
tx_thread: Mutex<(
mpsc::Sender<(Hash256, BeaconState<E>)>,
thread::JoinHandle<()>,
)>,
}
impl<E: EthSpec> Migrate<DiskStore, E> for BackgroundMigrator<E> {
fn new(db: Arc<DiskStore>) -> Self {
let tx_thread = Mutex::new(Self::spawn_thread(db.clone()));
Self { db, tx_thread }
}
/// Perform the freezing operation on the database,
fn freeze_to_state(
&self,
finalized_state_root: Hash256,
finalized_state: BeaconState<E>,
max_finality_distance: u64,
) {
if !self.needs_migration(finalized_state.slot, max_finality_distance) {
return;
}
let (ref mut tx, ref mut thread) = *self.tx_thread.lock();
if let Err(tx_err) = tx.send((finalized_state_root, finalized_state)) {
let (new_tx, new_thread) = Self::spawn_thread(self.db.clone());
drop(mem::replace(tx, new_tx));
let old_thread = mem::replace(thread, new_thread);
// Join the old thread, which will probably have panicked, or may have
// halted normally just now as a result of us dropping the old `mpsc::Sender`.
if let Err(thread_err) = old_thread.join() {
warn!(
self.db.log,
"Migration thread died, so it was restarted";
"reason" => format!("{:?}", thread_err)
);
}
// Retry at most once, we could recurse but that would risk overflowing the stack.
let _ = tx.send(tx_err.0);
}
}
}
impl<E: EthSpec> BackgroundMigrator<E> {
/// Return true if a migration needs to be performed, given a new `finalized_slot`.
fn needs_migration(&self, finalized_slot: Slot, max_finality_distance: u64) -> bool {
let finality_distance = finalized_slot - self.db.get_split_slot();
finality_distance > max_finality_distance
}
/// Spawn a new child thread to run the migration process.
///
/// Return a channel handle for sending new finalized states to the thread.
fn spawn_thread(
db: Arc<DiskStore>,
) -> (
mpsc::Sender<(Hash256, BeaconState<E>)>,
thread::JoinHandle<()>,
) {
let (tx, rx) = mpsc::channel();
let thread = thread::spawn(move || {
while let Ok((state_root, state)) = rx.recv() {
if let Err(e) = DiskStore::freeze_to_state(db.clone(), state_root, &state) {
warn!(
db.log,
"Database migration failed";
"error" => format!("{:?}", e)
);
}
}
});
(tx, thread)
}
}

217
beacon_node/store/src/partial_beacon_state.rs Normal file
View File

@@ -0,0 +1,217 @@
use crate::chunked_vector::{
load_variable_list_from_db, load_vector_from_db, BlockRoots, HistoricalRoots, RandaoMixes,
StateRoots,
};
use crate::{Error, Store};
use ssz_derive::{Decode, Encode};
use std::convert::TryInto;
use types::*;
/// Lightweight variant of the `BeaconState` that is stored in the database.
///
/// Utilises lazy-loading from separate storage for its vector fields.
///
/// Spec v0.9.1
#[derive(Debug, PartialEq, Clone, Encode, Decode)]
pub struct PartialBeaconState<T>
where
T: EthSpec,
{
// Versioning
pub genesis_time: u64,
pub slot: Slot,
pub fork: Fork,
// History
pub latest_block_header: BeaconBlockHeader,
#[ssz(skip_serializing)]
#[ssz(skip_deserializing)]
pub block_roots: Option<FixedVector<Hash256, T::SlotsPerHistoricalRoot>>,
#[ssz(skip_serializing)]
#[ssz(skip_deserializing)]
pub state_roots: Option<FixedVector<Hash256, T::SlotsPerHistoricalRoot>>,
#[ssz(skip_serializing)]
#[ssz(skip_deserializing)]
pub historical_roots: Option<VariableList<Hash256, T::HistoricalRootsLimit>>,
// Ethereum 1.0 chain data
pub eth1_data: Eth1Data,
pub eth1_data_votes: VariableList<Eth1Data, T::SlotsPerEth1VotingPeriod>,
pub eth1_deposit_index: u64,
// Registry
pub validators: VariableList<Validator, T::ValidatorRegistryLimit>,
pub balances: VariableList<u64, T::ValidatorRegistryLimit>,
// Shuffling
/// Randao value from the current slot, for patching into the per-epoch randao vector.
pub latest_randao_value: Hash256,
#[ssz(skip_serializing)]
#[ssz(skip_deserializing)]
pub randao_mixes: Option<FixedVector<Hash256, T::EpochsPerHistoricalVector>>,
// Slashings
slashings: FixedVector<u64, T::EpochsPerSlashingsVector>,
// Attestations
pub previous_epoch_attestations: VariableList<PendingAttestation<T>, T::MaxPendingAttestations>,
pub current_epoch_attestations: VariableList<PendingAttestation<T>, T::MaxPendingAttestations>,
// Finality
pub justification_bits: BitVector<T::JustificationBitsLength>,
pub previous_justified_checkpoint: Checkpoint,
pub current_justified_checkpoint: Checkpoint,
pub finalized_checkpoint: Checkpoint,
}
impl<T: EthSpec> PartialBeaconState<T> {
/// Convert a `BeaconState` to a `PartialBeaconState`, while dropping the optional fields.
pub fn from_state_forgetful(s: &BeaconState<T>) -> Self {
// TODO: could use references/Cow for fields to avoid cloning
PartialBeaconState {
genesis_time: s.genesis_time,
slot: s.slot,
fork: s.fork.clone(),
// History
latest_block_header: s.latest_block_header.clone(),
block_roots: None,
state_roots: None,
historical_roots: None,
// Eth1
eth1_data: s.eth1_data.clone(),
eth1_data_votes: s.eth1_data_votes.clone(),
eth1_deposit_index: s.eth1_deposit_index,
// Validator registry
validators: s.validators.clone(),
balances: s.balances.clone(),
// Shuffling
latest_randao_value: *s
.get_randao_mix(s.current_epoch())
.expect("randao at current epoch is OK"),
randao_mixes: None,
// Slashings
slashings: s.get_all_slashings().to_vec().into(),
// Attestations
previous_epoch_attestations: s.previous_epoch_attestations.clone(),
current_epoch_attestations: s.current_epoch_attestations.clone(),
// Finality
justification_bits: s.justification_bits.clone(),
previous_justified_checkpoint: s.previous_justified_checkpoint.clone(),
current_justified_checkpoint: s.current_justified_checkpoint.clone(),
finalized_checkpoint: s.finalized_checkpoint.clone(),
}
}
pub fn load_block_roots<S: Store>(&mut self, store: &S, spec: &ChainSpec) -> Result<(), Error> {
if self.block_roots.is_none() {
self.block_roots = Some(load_vector_from_db::<BlockRoots, T, _>(
store, self.slot, spec,
)?);
}
Ok(())
}
pub fn load_state_roots<S: Store>(&mut self, store: &S, spec: &ChainSpec) -> Result<(), Error> {
if self.state_roots.is_none() {
self.state_roots = Some(load_vector_from_db::<StateRoots, T, _>(
store, self.slot, spec,
)?);
}
Ok(())
}
pub fn load_historical_roots<S: Store>(
&mut self,
store: &S,
spec: &ChainSpec,
) -> Result<(), Error> {
if self.historical_roots.is_none() {
self.historical_roots = Some(load_variable_list_from_db::<HistoricalRoots, T, _>(
store, self.slot, spec,
)?);
}
Ok(())
}
pub fn load_randao_mixes<S: Store>(
&mut self,
store: &S,
spec: &ChainSpec,
) -> Result<(), Error> {
if self.randao_mixes.is_none() {
// Load the per-epoch values from the database
let mut randao_mixes =
load_vector_from_db::<RandaoMixes, T, _>(store, self.slot, spec)?;
// Patch the value for the current slot into the index for the current epoch
let current_epoch = self.slot.epoch(T::slots_per_epoch());
let len = randao_mixes.len();
randao_mixes[current_epoch.as_usize() % len] = self.latest_randao_value;
self.randao_mixes = Some(randao_mixes)
}
Ok(())
}
}
impl<E: EthSpec> TryInto<BeaconState<E>> for PartialBeaconState<E> {
type Error = Error;
fn try_into(self) -> Result<BeaconState<E>, Error> {
fn unpack<T>(x: Option<T>) -> Result<T, Error> {
x.ok_or(Error::PartialBeaconStateError)
}
Ok(BeaconState {
genesis_time: self.genesis_time,
slot: self.slot,
fork: self.fork,
// History
latest_block_header: self.latest_block_header,
block_roots: unpack(self.block_roots)?,
state_roots: unpack(self.state_roots)?,
historical_roots: unpack(self.historical_roots)?,
// Eth1
eth1_data: self.eth1_data,
eth1_data_votes: self.eth1_data_votes,
eth1_deposit_index: self.eth1_deposit_index,
// Validator registry
validators: self.validators,
balances: self.balances,
// Shuffling
randao_mixes: unpack(self.randao_mixes)?,
// Slashings
slashings: self.slashings,
// Attestations
previous_epoch_attestations: self.previous_epoch_attestations,
current_epoch_attestations: self.current_epoch_attestations,
// Finality
justification_bits: self.justification_bits,
previous_justified_checkpoint: self.previous_justified_checkpoint,
current_justified_checkpoint: self.current_justified_checkpoint,
finalized_checkpoint: self.finalized_checkpoint,
// Caching
committee_caches: <_>::default(),
pubkey_cache: <_>::default(),
exit_cache: <_>::default(),
tree_hash_cache: <_>::default(),
})
}
}