starknet_committer,apollo_committer: add timers for fetch patricia paths

Author: ArielElp

crates/apollo_committer/src/committer.rs

@@ -549,11 +549,15 @@ where
             // 4. Merge the two sets of patricia paths and write the result to the storage.
             // 5. Update the commitment offset and return the global root and the patricia proofs.
             CommitBlockHeightPlan::CommitTip { state_diff_commitment } => {
+                let mut block_measurements = SingleBlockMeasurements::default();
+                block_measurements.start_measurement(Action::EndToEnd);
+
                 let pre_roots = self
                     .forest_storage
                     .read_roots(ForestDB::InitialReadContext::create_empty())
                     .await
                     .map_err(|e| self.map_internal_error(e))?;
+                block_measurements.start_measurement(Action::FetchWitnessesFirstPass);
                 let mut patricia_proofs = self
                     .forest_storage
                     .fetch_patricia_witnesses(
@@ -569,16 +573,21 @@ where
                         height,
                         message: format!("pre-commit witness paths: {e:?}"),
                     })?;
+                block_measurements
+                    .attempt_to_stop_measurement(
+                        Action::FetchWitnessesFirstPass,
+                        patricia_proofs.len(),
+                    )
+                    .ok();
 
-                let mut block_measurements = SingleBlockMeasurements::default();
-                block_measurements.start_measurement(Action::EndToEnd);
                 let CommitStateDiffOutput { filled_forest, global_root, deleted_nodes } =
                     self.commit_state_diff(state_diff, &mut block_measurements).await?;
                 let post_roots = filled_forest.state_roots();
 
                 let forest_updates = ForestDB::serialize_forest(&filled_forest)
                     .map_err(|e| self.map_internal_error(e))?;
 
+                block_measurements.start_measurement(Action::FetchWitnessesSecondPass);
                 let proof_after = self
                     .forest_storage
                     .fetch_patricia_witnesses(
@@ -594,23 +603,24 @@ where
                         height,
                         message: format!("post-commit witness paths: {e:?}"),
                     })?;
+                block_measurements
+                    .attempt_to_stop_measurement(
+                        Action::FetchWitnessesSecondPass,
+                        proof_after.len(),
+                    )
+                    .ok();
 
                 patricia_proofs.extend(proof_after);
 
                 let (metadata, next_offset) =
                     commit_tip_metadata_bundle(height, global_root, state_diff_commitment);
-                let witness_node_count = patricia_proofs.classes_trie_proof.len()
-                    + patricia_proofs.contracts_trie_proof.nodes.len()
-                    + patricia_proofs.contracts_trie_proof.leaves.len()
-                    + patricia_proofs
-                        .contracts_trie_storage_proofs
-                        .values()
-                        .map(|proof| proof.len())
-                        .sum::<usize>();
+
                 info!(
-                    "For block number {height}, writing filled forest and {witness_node_count} \
-                     witness nodes to storage with metadata: {metadata:?}, delete {} nodes",
-                    deleted_nodes.len()
+                    "For block number {height}, writing filled forest and {witness_count} \
+                     witnesses to storage with metadata: {metadata:?}, delete \
+                     {deleted_nodes_count} nodes",
+                    witness_count = patricia_proofs.len(),
+                    deleted_nodes_count = deleted_nodes.len(),
                 );
                 block_measurements.start_measurement(Action::Write);
                 let n_write_entries = self
@@ -669,9 +679,16 @@ impl ComponentStarter for ApolloCommitter {
 }
 
 #[allow(clippy::as_conversions)]
+// TODO: Consider adding fetch witnesses measurements.
 fn update_metrics(
     height: BlockNumber,
-    BlockMeasurement { n_reads, n_writes, durations, modifications_counts }: &BlockMeasurement,
+    BlockMeasurement {
+        n_reads,
+        n_writes,
+        durations,
+        modifications_counts,
+        fetched_witnesses_count,
+    }: &BlockMeasurement,
 ) {
     BLOCKS_COMMITTED.increment(1);
     TOTAL_BLOCK_DURATION.increment((durations.block * 1000.0) as u64);
@@ -736,6 +753,7 @@ fn update_metrics(
         write_rate,
         modifications_counts,
         emptied_leaves_percentage,
+        *fetched_witnesses_count,
     );
 }
 
