metrics reporting for tpu-client-next

Author: Alex Pyattaev

Cargo.lock

@@ -4985,7 +4985,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_request_processor_new_tpu_client_next() {
-        rpc_request_processor_new::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_request_processor_new::<TpuClientNextClient>();
     }
 
     fn rpc_get_balance<Client: ClientWithCreator>() {
@@ -5022,7 +5022,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_get_balance_new_tpu_client_next() {
-        rpc_get_balance::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_get_balance::<TpuClientNextClient>();
     }
 
     fn rpc_get_balance_via_client<Client: ClientWithCreator>() {
@@ -5061,7 +5061,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_get_balance_via_client_tpu_client_next() {
-        rpc_get_balance_via_client::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_get_balance_via_client::<TpuClientNextClient>();
     }
 
     #[test]
@@ -5196,7 +5196,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_get_tx_count_tpu_client_next() {
-        rpc_get_tx_count::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_get_tx_count::<TpuClientNextClient>();
     }
 
     #[test]
@@ -6657,7 +6657,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_send_bad_tx_tpu_client_next() {
-        rpc_send_bad_tx::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_send_bad_tx::<TpuClientNextClient>();
     }
 
     fn rpc_send_transaction_preflight<Client: ClientWithCreator>() {
@@ -6841,7 +6841,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_send_transaction_preflight_with_tpu_client_next() {
-        rpc_send_transaction_preflight::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_send_transaction_preflight::<TpuClientNextClient>();
     }
 
     #[test]
@@ -7054,7 +7054,7 @@ pub mod tests {
 
     #[test]
     fn test_rpc_processor_get_block_commitment_with_tpu_client_next() {
-        rpc_processor_get_block_commitment::<TpuClientNextClient<NullTpuInfo>>();
+        rpc_processor_get_block_commitment::<TpuClientNextClient>();
     }
 
     #[test]

programs/sbf/Cargo.lock

@@ -21,7 +21,7 @@ solana-measure = { workspace = true }
 solana-metrics = { workspace = true }
 solana-runtime = { workspace = true }
 solana-sdk = { workspace = true }
-solana-tpu-client-next = { workspace = true }
+solana-tpu-client-next = { workspace = true, features = ["metrics"] }
 tokio = { workspace = true, features = ["full"] }
 tokio-util = { workspace = true }
 

rpc/src/rpc.rs

@@ -614,7 +614,7 @@ mod test {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn service_exit_with_tpu_client_next() {
-        service_exit::<TpuClientNextClient<NullTpuInfo>>(Some(Handle::current()));
+        service_exit::<TpuClientNextClient>(Some(Handle::current()));
     }
 
     fn validator_exit<C: ClientWithCreator>(maybe_runtime: Option<Handle>) {
@@ -665,7 +665,7 @@ mod test {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn validator_exit_with_tpu_client_next() {
-        validator_exit::<TpuClientNextClient<NullTpuInfo>>(Some(Handle::current()));
+        validator_exit::<TpuClientNextClient>(Some(Handle::current()));
     }
 
     fn process_transactions<C: ClientWithCreator>(maybe_runtime: Option<Handle>) {
@@ -924,7 +924,7 @@ mod test {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn process_transactions_with_tpu_client_next() {
-        process_transactions::<TpuClientNextClient<NullTpuInfo>>(Some(Handle::current()));
+        process_transactions::<TpuClientNextClient>(Some(Handle::current()));
     }
 
     fn retry_durable_nonce_transactions<C: ClientWithCreator>(maybe_runtime: Option<Handle>) {
@@ -1234,8 +1234,6 @@ mod test {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn retry_durable_nonce_transactions_with_tpu_client_next() {
-        retry_durable_nonce_transactions::<TpuClientNextClient<NullTpuInfo>>(Some(
-            Handle::current(),
-        ));
+        retry_durable_nonce_transactions::<TpuClientNextClient>(Some(Handle::current()));
     }
 }

send-transaction-service/Cargo.toml

@@ -43,7 +43,7 @@ impl CreateClient for ConnectionCacheClient<NullTpuInfo> {
     }
 }
 
-impl CreateClient for TpuClientNextClient<NullTpuInfo> {
+impl CreateClient for TpuClientNextClient {
     fn create_client(
         maybe_runtime: Option<Handle>,
         my_tpu_address: SocketAddr,
@@ -52,7 +52,7 @@ impl CreateClient for TpuClientNextClient<NullTpuInfo> {
     ) -> Self {
         let runtime_handle =
             maybe_runtime.expect("Runtime should be provided for the TpuClientNextClient.");
-        Self::new(
+        Self::new::<NullTpuInfo>(
             runtime_handle,
             my_tpu_address,
             tpu_peers,
@@ -74,10 +74,7 @@ where
     fn stop(&self) {}
 }
 
-impl<T> Stoppable for TpuClientNextClient<T>
-where
-    T: TpuInfoWithSendStatic + Clone,
-{
+impl Stoppable for TpuClientNextClient {
     fn stop(&self) {
         self.cancel().unwrap();
     }

send-transaction-service/src/send_transaction_service.rs

@@ -8,9 +8,7 @@ use {
     solana_measure::measure::Measure,
     solana_sdk::quic::NotifyKeyUpdate,
     solana_tpu_client_next::{
-        connection_workers_scheduler::{
-            ConnectionWorkersSchedulerConfig, Fanout, TransactionStatsAndReceiver,
-        },
+        connection_workers_scheduler::{ConnectionWorkersSchedulerConfig, Fanout},
         leader_updater::LeaderUpdater,
         transaction_batch::TransactionBatch,
         ConnectionWorkersScheduler, ConnectionWorkersSchedulerError,
@@ -230,27 +228,20 @@ where
 ///   scheduler. Most of the complexity of this structure arises from this
 ///   functionality.
 #[derive(Clone)]
-pub struct TpuClientNextClient<T>
-where
-    T: TpuInfoWithSendStatic + Clone,
-{
+pub struct TpuClientNextClient {
     runtime_handle: Handle,
     sender: mpsc::Sender<TransactionBatch>,
     // This handle is needed to implement `NotifyKeyUpdate` trait. It's only
     // method takes &self and thus we need to wrap with Mutex.
     join_and_cancel: Arc<Mutex<(Option<TpuClientJoinHandle>, CancellationToken)>>,
-    leader_updater: SendTransactionServiceLeaderUpdater<T>,
     leader_forward_count: u64,
 }
 
 type TpuClientJoinHandle =
-    TokioJoinHandle<Result<TransactionStatsAndReceiver, ConnectionWorkersSchedulerError>>;
+    TokioJoinHandle<Result<ConnectionWorkersScheduler, ConnectionWorkersSchedulerError>>;
 
-impl<T> TpuClientNextClient<T>
-where
-    T: TpuInfoWithSendStatic + Clone,
-{
-    pub fn new(
+impl TpuClientNextClient {
+    pub fn new<T>(
         runtime_handle: Handle,
         my_tpu_address: SocketAddr,
         tpu_peers: Option<Vec<SocketAddr>>,
@@ -275,18 +266,18 @@ where
                 tpu_peers,
             };
         let config = Self::create_config(identity, leader_forward_count as usize);
-        let handle = runtime_handle.spawn(ConnectionWorkersScheduler::run(
-            config,
-            Box::new(leader_updater.clone()),
-            receiver,
+        let scheduler = ConnectionWorkersScheduler::new(Box::new(leader_updater), receiver);
+        // leaking handle to this task, as it will run until the end of life of the process
+        runtime_handle.spawn(scheduler.stats.clone().report_to_influxdb(
+            "send-transaction-service-TPU-client",
+            Duration::from_secs(3),
             cancel.clone(),
         ));
-
+        let handle = runtime_handle.spawn(scheduler.run(config, cancel.clone()));
         Self {
             runtime_handle,
             join_and_cancel: Arc::new(Mutex::new((Some(handle), cancel))),
             sender,
-            leader_updater,
             leader_forward_count,
         }
     }
@@ -322,7 +313,6 @@ where
     async fn do_update_key(&self, identity: &Keypair) -> Result<(), Box<dyn std::error::Error>> {
         let runtime_handle = self.runtime_handle.clone();
         let config = Self::create_config(Some(identity), self.leader_forward_count as usize);
-        let leader_updater = self.leader_updater.clone();
         let handle = self.join_and_cancel.clone();
 
         let join_handle = {
@@ -340,14 +330,9 @@ where
             };
 
             match result {
-                Ok((_stats, receiver)) => {
+                Ok(scheduler) => {
                     let cancel = CancellationToken::new();
-                    let join_handle = runtime_handle.spawn(ConnectionWorkersScheduler::run(
-                        config,
-                        Box::new(leader_updater),
-                        receiver,
-                        cancel.clone(),
-                    ));
+                    let join_handle = runtime_handle.spawn(scheduler.run(config, cancel.clone()));
 
                     let Ok(mut lock) = handle.lock() else {
                         return Err("TpuClientNext task panicked.".into());
@@ -363,19 +348,13 @@ where
     }
 }
 
-impl<T> NotifyKeyUpdate for TpuClientNextClient<T>
-where
-    T: TpuInfoWithSendStatic + Clone,
-{
+impl NotifyKeyUpdate for TpuClientNextClient {
     fn update_key(&self, identity: &Keypair) -> Result<(), Box<dyn std::error::Error>> {
         self.runtime_handle.block_on(self.do_update_key(identity))
     }
 }
 
-impl<T> TransactionClient for TpuClientNextClient<T>
-where
-    T: TpuInfoWithSendStatic + Clone,
-{
+impl TransactionClient for TpuClientNextClient {
     fn send_transactions_in_batch(
         &self,
         wire_transactions: Vec<Vec<u8>>,
@@ -415,8 +394,11 @@ where
         };
         match self.runtime_handle.block_on(handle) {
             Ok(result) => match result {
-                Ok(stats) => {
-                    debug!("tpu-client-next statistics over all the connections: {stats:?}");
+                Ok(scheduler) => {
+                    debug!(
+                        "tpu-client-next statistics over all the connections: {:?}",
+                        scheduler.stats
+                    );
                 }
                 Err(error) => error!("tpu-client-next exits with error {error}."),
             },

send-transaction-service/src/test_utils.rs

@@ -18,6 +18,7 @@ solana-clock = { workspace = true }
 solana-connection-cache = { workspace = true }
 solana-keypair = { workspace = true }
 solana-measure = { workspace = true }
+solana-metrics ={ workspace = true, optional = true }
 solana-quic-definitions = { workspace = true }
 solana-rpc-client = { workspace = true }
 solana-streamer = { workspace = true }
@@ -39,3 +40,6 @@ solana-streamer = { workspace = true, features = ["dev-context-only-utils"] }
 
 [package.metadata.docs.rs]
 targets = ["x86_64-unknown-linux-gnu"]
+
+[features]
+metrics =["dep:solana-metrics"]