perf(l1): batch and stream the BAL code prefetch

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
 

crates/blockchain/vm.rs

@@ -308,6 +308,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<Vec<Option<Code>>, 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<H256> = 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<H256, Code> = 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",

crates/storage/store.rs

@@ -952,6 +952,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<Vec<Option<Code>>, StoreError> {
+        let mut out: Vec<Option<Code>> = vec![None; code_hashes.len()];
+        // Positions to fill per distinct hash, so a repeated hash costs one read.
+        let mut pending: HashMap<H256, Vec<usize>> = 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<H256> = 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<Vec<u8>>| -> Result<Option<Code>, 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<Result<Option<Code>, StoreError>> = if shards > parallelism {
+            let chunk = missing.len().div_ceil(shards);
+            let rv = read_view.as_ref();
+            let read_shard = |hashes: &[H256]| -> Vec<Result<Option<Code>, 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<Code> = 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).

crates/vm/backends/levm/db.rs

@@ -79,6 +79,24 @@ impl LevmDatabase for DatabaseLogger {
         self.store.as_ref().get_account_code(code_hash)
     }
 
+    fn prefetch_codes(&self, code_hashes: &[CoreH256]) -> Result<(), DatabaseError> {
+        // 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<CodeMetadata, DatabaseError> {
         // 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
@@ -141,6 +159,14 @@ impl LevmDatabase for DynVmDatabase {
             .map_err(|e| DatabaseError::Custom(e.to_string()))
     }
 
+    fn get_account_codes_batch(
+        &self,
+        code_hashes: &[CoreH256],
+    ) -> Result<Vec<Option<Code>>, DatabaseError> {
+        <dyn VmDatabase>::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<CodeMetadata, DatabaseError> {
         <dyn VmDatabase>::get_code_metadata(self.as_ref(), code_hash)
             .map_err(|e| DatabaseError::Custom(e.to_string()))

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;
@@ -2877,36 +2877,58 @@ 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<Address> = 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.
+        //
+        // Sized to saturate the read depth of the batched code fetch, because on the
+        // blocks this targets the warmer 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.
+        let parallelism = std::thread::available_parallelism().map_or(8, |p| p.get());
+        let warm_chunk_accounts = 256_usize.saturating_mul(parallelism.saturating_mul(2).max(64));
+
+        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<Address> = 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}")))?;
 
-        // 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<ethrex_common::H256> = 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<ethrex_common::H256> = states
+                .iter()
+                .map(|s| s.code_hash)
+                .filter(|h| *h != *EMPTY_KECCAK_HASH)
+                .collect();
+            code_hashes.sort_unstable();
+            code_hashes.dedup();
+
+            // Re-checked between the two reads, not only per chunk: a cancelled warmer
+            // should not still issue a chunk's worth of code reads.
+            if code_hashes.is_empty() || cancelled.load(Ordering::Relaxed) {
+                continue;
+            }
+
+            // 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.
+            let _ = store.prefetch_codes(&code_hashes);
+        }
 
         Ok(())
     }

crates/vm/db.rs

@@ -25,6 +25,17 @@ pub trait VmDatabase: Send + Sync + DynClone {
             .map(|a| self.get_account_state(*a))
             .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<Vec<Option<Code>>, EvmError> {
+        code_hashes
+            .iter()
+            .map(|h| self.get_account_code(*h).map(Some))
+            .collect()
+    }
 }
 
 dyn_clone::clone_trait_object!(VmDatabase);