all

2023-05-11 18:32:06 +05:30 · 2023-05-11 18:32:06 +05:30 · 8d83311428
parent 3eac7a391f
commit 8d83311428
11 changed files with 265 additions and 62 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -691,6 +691,12 @@ version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "13418e745008f7349ec7e449155f419a61b92b58a99cc3616942b926825ec76b"

+[[package]]
+name = "convert_case"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
+
 [[package]]
 name = "core-foundation-sys"
 version = "0.8.4"
@ -916,6 +922,19 @@ dependencies = [
 "syn 1.0.109",
 ]

+[[package]]
+name = "derive_more"
+version = "0.99.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4fb810d30a7c1953f91334de7244731fc3f3c10d7fe163338a35b9f640960321"
+dependencies = [
+ "convert_case",
+ "proc-macro2",
+ "quote",
+ "rustc_version 0.4.0",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "digest"
 version = "0.9.0"
@ -1018,6 +1037,19 @@ dependencies = [
 "syn 2.0.15",
 ]

+[[package]]
+name = "env_logger"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44533bbbb3bb3c1fa17d9f2e4e38bbbaf8396ba82193c4cb1b6445d711445d36"
+dependencies = [
+ "atty",
+ "humantime 1.3.0",
+ "log",
+ "regex",
+ "termcolor",
+]
+
 [[package]]
 name = "env_logger"
 version = "0.9.3"
@ -1025,7 +1057,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a12e6657c4c97ebab115a42dcee77225f7f482cdd841cf7088c657a42e9e00e7"
 dependencies = [
 "atty",
- "humantime",
+ "humantime 2.1.0",
 "log",
 "regex",
 "termcolor",
@ -1370,6 +1402,15 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"

+[[package]]
+name = "humantime"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df004cfca50ef23c36850aaaa59ad52cc70d0e90243c3c7737a4dd32dc7a3c4f"
+dependencies = [
+ "quick-error",
+]
+
 [[package]]
 name = "humantime"
 version = "2.1.0"
@ -1970,6 +2011,16 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de"

+[[package]]
+name = "pretty_env_logger"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "926d36b9553851b8b0005f1275891b392ee4d2d833852c417ed025477350fb9d"
+dependencies = [
+ "env_logger 0.7.1",
+ "log",
+]
+
 [[package]]
 name = "proc-macro-crate"
 version = "0.1.5"
@ -2007,6 +2058,12 @@ dependencies = [
 "percent-encoding",
 ]

+[[package]]
+name = "quick-error"
+version = "1.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
+
 [[package]]
 name = "quote"
 version = "1.0.26"
@ -2633,7 +2690,7 @@ version = "1.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "170714ca3612e4df75f57c2c14c8ab74654b3b66f668986aeed456cedcf24446"
 dependencies = [
- "env_logger",
+ "env_logger 0.9.3",
 "lazy_static",
 "log",
 ]
@ -2806,9 +2863,11 @@ dependencies = [
 "clap",
 "const_env",
 "dashmap",
+ "derive_more",
 "futures",
 "lazy_static",
 "log",
+ "pretty_env_logger",
 "rand 0.8.5",
 "rand_chacha 0.3.1",
 "serde",
--- a/Cargo.toml
+++ b/Cargo.toml
@ -3,8 +3,6 @@ name = "solana-rpc-testing"
 version = "0.1.0"
 edition = "2021"

-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
 serde = { version = "1.0.160", features = ["derive"] }
 serde_json = "1.0.96"
@ -20,7 +18,6 @@ log = "0.4.17"
 clap = { version = "4.2.4", features = ["derive"] }
 dashmap = "5.4.0"
 const_env = "0.1.2"
-tracing-subscriber = "0.3.16"
 chrono = "0.4.24"
 lazy_static = "1.4.0"
 rand = "0.8.5"
@ -29,3 +26,6 @@ async-trait = "0.1.68"

 solana-sdk = "1.15.2"
 solana-rpc-client = "1.15.2"
+derive_more = "0.99.17"
+pretty_env_logger = "0.4.0"
+tracing-subscriber = "0.3.17"
--- a/src/bencher.rs
+++ b/src/bencher.rs
@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use solana_rpc_client::nonblocking::rpc_client::RpcClient;
+
+use crate::cli::Args;
+use crate::metrics::{Metric, PartialMetric};
+
+#[async_trait::async_trait]
+pub trait BenchFn: Send + 'static {
+    async fn new(rpc_client: Arc<RpcClient>) -> anyhow::Result<Self>
+    where
+        Self: Sized;
+
+    async fn bench_fn(&mut self, rpc_client: Arc<RpcClient>) -> anyhow::Result<()>;
+}
+
+pub struct Bencher;
+
+impl Bencher {
+    pub async fn bench<B: BenchFn>(args: Args) -> anyhow::Result<Metric> {
+        let rpc_client = args.get_rpc_client();
+
+        let futs = (0..args.threads).map(|_| {
+            let rpc_client = rpc_client.clone();
+            let duration = args.get_duration_to_run_test();
+
+            tokio::spawn(async move {
+                let mut bench_fn = B::new(rpc_client.clone()).await.unwrap();
+
+                let mut part_metric = PartialMetric::default();
+                let thread_start = tokio::time::Instant::now();
+
+                while thread_start.elapsed() <= duration {
+                    let Err(err) = bench_fn.bench_fn(rpc_client.clone()).await else {
+                        part_metric.passed += 1;
+                        continue;
+                    };
+
+                    part_metric.failed += 1;
+                    log::warn!("{err}");
+                }
+
+                part_metric.total_time = thread_start.elapsed();
+                Metric::from(part_metric)
+            })
+        });
+
+        let avg_metric = futures::future::try_join_all(futs).await?.into_iter().sum::<Metric>() / args.threads;
+
+        Ok(avg_metric)
+    }
+}
--- a/src/cli.rs
+++ b/src/cli.rs
@ -36,7 +36,10 @@ pub struct Args {
    pub rpc_addr: String,

    #[arg(short = 'd', long, default_value_t = 60)]
-    pub duration_in_seconds: usize,
+    pub duration_in_seconds: u64,
+
+    #[arg(short = 't', long, default_value_t = 4)]
+    pub threads: u64,
 }

 impl Args {
@ -64,7 +67,7 @@ impl Args {

    #[inline]
    pub fn get_duration_to_run_test(&self) -> Duration {
-        Duration::from_secs(self.duration_in_seconds as u64)
+        Duration::from_secs(self.duration_in_seconds)
    }

    #[inline]
--- a/src/main.rs
+++ b/src/main.rs
@ -1,33 +1,36 @@
+pub mod bencher;
 mod cli;
 mod config;
 mod metrics;
 mod solana_runtime;
 mod test_registry;

+use anyhow::{bail, Context};
+use clap::Parser;
 use cli::Args;
 use config::Config;

-use clap::Parser;
-
 #[tokio::main(flavor = "multi_thread", worker_threads = 16)]
 async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+
    let args = Args::parse();

-    let config_file = args.config_file.clone();
+    assert_ne!(args.threads, 0, "Threads can't be 0");
+    assert_ne!(args.duration_in_seconds, 0, "Duration can't be 0");

-    let contents =
-        std::fs::read_to_string(config_file).expect("Should have been able to read the file");
-
-    let config_json: Config =
-        serde_json::from_str(contents.as_str()).expect("Config file not valid");
+    let contents = std::fs::read_to_string(&args.config_file)
+        .context("Should have been able to read the file")?;
+    let config_json: Config = serde_json::from_str(&contents).context("Config file not valid")?;

    if config_json.payers.is_empty() {
        log::error!("config file is missing payers");
-        return Err(anyhow::Error::msg("No payers"));
+        bail!("No payers");
    }

-    let registry = args.generate_test_registry();
-    registry.start_testing(args, config_json).await;
+    args.generate_test_registry()
+        .start_testing(args, config_json)
+        .await;

    Ok(())
 }
--- a/src/metrics.rs
+++ b/src/metrics.rs
@ -1,6 +1,59 @@
-#[derive(Default)]
-pub struct Metrics {
-    pub requests_per_second: u32,
-    pub time_per_request: u32,
-    pub total_transferred: u32,
+use derive_more::{Add, Sum};
+use std::time::Duration;
+
+use serde::Serialize;
+
+#[derive(Default, Serialize, Add, Sum)]
+pub struct Metric {
+    pub requests_per_second: f64,
+    pub time_per_request: f64,
+    pub total_transferred: u64,
+    #[serde(flatten)]
+    pub partial_metric: PartialMetric,
+}
+
+#[derive(Default, Serialize, Add, Sum)]
+pub struct PartialMetric {
+    pub total_time: Duration,
+    pub passed: u64,
+    pub failed: u64,
+}
+
+impl std::ops::Div<u64> for Metric {
+    type Output = Self;
+
+    fn div(self, rhs: u64) -> Self::Output {
+        Self {
+            requests_per_second: self.requests_per_second / rhs as f64,
+            time_per_request: self.time_per_request / rhs as f64,
+            total_transferred: self.total_transferred / rhs,
+            partial_metric: self.partial_metric / rhs,
+        }
+    }
+}
+
+impl std::ops::Div<u64> for PartialMetric {
+    type Output = Self;
+
+    fn div(self, rhs: u64) -> Self::Output {
+        Self {
+            total_time: Duration::from_secs_f64(self.total_time.as_secs_f64() / rhs as f64),
+            passed: self.passed / rhs,
+            failed: self.failed / rhs,
+        }
+    }
+}
+
+impl From<PartialMetric> for Metric {
+    fn from(partial_metric: PartialMetric) -> Self {
+        let total_transferred = partial_metric.passed + partial_metric.failed;
+        let total_time_secs = partial_metric.total_time.as_secs_f64();
+
+        Metric {
+            requests_per_second: total_transferred as f64 / total_time_secs,
+            time_per_request: total_time_secs / total_transferred as f64,
+            total_transferred,
+            partial_metric,
+        }
+    }
 }
--- a/src/solana_runtime/accounts_fetching.rs
+++ b/src/solana_runtime/accounts_fetching.rs
@ -11,7 +11,7 @@ use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::{pubkey::Pubkey, signature::Keypair, signer::Signer};
 use tokio::time::Instant;

-use crate::{config::Config, test_registry::TestingTask, metrics::Metrics};
+use crate::{config::Config, test_registry::TestingTask};

 #[from_env]
 const NB_ACCOUNT_FETCHING_TASKS: usize = 10;
@ -29,7 +29,7 @@ impl AccountsFetchingTests {

 #[async_trait]
 impl TestingTask for AccountsFetchingTests {
-    async fn test(&self, args: crate::cli::Args, config: Config) -> anyhow::Result<Metrics> {
+    async fn test(&self, args: crate::cli::Args, config: Config) -> anyhow::Result<()> {
        let rpc_client = Arc::new(RpcClient::new(args.rpc_addr.clone()));
        let total_fetches = Arc::new(AtomicU64::new(0));

@ -117,7 +117,7 @@ impl TestingTask for AccountsFetchingTests {
            NB_ACCOUNT_FETCHING_TASKS
        );

-        Ok(todo!())
+        Ok(())
    }

    fn get_name(&self) -> &'static str {
--- a/src/solana_runtime/get_block.rs
+++ b/src/solana_runtime/get_block.rs
@ -1,24 +1,46 @@
-use crate::{metrics::Metrics, cli::Args, config::Config, test_registry::TestingTask};
+use std::sync::Arc;
+
+use log::info;
+use solana_rpc_client::nonblocking::rpc_client::RpcClient;
+use solana_sdk::slot_history::Slot;
+
+use crate::{
+    bencher::{BenchFn, Bencher},
+    cli::Args,
+    config::Config,
+    test_registry::TestingTask,
+};

 pub struct GetBlockTest;

 #[async_trait::async_trait]
 impl TestingTask for GetBlockTest {
-    async fn test(&self, args: Args, _config: Config) -> anyhow::Result<Metrics> {
-        let rpc_client = args.get_rpc_client();
-
-        let futs = (0..12).map(|_| {
-            let rpc_client = rpc_client.clone();
-
-            tokio::spawn(async move { rpc_client.get_slot().await })
-        });
-
-        let _slots = futures::future::try_join_all(futs).await?;
-
-        Ok(Default::default())
+    async fn test(&self, args: Args, _config: Config) -> anyhow::Result<()> {
+        let metric = Bencher::bench::<GetBlockBench>(args).await?;
+        info!("{}", serde_json::to_string(&metric)?);
+        Ok(())
    }

    fn get_name(&self) -> &'static str {
        "GetBlockTest"
    }
 }
+
+pub struct GetBlockBench {
+    slot: Slot,
+}
+
+#[async_trait::async_trait]
+impl BenchFn for GetBlockBench {
+    async fn new(rpc_client: Arc<RpcClient>) -> anyhow::Result<Self> {
+        Ok(Self {
+            slot: rpc_client.get_slot().await?,
+        })
+    }
+
+    async fn bench_fn(&mut self, rpc_client: Arc<RpcClient>) -> anyhow::Result<()> {
+      //  self.slot += 1;
+        rpc_client.get_block(self.slot).await?;
+        Ok(())
+    }
+}
--- a/src/solana_runtime/get_slot.rs
+++ b/src/solana_runtime/get_slot.rs
@ -1,24 +1,34 @@
-use crate::{cli::Args, config::Config, metrics::Metrics, test_registry::TestingTask};
+use std::sync::Arc;
+
+use log::info;
+use solana_rpc_client::nonblocking::rpc_client::RpcClient;
+
+use crate::bencher::{BenchFn, Bencher};
+use crate::{cli::Args, config::Config, test_registry::TestingTask};

 pub struct GetSlotTest;

 #[async_trait::async_trait]
 impl TestingTask for GetSlotTest {
-    async fn test(&self, args: Args, _config: Config) -> anyhow::Result<Metrics> {
-        let rpc_client = args.get_rpc_client();
-
-        let futs = (0..12).map(|_| {
-            let rpc_client = rpc_client.clone();
-
-            tokio::spawn(async move { rpc_client.get_slot().await })
-        });
-
-        let _slots = futures::future::try_join_all(futs).await?;
-
-        Ok(Default::default())
+    async fn test(&self, args: Args, _config: Config) -> anyhow::Result<()> {
+        let metric = Bencher::bench::<Self>(args).await?;
+        info!("metric {}", serde_json::to_string(&metric)?);
+        Ok(())
    }

    fn get_name(&self) -> &'static str {
        "GetSlotTest"
    }
 }
+
+#[async_trait::async_trait]
+impl BenchFn for GetSlotTest {
+    async fn new(_: Arc<RpcClient>) -> anyhow::Result<Self> {
+        Ok(Self)
+    }
+
+    async fn bench_fn(&mut self, rpc_client: Arc<RpcClient>) -> anyhow::Result<()> {
+        rpc_client.get_slot().await?;
+        Ok(())
+    }
+}
--- a/src/solana_runtime/send_and_get_status_memo.rs
+++ b/src/solana_runtime/send_and_get_status_memo.rs
@ -1,4 +1,4 @@
-use crate::{metrics::Metrics, test_registry::TestingTask};
+use crate::test_registry::TestingTask;
 use async_trait::async_trait;
 use const_env::from_env;
 use dashmap::DashSet;
@ -114,7 +114,7 @@ impl TestingTask for SendAndConfrimTesting {
        &self,
        args: crate::cli::Args,
        config: crate::config::Config,
-    ) -> anyhow::Result<Metrics> {
+    ) -> anyhow::Result<()> {
        println!("started sending and confirming memo transactions");
        let rpc_client = Arc::new(RpcClient::new(args.rpc_addr.clone()));
        let block_hash: Arc<RwLock<Hash>> = Arc::new(RwLock::new(Hash::default()));
@ -143,14 +143,14 @@ impl TestingTask for SendAndConfrimTesting {
        let mut tasks = vec![];
        let payers = config.get_payers();
        for seed in 0..nb_runs {
-            let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(seed as u64);
+            let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(seed);
            let payer = payers.choose(&mut rng).unwrap();
            let payer = Keypair::from_bytes(payer.to_bytes().as_slice()).unwrap();
            tasks.push(tokio::spawn(send_and_confirm_transactions(
                rpc_client.clone(),
                NB_MEMO_TRANSACTIONS_SENT_PER_SECOND,
                payer,
-                seed as u64,
+                seed,
                block_hash.clone(),
            )));
            // wait for an interval
@ -170,7 +170,7 @@ impl TestingTask for SendAndConfrimTesting {

        println!("Memo transaction sent and confrim results \n Number of transaction confirmed : {}, \n Number of transactions unconfirmed {}, took {}s", total_txs_confirmed, total_txs_unconfirmed, duration.as_secs());

-        Ok(todo!())
+        Ok(())
    }

    fn get_name(&self) -> &'static str {
--- a/src/test_registry.rs
+++ b/src/test_registry.rs
@ -1,10 +1,10 @@
 use async_trait::async_trait;

-use crate::{cli::Args, config::Config, metrics::Metrics};
+use crate::{cli::Args, config::Config};

 #[async_trait]
 pub trait TestingTask: Send + Sync {
-    async fn test(&self, args: Args, config: Config) -> anyhow::Result<Metrics>;
+    async fn test(&self, args: Args, config: Config) -> anyhow::Result<()>;
    fn get_name(&self) -> &'static str;
 }

@ -22,16 +22,17 @@ impl TestRegistry {
        let tasks = self.tests.into_iter().map(|test| {
            let args = args.clone();
            let config = config.clone();
+            let name = test.get_name();

            tokio::spawn(async move {
-                log::info!("test {}", test.get_name());
+                log::info!("test {name}");

                match test.test(args, config).await {
                    Ok(_) => {
-                        println!("test {} passed", test.get_name());
+                        log::info!("test {name} passed");
                    }
                    Err(e) => {
-                        println!("test {} failed with error {}", test.get_name(), e);
+                        log::info!("test {} failed with error {name}", e);
                    }
                }
            })