feat(tree): measure different parts of canonicalization (#5266)

crates/storage/provider/Cargo.toml

@@ -25,6 +25,10 @@ tokio-stream = { workspace = true, features = ["sync"] }
 # tracing
 tracing.workspace = true
 
+# metrics
+reth-metrics.workspace = true
+metrics.workspace = true
+
 # misc
 auto_impl = "1.0"
 itertools.workspace = true

crates/storage/provider/src/providers/database/metrics.rs

@@ -0,0 +1,102 @@
+use metrics::Histogram;
+use reth_metrics::Metrics;
+use std::time::{Duration, Instant};
+
+#[derive(Debug)]
+pub(crate) struct DurationsRecorder {
+    start: Instant,
+    pub(crate) actions: Vec<(Action, Duration)>,
+    latest: Option<Duration>,
+}
+
+impl Default for DurationsRecorder {
+    fn default() -> Self {
+        Self { start: Instant::now(), actions: Vec::new(), latest: None }
+    }
+}
+
+impl DurationsRecorder {
+    /// Saves the provided duration for future logging and instantly reports as a metric with
+    /// `action` label.
+    pub(crate) fn record_duration(&mut self, action: Action, duration: Duration) {
+        self.actions.push((action, duration));
+        Metrics::new_with_labels(&[("action", format!("{action:?}"))]).duration.record(duration);
+        self.latest = Some(self.start.elapsed());
+    }
+
+    /// Records the duration since last record, saves it for future logging and instantly reports as
+    /// a metric with `action` label.
+    pub(crate) fn record_relative(&mut self, action: Action) {
+        let elapsed = self.start.elapsed();
+        let duration = elapsed - self.latest.unwrap_or_default();
+
+        self.actions.push((action, duration));
+        Metrics::new_with_labels(&[("action", action.as_str())]).duration.record(duration);
+
+        self.latest = Some(elapsed);
+    }
+}
+
+#[derive(Debug, Copy, Clone)]
+pub(crate) enum Action {
+    InsertStorageHashing,
+    InsertAccountHashing,
+    InsertMerkleTree,
+    InsertBlock,
+    InsertState,
+    InsertHashes,
+    InsertHistoryIndices,
+    UpdatePipelineStages,
+    InsertCanonicalHeaders,
+    InsertHeaders,
+    InsertHeaderNumbers,
+    InsertHeaderTD,
+    InsertBlockOmmers,
+    InsertTxSenders,
+    InsertTransactions,
+    InsertTxHashNumbers,
+    InsertBlockWithdrawals,
+    InsertBlockBodyIndices,
+    InsertTransactionBlock,
+
+    RecoverSigners,
+    GetNextTxNum,
+    GetParentTD,
+}
+
+impl Action {
+    fn as_str(&self) -> &'static str {
+        match self {
+            Action::InsertStorageHashing => "insert storage hashing",
+            Action::InsertAccountHashing => "insert account hashing",
+            Action::InsertMerkleTree => "insert merkle tree",
+            Action::InsertBlock => "insert block",
+            Action::InsertState => "insert state",
+            Action::InsertHashes => "insert hashes",
+            Action::InsertHistoryIndices => "insert history indices",
+            Action::UpdatePipelineStages => "update pipeline stages",
+            Action::InsertCanonicalHeaders => "insert canonical headers",
+            Action::InsertHeaders => "insert headers",
+            Action::InsertHeaderNumbers => "insert header numbers",
+            Action::InsertHeaderTD => "insert header TD",
+            Action::InsertBlockOmmers => "insert block ommers",
+            Action::InsertTxSenders => "insert tx senders",
+            Action::InsertTransactions => "insert transactions",
+            Action::InsertTxHashNumbers => "insert tx hash numbers",
+            Action::InsertBlockWithdrawals => "insert block withdrawals",
+            Action::InsertBlockBodyIndices => "insert block body indices",
+            Action::InsertTransactionBlock => "insert transaction block",
+            Action::RecoverSigners => "recover signers",
+            Action::GetNextTxNum => "get next tx num",
+            Action::GetParentTD => "get parent TD",
+        }
+    }
+}
+
+#[derive(Metrics)]
+#[metrics(scope = "storage.providers.database")]
+/// Database provider metrics
+struct Metrics {
+    /// The time it took to execute an action
+    duration: Histogram,
+}

crates/storage/provider/src/providers/database/mod.rs

@@ -21,7 +21,9 @@ use std::{
 };
 use tracing::trace;
 
+mod metrics;
 mod provider;
+
 pub use provider::{DatabaseProvider, DatabaseProviderRO, DatabaseProviderRW};
 
 /// A common provider that fetches data from a database.

crates/storage/provider/src/providers/database/provider.rs

