Track transaction performance through various stage using random mask

Cargo.toml

@@ -106,6 +106,7 @@ members = [
     "tokens",
     "tpu-client",
     "transaction-dos",
+    "transaction-metrics-tracker",
     "transaction-status",
     "turbine",
     "udp-client",
@@ -378,6 +379,7 @@ solana-system-program = { path = "programs/system", version = "=1.19.0" }
 solana-test-validator = { path = "test-validator", version = "=1.19.0" }
 solana-thin-client = { path = "thin-client", version = "=1.19.0" }
 solana-tpu-client = { path = "tpu-client", version = "=1.19.0", default-features = false }
+solana-transaction-metrics-tracker = { path = "transaction-metrics-tracker", version = "=1.19.0" }
 solana-transaction-status = { path = "transaction-status", version = "=1.19.0" }
 solana-turbine = { path = "turbine", version = "=1.19.0" }
 solana-udp-client = { path = "udp-client", version = "=1.19.0" }

core/Cargo.toml

@@ -67,6 +67,7 @@ solana-send-transaction-service = { workspace = true }
 solana-streamer = { workspace = true }
 solana-svm = { workspace = true }
 solana-tpu-client = { workspace = true }
+solana-transaction-metrics-tracker = { workspace = true }
 solana-transaction-status = { workspace = true }
 solana-turbine = { workspace = true }
 solana-unified-scheduler-pool = { workspace = true }

core/src/banking_stage/consumer.rs

@@ -208,6 +208,32 @@ impl Consumer {
             .slot_metrics_tracker
             .increment_retryable_packets_count(retryable_transaction_indexes.len() as u64);
 
+        // Now we track the performance for the interested transactions which is not in the retryable_transaction_indexes
+        // We assume the retryable_transaction_indexes is already sorted.
+        let mut retryable_idx = 0;
+        for (index, packet) in packets_to_process.iter().enumerate() {
+            if packet.original_packet().meta().is_perf_track_packet() {
+                if let Some(start_time) = packet.start_time() {
+                    if retryable_idx >= retryable_transaction_indexes.len()
+                        || retryable_transaction_indexes[retryable_idx] != index
+                    {
+                        let duration = Instant::now().duration_since(*start_time);
+
+                        debug!(
+                            "Banking stage processing took {duration:?} for transaction {:?}",
+                            packet.transaction().get_signatures().first()
+                        );
+                        payload
+                            .slot_metrics_tracker
+                            .increment_process_sampled_packets_us(duration.as_micros() as u64);
+                    } else {
+                        // This packet is retried, advance the retry index to the next, as the next packet's index will
+                        // certainly be > than this.
+                        retryable_idx += 1;
+                    }
+                }
+            }
+        }
         Some(retryable_transaction_indexes)
     }
 

core/src/banking_stage/immutable_deserialized_packet.rs

@@ -13,7 +13,7 @@ use {
             VersionedTransaction,
         },
     },
-    std::{cmp::Ordering, mem::size_of, sync::Arc},
+    std::{cmp::Ordering, mem::size_of, sync::Arc, time::Instant},
     thiserror::Error,
 };
 
@@ -41,10 +41,16 @@ pub struct ImmutableDeserializedPacket {
     message_hash: Hash,
     is_simple_vote: bool,
     compute_budget_details: ComputeBudgetDetails,
+    banking_stage_start_time: Option<Instant>,
 }
 
 impl ImmutableDeserializedPacket {
     pub fn new(packet: Packet) -> Result<Self, DeserializedPacketError> {
+        let banking_stage_start_time = packet
+            .meta()
+            .is_perf_track_packet()
+            .then_some(Instant::now());
+
         let versioned_transaction: VersionedTransaction = packet.deserialize_slice(..)?;
         let sanitized_transaction = SanitizedVersionedTransaction::try_from(versioned_transaction)?;
         let message_bytes = packet_message(&packet)?;
@@ -67,6 +73,7 @@ impl ImmutableDeserializedPacket {
             message_hash,
             is_simple_vote,
             compute_budget_details,
+            banking_stage_start_time,
         })
     }
 
@@ -98,6 +105,10 @@ impl ImmutableDeserializedPacket {
         self.compute_budget_details.clone()
     }
 
+    pub fn start_time(&self) -> &Option<Instant> {
+        &self.banking_stage_start_time
+    }
+
     // This function deserializes packets into transactions, computes the blake3 hash of transaction
     // messages, and verifies secp256k1 instructions.
     pub fn build_sanitized_transaction(

core/src/banking_stage/leader_slot_metrics.rs

@@ -936,6 +936,17 @@ impl LeaderSlotMetricsTracker {
             );
         }
     }
+
+    pub(crate) fn increment_process_sampled_packets_us(&mut self, us: u64) {
+        if let Some(leader_slot_metrics) = &mut self.leader_slot_metrics {
+            leader_slot_metrics
+                .timing_metrics
+                .process_packets_timings
+                .process_sampled_packets_us_hist
+                .increment(us)
+                .unwrap();
+        }
+    }
 }
 
 #[cfg(test)]

core/src/banking_stage/leader_slot_timing_metrics.rs

