v1.13: removes lazy-static thread-pool from sigverify-shreds (backpor…

…t of solana-labs#30787)

removes lazy-static thread-pool from sigverify-shreds solana-labs#30787

Instead the thread-pool is passed explicitly from higher in the call
stack so that
solana-labs#30786
can use the same thread-pool for shred deduplication.

(cherry picked from commit c6e7aaf)

core/src/sigverify_shreds.rs

@@ -6,20 +6,22 @@ use {
         sigverify_stage::{SigVerifier, SigVerifyServiceError},
     },
     crossbeam_channel::Sender,
+    rayon::{ThreadPool, ThreadPoolBuilder},
     solana_ledger::{
         leader_schedule_cache::LeaderScheduleCache, shred::Shred,
         sigverify_shreds::verify_shreds_gpu,
     },
     solana_perf::{self, packet::PacketBatch, recycler_cache::RecyclerCache},
+    solana_rayon_threadlimit::get_thread_count,
     solana_runtime::bank_forks::BankForks,
     std::{
         collections::{HashMap, HashSet},
         sync::{Arc, RwLock},
     },
 };
 
-#[derive(Clone)]
 pub struct ShredSigVerifier {
+    thread_pool: ThreadPool,
     bank_forks: Arc<RwLock<BankForks>>,
     leader_schedule_cache: Arc<LeaderScheduleCache>,
     recycler_cache: RecyclerCache,
@@ -32,8 +34,14 @@ impl ShredSigVerifier {
         leader_schedule_cache: Arc<LeaderScheduleCache>,
         packet_sender: Sender<Vec<PacketBatch>>,
     ) -> Self {
+        let thread_pool = ThreadPoolBuilder::new()
+            .num_threads(get_thread_count())
+            .thread_name(|i| format!("solSvrfyShred{i:02}"))
+            .build()
+            .unwrap();
         sigverify::init();
         Self {
+            thread_pool,
             bank_forks,
             leader_schedule_cache,
             recycler_cache: RecyclerCache::warmed(),
@@ -77,7 +85,12 @@ impl SigVerifier for ShredSigVerifier {
             .collect();
         leader_slots.insert(std::u64::MAX, [0u8; 32]);
 
-        let r = verify_shreds_gpu(&batches, &leader_slots, &self.recycler_cache);
+        let r = verify_shreds_gpu(
+            &self.thread_pool,
+            &batches,
+            &leader_slots,
+            &self.recycler_cache,
+        );
         solana_perf::sigverify::mark_disabled(&mut batches, &r);
         batches
     }

core/src/sigverify_stage.rs

@@ -234,7 +234,7 @@ impl SigVerifier for DisabledSigVerifier {
 
 impl SigVerifyStage {
     #[allow(clippy::new_ret_no_self)]
-    pub fn new<T: SigVerifier + 'static + Send + Clone>(
+    pub fn new<T: SigVerifier + 'static + Send>(
         packet_receiver: find_packet_sender_stake_stage::FindPacketSenderStakeReceiver,
         verifier: T,
         name: &'static str,
@@ -405,7 +405,7 @@ impl SigVerifyStage {
         Ok(())
     }
 
-    fn verifier_service<T: SigVerifier + 'static + Send + Clone>(
+    fn verifier_service<T: SigVerifier + 'static + Send>(
         packet_receiver: find_packet_sender_stake_stage::FindPacketSenderStakeReceiver,
         mut verifier: T,
         name: &'static str,
@@ -449,7 +449,7 @@ impl SigVerifyStage {
             .unwrap()
     }
 
-    fn verifier_services<T: SigVerifier + 'static + Send + Clone>(
+    fn verifier_services<T: SigVerifier + 'static + Send>(
         packet_receiver: find_packet_sender_stake_stage::FindPacketSenderStakeReceiver,
         verifier: T,
         name: &'static str,

ledger/Cargo.toml

@@ -19,7 +19,6 @@ crossbeam-channel = "0.5"
 fs_extra = "1.2.0"
 futures = "0.3.21"
 itertools = "0.10.3"
-lazy_static = "1.4.0"
 libc = "0.2.120"
 log = { version = "0.4.14" }
 lru = "0.7.5"

ledger/benches/sigverify_shreds.rs

@@ -2,6 +2,7 @@
 
 extern crate test;
 use {
+    rayon::ThreadPoolBuilder,
     solana_ledger::{
         shred::{Shred, SIZE_OF_DATA_SHRED_PAYLOAD},
         sigverify_shreds::{sign_shreds_cpu, sign_shreds_gpu, sign_shreds_gpu_pinned_keypair},
@@ -10,6 +11,7 @@ use {
         packet::{Packet, PacketBatch},
         recycler_cache::RecyclerCache,
     },
+    solana_rayon_threadlimit::get_thread_count,
     solana_sdk::signature::Keypair,
     std::sync::Arc,
     test::Bencher,
@@ -19,6 +21,10 @@ const NUM_PACKETS: usize = 256;
 const NUM_BATCHES: usize = 1;
 #[bench]
 fn bench_sigverify_shreds_sign_gpu(bencher: &mut Bencher) {
+    let thread_pool = ThreadPoolBuilder::new()
+        .num_threads(get_thread_count())
+        .build()
+        .unwrap();
     let recycler_cache = RecyclerCache::default();
 
     let mut packet_batch = PacketBatch::new_pinned_with_capacity(NUM_PACKETS);
@@ -44,15 +50,31 @@ fn bench_sigverify_shreds_sign_gpu(bencher: &mut Bencher) {
     let pinned_keypair = Some(Arc::new(pinned_keypair));
     //warmup
     for _ in 0..100 {
-        sign_shreds_gpu(&keypair, &pinned_keypair, &mut batches, &recycler_cache);
+        sign_shreds_gpu(
+            &thread_pool,
+            &keypair,
+            &pinned_keypair,
+            &mut batches,
+            &recycler_cache,
+        );
     }
     bencher.iter(|| {
-        sign_shreds_gpu(&keypair, &pinned_keypair, &mut batches, &recycler_cache);
+        sign_shreds_gpu(
+            &thread_pool,
+            &keypair,
+            &pinned_keypair,
+            &mut batches,
+            &recycler_cache,
+        );
     })
 }
 
 #[bench]
 fn bench_sigverify_shreds_sign_cpu(bencher: &mut Bencher) {
+    let thread_pool = ThreadPoolBuilder::new()
+        .num_threads(get_thread_count())
+        .build()
+        .unwrap();
     let mut packet_batch = PacketBatch::default();
     let slot = 0xdead_c0de;
     packet_batch.resize(NUM_PACKETS, Packet::default());
@@ -73,6 +95,6 @@ fn bench_sigverify_shreds_sign_cpu(bencher: &mut Bencher) {
     let mut batches = vec![packet_batch; NUM_BATCHES];
     let keypair = Keypair::new();
     bencher.iter(|| {
-        sign_shreds_cpu(&keypair, &mut batches);
+        sign_shreds_cpu(&thread_pool, &keypair, &mut batches);
     })
 }

ledger/src/lib.rs

@@ -35,8 +35,5 @@ extern crate solana_metrics;
 #[macro_use]
 extern crate log;
 
-#[macro_use]
-extern crate lazy_static;
-
 #[macro_use]
 extern crate solana_frozen_abi_macro;