Add hidden CLI arg to control number of shred sigverify threads

core/src/tvu.rs

@@ -92,6 +92,7 @@ pub struct TvuConfig {
     pub wait_for_vote_to_start_leader: bool,
     pub replay_forks_threads: NonZeroUsize,
     pub replay_transactions_threads: NonZeroUsize,
+    pub shred_sigverify_threads: NonZeroUsize,
 }
 
 impl Default for TvuConfig {
@@ -104,6 +105,7 @@ impl Default for TvuConfig {
             wait_for_vote_to_start_leader: false,
             replay_forks_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             replay_transactions_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
+            shred_sigverify_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
         }
     }
 }
@@ -198,6 +200,7 @@ impl Tvu {
             fetch_receiver,
             retransmit_sender.clone(),
             verified_sender,
+            tvu_config.shred_sigverify_threads,
         );
 
         let retransmit_stage = RetransmitStage::new(

core/src/validator.rs

@@ -77,7 +77,7 @@ use {
         poh_recorder::PohRecorder,
         poh_service::{self, PohService},
     },
-    solana_rayon_threadlimit::get_max_thread_count,
+    solana_rayon_threadlimit::{get_max_thread_count, get_thread_count},
     solana_rpc::{
         max_slots::MaxSlots,
         optimistically_confirmed_bank_tracker::{
@@ -281,6 +281,7 @@ pub struct ValidatorConfig {
     pub ip_echo_server_threads: NonZeroUsize,
     pub replay_forks_threads: NonZeroUsize,
     pub replay_transactions_threads: NonZeroUsize,
+    pub tvu_shred_sigverify_threads: NonZeroUsize,
     pub delay_leader_block_for_pending_fork: bool,
 }
 
@@ -353,6 +354,7 @@ impl Default for ValidatorConfig {
             ip_echo_server_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             replay_forks_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             replay_transactions_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
+            tvu_shred_sigverify_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             delay_leader_block_for_pending_fork: false,
         }
     }
@@ -368,6 +370,8 @@ impl ValidatorConfig {
             replay_forks_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
             replay_transactions_threads: NonZeroUsize::new(get_max_thread_count())
                 .expect("thread count is non-zero"),
+            tvu_shred_sigverify_threads: NonZeroUsize::new(get_thread_count())
+                .expect("thread count is non-zero"),
             ..Self::default()
         }
     }
@@ -1362,6 +1366,7 @@ impl Validator {
                 wait_for_vote_to_start_leader,
                 replay_forks_threads: config.replay_forks_threads,
                 replay_transactions_threads: config.replay_transactions_threads,
+                shred_sigverify_threads: config.tvu_shred_sigverify_threads,
             },
             &max_slots,
             block_metadata_notifier,

local-cluster/src/validator_configs.rs

@@ -72,6 +72,7 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
         ip_echo_server_threads: config.ip_echo_server_threads,
         replay_forks_threads: config.replay_forks_threads,
         replay_transactions_threads: config.replay_transactions_threads,
+        tvu_shred_sigverify_threads: config.tvu_shred_sigverify_threads,
         delay_leader_block_for_pending_fork: config.delay_leader_block_for_pending_fork,
     }
 }

turbine/src/sigverify_shreds.rs

@@ -13,7 +13,6 @@ use {
         sigverify_shreds::{verify_shreds_gpu, LruCache},
     },
     solana_perf::{self, deduper::Deduper, packet::PacketBatch, recycler_cache::RecyclerCache},
