From fe6a59604b2b678e446edab0a8f321c9bc78001c Mon Sep 17 00:00:00 2001 From: pasta Date: Mon, 31 Aug 2026 18:05:35 +0200 Subject: [PATCH] bench(drive-abci): per-block phase timing behind DRIVE_BLOCK_PERF Times each phase of ProcessProposal and FinalizeBlock and reports the means every DRIVE_BLOCK_PERF_EVERY blocks (default 500). Off unless DRIVE_BLOCK_PERF=1, and accumulated in memory rather than logged per block, so the measurement does not pay for a log line inside the spans it measures. This is what located the two per-block costs that scale with chain history: an unbounded withdrawal-document query and GroveDB checkpoint creation during replay. --- .../src/abci/handler/finalize_block.rs | 13 ++ .../engine/finalize_block_proposal/v0/mod.rs | 22 +++ .../engine/run_block_proposal/mod.rs | 8 + .../engine/run_block_proposal/v0/mod.rs | 44 +++++ packages/rs-drive-abci/src/lib.rs | 3 + packages/rs-drive-abci/src/perf.rs | 160 ++++++++++++++++++ 6 files changed, 250 insertions(+) create mode 100644 packages/rs-drive-abci/src/perf.rs diff --git a/packages/rs-drive-abci/src/abci/handler/finalize_block.rs b/packages/rs-drive-abci/src/abci/handler/finalize_block.rs index ade56bf1135..83306ec6896 100644 --- a/packages/rs-drive-abci/src/abci/handler/finalize_block.rs +++ b/packages/rs-drive-abci/src/abci/handler/finalize_block.rs @@ -18,6 +18,7 @@ where C: CoreRPCLike, { let _timer = crate::metrics::abci_request_duration("finalize_block"); + let mut laps = crate::perf::Laps::new(); let transaction_guard = app.transaction().read().unwrap(); let transaction = @@ -45,6 +46,8 @@ where let block_height = request_finalize_block.height; + laps.lap("fb_setup"); + let block_finalization_outcome = app.platform().finalize_block_proposal( request_finalize_block, block_execution_context, @@ -52,6 +55,8 @@ where platform_version, )?; + laps.lap("fb_proposal"); + drop(transaction_guard); //FIXME: tell tenderdash about the problem instead @@ -69,6 +74,8 @@ where let result = app.commit_transaction(platform_version); + laps.lap("fb_commit"); + // We had a sequence of errors on the mainnet started since block 32326. // We got RocksDB's "transaction is busy" error because of a bug (https://github.com/dashpay/platform/pull/2309). // Due to another bug in Tenderdash (https://github.com/dashpay/tenderdash/pull/966), @@ -92,6 +99,8 @@ where result.expect("commit transaction"); } + laps.lap("fb_commit_check"); + app.platform() .committed_block_height_guard .store(block_height, Ordering::Relaxed); @@ -101,6 +110,10 @@ where app.platform().create_grovedb_checkpoint(platform_version)?; } + laps.lap("fb_checkpoint"); + drop(laps); + crate::perf::end_block(block_height); + Ok(proto::ResponseFinalizeBlock { retain_height: 0 }) } diff --git a/packages/rs-drive-abci/src/execution/engine/finalize_block_proposal/v0/mod.rs b/packages/rs-drive-abci/src/execution/engine/finalize_block_proposal/v0/mod.rs index fbda2224c5c..0da94a5207e 100644 --- a/packages/rs-drive-abci/src/execution/engine/finalize_block_proposal/v0/mod.rs +++ b/packages/rs-drive-abci/src/execution/engine/finalize_block_proposal/v0/mod.rs @@ -63,6 +63,8 @@ where transaction: &Transaction, platform_version: &PlatformVersion, ) -> Result { + let mut laps = crate::perf::Laps::new(); + let mut validation_result = SimpleValidationResult::::new_with_errors(vec![]); let block_state_info = block_execution_context.block_state_info(); @@ -94,6 +96,8 @@ where .try_into() .expect("invalid sha256 length"); + laps.lap("fbp_msg_hash"); + //// Verification that commit is for our current executed block // When receiving the finalized block, we need to make sure info matches our current block @@ -136,6 +140,8 @@ where return Ok(validation_result.into()); } + laps.lap("fbp_basic_checks"); + // Verify votes extensions // We don't need to verify votes extension signatures once again after tenderdash // here, because we will do it bellow broadcasting withdrawal transactions. @@ -154,6 +160,8 @@ where return Ok(validation_result.into()); }; + laps.lap("fbp_vote_ext"); + // Verify commit // In production this will always be true @@ -188,6 +196,8 @@ where } } + laps.lap("fbp_verify_commit"); + if height == self.config.abci.genesis_height { self.drive .set_genesis_time(block_state_info.block_time_ms()); @@ -205,6 +215,8 @@ where to_commit_block_info.core_height = block_header.core_chain_locked_height; + laps.lap("fbp_block_info"); + if !transaction_to_extension_matches.is_empty() { self.append_signatures_and_broadcast_withdrawal_transactions( transaction_to_extension_matches, @@ -212,6 +224,8 @@ where )?; } + laps.lap("fbp_wd_broadcast"); + // Update platform (drive abci) state let extended_block_info = ExtendedBlockInfoV0 { @@ -225,12 +239,18 @@ where } .into(); + laps.lap("fbp_ext_block_info"); + self.update_drive_cache(&block_execution_context, platform_version)?; + laps.lap("fbp_drive_cache"); + // Check if we should create a checkpoint (must be done before consuming block_execution_context) let checkpoint_needed = self.should_checkpoint(&block_execution_context, platform_version)?; + laps.lap("fbp_should_checkpoint"); + let block_platform_state = block_execution_context.block_platform_state_owned(); self.update_state_cache( @@ -240,6 +260,8 @@ where platform_version, )?; + laps.lap("fbp_state_cache"); + // Gather some metrics crate::metrics::abci_last_block_time(block_header.time.seconds as u64); crate::metrics::abci_last_platform_height(height); diff --git a/packages/rs-drive-abci/src/execution/engine/run_block_proposal/mod.rs b/packages/rs-drive-abci/src/execution/engine/run_block_proposal/mod.rs index 68d87275ace..1e2df83bf18 100644 --- a/packages/rs-drive-abci/src/execution/engine/run_block_proposal/mod.rs +++ b/packages/rs-drive-abci/src/execution/engine/run_block_proposal/mod.rs @@ -53,6 +53,8 @@ where timer: Option<&HistogramTiming>, ) -> Result, Error> { + let mut laps = crate::perf::Laps::new(); + // Epoch information is always calculated with the last committed platform version // even if we are switching to a new version in this block. let last_committed_platform_version = platform_state.current_platform_version()?; @@ -66,6 +68,8 @@ where last_committed_platform_version, )?; + laps.lap("epoch_info"); + // Cleanup block cache before we execute a new proposal. // // This has to happen before `perform_events_on_first_block_of_protocol_change` below: @@ -74,9 +78,13 @@ where // them, leaving those reads to fall back to pre-change global cache entries. self.clear_drive_block_cache(last_committed_platform_version)?; + laps.lap("clear_block_cache"); + // Create a bock state from previous committed state let mut block_platform_state = platform_state.clone(); + laps.lap("state_clone"); + // Determine a platform version for this block let block_platform_version = if epoch_info.is_epoch_change_but_not_genesis() && platform_state.next_epoch_protocol_version() diff --git a/packages/rs-drive-abci/src/execution/engine/run_block_proposal/v0/mod.rs b/packages/rs-drive-abci/src/execution/engine/run_block_proposal/v0/mod.rs index b93125a08c9..1ec35e73786 100644 --- a/packages/rs-drive-abci/src/execution/engine/run_block_proposal/v0/mod.rs +++ b/packages/rs-drive-abci/src/execution/engine/run_block_proposal/v0/mod.rs @@ -70,6 +70,8 @@ where timer: Option<&HistogramTiming>, ) -> Result, Error> { + let mut laps = crate::perf::Laps::new(); + tracing::trace!( method = "run_block_proposal_v0", ?block_proposal, @@ -158,6 +160,8 @@ where platform_version, )?; + laps.lap("upgrade"); + // If there is a core chain lock update, we should start by verifying it if let Some(core_chain_lock_update) = core_chain_lock_update.as_ref() { if !known_from_us { @@ -242,6 +246,8 @@ where } } + laps.lap("chainlock"); + // Update the masternode list and create masternode identities and also update the active quorums self.update_core_info( Some(last_committed_platform_state), @@ -253,6 +259,8 @@ where platform_version, )?; + laps.lap("core_info"); + // Update the validator proposed app version // It should be called after protocol version upgrade self.drive @@ -266,6 +274,8 @@ where Error::Execution(ExecutionError::UpdateValidatorProposedAppVersionError(e)) })?; // This is a system error + laps.lap("val_app_ver"); + // Rebroadcast expired withdrawals if they exist // We do that before we mark withdrawals as expired // to rebroadcast them on the next block but not the same @@ -278,6 +288,8 @@ where platform_version, )?; + laps.lap("wd_rebroadcast"); + // Mark all previously broadcasted and chainlocked withdrawals as complete // only when we are on a new core height if block_state_info.core_chain_locked_height() != last_block_core_height { @@ -288,6 +300,8 @@ where )?; } + laps.lap("wd_status"); + // Preparing withdrawal transactions for signing and broadcasting // To process withdrawals we need to dequeue untiled transactions from the withdrawal transactions queue // Untiled transactions then converted to unsigned transactions, appending current block information @@ -304,6 +318,8 @@ where platform_version, )?; + laps.lap("wd_dequeue"); + // Run all dao platform events, such as vote tallying and distribution of contested documents // This must be done before state transition processing // Otherwise we would expect a proof after a successful vote that has since been cleaned up. @@ -315,6 +331,8 @@ where platform_version, )?; + laps.lap("dao"); + // Process transactions let state_transitions_result = self.process_raw_state_transitions( raw_state_transitions, @@ -326,6 +344,8 @@ where timer, )?; + laps.lap("state_transitions"); + // Store the address balances to recent block storage self.store_address_balances_to_recent_block_storage( &state_transitions_result.address_balances_updated, @@ -334,6 +354,8 @@ where platform_version, )?; + laps.lap("addr_store"); + // Clean up expired compacted address balance entries self.cleanup_recent_block_storage_address_balances( &block_info, @@ -341,6 +363,8 @@ where platform_version, )?; + laps.lap("addr_cleanup"); + // Record shielded pool anchor if the commitment tree changed this block. // This stores block_height → anchor_bytes so shielded transactions can // reference a recent anchor for spend authorization. @@ -350,9 +374,13 @@ where platform_version, )?; + laps.lap("shield_anchor"); + // Prune anchors older than the configured retention depth self.prune_shielded_pool_anchors(block_proposal.height, transaction, platform_version)?; + laps.lap("shield_prune"); + // Pool withdrawals into transactions queue // Takes queued withdrawals, creates untiled withdrawal transaction payload, saves them to queue @@ -364,6 +392,8 @@ where platform_version, )?; + laps.lap("wd_pool"); + // Cleans up the expired locks for withdrawal amounts // to update daily withdrawal limit // This is for example when we make a withdrawal for 30 Dash @@ -376,6 +406,8 @@ where platform_version, )?; + laps.lap("wd_locks"); + // Create a new block execution context let mut block_execution_context: BlockExecutionContext = @@ -389,6 +421,8 @@ where } .into(); + laps.lap("exec_ctx"); + // while we have the state transitions executed, we now need to process the block fees let block_fees_v0: BlockFeesV0 = state_transitions_result.aggregated_fees().clone().into(); @@ -402,6 +436,8 @@ where tracing::debug!(block_fees = ?processed_block_fees, "block fees are processed"); + laps.lap("fees"); + // Record the credits this block minted into Platform (asset locks funding state // transitions, epoch Core rewards) as a credit inflow: the daily withdrawal limit adds // inflows younger than its day-old base to the daily maximum, so it limits net outflow. @@ -415,6 +451,8 @@ where platform_version, )?; + laps.lap("credit_inflow"); + // Record the total credits in Platform if this block changed it: the daily withdrawal // limit is a share of the total credits Platform held a day ago, read from this history. // This runs after fees and epoch rewards, the last things in a block that can move the @@ -425,6 +463,8 @@ where platform_version, )?; + laps.lap("total_credits"); + let root_hash = self .drive .grove @@ -436,6 +476,8 @@ where .block_state_info_mut() .set_app_hash(Some(root_hash)); + laps.lap("root_hash"); + let validator_set_update = self.validator_set_update( block_proposal.proposer_pro_tx_hash, last_committed_platform_state, @@ -443,6 +485,8 @@ where platform_version, )?; + laps.lap("validator_set"); + if tracing::enabled!(tracing::Level::TRACE) { tracing::trace!( method = "run_block_proposal_v0", diff --git a/packages/rs-drive-abci/src/lib.rs b/packages/rs-drive-abci/src/lib.rs index ac1ac54661b..ade63f88abd 100644 --- a/packages/rs-drive-abci/src/lib.rs +++ b/packages/rs-drive-abci/src/lib.rs @@ -69,6 +69,9 @@ pub mod core; /// Metrics subsystem pub mod metrics; +/// Per-block phase timing, enabled with DRIVE_BLOCK_PERF=1 +pub mod perf; + /// Test helpers and fixtures #[cfg(any(feature = "mocks", test))] pub mod test; diff --git a/packages/rs-drive-abci/src/perf.rs b/packages/rs-drive-abci/src/perf.rs new file mode 100644 index 00000000000..1377752b1a1 --- /dev/null +++ b/packages/rs-drive-abci/src/perf.rs @@ -0,0 +1,160 @@ +//! Lightweight per-block phase timing. +//! +//! Enabled only when `DRIVE_BLOCK_PERF=1` is set in the environment. Phases are +//! accumulated in memory and reported as means every `DRIVE_BLOCK_PERF_EVERY` +//! blocks (default 500), so the measurement does not pay for a log line inside +//! the very spans it is measuring. + +use std::sync::{Mutex, OnceLock}; +use std::time::Instant; + +fn enabled() -> bool { + static ENABLED: OnceLock = OnceLock::new(); + *ENABLED.get_or_init(|| std::env::var("DRIVE_BLOCK_PERF").as_deref() == Ok("1")) +} + +fn report_every() -> u64 { + static EVERY: OnceLock = OnceLock::new(); + *EVERY.get_or_init(|| { + std::env::var("DRIVE_BLOCK_PERF_EVERY") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(500) + }) +} + +#[derive(Default)] +struct Totals { + blocks: u64, + /// (name, summed microseconds, samples), in first-seen order + phases: Vec<(&'static str, u64, u64)>, +} + +impl Totals { + fn add(&mut self, name: &'static str, micros: u64) { + if let Some(entry) = self.phases.iter_mut().find(|(n, _, _)| *n == name) { + entry.1 += micros; + entry.2 += 1; + } else { + self.phases.push((name, micros, 1)); + } + } +} + +fn totals() -> &'static Mutex { + static TOTALS: OnceLock> = OnceLock::new(); + TOTALS.get_or_init(|| Mutex::new(Totals::default())) +} + +/// Accumulates the elapsed time of successive phases of block execution. +/// +/// Timings are merged into the process-wide totals when the value is dropped. +pub struct Laps { + last: Instant, + on: bool, + buf: Vec<(&'static str, u64)>, +} + +impl Laps { + /// Start a new lap sequence. Cheap and inert when perf logging is off. + pub fn new() -> Self { + let on = enabled(); + Laps { + last: Instant::now(), + on, + buf: if on { + Vec::with_capacity(32) + } else { + Vec::new() + }, + } + } + + /// Record the time since the previous lap under `name`. + pub fn lap(&mut self, name: &'static str) { + if !self.on { + return; + } + let now = Instant::now(); + self.buf + .push((name, now.duration_since(self.last).as_micros() as u64)); + self.last = now; + } + + /// True when perf logging is enabled. + pub fn on(&self) -> bool { + self.on + } +} + +impl Default for Laps { + fn default() -> Self { + Self::new() + } +} + +impl Drop for Laps { + fn drop(&mut self) { + if !self.on || self.buf.is_empty() { + return; + } + let mut totals = totals().lock().expect("block perf totals poisoned"); + for (name, micros) in self.buf.drain(..) { + totals.add(name, micros); + } + } +} + +/// Record a non-timing value (e.g. a byte count) under `name`. +pub fn value(name: &'static str, v: u64) { + if !enabled() { + return; + } + totals() + .lock() + .expect("block perf totals poisoned") + .add(name, v); +} + +/// Called once per finalized block. Emits the means and resets every +/// `DRIVE_BLOCK_PERF_EVERY` blocks. +pub fn end_block(height: u64) { + if !enabled() { + return; + } + let every = report_every(); + let report = { + let mut totals = totals().lock().expect("block perf totals poisoned"); + totals.blocks += 1; + if totals.blocks < every { + None + } else { + let blocks = totals.blocks; + let mut line = String::with_capacity(totals.phases.len() * 20); + for (name, sum, samples) in &totals.phases { + if !line.is_empty() { + line.push(' '); + } + // mean over blocks, not over samples: a phase that only runs on + // some blocks should show its share of the per-block cost + line.push_str(name); + line.push('='); + line.push_str(&(*sum / blocks).to_string()); + line.push('/'); + line.push_str(&samples.to_string()); + } + totals.phases.clear(); + totals.blocks = 0; + Some((blocks, line)) + } + }; + if let Some((blocks, line)) = report { + tracing::info!( + block_perf = "agg", + height, + blocks, + phases = line, + "block perf" + ); + } +}