refactor: use `Arc` to eliminate potentially expensive cloning (#7638)

use arc for send_data_async

client/src/connection_cache.rs

@@ -168,7 +168,7 @@ pub(crate) use dispatch;
 impl ClientConnection for BlockingClientConnection {
     dispatch!(fn server_addr(&self) -> &SocketAddr);
     dispatch!(fn send_data(&self, buffer: &[u8]) -> TransportResult<()>);
-    dispatch!(fn send_data_async(&self, buffer: Vec<u8>) -> TransportResult<()>);
+    dispatch!(fn send_data_async(&self, buffer: Arc<Vec<u8>>) -> TransportResult<()>);
     dispatch!(fn send_data_batch(&self, buffers: &[Vec<u8>]) -> TransportResult<()>);
     dispatch!(fn send_data_batch_async(&self, buffers: Vec<Vec<u8>>) -> TransportResult<()>);
 }

connection-cache/src/client_connection.rs

@@ -1,7 +1,10 @@
 use {
     solana_metrics::MovingStat,
     solana_transaction_error::TransportResult,
-    std::{net::SocketAddr, sync::atomic::AtomicU64},
+    std::{
+        net::SocketAddr,
+        sync::{atomic::AtomicU64, Arc},
+    },
 };
 
 #[derive(Default)]
@@ -35,7 +38,7 @@ pub trait ClientConnection: Sync + Send {
 
     fn send_data(&self, buffer: &[u8]) -> TransportResult<()>;
 
-    fn send_data_async(&self, buffer: Vec<u8>) -> TransportResult<()>;
+    fn send_data_async(&self, buffer: Arc<Vec<u8>>) -> TransportResult<()>;
 
     fn send_data_batch(&self, buffers: &[Vec<u8>]) -> TransportResult<()>;
 

connection-cache/src/connection_cache.rs

@@ -650,7 +650,7 @@ mod tests {
         fn send_data(&self, _buffer: &[u8]) -> TransportResult<()> {
             unimplemented!()
         }
-        fn send_data_async(&self, _data: Vec<u8>) -> TransportResult<()> {
+        fn send_data_async(&self, _data: Arc<Vec<u8>>) -> TransportResult<()> {
             unimplemented!()
         }
         fn send_data_batch(&self, _buffers: &[Vec<u8>]) -> TransportResult<()> {

core/src/voting_service.rs

@@ -67,7 +67,7 @@ fn send_vote_transaction(
                 .tpu(connection_cache.protocol())
         })
         .ok_or(SendVoteError::InvalidTpuAddress)?;
-    let buf = serialize(transaction)?;
+    let buf = Arc::new(serialize(transaction)?);
     let client = connection_cache.get_connection(&tpu);
 
     client.send_data_async(buf).map_err(|err| {

quic-client/src/quic_client.rs

@@ -80,7 +80,7 @@ pub fn get_runtime() -> &'static Runtime {
 
 async fn send_data_async(
     connection: Arc<NonblockingQuicConnection>,
-    buffer: Vec<u8>,
+    buffer: Arc<Vec<u8>>,
 ) -> TransportResult<()> {
     let result = timeout(SEND_DATA_TIMEOUT, connection.send_data(&buffer)).await;
     ASYNC_TASK_SEMAPHORE.release();
@@ -160,7 +160,7 @@ impl ClientConnection for QuicClientConnection {
         Ok(())
     }
 
-    fn send_data_async(&self, data: Vec<u8>) -> TransportResult<()> {
+    fn send_data_async(&self, data: Arc<Vec<u8>>) -> TransportResult<()> {
         let _lock = ASYNC_TASK_SEMAPHORE.acquire();
         let inner = self.inner.clone();
 

tpu-client/src/tpu_client.rs

@@ -110,7 +110,7 @@ where
         transaction: &Transaction,
     ) -> TransportResult<()> {
         let wire_transaction =
-            bincode::serialize(&transaction).expect("should serialize transaction");
+            Arc::new(bincode::serialize(&transaction).expect("should serialize transaction"));
 
         let leaders = self
             .tpu_client

udp-client/src/udp_client.rs

@@ -30,7 +30,7 @@ impl ClientConnection for UdpClientConnection {
         &self.addr
     }
 
-    fn send_data_async(&self, data: Vec<u8>) -> TransportResult<()> {
+    fn send_data_async(&self, data: Arc<Vec<u8>>) -> TransportResult<()> {
         self.socket.send_to(data.as_ref(), self.addr)?;
         Ok(())
     }