Cache bloom filters (#598)

## 📝 Summary

Currently finalize time is often 30% bigger than root hash time but
ideally it should be really close.

One of the reasons is that we calculate bloom filters every single
finalize which can be expensive operation, this PR introduces caching
for bloom filters.

It also moves finalize to a separate thread so it can persist local
thread cache for the slot finalization and in addition the thread is not
named which makes profiling easier.

This shaves up to 5ms in the worst case from finalize but its a good
start for faster finalization

## 💡 Motivation and Context

<!--- (Optional) Why is this change required? What problem does it
solve? Remove this section if not applicable. -->

---

## ✅ I have completed the following steps:

* [ ] Run `make lint`
* [ ] Run `make test`
* [ ] Added tests (if applicable)

Author: dvush

crates/rbuilder/src/backtest/redistribute/mod.rs

@@ -18,7 +18,7 @@ use crate::{
     live_builder::{block_list_provider::BlockList, cli::LiveBuilderConfig},
     primitives::{Order, OrderId},
     provider::StateProviderFactory,
-    utils::{signed_uint_delta, u256decimal_serde_helper},
+    utils::{elapsed_s, signed_uint_delta, u256decimal_serde_helper},
 };
 use ahash::{HashMap, HashSet};
 use alloy_primitives::{utils::format_ether, Address, B256, I256, U256};
@@ -170,7 +170,7 @@ where
         config.base_config().evm_caching_enable,
     )?;
 
-    let time_preparation_s = start.elapsed().as_millis() as f64 / 1000.0;
+    let time_preparation_s = elapsed_s(start);
     let start = Instant::now();
 
     let results_without_exclusion = calculate_backtest_without_exclusion(
@@ -180,7 +180,7 @@ where
         block_data.clone(),
     )?;
 
-    let time_no_exclusion_s = start.elapsed().as_millis() as f64 / 1000.0;
+    let time_no_exclusion_s = elapsed_s(start);
     let start = Instant::now();
 
     let exclusion_results = calculate_backtest_identity_and_order_exclusion(
@@ -192,7 +192,7 @@ where
         &results_without_exclusion,
     )?;
 
-    let time_single_exclusion_s = start.elapsed().as_millis() as f64 / 1000.0;
+    let time_single_exclusion_s = elapsed_s(start);
     let start = Instant::now();
 
     let exclusion_results = calc_joint_exclusion_results(
@@ -206,7 +206,7 @@ where
         distribute_to_mempool_txs,
     )?;
 
