builds crds filters in parallel

core/Cargo.toml

@@ -24,7 +24,7 @@ crossbeam-channel = "0.4"
 ed25519-dalek = "=1.0.0-pre.4"
 fs_extra = "1.1.0"
 flate2 = "1.0"
-indexmap = "1.5"
+indexmap = { version = "1.5", features = ["rayon"] }
 itertools = "0.9.0"
 jsonrpc-core = "15.0.0"
 jsonrpc-core-client = { version = "15.0.0", features = ["ws"] }

core/benches/crds_gossip_pull.rs

@@ -3,19 +3,19 @@
 extern crate test;
 
 use rand::{thread_rng, Rng};
-use solana_core::crds_gossip_pull::CrdsFilter;
-use solana_sdk::hash::{Hash, HASH_BYTES};
+use rayon::ThreadPoolBuilder;
+use solana_core::cluster_info::MAX_BLOOM_SIZE;
+use solana_core::crds::Crds;
+use solana_core::crds_gossip_pull::{CrdsFilter, CrdsGossipPull};
+use solana_core::crds_value::CrdsValue;
+use solana_sdk::hash::Hash;
 use test::Bencher;
 
 #[bench]
 fn bench_hash_as_u64(bencher: &mut Bencher) {
     let mut rng = thread_rng();
-    let hashes: Vec<_> = (0..1000)
-        .map(|_| {
-            let mut buf = [0u8; HASH_BYTES];
-            rng.fill(&mut buf);
-            Hash::new(&buf)
-        })
+    let hashes: Vec<_> = std::iter::repeat_with(|| Hash::new_rand(&mut rng))
+        .take(1000)
         .collect();
     bencher.iter(|| {
         hashes
@@ -24,3 +24,30 @@ fn bench_hash_as_u64(bencher: &mut Bencher) {
             .collect::<Vec<_>>()
     });
 }
+
+#[bench]
+fn bench_build_crds_filters(bencher: &mut Bencher) {
+    let thread_pool = ThreadPoolBuilder::new().build().unwrap();
+    let mut rng = thread_rng();
+    let mut crds_gossip_pull = CrdsGossipPull::default();
+    let mut crds = Crds::default();
+    for _ in 0..50_000 {
+        crds_gossip_pull
+            .purged_values
+            .push_back((Hash::new_rand(&mut rng), rng.gen()));
+    }
+    let mut num_inserts = 0;
+    for _ in 0..90_000 {
+        if crds
+            .insert(CrdsValue::new_rand(&mut rng), rng.gen())
+            .is_ok()
+        {
+            num_inserts += 1;
+        }
+    }
+    assert_eq!(num_inserts, 90_000);
+    bencher.iter(|| {
+        let filters = crds_gossip_pull.build_crds_filters(&thread_pool, &crds, MAX_BLOOM_SIZE);
+        assert_eq!(filters.len(), 128);
+    });
+}

core/src/cluster_info.rs

@@ -1285,6 +1285,7 @@ impl ClusterInfo {
     // If the network entrypoint hasn't been discovered yet, add it to the crds table
     fn append_entrypoint_to_pulls(
         &self,
+        thread_pool: &ThreadPool,
         pulls: &mut Vec<(Pubkey, CrdsFilter, SocketAddr, CrdsValue)>,
     ) {
         let pull_from_entrypoint = {
@@ -1335,7 +1336,7 @@ impl ClusterInfo {
                     .unwrap_or_else(|| panic!("self_id invalid {}", self.id()));
                 r_gossip
                     .pull
-                    .build_crds_filters(&r_gossip.crds, MAX_BLOOM_SIZE)
+                    .build_crds_filters(thread_pool, &r_gossip.crds, MAX_BLOOM_SIZE)
                     .into_iter()
                     .for_each(|filter| pulls.push((id, filter, gossip, self_info.clone())));
             }
@@ -1384,7 +1385,7 @@ impl ClusterInfo {
 
     fn new_pull_requests(
         &self,
-        _thread_pool: &ThreadPool,
+        thread_pool: &ThreadPool,
         gossip_validators: Option<&HashSet<Pubkey>>,
         stakes: &HashMap<Pubkey, u64>,
     ) -> Vec<(SocketAddr, Protocol)> {
@@ -1393,7 +1394,7 @@ impl ClusterInfo {
             let r_gossip =
                 self.time_gossip_read_lock("new_pull_reqs", &self.stats.new_pull_requests);
             r_gossip
-                .new_pull_request(now, gossip_validators, stakes, MAX_BLOOM_SIZE)
+                .new_pull_request(thread_pool, now, gossip_validators, stakes, MAX_BLOOM_SIZE)
                 .ok()
                 .into_iter()
                 .filter_map(|(peer, filters, me)| {
@@ -1411,7 +1412,7 @@ impl ClusterInfo {
                 .flatten()
                 .collect()
         };
-        self.append_entrypoint_to_pulls(&mut pulls);
+        self.append_entrypoint_to_pulls(thread_pool, &mut pulls);
         self.stats
             .new_pull_requests_count
             .add_relaxed(pulls.len() as u64);
@@ -2881,6 +2882,7 @@ mod tests {
     //when constructed with keypairs
     #[test]
     fn test_gossip_signature_verification() {
+        let thread_pool = ThreadPoolBuilder::new().build().unwrap();
         //create new cluster info, leader, and peer
         let keypair = Keypair::new();
         let peer_keypair = Keypair::new();
@@ -2909,7 +2911,13 @@ mod tests {
             .gossip
             .write()
             .unwrap()
-            .new_pull_request(timestamp(), None, &HashMap::new(), MAX_BLOOM_SIZE)
+            .new_pull_request(
+                &thread_pool,
+                timestamp(),
+                None,
+                &HashMap::new(),
+                MAX_BLOOM_SIZE,
+            )
             .ok()
             .unwrap();
         assert!(val.verify());

core/src/crds_gossip.rs

@@ -10,6 +10,7 @@ use crate::{
     crds_gossip_push::{CrdsGossipPush, CRDS_GOSSIP_NUM_ACTIVE},
     crds_value::{CrdsValue, CrdsValueLabel},
 };
+use rayon::ThreadPool;
 use solana_sdk::pubkey::Pubkey;
 use std::collections::{HashMap, HashSet};
 
@@ -134,12 +135,14 @@ impl CrdsGossip {
     /// generate a random request
     pub fn new_pull_request(
         &self,
+        thread_pool: &ThreadPool,
         now: u64,
         gossip_validators: Option<&HashSet<Pubkey>>,
         stakes: &HashMap<Pubkey, u64>,
         bloom_size: usize,
     ) -> Result<(Pubkey, Vec<CrdsFilter>, CrdsValue), CrdsGossipError> {
         self.pull.new_pull_request(
+            thread_pool,
             &self.crds,
             &self.id,
             self.shred_version,