switch to solana-tpu-client from solana_client::tpu_client for bench-tps, dos/, LocalCluster, gossip/

bench-tps/Cargo.toml

@@ -28,6 +28,7 @@ solana-logger = { workspace = true }
 solana-measure = { workspace = true }
 solana-metrics = { workspace = true }
 solana-net-utils = { workspace = true }
+solana-quic-client = { workspace = true }
 solana-rpc = { workspace = true }
 solana-rpc-client = { workspace = true }
 solana-rpc-client-api = { workspace = true }

bench-tps/src/bench_tps_client/tpu_client.rs

@@ -1,6 +1,5 @@
 use {
     crate::bench_tps_client::{BenchTpsClient, BenchTpsError, Result},
-    solana_client::tpu_client::TpuClient,
     solana_connection_cache::connection_cache::{
         ConnectionManager, ConnectionPool, NewConnectionConfig,
     },
@@ -10,6 +9,7 @@ use {
         message::Message, pubkey::Pubkey, signature::Signature, slot_history::Slot,
         transaction::Transaction,
     },
+    solana_tpu_client::tpu_client::TpuClient,
     solana_transaction_status::UiConfirmedBlock,
 };
 

bench-tps/src/main.rs

@@ -8,10 +8,7 @@ use {
         keypairs::get_keypairs,
         send_batch::{generate_durable_nonce_accounts, generate_keypairs},
     },