-    let time_joint_exclusion_s = start.elapsed().as_millis() as f64 / 1000.0;
+    let time_joint_exclusion_s = elapsed_s(start);
     let start = Instant::now();
 
     let calculated_redistribution_result = apply_redistribution_formula(
@@ -237,7 +237,7 @@ where
         .map(|o| o.redistribution_value_received)
         .sum::<U256>();
 
-    let time_result_s = start.elapsed().as_millis() as f64 / 1000.0;
+    let time_result_s = elapsed_s(start);
 
     let time_total_s = time_preparation_s
         + time_no_exclusion_s

crates/rbuilder/src/bin/debug-bench-machine.rs

@@ -114,7 +114,7 @@ async fn main() -> eyre::Result<()> {
                 let build_time = build_time.elapsed();
 
                 let finalize_time = Instant::now();
-                let finalized_block = partial_block.finalize(&mut state, &ctx, &mut local_ctx)?;
+                let finalized_block = partial_block.finalize(state, &ctx, &mut local_ctx)?;
                 let finalize_time = finalize_time.elapsed();
 
                 debug!(

crates/rbuilder/src/building/builders/block_building_helper.rs

@@ -18,7 +18,7 @@ use crate::{
     },
     primitives::{SimValue, SimulatedOrder},
     telemetry::{self, add_block_fill_time, add_order_simulation_time},
-    utils::{check_block_hash_reader_health, HistoricalBlockError},
+    utils::{check_block_hash_reader_health, elapsed_ms, HistoricalBlockError},
 };
 
 use super::Block;
@@ -401,18 +401,22 @@ impl BlockBuildingHelper for BlockBuildingHelperFromProvider {
             return Err(BlockBuildingHelperError::PayoutTxNotAllowed);
         }
         let start_time = Instant::now();
+        let step_start = Instant::now();
 
         self.finalize_block_execution(local_ctx, payout_tx_value)?;
         // This could be moved outside of this func (pre finalize) since I don´t think the payout tx can change much.
         self.built_block_trace
             .verify_bundle_consistency(&self.building_ctx.blocklist)?;
 
+        let finalize_prep_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
+
         let sim_gas_used = self.partial_block.tracer.used_gas;
         let block_number = self.building_context().block();
         let finalized_block =
             match self
                 .partial_block
-                .finalize(&mut self.block_state, &self.building_ctx, local_ctx)
+                .finalize(self.block_state, &self.building_ctx, local_ctx)
             {
                 Ok(finalized_block) => finalized_block,
                 Err(err) => {
@@ -427,6 +431,15 @@ impl BlockBuildingHelper for BlockBuildingHelperFromProvider {
                     return Err(BlockBuildingHelperError::FinalizeError(err));
                 }
             };
+
+        let finalize_block_time_ms = elapsed_ms(step_start);
+        let finalize_time_ms = elapsed_ms(start_time);
+        trace!(
+            finalize_time_ms,
+            finalize_prep_time_ms,
+            finalize_block_time_ms,
+            "Block building helper finalized block"
+        );
         self.built_block_trace.update_orders_sealed_at();
         self.built_block_trace.root_hash_time = finalized_block.root_hash_time;
         self.built_block_trace.finalize_time = start_time.elapsed();

crates/rbuilder/src/building/builders/parallel_builder/block_building_result_assembler.rs

@@ -21,6 +21,7 @@ use crate::{
         BlockBuildingContext, ThreadBlockBuildingContext,
     },
     telemetry::mark_builder_considers_order,
+    utils::elapsed_ms,
 };
 
 /// Assembles block building results from the best orderings of order groups.
@@ -139,7 +140,7 @@ impl BlockBuildingResultAssembler {
                     trace!(
                         run_id = self.run_id,
                         version = version,
-                        time_ms = time_start.elapsed().as_millis(),
+                        time_ms = elapsed_ms(time_start),
                         profit = format_ether(value),
                         "Parallel builder built new block",
                     );

crates/rbuilder/src/building/builders/parallel_builder/conflict_resolving_pool.rs

@@ -15,8 +15,8 @@ use super::{
     simulation_cache::SharedSimulationCache, ConflictGroup, ConflictResolutionResultPerGroup,
     ConflictTask, GroupId, ResolutionResult, TaskPriority,
 };
-use crate::building::BlockBuildingContext;
 use crate::provider::StateProviderFactory;
+use crate::{building::BlockBuildingContext, utils::elapsed_ms};
 
 pub type TaskQueue = Arc<SegQueue<ConflictTask>>;
 
@@ -83,18 +83,18 @@ where
                             match group_result_sender.send((task_id, result)) {
                                 Ok(_) => {
                                     trace!(
-                                        task_id = %task_id,
-                                        time_taken_ms = %task_start.elapsed().as_millis(),
-                                        "Conflict resolving: successfully sent group result"
-                                    );
+                                                        task_id = %task_id,
+                                    time_taken_ms = %elapsed_ms(task_start),
+                                                        "Conflict resolving: successfully sent group result"
+                                                    );
                                 }
                                 Err(err) => {
                                     warn!(
-                                        task_id = %task_id,
-                                        error = ?err,
-                                        time_taken_ms = %task_start.elapsed().as_millis(),
-                                        "Conflict resolving: failed to send group result"
-                                    );
+                                                        task_id = %task_id,
+                                                        error = ?err,
+                                    time_taken_ms = %elapsed_ms(task_start),
+                                                        "Conflict resolving: failed to send group result"
+                                                    );
                                     return;
                                 }
                             }

crates/rbuilder/src/building/builders/parallel_builder/mod.rs

@@ -35,6 +35,7 @@ use crate::{
         LiveBuilderInput,
     },
     provider::StateProviderFactory,
+    utils::elapsed_ms,
 };
 
 use self::{
@@ -269,7 +270,7 @@ fn run_order_intake(
                 trace!(
                     new_orders_count = len,
                     groups_count = conflict_finder.get_order_groups().len(),
-                    time_taken_ms = %time_start.elapsed().as_millis(),
+                    time_taken_ms = %elapsed_ms(time_start),
                     "Order intake: added new orders and processing groups"
                 );
                 conflict_task_generator.process_groups(conflict_finder.get_order_groups());

crates/rbuilder/src/building/mod.rs

@@ -3,7 +3,11 @@ use crate::{
     primitives::{Order, OrderId, SimValue, SimulatedOrder, TransactionSignedEcRecoveredWithBlobs},
     provider::RootHasher,
     roothash::RootHashError,
-    utils::{a2r_withdrawal, default_cfg_env, timestamp_as_u64, Signer},
+    utils::{
+        a2r_withdrawal, default_cfg_env, elapsed_ms,
+        receipts::{calculate_receipt_root_and_block_logs_bloom, BloomCache},
+        timestamp_as_u64, Signer,
+    },
 };
 use alloy_consensus::{Header, EMPTY_OMMER_ROOT_HASH};
 use alloy_eips::{
@@ -49,6 +53,7 @@ use std::{
 };
 use thiserror::Error;
 use time::OffsetDateTime;
+use tracing::trace;
 use tx_sim_cache::TxExecutionCache;
 
 pub mod block_orders;
@@ -335,6 +340,7 @@ impl BlockBuildingContext {
 #[derive(Debug, Clone, Default)]
 pub struct ThreadBlockBuildingContext {
     pub cached_reads: LocalCachedReads,
+    pub bloom_cache: BloomCache,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -727,38 +733,53 @@ impl<Tracer: SimulationTracer> PartialBlock<Tracer> {
     #[allow(clippy::too_many_arguments)]
     pub fn finalize(
         self,
-        state: &mut BlockState,
+        mut state: BlockState,
         ctx: &BlockBuildingContext,
         local_ctx: &mut ThreadBlockBuildingContext,
     ) -> Result<FinalizeResult, FinalizeError> {
-        let (requests, withdrawals_root) = self.process_requests(state, ctx, local_ctx)?;
-        let bundle = state.clone_bundle();
+        let start = Instant::now();
+
+        let step_start = Instant::now();
+        let (requests, withdrawals_root) = self.process_requests(&mut state, ctx, local_ctx)?;
         let block_number = ctx.evm_env.block_env.number;
 
+        let request_processsing_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
+
         let requests_hash = requests.as_ref().map(|requests| requests.requests_hash());
+
+        let (bundle, _) = state.into_parts();
         let execution_outcome = ExecutionOutcome::new(
             bundle,
             vec![self.receipts],
             block_number,
             vec![requests.clone().unwrap_or_default()],
         );
 
-        // @TODO: Check ethereum_receipts_root since it could fail on Op. Check reth crates/optimism/payload/src/builder.rs?
-        let receipts_root = execution_outcome
-            .ethereum_receipts_root(block_number)
-            .expect("Number is in range");
-        let logs_bloom = execution_outcome
-            .block_logs_bloom(block_number)
-            .expect("Number is in range");
+        let exec_outcome_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
+
+        let (receipts_root, logs_bloom) = calculate_receipt_root_and_block_logs_bloom(
+            execution_outcome.receipts_by_block(block_number),
+            &mut local_ctx.bloom_cache,
+        );
+
+        let bloom_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
 
         // calculate the state root
-        let start = Instant::now();
         let state_root = ctx.root_hasher.state_root(&execution_outcome)?;
-        let root_hash_time = start.elapsed();
+        let root_hash_time = step_start.elapsed();
+
+        let root_hash_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
 
         // create the block header
         let transactions_root = proofs::calculate_transaction_root(&self.executed_tx);
 
+        let transactions_root_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
+
         // double check blocked txs
         for tx_with_blob in &self.executed_tx {
             if ctx.blocklist.contains(&tx_with_blob.signer()) {
@@ -788,6 +809,9 @@ impl<Tracer: SimulationTracer> PartialBlock<Tracer> {
             (None, None)
         };
 
+        let blobs_time_ms = elapsed_ms(step_start);
+        let step_start = Instant::now();
+
         let header = Header {
             parent_hash: ctx.attributes.parent,
             ommers_hash: EMPTY_OMMER_ROOT_HASH,
@@ -830,13 +854,30 @@ impl<Tracer: SimulationTracer> PartialBlock<Tracer> {
                 withdrawals,
             },
         };
-
-        Ok(FinalizeResult {
+        let result = FinalizeResult {
             sealed_block: block.seal_slow(),
             txs_blob_sidecars,
             root_hash_time,
             execution_requests: requests.map(|er| er.take()).unwrap_or_default(),
-        })
+        };
+
+        let block_seal_time_ms = elapsed_ms(step_start);
+
+        let total_time_ms = elapsed_ms(start);
+
+        trace!(
+            total_time_ms,
+            exec_outcome_time_ms,
+            bloom_time_ms,
+            request_processsing_time_ms,
+            root_hash_time_ms,
+            transactions_root_time_ms,
+            blobs_time_ms,
+            block_seal_time_ms,
+            "Partial block finalized block"
+        );
+
+        Ok(result)
     }
 
     pub fn pre_block_call(