Add tpu-client-next to SendTransactionService

Cargo.toml

@@ -521,6 +521,7 @@ solana-test-validator = { path = "test-validator", version = "=2.2.0" }
 solana-thin-client = { path = "thin-client", version = "=2.2.0" }
 solana-transaction-error = { path = "sdk/transaction-error", version = "=2.2.0" }
 solana-tpu-client = { path = "tpu-client", version = "=2.2.0", default-features = false }
+solana-tpu-client-next = { path = "tpu-client-next", version = "=2.2.0" }
 solana-transaction-status = { path = "transaction-status", version = "=2.2.0" }
 solana-transaction-status-client-types = { path = "transaction-status-client-types", version = "=2.2.0" }
 solana-transaction-metrics-tracker = { path = "transaction-metrics-tracker", version = "=2.2.0" }

banks-server/src/banks_server.rs

@@ -28,6 +28,7 @@ use {
     solana_send_transaction_service::{
         send_transaction_service::{SendTransactionService, TransactionInfo},
         tpu_info::NullTpuInfo,
+        transaction_client::ConnectionCacheClient,
     },
     std::{
         io,
@@ -453,17 +454,16 @@ pub async fn start_tcp_server(
         .map(move |chan| {
             let (sender, receiver) = unbounded();
 
-            SendTransactionService::new::<NullTpuInfo>(
+            let client = ConnectionCacheClient::<NullTpuInfo>::new(
+                connection_cache.clone(),
                 tpu_addr,
-                &bank_forks,
                 None,
-                receiver,
-                &connection_cache,
-                5_000,
+                None,
                 0,
-                exit.clone(),
             );
 
+            SendTransactionService::new(&bank_forks, receiver, client, 5_000, exit.clone());
+
             let server = BanksServer::new(
                 bank_forks.clone(),
                 block_commitment_cache.clone(),

quic-client/src/quic_client.rs

@@ -68,7 +68,7 @@ impl AsyncTaskSemaphore {
 lazy_static! {
     static ref ASYNC_TASK_SEMAPHORE: AsyncTaskSemaphore =
         AsyncTaskSemaphore::new(MAX_OUTSTANDING_TASK);
-    static ref RUNTIME: Runtime = tokio::runtime::Builder::new_multi_thread()
+    pub static ref RUNTIME: Runtime = tokio::runtime::Builder::new_multi_thread()
         .thread_name("solQuicClientRt")
         .enable_all()
         .build()

rpc/src/rpc.rs

@@ -83,6 +83,7 @@ use {
     solana_send_transaction_service::{
         send_transaction_service::{SendTransactionService, TransactionInfo},
         tpu_info::NullTpuInfo,
+        transaction_client::ConnectionCacheClient,
     },
     solana_stake_program,
     solana_storage_bigtable::Error as StorageError,
@@ -372,16 +373,14 @@ impl JsonRpcRequestProcessor {
             .tpu(connection_cache.protocol())
             .unwrap();
         let (sender, receiver) = unbounded();
-        SendTransactionService::new::<NullTpuInfo>(
+        let client = ConnectionCacheClient::<NullTpuInfo>::new(
+            connection_cache.clone(),
             tpu_address,
-            &bank_forks,
             None,
-            receiver,
-            &connection_cache,
-            1000,
+            None,
             1,
-            exit.clone(),
         );
+        SendTransactionService::new(&bank_forks, receiver, client, 1000, exit.clone());
 
         let leader_schedule_cache = Arc::new(LeaderScheduleCache::new_from_bank(&bank));
         let startup_verification_complete = Arc::clone(bank.get_startup_verification_complete());
@@ -6481,16 +6480,14 @@ pub mod tests {
             Arc::new(AtomicU64::default()),
             Arc::new(PrioritizationFeeCache::default()),
         );
-        SendTransactionService::new::<NullTpuInfo>(
+        let client = ConnectionCacheClient::<NullTpuInfo>::new(
+            connection_cache.clone(),
             tpu_address,
-            &bank_forks,
             None,
-            receiver,
-            &connection_cache,
-            1000,
+            None,
             1,
-            exit,
         );
+        SendTransactionService::new(&bank_forks, receiver, client, 1000, exit);
 
         let mut bad_transaction = system_transaction::transfer(
             &mint_keypair,
@@ -6755,16 +6752,14 @@ pub mod tests {
             Arc::new(AtomicU64::default()),
             Arc::new(PrioritizationFeeCache::default()),
         );
-        SendTransactionService::new::<NullTpuInfo>(
+        let client = ConnectionCacheClient::<NullTpuInfo>::new(
+            connection_cache.clone(),
             tpu_address,
-            &bank_forks,
             None,
-            receiver,
-            &connection_cache,
-            1000,
+            None,
             1,
-            exit,
         );
+        SendTransactionService::new(&bank_forks, receiver, client, 1000, exit);
         assert_eq!(
             request_processor.get_block_commitment(0),
             RpcBlockCommitment {

rpc/src/rpc_service.rs

@@ -36,7 +36,10 @@ use {
         exit::Exit, genesis_config::DEFAULT_GENESIS_DOWNLOAD_PATH, hash::Hash,
         native_token::lamports_to_sol,
     },
-    solana_send_transaction_service::send_transaction_service::{self, SendTransactionService},
+    solana_send_transaction_service::{
+        send_transaction_service::{self, SendTransactionService},
+        transaction_client::ConnectionCacheClient,
+    },
     solana_storage_bigtable::CredentialType,
     std::{
         net::SocketAddr,
@@ -474,15 +477,20 @@ impl JsonRpcService {
 
         let leader_info =
             poh_recorder.map(|recorder| ClusterTpuInfo::new(cluster_info.clone(), recorder));
-        let _send_transaction_service = Arc::new(SendTransactionService::new_with_config(
+        let client = ConnectionCacheClient::new(
+            connection_cache,
             tpu_address,
-            &bank_forks,
+            send_transaction_service_config.tpu_peers.clone(), //TODO(klykov): check if we can avoid cloning
             leader_info,
+            send_transaction_service_config.leader_forward_count,
+        );
+        let _send_transaction_service = SendTransactionService::new_with_config(
+            &bank_forks,
             receiver,
-            &connection_cache,
+            client,
             send_transaction_service_config,
             exit,
-        ));
+        );
 
         #[cfg(test)]
         let test_request_processor = request_processor.clone();

send-transaction-service/Cargo.toml

@@ -10,19 +10,26 @@ license = { workspace = true }
 edition = { workspace = true }
 
 [dependencies]
+# TODO(klykov): remove this dependency when trait for leader updater
+# will be changed
+async-trait = { workspace = true }
 crossbeam-channel = { workspace = true }
+itertools = { workspace = true }
 log = { workspace = true }
 solana-client = { workspace = true }
 solana-connection-cache = { workspace = true }
 solana-measure = { workspace = true }
 solana-metrics = { workspace = true }
 solana-runtime = { workspace = true }
 solana-sdk = { workspace = true }
-solana-tpu-client = { workspace = true }
+solana-tpu-client-next = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+tokio-util = { workspace = true }
 
 [dev-dependencies]
 solana-logger = { workspace = true }
 solana-runtime = { workspace = true, features = ["dev-context-only-utils"] }
+solana-quic-client = { workspace = true }
 
 [package.metadata.docs.rs]
 targets = ["x86_64-unknown-linux-gnu"]

send-transaction-service/src/lib.rs

@@ -1,6 +1,8 @@
 #![allow(clippy::arithmetic_side_effects)]
 pub mod send_transaction_service;
+pub mod send_transaction_service_stats;
 pub mod tpu_info;
+pub mod transaction_client;
 
 #[macro_use]
 extern crate solana_metrics;