transaction performance tracking -- streamer stage (#257)

* transaction performance tracking -- streamer stage

Author: lijunwangs

Cargo.lock

@@ -107,6 +107,7 @@ members = [
     "tokens",
     "tpu-client",
     "transaction-dos",
+    "transaction-metrics-tracker",
     "transaction-status",
     "turbine",
     "udp-client",
@@ -380,6 +381,7 @@ solana-test-validator = { path = "test-validator", version = "=2.0.0" }
 solana-thin-client = { path = "thin-client", version = "=2.0.0" }
 solana-tpu-client = { path = "tpu-client", version = "=2.0.0", default-features = false }
 solana-transaction-status = { path = "transaction-status", version = "=2.0.0" }
+solana-transaction-metrics-tracker = { path = "transaction-metrics-tracker", version = "=2.0.0" }
 solana-turbine = { path = "turbine", version = "=2.0.0" }
 solana-udp-client = { path = "udp-client", version = "=2.0.0" }
 solana-version = { path = "version", version = "=2.0.0" }

Cargo.toml

@@ -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)

programs/sbf/Cargo.lock

@@ -26,9 +26,11 @@ quinn = { workspace = true }
 quinn-proto = { workspace = true }
 rand = { workspace = true }
 rustls = { workspace = true, features = ["dangerous_configuration"] }
+solana-measure = { workspace = true }
 solana-metrics = { workspace = true }
 solana-perf = { workspace = true }
 solana-sdk = { workspace = true }
+solana-transaction-metrics-tracker = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 x509-parser = { workspace = true }

sdk/src/packet.rs

@@ -17,6 +17,7 @@ use {
     quinn::{Connecting, Connection, Endpoint, EndpointConfig, TokioRuntime, VarInt},
     quinn_proto::VarIntBoundsExceeded,
     rand::{thread_rng, Rng},
+    solana_measure::measure::Measure,
     solana_perf::packet::{PacketBatch, PACKETS_PER_BATCH},
     solana_sdk::{
         packet::{Meta, PACKET_DATA_SIZE},
@@ -27,9 +28,10 @@ use {
             QUIC_MIN_STAKED_CONCURRENT_STREAMS, QUIC_MIN_STAKED_RECEIVE_WINDOW_RATIO,
             QUIC_TOTAL_STAKED_CONCURRENT_STREAMS, QUIC_UNSTAKED_RECEIVE_WINDOW_RATIO,
         },
-        signature::Keypair,
+        signature::{Keypair, Signature},
         timing,
     },
+    solana_transaction_metrics_tracker::signature_if_should_track_packet,
     std::{
         iter::repeat_with,
         net::{IpAddr, SocketAddr, UdpSocket},
@@ -94,6 +96,7 @@ struct PacketChunk {
 struct PacketAccumulator {
     pub meta: Meta,
     pub chunks: Vec<PacketChunk>,
+    pub start_time: Instant,
 }
 
 #[derive(Copy, Clone, Debug)]
@@ -646,6 +649,7 @@ async fn packet_batch_sender(
     trace!("enter packet_batch_sender");
     let mut batch_start_time = Instant::now();
     loop {
+        let mut packet_perf_measure: Vec<([u8; 64], std::time::Instant)> = Vec::default();
         let mut packet_batch = PacketBatch::with_capacity(PACKETS_PER_BATCH);
         let mut total_bytes: usize = 0;
 
@@ -665,6 +669,8 @@ async fn packet_batch_sender(
                 || (!packet_batch.is_empty() && elapsed >= coalesce)
             {
                 let len = packet_batch.len();
+                track_streamer_fetch_packet_performance(&mut packet_perf_measure, &stats);
+
                 if let Err(e) = packet_sender.send(packet_batch) {
                     stats
                         .total_packet_batch_send_err
@@ -710,6 +716,14 @@ async fn packet_batch_sender(
 
                 total_bytes += packet_batch[i].meta().size;
 
+                if let Some(signature) = signature_if_should_track_packet(&packet_batch[i])
+                    .ok()
+                    .flatten()
+                {
+                    packet_perf_measure.push((*signature, packet_accumulator.start_time));
+                    // we set the PERF_TRACK_PACKET on
+                    packet_batch[i].meta_mut().set_track_performance(true);
+                }
                 stats
                     .total_chunks_processed_by_batcher
                     .fetch_add(num_chunks, Ordering::Relaxed);
@@ -718,6 +732,32 @@ async fn packet_batch_sender(
     }
 }
 
+fn track_streamer_fetch_packet_performance(
+    packet_perf_measure: &mut [([u8; 64], Instant)],
+    stats: &Arc<StreamStats>,
+) {
+    if packet_perf_measure.is_empty() {
+        return;
+    }
+    let mut measure = Measure::start("track_perf");
+    let mut process_sampled_packets_us_hist = stats.process_sampled_packets_us_hist.lock().unwrap();
+
+    for (signature, start_time) in packet_perf_measure.iter() {
+        let duration = Instant::now().duration_since(*start_time);
+        debug!(
+            "QUIC streamer fetch stage took {duration:?} for transaction {:?}",
+            Signature::from(*signature)
+        );
+        process_sampled_packets_us_hist
+            .increment(duration.as_micros() as u64)
+            .unwrap();
+    }
+    measure.stop();
+    stats
+        .perf_track_overhead_us
+        .fetch_add(measure.as_us(), Ordering::Relaxed);
+}
+
 async fn handle_connection(
     connection: Connection,
     remote_addr: SocketAddr,
@@ -872,6 +912,7 @@ async fn handle_chunk(
                     *packet_accum = Some(PacketAccumulator {
                         meta,
                         chunks: Vec::new(),
+                        start_time: Instant::now(),
                     });
                 }
 
@@ -1471,6 +1512,7 @@ pub mod test {
                     offset,
                     end_of_chunk: size,
                 }],
+                start_time: Instant::now(),
             };
             ptk_sender.send(packet_accum).await.unwrap();
         }

streamer/Cargo.toml

@@ -16,8 +16,8 @@ use {
     std::{
         net::UdpSocket,
         sync::{
-            atomic::{AtomicBool, AtomicUsize, Ordering},
-            Arc, RwLock,
+            atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
+            Arc, Mutex, RwLock,
         },
         thread,
         time::{Duration, SystemTime},
@@ -175,10 +175,19 @@ pub struct StreamStats {
     pub(crate) stream_load_ema: AtomicUsize,
     pub(crate) stream_load_ema_overflow: AtomicUsize,
     pub(crate) stream_load_capacity_overflow: AtomicUsize,
+    pub(crate) process_sampled_packets_us_hist: Mutex<histogram::Histogram>,
+    pub(crate) perf_track_overhead_us: AtomicU64,
 }
 
 impl StreamStats {
     pub fn report(&self, name: &'static str) {
+        let process_sampled_packets_us_hist = {
+            let mut metrics = self.process_sampled_packets_us_hist.lock().unwrap();
+            let process_sampled_packets_us_hist = metrics.clone();
+            metrics.clear();
+            process_sampled_packets_us_hist
+        };
+
         datapoint_info!(
             name,
             (
@@ -425,6 +434,38 @@ impl StreamStats {
                 self.stream_load_capacity_overflow.load(Ordering::Relaxed),
                 i64
             ),
+            (
+                "process_sampled_packets_us_90pct",
+                process_sampled_packets_us_hist
+                    .percentile(90.0)
+                    .unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_min",
+                process_sampled_packets_us_hist.minimum().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_max",
+                process_sampled_packets_us_hist.maximum().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_us_mean",
+                process_sampled_packets_us_hist.mean().unwrap_or(0),
+                i64
+            ),
+            (
+                "process_sampled_packets_count",
+                process_sampled_packets_us_hist.entries(),
+                i64
+            ),
+            (
+                "perf_track_overhead_us",
+                self.perf_track_overhead_us.swap(0, Ordering::Relaxed),
+                i64
+            ),
         );
     }
 }

streamer/src/nonblocking/quic.rs

@@ -0,0 +1,25 @@
+[package]
+name = "solana-transaction-metrics-tracker"
+description = "Solana transaction metrics tracker"
+documentation = "https://docs.rs/solana-transaction-metrics-tracker"
+version = { workspace = true }
+authors = { workspace = true }
+repository = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+edition = { workspace = true }
+publish = false
+
+[dependencies]
+Inflector = { workspace = true }
+base64 = { workspace = true }
+bincode = { workspace = true }
+# Update this borsh dependency to the workspace version once
+lazy_static = { workspace = true }
+log = { workspace = true }
+rand = { workspace = true }
+solana-perf = { workspace = true }
+solana-sdk = { workspace = true }
+
+[package.metadata.docs.rs]
+targets = ["x86_64-unknown-linux-gnu"]