Add CLI arg to control size of rayon global thread pool

The arg is hidden for now, and default behavior of one thread in the
pool per system thread remains

core/src/validator.rs

@@ -282,6 +282,7 @@ pub struct ValidatorConfig {
     pub wen_restart_coordinator: Option<Pubkey>,
     pub unified_scheduler_handler_threads: Option<usize>,
     pub ip_echo_server_threads: NonZeroUsize,
+    pub rayon_global_threads: NonZeroUsize,
     pub replay_forks_threads: NonZeroUsize,
     pub replay_transactions_threads: NonZeroUsize,
     pub tvu_shred_sigverify_threads: NonZeroUsize,
@@ -354,6 +355,7 @@ impl Default for ValidatorConfig {
             wen_restart_coordinator: None,
             unified_scheduler_handler_threads: None,
             ip_echo_server_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
+            rayon_global_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"),
@@ -364,15 +366,18 @@ impl Default for ValidatorConfig {
 
 impl ValidatorConfig {
     pub fn default_for_test() -> Self {
+        let max_thread_count =
+            NonZeroUsize::new(get_max_thread_count()).expect("thread count is non-zero");
+
         Self {
             enforce_ulimit_nofile: false,
             accounts_db_config: Some(ACCOUNTS_DB_CONFIG_FOR_TESTING),
             rpc_config: JsonRpcConfig::default_for_test(),
             block_production_method: BlockProductionMethod::default(),
             enable_block_production_forwarding: true, // enable forwarding by default for tests
+            rayon_global_threads: max_thread_count,
             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"),
+            replay_transactions_threads: max_thread_count,
             tvu_shred_sigverify_threads: NonZeroUsize::new(get_thread_count())
                 .expect("thread count is non-zero"),
             ..Self::default()
@@ -533,6 +538,18 @@ impl Validator {
     ) -> Result<Self> {
         let start_time = Instant::now();
 
+        // Initialize the global rayon pool first to ensure the value in config
+        // is honored. Otherwise, some code accessing the global pool could
+        // cause it to get initialized with Rayon's default (not ours)
+        if rayon::ThreadPoolBuilder::new()
+            .thread_name(|i| format!("solRayonGlob{i:02}"))
+            .num_threads(config.rayon_global_threads.get())
+            .build_global()
+            .is_err()
+        {
+            warn!("Rayon global thread pool already initialized");
+        }
+
         let id = identity_keypair.pubkey();
         assert_eq!(&id, node.info.pubkey());
 
@@ -582,14 +599,6 @@ impl Validator {
             info!("entrypoint: {:?}", cluster_entrypoint);
         }
 
-        if rayon::ThreadPoolBuilder::new()
-            .thread_name(|i| format!("solRayonGlob{i:02}"))
-            .build_global()
-            .is_err()
-        {
-            warn!("Rayon global thread pool already initialized");
-        }
-
         if solana_perf::perf_libs::api().is_some() {
             info!("Initializing sigverify, this could take a while...");
         } else {

local-cluster/src/validator_configs.rs

@@ -69,6 +69,7 @@ pub fn safe_clone_config(config: &ValidatorConfig) -> ValidatorConfig {
         wen_restart_coordinator: config.wen_restart_coordinator,
         unified_scheduler_handler_threads: config.unified_scheduler_handler_threads,
         ip_echo_server_threads: config.ip_echo_server_threads,
+        rayon_global_threads: config.rayon_global_threads,
         replay_forks_threads: config.replay_forks_threads,
         replay_transactions_threads: config.replay_transactions_threads,
         tvu_shred_sigverify_threads: config.tvu_shred_sigverify_threads,

validator/src/cli/thread_args.rs

@@ -15,6 +15,7 @@ pub struct DefaultThreadArgs {
     pub accounts_db_hash_threads: String,
     pub accounts_index_flush_threads: String,
     pub ip_echo_server_threads: String,
+    pub rayon_global_threads: String,
     pub replay_forks_threads: String,
     pub replay_transactions_threads: String,
     pub tvu_receive_threads: String,
@@ -31,6 +32,7 @@ impl Default for DefaultThreadArgs {
             accounts_index_flush_threads: AccountsIndexFlushThreadsArg::bounded_default()
                 .to_string(),
             ip_echo_server_threads: IpEchoServerThreadsArg::bounded_default().to_string(),
+            rayon_global_threads: RayonGlobalThreadsArg::bounded_default().to_string(),
             replay_forks_threads: ReplayForksThreadsArg::bounded_default().to_string(),
             replay_transactions_threads: ReplayTransactionsThreadsArg::bounded_default()
                 .to_string(),
@@ -47,6 +49,7 @@ pub fn thread_args<'a>(defaults: &DefaultThreadArgs) -> Vec<Arg<'_, 'a>> {
         new_thread_arg::<AccountsDbHashThreadsArg>(&defaults.accounts_db_hash_threads),
         new_thread_arg::<AccountsIndexFlushThreadsArg>(&defaults.accounts_index_flush_threads),
         new_thread_arg::<IpEchoServerThreadsArg>(&defaults.ip_echo_server_threads),
+        new_thread_arg::<RayonGlobalThreadsArg>(&defaults.rayon_global_threads),
         new_thread_arg::<ReplayForksThreadsArg>(&defaults.replay_forks_threads),
         new_thread_arg::<ReplayTransactionsThreadsArg>(&defaults.replay_transactions_threads),
         new_thread_arg::<TvuReceiveThreadsArg>(&defaults.tvu_receive_threads),
@@ -71,6 +74,7 @@ pub struct NumThreadConfig {
     pub accounts_db_hash_threads: NonZeroUsize,
     pub accounts_index_flush_threads: NonZeroUsize,
     pub ip_echo_server_threads: NonZeroUsize,
+    pub rayon_global_threads: NonZeroUsize,
     pub replay_forks_threads: NonZeroUsize,
     pub replay_transactions_threads: NonZeroUsize,
     pub tvu_receive_threads: NonZeroUsize,
@@ -104,6 +108,7 @@ pub fn parse_num_threads_args(matches: &ArgMatches) -> NumThreadConfig {
             IpEchoServerThreadsArg::NAME,
             NonZeroUsize
         ),
+        rayon_global_threads: value_t_or_exit!(matches, RayonGlobalThreadsArg::NAME, NonZeroUsize),
         replay_forks_threads: if matches.is_present("replay_slots_concurrently") {
             NonZeroUsize::new(4).expect("4 is non-zero")
         } else {
@@ -213,6 +218,17 @@ impl ThreadArg for IpEchoServerThreadsArg {
     }
 }
 
+struct RayonGlobalThreadsArg;
+impl ThreadArg for RayonGlobalThreadsArg {
+    const NAME: &'static str = "rayon_global_threads";
+    const LONG_NAME: &'static str = "rayon-global-threads";
+    const HELP: &'static str = "Number of threads to use for the global rayon thread pool";
+
+    fn default() -> usize {
+        get_max_thread_count()
+    }
+}
+
 struct ReplayForksThreadsArg;
 impl ThreadArg for ReplayForksThreadsArg {
     const NAME: &'static str = "replay_forks_threads";

validator/src/main.rs

@@ -915,6 +915,7 @@ pub fn main() {
         accounts_db_hash_threads,
         accounts_index_flush_threads,
         ip_echo_server_threads,
+        rayon_global_threads,
         replay_forks_threads,
         replay_transactions_threads,
         tvu_receive_threads,
@@ -1542,6 +1543,7 @@ pub fn main() {
             UseSnapshotArchivesAtStartup
         ),
         ip_echo_server_threads,
+        rayon_global_threads,
         replay_forks_threads,
         replay_transactions_threads,
         tvu_shred_sigverify_threads: tvu_sigverify_threads,