Skip to content

perf(l1): run batch block execution on spawn_blocking #7117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NikhilSharmaWe wants to merge 1 commit into
base: main
Choose a base branch
from
Open

perf(l1): run batch block execution on spawn_blocking #7117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion cmd/ethrex/cli.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -996,9 +996,10 @@ pub async fn import_blocks(
block_batch.push(block);
if block_batch.len() >= IMPORT_BATCH_SIZE || index + MIN_FULL_BLOCKS + 1 == size {
blockchain
.clone()
.add_blocks_in_batch(
mem::take(&mut block_batch),
&[],
vec![],
CancellationToken::new(),
)
.await
Expand Down
22 changes: 19 additions & 3 deletions crates/blockchain/blockchain.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2881,12 +2881,29 @@ impl Blockchain {
/// the failed block hash and the last successfully-imported block hash.
///
/// `bals` holds the per-block Block Access Lists fetched during sync, aligned by index with
/// `blocks`. Pass an empty slice when no BALs are available (e.g. block import from RLP); the
/// `blocks`. Pass an empty vec when no BALs are available (e.g. block import from RLP); the
/// pipeline then rebuilds each BAL. Only a BAL matching its block's header commitment is used.
///
/// The per-block pipeline is CPU/DB-bound, so the loop runs on Tokio's blocking pool and
/// does not hold an async worker for the duration of each block.
pub async fn add_blocks_in_batch(
self: Arc<Self>,
blocks: Vec<Block>,
bals: Vec<Option<BlockAccessList>>,
cancellation_token: CancellationToken,
) -> Result<(), (ChainError, Option<BatchBlockProcessingFailure>)> {
tokio::task::spawn_blocking(move || {
self.add_blocks_in_batch_sync(blocks, bals, cancellation_token)
})
.await
.map_err(|e| (ChainError::Custom(e.to_string()), None))?
}

/// Blocking loop for [`Self::add_blocks_in_batch`].
fn add_blocks_in_batch_sync(
&self,
blocks: Vec<Block>,
bals: &[Option<BlockAccessList>],
bals: Vec<Option<BlockAccessList>>,
cancellation_token: CancellationToken,
) -> Result<(), (ChainError, Option<BatchBlockProcessingFailure>)> {
debug_assert!(
Expand Down Expand Up @@ -2949,7 +2966,6 @@ impl Blockchain {
transactions_count += block_tx_count;

log_batch_progress(blocks_len as u32, i as u32);
tokio::task::yield_now().await;
}

let elapsed_seconds = interval.elapsed().as_secs_f64();
Expand Down
54 changes: 6 additions & 48 deletions crates/networking/p2p/sync/full.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ use ethrex_common::{
H256,
types::{Block, BlockBody, BlockHeader, block_access_list::BlockAccessList},
};
use ethrex_storage::{DB_COMMIT_THRESHOLD, Store};
use ethrex_storage::Store;
use tokio::sync::RwLock;
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;
Expand Down Expand Up @@ -195,7 +195,6 @@ pub async fn sync_cycle_full(
blockchain.clone(),
cancel_token.clone(),
pending_blocks,
true,
store.clone(),
peers,
)
Expand Down Expand Up @@ -538,7 +537,6 @@ pub async fn sync_cycle_full(
blockchain.clone(),
cancel_token.clone(),
blocks,
final_batch,
store.clone(),
peers,
)
Expand Down Expand Up @@ -581,7 +579,6 @@ pub async fn sync_cycle_full(
blockchain.clone(),
cancel_token.clone(),
pending_blocks,
true,
store.clone(),
peers,
)
Expand Down Expand Up @@ -615,7 +612,6 @@ async fn add_blocks_in_batch(
blockchain: Arc<Blockchain>,
cancel_token: CancellationToken,
blocks: Vec<Block>,
final_batch: bool,
store: Store,
peers: &mut PeerHandler,
) -> Result<(), SyncError> {
Expand All @@ -638,8 +634,8 @@ async fn add_blocks_in_batch(
let blocks_hashes = blocks.iter().map(|block| block.hash()).collect::<Vec<_>>();
let chain_config = store.get_chain_config();
let bals: Vec<Option<BlockAccessList>> = {
// Fetch BALs for every Amsterdam batch (not just the final one): both the
// batch path and `run_blocks_pipeline` now persist them, so peers can serve
// Fetch BALs for every Amsterdam batch (not just the final one):
// `add_blocks_in_batch` persists them, so peers can serve
// these blocks over eth/71 later without regenerating against pruned state.
let any_amsterdam = blocks
.iter()
Expand All @@ -658,7 +654,7 @@ async fn add_blocks_in_batch(
};
// Run the batch
if let Err((err, batch_failure)) =
add_blocks(blockchain.clone(), blocks, bals, final_batch, cancel_token).await
add_blocks(blockchain.clone(), blocks, bals, cancel_token).await
{
if let Some(batch_failure) = batch_failure {
warn!("Failed to add block during FullSync: {err}");
Expand Down Expand Up @@ -709,52 +705,14 @@ async fn add_blocks_in_batch(
Ok(())
}

/// Executes the given blocks and stores them.
///
/// Both paths execute block-by-block through the same validated pipeline
/// (`add_block_pipeline_bounded`), which builds fresh per-block VM state. When the sync
/// head is found the blocks run sequentially on a blocking thread; otherwise
/// `add_blocks_in_batch` runs them with BAL fetching, progress logging and cancellation.
/// Execute and store `blocks` via [`Blockchain::add_blocks_in_batch`].
async fn add_blocks(
blockchain: Arc<Blockchain>,
blocks: Vec<Block>,
bals: Vec<Option<BlockAccessList>>,
sync_head_found: bool,
cancel_token: CancellationToken,
) -> Result<(), (ChainError, Option<BatchBlockProcessingFailure>)> {
if sync_head_found {
return run_blocks_pipeline(blockchain, blocks, bals).await;
}
blockchain
.add_blocks_in_batch(blocks, &bals, cancel_token)
.add_blocks_in_batch(blocks, bals, cancel_token)
.await
}

async fn run_blocks_pipeline(
blockchain: Arc<Blockchain>,
blocks: Vec<Block>,
bals: Vec<Option<BlockAccessList>>,
) -> Result<(), (ChainError, Option<BatchBlockProcessingFailure>)> {
tokio::task::spawn_blocking(move || {
let mut last_valid_hash = H256::default();
for (block, bal) in blocks.into_iter().zip(bals.into_iter()) {
let block_hash = block.hash();
blockchain
.add_block_pipeline_bounded(block, bal.map(Arc::new), DB_COMMIT_THRESHOLD)
.map(|_| ())
.map_err(|e| {
(
e,
Some(BatchBlockProcessingFailure {
last_valid_hash,
failed_block_hash: block_hash,
}),
)
})?;
last_valid_hash = block_hash;
}
Ok(())
})
.await
.map_err(|e| (ChainError::Custom(e.to_string()), None))?
}
18 changes: 10 additions & 8 deletions test/tests/blockchain/batch_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
use std::{fs::File, io::BufReader, path::PathBuf};
use std::{fs::File, io::BufReader, path::PathBuf, sync::Arc};

use bytes::Bytes;
use ethrex_blockchain::{
Expand Down Expand Up @@ -234,10 +234,11 @@ async fn batch_selfdestruct_created_account_no_spurious_state() {
// 7. Create a fresh store B and re-execute both blocks in batch.
// This is the code path that was buggy before the fix.
let (store_b, _) = setup_store(sender).await;
let blockchain_b = Blockchain::default_with_store(store_b);
let blockchain_b = Arc::new(Blockchain::default_with_store(store_b));

let result = blockchain_b
.add_blocks_in_batch(vec![block1, block2], &[], CancellationToken::new())
.clone()
.add_blocks_in_batch(vec![block1, block2], vec![], CancellationToken::new())
.await;

assert!(
Expand Down Expand Up @@ -382,10 +383,11 @@ async fn batch_cross_batch_blockhash_regression() {
// Batch 1: blocks 1-2, Batch 2: block 3
// Block 3 needs blockhash(2) which is only in batch 1.
let (store_b, _) = setup_store(sender).await;
let blockchain_b = Blockchain::default_with_store(store_b);
let blockchain_b = Arc::new(Blockchain::default_with_store(store_b));

let result1 = blockchain_b
.add_blocks_in_batch(vec![block1, block2], &[], CancellationToken::new())
.clone()
.add_blocks_in_batch(vec![block1, block2], vec![], CancellationToken::new())
.await;
assert!(
result1.is_ok(),
Expand All @@ -394,7 +396,7 @@ async fn batch_cross_batch_blockhash_regression() {
);

let result2 = blockchain_b
.add_blocks_in_batch(vec![block3], &[], CancellationToken::new())
.add_blocks_in_batch(vec![block3], vec![], CancellationToken::new())
.await;
assert!(
result2.is_ok(),
Expand Down Expand Up @@ -438,10 +440,10 @@ async fn batch_single_block_selfdestruct() {

// Batch path: re-execute the same block on a fresh store.
let (store_b, _) = setup_store(sender).await;
let blockchain_b = Blockchain::default_with_store(store_b);
let blockchain_b = Arc::new(Blockchain::default_with_store(store_b));

let result = blockchain_b
.add_blocks_in_batch(vec![block1], &[], CancellationToken::new())
.add_blocks_in_batch(vec![block1], vec![], CancellationToken::new())
.await;

assert!(
Expand Down
2 changes: 1 addition & 1 deletion test/tests/blockchain/canonical_commit_gate_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,7 @@ async fn bounded_reexec_without_fcu_bounds_memory_and_serves_window() {
);

// Re-execute a single canonical chain through the BOUNDED path, block by block, with
// NO forkchoice_update — exactly what regenerate_head_state / run_blocks_pipeline /
// NO forkchoice_update — exactly what regenerate_head_state / full-sync batch import /
// import do. Record each block's state root to query the retained window later.
let mut parent_header = genesis_header;
let mut roots: Vec<H256> = Vec::with_capacity(BLOCKS as usize); // roots[N-1] == block N
Expand Down