v1.14: generalizes deduper to work with any hashable type (backport of #30753) (#30856)

* generalizes deduper to work with any hashable type (#30753)

generalizes Deduper to work with any hashable type

Current Deduper is hard-coded only for Packet type. In order to use
Deduper in retransmit-stage, we need to dedup types other than Packet.
The commit generalizes Deduper to any hashable type.

(cherry picked from commit 9ad7748)

# Conflicts:
#	core/src/sigverify_stage.rs
#	perf/benches/dedup.rs
#	perf/src/deduper.rs

* resolves mergify merge conflicts

---------

Co-authored-by: behzad nouri <behzadnouri@gmail.com>

Author: mergify[bot]

core/src/shred_fetch_stage.rs

@@ -48,7 +48,7 @@ impl ShredFetchStage {
     ) {
         const STATS_SUBMIT_CADENCE: Duration = Duration::from_secs(1);
         let mut rng = rand::thread_rng();
-        let mut deduper = Deduper::<2>::new(&mut rng, DEDUPER_NUM_BITS);
+        let mut deduper = Deduper::<2, [u8]>::new(&mut rng, DEDUPER_NUM_BITS);
         let mut last_updated = Instant::now();
         let mut keypair = repair_context
             .as_ref()
@@ -252,7 +252,7 @@ fn should_discard_packet<const K: usize>(
     root: Slot,
     max_slot: Slot, // Max slot to ingest shreds for.
     shred_version: u16,
-    deduper: &Deduper<K>,
+    deduper: &Deduper<K, [u8]>,
     should_drop_merkle_shreds: impl Fn(Slot) -> bool,
     stats: &mut ShredFetchStats,
 ) -> bool {
@@ -266,7 +266,11 @@ fn should_discard_packet<const K: usize>(
     ) {
         return true;
     }
-    if deduper.dedup_packet(packet) {
+    if packet
+        .data(..)
+        .map(|data| deduper.dedup(data))
+        .unwrap_or(true)
+    {
         stats.duplicate_shred += 1;
         true
     } else {
@@ -301,7 +305,7 @@ mod tests {
     fn test_data_code_same_index() {
         solana_logger::setup();
         let mut rng = rand::thread_rng();
-        let deduper = Deduper::<2>::new(&mut rng, /*num_bits:*/ 640_007);
+        let deduper = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 640_007);
         let mut packet = Packet::default();
         let mut stats = ShredFetchStats::default();
 
@@ -353,7 +357,7 @@ mod tests {
     fn test_shred_filter() {
         solana_logger::setup();
         let mut rng = rand::thread_rng();
-        let deduper = Deduper::<2>::new(&mut rng, /*num_bits:*/ 640_007);
+        let deduper = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 640_007);
         let mut packet = Packet::default();
         let mut stats = ShredFetchStats::default();
         let last_root = 0;

core/src/sigverify_stage.rs

@@ -14,8 +14,8 @@ use {
     solana_perf::{
         packet::{Packet, PacketBatch},
         sigverify::{
-            count_discarded_packets, count_packets_in_batches, count_valid_packets, shrink_batches,
-            Deduper,
+            count_discarded_packets, count_packets_in_batches, count_valid_packets,
+            dedup_packets_and_count_discards, shrink_batches, Deduper,
         },
     },
     solana_sdk::timing,
@@ -291,8 +291,8 @@ impl SigVerifyStage {
         (shrink_time.as_us(), shrink_total)
     }
 
-    fn verifier<T: SigVerifier>(
-        deduper: &Deduper<2>,
+    fn verifier<const K: usize, T: SigVerifier>(
+        deduper: &Deduper<K, [u8]>,
         recvr: &find_packet_sender_stake_stage::FindPacketSenderStakeReceiver,
         verifier: &mut T,
         stats: &mut SigVerifierStats,
@@ -316,7 +316,8 @@ impl SigVerifyStage {
         discard_random_time.stop();
 
         let mut dedup_time = Measure::start("sigverify_dedup_time");
-        let discard_or_dedup_fail = deduper.dedup_packets_and_count_discards(
+        let discard_or_dedup_fail = dedup_packets_and_count_discards(
+            deduper,
             &mut batches,
             #[inline(always)]
             |received_packet, removed_before_sigverify_stage, is_dup| {
@@ -418,7 +419,7 @@ impl SigVerifyStage {
             .name("solSigVerifier".to_string())
             .spawn(move || {
                 let mut rng = rand::thread_rng();
-                let mut deduper = Deduper::<2>::new(&mut rng, DEDUPER_NUM_BITS);
+                let mut deduper = Deduper::<2, [u8]>::new(&mut rng, DEDUPER_NUM_BITS);
                 loop {
                     if deduper.maybe_reset(&mut rng, DEDUPER_FALSE_POSITIVE_RATE, MAX_DEDUPER_AGE) {
                         stats.num_deduper_saturations += 1;

perf/benches/dedup.rs

@@ -7,7 +7,7 @@ use {
     rand::prelude::*,
     solana_perf::{
         packet::{to_packet_batches, PacketBatch},
-        sigverify::Deduper,
+        sigverify::{self, Deduper},
     },
     std::time::Duration,
     test::Bencher,
@@ -25,9 +25,10 @@ fn test_packet_with_size(size: usize, rng: &mut ThreadRng) -> Vec<u8> {
 fn do_bench_dedup_packets(bencher: &mut Bencher, mut batches: Vec<PacketBatch>) {
     // verify packets
     let mut rng = rand::thread_rng();
-    let mut deduper = Deduper::<2>::new(&mut rng, /*num_bits:*/ 63_999_979);
+    let mut deduper = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 63_999_979);
     bencher.iter(|| {
-        let _ans = deduper.dedup_packets_and_count_discards(&mut batches, |_, _, _| ());
+        let _ans =
+            sigverify::dedup_packets_and_count_discards(&deduper, &mut batches, |_, _, _| ());
         deduper.maybe_reset(
             &mut rng,
             0.001,                  // false_positive_rate
@@ -118,7 +119,7 @@ fn bench_dedup_baseline(bencher: &mut Bencher) {
 #[ignore]
 fn bench_dedup_reset(bencher: &mut Bencher) {
     let mut rng = rand::thread_rng();
-    let mut deduper = Deduper::<2>::new(&mut rng, /*num_bits:*/ 63_999_979);
+    let mut deduper = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 63_999_979);
     bencher.iter(|| {
         deduper.maybe_reset(
             &mut rng,

perf/src/sigverify.rs

@@ -24,8 +24,9 @@ use {
     },
     std::{
         convert::TryFrom,
-        hash::{Hash as _, Hasher},
+        hash::Hasher,
         iter::repeat_with,
+        marker::PhantomData,
         mem::size_of,
         sync::atomic::{AtomicU64, Ordering},
         time::{Duration, Instant},
@@ -495,15 +496,16 @@ pub fn generate_offsets(
     )
 }
 
-pub struct Deduper<const K: usize> {
+pub struct Deduper<const K: usize, T: ?Sized> {
     num_bits: u64,
     bits: Vec<AtomicU64>,
     seeds: [(u128, u128); K],
     clock: Instant,
     popcount: AtomicU64, // Number of one bits in self.bits.
+    _phantom: PhantomData<T>,
 }
 
-impl<const K: usize> Deduper<K> {
+impl<const K: usize, T: ?Sized + std::hash::Hash> Deduper<K, T> {
     pub fn new<R: Rng>(rng: &mut R, num_bits: u64) -> Self {
         let size = num_bits.checked_add(63).unwrap() / 64;
         let size = usize::try_from(size).unwrap();
@@ -513,6 +515,7 @@ impl<const K: usize> Deduper<K> {
             clock: Instant::now(),
             bits: repeat_with(AtomicU64::default).take(size).collect(),
             popcount: AtomicU64::default(),
+            _phantom: PhantomData::<T>::default(),
         }
     }
 
@@ -542,16 +545,14 @@ impl<const K: usize> Deduper<K> {
         saturated
     }
 
-    // Returns true if the packet is duplicate.
+    // Returns true if the data is duplicate.
     #[must_use]
     #[allow(clippy::integer_arithmetic)]
-    pub fn dedup_packet(&self, packet: &Packet) -> bool {
-        // Should not dedup packet if already discarded.
-        debug_assert!(!packet.meta.discard());
+    pub fn dedup(&self, data: &T) -> bool {
         let mut out = true;
         for seed in self.seeds {
             let mut hasher = AHasher::new_with_keys(seed.0, seed.1);
-            packet.data(..).unwrap_or_default().hash(&mut hasher);
+            data.hash(&mut hasher);
             let hash: u64 = hasher.finish() % self.num_bits;
             let index = (hash >> 6) as usize;
             let mask: u64 = 1u64 << (hash & 63);
@@ -563,28 +564,32 @@ impl<const K: usize> Deduper<K> {
         }
         out
     }
+}
 
-    pub fn dedup_packets_and_count_discards(
-        &self,
-        batches: &mut [PacketBatch],
-        mut process_received_packet: impl FnMut(&mut Packet, bool, bool),
-    ) -> u64 {
-        batches
-            .iter_mut()
-            .flat_map(PacketBatch::iter_mut)
-            .map(|packet| {
-                if packet.meta.discard() {
-                    process_received_packet(packet, true, false);
-                } else if self.dedup_packet(packet) {
-                    packet.meta.set_discard(true);
-                    process_received_packet(packet, false, true);
-                } else {
-                    process_received_packet(packet, false, false);
-                }
-                u64::from(packet.meta.discard())
-            })
-            .sum()
-    }
+pub fn dedup_packets_and_count_discards<const K: usize>(
+    deduper: &Deduper<K, [u8]>,
+    batches: &mut [PacketBatch],
+    mut process_received_packet: impl FnMut(&mut Packet, bool, bool),
+) -> u64 {
+    batches
+        .iter_mut()
+        .flat_map(PacketBatch::iter_mut)
+        .map(|packet| {
+            if packet.meta.discard() {
+                process_received_packet(packet, true, false);
+            } else if packet
+                .data(..)
+                .map(|data| deduper.dedup(data))
+                .unwrap_or(true)
+            {
+                packet.meta.set_discard(true);
+                process_received_packet(packet, false, true);
+            } else {
+                process_received_packet(packet, false, false);
+            }
+            u64::from(packet.meta.discard())
+        })
+        .sum()
 }
 
 //inplace shrink a batch of packets
@@ -1606,9 +1611,10 @@ mod tests {
             to_packet_batches(&std::iter::repeat(tx).take(1024).collect::<Vec<_>>(), 128);
         let packet_count = sigverify::count_packets_in_batches(&batches);
         let mut rng = rand::thread_rng();
-        let filter = Deduper::<2>::new(&mut rng, /*num_bits:*/ 63_999_979);
+        let filter = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 63_999_979);
         let mut num_deduped = 0;
-        let discard = filter.dedup_packets_and_count_discards(
+        let discard = dedup_packets_and_count_discards(
+            &filter,
             &mut batches,
             |_deduped_packet, _removed_before_sigverify_stage, _is_dup| {
                 num_deduped += 1;
@@ -1621,9 +1627,10 @@ mod tests {
     #[test]
     fn test_dedup_diff() {
         let mut rng = rand::thread_rng();
-        let mut filter = Deduper::<2>::new(&mut rng, /*num_bits:*/ 63_999_979);
+        let mut filter = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 63_999_979);
         let mut batches = to_packet_batches(&(0..1024).map(|_| test_tx()).collect::<Vec<_>>(), 128);
-        let discard = filter.dedup_packets_and_count_discards(&mut batches, |_, _, _| ()) as usize;
+        let discard =
+            dedup_packets_and_count_discards(&filter, &mut batches, |_, _, _| ()) as usize;
         // because dedup uses a threadpool, there maybe up to N threads of txs that go through
         assert_eq!(discard, 0);
         assert!(!filter.maybe_reset(
@@ -1646,14 +1653,15 @@ mod tests {
         const NUM_BITS: u64 = 63_999_979;
         const FALSE_POSITIVE_RATE: f64 = 0.001;
         let mut rng = rand::thread_rng();
-        let mut filter = Deduper::<2>::new(&mut rng, NUM_BITS);
+        let mut filter = Deduper::<2, [u8]>::new(&mut rng, NUM_BITS);
         let capacity = get_capacity::<2>(NUM_BITS, FALSE_POSITIVE_RATE);
         let mut discard = 0;
         assert!(filter.popcount.load(Ordering::Relaxed) < capacity);
         for i in 0..1000 {
             let mut batches =
                 to_packet_batches(&(0..1000).map(|_| test_tx()).collect::<Vec<_>>(), 128);
-            discard += filter.dedup_packets_and_count_discards(&mut batches, |_, _, _| ()) as usize;
+            discard +=
+                dedup_packets_and_count_discards(&filter, &mut batches, |_, _, _| ()) as usize;
             trace!("{} {}", i, discard);
             if filter.popcount.load(Ordering::Relaxed) > capacity {
                 break;
@@ -1671,12 +1679,13 @@ mod tests {
     #[test]
     fn test_dedup_false_positive() {
         let mut rng = rand::thread_rng();
-        let filter = Deduper::<2>::new(&mut rng, /*num_bits:*/ 63_999_979);
+        let filter = Deduper::<2, [u8]>::new(&mut rng, /*num_bits:*/ 63_999_979);
         let mut discard = 0;
         for i in 0..10 {
             let mut batches =
                 to_packet_batches(&(0..1024).map(|_| test_tx()).collect::<Vec<_>>(), 128);
-            discard += filter.dedup_packets_and_count_discards(&mut batches, |_, _, _| ()) as usize;
+            discard +=
+                dedup_packets_and_count_discards(&filter, &mut batches, |_, _, _| ()) as usize;
             debug!("false positive rate: {}/{}", discard, i * 1024);
         }
         //allow for 1 false positive even if extremely unlikely
@@ -1697,7 +1706,7 @@ mod tests {
     fn test_dedup_capacity(num_bits: u64, false_positive_rate: f64, capacity: u64) {
         let mut rng = rand::thread_rng();
         assert_eq!(get_capacity::<2>(num_bits, false_positive_rate), capacity);
-        let mut deduper = Deduper::<2>::new(&mut rng, num_bits);
+        let mut deduper = Deduper::<2, [u8]>::new(&mut rng, num_bits);
         assert_eq!(deduper.false_positive_rate(), 0.0);
         deduper.popcount.store(capacity, Ordering::Relaxed);
         assert!(deduper.false_positive_rate() < false_positive_rate);
@@ -1726,18 +1735,18 @@ mod tests {
     ) {
         const FALSE_POSITIVE_RATE: f64 = 0.001;
         let mut rng = ChaChaRng::from_seed(seed);
-        let mut deduper = Deduper::<2>::new(&mut rng, num_bits);
+        let mut deduper = Deduper::<2, [u8]>::new(&mut rng, num_bits);
         assert_eq!(get_capacity::<2>(num_bits, FALSE_POSITIVE_RATE), capacity);
         let mut packet = Packet::new([0u8; PACKET_DATA_SIZE], Meta::default());
         let mut dup_count = 0usize;
         for _ in 0..num_packets {
             let size = rng.gen_range(0, PACKET_DATA_SIZE);
             packet.meta.size = size;
             rng.fill(&mut packet.buffer_mut()[0..size]);
-            if deduper.dedup_packet(&packet) {
+            if deduper.dedup(packet.data(..).unwrap()) {
                 dup_count += 1;
             }
-            assert!(deduper.dedup_packet(&packet));
+            assert!(deduper.dedup(packet.data(..).unwrap()));
         }
         assert_eq!(dup_count, num_dups);
         assert_eq!(deduper.popcount.load(Ordering::Relaxed), popcount);