-    solana_client::{
-        connection_cache::ConnectionCache,
-        tpu_client::{TpuClient, TpuClientConfig},
-    },
+    solana_client::connection_cache::ConnectionCache,
     solana_genesis::Base64Account,
     solana_rpc_client::rpc_client::RpcClient,
     solana_sdk::{
@@ -22,6 +19,7 @@ use {
         system_program,
     },
     solana_streamer::streamer::StakedNodes,
+    solana_tpu_client::tpu_client::{TpuClient, TpuClientConfig},
     std::{
         collections::HashMap,
         fs::File,

bench-tps/tests/bench_tps.rs

@@ -7,16 +7,15 @@ use {
         cli::{Config, InstructionPaddingConfig},
         send_batch::generate_durable_nonce_accounts,
     },
-    solana_client::{
-        connection_cache::ConnectionCache,
-        tpu_client::{TpuClient, TpuClientConfig},
-    },
+    solana_connection_cache::connection_cache::{ConnectionCache, NewConnectionConfig},
     solana_core::validator::ValidatorConfig,
     solana_faucet::faucet::run_local_faucet,
     solana_local_cluster::{
+        cluster::Cluster,
         local_cluster::{ClusterConfig, LocalCluster},
         validator_configs::make_identical_validator_configs,
     },
+    solana_quic_client::{QuicConfig, QuicConnectionManager},
     solana_rpc::rpc::JsonRpcConfig,
     solana_rpc_client::rpc_client::RpcClient,
     solana_sdk::{
@@ -28,6 +27,7 @@ use {
     },
     solana_streamer::socket::SocketAddrSpace,
     solana_test_validator::TestValidatorGenesis,
+    solana_tpu_client::tpu_client::{TpuClient, TpuClientConfig, DEFAULT_TPU_CONNECTION_POOL_SIZE},
     std::{sync::Arc, time::Duration},
 };
 
@@ -78,24 +78,9 @@ fn test_bench_tps_local_cluster(config: Config) {
 
     cluster.transfer(&cluster.funding_keypair, &faucet_pubkey, 100_000_000);
 
-    let ConnectionCache::Quic(cache) = &*cluster.connection_cache else {
-        panic!("Expected a Quic ConnectionCache.");
-    };
-
-    let rpc_pubsub_url = format!("ws://{}/", cluster.entry_point_info.rpc_pubsub().unwrap());
-    let rpc_url = format!("http://{}", cluster.entry_point_info.rpc().unwrap());
-
-    let client = Arc::new(
-        TpuClient::new_with_connection_cache(
-            Arc::new(RpcClient::new(rpc_url)),
-            rpc_pubsub_url.as_str(),
-            TpuClientConfig::default(),
-            cache.clone(),
-        )
-        .unwrap_or_else(|err| {
-            panic!("Could not create TpuClient {err:?}");
-        }),
-    );
+    let client = Arc::new(cluster.build_tpu_quic_client().unwrap_or_else(|err| {
+        panic!("Could not create TpuClient with Quic Cache {err:?}");
+    }));
 
     let lamports_per_account = 100;
 
@@ -141,8 +126,25 @@ fn test_bench_tps_test_validator(config: Config) {
         CommitmentConfig::processed(),
     ));
     let websocket_url = test_validator.rpc_pubsub_url();
-    let client =
-        Arc::new(TpuClient::new(rpc_client, &websocket_url, TpuClientConfig::default()).unwrap());
+
+    let connection_cache = Arc::new(
+        ConnectionCache::new(
+            "connection_cache_bench_tps_quic",
+            QuicConnectionManager::new_with_connection_config(QuicConfig::new().unwrap()),
+            DEFAULT_TPU_CONNECTION_POOL_SIZE,
+        )
+        .unwrap(),
+    );
+
+    let client = Arc::new(
+        TpuClient::new_with_connection_cache(
+            rpc_client,
+            &websocket_url,
+            TpuClientConfig::default(),
+            connection_cache,
+        )
+        .unwrap(),
+    );
 
     let lamports_per_account = 1000;
 

client/src/nonblocking/tpu_client.rs

@@ -1,6 +1,7 @@
 pub use solana_tpu_client::nonblocking::tpu_client::{LeaderTpuService, TpuSenderError};
 use {
     crate::{connection_cache::ConnectionCache, tpu_client::TpuClientConfig},
+    // crate::tpu_client::TpuClientConfig,
     solana_connection_cache::connection_cache::{
         ConnectionCache as BackendConnectionCache, ConnectionManager, ConnectionPool,
         NewConnectionConfig,

client/src/thin_client.rs

@@ -5,6 +5,7 @@
 #[allow(deprecated)]
 use {
     crate::connection_cache::{dispatch, ConnectionCache},
+    // crate::connection_cache::dispatch,
     solana_quic_client::{QuicConfig, QuicConnectionManager, QuicPool},
     solana_rpc_client::rpc_client::RpcClient,
     solana_rpc_client_api::config::RpcProgramAccountsConfig,

client/src/tpu_client.rs

@@ -13,19 +13,13 @@ use {
         transport::Result as TransportResult,
     },
     solana_tpu_client::tpu_client::{Result, TpuClient as BackendTpuClient},
-    solana_udp_client::{UdpConfig, UdpConnectionManager, UdpPool},
     std::sync::Arc,
 };
 pub use {
     crate::nonblocking::tpu_client::TpuSenderError,
     solana_tpu_client::tpu_client::{TpuClientConfig, DEFAULT_FANOUT_SLOTS, MAX_FANOUT_SLOTS},
 };
 
-pub enum TpuClientWrapper {
-    Quic(TpuClient<QuicPool, QuicConnectionManager, QuicConfig>),
-    Udp(TpuClient<UdpPool, UdpConnectionManager, UdpConfig>),
-}
-
 /// Client which sends transactions directly to the current leader's TPU port over UDP.
 /// The client uses RPC to determine the current leader and fetch node contact info
 /// This is just a thin wrapper over the "BackendTpuClient", use that directly for more efficiency.

dos/Cargo.toml

@@ -19,6 +19,7 @@ rand = { workspace = true }
 serde = { workspace = true }
 solana-bench-tps = { workspace = true }
 solana-client = { workspace = true }
+solana-connection-cache = { workspace = true }
 solana-core = { workspace = true }
 solana-faucet = { workspace = true }
 solana-gossip = { workspace = true }
@@ -32,6 +33,7 @@ solana-rpc-client = { workspace = true }
 solana-sdk = { workspace = true }
 solana-streamer = { workspace = true }
 solana-tpu-client = { workspace = true }
+solana-udp-client = { workspace = true }
 solana-version = { workspace = true }
 
 [package.metadata.docs.rs]

dos/src/main.rs

@@ -46,10 +46,7 @@ use {
     log::*,
     rand::{thread_rng, Rng},
     solana_bench_tps::{bench::generate_and_fund_keypairs, bench_tps_client::BenchTpsClient},
-    solana_client::{
-        connection_cache::ConnectionCache, tpu_client::TpuClientWrapper,
-        tpu_connection::TpuConnection,
-    },
+    solana_client::{connection_cache::ConnectionCache, tpu_connection::TpuConnection},
     solana_core::repair::serve_repair::{RepairProtocol, RepairRequestHeader, ServeRepair},
     solana_dos::cli::*,
     solana_gossip::{
@@ -72,7 +69,7 @@ use {
         transaction::Transaction,
     },
     solana_streamer::socket::SocketAddrSpace,
-    solana_tpu_client::tpu_client::DEFAULT_TPU_CONNECTION_POOL_SIZE,
+    solana_tpu_client::tpu_client::{TpuClientWrapper, DEFAULT_TPU_CONNECTION_POOL_SIZE},
     std::{
         net::{SocketAddr, UdpSocket},
         process::exit,
@@ -795,6 +792,7 @@ fn main() {
                 DEFAULT_TPU_CONNECTION_POOL_SIZE,
             ),
         };
+
         let client = get_client(&validators, Arc::new(connection_cache));
         (gossip_nodes, Some(client))
     } else {
@@ -818,7 +816,6 @@ fn main() {
 pub mod test {
     use {
         super::*,
-        solana_client::tpu_client::TpuClient,
         solana_core::validator::ValidatorConfig,
         solana_faucet::faucet::run_local_faucet,
         solana_gossip::contact_info::LegacyContactInfo,
@@ -827,43 +824,17 @@ pub mod test {
             local_cluster::{ClusterConfig, LocalCluster},
             validator_configs::make_identical_validator_configs,
         },
-        solana_quic_client::{QuicConfig, QuicConnectionManager, QuicPool},
         solana_rpc::rpc::JsonRpcConfig,
         solana_sdk::timing::timestamp,
-        solana_tpu_client::tpu_client::TpuClientConfig,
+        solana_tpu_client::tpu_client::QuicTpuClient,
     };
 
     const TEST_SEND_BATCH_SIZE: usize = 1;
 
     // thin wrapper for the run_dos function
     // to avoid specifying everywhere generic parameters
     fn run_dos_no_client(nodes: &[ContactInfo], iterations: usize, params: DosClientParameters) {
-        run_dos::<TpuClient<QuicPool, QuicConnectionManager, QuicConfig>>(
-            nodes, iterations, None, params,
-        );
-    }
-
-    fn build_tpu_quic_client(
-        cluster: &LocalCluster,
-    ) -> Arc<TpuClient<QuicPool, QuicConnectionManager, QuicConfig>> {
-        let rpc_pubsub_url = format!("ws://{}/", cluster.entry_point_info.rpc_pubsub().unwrap());
-        let rpc_url = format!("http://{}", cluster.entry_point_info.rpc().unwrap());
-
-        let ConnectionCache::Quic(cache) = &*cluster.connection_cache else {
-            panic!("Expected a Quic ConnectionCache.");
-        };
-
-        Arc::new(
-            TpuClient::new_with_connection_cache(
-                Arc::new(RpcClient::new(rpc_url)),
-                rpc_pubsub_url.as_str(),
-                TpuClientConfig::default(),
-                cache.clone(),
-            )
-            .unwrap_or_else(|err| {
-                panic!("Could not create TpuClient with Quic Cache {err:?}");
-            }),
-        )
+        run_dos::<QuicTpuClient>(nodes, iterations, None, params);
     }
 
     #[test]
@@ -1003,7 +974,9 @@ pub mod test {
             .unwrap();
         let nodes_slice = [node];
 
-        let client = build_tpu_quic_client(&cluster);
+        let client = Arc::new(cluster.build_tpu_quic_client().unwrap_or_else(|err| {
+            panic!("Could not create TpuClient with Quic Cache {err:?}");
+        }));
 
         // creates one transaction with 8 valid signatures and sends it 10 times
         run_dos(
@@ -1135,7 +1108,9 @@ pub mod test {
             .unwrap();
         let nodes_slice = [node];
 
-        let client = build_tpu_quic_client(&cluster);
+        let client = Arc::new(cluster.build_tpu_quic_client().unwrap_or_else(|err| {
+            panic!("Could not create TpuClient with Quic Cache {err:?}");
+        }));
 
         // creates one transaction and sends it 10 times
         // this is done in single thread

gossip/src/gossip_service.rs

@@ -4,11 +4,7 @@ use {
     crate::{cluster_info::ClusterInfo, legacy_contact_info::LegacyContactInfo as ContactInfo},
     crossbeam_channel::{unbounded, Sender},
     rand::{thread_rng, Rng},
-    solana_client::{
-        connection_cache::ConnectionCache,
-        rpc_client::RpcClient,
-        tpu_client::{TpuClient, TpuClientConfig, TpuClientWrapper},
-    },
+    solana_client::{connection_cache::ConnectionCache, rpc_client::RpcClient},
     solana_perf::recycler::Recycler,
     solana_runtime::bank_forks::BankForks,
     solana_sdk::{
@@ -19,6 +15,7 @@ use {
         socket::SocketAddrSpace,
         streamer::{self, StreamerReceiveStats},
     },
+    solana_tpu_client::tpu_client::{TpuClient, TpuClientConfig, TpuClientWrapper},
     std::{
         collections::HashSet,
         net::{SocketAddr, TcpListener, UdpSocket},

local-cluster/Cargo.toml

@@ -24,6 +24,7 @@ solana-gossip = { workspace = true }
 solana-ledger = { workspace = true }
 solana-logger = { workspace = true }
 solana-pubsub-client = { workspace = true }
+solana-quic-client = { workspace = true }
 solana-rpc-client = { workspace = true }
 solana-rpc-client-api = { workspace = true }
 solana-runtime = { workspace = true }

local-cluster/src/cluster.rs

@@ -3,9 +3,10 @@ use {
     solana_core::validator::{Validator, ValidatorConfig},
     solana_gossip::{cluster_info::Node, contact_info::ContactInfo},
     solana_ledger::shred::Shred,
-    solana_sdk::{pubkey::Pubkey, signature::Keypair},
+    solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Keypair},
     solana_streamer::socket::SocketAddrSpace,
-    std::{path::PathBuf, sync::Arc},
+    solana_tpu_client::tpu_client::QuicTpuClient,
+    std::{io::Result, path::PathBuf, sync::Arc},
 };
 
 pub struct ValidatorInfo {
@@ -38,6 +39,11 @@ impl ClusterValidatorInfo {
 pub trait Cluster {
     fn get_node_pubkeys(&self) -> Vec<Pubkey>;
     fn get_validator_client(&self, pubkey: &Pubkey) -> Option<ThinClient>;
+    fn build_tpu_quic_client(&self) -> Result<QuicTpuClient>;
+    fn build_tpu_quic_client_with_commitment(
+        &self,
+        commitment_config: CommitmentConfig,
+    ) -> Result<QuicTpuClient>;
     fn get_contact_info(&self, pubkey: &Pubkey) -> Option<&ContactInfo>;
     fn exit_node(&mut self, pubkey: &Pubkey) -> ClusterValidatorInfo;
     fn restart_node(