Add CLI command for measuring TpuClient slot estimation

Cargo.toml

@@ -117,6 +117,7 @@ members = [
     "tokens",
     "tps-client",
     "tpu-client",
+    "tpu-client-test",
     "transaction-dos",
     "transaction-metrics-tracker",
     "transaction-status",

core/src/validator.rs

@@ -1658,6 +1658,10 @@ impl Validator {
             .join()
             .expect("poh_timing_report_service");
     }
+
+    pub fn poh_recorder(&self) -> Arc<RwLock<PohRecorder>> {
+        self.poh_recorder.clone()
+    }
 }
 
 fn active_vote_account_exists_in_bank(bank: &Bank, vote_account: &Pubkey) -> bool {

poh/src/poh_recorder.rs

@@ -366,6 +366,10 @@ impl PohRecorder {
             )
     }
 
+    pub fn current_slot(&self) -> Slot {
+        self.slot_for_tick_height(self.tick_height)
+    }
+
     // Return the slot for a given tick height
     fn slot_for_tick_height(&self, tick_height: u64) -> Slot {
         // We need to subtract by one here because, assuming ticks per slot is 64,

test-validator/Cargo.toml

@@ -33,6 +33,7 @@ solana-runtime = { workspace = true }
 solana-sdk = { workspace = true }
 solana-streamer = { workspace = true }
 solana-tpu-client = { workspace = true }
+solana-poh = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 
 [package.metadata.docs.rs]

test-validator/src/lib.rs

@@ -30,6 +30,7 @@ use {
         create_new_tmp_ledger,
     },
     solana_net_utils::PortRange,
+    solana_poh::poh_recorder::PohRecorder,
     solana_rpc::{rpc::JsonRpcConfig, rpc_pubsub_service::PubSubConfig},
     solana_rpc_client::{nonblocking, rpc_client::RpcClient},
     solana_runtime::{
@@ -694,6 +695,14 @@ impl TestValidator {
             .expect("validator start failed")
     }
 
+    pub fn current_slot(&self) -> u64 {
+        self.poh_recorder().read().unwrap().current_slot()
+    }
+
+    pub fn poh_recorder(&self) -> Arc<RwLock<PohRecorder>> {
+        self.validator.as_ref().unwrap().poh_recorder()
+    }
+
     /// Create a test validator using udp for TPU.
     pub fn with_no_fees_udp(
         mint_address: Pubkey,

tpu-client-test/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "solana-tpu-client-test"
+description = "Solana TPU Client Test"
+publish = false
+version = { workspace = true }
+authors = { workspace = true }
+repository = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+solana-client = { workspace = true }
+solana-sdk = { workspace = true }
+solana-logger = { workspace = true }
+solana-test-validator = { workspace = true }
+clap = { version = "3.1.5", features = ["cargo", "derive"] }
+tokio = { workspace = true, features = ["full"] }
+log = { workspace = true }

tpu-client-test/src/main.rs

@@ -0,0 +1,45 @@
+use clap::{crate_description, crate_name, crate_version, Parser, Subcommand};
+
+#[derive(Parser, Debug)]
+#[clap(name = crate_name!(),
+    version = crate_version!(),
+    about = crate_description!(),
+    rename_all = "kebab-case"
+)]
+struct TpuClientTestArgs {
+    #[clap(subcommand)]
+    pub mode: Mode,
+}
+
+#[derive(Subcommand, Debug)]
+enum Mode {
+    /// Test the TpuClient's slot estimation accuracy.
+    SlotEstimationAccuracy {
+        #[clap(
+            short,
+            long,
+            default_value_t = 0.15,
+            help = "Accuracy threshold for slot estimation"
+        )]
+        accuracy_threshold: f64,
+        #[clap(short, long, default_value_t = 100, help = "Number of samples to take")]
+        num_samples: usize,
+    },
+}
+
+mod modes;
+use modes::*;
+
+#[tokio::main]
+async fn main() {
+    solana_logger::setup_with("solana_tpu_client_test=info");
+
+    let args = TpuClientTestArgs::parse();
+
+    match args.mode {
+        Mode::SlotEstimationAccuracy {
+            accuracy_threshold,
+            num_samples,
+        } => slot_estimation_accuracy::run(accuracy_threshold, num_samples).await,
+    }
+}

tpu-client-test/src/modes/mod.rs

@@ -0,0 +1 @@
+pub mod slot_estimation_accuracy;

tpu-client-test/src/modes/slot_estimation_accuracy.rs

@@ -0,0 +1,87 @@
+use {
+    log::info,
+    solana_client::{connection_cache::Protocol, nonblocking::tpu_client::LeaderTpuService},
+    solana_sdk::clock::DEFAULT_MS_PER_SLOT,
+    solana_test_validator::{TestValidator, TestValidatorGenesis},
+    std::{
+        sync::{
+            atomic::{AtomicBool, Ordering},
+            Arc,
+        },
+        time::Duration,
+    },
+    tokio::time::sleep,
+};
+
+struct SlotEstimationAccuracy {
+    leader_tpu_service: LeaderTpuService,
+    validator: TestValidator,
+    exit: Arc<AtomicBool>,
+}
+
+impl SlotEstimationAccuracy {
+    async fn new() -> Self {
+        let validator = TestValidatorGenesis::default().start_async().await.0;
+        let rpc_client = Arc::new(validator.get_async_rpc_client());
+        let exit = Arc::new(AtomicBool::new(false));
+        let leader_tpu_service = LeaderTpuService::new(
+            rpc_client,
+            &validator.rpc_pubsub_url(),
+            Protocol::QUIC,
+            exit.clone(),
+        )
+        .await
+        .unwrap();
+
+        Self {
+            leader_tpu_service,
+            validator,
+            exit,
+        }
+    }
+}
+
+pub(crate) async fn run(accuracy_threshold: f64, num_samples: usize) {
+    info!("bootstrapping test validator");
+
+    let SlotEstimationAccuracy {
+        mut leader_tpu_service,
+        validator,
+        exit,
+    } = SlotEstimationAccuracy::new().await;
+
+    let sleep_time = Duration::from_millis(DEFAULT_MS_PER_SLOT);
+    let mut result_pairs = vec![];
+
+    let mut actual = validator.current_slot();
+    while (actual as usize) < num_samples {
+        actual = validator.current_slot();
+        let estimated = leader_tpu_service.estimated_current_slot();
+        result_pairs.push((estimated, actual));
+        info!(
+            "estimated: {}, actual: {} {}",
+            estimated,
+            actual,
+            if estimated == actual { "✅" } else { "❌" }
+        );
+
+        sleep(sleep_time).await;
+    }
+
+    let failure_rate =
+        result_pairs.iter().filter(|(a, b)| a != b).count() as f64 / result_pairs.len() as f64;
+
+    info!(
+        "failure rate: {failure_rate} <= {accuracy_threshold} {}",
+        if failure_rate <= accuracy_threshold {
+            "✅"
+        } else {
+            "❌"
+        }
+    );
+
+    info!("cleaning up and exiting...");
+
+    exit.store(true, Ordering::Relaxed);
+    leader_tpu_service.join().await;
+}