Dedup v1.8 (#22584)

backport

Cargo.lock

@@ -589,7 +589,7 @@ version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4b9434b9a5aa1450faa3f9cb14ea0e8c53bb5d2b3c1bfd1ab4fc03e9f33fbfb0"
 dependencies = [
- "rustc_version",
+ "rustc_version 0.2.3",
 ]
 
 [[package]]
@@ -2798,7 +2798,7 @@ checksum = "f842b1982eb6c2fe34036a4fbfb06dd185a3f5c8edfaacdf7d1ea10b07de6252"
 dependencies = [
  "lock_api 0.3.4",
  "parking_lot_core 0.6.2",
- "rustc_version",
+ "rustc_version 0.2.3",
 ]
 
 [[package]]
@@ -2832,7 +2832,7 @@ dependencies = [
  "cloudabi",
  "libc",
  "redox_syscall 0.1.56",
- "rustc_version",
+ "rustc_version 0.2.3",
  "smallvec 0.6.14",
  "winapi 0.3.9",
 ]
@@ -3718,6 +3718,15 @@ dependencies = [
  "semver 0.9.0",
 ]
 
+[[package]]
+name = "rustc_version"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
+dependencies = [
+ "semver 1.0.4",
+]
+
 [[package]]
 name = "rustls"
 version = "0.19.0"
@@ -3881,9 +3890,9 @@ dependencies = [
 
 [[package]]
 name = "serde"
-version = "1.0.130"
+version = "1.0.133"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f12d06de37cf59146fbdecab66aa99f9fe4f78722e3607577a5375d66bd0c913"
+checksum = "97565067517b60e2d1ea8b268e59ce036de907ac523ad83a0475da04e818989a"
 dependencies = [
  "serde_derive",
 ]
@@ -3909,9 +3918,9 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.130"
+version = "1.0.133"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7bc1a1ab1961464eae040d96713baa5a724a8152c1222492465b54322ec508b"
+checksum = "ed201699328568d8d08208fdd080e3ff594e6c422e438b6705905da01005d537"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
@@ -4403,6 +4412,23 @@ dependencies = [
  "solana-version",
 ]
 
+[[package]]
+name = "solana-bloom"
+version = "1.8.14"
+dependencies = [
+ "bv",
+ "fnv",
+ "log 0.4.14",
+ "rand 0.7.3",
+ "rayon",
+ "rustc_version 0.4.0",
+ "serde",
+ "serde_derive",
+ "solana-frozen-abi 1.8.14",
+ "solana-frozen-abi-macro 1.8.14",
+ "solana-sdk",
+]
+
 [[package]]
 name = "solana-bpf-loader-program"
 version = "1.8.14"
@@ -4643,7 +4669,7 @@ dependencies = [
  "rayon",
  "reqwest",
  "retain_mut",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_bytes",
  "serde_derive",
@@ -4652,6 +4678,7 @@ dependencies = [
  "solana-account-decoder",
  "solana-accountsdb-plugin-manager",
  "solana-banks-server",
+ "solana-bloom",
  "solana-clap-utils",
  "solana-client",
  "solana-frozen-abi 1.8.14",
@@ -4804,7 +4831,7 @@ dependencies = [
  "generic-array 0.14.3",
  "log 0.4.14",
  "memmap2",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "sha2",
@@ -4822,7 +4849,7 @@ dependencies = [
  "generic-array 0.14.3",
  "log 0.4.14",
  "memmap2",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "sha2",
@@ -4839,7 +4866,7 @@ checksum = "ceac6e8ad1a784c92ff5f3d6ad68a8d664d389b08055b674c38b2b9abb69e6d4"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "rustc_version",
+ "rustc_version 0.2.3",
  "syn 1.0.60",
 ]
 
@@ -4849,7 +4876,7 @@ version = "1.8.14"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "rustc_version",
+ "rustc_version 0.2.3",
  "syn 1.0.60",
 ]
 
@@ -4903,11 +4930,12 @@ dependencies = [
  "rand 0.7.3",
  "rand_chacha 0.2.2",
  "rayon",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_bytes",
  "serde_derive",
  "serial_test",
+ "solana-bloom",
  "solana-clap-utils",
  "solana-client",
  "solana-frozen-abi 1.8.14",
@@ -5006,7 +5034,7 @@ dependencies = [
  "rayon",
  "reed-solomon-erasure",
  "rocksdb",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_bytes",
  "sha2",
@@ -5239,6 +5267,7 @@ dependencies = [
  "rand 0.7.3",
  "rayon",
  "serde",
+ "solana-bloom",
  "solana-logger 1.8.14",
  "solana-metrics",
  "solana-rayon-threadlimit",
@@ -5305,7 +5334,7 @@ dependencies = [
  "num-derive",
  "num-traits",
  "rand 0.7.3",
- "rustc_version",
+ "rustc_version 0.2.3",
  "rustversion",
  "serde",
  "serde_bytes",
@@ -5341,7 +5370,7 @@ dependencies = [
  "num-derive",
  "num-traits",
  "rand 0.7.3",
- "rustc_version",
+ "rustc_version 0.2.3",
  "rustversion",
  "serde",
  "serde_bytes",
@@ -5491,9 +5520,10 @@ dependencies = [
  "rand 0.7.3",
  "rayon",
  "regex",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
+ "solana-bloom",
  "solana-compute-budget-program",
  "solana-config-program",
  "solana-ed25519-program",
@@ -5556,7 +5586,7 @@ dependencies = [
  "rand 0.7.3",
  "rand_chacha 0.2.2",
  "rand_core 0.6.2",
- "rustc_version",
+ "rustc_version 0.2.3",
  "rustversion",
  "serde",
  "serde_bytes",
@@ -5633,7 +5663,7 @@ dependencies = [
  "num-derive",
  "num-traits",
  "proptest",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "solana-config-program",
@@ -5856,7 +5886,7 @@ name = "solana-version"
 version = "1.8.14"
 dependencies = [
  "log 0.4.14",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "solana-frozen-abi 1.8.14",
@@ -5873,7 +5903,7 @@ dependencies = [
  "log 0.4.14",
  "num-derive",
  "num-traits",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "solana-frozen-abi 1.8.14",
@@ -5988,7 +6018,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d022496b16281348b52d0e30ae99e01a73d737b2f45d38fed4edf79f9325a1d5"
 dependencies = [
  "discard",
- "rustc_version",
+ "rustc_version 0.2.3",
  "stdweb-derive",
  "stdweb-internal-macros",
  "stdweb-internal-runtime",

Cargo.toml

@@ -12,6 +12,7 @@ members = [
     "banks-client",
     "banks-interface",
     "banks-server",
+    "bloom",
     "clap-utils",
     "cli-config",
     "cli-output",

bloom/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "solana-bloom"
+version = "1.8.14"
+description = "Solana bloom filter"
+authors = ["Solana Maintainers <maintainers@solana.foundation>"]
+repository = "https://github.com/solana-labs/solana"
+license = "Apache-2.0"
+homepage = "https://solana.com/"
+documentation = "https://docs.rs/solana-bloom"
+edition = "2018"
+
+[dependencies]
+bv = { version = "0.11.1", features = ["serde"] }
+fnv = "1.0.7"
+rand = "0.7.0"
+serde = { version = "1.0.133", features = ["rc"] }
+rayon = "1.5.1"
+serde_derive = "1.0.103"
+solana-frozen-abi = { path = "../frozen-abi", version = "=1.8.14" }
+solana-frozen-abi-macro = { path = "../frozen-abi/macro", version = "=1.8.14" }
+solana-sdk = { path = "../sdk", version = "=1.8.14" }
+log = "0.4.14"
+
+[lib]
+crate-type = ["lib"]
+name = "solana_bloom"
+
+[package.metadata.docs.rs]
+targets = ["x86_64-unknown-linux-gnu"]
+
+[build-dependencies]
+rustc_version = "0.4"

bloom/benches/bloom.rs

@@ -5,7 +5,7 @@ use {
     bv::BitVec,
     fnv::FnvHasher,
     rand::Rng,
-    solana_runtime::bloom::{AtomicBloom, Bloom, BloomHashIndex},
+    solana_bloom::bloom::{AtomicBloom, Bloom, BloomHashIndex},
     solana_sdk::{
         hash::{hash, Hash},
         signature::Signature,

bloom/build.rs

@@ -0,0 +1 @@
+../frozen-abi/build.rs

bloom/src/bloom.rs

@@ -6,7 +6,9 @@ use {
     serde::{Deserialize, Serialize},
     solana_sdk::sanitize::{Sanitize, SanitizeError},
     std::{
-        cmp, fmt,
+        cmp,
+        convert::TryFrom,
+        fmt,
         hash::Hasher,
         marker::PhantomData,
         sync::atomic::{AtomicU64, Ordering},
@@ -99,7 +101,7 @@ impl<T: BloomHashIndex> Bloom<T> {
         }
     }
     fn pos(&self, key: &T, k: u64) -> u64 {
-        key.hash_at_index(k) % self.bits.len()
+        key.hash_at_index(k).wrapping_rem(self.bits.len())
     }
     pub fn clear(&mut self) {
         self.bits = BitVec::new_fill(false, self.bits.len());
@@ -109,7 +111,7 @@ impl<T: BloomHashIndex> Bloom<T> {
         for k in &self.keys {
             let pos = self.pos(key, *k);
             if !self.bits.get(pos) {
-                self.num_bits_set += 1;
+                self.num_bits_set = self.num_bits_set.saturating_add(1);
                 self.bits.set(pos, true);
             }
         }
@@ -162,13 +164,13 @@ impl<T: BloomHashIndex> From<Bloom<T>> for AtomicBloom<T> {
 
 impl<T: BloomHashIndex> AtomicBloom<T> {
     fn pos(&self, key: &T, hash_index: u64) -> (usize, u64) {
-        let pos = key.hash_at_index(hash_index) % self.num_bits;
+        let pos = key.hash_at_index(hash_index).wrapping_rem(self.num_bits);
         // Divide by 64 to figure out which of the
         // AtomicU64 bit chunks we need to modify.
-        let index = pos >> 6;
+        let index = pos.wrapping_shr(6);
         // (pos & 63) is equivalent to mod 64 so that we can find
         // the index of the bit within the AtomicU64 to modify.
-        let mask = 1u64 << (pos & 63);
+        let mask = 1u64.wrapping_shl(u32::try_from(pos & 63).unwrap());
         (index as usize, mask)
     }
 

bloom/src/lib.rs

@@ -0,0 +1,5 @@
+#![cfg_attr(RUSTC_WITH_SPECIALIZATION, feature(min_specialization))]
+pub mod bloom;
+
+#[macro_use]
+extern crate solana_frozen_abi_macro;

core/Cargo.toml

@@ -48,6 +48,7 @@ solana-account-decoder = { path = "../account-decoder", version = "=1.8.14" }
 solana-accountsdb-plugin-manager = { path = "../accountsdb-plugin-manager", version = "=1.8.14" }
 solana-banks-server = { path = "../banks-server", version = "=1.8.14" }
 solana-clap-utils = { path = "../clap-utils", version = "=1.8.14" }
+solana-bloom = { path = "../bloom", version = "=1.8.14" }
 solana-client = { path = "../client", version = "=1.8.14" }
 solana-gossip = { path = "../gossip", version = "=1.8.14" }
 solana-ledger = { path = "../ledger", version = "=1.8.14" }

core/src/banking_stage.rs

@@ -2,11 +2,9 @@
 //! to contruct a software pipeline. The stage uses all available CPU cores and
 //! can do its processing in parallel with signature verification on the GPU.
 use {
-    crate::packet_hasher::PacketHasher,
     crossbeam_channel::{Receiver as CrossbeamReceiver, RecvTimeoutError},
     histogram::Histogram,
     itertools::Itertools,
-    lru::LruCache,
     retain_mut::RetainMut,
     solana_gossip::{cluster_info::ClusterInfo, contact_info::ContactInfo},
     solana_ledger::{blockstore_processor::TransactionStatusSender, entry::hash_transactions},
@@ -53,7 +51,6 @@ use {
         env,
         mem::size_of,
         net::{SocketAddr, UdpSocket},
-        ops::DerefMut,
         sync::{
             atomic::{AtomicU64, AtomicUsize, Ordering},
             Arc, Mutex, RwLock, RwLockReadGuard,
@@ -80,8 +77,6 @@ const TOTAL_BUFFERED_PACKETS: usize = 500_000;
 
 const MAX_NUM_TRANSACTIONS_PER_BATCH: usize = 128;
 
-const DEFAULT_LRU_SIZE: usize = 200_000;
-
 const NUM_VOTE_PROCESSING_THREADS: u32 = 2;
 const MIN_THREADS_BANKING: u32 = 1;
 
@@ -93,7 +88,6 @@ pub struct BankingStageStats {
     new_tx_count: AtomicUsize,
     dropped_packet_batches_count: AtomicUsize,
     dropped_packets_count: AtomicUsize,
-    dropped_duplicated_packets_count: AtomicUsize,
     newly_buffered_packets_count: AtomicUsize,
     current_buffered_packets_count: AtomicUsize,
     current_buffered_packet_batches_count: AtomicUsize,
@@ -108,7 +102,6 @@ pub struct BankingStageStats {
     process_packets_elapsed: AtomicU64,
     handle_retryable_packets_elapsed: AtomicU64,
     filter_pending_packets_elapsed: AtomicU64,
-    packet_duplicate_check_elapsed: AtomicU64,
     packet_conversion_elapsed: AtomicU64,
     unprocessed_packet_conversion_elapsed: AtomicU64,
     transaction_processing_elapsed: AtomicU64,
@@ -154,12 +147,6 @@ impl BankingStageStats {
                     self.dropped_packets_count.swap(0, Ordering::Relaxed) as i64,
                     i64
                 ),
-                (
-                    "dropped_duplicated_packets_count",
-                    self.dropped_duplicated_packets_count
-                        .swap(0, Ordering::Relaxed) as i64,
-                    i64
-                ),
                 (
                     "newly_buffered_packets_count",
                     self.newly_buffered_packets_count.swap(0, Ordering::Relaxed) as i64,
@@ -222,12 +209,6 @@ impl BankingStageStats {
                         .swap(0, Ordering::Relaxed) as i64,
                     i64
                 ),
-                (
-                    "packet_duplicate_check_elapsed",
-                    self.packet_duplicate_check_elapsed
-                        .swap(0, Ordering::Relaxed) as i64,
-                    i64
-                ),
                 (
                     "packet_conversion_elapsed",
                     self.packet_conversion_elapsed.swap(0, Ordering::Relaxed) as i64,
@@ -348,10 +329,6 @@ impl BankingStage {
         // This thread talks to poh_service and broadcasts the entries once they have been recorded.
         // Once an entry has been recorded, its blockhash is registered with the bank.
         let my_pubkey = cluster_info.id();
-        let duplicates = Arc::new(Mutex::new((
-            LruCache::new(DEFAULT_LRU_SIZE),
-            PacketHasher::default(),
-        )));
         let data_budget = Arc::new(DataBudget::default());
         // Many banks that process transactions in parallel.
         assert!(num_threads >= NUM_VOTE_PROCESSING_THREADS + MIN_THREADS_BANKING);
@@ -375,7 +352,6 @@ impl BankingStage {
                 let mut recv_start = Instant::now();
                 let transaction_status_sender = transaction_status_sender.clone();
                 let gossip_vote_sender = gossip_vote_sender.clone();
-                let duplicates = duplicates.clone();
                 let data_budget = data_budget.clone();
                 let cost_model = cost_model.clone();
                 Builder::new()
@@ -392,7 +368,6 @@ impl BankingStage {
                             batch_limit,
                             transaction_status_sender,
                             gossip_vote_sender,
-                            &duplicates,
                             &data_budget,
                             cost_model,
                         );
@@ -743,7 +718,6 @@ impl BankingStage {
         batch_limit: usize,
         transaction_status_sender: Option<TransactionStatusSender>,
         gossip_vote_sender: ReplayVoteSender,
-        duplicates: &Arc<Mutex<(LruCache<u64, ()>, PacketHasher)>>,
         data_budget: &DataBudget,
         cost_model: Arc<RwLock<CostModel>>,
     ) {
@@ -798,7 +772,6 @@ impl BankingStage {
                 &gossip_vote_sender,
                 &mut buffered_packet_batches,
                 &mut banking_stage_stats,
-                duplicates,
                 &recorder,
                 &cost_model,
             ) {
@@ -1456,7 +1429,6 @@ impl BankingStage {
         gossip_vote_sender: &ReplayVoteSender,
         buffered_packet_batches: &mut UnprocessedPacketBatches,
         banking_stage_stats: &mut BankingStageStats,
-        duplicates: &Arc<Mutex<(LruCache<u64, ()>, PacketHasher)>>,
         recorder: &TransactionRecorder,
         cost_model: &Arc<RwLock<CostModel>>,
     ) -> Result<(), RecvTimeoutError> {
@@ -1493,7 +1465,6 @@ impl BankingStage {
                     &mut dropped_packets_count,
                     &mut newly_buffered_packets_count,
                     batch_limit,
-                    duplicates,
                     banking_stage_stats,
                 );
                 continue;
@@ -1524,7 +1495,6 @@ impl BankingStage {
                 &mut dropped_packets_count,
                 &mut newly_buffered_packets_count,
                 batch_limit,
-                duplicates,
                 banking_stage_stats,
             );
 
@@ -1553,7 +1523,6 @@ impl BankingStage {
                         &mut dropped_packets_count,
                         &mut newly_buffered_packets_count,
                         batch_limit,
-                        duplicates,
                         banking_stage_stats,
                     );
                 }
@@ -1610,40 +1579,13 @@ impl BankingStage {
     fn push_unprocessed(
         unprocessed_packet_batches: &mut UnprocessedPacketBatches,
         packet_batch: PacketBatch,
-        mut packet_indexes: Vec<usize>,
+        packet_indexes: Vec<usize>,
         dropped_packet_batches_count: &mut usize,
         dropped_packets_count: &mut usize,
         newly_buffered_packets_count: &mut usize,
         batch_limit: usize,
-        duplicates: &Arc<Mutex<(LruCache<u64, ()>, PacketHasher)>>,
         banking_stage_stats: &mut BankingStageStats,
     ) {
-        {
-            let original_packets_count = packet_indexes.len();
-            let mut packet_duplicate_check_time = Measure::start("packet_duplicate_check");
-            let mut duplicates = duplicates.lock().unwrap();
-            let (cache, hasher) = duplicates.deref_mut();
-            packet_indexes.retain(|i| {
-                let packet_hash = hasher.hash_packet(&packet_batch.packets[*i]);
-                match cache.get_mut(&packet_hash) {
-                    Some(_hash) => false,
-                    None => {
-                        cache.put(packet_hash, ());
-                        true
-                    }
-                }
-            });
-            packet_duplicate_check_time.stop();
-            banking_stage_stats
-                .packet_duplicate_check_elapsed
-                .fetch_add(packet_duplicate_check_time.as_us(), Ordering::Relaxed);
-            banking_stage_stats
-                .dropped_duplicated_packets_count
-                .fetch_add(
-                    original_packets_count.saturating_sub(packet_indexes.len()),
-                    Ordering::Relaxed,
-                );
-        }
         if Self::packet_has_more_unprocessed_transactions(&packet_indexes) {
             if unprocessed_packet_batches.len() >= batch_limit {
                 *dropped_packet_batches_count += 1;
@@ -3221,10 +3163,6 @@ mod tests {
         let new_packet_batch = PacketBatch::new(vec![Packet::default()]);
         let packet_indexes = vec![];
 
-        let duplicates = Arc::new(Mutex::new((
-            LruCache::new(DEFAULT_LRU_SIZE),
-            PacketHasher::default(),
-        )));
         let mut dropped_packet_batches_count = 0;
         let mut dropped_packets_count = 0;
         let mut newly_buffered_packets_count = 0;
@@ -3239,7 +3177,6 @@ mod tests {
             &mut dropped_packets_count,
             &mut newly_buffered_packets_count,
             batch_limit,
-            &duplicates,
             &mut banking_stage_stats,
         );
         assert_eq!(unprocessed_packets.len(), 1);
@@ -3258,7 +3195,6 @@ mod tests {
             &mut dropped_packets_count,
             &mut newly_buffered_packets_count,
             batch_limit,
-            &duplicates,
             &mut banking_stage_stats,
         );
         assert_eq!(unprocessed_packets.len(), 2);
@@ -3274,27 +3210,6 @@ mod tests {
         )
         .unwrap()]);
         assert_eq!(unprocessed_packets.len(), batch_limit);
-        BankingStage::push_unprocessed(
-            &mut unprocessed_packets,
-            new_packet_batch.clone(),
-            packet_indexes.clone(),
-            &mut dropped_packet_batches_count,
-            &mut dropped_packets_count,
-            &mut newly_buffered_packets_count,
-            batch_limit,
-            &duplicates,
-            &mut banking_stage_stats,
-        );
-        assert_eq!(unprocessed_packets.len(), 2);
-        assert_eq!(
-            unprocessed_packets[1].0.packets[0],
-            new_packet_batch.packets[0]
-        );
-        assert_eq!(dropped_packet_batches_count, 1);
-        assert_eq!(dropped_packets_count, 2);
-        assert_eq!(newly_buffered_packets_count, 2);
-
-        // Check duplicates are dropped (newly buffered shouldn't change)
         BankingStage::push_unprocessed(
             &mut unprocessed_packets,
             new_packet_batch.clone(),
@@ -3302,8 +3217,7 @@ mod tests {
             &mut dropped_packet_batches_count,
             &mut dropped_packets_count,
             &mut newly_buffered_packets_count,
-            3,
+            batch_limit,
-            &duplicates,
             &mut banking_stage_stats,
         );
         assert_eq!(unprocessed_packets.len(), 2);

core/src/sigverify_stage.rs

@@ -7,10 +7,13 @@
 
 use {
     crate::sigverify,
+    core::time::Duration,
     crossbeam_channel::{SendError, Sender as CrossbeamSender},
     itertools::Itertools,
+    solana_bloom::bloom::{AtomicBloom, Bloom},
     solana_measure::measure::Measure,
     solana_perf::packet::PacketBatch,
+    solana_perf::sigverify::dedup_packets,
     solana_sdk::timing,
     solana_streamer::streamer::{self, PacketBatchReceiver, StreamerError},
     std::{
@@ -50,10 +53,13 @@ struct SigVerifierStats {
     recv_batches_us_hist: histogram::Histogram, // time to call recv_batch
     verify_batches_pp_us_hist: histogram::Histogram, // per-packet time to call verify_batch
     discard_packets_pp_us_hist: histogram::Histogram, // per-packet time to call verify_batch
+    dedup_packets_pp_us_hist: histogram::Histogram, // per-packet time to call verify_batch
     batches_hist: histogram::Histogram,         // number of packet batches per verify call
     packets_hist: histogram::Histogram,         // number of packets per verify call
     total_batches: usize,
     total_packets: usize,
+    total_dedup: usize,
+    total_excess_fail: usize,
 }
 
 impl SigVerifierStats {
@@ -122,6 +128,26 @@ impl SigVerifierStats {
                 self.discard_packets_pp_us_hist.mean().unwrap_or(0),
                 i64
             ),
+            (
+                "dedup_packets_pp_us_90pct",
+                self.dedup_packets_pp_us_hist.percentile(90.0).unwrap_or(0),
+                i64
+            ),
+            (
+                "dedup_packets_pp_us_min",
+                self.dedup_packets_pp_us_hist.minimum().unwrap_or(0),
+                i64
+            ),
+            (
+                "dedup_packets_pp_us_max",
+                self.dedup_packets_pp_us_hist.maximum().unwrap_or(0),
+                i64
+            ),
+            (
+                "dedup_packets_pp_us_mean",
+                self.dedup_packets_pp_us_hist.mean().unwrap_or(0),
+                i64
+            ),
             (
                 "batches_90pct",
                 self.batches_hist.percentile(90.0).unwrap_or(0),
@@ -140,6 +166,8 @@ impl SigVerifierStats {
             ("packets_mean", self.packets_hist.mean().unwrap_or(0), i64),
             ("total_batches", self.total_batches, i64),
             ("total_packets", self.total_packets, i64),
+            ("total_dedup", self.total_dedup, i64),
+            ("total_excess_fail", self.total_excess_fail, i64),
         );
     }
 }
@@ -187,22 +215,36 @@ impl SigVerifyStage {
     }
 
     fn verifier<T: SigVerifier>(
+        bloom: &AtomicBloom<&[u8]>,
         recvr: &PacketBatchReceiver,
         sendr: &CrossbeamSender<Vec<PacketBatch>>,
         verifier: &T,
         stats: &mut SigVerifierStats,
     ) -> Result<()> {
-        let (mut batches, len, recv_time) = streamer::recv_batch(recvr)?;
+        let (mut batches, num_packets, recv_time) = streamer::recv_batch(recvr)?;
+
+        let batches_len = batches.len();
+        debug!(
+            "@{:?} verifier: verifying: {}",
+            timing::timestamp(),
+            num_packets,
+        );
+
+        let mut dedup_time = Measure::start("sigverify_dedup_time");
+        let dedup_fail = dedup_packets(bloom, &mut batches) as usize;
+        dedup_time.stop();
+        let valid_packets = num_packets.saturating_sub(dedup_fail);
+
+        let mut discard_time = Measure::start("sigverify_discard_time");
+        if valid_packets > MAX_SIGVERIFY_BATCH {
+            Self::discard_excess_packets(&mut batches, MAX_SIGVERIFY_BATCH)
+        };
+        let excess_fail = valid_packets.saturating_sub(MAX_SIGVERIFY_BATCH);
+        discard_time.stop();
 
         let mut verify_batch_time = Measure::start("sigverify_batch_time");
-        let batches_len = batches.len();
+        let batches = verifier.verify_batches(batches);
-        debug!("@{:?} verifier: verifying: {}", timing::timestamp(), len,);
+        sendr.send(batches)?;
-        let mut discard_time = Measure::start("sigverify_discard_time");
-        if len > MAX_SIGVERIFY_BATCH {
-            Self::discard_excess_packets(&mut batches, MAX_SIGVERIFY_BATCH);
-        }
-        discard_time.stop();
-        sendr.send(verifier.verify_batches(batches))?;
         verify_batch_time.stop();
 
         debug!(
@@ -210,14 +252,14 @@ impl SigVerifyStage {
             timing::timestamp(),
             batches_len,
             verify_batch_time.as_ms(),
-            len,
+            num_packets,
-            (len as f32 / verify_batch_time.as_s())
+            (num_packets as f32 / verify_batch_time.as_s())
         );
 
         datapoint_debug!(
             "sigverify_stage-total_verify_time",
             ("num_batches", batches_len, i64),
-            ("num_packets", len, i64),
+            ("num_packets", num_packets, i64),
             ("verify_time_ms", verify_batch_time.as_ms(), i64),
             ("recv_time", recv_time, i64),
         );
@@ -228,16 +270,22 @@ impl SigVerifyStage {
             .unwrap();
         stats
             .verify_batches_pp_us_hist
-            .increment(verify_batch_time.as_us() / (len as u64))
+            .increment(verify_batch_time.as_us() / (num_packets as u64))
             .unwrap();
         stats
             .discard_packets_pp_us_hist
-            .increment(discard_time.as_us() / (len as u64))
+            .increment(discard_time.as_us() / (num_packets as u64))
+            .unwrap();
+        stats
+            .dedup_packets_pp_us_hist
+            .increment(dedup_time.as_us() / (num_packets as u64))
             .unwrap();
         stats.batches_hist.increment(batches_len as u64).unwrap();
-        stats.packets_hist.increment(len as u64).unwrap();
+        stats.packets_hist.increment(num_packets as u64).unwrap();
         stats.total_batches += batches_len;
-        stats.total_packets += len;
+        stats.total_packets += num_packets;
+        stats.total_dedup += dedup_fail;
+        stats.total_excess_fail += excess_fail;
 
         Ok(())
     }
@@ -250,29 +298,48 @@ impl SigVerifyStage {
         let verifier = verifier.clone();
         let mut stats = SigVerifierStats::default();
         let mut last_print = Instant::now();
+        const MAX_BLOOM_AGE: Duration = Duration::from_millis(2_000);
+        const MAX_BLOOM_ITEMS: usize = 1_000_000;
+        const MAX_BLOOM_FAIL: f64 = 0.0001;
+        const MAX_BLOOM_BITS: usize = 8 << 22;
         Builder::new()
             .name("solana-verifier".to_string())
-            .spawn(move || loop {
+            .spawn(move || {
-                if let Err(e) =
+                let mut bloom =
-                    Self::verifier(&packet_receiver, &verified_sender, &verifier, &mut stats)
+                    Bloom::random(MAX_BLOOM_ITEMS, MAX_BLOOM_FAIL, MAX_BLOOM_BITS).into();
-                {
+                let mut bloom_age = Instant::now();
-                    match e {
+                loop {
-                        SigVerifyServiceError::Streamer(StreamerError::RecvTimeout(
+                    let now = Instant::now();
-                            RecvTimeoutError::Disconnected,
+                    if now.duration_since(bloom_age) > MAX_BLOOM_AGE {
-                        )) => break,
+                        bloom =
-                        SigVerifyServiceError::Streamer(StreamerError::RecvTimeout(
+                            Bloom::random(MAX_BLOOM_ITEMS, MAX_BLOOM_FAIL, MAX_BLOOM_BITS).into();
-                            RecvTimeoutError::Timeout,
+                        bloom_age = now;
-                        )) => (),
+                    }
-                        SigVerifyServiceError::Send(_) => {
+                    if let Err(e) = Self::verifier(
-                            break;
+                        &bloom,
-                        }
+                        &packet_receiver,
-                        _ => error!("{:?}", e),
+                        &verified_sender,
+                        &verifier,
+                        &mut stats,
+                    ) {
+                        match e {
+                            SigVerifyServiceError::Streamer(StreamerError::RecvTimeout(
+                                RecvTimeoutError::Disconnected,
+                            )) => break,
+                            SigVerifyServiceError::Streamer(StreamerError::RecvTimeout(
+                                RecvTimeoutError::Timeout,
+                            )) => (),
+                            SigVerifyServiceError::Send(_) => {
+                                break;
+                            }
+                            _ => error!("{:?}", e),
+                        }
+                    }
+                    if last_print.elapsed().as_secs() > 2 {
+                        stats.report();
+                        stats = SigVerifierStats::default();
+                        last_print = Instant::now();
                     }
-                }
-                if last_print.elapsed().as_secs() > 2 {
-                    stats.report();
-                    stats = SigVerifierStats::default();
-                    last_print = Instant::now();
                 }
             })
             .unwrap()

gossip/Cargo.toml

@@ -42,6 +42,7 @@ solana-streamer = { path = "../streamer", version = "=1.8.14" }
 solana-sdk = { path = "../sdk", version = "=1.8.14" }
 solana-version = { path = "../version", version = "=1.8.14" }
 solana-vote-program = { path = "../programs/vote", version = "=1.8.14" }
+solana-bloom = { path = "../bloom", version = "=1.8.14" }
 thiserror = "1.0"
 
 [dev-dependencies]

gossip/src/cluster_info.rs

@@ -257,7 +257,7 @@ pub fn make_accounts_hashes_message(
 pub(crate) type Ping = ping_pong::Ping<[u8; GOSSIP_PING_TOKEN_SIZE]>;
 
 // TODO These messages should go through the gpu pipeline for spam filtering
-#[frozen_abi(digest = "7cgH6JHdpxMSuPs6LEZzV5ShLXQMcZftb95s5PZKR5qB")]
+#[frozen_abi(digest = "2trrZsxmuF9SLAEeKsj5dgF8ufXBHL7bcVpkVYvJgTKP")]
 #[derive(Serialize, Deserialize, Debug, AbiEnumVisitor, AbiExample)]
 #[allow(clippy::large_enum_variant)]
 pub(crate) enum Protocol {

gossip/src/crds_gossip_pull.rs

@@ -23,7 +23,7 @@ use {
     lru::LruCache,
     rand::Rng,
     rayon::{prelude::*, ThreadPool},
-    solana_runtime::bloom::{AtomicBloom, Bloom},
+    solana_bloom::bloom::{AtomicBloom, Bloom},
     solana_sdk::{
         hash::{hash, Hash},
         pubkey::Pubkey,

gossip/src/crds_gossip_push.rs

@@ -22,7 +22,7 @@ use {
     indexmap::map::IndexMap,
     lru::LruCache,
     rand::{seq::SliceRandom, Rng},
-    solana_runtime::bloom::{AtomicBloom, Bloom},
+    solana_bloom::bloom::{AtomicBloom, Bloom},
     solana_sdk::{packet::PACKET_DATA_SIZE, pubkey::Pubkey, timing::timestamp},
     solana_streamer::socket::SocketAddrSpace,
     std::{

perf/Cargo.toml

@@ -23,6 +23,7 @@ solana-logger = { path = "../logger", version = "=1.8.14" }
 solana-metrics = { path = "../metrics", version = "=1.8.14" }
 solana-sdk = { path = "../sdk", version = "=1.8.14" }
 solana-rayon-threadlimit = { path = "../rayon-threadlimit", version = "=1.8.14" }
+solana-bloom = { path = "../bloom", version = "=1.8.14" }
 solana-vote-program = { path = "../programs/vote", version = "=1.8.14" }
 
 [target."cfg(target_os = \"linux\")".dependencies]

perf/benches/dedup.rs

@@ -0,0 +1,45 @@
+#![feature(test)]
+
+extern crate test;
+
+use {
+    solana_bloom::bloom::{AtomicBloom, Bloom},
+    solana_perf::{packet::to_packet_batches, sigverify, test_tx::test_tx},
+    test::Bencher,
+};
+
+#[bench]
+fn bench_dedup_same(bencher: &mut Bencher) {
+    let tx = test_tx();
+
+    // generate packet vector
+    let mut batches = to_packet_batches(
+        &std::iter::repeat(tx).take(64 * 1024).collect::<Vec<_>>(),
+        128,
+    );
+    let packet_count = sigverify::count_packets_in_batches(&batches);
+    let bloom: AtomicBloom<&[u8]> = Bloom::random(1_000_000, 0.0001, 8 << 22).into();
+
+    println!("packet_count {} {}", packet_count, batches.len());
+
+    // verify packets
+    bencher.iter(|| {
+        let _ans = sigverify::dedup_packets(&bloom, &mut batches);
+    })
+}
+
+#[bench]
+fn bench_dedup_diff(bencher: &mut Bencher) {
+    // generate packet vector
+    let mut batches =
+        to_packet_batches(&(0..64 * 1024).map(|_| test_tx()).collect::<Vec<_>>(), 128);
+    let packet_count = sigverify::count_packets_in_batches(&batches);
+    let bloom: AtomicBloom<&[u8]> = Bloom::random(1_000_000, 0.0001, 8 << 22).into();
+
+    println!("packet_count {} {}", packet_count, batches.len());
+
+    // verify packets
+    bencher.iter(|| {
+        let _ans = sigverify::dedup_packets(&bloom, &mut batches);
+    })
+}

perf/src/sigverify.rs

@@ -4,6 +4,7 @@
 //! to the GPU.
 //!
 
+use solana_bloom::bloom::AtomicBloom;
 #[cfg(test)]
 use solana_sdk::transaction::Transaction;
 use {
@@ -20,6 +21,7 @@ use {
         hash::Hash, message::MESSAGE_HEADER_LENGTH, pubkey::Pubkey, short_vec::decode_shortu16_len,
         signature::Signature,
     },
+    std::sync::atomic::{AtomicU64, Ordering},
     std::{convert::TryFrom, mem::size_of},
 };
 
@@ -388,6 +390,37 @@ pub fn generate_offsets(
     )
 }
 
+fn dedup_packet(count: &AtomicU64, packet: &mut Packet, bloom: &AtomicBloom<&[u8]>) {
+    // If this packet was already marked as discard, drop it
+    if packet.meta.discard {
+        return;
+    }
+
+    if bloom.contains(&&packet.data[..]) {
+        packet.meta.discard = true;
+        count.fetch_add(1, Ordering::Relaxed);
+        return;
+    }
+    bloom.add(&&packet.data[..]);
+}
+
+pub fn dedup_packets(bloom: &AtomicBloom<&[u8]>, batches: &mut [PacketBatch]) -> u64 {
+    use rayon::prelude::*;
+    let packet_count = count_packets_in_batches(batches);
+    // machine specific random offset to read the u64 from the packet signature
+    let count = AtomicU64::new(0);
+    PAR_THREAD_POOL.install(|| {
+        batches.into_par_iter().for_each(|batch| {
+            batch
+                .packets
+                .par_iter_mut()
+                .for_each(|p| dedup_packet(&count, p, bloom))
+        })
+    });
+    inc_new_counter_debug!("dedup_packets_total", packet_count);
+    count.load(Ordering::Relaxed)
+}
+
 pub fn ed25519_verify_cpu(batches: &mut [PacketBatch], reject_non_vote: bool) {
     use rayon::prelude::*;
     let packet_count = count_packets_in_batches(batches);
@@ -567,11 +600,12 @@ mod tests {
     use {
         super::*,
         crate::{
-            packet::{Packet, PacketBatch},
+            packet::{to_packet_batches, Packet, PacketBatch},
             sigverify::{self, PacketOffsets},
             test_tx::{test_multisig_tx, test_tx, vote_tx},
         },
         bincode::{deserialize, serialize},
+        solana_bloom::bloom::{AtomicBloom, Bloom},
         solana_sdk::{
             instruction::CompiledInstruction,
             message::{Message, MessageHeader},
@@ -1208,4 +1242,31 @@ mod tests {
                 current_offset = current_offset.saturating_add(size_of::<Packet>());
             });
     }
+
+    #[test]
+    fn test_dedup_same() {
+        let tx = test_tx();
+
+        // generate packet vector
+        let mut batches =
+            to_packet_batches(&std::iter::repeat(tx).take(1024).collect::<Vec<_>>(), 128);
+        let packet_count = sigverify::count_packets_in_batches(&batches);
+        let bloom: AtomicBloom<&[u8]> = Bloom::random(1_000_000, 0.0001, 8 << 20).into();
+        let discard = sigverify::dedup_packets(&bloom, &mut batches) as usize;
+        // because dedup uses a threadpool, there maybe up to N threads of txs that go through
+        let n = get_thread_count();
+        assert!(packet_count < discard + n * 2);
+    }
+
+    #[test]
+    fn test_dedup_diff() {
+        // generate packet vector
+        let mut batches = to_packet_batches(&(0..1024).map(|_| test_tx()).collect::<Vec<_>>(), 128);
+
+        let bloom: AtomicBloom<&[u8]> = Bloom::random(1_000_000, 0.0001, 8 << 20).into();
+        let discard = sigverify::dedup_packets(&bloom, &mut batches) as usize;
+        // because dedup uses a threadpool, there maybe up to N threads of txs that go through
+        let n = get_thread_count();
+        assert!(discard < n * 2);
+    }
 }

programs/bpf/Cargo.lock

@@ -1590,11 +1590,11 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.11"
+version = "0.4.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fabed175da42fed1fa0746b0ea71f412aa9d35e76e95e59b192c64b9dc2bf8b"
+checksum = "51b9bbe6c47d51fc3e1a9b945965946b4c44142ab8792c50835a980d362c2710"
 dependencies = [
- "cfg-if 0.1.10",
+ "cfg-if 1.0.0",
 ]
 
 [[package]]
@@ -2229,9 +2229,9 @@ dependencies = [
 
 [[package]]
 name = "rayon"
-version = "1.5.0"
+version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b0d8e0819fadc20c74ea8373106ead0600e3a67ef1fe8da56e39b9ae7275674"
+checksum = "c06aca804d41dbc8ba42dfd964f0d01334eceb64314b9ecf7c5fad5188a06d90"
 dependencies = [
  "autocfg",
  "crossbeam-deque",
@@ -2241,9 +2241,9 @@ dependencies = [
 
 [[package]]
 name = "rayon-core"
-version = "1.9.0"
+version = "1.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ab346ac5921dc62ffa9f89b7a773907511cdfa5490c572ae9be1be33e8afa4a"
+checksum = "d78120e2c850279833f1dd3582f730c4ab53ed95aeaaaa862a2a5c71b1656d8e"
 dependencies = [
  "crossbeam-channel 0.5.0",
  "crossbeam-deque",
@@ -2385,6 +2385,15 @@ dependencies = [
  "semver 0.9.0",
 ]
 
+[[package]]
+name = "rustc_version"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
+dependencies = [
+ "semver 1.0.4",
+]
+
 [[package]]
 name = "rustls"
 version = "0.19.0"
@@ -2506,6 +2515,12 @@ dependencies = [
  "semver-parser 0.10.2",
 ]
 
+[[package]]
+name = "semver"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "568a8e6258aa33c13358f81fd834adb854c6f7c9468520910a9b1e8fac068012"
+
 [[package]]
 name = "semver-parser"
 version = "0.7.0"
@@ -2523,9 +2538,9 @@ dependencies = [
 
 [[package]]
 name = "serde"
-version = "1.0.130"
+version = "1.0.133"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f12d06de37cf59146fbdecab66aa99f9fe4f78722e3607577a5375d66bd0c913"
+checksum = "97565067517b60e2d1ea8b268e59ce036de907ac523ad83a0475da04e818989a"
 dependencies = [
  "serde_derive",
 ]
@@ -2541,9 +2556,9 @@ dependencies = [
 
 [[package]]
 name = "serde_derive"
-version = "1.0.130"
+version = "1.0.133"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7bc1a1ab1961464eae040d96713baa5a724a8152c1222492465b54322ec508b"
+checksum = "ed201699328568d8d08208fdd080e3ff594e6c422e438b6705905da01005d537"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.6",
@@ -2737,6 +2752,23 @@ dependencies = [
  "tokio-stream",
 ]
 
+[[package]]
+name = "solana-bloom"
+version = "1.8.14"
+dependencies = [
+ "bv",
+ "fnv",
+ "log",
+ "rand 0.7.3",
+ "rayon",
+ "rustc_version 0.4.0",
+ "serde",
+ "serde_derive",
+ "solana-frozen-abi 1.8.14",
+ "solana-frozen-abi-macro 1.8.14",
+ "solana-sdk",
+]
+
 [[package]]
 name = "solana-bpf-loader-program"
 version = "1.8.14"
@@ -3232,7 +3264,7 @@ dependencies = [
  "generic-array 0.14.3",
  "log",
  "memmap2",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "sha2",
@@ -3250,7 +3282,7 @@ dependencies = [
  "generic-array 0.14.3",
  "log",
  "memmap2",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "sha2",
@@ -3267,7 +3299,7 @@ checksum = "ceac6e8ad1a784c92ff5f3d6ad68a8d664d389b08055b674c38b2b9abb69e6d4"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.6",
- "rustc_version",
+ "rustc_version 0.2.3",
  "syn 1.0.67",
 ]
 
@@ -3277,7 +3309,7 @@ version = "1.8.14"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.6",
- "rustc_version",
+ "rustc_version 0.2.3",
  "syn 1.0.67",
 ]
 
@@ -3358,6 +3390,7 @@ dependencies = [
  "rand 0.7.3",
  "rayon",
  "serde",
+ "solana-bloom",
  "solana-logger 1.8.14",
  "solana-metrics",
  "solana-rayon-threadlimit",
@@ -3386,7 +3419,7 @@ dependencies = [
  "num-derive 0.3.0",
  "num-traits",
  "rand 0.7.3",
- "rustc_version",
+ "rustc_version 0.2.3",
  "rustversion",
  "serde",
  "serde_bytes",
@@ -3421,7 +3454,7 @@ dependencies = [
  "num-derive 0.3.0",
  "num-traits",
  "rand 0.7.3",
- "rustc_version",
+ "rustc_version 0.2.3",
  "rustversion",
  "serde",
  "serde_bytes",
@@ -3514,9 +3547,10 @@ dependencies = [
  "rand 0.7.3",
  "rayon",
  "regex",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
+ "solana-bloom",
  "solana-compute-budget-program",
  "solana-config-program",
  "solana-ed25519-program",
@@ -3570,7 +3604,7 @@ dependencies = [
  "rand 0.7.3",
  "rand_chacha 0.2.2",
  "rand_core 0.6.2",
- "rustc_version",
+ "rustc_version 0.2.3",
  "rustversion",
  "serde",
  "serde_bytes",
@@ -3627,7 +3661,7 @@ dependencies = [
  "log",
  "num-derive 0.3.0",
  "num-traits",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "solana-config-program",
@@ -3669,7 +3703,7 @@ name = "solana-version"
 version = "1.8.14"
 dependencies = [
  "log",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "solana-frozen-abi 1.8.14",
@@ -3686,7 +3720,7 @@ dependencies = [
  "log",
  "num-derive 0.3.0",
  "num-traits",
- "rustc_version",
+ "rustc_version 0.2.3",
  "serde",
  "serde_derive",
  "solana-frozen-abi 1.8.14",

runtime/Cargo.toml

@@ -38,6 +38,7 @@ serde = { version = "1.0.122", features = ["rc"] }
 serde_derive = "1.0.103"
 solana-config-program = { path = "../programs/config", version = "=1.8.14" }
 solana-compute-budget-program = { path = "../programs/compute-budget", version = "=1.8.14" }
+solana-bloom = { path = "../bloom", version = "=1.8.14" }
 solana-ed25519-program = { path = "../programs/ed25519", version = "=1.8.14" }
 solana-frozen-abi = { path = "../frozen-abi", version = "=1.8.14" }
 solana-frozen-abi-macro = { path = "../frozen-abi/macro", version = "=1.8.14" }

runtime/src/lib.rs

@@ -15,7 +15,6 @@ pub mod bank_forks;
 pub mod bank_utils;
 pub mod block_cost_limits;
 pub mod blockhash_queue;
-pub mod bloom;
 pub mod builtins;
 pub mod commitment;
 pub mod contains;