Use connection cache in send transaction (#23712)

Use connection cache in send transaction (#23712)

Cargo.lock

@@ -5617,7 +5617,9 @@ version = "1.11.0"
 dependencies = [
  "crossbeam-channel",
  "log",
+ "solana-client",
  "solana-logger",
+ "solana-measure",
  "solana-metrics",
  "solana-runtime",
  "solana-sdk",

core/src/replay_stage.rs

@@ -2001,7 +2001,7 @@ impl ReplayStage {
             );
             voting_sender
                 .send(VoteOp::RefreshVote {
-                    tx: vote_tx,
+                    tx: vote_tx.into(),
                     last_voted_slot,
                 })
                 .unwrap_or_else(|err| warn!("Error: {:?}", err));
@@ -2044,7 +2044,7 @@ impl ReplayStage {
             let tower_slots = tower.tower_slots();
             voting_sender
                 .send(VoteOp::PushVote {
-                    tx: vote_tx,
+                    tx: vote_tx.into(),
                     tower_slots,
                     saved_tower: SavedTowerVersions::from(saved_tower),
                 })

core/src/voting_service.rs

@@ -6,7 +6,7 @@ use {
     solana_measure::measure::Measure,
     solana_poh::poh_recorder::PohRecorder,
     solana_runtime::bank_forks::BankForks,
-    solana_sdk::{clock::Slot, transaction::Transaction},
+    solana_sdk::{clock::Slot, transaction::VersionedTransaction},
     std::{
         sync::{Arc, Mutex, RwLock},
         thread::{self, Builder, JoinHandle},
@@ -15,18 +15,18 @@ use {
 
 pub enum VoteOp {
     PushVote {
-        tx: Transaction,
+        tx: VersionedTransaction,
         tower_slots: Vec<Slot>,
         saved_tower: SavedTowerVersions,
     },
     RefreshVote {
-        tx: Transaction,
+        tx: VersionedTransaction,
         last_voted_slot: Slot,
     },
 }
 
 impl VoteOp {
-    fn tx(&self) -> &Transaction {
+    fn tx(&self) -> &VersionedTransaction {
         match self {
             VoteOp::PushVote { tx, .. } => tx,
             VoteOp::RefreshVote { tx, .. } => tx,
@@ -90,7 +90,7 @@ impl VotingService {
 
         let mut measure = Measure::start("vote_tx_send-ms");
         let target_address = target_address.unwrap_or_else(|| cluster_info.my_contact_info().tpu);
-        let _ = get_connection(&target_address).send_transaction(vote_op.tx());
+        let _ = get_connection(&target_address).serialize_and_send_transaction(vote_op.tx());
         measure.stop();
         inc_new_counter_info!("vote_tx_send-ms", measure.as_ms() as usize);
 
@@ -98,12 +98,18 @@ impl VotingService {
             VoteOp::PushVote {
                 tx, tower_slots, ..
             } => {
+                // we can safely unwrap here because the vote tx is constructed
+                // from a legacy transaction in replay stage
+                let tx = tx.into_legacy_transaction().unwrap();
                 cluster_info.push_vote(&tower_slots, tx);
             }
             VoteOp::RefreshVote {
                 tx,
                 last_voted_slot,
             } => {
+                // we can safely unwrap here because the vote tx is constructed
+                // from a legacy transaction in replay stage
+                let tx = tx.into_legacy_transaction().unwrap();
                 cluster_info.refresh_vote(tx, last_voted_slot);
             }
         }

programs/bpf/Cargo.lock

@@ -3595,6 +3595,8 @@ version = "1.11.0"
 dependencies = [
  "crossbeam-channel",
  "log",
+ "solana-client",
+ "solana-measure",
  "solana-metrics",
  "solana-runtime",
  "solana-sdk",

send-transaction-service/Cargo.toml

@@ -12,10 +12,13 @@ edition = "2021"
 [dependencies]
 crossbeam-channel = "0.5"
 log = "0.4.14"
+solana-client = { path = "../client", version = "=1.11.0" }
+solana-measure = { path = "../measure", version = "=1.11.0" }
 solana-metrics = { path = "../metrics", version = "=1.11.0" }
 solana-runtime = { path = "../runtime", version = "=1.11.0" }
 solana-sdk = { path = "../sdk", version = "=1.11.0" }
 
+
 [dev-dependencies]
 solana-logger = { path = "../logger", version = "=1.11.0" }
 

send-transaction-service/src/send_transaction_service.rs

@@ -2,12 +2,14 @@ use {
     crate::tpu_info::TpuInfo,
     crossbeam_channel::{Receiver, RecvTimeoutError},
     log::*,
+    solana_client::connection_cache,
+    solana_measure::measure::Measure,
     solana_metrics::{datapoint_warn, inc_new_counter_info},
     solana_runtime::{bank::Bank, bank_forks::BankForks},
     solana_sdk::{hash::Hash, nonce_account, pubkey::Pubkey, signature::Signature},
     std::{
         collections::hash_map::{Entry, HashMap},
-        net::{SocketAddr, UdpSocket},
+        net::SocketAddr,
         sync::{Arc, RwLock},
         thread::{self, Builder, JoinHandle},
         time::{Duration, Instant},
@@ -134,12 +136,11 @@ impl SendTransactionService {
         let mut last_status_check = Instant::now();
         let mut last_leader_refresh = Instant::now();
         let mut transactions = HashMap::new();
-        let send_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
 
         if let Some(leader_info) = leader_info.as_mut() {
             leader_info.refresh_recent_peers();
         }
-
+        connection_cache::set_use_quic(config.use_quic);
         Builder::new()
             .name("send-tx-sv2".to_string())
             .spawn(move || loop {
@@ -164,11 +165,7 @@ impl SendTransactionService {
                                 })
                                 .unwrap_or_else(|| vec![&tpu_address]);
                             for address in addresses {
-                                Self::send_transaction(
-                                    &send_socket,
-                                    address,
-                                    &transaction_info.wire_transaction,
-                                );
+                                Self::send_transaction(address, &transaction_info.wire_transaction);
                             }
                             if transactions_len < MAX_TRANSACTION_QUEUE_SIZE {
                                 inc_new_counter_info!("send_transaction_service-insert-tx", 1);
@@ -199,7 +196,6 @@ impl SendTransactionService {
                         let _result = Self::process_transactions(
                             &working_bank,
                             &root_bank,
-                            &send_socket,
                             &tpu_address,
                             &mut transactions,
                             &leader_info,
@@ -221,7 +217,6 @@ impl SendTransactionService {
     fn process_transactions<T: TpuInfo>(
         working_bank: &Arc<Bank>,
         root_bank: &Arc<Bank>,
-        send_socket: &UdpSocket,
         tpu_address: &SocketAddr,
         transactions: &mut HashMap<Signature, TransactionInfo>,
         leader_info: &Option<T>,
@@ -292,11 +287,7 @@ impl SendTransactionService {
                         })
                         .unwrap_or_else(|| vec![tpu_address]);
                     for address in addresses {
-                        Self::send_transaction(
-                            send_socket,
-                            address,
-                            &transaction_info.wire_transaction,
-                        );
+                        Self::send_transaction(address, &transaction_info.wire_transaction);
                     }
                     true
                 }
@@ -317,14 +308,20 @@ impl SendTransactionService {
         result
     }
 
-    fn send_transaction(
-        send_socket: &UdpSocket,
-        tpu_address: &SocketAddr,
-        wire_transaction: &[u8],
-    ) {
-        if let Err(err) = send_socket.send_to(wire_transaction, tpu_address) {
+    fn send_transaction(tpu_address: &SocketAddr, wire_transaction: &[u8]) {
+        let mut measure = Measure::start("send_transaction_service-us");
+        let connection = connection_cache::get_connection(tpu_address);
+
+        if let Err(err) = connection.send_wire_transaction(wire_transaction) {
             warn!("Failed to send transaction to {}: {:?}", tpu_address, err);
         }
+        measure.stop();
+        inc_new_counter_info!(
+            "send_transaction_service-us",
+            measure.as_us() as usize,
+            1000,
+            1000
+        );
     }
 
     pub fn join(self) -> thread::Result<()> {
@@ -372,7 +369,6 @@ mod test {
         let (genesis_config, mint_keypair) = create_genesis_config(4);
         let bank = Bank::new_for_tests(&genesis_config);
         let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
-        let send_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
         let tpu_address = "127.0.0.1:0".parse().unwrap();
         let config = Config {
             leader_forward_count: 1,
@@ -419,7 +415,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -448,7 +443,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -477,7 +471,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -506,7 +499,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -536,7 +528,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -576,7 +567,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -594,7 +584,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -617,7 +606,6 @@ mod test {
         let (genesis_config, mint_keypair) = create_genesis_config(4);
         let bank = Bank::new_for_tests(&genesis_config);
         let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
-        let send_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
         let tpu_address = "127.0.0.1:0".parse().unwrap();
         let config = Config {
             leader_forward_count: 1,
@@ -674,7 +662,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -702,7 +689,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -732,7 +718,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -760,7 +745,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -789,7 +773,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -818,7 +801,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -848,7 +830,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,
@@ -873,7 +854,6 @@ mod test {
         let result = SendTransactionService::process_transactions::<NullTpuInfo>(
             &working_bank,
             &root_bank,
-            &send_socket,
             &tpu_address,
             &mut transactions,
             &None,