-    solana_rayon_threadlimit::get_thread_count,
     solana_runtime::{
         bank::{Bank, MAX_LEADER_SCHEDULE_STAKES},
         bank_forks::BankForks,
@@ -26,6 +25,7 @@ use {
     static_assertions::const_assert_eq,
     std::{
         collections::HashMap,
+        num::NonZeroUsize,
         sync::{
             atomic::{AtomicUsize, Ordering},
             Arc, RwLock,
@@ -66,6 +66,7 @@ pub fn spawn_shred_sigverify(
     shred_fetch_receiver: Receiver<PacketBatch>,
     retransmit_sender: Sender<Vec</*shred:*/ Vec<u8>>>,
     verified_sender: Sender<Vec<PacketBatch>>,
+    num_sigverify_threads: NonZeroUsize,
 ) -> JoinHandle<()> {
     let recycler_cache = RecyclerCache::warmed();
     let mut stats = ShredSigVerifyStats::new(Instant::now());
@@ -75,10 +76,10 @@ pub fn spawn_shred_sigverify(
         CLUSTER_NODES_CACHE_TTL,
     );
     let thread_pool = ThreadPoolBuilder::new()
-        .num_threads(get_thread_count())
+        .num_threads(num_sigverify_threads.get())
         .thread_name(|i| format!("solSvrfyShred{i:02}"))
         .build()
-        .unwrap();
+        .expect("new rayon threadpool");
     let run_shred_sigverify = move || {
         let mut rng = rand::thread_rng();
         let mut deduper = Deduper::<2, [u8]>::new(&mut rng, DEDUPER_NUM_BITS);

validator/src/cli/thread_args.rs

@@ -3,7 +3,7 @@
 use {
     clap::{value_t_or_exit, Arg, ArgMatches},
     solana_clap_utils::{hidden_unless_forced, input_validators::is_within_range},
-    solana_rayon_threadlimit::get_max_thread_count,
+    solana_rayon_threadlimit::{get_max_thread_count, get_thread_count},
     std::{num::NonZeroUsize, ops::RangeInclusive},
 };
 
@@ -13,6 +13,7 @@ pub struct DefaultThreadArgs {
     pub replay_forks_threads: String,
     pub replay_transactions_threads: String,
     pub tvu_receive_threads: String,
+    pub tvu_sigverify_threads: String,
 }
 
 impl Default for DefaultThreadArgs {
@@ -23,6 +24,7 @@ impl Default for DefaultThreadArgs {
             replay_transactions_threads: ReplayTransactionsThreadsArg::bounded_default()
                 .to_string(),
             tvu_receive_threads: TvuReceiveThreadsArg::bounded_default().to_string(),
+            tvu_sigverify_threads: TvuShredSigverifyThreadsArg::bounded_default().to_string(),
         }
     }
 }
@@ -33,6 +35,7 @@ pub fn thread_args<'a>(defaults: &DefaultThreadArgs) -> Vec<Arg<'_, 'a>> {
         new_thread_arg::<ReplayForksThreadsArg>(&defaults.replay_forks_threads),
         new_thread_arg::<ReplayTransactionsThreadsArg>(&defaults.replay_transactions_threads),
         new_thread_arg::<TvuReceiveThreadsArg>(&defaults.tvu_receive_threads),
+        new_thread_arg::<TvuShredSigverifyThreadsArg>(&defaults.tvu_sigverify_threads),
     ]
 }
 
@@ -52,6 +55,7 @@ pub struct NumThreadConfig {
     pub replay_forks_threads: NonZeroUsize,
     pub replay_transactions_threads: NonZeroUsize,
     pub tvu_receive_threads: NonZeroUsize,
+    pub tvu_sigverify_threads: NonZeroUsize,
 }
 
 pub fn parse_num_threads_args(matches: &ArgMatches) -> NumThreadConfig {
@@ -72,6 +76,11 @@ pub fn parse_num_threads_args(matches: &ArgMatches) -> NumThreadConfig {
             NonZeroUsize
         ),
         tvu_receive_threads: value_t_or_exit!(matches, TvuReceiveThreadsArg::NAME, NonZeroUsize),
+        tvu_sigverify_threads: value_t_or_exit!(
+            matches,
+            TvuShredSigverifyThreadsArg::NAME,
+            NonZeroUsize
+        ),
     }
 }
 
@@ -163,3 +172,15 @@ impl ThreadArg for TvuReceiveThreadsArg {
         solana_gossip::cluster_info::MINIMUM_NUM_TVU_SOCKETS.get()
     }
 }
+
+struct TvuShredSigverifyThreadsArg;
+impl ThreadArg for TvuShredSigverifyThreadsArg {
+    const NAME: &'static str = "tvu_shred_sigverify_threads";
+    const LONG_NAME: &'static str = "tvu-shred-sigverify-threads";
+    const HELP: &'static str =
+        "Number of threads to use for performing signature verification of received shreds";
+
+    fn default() -> usize {
+        get_thread_count()
+    }
+}

validator/src/main.rs

@@ -1390,6 +1390,7 @@ pub fn main() {
         replay_forks_threads,
         replay_transactions_threads,
         tvu_receive_threads,
+        tvu_sigverify_threads,
     } = cli::thread_args::parse_num_threads_args(&matches);
 
     let mut validator_config = ValidatorConfig {
@@ -1533,6 +1534,7 @@ pub fn main() {
         ip_echo_server_threads,
         replay_forks_threads,
         replay_transactions_threads,
+        tvu_shred_sigverify_threads: tvu_sigverify_threads,
         delay_leader_block_for_pending_fork: matches
             .is_present("delay_leader_block_for_pending_fork"),
         wen_restart_proto_path: value_t!(matches, "wen_restart", PathBuf).ok(),