diff --git a/CHANGELOG.md b/CHANGELOG.md index e4b84cadfde..642a154cfdb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ ### 2026-08-04 - Cut the cost of a cold contract-code access: store jump destinations as a 1-bit-per-byte bitmap instead of a persisted RLP list of `u32` offsets, count the bytecode in the code cache's byte budget, answer `EXTCODESIZE` from the code-length table instead of materializing the bytecode, and give the account-code column families a bloom filter (4KB data blocks on the blob-backed bytecode CF). Raises the code cache's byte budget from an effective 64 MiB of jump tables to 256 MiB of bytecode, and bumps the store schema version so an older binary warns rather than failing on the new value format. `COLD_ACCOUNT_CODE_ACCESS` drops from 7736 to 4652 gas in the EIP-8038 repricing fit, and `COLD_ACCOUNT_CODE_WRITE` from 10415 to 6355 [#7095](https://github.com/lambdaclass/ethrex/pull/7095) +- Batch and stream the BAL contract-code prefetch: warm accounts and their code in chunks instead of reading every access-list account before the first bytecode, take code hashes from the account read rather than a second lookup per account, and add a batched bytecode read that resolves the buffer and code cache first, then either fans out parallel point gets or shards the remainder across concurrent `multi_get`s, whichever reaches the greater read queue depth for the batch size on this host [#7099](https://github.com/lambdaclass/ethrex/pull/7099) ### 2026-07-22 diff --git a/crates/blockchain/vm.rs b/crates/blockchain/vm.rs index 18985c112f2..bf4d3a114d3 100644 --- a/crates/blockchain/vm.rs +++ b/crates/blockchain/vm.rs @@ -130,6 +130,10 @@ impl StoreVmDatabase { } impl VmDatabase for StoreVmDatabase { + fn code_cache_budget_bytes(&self) -> u64 { + self.store.code_cache_budget_bytes() + } + #[instrument( level = "trace", name = "Account read", @@ -361,6 +365,44 @@ impl VmDatabase for StoreVmDatabase { } } + #[instrument( + level = "trace", + name = "Account codes batch read", + skip_all, + fields(namespace = "block_execution") + )] + fn get_account_codes_batch(&self, code_hashes: &[H256]) -> Result>, EvmError> { + // The empty hash is answered here rather than sent to the store, matching + // `get_account_code`, so a batch containing EOAs does not fault on it. + let to_read: Vec = code_hashes + .iter() + .copied() + .filter(|h| *h != *EMPTY_KECCAK_HASH) + .collect(); + let read = self + .store + .get_account_codes_batch(&to_read) + .map_err(|e| EvmError::DB(e.to_string()))?; + + let mut by_hash: FxHashMap = FxHashMap::default(); + for (hash, code) in to_read.iter().zip(read.into_iter()) { + if let Some(code) = code { + by_hash.insert(*hash, code); + } + } + + Ok(code_hashes + .iter() + .map(|h| { + if *h == *EMPTY_KECCAK_HASH { + Some(Code::default()) + } else { + by_hash.get(h).cloned() + } + }) + .collect()) + } + #[instrument( level = "trace", name = "Code metadata read", diff --git a/crates/storage/store.rs b/crates/storage/store.rs index 6b970cb1f46..25db6618184 100644 --- a/crates/storage/store.rs +++ b/crates/storage/store.rs @@ -923,6 +923,19 @@ impl Store { .map_err(StoreError::from) } + /// Capacity of the bytecode cache, in [`Code::size`] bytes, which is how much + /// bytecode this node is configured to keep resident and so bounds what a block warm + /// reads speculatively. + /// + /// A poisoned cache reports zero, which stops warming rather than warming into a + /// cache no reader can reach. + pub fn code_cache_budget_bytes(&self) -> u64 { + self.account_code_cache + .lock() + .map(|cache| cache.max_size) + .unwrap_or(0) + } + /// Get account code by its hash. /// /// Checks the in-memory block-data buffer first, then the LRU cache @@ -966,6 +979,154 @@ impl Store { Ok(Some(code)) } + /// Batched [`Self::get_account_code`]. + /// + /// Resolves the buffer and the LRU first, then reads whatever is left by whichever + /// of two strategies gets more of those reads in flight for a batch this size: a + /// parallel fan-out of point gets, or sorted keys split into contiguous shards read + /// concurrently. See the comment on the read below for how the choice is made. The + /// LRU is locked once for the whole batch rather than twice per code. + /// + /// Results are returned in the order of `code_hashes`. Duplicate hashes are read + /// once. `None` means the hash is absent from the database. + pub fn get_account_codes_batch( + &self, + code_hashes: &[H256], + ) -> Result>, StoreError> { + let mut out: Vec> = vec![None; code_hashes.len()]; + // Positions to fill per distinct hash, so a repeated hash costs one read. + let mut pending: HashMap> = HashMap::new(); + + { + let buffer = self.buffer()?; + let mut cache = self + .account_code_cache + .lock() + .map_err(|_| StoreError::LockError)?; + for (i, hash) in code_hashes.iter().enumerate() { + if let Some(code) = buffer.get_code(hash) { + out[i] = Some(code); + } else if let Some(code) = cache.get(hash)? { + out[i] = Some(code); + } else { + pending.entry(*hash).or_default().push(i); + } + } + } + + if pending.is_empty() { + return Ok(out); + } + + let mut missing: Vec = pending.keys().copied().collect(); + missing.sort_unstable(); + + // Cold blob reads here are latency-bound, so what matters is how many are in + // flight. Two ways to get there, and which one wins depends on the batch size: + // + // * A parallel fan-out of point gets reaches queue depth ~= core count. Cheap + // for any batch: rayon's pool is already warm. + // * Contiguous shards of the SORTED keys, one blocking thread each, reach queue + // depth ~= shard count and share RocksDB blocks within a shard. This is the + // only way past core count (async_io is OFF in this build, so a single + // `multi_get` runs the whole batch at queue depth 1), but it pays a thread + // spawn per shard. + // + // So shard only once sharding can actually beat the fan-out, i.e. once the batch + // is wide enough to form more shards than there are cores. Below that the + // fan-out is both deeper and cheaper, and a single serial `multi_get` would be + // far worse than either. + // + // The shard cap has to sit above the core count, or on a host with at least that + // many cores the fan-out would always win and this path would be unreachable. + // Twice the cores guarantees that, with a floor so a small host keeps the depth + // it can already reach: these threads block on I/O rather than compute, so more + // of them than cores is the point. + const KEYS_PER_SHARD: usize = 256; + const MIN_SHARD_CAP: usize = 64; + let parallelism = std::thread::available_parallelism().map_or(8, |p| p.get()); + let max_shards = parallelism.saturating_mul(2).max(MIN_SHARD_CAP); + let shards = missing.len().div_ceil(KEYS_PER_SHARD).min(max_shards); + let read_view = self.backend.begin_read()?; + // Both paths decode in whatever thread did the read, so rebuilding the jumpdest + // bitmap for a legacy entry stays off the caller's thread and stays parallel. + let decode = |hash: &H256, value: Option>| -> Result, StoreError> { + let Some(bytes) = value else { return Ok(None) }; + let (bytecode_slice, jumpdests) = decode_bytes(&bytes)?; + Ok(Some(Code::from_parts_unchecked( + *hash, + bytecode_slice, + decode_jumpdests(bytecode_slice, jumpdests)?, + ))) + }; + let decoded: Vec, StoreError>> = if shards > parallelism { + let chunk = missing.len().div_ceil(shards); + let rv = read_view.as_ref(); + let read_shard = |hashes: &[H256]| -> Vec, StoreError>> { + let keys: Vec<&[u8]> = hashes.iter().map(|h| h.as_bytes()).collect(); + rv.multi_get(ACCOUNT_CODES, &keys) + .into_iter() + .zip(hashes.iter()) + .map(|(value, hash)| decode(hash, value?)) + .collect() + }; + std::thread::scope(|scope| { + let handles: Vec<_> = missing + .chunks(chunk) + .map(|ck| (ck, scope.spawn(move || read_shard(ck)))) + .collect(); + handles + .into_iter() + .flat_map(|(shard, handle)| { + // A panicked shard becomes an `Err` per key it covered, keeping + // the results aligned with `missing` and leaving the caller's + // best-effort handling to decide. Re-panicking here would + // escalate a warm into taking down whatever runs it. + handle.join().unwrap_or_else(|_| { + shard + .iter() + .map(|_| { + Err(StoreError::Custom( + "account code shard panicked".to_string(), + )) + }) + .collect() + }) + }) + .collect() + }) + } else { + // One key per read, so `get` rather than a single-key `multi_get`: the + // batched call sets up its own result buffers, which is pure overhead here. + let rv = read_view.as_ref(); + missing + .par_iter() + .map(|hash| decode(hash, rv.get(ACCOUNT_CODES, hash.as_bytes())?)) + .collect() + }; + + let mut fetched: Vec = Vec::new(); + for (hash, code) in missing.iter().zip(decoded.into_iter()) { + let Some(code) = code? else { continue }; + for &i in pending.get(hash).into_iter().flatten() { + out[i] = Some(code.clone()); + } + fetched.push(code); + } + + if !fetched.is_empty() { + let mut cache = self + .account_code_cache + .lock() + .map_err(|_| StoreError::LockError)?; + for code in &fetched { + cache.insert(code)?; + } + } + + Ok(out) + } + /// Check if account code exists by its hash, without constructing the full `Code` struct. /// More efficient than `get_account_code` for existence checks since it skips /// RLP decoding and `Code` struct construction (no jumpdest-bitmap decoding). diff --git a/crates/vm/backends/levm/db.rs b/crates/vm/backends/levm/db.rs index 387b341c32a..236d0fd9585 100644 --- a/crates/vm/backends/levm/db.rs +++ b/crates/vm/backends/levm/db.rs @@ -79,6 +79,28 @@ impl LevmDatabase for DatabaseLogger { self.store.as_ref().get_account_code(code_hash) } + fn code_cache_budget_bytes(&self) -> u64 { + self.store.as_ref().code_cache_budget_bytes() + } + + fn prefetch_codes(&self, code_hashes: &[CoreH256]) -> Result { + // Record before delegating, exactly as the per-hash path does, so a batched + // read cannot leave the witness short of a bytecode it observed. + { + let mut code_accessed = self + .code_accessed + .lock() + .map_err(|_| DatabaseError::Custom("Could not lock mutex".to_string()))?; + code_accessed.extend( + code_hashes + .iter() + .filter(|h| **h != *EMPTY_KECCAK_HASH) + .copied(), + ); + } + self.store.as_ref().prefetch_codes(code_hashes) + } + fn get_code_metadata(&self, code_hash: CoreH256) -> Result { // A size-only read still observes the bytecode, so the witness must carry it: // EIP-8025 stateless validation recomputes the length from the code itself, and @@ -114,6 +136,10 @@ impl LevmDatabase for DynVmDatabase { .collect()) } + fn code_cache_budget_bytes(&self) -> u64 { + ::code_cache_budget_bytes(self.as_ref()) + } + fn get_storage_value( &self, address: CoreAddress, @@ -153,6 +179,14 @@ impl LevmDatabase for DynVmDatabase { .map_err(|e| DatabaseError::Custom(e.to_string())) } + fn get_account_codes_batch( + &self, + code_hashes: &[CoreH256], + ) -> Result>, DatabaseError> { + ::get_account_codes_batch(self.as_ref(), code_hashes) + .map_err(|e| DatabaseError::Custom(e.to_string())) + } + fn get_code_metadata(&self, code_hash: CoreH256) -> Result { ::get_code_metadata(self.as_ref(), code_hash) .map_err(|e| DatabaseError::Custom(e.to_string())) diff --git a/crates/vm/backends/levm/mod.rs b/crates/vm/backends/levm/mod.rs index aa516c30e11..58e18f196ed 100644 --- a/crates/vm/backends/levm/mod.rs +++ b/crates/vm/backends/levm/mod.rs @@ -67,7 +67,7 @@ use ethrex_levm::{ vm::VM, }; #[cfg(all(feature = "rayon", not(feature = "eip-8025")))] -use rayon::iter::{IntoParallelIterator, IntoParallelRefIterator, ParallelIterator}; +use rayon::iter::{IntoParallelIterator, ParallelIterator}; #[cfg(all(feature = "rayon", not(feature = "eip-8025")))] use rustc_hash::{FxHashMap, FxHashSet}; use std::cmp::min; @@ -2878,36 +2878,79 @@ impl LEVM { return Ok(()); } - // Phase 1: Prefetch all account states — parallel inner fetch + single write-lock. - // This warms the CachingDatabase account cache and the TrieLayerCache with - // state trie nodes. Storage slots are prefetched synchronously before the - // executor starts (see `bal_storage_slots` at the call site), so this warmer - // only needs to cover account states and contract code, which overlap exec. - let account_addresses: Vec
= accounts.iter().map(|ac| ac.address).collect(); - store - .prefetch_accounts(&account_addresses) - .map_err(|e| EvmError::Custom(format!("prefetch_accounts: {e}")))?; + // Accounts and their code are warmed in chunks rather than as two whole-block + // phases. Code hashes come from account states, so a phase that reads every + // address before the first code read leaves the executor faulting in code itself + // for the whole of that phase, however early the warmer finishes in aggregate. + // Chunking starts the code reads after the first slice of addresses instead. + // + // Storage slots are prefetched synchronously before the executor starts (see + // `bal_storage_slots` at the call site), so this warmer covers only account + // states and contract code, which overlap execution. + // + let parallelism = std::thread::available_parallelism().map_or(8, |p| p.get()); + let warm_chunk_accounts = warm_chunk_accounts(parallelism); + + // A BAL carries every accessed address, and it does not say which of them were + // accessed for their bytecode: an opcode reading only account state (BALANCE, + // EXTCODEHASH) leaves the whole of a block's bytecode unread. So an unbounded warm + // can spend the block's entire read bandwidth on bytes nothing consumes, competing + // with the account reads the executor is waiting on, and hold all of it in the + // per-block cache besides. + // + // The bound is the bytecode cache's capacity, which is the one dial already sized + // against host memory. Normal blocks sit orders of magnitude under it and warm + // exactly as they did before; it binds only where a block's distinct bytecode + // outgrows what the node was configured to keep resident at all. + let code_budget_bytes = store.code_cache_budget_bytes(); + let mut code_warmed_bytes = 0u64; + + for chunk in accounts.chunks(warm_chunk_accounts) { + if cancelled.load(Ordering::Relaxed) { + return Ok(()); + } - if cancelled.load(Ordering::Relaxed) { - return Ok(()); - } + let addresses: Vec
= chunk.iter().map(|ac| ac.address).collect(); + // Returns the states as well as caching them, so the code hashes come out + // of this read instead of costing a second lookup per account. + let states = store + .get_account_states_batch(&addresses) + .map_err(|e| EvmError::Custom(format!("get_account_states_batch: {e}")))?; + + // Accounts stay warmed for the rest of the block once the code budget is + // spent: they are a fraction of the bytes, and every opcode reaching an + // address needs its state whether or not it reads the bytecode. + if code_warmed_bytes >= code_budget_bytes { + continue; + } - // Phase 2: Code prefetch — collect code hashes from Phase 1 account states - // (already cached after Phase 1 prefetch), then batch-fetch codes in parallel. - // Uses par_iter for collection since blocks can have thousands of accounts. - let code_hashes: Vec = accounts - .par_iter() - .filter_map(|ac| { - store - .get_account_state(ac.address) - .ok() - .filter(|s| s.code_hash != *EMPTY_KECCAK_HASH) - .map(|s| s.code_hash) - }) - .collect(); - code_hashes.par_iter().for_each(|&h| { - let _ = store.get_account_code(h); - }); + // Distinct hashes only: accounts sharing one bytecode are common, and + // without this the same code would be requested once per account holding it. + let mut code_hashes: Vec = states + .iter() + .map(|s| s.code_hash) + .filter(|h| *h != *EMPTY_KECCAK_HASH) + .collect(); + code_hashes.sort_unstable(); + code_hashes.dedup(); + + for slice in code_hashes.chunks(CODE_SLICE_HASHES) { + if code_warmed_bytes >= code_budget_bytes { + break; + } + + // Re-checked between the two reads, not only per chunk: a cancelled + // warmer should not still issue a slice's worth of code reads. + if cancelled.load(Ordering::Relaxed) { + return Ok(()); + } + + // Best-effort: a code hash present in a BAL but absent from the database + // must not fail the warmer, since the executor reports that itself. + code_warmed_bytes = + code_warmed_bytes.saturating_add(store.prefetch_codes(slice).unwrap_or(0)); + } + } Ok(()) } @@ -4026,6 +4069,196 @@ fn describe_balance_diff(expected: U256, actual: U256) -> String { format!("{sign}{mag_u128} wei") } +/// Addresses per chunk in [`LEVM::warm_block_from_bal`]. +/// +/// Sized to saturate the read depth of the batched code fetch, because on the blocks the +/// warmer targets it is throughput-bound rather than latency-bound: it occupies the whole +/// execution window, so total read concurrency matters more than how soon the first code +/// lands. That fetch shards at 256 keys and caps its shard count at `max(64, 2 * cores)`, +/// and its depth is bounded by the distinct uncached hashes a chunk carries, so a chunk +/// narrower than the cap's worth of keys buys an earlier start by giving up depth for the +/// whole block. +/// +/// The width also routes the account read, which is why it must not shrink: +/// `prefetch_accounts` sends a chunk to the sorted sharded batch only once its cold +/// addresses reach [`BLOATED_BATCH_THRESHOLD`] and to parallel point-gets below that, and +/// on a cold account-heavy block the sharded batch is worth several times the point-gets. +#[cfg(all(feature = "rayon", not(feature = "eip-8025")))] +fn warm_chunk_accounts(parallelism: usize) -> usize { + 256_usize.saturating_mul(parallelism.saturating_mul(2).max(64)) +} + +/// Code hashes per batched fetch in [`LEVM::warm_block_from_bal`]. +/// +/// A chunk's code hashes are fetched in slices rather than in one call, because the chunk +/// has to stay wide for the account batch (see [`warm_chunk_accounts`]) while the byte +/// budget and the cancellation flag are only observed between calls. The slice is +/// therefore what bounds how far a warm can overshoot its budget and how much I/O a +/// cancelled warmer still commits to; a whole chunk of max-size bytecode is an order of +/// magnitude past the budget itself. +#[cfg(all(feature = "rayon", not(feature = "eip-8025")))] +const CODE_SLICE_HASHES: usize = 4096; + +#[cfg(all(test, feature = "rayon", not(feature = "eip-8025")))] +mod warm_budget_tests { + use super::{warm_chunk_accounts, *}; + use ethrex_common::H256; + use ethrex_common::types::{AccountState, CodeMetadata, block_access_list::AccountChanges}; + use ethrex_levm::db::BLOATED_BATCH_THRESHOLD; + use ethrex_levm::errors::DatabaseError; + use std::sync::atomic::AtomicUsize; + + /// A chunk narrower than the threshold would route every account read to the + /// point-gets path and leave the sharded batch permanently unreachable. + #[test] + fn chunk_reaches_the_sharded_account_batch_at_every_core_count() { + for parallelism in [1, 2, 4, 8, 16, 32, 64, 128, 256] { + assert!( + warm_chunk_accounts(parallelism) >= BLOATED_BATCH_THRESHOLD, + "chunk of {} accounts is below the batch threshold {BLOATED_BATCH_THRESHOLD} \ + at {parallelism} cores", + warm_chunk_accounts(parallelism), + ); + } + } + + const CODE_LEN: usize = 1024; + + /// Every account holds a distinct bytecode of [`CODE_LEN`], and both read paths are + /// counted so a test can tell warmed code from warmed account state. + struct CountingStore { + budget: u64, + states_read: AtomicUsize, + codes_read: AtomicUsize, + } + + impl CountingStore { + fn with_budget(budget: u64) -> Self { + Self { + budget, + states_read: AtomicUsize::new(0), + codes_read: AtomicUsize::new(0), + } + } + } + + impl Database for CountingStore { + fn code_cache_budget_bytes(&self) -> u64 { + self.budget + } + fn get_account_state(&self, address: Address) -> Result { + self.states_read.fetch_add(1, Ordering::Relaxed); + Ok(AccountState { + // A distinct hash per address, so no dedup collapses the batch. + code_hash: H256::from_low_u64_be(address.to_low_u64_be() + 1), + ..Default::default() + }) + } + fn get_account_code(&self, hash: H256) -> Result { + self.codes_read.fetch_add(1, Ordering::Relaxed); + Ok(Code::from_bytecode_unchecked( + Bytes::from(vec![0u8; CODE_LEN]), + hash, + )) + } + fn get_storage_value(&self, _: Address, _: H256) -> Result { + Ok(U256::zero()) + } + fn get_block_hash(&self, _: u64) -> Result { + Ok(H256::zero()) + } + fn get_chain_config(&self) -> Result { + Err(DatabaseError::Custom("not implemented".into())) + } + fn get_code_metadata(&self, _: H256) -> Result { + Ok(CodeMetadata { + length: CODE_LEN as u64, + }) + } + } + + fn bal_with_accounts(n: u64) -> BlockAccessList { + BlockAccessList::from_accounts( + (0..n) + .map(|i| AccountChanges::new(Address::from_low_u64_be(i))) + .collect(), + ) + } + + /// Two slices' worth of distinct bytecodes against a budget the first slice already + /// exceeds: the second slice must not be read. + #[test] + fn code_warming_stops_once_the_budget_is_spent() { + let accounts = 2 * super::CODE_SLICE_HASHES; + let bal = bal_with_accounts(accounts as u64); + let store = Arc::new(CountingStore::with_budget(1024 * 1024)); + let counter = store.clone(); + + LEVM::warm_block_from_bal(&bal, store, &AtomicBool::new(false)).expect("warm"); + + assert_eq!( + counter.states_read.load(Ordering::Relaxed), + accounts, + "every account state should still be warmed" + ); + assert_eq!( + counter.codes_read.load(Ordering::Relaxed), + super::CODE_SLICE_HASHES, + "the warm should stop after the slice that spends the budget" + ); + } + + /// The budget must not touch ordinary blocks. A block whose distinct bytecode fits the + /// smallest budget any backend reports warms every code exactly once, in one batch, + /// which is what the warm did before there was a budget at all. + /// + /// A 60M-gas block reaches at most ~23k cold account accesses at 2600 gas each, and a + /// real one spreads those over a few hundred distinct contracts of a few KB: single + /// digit MB against a 64 MiB floor. Only the adversarial shape, every access landing on + /// its own max-size contract, reaches ~567 MB and outgrows the budget, which is the + /// case it exists for and the one where warming all of it was measured to buy nothing. + #[test] + fn a_normal_block_warms_every_bytecode() { + const MIN_BUDGET: u64 = 64 * 1024 * 1024; + let accounts = 2048; + assert!( + accounts < super::CODE_SLICE_HASHES, + "a normal block must fit one batch" + ); + assert!( + (accounts as u64).saturating_mul(CODE_LEN as u64) < MIN_BUDGET, + "a normal block must fit the smallest budget" + ); + + let bal = bal_with_accounts(accounts as u64); + let store = Arc::new(CountingStore::with_budget(MIN_BUDGET)); + let counter = store.clone(); + + LEVM::warm_block_from_bal(&bal, store, &AtomicBool::new(false)).expect("warm"); + + assert_eq!(counter.states_read.load(Ordering::Relaxed), accounts); + assert_eq!( + counter.codes_read.load(Ordering::Relaxed), + accounts, + "every bytecode should still be warmed, once each" + ); + } + + /// A backend reporting no bytecode cache gets no bytecode warming, while its account + /// states are still warmed. + #[test] + fn a_zero_budget_warms_accounts_and_no_code() { + let bal = bal_with_accounts(64); + let store = Arc::new(CountingStore::with_budget(0)); + let counter = store.clone(); + + LEVM::warm_block_from_bal(&bal, store, &AtomicBool::new(false)).expect("warm"); + + assert_eq!(counter.states_read.load(Ordering::Relaxed), 64); + assert_eq!(counter.codes_read.load(Ordering::Relaxed), 0); + } +} + // Exercises the rayon-parallel-BAL execution path (and shares its // `not(eip-8025)`-gated imports), so it only builds in the non-guest test profile. #[cfg(all(test, not(feature = "eip-8025")))] diff --git a/crates/vm/db.rs b/crates/vm/db.rs index da444b9787e..e40748bf464 100644 --- a/crates/vm/db.rs +++ b/crates/vm/db.rs @@ -13,6 +13,13 @@ pub trait VmDatabase: Send + Sync + DynClone { fn get_account_code(&self, code_hash: H256) -> Result; fn get_code_metadata(&self, code_hash: H256) -> Result; + /// Capacity of this backend's bytecode cache, in [`Code::size`] bytes, which bounds + /// how much bytecode a block warm reads speculatively. Default is the floor for a + /// backend holding no such cache. + fn code_cache_budget_bytes(&self) -> u64 { + 64 * 1024 * 1024 + } + /// Batch account-state lookup. Default impl loops `get_account_state`. /// Backends that can amortize per-key cost (e.g. rocksdb `multi_get_cf` on /// the flat key-value table) should override this. @@ -26,6 +33,17 @@ pub trait VmDatabase: Send + Sync + DynClone { .collect() } + /// Batch bytecode lookup, aligned to `code_hashes`. `None` means the hash is absent + /// from the database, which callers warming a cache treat as nothing to warm rather + /// than as failure. Default impl loops `get_account_code`; backends that can amortize + /// per-key cost (e.g. rocksdb `multi_get_cf` on the account-codes table) override it. + fn get_account_codes_batch(&self, code_hashes: &[H256]) -> Result>, EvmError> { + code_hashes + .iter() + .map(|h| self.get_account_code(*h).map(Some)) + .collect() + } + /// Batch storage-slot lookup. Default impl loops `get_storage_slot`. /// Backends that can amortize per-key cost (e.g. rocksdb `multi_get_cf` on /// the flat key-value table) should override this. diff --git a/crates/vm/levm/src/db/mod.rs b/crates/vm/levm/src/db/mod.rs index f4ad12a7929..bee4602cee0 100644 --- a/crates/vm/levm/src/db/mod.rs +++ b/crates/vm/levm/src/db/mod.rs @@ -3,7 +3,7 @@ use ethrex_common::{ Address, H256, U256, types::{AccountState, ChainConfig, Code, CodeMetadata}, }; -use rustc_hash::FxHashMap; +use rustc_hash::{FxHashMap, FxHashSet}; use std::sync::{Arc, OnceLock, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard}; pub mod gen_db; @@ -51,6 +51,19 @@ pub trait Database: Send + Sync { .map(|a| self.get_account_state(*a)) .collect() } + /// Batch bytecode lookup, aligned to `code_hashes`. `None` means the hash is absent + /// from the database. Default: loop. Backends with a batched read path (e.g. rocksdb + /// `multi_get_cf` on the account-codes table) should override this; the caching layer + /// above dispatches to it from [`Self::prefetch_codes`]. + fn get_account_codes_batch( + &self, + code_hashes: &[H256], + ) -> Result>, DatabaseError> { + code_hashes + .iter() + .map(|h| self.get_account_code(*h).map(Some)) + .collect() + } /// Batch storage-slot lookup. Default: loop. Backends with a batched read /// path (e.g. rocksdb `multi_get_cf` on the storage flat key-value table) /// should override this and the caching layer above will dispatch to it. @@ -62,6 +75,31 @@ pub trait Database: Send + Sync { .map(|&(addr, key)| self.get_storage_value(addr, key)) .collect() } + + /// Byte budget for warming bytecode ahead of execution, as measured by [`Code::size`]. + /// + /// Bounds the bytecode a block warm reads speculatively, since a block access list + /// does not say which of its addresses were accessed for their code. Backends holding + /// a bytecode cache report its capacity, which is how much the node was configured to + /// keep resident; the default is the floor for a backend without one. + fn code_cache_budget_bytes(&self) -> u64 { + 64 * 1024 * 1024 + } + + /// Prefetch a batch of bytecodes into the cache. Default: sequential fallback. + /// + /// Returns the total [`Code::size`] warmed, so a caller warming a whole block can + /// hold itself to [`Self::code_cache_budget_bytes`]. An absent hash is not an error: + /// warming has nothing to do for it and the executor reports it if reached. + fn prefetch_codes(&self, code_hashes: &[H256]) -> Result { + let mut warmed = 0u64; + for &hash in code_hashes { + if let Ok(code) = self.get_account_code(hash) { + warmed = warmed.saturating_add(u64::try_from(code.size()).unwrap_or(u64::MAX)); + } + } + Ok(warmed) + } /// Prefetch a batch of accounts into the cache. Default: sequential fallback. fn prefetch_accounts(&self, addresses: &[Address]) -> Result<(), DatabaseError> { for &addr in addresses { @@ -322,6 +360,68 @@ impl Database for CachingDatabase { self.precompile_cache.as_ref() } + /// Warms every address through [`Self::prefetch_accounts`], then answers from the + /// cache under one read lock. Repeated addresses cost a single database read, and a + /// caller that wants the states pays one map lookup each rather than a full + /// `get_account_state` round trip per address. + /// + /// The read lock spans the whole assembly, so callers should keep `addresses` to a + /// bounded slice: a writer (an executor filling the same cache) waits on it. + fn get_account_states_batch( + &self, + addresses: &[Address], + ) -> Result, DatabaseError> { + self.prefetch_accounts(addresses)?; + let cache = self.read_accounts()?; + addresses + .iter() + .map(|addr| { + cache.get(addr).copied().ok_or_else(|| { + DatabaseError::Custom(format!("account {addr:?} missing after prefetch")) + }) + }) + .collect() + } + + fn code_cache_budget_bytes(&self) -> u64 { + self.inner.code_cache_budget_bytes() + } + + /// Fetches the uncached bytecodes in one batch and inserts them, reading each + /// distinct hash once however many entries share it. + /// + /// Returns only the warmed byte count, not the bytecodes: the executor reads code + /// through [`Self::get_account_code`], which this turns into a cache hit. Assembling + /// a result vector here would hold the write lock that guards every executor code + /// read for the length of the batch, to build something a warming caller discards. + fn prefetch_codes(&self, code_hashes: &[H256]) -> Result { + let missing: Vec = { + let cache = self.read_code()?; + let mut seen: FxHashSet = FxHashSet::default(); + code_hashes + .iter() + .copied() + .filter(|h| !cache.contains_key(h) && seen.insert(*h)) + .collect() + }; + + if missing.is_empty() { + return Ok(0); + } + + let codes = self.inner.get_account_codes_batch(&missing)?; + let mut warmed = 0u64; + let mut cache = self.write_code()?; + for (hash, code) in missing.into_iter().zip(codes.into_iter()) { + // An absent hash has nothing to warm; the executor reports it if reached. + if let Some(code) = code { + warmed = warmed.saturating_add(u64::try_from(code.size()).unwrap_or(u64::MAX)); + cache.entry(hash).or_insert(code); + } + } + Ok(warmed) + } + fn prefetch_accounts(&self, addresses: &[Address]) -> Result<(), DatabaseError> { // Filter out already-cached addresses before issuing the batch read. let missing: Vec
= { diff --git a/test/tests/storage/account_code_batch_tests.rs b/test/tests/storage/account_code_batch_tests.rs new file mode 100644 index 00000000000..68e2ce3626e --- /dev/null +++ b/test/tests/storage/account_code_batch_tests.rs @@ -0,0 +1,197 @@ +//! Correctness parity between the batched bytecode lookup +//! (`Store::get_account_codes_batch`) used by the BAL code prefetch and the per-hash +//! single-get path (`Store::get_account_code`) the executor reads through. +//! +//! The prefetch warms a cache that execution trusts, so the batched path must return the +//! same code (and the same jump-destination bitmap) for every hash, including "absent +//! code" -> None, and must keep results aligned with the caller's order rather than the +//! order the keys happen to be read in. + +use bytes::Bytes; +use ethrex_common::{H256, types::Code}; +use ethrex_storage::{EngineType, Store}; + +/// Number of distinct codes written. +/// +/// The batched read picks between a parallel fan-out of point gets and sharded blocking +/// reads, switching once the batch forms more 256-key shards than there are cores. Sized +/// past that switch so the parity checks cover the sharded path; the fan-out path is +/// covered by [`batch_matches_the_single_get_below_the_shard_threshold`]. +fn code_count() -> u64 { + let parallelism = + u64::try_from(std::thread::available_parallelism().map_or(8, |p| p.get())).unwrap_or(8); + // One shard per 256 keys, and sharding is chosen above `parallelism` shards. + 256 * (parallelism + 1) +} + +const JUMPDEST: u8 = 0x5b; +const PUSH1: u8 = 0x60; + +/// A distinct code per `id`. The length varies with `id` and a `PUSH1` is planted so +/// that the byte after it is *not* a valid jump destination, which makes the bitmap +/// depend on the bytecode rather than being a constant. +fn code_of(id: u64) -> Code { + let len = 32 + (id as usize % 97); + let mut bytecode = vec![JUMPDEST; len]; + bytecode[0] = PUSH1; + let bytecode: Bytes = bytecode.into(); + Code::from_bytecode_unchecked(bytecode, H256::from_low_u64_be(id)) +} + +/// A hash never written, to cover the absent case. +fn absent_hash(id: u64) -> H256 { + H256::from_low_u64_be(1_000_000 + id) +} + +async fn store_with_codes(count: u64) -> Store { + let dir = tempfile::tempdir().expect("tempdir"); + let store = Store::new(dir.path(), EngineType::InMemory).expect("store"); + for id in 0..count { + store + .add_account_code(code_of(id)) + .await + .expect("add_account_code"); + } + store +} + +/// The batched read SHALL agree with the per-hash read on the bytecode, the jumpdest +/// bitmap, and absence, for a batch spanning several shards. +#[tokio::test] +async fn batch_matches_the_single_get_including_absent_hashes() { + let count = code_count(); + let store = store_with_codes(count).await; + + // Present and absent hashes interleaved, so a shard cannot be all-present. + let mut requested = Vec::new(); + for id in 0..count { + requested.push(H256::from_low_u64_be(id)); + requested.push(absent_hash(id)); + } + + // Read per-hash FIRST, on a cold cache, so these are real database reads rather + // than the batch's own inserts read back. + let singles: Vec> = requested + .iter() + .map(|hash| store.get_account_code(*hash).expect("get_account_code")) + .collect(); + + let batched = store + .get_account_codes_batch(&requested) + .expect("get_account_codes_batch"); + assert_eq!(batched.len(), requested.len()); + + for ((hash, batched), single) in requested.iter().zip(batched.iter()).zip(singles.iter()) { + // Independent expectation: absent hashes yield None, present ones the code we + // wrote, with the bitmap recomputed from the bytecode rather than read back. + let expected = (*hash != absent_hash(0) && hash.to_low_u64_be() < count) + .then(|| code_of(hash.to_low_u64_be())); + + assert_eq!( + batched.as_ref().map(|c| c.code()), + expected.as_ref().map(|c| c.code()), + "bytecode mismatch for {hash:?}" + ); + assert_eq!( + batched.as_ref().map(|c| c.jumpdests()), + expected.as_ref().map(|c| c.jumpdests()), + "jumpdest bitmap mismatch for {hash:?}" + ); + assert_eq!( + batched.as_ref().map(|c| c.code()), + single.as_ref().map(|c| c.code()), + "batch and per-hash read disagree for {hash:?}" + ); + } +} + +/// Results SHALL follow the caller's order. The batched read sorts internally, so a +/// result vector keyed by read order would silently mismatch the request. +#[tokio::test] +async fn batch_results_follow_the_requested_order() { + let count = code_count(); + let store = store_with_codes(count).await; + + let requested: Vec = (0..count).rev().map(H256::from_low_u64_be).collect(); + let batched = store + .get_account_codes_batch(&requested) + .expect("get_account_codes_batch"); + + for (id, batched) in (0..count).rev().zip(batched.iter()) { + let expected = code_of(id); + assert_eq!( + batched.as_ref().map(|c| c.code()), + Some(expected.code()), + "wrong code returned for id {id}" + ); + assert_eq!( + batched.as_ref().map(|c| c.jumpdests()), + Some(expected.jumpdests()), + "wrong bitmap returned for id {id}" + ); + } +} + +/// A hash repeated within one batch SHALL be answered at every position it occupies, +/// since the read deduplicates before going to the database. +#[tokio::test] +async fn batch_answers_every_position_of_a_repeated_hash() { + let store = store_with_codes(code_count()).await; + + let repeated = H256::from_low_u64_be(3); + let other = H256::from_low_u64_be(4); + let requested = vec![ + repeated, + other, + repeated, + repeated, + absent_hash(0), + repeated, + ]; + + let batched = store + .get_account_codes_batch(&requested) + .expect("get_account_codes_batch"); + + for i in [0, 2, 3, 5] { + assert_eq!( + batched[i].as_ref().map(|c| c.code()), + Some(code_of(3).code()), + "position {i} lost the repeated hash" + ); + } + assert_eq!( + batched[1].as_ref().map(|c| c.code()), + Some(code_of(4).code()) + ); + assert!(batched[4].is_none()); +} + +/// The same parity, for a batch small enough to take the parallel fan-out instead of the +/// sharded reads. Ordinary blocks land here, so it is the path that must not regress. +#[tokio::test] +async fn batch_matches_the_single_get_below_the_shard_threshold() { + let store = store_with_codes(64).await; + + let mut requested: Vec = (0..64).map(H256::from_low_u64_be).collect(); + requested.push(absent_hash(0)); + + let batched = store + .get_account_codes_batch(&requested) + .expect("get_account_codes_batch"); + + for (hash, batched) in requested.iter().zip(batched.iter()) { + let id = hash.to_low_u64_be(); + let expected = (id < 64).then(|| code_of(id)); + assert_eq!( + batched.as_ref().map(|c| c.code()), + expected.as_ref().map(|c| c.code()), + "bytecode mismatch for {hash:?}" + ); + assert_eq!( + batched.as_ref().map(|c| c.jumpdests()), + expected.as_ref().map(|c| c.jumpdests()), + "jumpdest bitmap mismatch for {hash:?}" + ); + } +} diff --git a/test/tests/storage/mod.rs b/test/tests/storage/mod.rs index 016b95af242..7f8db36d6a0 100644 --- a/test/tests/storage/mod.rs +++ b/test/tests/storage/mod.rs @@ -1,3 +1,4 @@ +mod account_code_batch_tests; mod deferred_persistence_tests; mod fcu_race_tests; mod storage_batch_tests;