Track transaction performance.v1.17.enhanced

Cargo.toml

@@ -364,6 +364,7 @@ solana-system-program = { path = "programs/system", version = "=1.17.30" }
 solana-test-validator = { path = "test-validator", version = "=1.17.30" }
 solana-thin-client = { path = "thin-client", version = "=1.17.30" }
 solana-tpu-client = { path = "tpu-client", version = "=1.17.30", default-features = false }
+solana-transaction-metrics-tracker = { path = "transaction-metrics-tracker", version = "=1.17.30" }
 solana-transaction-status = { path = "transaction-status", version = "=1.17.30" }
 solana-turbine = { path = "turbine", version = "=1.17.30" }
 solana-udp-client = { path = "udp-client", version = "=1.17.30" }

core/Cargo.toml

@@ -66,6 +66,7 @@ solana-sdk = { workspace = true }
 solana-send-transaction-service = { workspace = true }
 solana-streamer = { workspace = true }
 solana-tpu-client = { workspace = true }
+solana-transaction-metrics-tracker = { workspace = true }
 solana-transaction-status = { workspace = true }
 solana-turbine = { workspace = true }
 solana-version = { workspace = true }

core/src/banking_stage/consumer.rs

@@ -201,6 +201,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

@@ -17,7 +17,7 @@ use {
             VersionedTransaction,
         },
     },
-    std::{cmp::Ordering, mem::size_of, sync::Arc},
+    std::{cmp::Ordering, mem::size_of, sync::Arc, time::Instant},
     thiserror::Error,
 };
 
@@ -45,10 +45,16 @@ pub struct ImmutableDeserializedPacket {
     message_hash: Hash,
     is_simple_vote: bool,
     priority_details: TransactionPriorityDetails,
+    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)?;
@@ -71,6 +77,7 @@ impl ImmutableDeserializedPacket {
             message_hash,
             is_simple_vote,
             priority_details,
+            banking_stage_start_time,
         })
     }
 
@@ -114,6 +121,10 @@ impl ImmutableDeserializedPacket {
         self.compute_unit_limit() >= static_builtin_cost_sum
     }
 
+    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

@@ -839,6 +839,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

@@ -876,6 +876,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)

sdk/src/packet.rs

@@ -33,6 +33,8 @@ bitflags! {
         /// the packet is built.
         /// This field can be removed when the above feature gate is adopted by mainnet-beta.
         const ROUND_COMPUTE_UNIT_PRICE = 0b0010_0000;
+        /// For tracking performance
+        const PERF_TRACK_PACKET  = 0b0100_0000;
     }
 }
 
@@ -228,6 +230,12 @@ impl Meta {
         self.flags.set(PacketFlags::TRACER_PACKET, is_tracer);
     }
 
+    #[inline]
+    pub fn set_track_performance(&mut self, is_performance_track: bool) {
+        self.flags
+            .set(PacketFlags::PERF_TRACK_PACKET, is_performance_track);
+    }
+
     #[inline]
     pub fn set_simple_vote(&mut self, is_simple_vote: bool) {
         self.flags.set(PacketFlags::SIMPLE_VOTE_TX, is_simple_vote);
@@ -261,6 +269,11 @@ impl Meta {
         self.flags.contains(PacketFlags::TRACER_PACKET)
     }
 
+    #[inline]
+    pub fn is_perf_track_packet(&self) -> bool {
+        self.flags.contains(PacketFlags::PERF_TRACK_PACKET)
+    }
+
     #[inline]
     pub fn round_compute_unit_price(&self) -> bool {
         self.flags.contains(PacketFlags::ROUND_COMPUTE_UNIT_PRICE)