Skip to content

Commit 17af7c3

Browse files
authored
perf(storage): borrow values on read, make has_state skip snapshot reads, keep states in blob files (#658)
## Motivation On the Hoodi follower, the import guard's `has_state(parent)` check costs 0-1 ms on most blocks but **p50 180 ms, p90 332 ms, max 536 ms** on the block after each epoch's first block (48h of import timings, 10-03 to 10-05): | slot % 32 | `guards` p50 | `guards` p90 | |---|---|---| | 0 | 0 ms | 1 ms | | **1** | **180 ms** | **332 ms** | | 2 | 0 ms | 177 ms (when slot 1 is empty) | | 3..31 | 0 ms | 1-8 ms | An epoch-crossing block's state is stored as a full snapshot only, and `has_state` did `get(States)` on it: RocksDB read and copied the whole ~150 MB state to answer a yes/no question, although that state was still in the store's state cache. On Plataberget the same function costs ~30 ms on every block, since `States` has no bloom filter and a miss still reads a snapshot-sized data block per L0 file. ## Changes 1. **`has_state` is an existence check.** It consults the state cache (`peek`, so the LRU order is untouched), then `StateDiffs` before `States`, through `contains`, and never copies a value. A cached `BlockState` always means the state is pending or persisted, and nothing deletes a persisted state, so the answer is unchanged. 2. **`StorageReadView::read` is the one required lookup.** It lends the value to a `&mut dyn FnMut(&[u8])` callback, which keeps the trait usable as `dyn StorageReadView`. `get` and `contains` are default methods on top of it, so both backends implement `read` only. RocksDB uses `get_pinned_cf`. 3. **Decodes read from the backend's buffer.** `StorageReadViewExt::read_with` decodes inside the borrow, and every get-then-decode site uses it. Beacon state reconstruction folds its deltas while the snapshot is still borrowed and hands the result over as a `Cow`: a snapshot root decodes straight from RocksDB's buffer, and the writer's parent-bytes path takes ownership without a second copy. `get(..).is_some()` checks became `contains`. `get` stays where the bytes outlive the read (the beacon walk's delta records, the pending-column take) and for the stored config, which must not be decoded before the version and preset checks. 4. **`States` and `StateDiffs` use RocksDB blob files.** | setting | value | why | |---|---|---| | `min_blob_size` | 4 KiB | one default data block; anything larger gets an oversized block anyway | | blob compression | LZ4 | blob files default to none, while SST blocks get Snappy; the values are raw SSZ | | blob GC | off | nothing deletes or overwrites a state, so GC would only relocate live blobs | | blob cache | none | the store caches decoded states, and a snapshot-sized entry would evict the whole block cache | ## Compatibility No `DB_VERSION` bump. RocksDB applies the blob options to an existing directory as it goes: new writes land in blob files, and inline values move out as compaction rewrites their SSTs. `a_directory_written_without_blob_files_still_reads` opens a directory written with plain options and reads both old inline and new blob values. `estimate_table_bytes` now adds `rocksdb.live-blob-file-size`, which `estimate-live-data-size` leaves out, so the table-size metric keeps counting state bytes. ## Testing - New tests: `has_state` on cached, diff, snapshot-only and absent roots, through a counting backend that proves the cached path reads neither table; `read` on both backends (value, absent key, callback error); a cold-cache beacon reconstruction across a whole snapshot interval; blob placement and the pre-blob directory. - `cargo test --workspace --profile release-fast --lib --bins`: 0 failures. Clippy `-D warnings` clean. - Spec-test suites not run locally. ## Not covered - A bloom filter on `States`: blob files make misses cheap here, but other large-value tables still have none. - The live effect on `guards`: worth re-measuring on a follower after deploy.
1 parent ffac634 commit 17af7c3

9 files changed

Lines changed: 714 additions & 154 deletions

File tree

‎crates/storage/src/api/mod.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,4 +17,6 @@ mod tables;
1717
mod traits;
1818

1919
pub use tables::{ALL_TABLES, Table};
20-
pub use traits::{Error, PrefixResult, StorageBackend, StorageReadView, StorageWriteBatch};
20+
pub use traits::{
21+
Error, PrefixResult, StorageBackend, StorageReadView, StorageReadViewExt, StorageWriteBatch,
22+
};

‎crates/storage/src/api/traits.rs‎

Lines changed: 59 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,38 @@ pub trait StorageBackend: Send + Sync {
2424

2525
/// A read-only view of the storage.
2626
pub trait StorageReadView {
27-
/// Get a value by key from a table.
28-
fn get(&self, table: Table, key: &[u8]) -> Result<Option<Vec<u8>>, Error>;
27+
/// Calls `read_fn` with a borrow of the value stored under `key`, if any,
28+
/// and returns whether the key was present.
29+
///
30+
/// The value is never copied: `read_fn` sees the backend's own buffer, so
31+
/// a caller that only decodes or inspects the bytes avoids allocating a
32+
/// value-sized `Vec` (full state snapshots are 100+ MB on mainnet-sized
33+
/// beacon chains). `read_fn` runs at most once, and its error is returned
34+
/// as-is. A `&mut dyn FnMut` rather than a generic closure, so the trait
35+
/// stays usable as `dyn StorageReadView`.
36+
fn read(
37+
&self,
38+
table: Table,
39+
key: &[u8],
40+
read_fn: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
41+
) -> Result<bool, Error>;
42+
43+
/// Get a value by key from a table, copied into an owned `Vec`.
44+
///
45+
/// Prefer [`StorageReadViewExt::read_with`] when the bytes are only
46+
/// decoded: it decodes from the backend's buffer without the copy.
47+
fn get(&self, table: Table, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
48+
self.read_with(table, key, <[u8]>::to_vec)
49+
}
50+
51+
/// Whether `key` is present in a table.
52+
///
53+
/// Same answer as `get(..)?.is_some()` but never materializes the value, so
54+
/// the cost does not scale with its size. Prefer it for pure existence
55+
/// checks on large values.
56+
fn contains(&self, table: Table, key: &[u8]) -> Result<bool, Error> {
57+
self.read(table, key, &mut |_| Ok(()))
58+
}
2959

3060
/// Iterate over all entries with a given key prefix.
3161
fn prefix_iterator(
@@ -35,6 +65,33 @@ pub trait StorageReadView {
3565
) -> Result<Box<dyn Iterator<Item = PrefixResult> + '_>, Error>;
3666
}
3767

68+
/// Generic conveniences over [`StorageReadView::read`].
69+
///
70+
/// A separate trait because its methods are generic, which would make
71+
/// `StorageReadView` itself unusable as a trait object. The blanket impl
72+
/// covers every view, `dyn StorageReadView` included.
73+
pub trait StorageReadViewExt: StorageReadView {
74+
/// Decodes the value stored under `key` straight from the backend's
75+
/// buffer, without copying it first. `None` when the key is absent.
76+
fn read_with<T>(
77+
&self,
78+
table: Table,
79+
key: &[u8],
80+
decode: impl FnOnce(&[u8]) -> T,
81+
) -> Result<Option<T>, Error> {
82+
let mut decode = Some(decode);
83+
let mut value = None;
84+
self.read(table, key, &mut |bytes| {
85+
let decode = decode.take().expect("read calls read_fn at most once");
86+
value = Some(decode(bytes));
87+
Ok(())
88+
})?;
89+
Ok(value)
90+
}
91+
}
92+
93+
impl<V: StorageReadView + ?Sized> StorageReadViewExt for V {}
94+
3895
/// A write batch that can be committed atomically.
3996
pub trait StorageWriteBatch: Send {
4097
/// Put multiple key-value pairs into a table.

‎crates/storage/src/backend/in_memory.rs‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -66,13 +66,17 @@ struct InMemoryReadView<'a> {
6666
}
6767

6868
impl StorageReadView for InMemoryReadView<'_> {
69-
fn get(&self, table: Table, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
70-
Ok(self
71-
.guard
72-
.get(&table)
73-
.expect("table exists")
74-
.get(key)
75-
.cloned())
69+
fn read(
70+
&self,
71+
table: Table,
72+
key: &[u8],
73+
read_fn: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
74+
) -> Result<bool, Error> {
75+
let Some(value) = self.guard.get(&table).expect("table exists").get(key) else {
76+
return Ok(false);
77+
};
78+
read_fn(value)?;
79+
Ok(true)
7680
}
7781

7882
fn prefix_iterator(

‎crates/storage/src/backend/rocksdb.rs‎

Lines changed: 140 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,8 @@ use crate::api::{
44
ALL_TABLES, Error, PrefixResult, StorageBackend, StorageReadView, StorageWriteBatch, Table,
55
};
66
use rocksdb::{
7-
BlockBasedOptions, Cache, ColumnFamilyDescriptor, DBWithThreadMode, MultiThreaded, Options,
8-
WriteBatch, WriteOptions,
7+
BlockBasedOptions, Cache, ColumnFamilyDescriptor, DBCompressionType, DBWithThreadMode,
8+
MultiThreaded, Options, WriteBatch, WriteOptions,
99
};
1010
use std::path::Path;
1111
use std::sync::Arc;
@@ -18,6 +18,43 @@ fn cf_name(table: Table) -> &'static str {
1818
table.name()
1919
}
2020

21+
/// The smallest value a blob-file table stores out of line.
22+
///
23+
/// RocksDB's default data block size: a larger value would get an oversized
24+
/// block of its own anyway, so it gains nothing from staying inline.
25+
const MIN_BLOB_SIZE: u64 = 4 * 1024;
26+
27+
/// Whether a table keeps its values in blob files rather than inline in SSTs.
28+
///
29+
/// `States` holds full state snapshots (100+ MB on mainnet-sized beacon
30+
/// chains) and `StateDiffs` the deltas between them (hundreds of KB). Inline,
31+
/// every compaction that touches a file rewrites those values, and a lookup
32+
/// that misses still reads the data block around the key, which here can be
33+
/// a whole snapshot. In a blob file a value is written once, and the SSTs
34+
/// hold only small references to it.
35+
fn stores_values_in_blob_files(table: Table) -> bool {
36+
matches!(table, Table::States | Table::StateDiffs)
37+
}
38+
39+
/// Moves a column family's large values into blob files.
40+
///
41+
/// RocksDB applies the change to an existing database as it goes: new
42+
/// writes land in blob files, and inline values move out as compaction
43+
/// rewrites their SSTs. So a data directory written without it opens
44+
/// unchanged, and no `DB_VERSION` bump is needed.
45+
fn enable_blob_files(cf_opts: &mut Options) {
46+
cf_opts.set_enable_blob_files(true);
47+
cf_opts.set_min_blob_size(MIN_BLOB_SIZE);
48+
// SST blocks get RocksDB's default Snappy, but blob files default to no
49+
// compression, so moving the values out would otherwise grow the tables
50+
// on disk. The values are raw SSZ.
51+
cf_opts.set_blob_compression_type(DBCompressionType::Lz4);
52+
// No blob garbage collection: nothing deletes or overwrites a state, so it
53+
// would only relocate live blobs during compaction. Revisit if states are
54+
// ever pruned. No blob cache either: the store caches decoded states
55+
// itself, and a snapshot-sized entry would evict the whole block cache.
56+
}
57+
2158
/// RocksDB storage backend.
2259
#[derive(Clone)]
2360
pub struct RocksDBBackend {
@@ -52,6 +89,9 @@ impl RocksDBBackend {
5289
.map(|t| {
5390
let mut cf_opts = Options::default();
5491
cf_opts.set_block_based_table_factory(&block_opts);
92+
if stores_values_in_blob_files(*t) {
93+
enable_blob_files(&mut cf_opts);
94+
}
5595
ColumnFamilyDescriptor::new(cf_name(*t), cf_opts)
5696
})
5797
.collect();
@@ -93,7 +133,15 @@ impl StorageBackend for RocksDBBackend {
93133
.ok()
94134
.flatten()
95135
.unwrap_or(0);
96-
sst_bytes + memtable_bytes
136+
// `estimate-live-data-size` counts SST files only, so a blob-file
137+
// table's values would otherwise vanish from the estimate.
138+
let blob_bytes = self
139+
.db
140+
.property_int_value_cf(&cf, "rocksdb.live-blob-file-size")
141+
.ok()
142+
.flatten()
143+
.unwrap_or(0);
144+
sst_bytes + memtable_bytes + blob_bytes
97145
}
98146
}
99147

@@ -103,13 +151,24 @@ struct RocksDBReadView {
103151
}
104152

105153
impl StorageReadView for RocksDBReadView {
106-
fn get(&self, table: Table, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
154+
fn read(
155+
&self,
156+
table: Table,
157+
key: &[u8],
158+
read_fn: &mut dyn FnMut(&[u8]) -> Result<(), Error>,
159+
) -> Result<bool, Error> {
107160
let cf = self
108161
.db
109162
.cf_handle(cf_name(table))
110163
.ok_or_else(|| format!("Column family {} not found", cf_name(table)))?;
111164

112-
Ok(self.db.get_cf(&cf, key)?)
165+
// Pinned: references RocksDB's own buffer instead of copying the
166+
// value into a `Vec`.
167+
let Some(value) = self.db.get_pinned_cf(&cf, key)? else {
168+
return Ok(false);
169+
};
170+
read_fn(&value)?;
171+
Ok(true)
113172
}
114173

115174
fn prefix_iterator(
@@ -200,6 +259,82 @@ mod tests {
200259
run_backend_tests(&backend);
201260
}
202261

262+
/// A value big enough for a blob file, and the property counting them.
263+
const BLOB_VALUE_LEN: usize = 64 * 1024;
264+
const NUM_BLOB_FILES: &str = "rocksdb.num-blob-files";
265+
266+
fn num_blob_files(backend: &RocksDBBackend, table: Table) -> u64 {
267+
let cf = backend.db.cf_handle(cf_name(table)).unwrap();
268+
backend
269+
.db
270+
.property_int_value_cf(&cf, NUM_BLOB_FILES)
271+
.unwrap()
272+
.unwrap()
273+
}
274+
275+
fn put_and_flush(backend: &RocksDBBackend, table: Table, key: &[u8], value: Vec<u8>) {
276+
let mut batch = backend.begin_write().unwrap();
277+
batch.put_batch(table, vec![(key.to_vec(), value)]).unwrap();
278+
batch.commit().unwrap();
279+
let cf = backend.db.cf_handle(cf_name(table)).unwrap();
280+
backend.db.flush_cf(&cf).unwrap();
281+
}
282+
283+
#[test]
284+
fn large_state_values_live_in_blob_files() {
285+
let dir = tempdir().unwrap();
286+
let backend = RocksDBBackend::open(dir.path()).unwrap();
287+
let value: Vec<u8> = (0..BLOB_VALUE_LEN).map(|i| (i % 251) as u8).collect();
288+
289+
for table in [Table::States, Table::StateDiffs] {
290+
put_and_flush(&backend, table, b"big", value.clone());
291+
assert_eq!(num_blob_files(&backend, table), 1, "{table:?}");
292+
293+
let view = backend.begin_read().unwrap();
294+
assert_eq!(view.get(table, b"big").unwrap(), Some(value.clone()));
295+
assert!(view.contains(table, b"big").unwrap());
296+
assert!(!view.contains(table, b"absent").unwrap());
297+
}
298+
assert!(backend.estimate_table_bytes(Table::States) > 0);
299+
300+
// Small values, and every other table, stay inline.
301+
put_and_flush(&backend, Table::States, b"small", vec![7; 16]);
302+
assert_eq!(num_blob_files(&backend, Table::States), 1);
303+
put_and_flush(&backend, Table::BlockHeaders, b"big", value);
304+
assert_eq!(num_blob_files(&backend, Table::BlockHeaders), 0);
305+
}
306+
307+
#[test]
308+
fn a_directory_written_without_blob_files_still_reads() {
309+
let dir = tempdir().unwrap();
310+
let value: Vec<u8> = (0..BLOB_VALUE_LEN).map(|i| (i % 251) as u8).collect();
311+
312+
// The layout a data directory had before blob files: every table
313+
// with plain options.
314+
{
315+
let mut opts = Options::default();
316+
opts.create_if_missing(true);
317+
opts.create_missing_column_families(true);
318+
let cfs = ALL_TABLES.iter().map(|t| cf_name(*t));
319+
let db = DBWithThreadMode::<MultiThreaded>::open_cf(&opts, dir.path(), cfs).unwrap();
320+
let cf = db.cf_handle(cf_name(Table::States)).unwrap();
321+
db.put_cf(&cf, b"old", &value).unwrap();
322+
db.flush_cf(&cf).unwrap();
323+
}
324+
325+
let backend = RocksDBBackend::open(dir.path()).unwrap();
326+
assert_eq!(num_blob_files(&backend, Table::States), 0);
327+
put_and_flush(&backend, Table::States, b"new", value.clone());
328+
assert_eq!(num_blob_files(&backend, Table::States), 1);
329+
330+
let view = backend.begin_read().unwrap();
331+
assert_eq!(
332+
view.get(Table::States, b"old").unwrap(),
333+
Some(value.clone())
334+
);
335+
assert_eq!(view.get(Table::States, b"new").unwrap(), Some(value));
336+
}
337+
203338
#[test]
204339
fn test_persistence() {
205340
let dir = tempdir().unwrap();

‎crates/storage/src/backend/tests.rs‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ pub fn run_backend_tests(backend: &dyn StorageBackend) {
1717
test_delete(backend);
1818
test_prefix_iterator(backend);
1919
test_nonexistent_key(backend);
20+
test_contains(backend);
21+
test_read(backend);
2022
test_delete_then_put(backend);
2123
test_put_then_delete(backend);
2224
test_delete_range(backend);
@@ -45,6 +47,72 @@ fn test_put_and_get(backend: &dyn StorageBackend) {
4547
}
4648
}
4749

50+
fn test_contains(backend: &dyn StorageBackend) {
51+
{
52+
let mut batch = backend.begin_write().unwrap();
53+
batch
54+
.put_batch(
55+
Table::BlockHeaders,
56+
vec![(b"test_contains_key".to_vec(), b"value1".to_vec())],
57+
)
58+
.unwrap();
59+
batch.commit().unwrap();
60+
}
61+
let view = backend.begin_read().unwrap();
62+
assert!(
63+
view.contains(Table::BlockHeaders, b"test_contains_key")
64+
.unwrap()
65+
);
66+
assert!(
67+
!view
68+
.contains(Table::BlockHeaders, b"test_contains_missing")
69+
.unwrap()
70+
);
71+
}
72+
73+
fn test_read(backend: &dyn StorageBackend) {
74+
{
75+
let mut batch = backend.begin_write().unwrap();
76+
batch
77+
.put_batch(
78+
Table::BlockHeaders,
79+
vec![(b"test_read_key".to_vec(), b"value1".to_vec())],
80+
)
81+
.unwrap();
82+
batch.commit().unwrap();
83+
}
84+
let view = backend.begin_read().unwrap();
85+
86+
let mut seen = Vec::new();
87+
let found = view
88+
.read(Table::BlockHeaders, b"test_read_key", &mut |bytes| {
89+
seen.push(bytes.to_vec());
90+
Ok(())
91+
})
92+
.unwrap();
93+
assert!(found);
94+
assert_eq!(seen, vec![b"value1".to_vec()]);
95+
96+
// A missing key never reaches the callback.
97+
let mut calls = 0;
98+
let found = view
99+
.read(Table::BlockHeaders, b"test_read_missing", &mut |_| {
100+
calls += 1;
101+
Ok(())
102+
})
103+
.unwrap();
104+
assert!(!found);
105+
assert_eq!(calls, 0);
106+
107+
// The callback's own error is what `read` returns.
108+
let err = view
109+
.read(Table::BlockHeaders, b"test_read_key", &mut |_| {
110+
Err("decode failed".into())
111+
})
112+
.unwrap_err();
113+
assert_eq!(err.to_string(), "decode failed");
114+
}
115+
48116
fn test_delete(backend: &dyn StorageBackend) {
49117
// Write data
50118
{

‎crates/storage/src/lib.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,9 @@ mod state_diff;
99
mod state_writer;
1010
mod store;
1111

12-
pub use api::{ALL_TABLES, StorageBackend, StorageReadView, StorageWriteBatch, Table};
12+
pub use api::{
13+
ALL_TABLES, StorageBackend, StorageReadView, StorageReadViewExt, StorageWriteBatch, Table,
14+
};
1315
pub use committee_cache::{CommitteeCache, Lookup, ShufflingKey};
1416
/// Error type returned by the fallible [`Store`] operations, exported so
1517
/// callers can match on it (e.g. to distinguish [`Error::DbVersionMismatch`]).

0 commit comments

Comments
 (0)