validator: Add experimental flag to select PoH pinned core

Cargo.lock

@@ -5099,6 +5099,7 @@ dependencies = [
  "chrono",
  "clap",
  "console",
+ "core_affinity",
  "fd-lock",
  "indicatif",
  "libc",

core/src/banking_stage.rs

@@ -4,7 +4,7 @@
 use crate::{
     cluster_info::ClusterInfo,
     poh_recorder::{PohRecorder, PohRecorderError, WorkingBankEntry},
-    poh_service::PohService,
+    poh_service::{self, PohService},
 };
 use crossbeam_channel::{Receiver as CrossbeamReceiver, RecvTimeoutError};
 use itertools::Itertools;
@@ -1093,6 +1093,7 @@ pub fn create_test_recorder(
         &poh_config,
         &exit,
         bank.ticks_per_slot(),
+        poh_service::DEFAULT_PINNED_CPU_CORE,
     );
 
     (exit, poh_recorder, poh_service, entry_receiver)

core/src/poh_service.rs

@@ -19,12 +19,15 @@ pub struct PohService {
 // See benches/poh.rs for some benchmarks that attempt to justify this magic number.
 pub const NUM_HASHES_PER_BATCH: u64 = 1;
 
+pub const DEFAULT_PINNED_CPU_CORE: usize = 0;
+
 impl PohService {
     pub fn new(
         poh_recorder: Arc<Mutex<PohRecorder>>,
         poh_config: &Arc<PohConfig>,
         poh_exit: &Arc<AtomicBool>,
         ticks_per_slot: u64,
+        pinned_cpu_core: usize,
     ) -> Self {
         let poh_exit_ = poh_exit.clone();
         let poh_config = poh_config.clone();
@@ -47,7 +50,7 @@ impl PohService {
                     // Let's dedicate one of the CPU cores to this thread so that it can gain
                     // from cache performance.
                     if let Some(cores) = core_affinity::get_core_ids() {
-                        core_affinity::set_for_current(cores[0]);
+                        core_affinity::set_for_current(cores[pinned_cpu_core]);
                     }
                     Self::tick_producer(
                         poh_recorder,
@@ -215,7 +218,13 @@ mod tests {
                     .unwrap()
             };
 
-            let poh_service = PohService::new(poh_recorder.clone(), &poh_config, &exit, 0);
+            let poh_service = PohService::new(
+                poh_recorder.clone(),
+                &poh_config,
+                &exit,
+                0,
+                DEFAULT_PINNED_CPU_CORE,
+            );
             poh_recorder.lock().unwrap().set_working_bank(working_bank);
 
             // get some events

core/src/validator.rs

@@ -16,7 +16,7 @@ use crate::{
         OptimisticallyConfirmedBank, OptimisticallyConfirmedBankTracker,
     },
     poh_recorder::{PohRecorder, GRACE_TICKS_FACTOR, MAX_GRACE_SLOTS},
-    poh_service::PohService,
+    poh_service::{self, PohService},
     rewards_recorder_service::{RewardsRecorderSender, RewardsRecorderService},
     rpc::JsonRpcConfig,
     rpc_pubsub_service::{PubSubConfig, PubSubService},
@@ -116,6 +116,7 @@ pub struct ValidatorConfig {
     pub send_transaction_retry_ms: u64,
     pub send_transaction_leader_forward_count: u64,
     pub no_poh_speed_test: bool,
+    pub poh_pinned_cpu_core: usize,
 }
 
 impl Default for ValidatorConfig {
@@ -159,6 +160,7 @@ impl Default for ValidatorConfig {
             send_transaction_retry_ms: 2000,
             send_transaction_leader_forward_count: 2,
             no_poh_speed_test: true,
+            poh_pinned_cpu_core: poh_service::DEFAULT_PINNED_CPU_CORE,
         }
     }
 }
@@ -547,6 +549,7 @@ impl Validator {
             &poh_config,
             &exit,
             bank.ticks_per_slot(),
+            config.poh_pinned_cpu_core,
         );
         assert_eq!(
             blockstore.new_shreds_signals.len(),

validator/Cargo.toml

@@ -15,6 +15,7 @@ bincode = "1.3.1"
 clap = "2.33.1"
 chrono = { version = "0.4.11", features = ["serde"] }
 console = "0.11.3"
+core_affinity = "0.5.10"
 fd-lock = "1.1.1"
 indicatif = "0.15.0"
 log = "0.4.11"

validator/src/main.rs

@@ -5,7 +5,7 @@ use clap::{
 use log::*;
 use rand::{seq::SliceRandom, thread_rng, Rng};
 use solana_clap_utils::{
-    input_parsers::{keypair_of, keypairs_of, pubkey_of},
+    input_parsers::{keypair_of, keypairs_of, pubkey_of, value_of},
     input_validators::{
         is_keypair_or_ask_keyword, is_parsable, is_pubkey, is_pubkey_or_keypair, is_slot,
     },
@@ -19,6 +19,7 @@ use solana_core::{
     cluster_info::{ClusterInfo, Node, MINIMUM_VALIDATOR_PORT_RANGE_WIDTH, VALIDATOR_PORT_RANGE},
     contact_info::ContactInfo,
     gossip_service::GossipService,
+    poh_service,
     rpc::JsonRpcConfig,
     rpc_pubsub_service::PubSubConfig,
     validator::{is_snapshot_config_invalid, Validator, ValidatorConfig},
@@ -1395,6 +1396,22 @@ pub fn main() {
                 .takes_value(false)
                 .help("Use the just-in-time compiler instead of the interpreter for BPF."),
         )
+        .arg(
+            Arg::with_name("poh_pinned_cpu_core")
+                .hidden(true)
+                .long("experimental-poh-pinned-cpu-core")
+                .takes_value(true)
+                .value_name("CPU_CORE_INDEX")
+                .validator(|s| {
+                    let core_index = usize::from_str(&s).map_err(|e| e.to_string())?;
+                    let max_index = core_affinity::get_core_ids().map(|cids| cids.len() - 1).unwrap_or(0);
+                    if core_index > max_index {
+                        return Err(format!("core index must be in the range [0, {}]", max_index));
+                    }
+                    Ok(())
+                })
+                .help("EXPERIMENTAL: Specify which CPU core PoH is pinned to")
+        )
         .get_matches();
 
     let identity_keypair = Arc::new(keypair_of(&matches, "identity").unwrap_or_else(Keypair::new));
@@ -1550,6 +1567,8 @@ pub fn main() {
             u64
         ),
         no_poh_speed_test: matches.is_present("no_poh_speed_test"),
+        poh_pinned_cpu_core: value_of(&matches, "poh_pinned_cpu_core")
+            .unwrap_or(poh_service::DEFAULT_PINNED_CPU_CORE),
         ..ValidatorConfig::default()
     };