@@ -1,5 +1,6 @@
 use crate::{
     bundle_state::{BundleStateInit, BundleStateWithReceipts, RevertsInit},
+    providers::database::metrics,
     traits::{
         AccountExtReader, BlockSource, ChangeSetReader, ReceiptProvider, StageCheckpointWriter,
     },
@@ -48,7 +49,9 @@ use std::{
     fmt::Debug,
     ops::{Deref, DerefMut, Range, RangeBounds, RangeInclusive},
     sync::{mpsc, Arc},
+    time::{Duration, Instant},
 };
+use tracing::debug;
 
 /// A [`DatabaseProvider`] that holds a read-only database transaction.
 pub type DatabaseProviderRO<'this, DB> = DatabaseProvider<<DB as DatabaseGAT<'this>>::TX>;
@@ -1600,6 +1603,8 @@ impl<TX: DbTxMut + DbTx> HashingWriter for DatabaseProvider<TX> {
         let mut storage_prefix_set: HashMap<B256, PrefixSetMut> = HashMap::default();
         let mut destroyed_accounts = HashSet::default();
 
+        let mut durations_recorder = metrics::DurationsRecorder::default();
+
         // storage hashing stage
         {
             let lists = self.changed_storages_with_range(range.clone())?;
@@ -1615,6 +1620,7 @@ impl<TX: DbTxMut + DbTx> HashingWriter for DatabaseProvider<TX> {
                 }
             }
         }
+        durations_recorder.record_relative(metrics::Action::InsertStorageHashing);
 
         // account hashing stage
         {
@@ -1628,6 +1634,7 @@ impl<TX: DbTxMut + DbTx> HashingWriter for DatabaseProvider<TX> {
                 }
             }
         }
+        durations_recorder.record_relative(metrics::Action::InsertAccountHashing);
 
         // merkle tree
         {
@@ -1652,6 +1659,10 @@ impl<TX: DbTxMut + DbTx> HashingWriter for DatabaseProvider<TX> {
             }
             trie_updates.flush(&self.tx)?;
         }
+        durations_recorder.record_relative(metrics::Action::InsertMerkleTree);
+
+        debug!(target: "providers::db", ?range, actions = ?durations_recorder.actions, "Inserted hashes");
+
         Ok(())
     }
 
