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..2ae9d542fd6 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,8 @@ where C: CoreRPCLike, { let _timer = crate::metrics::abci_request_duration("finalize_block"); + #[cfg(debug_assertions)] + let mut laps = crate::perf::Laps::new(); let transaction_guard = app.transaction().read().unwrap(); let transaction = @@ -45,6 +47,9 @@ where let block_height = request_finalize_block.height; + #[cfg(debug_assertions)] + laps.lap("fb_setup"); + let block_finalization_outcome = app.platform().finalize_block_proposal( request_finalize_block, block_execution_context, @@ -52,6 +57,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("fb_proposal"); + drop(transaction_guard); //FIXME: tell tenderdash about the problem instead @@ -69,6 +77,9 @@ where let result = app.commit_transaction(platform_version); + #[cfg(debug_assertions)] + 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 +103,9 @@ where result.expect("commit transaction"); } + #[cfg(debug_assertions)] + laps.lap("fb_commit_check"); + app.platform() .committed_block_height_guard .store(block_height, Ordering::Relaxed); @@ -101,6 +115,16 @@ where app.platform().create_grovedb_checkpoint(platform_version)?; } + #[cfg(debug_assertions)] + laps.lap_if( + block_finalization_outcome.checkpoint_needed, + "fb_checkpoint", + ); + #[cfg(debug_assertions)] + drop(laps); + #[cfg(debug_assertions)] + 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..ecbb5b913c6 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,9 @@ where transaction: &Transaction, platform_version: &PlatformVersion, ) -> Result { + #[cfg(debug_assertions)] + 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 +97,9 @@ where .try_into() .expect("invalid sha256 length"); + #[cfg(debug_assertions)] + 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 +142,9 @@ where return Ok(validation_result.into()); } + #[cfg(debug_assertions)] + 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 +163,9 @@ where return Ok(validation_result.into()); }; + #[cfg(debug_assertions)] + laps.lap("fbp_vote_ext"); + // Verify commit // In production this will always be true @@ -188,6 +200,9 @@ where } } + #[cfg(debug_assertions)] + laps.lap("fbp_verify_commit"); + if height == self.config.abci.genesis_height { self.drive .set_genesis_time(block_state_info.block_time_ms()); @@ -205,13 +220,20 @@ where to_commit_block_info.core_height = block_header.core_chain_locked_height; - if !transaction_to_extension_matches.is_empty() { + #[cfg(debug_assertions)] + laps.lap("fbp_block_info"); + + let broadcast_withdrawals = !transaction_to_extension_matches.is_empty(); + if broadcast_withdrawals { self.append_signatures_and_broadcast_withdrawal_transactions( transaction_to_extension_matches, platform_version, )?; } + #[cfg(debug_assertions)] + laps.lap_if(broadcast_withdrawals, "fbp_wd_broadcast"); + // Update platform (drive abci) state let extended_block_info = ExtendedBlockInfoV0 { @@ -225,12 +247,21 @@ where } .into(); + #[cfg(debug_assertions)] + laps.lap("fbp_ext_block_info"); + self.update_drive_cache(&block_execution_context, platform_version)?; + #[cfg(debug_assertions)] + 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)?; + #[cfg(debug_assertions)] + laps.lap("fbp_should_checkpoint"); + let block_platform_state = block_execution_context.block_platform_state_owned(); self.update_state_cache( @@ -240,6 +271,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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..341d051a132 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,9 @@ where timer: Option<&HistogramTiming>, ) -> Result, Error> { + #[cfg(debug_assertions)] + 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 +69,9 @@ where last_committed_platform_version, )?; + #[cfg(debug_assertions)] + 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 +80,15 @@ where // them, leaving those reads to fall back to pre-change global cache entries. self.clear_drive_block_cache(last_committed_platform_version)?; + #[cfg(debug_assertions)] + laps.lap("clear_block_cache"); + // Create a bock state from previous committed state let mut block_platform_state = platform_state.clone(); + #[cfg(debug_assertions)] + 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..f3d8aed7a63 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,9 @@ where timer: Option<&HistogramTiming>, ) -> Result, Error> { + #[cfg(debug_assertions)] + let mut laps = crate::perf::Laps::new(); + tracing::trace!( method = "run_block_proposal_v0", ?block_proposal, @@ -158,6 +161,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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 +248,13 @@ where } } + // The verification only runs for a chain lock we did not propose ourselves. + #[cfg(debug_assertions)] + laps.lap_if( + core_chain_lock_update.is_some() && !known_from_us, + "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 +266,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("core_info"); + // Update the validator proposed app version // It should be called after protocol version upgrade self.drive @@ -266,6 +282,9 @@ where Error::Execution(ExecutionError::UpdateValidatorProposedAppVersionError(e)) })?; // This is a system error + #[cfg(debug_assertions)] + 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,9 +297,14 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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 { + let core_height_advanced = + block_state_info.core_chain_locked_height() != last_block_core_height; + if core_height_advanced { self.update_broadcasted_withdrawal_statuses( &block_info, transaction, @@ -288,6 +312,9 @@ where )?; } + #[cfg(debug_assertions)] + laps.lap_if(core_height_advanced, "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 +331,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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 +345,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("dao"); + // Process transactions let state_transitions_result = self.process_raw_state_transitions( raw_state_transitions, @@ -326,6 +359,9 @@ where timer, )?; + #[cfg(debug_assertions)] + 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 +370,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("addr_store"); + // Clean up expired compacted address balance entries self.cleanup_recent_block_storage_address_balances( &block_info, @@ -341,6 +380,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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 +392,15 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("shield_anchor"); + // Prune anchors older than the configured retention depth self.prune_shielded_pool_anchors(block_proposal.height, transaction, platform_version)?; + #[cfg(debug_assertions)] + laps.lap("shield_prune"); + // Pool withdrawals into transactions queue // Takes queued withdrawals, creates untiled withdrawal transaction payload, saves them to queue @@ -364,6 +412,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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 +427,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("wd_locks"); + // Create a new block execution context let mut block_execution_context: BlockExecutionContext = @@ -389,6 +443,9 @@ where } .into(); + #[cfg(debug_assertions)] + 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 +459,9 @@ where tracing::debug!(block_fees = ?processed_block_fees, "block fees are processed"); + #[cfg(debug_assertions)] + 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 +475,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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 +488,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + laps.lap("total_credits"); + let root_hash = self .drive .grove @@ -436,6 +502,9 @@ where .block_state_info_mut() .set_app_hash(Some(root_hash)); + #[cfg(debug_assertions)] + laps.lap("root_hash"); + let validator_set_update = self.validator_set_update( block_proposal.proposer_pro_tx_hash, last_committed_platform_state, @@ -443,6 +512,9 @@ where platform_version, )?; + #[cfg(debug_assertions)] + 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..9e17460ef7c 100644 --- a/packages/rs-drive-abci/src/lib.rs +++ b/packages/rs-drive-abci/src/lib.rs @@ -69,6 +69,10 @@ pub mod core; /// Metrics subsystem pub mod metrics; +/// Per-block phase timing for debug builds, enabled with DRIVE_BLOCK_PERF=1. +#[cfg(debug_assertions)] +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..94a5de67aba --- /dev/null +++ b/packages/rs-drive-abci/src/perf.rs @@ -0,0 +1,249 @@ +//! Lightweight per-block phase timing for debug builds. +//! +//! This module and all timing call sites are compiled only with debug assertions +//! enabled. Standard release builds exclude the instrumentation entirely, even +//! when `DRIVE_BLOCK_PERF=1` is set. +//! +//! In debug builds, 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. +//! +//! This is read straight from the environment rather than through +//! `PlatformConfig` on purpose: it is a developer switch for replay benchmarks, +//! it must cost nothing when off, and it should not need a config change to be +//! flipped on a node under investigation. +//! +//! Phases nest where a handler times a call whose body is itself timed: the +//! `fb_proposal` lap in `finalize_block` covers all of the `fbp_*` laps taken +//! inside `finalize_block_proposal`. Add up laps from one level only. + +use std::sync::{Mutex, OnceLock, PoisonError}; +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)); + } + } + + /// One `name=mean/samples` term per phase, space separated. The mean is + /// over blocks, not over samples: a phase that only runs on some blocks + /// shows its share of the per-block cost, and the sample count shows how + /// often it ran. Only called from `end_block`, after a block was counted. + fn report_line(&self) -> String { + debug_assert!(self.blocks > 0, "report_line before any block was counted"); + let mut line = String::with_capacity(self.phases.len() * 20); + for (name, sum, samples) in &self.phases { + if !line.is_empty() { + line.push(' '); + } + line.push_str(name); + line.push('='); + line.push_str(&(*sum / self.blocks.max(1)).to_string()); + line.push('/'); + line.push_str(&samples.to_string()); + } + line + } + + /// Counts a finished block. Returns the report and resets when the + /// reporting interval is reached. An interval of zero reports every block. + fn end_block(&mut self, every: u64) -> Option<(u64, String)> { + self.blocks += 1; + if self.blocks < every { + return None; + } + let report = (self.blocks, self.report_line()); + self.phases.clear(); + self.blocks = 0; + Some(report) + } +} + +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) { + self.lap_if(true, name); + } + + /// Like [`lap`](Self::lap), but only records a sample when `ran` is true. + /// Use it after work that runs on some blocks only, so the sample count in + /// the report is the number of blocks the work actually ran on. The lap + /// boundary moves either way. + pub fn lap_if(&mut self, ran: bool, name: &'static str) { + if !self.on { + return; + } + let now = Instant::now(); + if ran { + self.buf + .push((name, now.duration_since(self.last).as_micros() as u64)); + } + self.last = now; + } +} + +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; + } + // Telemetry only: a panic elsewhere while the lock was held must not + // turn into a second panic here, least of all during unwinding. + let mut totals = totals().lock().unwrap_or_else(PoisonError::into_inner); + for (name, micros) in self.buf.drain(..) { + totals.add(name, micros); + } + } +} + +/// 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 report = totals() + .lock() + .unwrap_or_else(PoisonError::into_inner) + .end_block(report_every()); + if let Some((blocks, line)) = report { + tracing::info!( + block_perf = "agg", + height, + blocks, + phases = line, + "block perf" + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn phases_keep_first_seen_order_and_sum_samples() { + let mut totals = Totals::default(); + totals.add("b", 10); + totals.add("a", 5); + totals.add("b", 20); + + assert_eq!(totals.phases, vec![("b", 30, 2), ("a", 5, 1)]); + } + + #[test] + fn report_means_over_blocks_not_over_samples() { + let mut totals = Totals::default(); + // Ran on one block out of four, costing 400 µs that time. + totals.add("rare", 400); + // Ran on every block. + for _ in 0..4 { + totals.add("common", 10); + } + totals.blocks = 4; + + assert_eq!(totals.report_line(), "rare=100/1 common=10/4"); + } + + #[test] + fn end_block_reports_and_resets_at_the_interval() { + let mut totals = Totals::default(); + totals.add("x", 30); + assert_eq!(totals.end_block(3), None); + totals.add("x", 30); + assert_eq!(totals.end_block(3), None); + totals.add("x", 30); + + assert_eq!(totals.end_block(3), Some((3, "x=30/3".to_string()))); + assert_eq!(totals.blocks, 0); + assert!(totals.phases.is_empty()); + } + + #[test] + fn a_lap_that_did_not_run_moves_the_boundary_without_a_sample() { + let mut laps = Laps { + last: Instant::now(), + on: true, + buf: Vec::new(), + }; + let before = laps.last; + laps.lap_if(false, "skipped"); + assert!(laps.buf.is_empty()); + assert!(laps.last >= before); + + laps.lap_if(true, "ran"); + assert_eq!(laps.buf.len(), 1); + assert_eq!(laps.buf[0].0, "ran"); + // Drop must not merge test laps into the process-wide totals. + laps.buf.clear(); + } + + #[test] + fn report_line_is_empty_when_nothing_was_recorded() { + let mut totals = Totals::default(); + assert_eq!(totals.end_block(1), Some((1, String::new()))); + } +}