@@ -749,12 +767,24 @@ fn log_block_measurements(
     write_rate: Option<f64>,
     modifications_counts: &BlockModificationsCounts,
     emptied_leaves_percentage: Option<f64>,
+    fetched_witnesses_count: usize,
 ) {
+    #[cfg(feature = "os_input")]
+    let witness_log = format!(
+        "witness fetch ms (pre-commit/post-commit): {:.0}/{:.0}, witness entries: {}",
+        durations.fetch_witnesses_first_pass * 1000.0,
+        durations.fetch_witnesses_second_pass * 1000.0,
+        fetched_witnesses_count,
+    );
+    #[cfg(not(feature = "os_input"))]
+    let witness_log = String::new();
+
     debug!(
         "Block {height} stats: durations in ms (total/read/compute/write): \
          {:.0}/{:.0}/{:.0}/{:.0}, total block duration per modification in µs: {}, rates in \
          entries/sec (read/compute/write): {}/{}/{}, modifications count \
-         (storage_tries/contracts_trie/classes_trie/emptied_storage_leaves): {}/{}/{}/{}{}",
+         (storage_tries/contracts_trie/classes_trie/emptied_storage_leaves): {}/{}/{}/{}{}, \
+         {witness_log}",
         durations.block * 1000.0,
         durations.read * 1000.0,
         durations.compute * 1000.0,
@@ -767,6 +797,7 @@ fn log_block_measurements(
         modifications_counts.contracts_trie,
         modifications_counts.classes_trie,
         modifications_counts.emptied_storage_leaves,
-        emptied_leaves_percentage.map_or(String::new(), |p| format!(" ({p:.2}%)"))
+        emptied_leaves_percentage.map_or(String::new(), |p| format!(" ({p:.2}%)")),
+        witness_log = witness_log,
     );
 }

crates/starknet_committer/src/block_committer/measurements_util.rs

@@ -11,6 +11,10 @@ pub enum Action {
     Read,
     Compute,
     Write,
+    #[cfg(feature = "os_input")]
+    FetchWitnessesFirstPass,
+    #[cfg(feature = "os_input")]
+    FetchWitnessesSecondPass,
 }
 
 #[derive(Default)]
@@ -19,6 +23,10 @@ pub struct BlockTimers {
     pub read_timer: Option<Instant>,
     pub compute_timer: Option<Instant>,
     pub writer_timer: Option<Instant>,
+    #[cfg(feature = "os_input")]
+    pub fetch_witnesses_first_pass_timer: Option<Instant>,
+    #[cfg(feature = "os_input")]
+    pub fetch_witnesses_second_pass_timer: Option<Instant>,
 }
 
 impl BlockTimers {
@@ -28,6 +36,10 @@ impl BlockTimers {
             Action::Read => &mut self.read_timer,
             Action::Compute => &mut self.compute_timer,
             Action::Write => &mut self.writer_timer,
+            #[cfg(feature = "os_input")]
+            Action::FetchWitnessesFirstPass => &mut self.fetch_witnesses_first_pass_timer,
+            #[cfg(feature = "os_input")]
+            Action::FetchWitnessesSecondPass => &mut self.fetch_witnesses_second_pass_timer,
         }
     }
 
@@ -91,6 +103,12 @@ pub struct BlockDurations {
     pub read: f64,    // Duration of a read phase (seconds).
     pub compute: f64, // Duration of a computation phase (seconds).
     pub write: f64,   // Duration of a write phase (seconds).
+    #[cfg(feature = "os_input")]
+    // Duration of fetching witnesses w.r.t the old root (seconds).
+    pub fetch_witnesses_first_pass: f64,
+    #[cfg(feature = "os_input")]
+    // Duration of fetching witnesses w.r.t the new root (seconds).
+    pub fetch_witnesses_second_pass: f64,
 }
 
 #[derive(Default, Clone, Debug, PartialEq, Eq)]
@@ -113,6 +131,8 @@ pub struct BlockMeasurement {
     pub n_reads: usize,
     pub durations: BlockDurations,
     pub modifications_counts: BlockModificationsCounts,
+    // Number of witnesses fetched in the first pass (pre-commit).
+    pub fetched_witnesses_count: usize,
 }
 
 impl BlockMeasurement {
@@ -137,6 +157,15 @@ impl BlockMeasurement {
             Action::EndToEnd => {
                 self.durations.block = duration_in_seconds;
             }
+            #[cfg(feature = "os_input")]
+            Action::FetchWitnessesFirstPass => {
+                self.durations.fetch_witnesses_first_pass = duration_in_seconds;
+                self.fetched_witnesses_count += entries_count;
+            }
+            #[cfg(feature = "os_input")]
+            Action::FetchWitnessesSecondPass => {
+                self.durations.fetch_witnesses_second_pass = duration_in_seconds;
+            }
         }
     }
 }

crates/starknet_committer/src/patricia_merkle_tree/types.rs

@@ -65,6 +65,15 @@ impl StarknetForestProofs {
             self.contracts_trie_storage_proofs.entry(address).or_default().extend(proof);
         }
     }
+
+    pub fn len(&self) -> usize {
+        self.classes_trie_proof.len()
+            + self.contracts_trie_proof.nodes.len()
+            + self
+                .contracts_trie_storage_proofs
+                .values()
+                .fold(0, |count, proofs| count + proofs.len())
+    }
 }
 
 pub struct RootHashes {

crates/starknet_committer_cli/Cargo.toml

@@ -31,3 +31,6 @@ starknet_patricia_storage = { workspace = true, features = [
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 tracing.workspace = true
 tracing-subscriber.workspace = true
+
+[features]
+os_input = ["starknet_committer/os_input"]