poc for use tpu client next forwarding

Cargo.toml

@@ -492,6 +492,7 @@ solana-system-program = { path = "programs/system", version = "=2.1.0" }
 solana-test-validator = { path = "test-validator", version = "=2.1.0" }
 solana-thin-client = { path = "thin-client", version = "=2.1.0" }
 solana-tpu-client = { path = "tpu-client", version = "=2.1.0", default-features = false }
+solana-tpu-client-next = { path = "tpu-client-next", version = "=2.1.0" }
 solana-transaction-status = { path = "transaction-status", version = "=2.1.0" }
 solana-transaction-status-client-types = { path = "transaction-status-client-types", version = "=2.1.0" }
 solana-transaction-metrics-tracker = { path = "transaction-metrics-tracker", version = "=2.1.0" }

banking-bench/src/main.rs

@@ -5,7 +5,6 @@ use {
     log::*,
     rand::{thread_rng, Rng},
     rayon::prelude::*,
-    solana_client::connection_cache::ConnectionCache,
     solana_core::{
         banking_stage::BankingStage,
         banking_trace::{BankingPacketBatch, BankingTracer, BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT},
@@ -35,7 +34,6 @@ use {
         transaction::Transaction,
     },
     solana_streamer::socket::SocketAddrSpace,
-    solana_tpu_client::tpu_client::DEFAULT_TPU_CONNECTION_POOL_SIZE,
     std::{
         sync::{atomic::Ordering, Arc, RwLock},
         thread::sleep,
@@ -449,17 +447,6 @@ fn main() {
         ClusterInfo::new(node.info, keypair, SocketAddrSpace::Unspecified)
     };
     let cluster_info = Arc::new(cluster_info);
-    let tpu_disable_quic = matches.is_present("tpu_disable_quic");
-    let connection_cache = match tpu_disable_quic {
-        false => ConnectionCache::new_quic(
-            "connection_cache_banking_bench_quic",
-            DEFAULT_TPU_CONNECTION_POOL_SIZE,
-        ),
-        true => ConnectionCache::with_udp(
-            "connection_cache_banking_bench_udp",
-            DEFAULT_TPU_CONNECTION_POOL_SIZE,
-        ),
-    };
     let banking_stage = BankingStage::new_num_threads(
         block_production_method,
         &cluster_info,
@@ -471,10 +458,8 @@ fn main() {
         None,
         replay_vote_sender,
         None,
-        Arc::new(connection_cache),
         bank_forks.clone(),
         &Arc::new(PrioritizationFeeCache::new(0u64)),
-        false,
     );
 
     // This is so that the signal_receiver does not go out of scope after the closure.

core/Cargo.toml

@@ -17,6 +17,7 @@ codecov = { repository = "solana-labs/solana", branch = "master", service = "git
 ahash = { workspace = true }
 anyhow = { workspace = true }
 arrayvec = { workspace = true }
+async-trait = { workspace = true }
 base64 = { workspace = true }
 bincode = { workspace = true }
 bs58 = { workspace = true }
@@ -83,6 +84,7 @@ solana-svm = { workspace = true }
 solana-svm-transaction = { workspace = true }
 solana-timings = { workspace = true }
 solana-tpu-client = { workspace = true }
+solana-tpu-client-next = { workspace = true }
 solana-transaction-status = { workspace = true }
 solana-turbine = { workspace = true }
 solana-unified-scheduler-pool = { workspace = true }
@@ -96,6 +98,7 @@ sys-info = { workspace = true }
 tempfile = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
+tokio-util = { workspace = true }
 trees = { workspace = true }
 
 [dev-dependencies]

core/benches/banking_stage.rs

@@ -13,7 +13,6 @@ use {
     log::*,
     rand::{thread_rng, Rng},
     rayon::prelude::*,
-    solana_client::connection_cache::ConnectionCache,
     solana_core::{
         banking_stage::{
             committer::Committer,
@@ -300,10 +299,8 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
         None,
         s,
         None,
-        Arc::new(ConnectionCache::new("connection_cache_test")),
         bank_forks,
         &Arc::new(PrioritizationFeeCache::new(0u64)),
-        false,
     );
 
     let chunk_len = verified.len() / CHUNKS;

core/benches/sigverify_stage.rs

@@ -158,7 +158,7 @@ fn bench_sigverify_stage(bencher: &mut Bencher, use_same_tx: bool) {
     trace!("start");
     let (packet_s, packet_r) = unbounded();
     let (verified_s, verified_r) = BankingTracer::channel_for_test();
-    let verifier = TransactionSigVerifier::new(verified_s);
+    let verifier = TransactionSigVerifier::new(verified_s, None);
     let stage = SigVerifyStage::new(packet_r, verifier, "solSigVerBench", "bench");
 
     bencher.iter(move || {
@@ -237,7 +237,7 @@ fn prepare_batches(discard_factor: i32) -> (Vec<PacketBatch>, usize) {
 fn bench_shrink_sigverify_stage_core(bencher: &mut Bencher, discard_factor: i32) {
     let (batches0, num_valid_packets) = prepare_batches(discard_factor);
     let (verified_s, _verified_r) = BankingTracer::channel_for_test();
-    let verifier = TransactionSigVerifier::new(verified_s);
+    let verifier = TransactionSigVerifier::new(verified_s, None);
 
     let mut c = 0;
     let mut total_shrink_time = 0;

core/src/banking_simulation.rs

@@ -12,7 +12,6 @@ use {
     crossbeam_channel::{unbounded, Sender},
     itertools::Itertools,
     log::*,
-    solana_client::connection_cache::ConnectionCache,
     solana_gossip::{
         cluster_info::{ClusterInfo, Node},
         contact_info::ContactInfo,
@@ -762,7 +761,6 @@ impl BankingSimulator {
         let (tpu_vote_sender, tpu_vote_receiver) = retracer.create_channel_tpu_vote();
         let (gossip_vote_sender, gossip_vote_receiver) = retracer.create_channel_gossip_vote();
 
-        let connection_cache = Arc::new(ConnectionCache::new("connection_cache_sim"));
         let (replay_vote_sender, _replay_vote_receiver) = unbounded();
         let (retransmit_slots_sender, retransmit_slots_receiver) = unbounded();
         let shred_version = compute_shred_version(
@@ -811,10 +809,8 @@ impl BankingSimulator {
             None,
             replay_vote_sender,
             None,
-            connection_cache,
             bank_forks.clone(),
             prioritization_fee_cache,
-            false,
         );
 
         let (&_slot, &raw_base_event_time) = freeze_time_by_slot