shrink batches when over 80% of the space is wasted

core/src/sigverify_stage.rs

@@ -12,7 +12,7 @@ use {
     itertools::Itertools,
     solana_measure::measure::Measure,
     solana_perf::packet::PacketBatch,
-    solana_perf::sigverify::Deduper,
+    solana_perf::sigverify::{count_valid_packets, shrink_batches, Deduper},
     solana_sdk::timing,
     solana_streamer::streamer::{self, PacketBatchReceiver, StreamerError},
     std::{
@@ -58,6 +58,9 @@ struct SigVerifierStats {
     total_packets: usize,
     total_dedup: usize,
     total_excess_fail: usize,
+    total_shrink_time: usize,
+    total_shrinks: usize,
+    total_valid_packets: usize,
 }
 
 impl SigVerifierStats {
@@ -166,6 +169,9 @@ impl SigVerifierStats {
             ("total_packets", self.total_packets, i64),
             ("total_dedup", self.total_dedup, i64),
             ("total_excess_fail", self.total_excess_fail, i64),
+            ("total_shrink_time", self.total_shrink_time, i64),
+            ("total_shrinks", self.total_shrinks, i64),
+            ("total_valid_packets", self.total_valid_packets, i64),
         );
     }
 }
@@ -248,7 +254,20 @@ impl SigVerifyStage {
         discard_time.stop();
 
         let mut verify_batch_time = Measure::start("sigverify_batch_time");
-        let batches = verifier.verify_batches(batches, num_valid_packets);
+        let mut batches = verifier.verify_batches(batches, num_valid_packets);
+        verify_batch_time.stop();
+
+        let mut shrink_time = Measure::start("sigverify_shrink_time");
+        let num_valid_packets = count_valid_packets(&batches);
+        let start_len = batches.len();
+        const MAX_EMPTY_BATCH_RATIO: usize = 4;
+        if num_packets > num_valid_packets.saturating_mul(MAX_EMPTY_BATCH_RATIO) {
+            let valid = shrink_batches(&mut batches);
+            batches.truncate(valid);
+        }
+        let total_shrinks = start_len.saturating_sub(batches.len());
+        shrink_time.stop();
+
         sendr.send(batches)?;
         verify_batch_time.stop();
 
@@ -282,7 +301,10 @@ impl SigVerifyStage {
         stats.total_batches += batches_len;
         stats.total_packets += num_packets;
         stats.total_dedup += dedup_fail;
+        stats.total_valid_packets += num_valid_packets;
         stats.total_excess_fail += excess_fail;
+        stats.total_shrink_time += shrink_time.as_us() as usize;
+        stats.total_shrinks += total_shrinks;
 
         Ok(())
     }

perf/benches/shrink.rs

@@ -0,0 +1,86 @@
+#![allow(clippy::integer_arithmetic)]
+#![feature(test)]
+
+extern crate test;
+
+use {
+    rand::prelude::*,
+    solana_perf::{
+        packet::{to_packet_batches, PacketBatch, PACKETS_PER_BATCH},
+        sigverify,
+    },
+    test::Bencher,
+};
+
+const NUM_PACKETS: usize = 1024 * 4;
+
+fn test_packet_with_size(size: usize, rng: &mut ThreadRng) -> Vec<u8> {
+    // subtract 8 bytes because the length will get serialized as well
+    (0..size.checked_sub(8).unwrap())
+        .map(|_| rng.gen())
+        .collect()
+}
+
+fn do_bench_shrink_packets(bencher: &mut Bencher, mut batches: Vec<PacketBatch>) {
+    // verify packets
+    bencher.iter(|| {
+        let _ans = sigverify::shrink_batches(&mut batches);
+        batches.iter_mut().for_each(|b| {
+            b.packets
+                .iter_mut()
+                .for_each(|p| p.meta.set_discard(thread_rng().gen()))
+        });
+    });
+}
+
+#[bench]
+#[ignore]
+fn bench_shrink_diff_small_packets(bencher: &mut Bencher) {
+    let mut rng = rand::thread_rng();
+
+    let batches = to_packet_batches(
+        &(0..NUM_PACKETS)
+            .map(|_| test_packet_with_size(128, &mut rng))
+            .collect::<Vec<_>>(),
+        PACKETS_PER_BATCH,
+    );
+
+    do_bench_shrink_packets(bencher, batches);
+}
+
+#[bench]
+#[ignore]
+fn bench_shrink_diff_big_packets(bencher: &mut Bencher) {
+    let mut rng = rand::thread_rng();
+
+    let batches = to_packet_batches(
+        &(0..NUM_PACKETS)
+            .map(|_| test_packet_with_size(1024, &mut rng))
+            .collect::<Vec<_>>(),
+        PACKETS_PER_BATCH,
+    );
+
+    do_bench_shrink_packets(bencher, batches);
+}
+
+#[bench]
+#[ignore]
+fn bench_shrink_count_packets(bencher: &mut Bencher) {
+    let mut rng = rand::thread_rng();
+
+    let mut batches = to_packet_batches(
+        &(0..NUM_PACKETS)
+            .map(|_| test_packet_with_size(128, &mut rng))
+            .collect::<Vec<_>>(),
+        PACKETS_PER_BATCH,
+    );
+    batches.iter_mut().for_each(|b| {
+        b.packets
+            .iter_mut()
+            .for_each(|p| p.meta.set_discard(thread_rng().gen()))
+    });
+
+    bencher.iter(|| {
+        let _ = sigverify::count_valid_packets(&batches);
+    });
+}