Update TpuConnection interface to be compatible with versioned txs (#23760)

* Update TpuConnection interface to be compatible with versioned txs

* Add convenience method for sending txs

* use parallel iterator to serialize transactions

client/src/tpu_connection.rs

@@ -1,5 +1,6 @@
 use {
-    solana_sdk::{transaction::Transaction, transport::Result as TransportResult},
+    rayon::iter::{IntoParallelRefIterator, ParallelIterator},
+    solana_sdk::{transaction::VersionedTransaction, transport::Result as TransportResult},
     std::net::{SocketAddr, UdpSocket},
 };
 
@@ -10,12 +11,35 @@ pub trait TpuConnection {
 
     fn tpu_addr(&self) -> &SocketAddr;
 
-    fn send_transaction(&self, tx: &Transaction) -> TransportResult<()> {
-        let data = bincode::serialize(tx).expect("serialize Transaction in send_transaction");
-        self.send_wire_transaction(&data)
+    fn serialize_and_send_transaction(
+        &self,
+        transaction: &VersionedTransaction,
+    ) -> TransportResult<()> {
+        let wire_transaction =
+            bincode::serialize(transaction).expect("serialize Transaction in send_batch");
+        self.send_wire_transaction(&wire_transaction)
     }
 
-    fn send_wire_transaction(&self, data: &[u8]) -> TransportResult<()>;
+    fn send_wire_transaction(&self, wire_transaction: &[u8]) -> TransportResult<()>;
 
-    fn send_batch(&self, transactions: &[Transaction]) -> TransportResult<()>;
+    fn par_serialize_and_send_transaction_batch(
+        &self,
+        transaction_batch: &[VersionedTransaction],
+    ) -> TransportResult<()> {
+        let wire_transaction_batch: Vec<_> = transaction_batch
+            .par_iter()
+            .map(|tx| bincode::serialize(&tx).expect("serialize Transaction in send_batch"))
+            .collect();
+        self.send_wire_transaction_batch(&wire_transaction_batch)
+    }
+
+    fn send_wire_transaction_batch(
+        &self,
+        wire_transaction_batch: &[Vec<u8>],
+    ) -> TransportResult<()> {
+        for wire_transaction in wire_transaction_batch {
+            self.send_wire_transaction(wire_transaction)?;
+        }
+        Ok(())
+    }
 }