@@ -244,6 +244,9 @@ pub(crate) struct ProcessPacketsTimings {
     // Time spent running the cost model in processing transactions before executing
     // transactions
     pub cost_model_us: u64,
+
+    // banking stage processing time histogram for sampled packets
+    pub process_sampled_packets_us_hist: histogram::Histogram,
 }
 
 impl ProcessPacketsTimings {
@@ -264,6 +267,28 @@ impl ProcessPacketsTimings {
                 i64
             ),
             ("cost_model_us", self.cost_model_us, i64),
+            (
+                "process_sampled_packets_us_90pct",
+                self.process_sampled_packets_us_hist
+                    .percentile(90.0)
+                    .unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_min",
+                self.process_sampled_packets_us_hist.minimum().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_max",
+                self.process_sampled_packets_us_hist.maximum().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_mean",
+                self.process_sampled_packets_us_hist.mean().unwrap_or(0),
+                i64
+            ),
         );
     }
 }

core/src/banking_stage/unprocessed_transaction_storage.rs

@@ -924,6 +924,7 @@ impl ThreadLocalUnprocessedPackets {
                 .iter()
                 .map(|p| (*p).clone())
                 .collect_vec();
+
             let retryable_packets = if let Some(retryable_transaction_indexes) =
                 processing_function(&packets_to_process, payload)
             {

core/src/sigverify_stage.rs

@@ -18,8 +18,9 @@ use {
             count_discarded_packets, count_packets_in_batches, count_valid_packets, shrink_batches,
         },
     },
-    solana_sdk::timing,
+    solana_sdk::{signature::Signature, timing},
     solana_streamer::streamer::{self, StreamerError},
+    solana_transaction_metrics_tracker::get_signature_from_packet,
     std::{
         thread::{self, Builder, JoinHandle},
         time::Instant,
@@ -78,8 +79,9 @@ struct SigVerifierStats {
     verify_batches_pp_us_hist: histogram::Histogram, // per-packet time to call verify_batch
     discard_packets_pp_us_hist: histogram::Histogram, // per-packet time to call verify_batch
     dedup_packets_pp_us_hist: histogram::Histogram, // per-packet time to call verify_batch
-    batches_hist: histogram::Histogram,         // number of packet batches per verify call
-    packets_hist: histogram::Histogram,         // number of packets per verify call
+    process_sampled_packets_us_hist: histogram::Histogram, // per-packet time do do overall verify for sampled packets
+    batches_hist: histogram::Histogram, // number of packet batches per verify call
+    packets_hist: histogram::Histogram, // number of packets per verify call
     num_deduper_saturations: usize,
     total_batches: usize,
     total_packets: usize,
@@ -93,6 +95,7 @@ struct SigVerifierStats {
     total_discard_random_time_us: usize,
     total_verify_time_us: usize,
     total_shrink_time_us: usize,
+    perf_track_overhead_us: usize,
 }
 
 impl SigVerifierStats {
@@ -181,6 +184,28 @@ impl SigVerifierStats {
                 self.dedup_packets_pp_us_hist.mean().unwrap_or(0),
                 i64
             ),
+            (
+                "process_sampled_packets_us_90pct",
+                self.process_sampled_packets_us_hist
+                    .percentile(90.0)
+                    .unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_min",
+                self.process_sampled_packets_us_hist.minimum().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_max",
+                self.process_sampled_packets_us_hist.maximum().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_mean",
+                self.process_sampled_packets_us_hist.mean().unwrap_or(0),
+                i64
+            ),
             (
                 "batches_90pct",
                 self.batches_hist.percentile(90.0).unwrap_or(0),
@@ -214,6 +239,7 @@ impl SigVerifierStats {
             ),
             ("total_verify_time_us", self.total_verify_time_us, i64),
             ("total_shrink_time_us", self.total_shrink_time_us, i64),
+            ("perf_track_overhead_us", self.perf_track_overhead_us, i64),
         );
     }
 }
@@ -296,8 +322,26 @@ impl SigVerifyStage {
         verifier: &mut T,
         stats: &mut SigVerifierStats,
     ) -> Result<(), T::SendType> {
+        let mut packet_perf_measure: Vec<([u8; 64], std::time::Instant)> = Vec::default();
+
         let (mut batches, num_packets, recv_duration) = streamer::recv_packet_batches(recvr)?;
 
+        let mut start_perf_track_measure = Measure::start("start_perf_track");
+        // track sigverify start time for interested packets
+        for batch in &batches {
+            for packet in batch.iter() {
+                if packet.meta().is_perf_track_packet() {
+                    let signature = get_signature_from_packet(packet);
+                    if let Ok(signature) = signature {
+                        packet_perf_measure.push((*signature, Instant::now()));
+                    }
+                }
+            }
+        }
+        start_perf_track_measure.stop();
+
+        stats.perf_track_overhead_us = start_perf_track_measure.as_us() as usize;
+
         let batches_len = batches.len();
         debug!(
             "@{:?} verifier: verifying: {}",
@@ -370,6 +414,22 @@ impl SigVerifyStage {
             (num_packets as f32 / verify_time.as_s())
         );
 
+        let mut perf_track_end_measure = Measure::start("perf_track_end");
+        for (signature, start_time) in packet_perf_measure.iter() {
+            let duration = Instant::now().duration_since(*start_time);
+            debug!(
+                "Sigverify took {duration:?} for transaction {:?}",
+                Signature::from(*signature)
+            );
+            stats
+                .process_sampled_packets_us_hist
+                .increment(duration.as_micros() as u64)
+                .unwrap();
+        }
+
+        perf_track_end_measure.stop();
+        stats.perf_track_overhead_us += perf_track_end_measure.as_us() as usize;
+
         stats
             .recv_batches_us_hist
             .increment(recv_duration.as_micros() as u64)