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(),

core/src/validator.rs

@@ -81,8 +81,10 @@ use {
         poh_recorder::PohRecorder,
         poh_service::{self, PohService},
     },
+    solana_quic_client::quic_client::RUNTIME,
     solana_rayon_threadlimit::{get_max_thread_count, get_thread_count},
     solana_rpc::{
+        cluster_tpu_info::ClusterTpuInfo,
         max_slots::MaxSlots,
         optimistically_confirmed_bank_tracker::{
             BankNotificationSenderConfig, OptimisticallyConfirmedBank,
@@ -124,7 +126,10 @@ use {
         signature::{Keypair, Signer},
         timing::timestamp,
     },
-    solana_send_transaction_service::send_transaction_service,
+    solana_send_transaction_service::{
+        send_transaction_service,
+        transaction_client::{spawn_tpu_client_send_txs, ConnectionCacheClient},
+    },
     solana_streamer::{socket::SocketAddrSpace, streamer::StakedNodes},
     solana_turbine::{self, broadcast_stage::BroadcastStageType},
     solana_unified_scheduler_pool::DefaultSchedulerPool,
@@ -287,6 +292,7 @@ pub struct ValidatorConfig {
     pub replay_transactions_threads: NonZeroUsize,
     pub tvu_shred_sigverify_threads: NonZeroUsize,
     pub delay_leader_block_for_pending_fork: bool,
+    pub use_tpu_client_next: bool,
 }
 
 impl Default for ValidatorConfig {
@@ -360,6 +366,7 @@ impl Default for ValidatorConfig {
             replay_transactions_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             tvu_shred_sigverify_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             delay_leader_block_for_pending_fork: false,
+            use_tpu_client_next: false,
         }
     }
 }
@@ -991,6 +998,8 @@ impl Validator {
 
         let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
 
+        // ConnectionCache might be used for JsonRpc and for Forwarding. Since the later is not migrated yet to the tpu-client-next,
+        // create ConnectionCache regardless of config.use_tpu_client_next for now
         let connection_cache = match use_quic {
             true => {
                 let connection_cache = ConnectionCache::new_with_client_options(
@@ -1044,31 +1053,82 @@ impl Validator {
                 None
             };
 
-            let json_rpc_service = JsonRpcService::new(
-                rpc_addr,
-                config.rpc_config.clone(),
-                Some(config.snapshot_config.clone()),
-                bank_forks.clone(),
-                block_commitment_cache.clone(),
-                blockstore.clone(),
-                cluster_info.clone(),
-                Some(poh_recorder.clone()),
-                genesis_config.hash(),
-                ledger_path,
-                config.validator_exit.clone(),
-                exit.clone(),
-                rpc_override_health_check.clone(),
-                startup_verification_complete,
-                optimistically_confirmed_bank.clone(),
-                config.send_transaction_service_config.clone(),
-                max_slots.clone(),
-                leader_schedule_cache.clone(),
-                connection_cache.clone(),
-                max_complete_transaction_status_slot,
-                max_complete_rewards_slot,
-                prioritization_fee_cache.clone(),
-            )
-            .map_err(ValidatorError::Other)?;
+            let leader_info = ClusterTpuInfo::new(cluster_info.clone(), poh_recorder.clone());
+            // TODO(klykov): consider using Box<dyn TransactionClient> to make this shorter?
+            let json_rpc_service = if config.use_tpu_client_next {
+                let my_tpu_address = cluster_info
+                    .my_contact_info()
+                    .tpu(Protocol::QUIC)
+                    .map_err(|err| ValidatorError::Other(format!("{err}")))?;
+                let client = spawn_tpu_client_send_txs(
+                    &*RUNTIME, // use the same runtime as ConnectionCache
+                    my_tpu_address,
+                    config.send_transaction_service_config.tpu_peers.clone(),
+                    Some(leader_info),
+                    config.send_transaction_service_config.leader_forward_count,
+                    Some((*identity_keypair).insecure_clone()),
+                );
+                JsonRpcService::new(
+                    rpc_addr,
+                    config.rpc_config.clone(),
+                    Some(config.snapshot_config.clone()),
+                    bank_forks.clone(),
+                    block_commitment_cache.clone(),
+                    blockstore.clone(),
+                    cluster_info.clone(),
+                    genesis_config.hash(),
+                    ledger_path,
+                    config.validator_exit.clone(),
+                    exit.clone(),
+                    rpc_override_health_check.clone(),
+                    startup_verification_complete,
+                    optimistically_confirmed_bank.clone(),
+                    config.send_transaction_service_config.clone(),
+                    max_slots.clone(),
+                    leader_schedule_cache.clone(),
+                    client,
+                    max_complete_transaction_status_slot,
+                    max_complete_rewards_slot,
+                    prioritization_fee_cache.clone(),
+                )
+                .map_err(ValidatorError::Other)?
+            } else {
+                let my_tpu_address = cluster_info
+                    .my_contact_info()
+                    .tpu(connection_cache.protocol())
+                    .map_err(|err| ValidatorError::Other(format!("{err}")))?;
+                let client = ConnectionCacheClient::new(
+                    connection_cache.clone(),
+                    my_tpu_address,
+                    config.send_transaction_service_config.tpu_peers.clone(),
+                    Some(leader_info),
+                    config.send_transaction_service_config.leader_forward_count,
+                );
+                JsonRpcService::new(
+                    rpc_addr,
+                    config.rpc_config.clone(),
+                    Some(config.snapshot_config.clone()),
+                    bank_forks.clone(),
+                    block_commitment_cache.clone(),
+                    blockstore.clone(),
+                    cluster_info.clone(),
+                    genesis_config.hash(),
+                    ledger_path,
+                    config.validator_exit.clone(),
+                    exit.clone(),
+                    rpc_override_health_check.clone(),
+                    startup_verification_complete,
+                    optimistically_confirmed_bank.clone(),
+                    config.send_transaction_service_config.clone(),
+                    max_slots.clone(),
+                    leader_schedule_cache.clone(),
+                    client,
+                    max_complete_transaction_status_slot,
+                    max_complete_rewards_slot,
+                    prioritization_fee_cache.clone(),
+                )
+                .map_err(ValidatorError::Other)?
+            };
 
             let pubsub_service = if !config.rpc_config.full_api {
                 None
@@ -1416,7 +1476,7 @@ impl Validator {
             config.wait_to_vote_slot,
             accounts_background_request_sender.clone(),
             config.runtime_config.log_messages_bytes_limit,
-            json_rpc_service.is_some().then_some(&connection_cache), // for the cache warmer only used for STS for RPC service
+            (json_rpc_service.is_some() && config.use_tpu_client_next).then_some(&connection_cache), // for the cache warmer only used for STS for RPC service
             &prioritization_fee_cache,
             banking_tracer.clone(),
             turbine_quic_endpoint_sender.clone(),
@@ -1508,7 +1568,11 @@ impl Validator {
         );
 
         *start_progress.write().unwrap() = ValidatorStartProgress::Running;
-        key_notifies.push(connection_cache);
+        if config.use_tpu_client_next {
+            unimplemented!();
+        } else {
+            key_notifies.push(connection_cache);
+        }
 
         *admin_rpc_service_post_init.write().unwrap() = Some(AdminRpcRequestMetadataPostInit {
             bank_forks: bank_forks.clone(),

local-cluster/src/validator_configs.rs

@@ -74,6 +74,7 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
         replay_transactions_threads: config.replay_transactions_threads,
         tvu_shred_sigverify_threads: config.tvu_shred_sigverify_threads,
         delay_leader_block_for_pending_fork: config.delay_leader_block_for_pending_fork,
+        use_tpu_client_next: config.use_tpu_client_next,
     }
 }
 

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/lib.rs

@@ -1,5 +1,5 @@
 #![allow(clippy::arithmetic_side_effects)]
-mod cluster_tpu_info;
+pub mod cluster_tpu_info;
 pub mod filter;
 pub mod max_slots;
 pub mod optimistically_confirmed_bank_tracker;