Pr634 1.17 new filter (experiment no merging)

Cargo.toml

@@ -205,6 +205,7 @@ generic-array = { version = "0.14.7", default-features = false }
 gethostname = "0.2.3"
 getrandom = "0.2.10"
 goauth = "0.13.1"
+governor = "0.6.3"
 hex = "0.4.3"
 hidapi = { version = "2.4.1", default-features = false }
 histogram = "0.6.9"
@@ -265,8 +266,10 @@ prost-types = "0.11.9"
 protobuf-src = "1.1.0"
 qstring = "0.7.2"
 qualifier_attr = { version = "0.2.2", default-features = false }
-quinn = "0.10.2"
-quinn-proto = "0.10.5"
+#quinn = "0.10.2"
+#quinn-proto = "0.10.5"
+quinn = {git = "https://github.com/lijunwangs/quinn.git", rev = "5980492919744a8366ce74a837b64e052aff9944"}
+quinn-proto = {git = "https://github.com/lijunwangs/quinn.git", rev = "5980492919744a8366ce74a837b64e052aff9944"}
 quote = "1.0"
 rand = "0.8.5"
 rand_chacha = "0.3.1"
@@ -296,7 +299,7 @@ sha3 = "0.10.4"
 signal-hook = "0.3.17"
 siphasher = "0.3.11"
 smpl_jwt = "0.7.1"
-socket2 = "0.5.4"
+socket2 = { version= "0.5.4", features=["all"]}
 soketto = "0.7"
 solana_rbpf = "=0.8.0"
 solana-account-decoder = { path = "account-decoder", version = "=1.17.30" }
@@ -365,6 +368,7 @@ 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-status = { path = "transaction-status", version = "=1.17.30" }
+solana-transaction-metrics-tracker = { path = "transaction-metrics-tracker", version = "=1.17.30" }
 solana-turbine = { path = "turbine", version = "=1.17.30" }
 solana-udp-client = { path = "udp-client", version = "=1.17.30" }
 solana-version = { path = "version", version = "=1.17.30" }

cli-output/src/display.rs