@@ -1824,7 +1835,7 @@ impl<TX: DbTxMut + DbTx> HashingWriter for DatabaseProvider<TX> {
 }
 
 impl<TX: DbTxMut + DbTx> HistoryWriter for DatabaseProvider<TX> {
-    fn calculate_history_indices(&self, range: RangeInclusive<BlockNumber>) -> RethResult<()> {
+    fn update_history_indices(&self, range: RangeInclusive<BlockNumber>) -> RethResult<()> {
         // account history stage
         {
             let indices = self.changed_accounts_and_blocks_with_range(range.clone())?;
@@ -2062,28 +2073,39 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
         prune_modes: Option<&PruneModes>,
     ) -> RethResult<StoredBlockBodyIndices> {
         let block_number = block.number;
-        self.tx.put::<tables::CanonicalHeaders>(block.number, block.hash())?;
+
+        let mut durations_recorder = metrics::DurationsRecorder::default();
+
+        self.tx.put::<tables::CanonicalHeaders>(block_number, block.hash())?;
+        durations_recorder.record_relative(metrics::Action::InsertCanonicalHeaders);
+
         // Put header with canonical hashes.
-        self.tx.put::<tables::Headers>(block.number, block.header.as_ref().clone())?;
-        self.tx.put::<tables::HeaderNumbers>(block.hash(), block.number)?;
+        self.tx.put::<tables::Headers>(block_number, block.header.as_ref().clone())?;
+        durations_recorder.record_relative(metrics::Action::InsertHeaders);
+
+        self.tx.put::<tables::HeaderNumbers>(block.hash(), block_number)?;
+        durations_recorder.record_relative(metrics::Action::InsertHeaderNumbers);
 
         // total difficulty
-        let ttd = if block.number == 0 {
+        let ttd = if block_number == 0 {
             block.difficulty
         } else {
-            let parent_block_number = block.number - 1;
+            let parent_block_number = block_number - 1;
             let parent_ttd = self.header_td_by_number(parent_block_number)?.unwrap_or_default();
+            durations_recorder.record_relative(metrics::Action::GetParentTD);
             parent_ttd + block.difficulty
         };
 
-        self.tx.put::<tables::HeaderTD>(block.number, ttd.into())?;
+        self.tx.put::<tables::HeaderTD>(block_number, ttd.into())?;
+        durations_recorder.record_relative(metrics::Action::InsertHeaderTD);
 
         // insert body ommers data
         if !block.ommers.is_empty() {
             self.tx.put::<tables::BlockOmmers>(
-                block.number,
+                block_number,
                 StoredBlockOmmers { ommers: block.ommers },
             )?;
+            durations_recorder.record_relative(metrics::Action::InsertBlockOmmers);
         }
 
         let mut next_tx_num = self
@@ -2092,6 +2114,7 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
             .last()?
             .map(|(n, _)| n + 1)
             .unwrap_or_default();
+        durations_recorder.record_relative(metrics::Action::GetNextTxNum);
         let first_tx_num = next_tx_num;
 
         let tx_count = block.body.len() as u64;
@@ -2103,10 +2126,14 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
             let senders = TransactionSigned::recover_signers(&block.body, block.body.len()).ok_or(
                 BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError),
             )?;
+            durations_recorder.record_relative(metrics::Action::RecoverSigners);
             debug_assert_eq!(senders.len(), block.body.len(), "missing one or more senders");
             block.body.into_iter().zip(senders).collect()
         };
 
+        let mut tx_senders_elapsed = Duration::default();
+        let mut transactions_elapsed = Duration::default();
+        let mut tx_hash_numbers_elapsed = Duration::default();
         for (transaction, sender) in tx_iter {
             let hash = transaction.hash();
 
@@ -2115,20 +2142,31 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
                 .filter(|prune_mode| prune_mode.is_full())
                 .is_none()
             {
+                let start = Instant::now();
                 self.tx.put::<tables::TxSenders>(next_tx_num, sender)?;
+                tx_senders_elapsed += start.elapsed();
             }
 
+            let start = Instant::now();
             self.tx.put::<tables::Transactions>(next_tx_num, transaction.into())?;
+            transactions_elapsed += start.elapsed();
 
             if prune_modes
                 .and_then(|modes| modes.transaction_lookup)
                 .filter(|prune_mode| prune_mode.is_full())
                 .is_none()
             {
+                let start = Instant::now();
                 self.tx.put::<tables::TxHashNumber>(hash, next_tx_num)?;
+                tx_hash_numbers_elapsed += start.elapsed();
             }
             next_tx_num += 1;
         }
+        durations_recorder.record_duration(metrics::Action::InsertTxSenders, tx_senders_elapsed);
+        durations_recorder
+            .record_duration(metrics::Action::InsertTransactions, transactions_elapsed);
+        durations_recorder
+            .record_duration(metrics::Action::InsertTxHashNumbers, tx_hash_numbers_elapsed);
 
         if let Some(withdrawals) = block.withdrawals {
             if !withdrawals.is_empty() {
@@ -2136,16 +2174,26 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
                     block_number,
                     StoredBlockWithdrawals { withdrawals },
                 )?;
+                durations_recorder.record_relative(metrics::Action::InsertBlockWithdrawals);
             }
         }
 
         let block_indices = StoredBlockBodyIndices { first_tx_num, tx_count };
         self.tx.put::<tables::BlockBodyIndices>(block_number, block_indices.clone())?;
+        durations_recorder.record_relative(metrics::Action::InsertBlockBodyIndices);
 
         if !block_indices.is_empty() {
             self.tx.put::<tables::TransactionBlock>(block_indices.last_tx_num(), block_number)?;
+            durations_recorder.record_relative(metrics::Action::InsertTransactionBlock);
         }
 
+        debug!(
+            target: "providers::db",
+            ?block_number,
+            actions = ?durations_recorder.actions,
+            "Inserted block"
+        );
+
         Ok(block_indices)
     }
 
@@ -2168,22 +2216,31 @@ impl<TX: DbTxMut + DbTx> BlockWriter for DatabaseProvider<TX> {
         let last_block_hash = last.hash();
         let expected_state_root = last.state_root;
 
+        let mut durations_recorder = metrics::DurationsRecorder::default();
+
         // Insert the blocks
         for block in blocks {
             let (block, senders) = block.into_components();
             self.insert_block(block, Some(senders), prune_modes)?;
+            durations_recorder.record_relative(metrics::Action::InsertBlock);
         }
 
         // Write state and changesets to the database.
         // Must be written after blocks because of the receipt lookup.
         state.write_to_db(self.tx_ref(), OriginalValuesKnown::No)?;
+        durations_recorder.record_relative(metrics::Action::InsertState);
 
         self.insert_hashes(first_number..=last_block_number, last_block_hash, expected_state_root)?;
+        durations_recorder.record_relative(metrics::Action::InsertHashes);
 
-        self.calculate_history_indices(first_number..=last_block_number)?;
+        self.update_history_indices(first_number..=last_block_number)?;
+        durations_recorder.record_relative(metrics::Action::InsertHistoryIndices);
 
         // Update pipeline progress
         self.update_pipeline_stages(new_tip_number, false)?;
+        durations_recorder.record_relative(metrics::Action::UpdatePipelineStages);
+
+        debug!(target: "providers::db", actions = ?durations_recorder.actions, "Appended blocks");
 
         Ok(())
     }

crates/storage/provider/src/traits/history.rs

@@ -37,5 +37,5 @@ pub trait HistoryWriter: Send + Sync {
     ) -> RethResult<()>;
 
     /// Read account/storage changesets and update account/storage history indices.
-    fn calculate_history_indices(&self, range: RangeInclusive<BlockNumber>) -> RethResult<()>;
+    fn update_history_indices(&self, range: RangeInclusive<BlockNumber>) -> RethResult<()>;
 }