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 and 4KB data blocks. `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 [#7099](https://github.com/lambdaclass/ethrex/pull/7099)
 
 ### 2026-07-22
 

crates/blockchain/vm.rs

@@ -308,6 +308,49 @@ 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<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()) {
+            match code {
+                Some(code) => {
+                    by_hash.insert(*hash, code);
+                }
+                None => return Err(EvmError::DB(format!("Code not found for hash: {hash:?}"))),
+            }
+        }
+
+        code_hashes
+            .iter()
+            .map(|h| {
+                if *h == *EMPTY_KECCAK_HASH {
+                    return Ok(Code::default());
+                }
+                by_hash
+                    .get(h)
+                    .cloned()
+                    .ok_or_else(|| EvmError::DB(format!("Code not found for hash: {h:?}")))
+            })
+            .collect()
+    }
+
     #[instrument(
         level = "trace",
         name = "Code metadata read",

crates/storage/store.rs

@@ -943,6 +943,132 @@ 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.
+        const KEYS_PER_SHARD: usize = 256;
+        const MAX_SHARDS: usize = 64;
+        let parallelism = std::thread::available_parallelism().map_or(8, |p| p.get());
+        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| scope.spawn(move || read_shard(ck)))
+                    .collect();
+                handles
+                    .into_iter()
+                    .flat_map(|h| h.join().expect("account code shard panicked"))
+                    .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 `jump_targets` deserialization).

crates/vm/backends/levm/db.rs

@@ -79,6 +79,27 @@ impl LevmDatabase for DatabaseLogger {
         self.store.as_ref().get_account_code(code_hash)
     }
 
+    fn get_account_codes_batch(
+        &self,
+        code_hashes: &[CoreH256],
+    ) -> Result<Vec<Code>, 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().get_account_codes_batch(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 +162,14 @@ impl LevmDatabase for DynVmDatabase {
             .map_err(|e| DatabaseError::Custom(e.to_string()))
     }
 
+    fn get_account_codes_batch(
+        &self,
+        code_hashes: &[CoreH256],
+    ) -> Result<Vec<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

@@ -66,7 +66,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;
@@ -2823,36 +2823,60 @@ 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 against the batched code read below (`Store::get_account_codes_batch`),
+        // which shards at 256 keys and caps at 64 shards: narrower chunks would cap its
+        // read concurrency, trading the executor's head start for shallower reads. This
+        // is the smallest chunk that can still reach that cap, and only does so when
+        // most accounts in a chunk hold distinct code; a chunk of EOAs yields no code
+        // hashes at all and the boundary costs nothing either way.
+        const WARM_CHUNK_ACCOUNTS: usize = 256 * 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();
+
+            if code_hashes.is_empty() {
+                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. The
+            // batch is all-or-nothing, so fall back to per-hash reads rather than
+            // leaving the whole chunk cold because of one absent entry.
+            if store.get_account_codes_batch(&code_hashes).is_err() {
+                for hash in &code_hashes {
+                    let _ = store.get_account_code(*hash);
+                }
+            }
+        }
 
         Ok(())
     }

crates/vm/db.rs

@@ -25,6 +25,16 @@ pub trait VmDatabase: Send + Sync + DynClone {
             .map(|a| self.get_account_state(*a))
             .collect()
     }
+
+    /// Batch bytecode lookup. Default impl loops `get_account_code`.
+    /// Backends that can amortize per-key cost (e.g. rocksdb `multi_get_cf` on
+    /// the account-codes table) should override this.
+    fn get_account_codes_batch(&self, code_hashes: &[H256]) -> Result<Vec<Code>, EvmError> {
+        code_hashes
+            .iter()
+            .map(|h| self.get_account_code(*h))
+            .collect()
+    }
 }
 
 dyn_clone::clone_trait_object!(VmDatabase);