@@ -1,7 +1,7 @@
 use {
     crate::cli_output::CliSignatureVerificationStatus,
     base64::{prelude::BASE64_STANDARD, Engine},
-    chrono::{Local, NaiveDateTime, SecondsFormat, TimeZone, Utc},
+    chrono::{DateTime, Local, SecondsFormat, TimeZone, Utc},
     console::style,
     indicatif::{ProgressBar, ProgressStyle},
     solana_cli_config::SettingType,
@@ -715,10 +715,8 @@ pub fn new_spinner_progress_bar() -> ProgressBar {
 }
 
 pub fn unix_timestamp_to_string(unix_timestamp: UnixTimestamp) -> String {
-    match NaiveDateTime::from_timestamp_opt(unix_timestamp, 0) {
-        Some(ndt) => Utc
-            .from_utc_datetime(&ndt)
-            .to_rfc3339_opts(SecondsFormat::Secs, true),
+    match DateTime::from_timestamp(unix_timestamp, 0) {
+        Some(ndt) => ndt.to_rfc3339_opts(SecondsFormat::Secs, true),
         None => format!("UnixTimestamp {unix_timestamp}"),
     }
 }

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/repair/quic_endpoint.rs

@@ -183,7 +183,6 @@ fn new_server_config(cert: Certificate, key: PrivateKey) -> Result<ServerConfig,
     let mut config = ServerConfig::with_crypto(Arc::new(config));
     config
         .transport_config(Arc::new(new_transport_config()))
-        .use_retry(true)
         .migration(false);
     Ok(config)
 }
@@ -224,17 +223,29 @@ async fn run_server(
     let stats = Arc::<RepairQuicStats>::default();
     let report_metrics_task =
         tokio::task::spawn(report_metrics_task("repair_quic_server", stats.clone()));
-    while let Some(connecting) = endpoint.accept().await {
-        tokio::task::spawn(handle_connecting_task(
-            endpoint.clone(),
-            connecting,
-            remote_request_sender.clone(),
-            bank_forks.clone(),
-            prune_cache_pending.clone(),
-            router.clone(),
-            cache.clone(),
-            stats.clone(),
-        ));
+    while let Some(incoming) = endpoint.accept().await {
+        let remote_addr: SocketAddr = incoming.remote_address();
+        let connecting = incoming.accept();
+        match connecting {
+            Ok(connecting) => {
+                tokio::task::spawn(handle_connecting_task(
+                    endpoint.clone(),
+                    connecting,
+                    remote_request_sender.clone(),
+                    bank_forks.clone(),
+                    prune_cache_pending.clone(),
+                    router.clone(),
+                    cache.clone(),
+                    stats.clone(),
+                ));
+            }
+            Err(error) => {
+                debug!(
+                    "Error while accepting incoming connection: {error:?} from {}",
+                    remote_addr
+                );
+            }
+        }
     }
     report_metrics_task.abort();
 }
@@ -775,14 +786,15 @@ impl<T> From<crossbeam_channel::SendError<T>> for Error {
 struct RepairQuicStats {
     connect_error_invalid_remote_address: AtomicU64,
     connect_error_other: AtomicU64,
-    connect_error_too_many_connections: AtomicU64,
+    connect_error_cids_exhausted: AtomicU64,
     connection_error_application_closed: AtomicU64,
     connection_error_connection_closed: AtomicU64,
     connection_error_locally_closed: AtomicU64,
     connection_error_reset: AtomicU64,
     connection_error_timed_out: AtomicU64,
     connection_error_transport_error: AtomicU64,
     connection_error_version_mismatch: AtomicU64,
+    connection_error_cids_exhausted: AtomicU64,
     invalid_identity: AtomicU64,
     no_response_received: AtomicU64,
     read_to_end_error_connection_lost: AtomicU64,
@@ -814,9 +826,6 @@ fn record_error(err: &Error, stats: &RepairQuicStats) {
         Error::ConnectError(ConnectError::EndpointStopping) => {
             add_metric!(stats.connect_error_other)
         }
-        Error::ConnectError(ConnectError::TooManyConnections) => {
-            add_metric!(stats.connect_error_too_many_connections)
-        }
         Error::ConnectError(ConnectError::InvalidDnsName(_)) => {
             add_metric!(stats.connect_error_other)
         }
@@ -829,6 +838,12 @@ fn record_error(err: &Error, stats: &RepairQuicStats) {
         Error::ConnectError(ConnectError::UnsupportedVersion) => {
             add_metric!(stats.connect_error_other)
         }
+        Error::ConnectError(ConnectError::CidsExhausted) => {
+            add_metric!(stats.connect_error_cids_exhausted)
+        }
+        Error::ConnectionError(ConnectionError::CidsExhausted) => {
+            add_metric!(stats.connection_error_cids_exhausted)
+        }
         Error::ConnectionError(ConnectionError::VersionMismatch) => {
             add_metric!(stats.connection_error_version_mismatch)
         }
@@ -903,8 +918,8 @@ fn report_metrics(name: &'static str, stats: &RepairQuicStats) {
             i64
         ),
         (
-            "connect_error_too_many_connections",
-            reset_metric!(stats.connect_error_too_many_connections),
+            "connect_error_cids_exhausted",
+            reset_metric!(stats.connect_error_cids_exhausted),
             i64
         ),
         (
@@ -942,6 +957,11 @@ fn report_metrics(name: &'static str, stats: &RepairQuicStats) {
             reset_metric!(stats.connection_error_version_mismatch),
             i64
         ),
+        (
+            "connection_error_cids_exhausted",
+            reset_metric!(stats.connection_error_cids_exhausted),
+            i64
+        ),
         (
             "invalid_identity",
             reset_metric!(stats.invalid_identity),