Optimize packet dedup (#22571)

* Use bloom filter to dedup packets * dedup first * Update bloom/src/bloom.rs Co-authored-by: Trent Nelson <trent.a.b.nelson@gmail.com> * Update core/src/sigverify_stage.rs Co-authored-by: Trent Nelson <trent.a.b.nelson@gmail.com> * Update core/src/sigverify_stage.rs Co-authored-by: Trent Nelson <trent.a.b.nelson@gmail.com> * Update core/src/sigverify_stage.rs Co-authored-by: Trent Nelson <trent.a.b.nelson@gmail.com> * fixup * fixup * fixup Co-authored-by: Trent Nelson <trent.a.b.nelson@gmail.com>
solana-labs · Jan 19, 2022 · d343713 · d343713
1 parent b448472
commit d343713
Show file tree

Hide file tree

Showing 25 changed files with 295 additions and 151 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -11,6 +11,7 @@ members = [
     "banks-interface",
     "banks-server",
     "bucket_map",
+    "bloom",
     "clap-utils",
     "cli-config",
     "cli-output",

diff --git a/banking-bench/src/main.rs b/banking-bench/src/main.rs
@@ -5,7 +5,7 @@ use {
     log::*,
     rand::{thread_rng, Rng},
     rayon::prelude::*,
-    solana_core::{banking_stage::BankingStage, packet_deduper::PacketDeduper},
+    solana_core::banking_stage::BankingStage,
     solana_gossip::cluster_info::{ClusterInfo, Node},
     solana_ledger::{
         blockstore::Blockstore,
@@ -226,7 +226,6 @@ fn main() {
             SocketAddrSpace::Unspecified,
         );
         let cluster_info = Arc::new(cluster_info);
-        let packet_deduper = PacketDeduper::default();
         let banking_stage = BankingStage::new(
             &cluster_info,
             &poh_recorder,
@@ -236,7 +235,6 @@ fn main() {
             None,
             replay_vote_sender,
             Arc::new(RwLock::new(CostModel::default())),
-            packet_deduper.clone(),
         );
         poh_recorder.lock().unwrap().set_bank(&bank);
 
@@ -351,7 +349,6 @@ fn main() {
             // in this chunk, but since we rotate between CHUNKS then
             // we should clear them by the time we come around again to re-use that chunk.
             bank.clear_signatures();
-            packet_deduper.reset();
             total_us += duration_as_us(&now.elapsed());
             debug!(
                 "time: {} us checked: {} sent: {}",

diff --git a/bloom/Cargo.toml b/bloom/Cargo.toml
@@ -0,0 +1,32 @@
+[package]
+name = "solana-bloom"
+version = "1.10.0"
+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 = "2021"
+
+[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.10.0" }
+solana-frozen-abi-macro = { path = "../frozen-abi/macro", version = "=1.10.0" }
+solana-sdk = { path = "../sdk", version = "=1.10.0" }
+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"
diff --git a/runtime/benches/bloom.rs → bloom/benches/bloom.rs b/runtime/benches/bloom.rs → 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,

diff --git a/bloom/build.rs b/bloom/build.rs
@@ -0,0 +1 @@
+../frozen-abi/build.rs
diff --git a/runtime/src/bloom.rs → bloom/src/bloom.rs b/runtime/src/bloom.rs → bloom/src/bloom.rs
@@ -101,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());
@@ -111,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);
             }
         }
@@ -164,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)
     }
 

diff --git a/bloom/src/lib.rs b/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;
diff --git a/core/Cargo.toml b/core/Cargo.toml
@@ -34,6 +34,7 @@ retain_mut = "0.1.5"
 serde = "1.0.133"
 serde_derive = "1.0.103"
 solana-address-lookup-table-program = { path = "../programs/address-lookup-table", version = "=1.10.0" }
+solana-bloom = { path = "../bloom", version = "=1.10.0" }
 solana-accountsdb-plugin-manager = { path = "../accountsdb-plugin-manager", version = "=1.10.0" }
 solana-client = { path = "../client", version = "=1.10.0" }
 solana-entry = { path = "../entry", version = "=1.10.0" }

diff --git a/core/benches/banking_stage.rs b/core/benches/banking_stage.rs
@@ -10,7 +10,6 @@ use {
     rayon::prelude::*,
     solana_core::{
         banking_stage::{BankingStage, BankingStageStats},
-        packet_deduper::PacketDeduper,
         qos_service::QosService,
     },
     solana_entry::entry::{next_hash, Entry},
@@ -222,7 +221,6 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
         );
         let cluster_info = Arc::new(cluster_info);
         let (s, _r) = unbounded();
-        let packet_deduper = PacketDeduper::default();
         let _banking_stage = BankingStage::new(
             &cluster_info,
             &poh_recorder,
@@ -232,7 +230,6 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
             None,
             s,
             Arc::new(RwLock::new(CostModel::default())),
-            packet_deduper.clone(),
         );
         poh_recorder.lock().unwrap().set_bank(&bank);
 
@@ -267,7 +264,6 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
             // in this chunk, but since we rotate between CHUNKS then
             // we should clear them by the time we come around again to re-use that chunk.
             bank.clear_signatures();
-            packet_deduper.reset();
             trace!(
                 "time: {} checked: {} sent: {}",
                 duration_as_us(&now.elapsed()),