refactor(engine): adjust prefetch and state update batch sizes

Increase the maximum number of targets for prefetch batching from 500 to 512 and decrease the maximum for state updates from 128 to 64. These changes aim to optimize the batching logic and improve overall performance in handling proof requests and state updates.

Author: yongkangc

crates/engine/tree/src/tree/payload_processor/multiproof.rs

@@ -29,7 +29,7 @@ use tracing::{debug, error, instrument, trace};
 /// Maximum number of targets to batch together for prefetch batching.
 /// Prefetches are just proof requests (no state merging), so we allow a higher cap than state
 /// updates
-const PREFETCH_MAX_BATCH_TARGETS: usize = 500;
+const PREFETCH_MAX_BATCH_TARGETS: usize = 512;
 
 /// Maximum number of prefetch messages to batch together.
 /// Prevents excessive batching even with small messages.
@@ -38,7 +38,7 @@ const PREFETCH_MAX_BATCH_MESSAGES: usize = 16;
 /// Maximum number of targets to batch together for state updates.
 /// Lower than prefetch because state updates require additional processing (hashing, state
 /// partitioning) before dispatch.
-const STATE_UPDATE_MAX_BATCH_TARGETS: usize = 128;
+const STATE_UPDATE_MAX_BATCH_TARGETS: usize = 64;
 
 /// The default max targets, for limiting the number of account and storage proof targets to be
 /// fetched by a single worker. If exceeded, chunking is forced regardless of worker availability.
@@ -215,8 +215,8 @@ fn can_batch_state_update(
     }
 
     match (batch_source, next_source) {
-        (StateChangeSource::PreBlock(_), StateChangeSource::PreBlock(_)) |
-        (StateChangeSource::PostBlock(_), StateChangeSource::PostBlock(_)) => {
+        (StateChangeSource::PreBlock(_), StateChangeSource::PreBlock(_))
+        | (StateChangeSource::PostBlock(_), StateChangeSource::PostBlock(_)) => {
             batch_update == next_update
         }
         _ => true,
@@ -271,13 +271,13 @@ fn dispatch_with_chunking<T, I>(
 where
     I: IntoIterator<Item = T>,
 {
-    let should_chunk = chunking_len > max_targets_for_chunking ||
-        available_account_workers > 1 ||
-        available_storage_workers > 1;
+    let should_chunk = chunking_len > max_targets_for_chunking
+        || available_account_workers > 1
+        || available_storage_workers > 1;
 
-    if should_chunk &&
-        let Some(chunk_size) = chunk_size &&
-        chunking_len > chunk_size
+    if should_chunk
+        && let Some(chunk_size) = chunk_size
+        && chunking_len > chunk_size
     {
         let mut dispatched = 0usize;
         for chunk in chunker(items, chunk_size) {
@@ -1134,12 +1134,12 @@ impl MultiProofTask {
                 accumulated_targets.push(targets);
 
                 // Fast-path dispatch if the first message already fills the batch.
-                if accumulated_count < PREFETCH_MAX_BATCH_TARGETS &&
-                    accumulated_targets.len() < PREFETCH_MAX_BATCH_MESSAGES
+                if accumulated_count < PREFETCH_MAX_BATCH_TARGETS
+                    && accumulated_targets.len() < PREFETCH_MAX_BATCH_MESSAGES
                 {
                     loop {
-                        if accumulated_count >= PREFETCH_MAX_BATCH_TARGETS ||
-                            accumulated_targets.len() >= PREFETCH_MAX_BATCH_MESSAGES
+                        if accumulated_count >= PREFETCH_MAX_BATCH_TARGETS
+                            || accumulated_targets.len() >= PREFETCH_MAX_BATCH_MESSAGES
                         {
                             break;
                         }
@@ -1205,7 +1205,7 @@ impl MultiProofTask {
                 // Accumulate messages including the first one
                 let mut accumulated_targets = estimate_evm_state_targets(&update);
                 // Preallocate modestly; state updates are heavier per message, but we can see small
-                // bursts. 
+                // bursts.
                 let mut accumulated_updates: Vec<(StateChangeSource, EvmState)> =
                     Vec::with_capacity(16);
                 accumulated_updates.push((source, update));
@@ -1572,8 +1572,8 @@ fn get_proof_targets(
             .storage
             .keys()
             .filter(|slot| {
-                !fetched.is_some_and(|f| f.contains(*slot)) ||
-                    storage_added_removed_keys.is_some_and(|k| k.is_removed(slot))
+                !fetched.is_some_and(|f| f.contains(*slot))
+                    || storage_added_removed_keys.is_some_and(|k| k.is_removed(slot))
             })
             .peekable();
 
@@ -2249,8 +2249,8 @@ mod tests {
                 }
                 match task.rx.try_recv() {
                     Ok(MultiProofMessage::StateUpdate(next_source, next_update)) => {
-                        if let Some((batch_source, batch_update)) = accumulated_updates.first() &&
-                            !can_batch_state_update(
+                        if let Some((batch_source, batch_update)) = accumulated_updates.first()
+                            && !can_batch_state_update(
                                 *batch_source,
                                 batch_update,
                                 next_source,
@@ -2268,8 +2268,8 @@ mod tests {
                                 Some(MultiProofMessage::StateUpdate(next_source, next_update));
                             break;
                         }
-                        if accumulated_targets + next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS &&
-                            !accumulated_updates.is_empty()
+                        if accumulated_targets + next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS
+                            && !accumulated_updates.is_empty()
                         {
                             pending_msg =
                                 Some(MultiProofMessage::StateUpdate(next_source, next_update));
@@ -2371,8 +2371,8 @@ mod tests {
                 }
                 match task.rx.try_recv() {
                     Ok(MultiProofMessage::StateUpdate(next_source, next_update)) => {
-                        if let Some((batch_source, batch_update)) = accumulated_updates.first() &&
-                            !can_batch_state_update(
+                        if let Some((batch_source, batch_update)) = accumulated_updates.first()
+                            && !can_batch_state_update(
                                 *batch_source,
                                 batch_update,
                                 next_source,
@@ -2390,8 +2390,8 @@ mod tests {
                                 Some(MultiProofMessage::StateUpdate(next_source, next_update));
                             break;
                         }
-                        if accumulated_targets + next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS &&
-                            !accumulated_updates.is_empty()
+                        if accumulated_targets + next_estimate > STATE_UPDATE_MAX_BATCH_TARGETS
+                            && !accumulated_updates.is_empty()
                         {
                             pending_msg =
                                 Some(MultiProofMessage::StateUpdate(next_source, next_update));