Rename Packets to PacketBatch (backport #21794)

banking-bench/src/main.rs

@@ -13,7 +13,7 @@ use {
         get_tmp_ledger_path,
     },
     solana_measure::measure::Measure,
-    solana_perf::packet::to_packets_chunked,
+    solana_perf::packet::to_packet_batches,
     solana_poh::poh_recorder::{create_test_recorder, PohRecorder, WorkingBankEntry},
     solana_runtime::{
         accounts_background_service::AbsRequestSender, bank::Bank, bank_forks::BankForks,
@@ -212,7 +212,7 @@ fn main() {
         bank.clear_signatures();
     }
 
-    let mut verified: Vec<_> = to_packets_chunked(&transactions, packets_per_chunk);
+    let mut verified: Vec<_> = to_packet_batches(&transactions, packets_per_chunk);
     let ledger_path = get_tmp_ledger_path!();
     {
         let blockstore = Arc::new(
@@ -364,7 +364,7 @@ fn main() {
                     let sig: Vec<u8> = (0..64).map(|_| thread_rng().gen::<u8>()).collect();
                     tx.signatures[0] = Signature::new(&sig[0..64]);
                 }
-                verified = to_packets_chunked(&transactions.clone(), packets_per_chunk);
+                verified = to_packet_batches(&transactions.clone(), packets_per_chunk);
             }
 
             start += chunk_len;

bench-streamer/src/main.rs

@@ -2,8 +2,8 @@
 use {
     clap::{crate_description, crate_name, App, Arg},
     solana_streamer::{
-        packet::{Packet, Packets, PacketsRecycler, PACKET_DATA_SIZE},
-        streamer::{receiver, PacketReceiver},
+        packet::{Packet, PacketBatch, PacketBatchRecycler, PACKET_DATA_SIZE},
+        streamer::{receiver, PacketBatchReceiver},
     },
     std::{
         cmp::max,
@@ -20,19 +20,19 @@ use {
 
 fn producer(addr: &SocketAddr, exit: Arc<AtomicBool>) -> JoinHandle<()> {
     let send = UdpSocket::bind("0.0.0.0:0").unwrap();
-    let mut msgs = Packets::default();
-    msgs.packets.resize(10, Packet::default());
-    for w in msgs.packets.iter_mut() {
+    let mut packet_batch = PacketBatch::default();
+    packet_batch.packets.resize(10, Packet::default());
+    for w in packet_batch.packets.iter_mut() {
         w.meta.size = PACKET_DATA_SIZE;
         w.meta.set_addr(addr);
     }
-    let msgs = Arc::new(msgs);
+    let packet_batch = Arc::new(packet_batch);
     spawn(move || loop {
         if exit.load(Ordering::Relaxed) {
             return;
         }
         let mut num = 0;
-        for p in &msgs.packets {
+        for p in &packet_batch.packets {
             let a = p.meta.addr();
             assert!(p.meta.size <= PACKET_DATA_SIZE);
             send.send_to(&p.data[..p.meta.size], &a).unwrap();
@@ -42,14 +42,14 @@ fn producer(addr: &SocketAddr, exit: Arc<AtomicBool>) -> JoinHandle<()> {
     })
 }
 
-fn sink(exit: Arc<AtomicBool>, rvs: Arc<AtomicUsize>, r: PacketReceiver) -> JoinHandle<()> {
+fn sink(exit: Arc<AtomicBool>, rvs: Arc<AtomicUsize>, r: PacketBatchReceiver) -> JoinHandle<()> {
     spawn(move || loop {
         if exit.load(Ordering::Relaxed) {
             return;
         }
         let timer = Duration::new(1, 0);
-        if let Ok(msgs) = r.recv_timeout(timer) {
-            rvs.fetch_add(msgs.packets.len(), Ordering::Relaxed);
+        if let Ok(packet_batch) = r.recv_timeout(timer) {
+            rvs.fetch_add(packet_batch.packets.len(), Ordering::Relaxed);
         }
     })
 }
@@ -81,7 +81,7 @@ fn main() -> Result<()> {
 
     let mut read_channels = Vec::new();
     let mut read_threads = Vec::new();
-    let recycler = PacketsRecycler::default();
+    let recycler = PacketBatchRecycler::default();
     for _ in 0..num_sockets {
         let read = solana_net_utils::bind_to(ip_addr, port, false).unwrap();
         read.set_read_timeout(Some(Duration::new(1, 0))).unwrap();

core/benches/banking_stage.rs

@@ -20,7 +20,7 @@ use {
         genesis_utils::{create_genesis_config, GenesisConfigInfo},
         get_tmp_ledger_path,
     },
-    solana_perf::{packet::to_packets_chunked, test_tx::test_tx},
+    solana_perf::{packet::to_packet_batches, test_tx::test_tx},
     solana_poh::poh_recorder::{create_test_recorder, WorkingBankEntry},
     solana_runtime::{bank::Bank, cost_model::CostModel},
     solana_sdk::{
@@ -77,11 +77,11 @@ fn bench_consume_buffered(bencher: &mut Bencher) {
         let tx = test_tx();
         let len = 4096;
         let chunk_size = 1024;
-        let batches = to_packets_chunked(&vec![tx; len], chunk_size);
-        let mut packets = VecDeque::new();
+        let batches = to_packet_batches(&vec![tx; len], chunk_size);
+        let mut packet_batches = VecDeque::new();
         for batch in batches {
             let batch_len = batch.packets.len();
-            packets.push_back((batch, vec![0usize; batch_len], false));
+            packet_batches.push_back((batch, vec![0usize; batch_len], false));
         }
         let (s, _r) = unbounded();
         // This tests the performance of buffering packets.
@@ -91,7 +91,7 @@ fn bench_consume_buffered(bencher: &mut Bencher) {
                 &my_pubkey,
                 std::u128::MAX,
                 &poh_recorder,
-                &mut packets,
+                &mut packet_batches,
                 None,
                 &s,
                 None::<Box<dyn Fn()>>,
@@ -206,7 +206,7 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
         assert!(r.is_ok(), "sanity parallel execution");
     }
     bank.clear_signatures();
-    let verified: Vec<_> = to_packets_chunked(&transactions, PACKETS_PER_BATCH);
+    let verified: Vec<_> = to_packet_batches(&transactions, PACKETS_PER_BATCH);
     let ledger_path = get_tmp_ledger_path!();
     {
         let blockstore = Arc::new(

core/benches/sigverify_stage.rs

@@ -8,7 +8,7 @@ use {
     log::*,
     rand::{thread_rng, Rng},
     solana_core::{sigverify::TransactionSigVerifier, sigverify_stage::SigVerifyStage},
-    solana_perf::{packet::to_packets_chunked, test_tx::test_tx},
+    solana_perf::{packet::to_packet_batches, test_tx::test_tx},
     solana_sdk::{
         hash::Hash,
         signature::{Keypair, Signer},
@@ -28,7 +28,7 @@ fn bench_packet_discard(bencher: &mut Bencher) {
     let len = 30 * 1000;
     let chunk_size = 1024;
     let tx = test_tx();
-    let mut batches = to_packets_chunked(&vec![tx; len], chunk_size);
+    let mut batches = to_packet_batches(&vec![tx; len], chunk_size);
 
     let mut total = 0;
 
@@ -74,7 +74,7 @@ fn bench_sigverify_stage(bencher: &mut Bencher) {
     let chunk_size = 1024;
     let mut batches = if use_same_tx {
         let tx = test_tx();
-        to_packets_chunked(&vec![tx; len], chunk_size)
+        to_packet_batches(&vec![tx; len], chunk_size)
     } else {
         let from_keypair = Keypair::new();
         let to_keypair = Keypair::new();
@@ -89,7 +89,7 @@ fn bench_sigverify_stage(bencher: &mut Bencher) {
                 )
             })
             .collect();
-        to_packets_chunked(&txs, chunk_size)
+        to_packet_batches(&txs, chunk_size)
     };
 
     trace!(

core/src/ancestor_hashes_service.rs

@@ -14,7 +14,7 @@ use {
     solana_ledger::{blockstore::Blockstore, shred::SIZE_OF_NONCE},
     solana_measure::measure::Measure,
     solana_perf::{
-        packet::{limited_deserialize, Packet, Packets},
+        packet::{limited_deserialize, Packet, PacketBatch},
         recycler::Recycler,
     },
     solana_runtime::bank::Bank,
@@ -23,7 +23,7 @@ use {
         pubkey::Pubkey,
         timing::timestamp,
     },
-    solana_streamer::streamer::{self, PacketReceiver},
+    solana_streamer::streamer::{self, PacketBatchReceiver},
     std::{
         collections::HashSet,
         net::UdpSocket,
@@ -197,7 +197,7 @@ impl AncestorHashesService {
     /// Listen for responses to our ancestors hashes repair requests
     fn run_responses_listener(
         ancestor_hashes_request_statuses: Arc<DashMap<Slot, DeadSlotAncestorRequestStatus>>,
-        response_receiver: PacketReceiver,
+        response_receiver: PacketBatchReceiver,
         blockstore: Arc<Blockstore>,
         outstanding_requests: Arc<RwLock<OutstandingAncestorHashesRepairs>>,
         exit: Arc<AtomicBool>,
@@ -240,7 +240,7 @@ impl AncestorHashesService {
     /// Process messages from the network
     fn process_new_packets_from_channel(
         ancestor_hashes_request_statuses: &DashMap<Slot, DeadSlotAncestorRequestStatus>,
-        response_receiver: &PacketReceiver,
+        response_receiver: &PacketBatchReceiver,
         blockstore: &Blockstore,
         outstanding_requests: &RwLock<OutstandingAncestorHashesRepairs>,
         stats: &mut AncestorHashesResponsesStats,
@@ -249,28 +249,28 @@ impl AncestorHashesService {
         retryable_slots_sender: &RetryableSlotsSender,
     ) -> Result<()> {
         let timeout = Duration::new(1, 0);
-        let mut responses = vec![response_receiver.recv_timeout(timeout)?];
-        let mut total_packets = responses[0].packets.len();
+        let mut packet_batches = vec![response_receiver.recv_timeout(timeout)?];
+        let mut total_packets = packet_batches[0].packets.len();
 
         let mut dropped_packets = 0;
-        while let Ok(more) = response_receiver.try_recv() {
-            total_packets += more.packets.len();
+        while let Ok(batch) = response_receiver.try_recv() {
+            total_packets += batch.packets.len();
             if total_packets < *max_packets {
                 // Drop the rest in the channel in case of DOS
-                responses.push(more);
+                packet_batches.push(batch);
             } else {
-                dropped_packets += more.packets.len();
+                dropped_packets += batch.packets.len();
             }
         }
 
         stats.dropped_packets += dropped_packets;
         stats.total_packets += total_packets;
 
         let mut time = Measure::start("ancestor_hashes::handle_packets");
-        for response in responses {
-            Self::process_single_packets(
+        for packet_batch in packet_batches {
+            Self::process_packet_batch(
                 ancestor_hashes_request_statuses,
-                response,
+                packet_batch,
                 stats,
                 outstanding_requests,
                 blockstore,
@@ -289,16 +289,16 @@ impl AncestorHashesService {
         Ok(())
     }
 
-    fn process_single_packets(
+    fn process_packet_batch(
         ancestor_hashes_request_statuses: &DashMap<Slot, DeadSlotAncestorRequestStatus>,
-        packets: Packets,
+        packet_batch: PacketBatch,
         stats: &mut AncestorHashesResponsesStats,
         outstanding_requests: &RwLock<OutstandingAncestorHashesRepairs>,
         blockstore: &Blockstore,
         duplicate_slots_reset_sender: &DuplicateSlotsResetSender,
         retryable_slots_sender: &RetryableSlotsSender,
     ) {
-        packets.packets.iter().for_each(|packet| {
+        packet_batch.packets.iter().for_each(|packet| {
             let decision = Self::verify_and_process_ancestor_response(
                 packet,
                 ancestor_hashes_request_statuses,
@@ -871,7 +871,7 @@ mod test {
         t_listen: JoinHandle<()>,
         exit: Arc<AtomicBool>,
         responder_info: ContactInfo,
-        response_receiver: PacketReceiver,
+        response_receiver: PacketBatchReceiver,
         correct_bank_hashes: HashMap<Slot, Hash>,
     }