Updating bencher and stats

src/bencher.rs

@@ -1,19 +1,24 @@
+use crate::cli::Args;
+use itertools::Itertools;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use serde::Serialize;
+use solana_program::hash::Hash;
+use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
+use tokio::sync::RwLock;
 
-use itertools::Itertools;
-use serde::Serialize;
-use solana_rpc_client::nonblocking::rpc_client::RpcClient;
-
-use crate::cli::Args;
+pub type BlockHashGetter = Arc<RwLock<Hash>>;
 
 #[async_trait::async_trait]
-pub trait Benchmark: Send + 'static {
-    async fn prepare(rpc_client: Arc<RpcClient>) -> anyhow::Result<Self>
-    where
-        Self: Sized;
-
-    async fn run(&mut self, rpc_client: Arc<RpcClient>, duration: Duration) -> anyhow::Result<Run>;
+pub trait Benchmark: Clone + Send + 'static {
+    async fn run(
+        self,
+        rpc_client: Arc<RpcClient>,
+        duration: Duration,
+        random_number: u64,
+    ) -> anyhow::Result<Run>;
 }
 
 #[derive(Default, Serialize)]
@@ -25,29 +30,39 @@ pub struct Run {
     pub errors: Vec<String>,
 }
 
-#[derive(Default, Serialize)]
+#[derive(Default, Serialize, Clone)]
 pub struct Stats {
     pub total_requests: u64,
     pub requests_per_second: f64,
     pub time_per_request: f64,
     pub total_transferred: u64,
     pub top_5_errors: Vec<(String, usize)>,
-    #[serde(flatten)]
-    pub all_runs: Vec<Run>,
+    pub average_number_of_requests_by_a_task: f64,
+    pub total_requests_succeded: u64,
+    pub total_requests_failed: u64,
+    pub average_succeds_per_task: f64,
+    pub average_failed_per_task: f64,
 }
 
 pub struct Bencher;
 
 impl Bencher {
-    pub async fn bench<B: Benchmark>(args: Args) -> anyhow::Result<Stats> {
+    pub async fn bench<B: Benchmark + Send + Clone>(
+        instant: B,
+        args: Args,
+    ) -> anyhow::Result<Stats> {
         let start = Instant::now();
+        let mut random = StdRng::seed_from_u64(0);
         let futs = (0..args.threads).map(|_| {
             let rpc_client = args.get_rpc_client();
             let duration = args.get_duration_to_run_test();
-
+            let random_number = random.gen();
+            let instance = instant.clone();
             tokio::spawn(async move {
-                let mut benchmark = B::prepare(rpc_client.clone()).await.unwrap();
-                benchmark.run(rpc_client.clone(), duration).await.unwrap()
+                instance
+                    .run(rpc_client.clone(), duration, random_number)
+                    .await
+                    .unwrap()
             })
         });
 
@@ -58,6 +73,12 @@ impl Bencher {
         let total_requests = all_results
             .iter()
             .fold(0, |acc, r| acc + r.requests_completed + r.requests_failed);
+
+        let total_requests_succeded = all_results
+            .iter()
+            .fold(0, |acc, r| acc + r.requests_completed);
+        let total_requests_failed = all_results.iter().fold(0, |acc, r| acc + r.requests_failed);
+
         let total_transferred = all_results
             .iter()
             .fold(0, |acc, r| acc + r.bytes_sent + r.bytes_received);
@@ -73,10 +94,14 @@ impl Bencher {
         Ok(Stats {
             total_requests,
             requests_per_second: total_requests as f64 / time.as_secs_f64(),
-            time_per_request: time.as_secs_f64() / total_requests as f64,
+            time_per_request: time.as_secs_f64() / (total_requests as f64 / args.threads as f64),
             total_transferred,
             top_5_errors,
-            all_runs: all_results,
+            average_number_of_requests_by_a_task: (total_requests as f64) / (args.threads as f64),
+            total_requests_succeded,
+            total_requests_failed,
+            average_succeds_per_task: total_requests_succeded as f64 / args.threads as f64,
+            average_failed_per_task: total_requests_failed as f64 / args.threads as f64,
         })
     }
 }

src/cli.rs

@@ -54,6 +54,9 @@ pub struct Args {
     #[arg(short = 't', long, default_value_t = 32)]
     pub threads: u64,
 
+    #[arg(short = 'o', long, default_value_t = String::from("out.json"))]
+    pub output_file: String,
+
     #[arg(short = 'p', long)]
     pub print_logs: bool,
 }

src/main.rs

@@ -33,6 +33,11 @@ async fn main() -> anyhow::Result<()> {
         bail!("No payers");
     }
 
+    if config_json.markets.is_empty() {
+        log::error!("Config file is missing markets");
+        bail!("No markets")
+    }
+
     let rpc_client = args.get_rpc_client();
     let current_hash = rpc_client.get_latest_blockhash().await.unwrap();
     let block_hash: Arc<RwLock<Hash>> = Arc::new(RwLock::new(current_hash));

src/openbook/simulate_place_orders.rs

@@ -1,7 +1,13 @@
+use crate::bencher::{Bencher, Benchmark, Run, Stats};
+use crate::config::{Market, User};
 use crate::test_registry::TestingTask;
 use crate::utils::noop;
 use async_trait::async_trait;
+use rand::rngs::StdRng;
+use rand::seq::SliceRandom;
+use rand::{Rng, SeedableRng};
 use serde::{Deserialize, Serialize};
+use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::compute_budget;
 use solana_sdk::hash::Hash;
 use solana_sdk::{
@@ -12,7 +18,6 @@ use solana_sdk::{
     transaction::Transaction,
 };
 use std::mem::size_of;
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::{str::FromStr, sync::Arc, time::Instant};
 use tokio::sync::RwLock;
 
@@ -51,15 +56,14 @@ impl TestingTask for SimulateOpenbookV2PlaceOrder {
         &self,
         args: crate::cli::Args,
         config: crate::config::Config,
-    ) -> anyhow::Result<()> {
-        let mut tasks = vec![];
+    ) -> anyhow::Result<Stats> {
         let openbook_data = config
             .programs
             .iter()
             .find(|x| x.name == "openbook_v2")
             .unwrap()
             .clone();
-        let openbook_pk = Pubkey::from_str(openbook_data.program_id.as_str()).unwrap();
+        let openbook_pid: Pubkey = Pubkey::from_str(openbook_data.program_id.as_str()).unwrap();
         let place_order_cmd = openbook_data
             .commands
             .iter()
@@ -68,134 +72,135 @@ impl TestingTask for SimulateOpenbookV2PlaceOrder {
             .clone();
 
         assert!(place_order_cmd.instruction.len() == 8 + size_of::<PlaceOrderArgs>());
-        let successful_orders_count = Arc::new(AtomicU64::new(0));
 
-        for user in &config.users {
-            for market in &config.markets {
-                let market = market.clone();
-                let user = user.clone();
-                let args = args.clone();
-                let place_order_cmd = place_order_cmd.clone();
-                let openbook_pk = openbook_pk.clone();
-                let block_hash = self.block_hash.clone();
-                let open_orders = user.open_orders[market.market_index]
-                    .open_orders
-                    .to_pubkey();
-                let base_token_account = user.token_data[market.market_index + 1]
-                    .token_account
-                    .to_pubkey();
-                let quote_token_account = user.token_data[0].token_account.to_pubkey();
-                let successful_orders_count = successful_orders_count.clone();
-
-                let task = tokio::spawn(async move {
-                    let start = Instant::now();
-
-                    let mut place_order_ix = place_order_cmd.instruction;
-                    let rpc_client = args.get_rpc_client();
-                    let mut side = false;
-                    let mut price_diff: i64 = 1;
-                    let mut max_base_lots = 1;
-                    while start.elapsed().as_secs() < args.duration_in_seconds {
-                        let order_type: u8 = 0;
-
-                        let price_lots: i64 = if side {
-                            1000 - price_diff
-                        } else {
-                            1000 + price_diff
-                        };
-                        let place_order_params = PlaceOrderArgs {
-                            client_order_id: 100,
-                            side: side as u8,
-                            price_lots,
-                            max_base_lots,
-                            max_quote_lots_including_fees: i64::MAX,
-                            order_type,
-                            reduce_only: false,
-                            expiry_timestamp: u64::MAX,
-                            limit: 10,
-                        };
-                        let bytes: Vec<u8> = bincode::serialize(&place_order_params).unwrap();
-                        assert!(bytes.len() + 8 == place_order_ix.len());
-
-                        // copy the instruction data
-                        for i in 0..bytes.len() {
-                            place_order_ix[8 + i] = bytes[i];
-                        }
-
-                        let token_account = if side {
-                            base_token_account
-                        } else {
-                            quote_token_account
-                        };
-                        let accounts = vec![
-                            AccountMeta::new(open_orders, false),
-                            AccountMeta::new(user.pubkey(), false),
-                            AccountMeta::new(market.market_pk.to_pubkey(), false),
-                            AccountMeta::new(market.bids.to_pubkey(), false),
-                            AccountMeta::new(market.asks.to_pubkey(), false),
-                            AccountMeta::new(token_account, false),
-                            AccountMeta::new(market.base_vault.to_pubkey(), false),
-                            AccountMeta::new(market.quote_vault.to_pubkey(), false),
-                            AccountMeta::new(market.event_queue.to_pubkey(), false),
-                            AccountMeta::new_readonly(market.oracle.to_pubkey(), false),
-                            AccountMeta::new_readonly(spl_token::ID, false),
-                            AccountMeta::new_readonly(solana_sdk::system_program::id(), false),
-                        ];
-
-                        let ix = Instruction::new_with_bytes(
-                            openbook_pk,
-                            place_order_ix.as_slice(),
-                            accounts,
-                        );
-
-                        let recent_blockhash = *block_hash.read().await;
-
-                        // to generate new signature each time
-                        let noop_ix = noop::timestamp();
-                        // to have higher compute budget limit
-                        let cu_limits_ix =
-                            compute_budget::ComputeBudgetInstruction::set_compute_unit_limit(
-                                1000000,
-                            );
-
-                        let transaction = Transaction::new(
-                            &[&user.get_keypair()],
-                            Message::new(&[noop_ix, cu_limits_ix, ix], Some(&user.pubkey())),
-                            recent_blockhash,
-                        );
-                        let signature = transaction.signatures[0];
-                        if let Err(e) = rpc_client.simulate_transaction(&transaction).await {
-                            log::error!(
-                                "error while simulating openbook place order {} sig {}",
-                                e,
-                                signature
-                            );
-                        } else {
-                            successful_orders_count.fetch_add(1, Ordering::Relaxed);
-                        }
-
-                        // update side and price diff
-                        side = !side;
-                        price_diff = price_diff % 6 + 1;
-                        max_base_lots = max_base_lots % 10 + 1;
-                    }
-                });
-                tasks.push(task);
-            }
-        }
-
-        futures::future::join_all(tasks).await;
-
-        log::info!(
-            "Simulated {} transactions with {} users and {} markets",
-            successful_orders_count.load(Ordering::Relaxed),
-            config.users.len(),
-            config.markets.len()
-        );
-        Ok(())
+        let instant = SimulateOpenbookV2PlaceOrderBench {
+            block_hash: self.block_hash.clone(),
+            markets: config.markets.clone(),
+            users: config.users.clone(),
+            place_order_cmd: place_order_cmd.instruction,
+            openbook_pid,
+        };
+        let metric = Bencher::bench::<SimulateOpenbookV2PlaceOrderBench>(instant, args).await?;
+        log::info!("{} {}", self.get_name(), serde_json::to_string(&metric)?);
+        Ok(metric)
     }
 
     fn get_name(&self) -> String {
         format!("Simulating openbook place orders")
     }
 }
+
+#[derive(Clone)]
+pub struct SimulateOpenbookV2PlaceOrderBench {
+    pub block_hash: Arc<RwLock<Hash>>,
+    pub markets: Vec<Market>,
+    pub users: Vec<User>,
+    pub place_order_cmd: Vec<u8>,
+    pub openbook_pid: Pubkey,
+}
+
+#[async_trait::async_trait]
+impl Benchmark for SimulateOpenbookV2PlaceOrderBench {
+    async fn run(
+        self,
+        rpc_client: Arc<RpcClient>,
+        duration: std::time::Duration,
+        random_number: u64,
+    ) -> anyhow::Result<crate::bencher::Run> {
+        let mut result = Run::default();
+
+        let mut rng = StdRng::seed_from_u64(random_number);
+        let start = Instant::now();
+        while start.elapsed() < duration {
+            let mut place_order_ix = self.place_order_cmd.clone();
+            let market = self.markets.choose(&mut rng).cloned().unwrap();
+            let user = self.users.choose(&mut rng).cloned().unwrap();
+
+            let open_orders = user.open_orders[market.market_index]
+                .open_orders
+                .to_pubkey();
+            let base_token_account = user.token_data[market.market_index + 1]
+                .token_account
+                .to_pubkey();
+            let quote_token_account = user.token_data[0].token_account.to_pubkey();
+
+            let side = rng.gen_bool(0.5);
+            let order_type: u8 = 0;
+            let price_diff = 3;
+            let price_lots: i64 = if side {
+                1000 - price_diff
+            } else {
+                1000 + price_diff
+            };
+            let max_base_lots = 5;
+            let place_order_params = PlaceOrderArgs {
+                client_order_id: 100,
+                side: side as u8,
+                price_lots,
+                max_base_lots,
+                max_quote_lots_including_fees: i64::MAX,
+                order_type,
+                reduce_only: false,
+                expiry_timestamp: u64::MAX,
+                limit: 10,
+            };
+            let bytes: Vec<u8> = bincode::serialize(&place_order_params).unwrap();
+            assert!(bytes.len() + 8 == place_order_ix.len());
+
+            // copy the instruction data
+            for i in 0..bytes.len() {
+                place_order_ix[8 + i] = bytes[i];
+            }
+
+            let token_account = if side {
+                base_token_account
+            } else {
+                quote_token_account
+            };
+            let accounts = vec![
+                AccountMeta::new(open_orders, false),
+                AccountMeta::new(user.pubkey(), false),
+                AccountMeta::new(market.market_pk.to_pubkey(), false),
+                AccountMeta::new(market.bids.to_pubkey(), false),
+                AccountMeta::new(market.asks.to_pubkey(), false),
+                AccountMeta::new(token_account, false),
+                AccountMeta::new(market.base_vault.to_pubkey(), false),
+                AccountMeta::new(market.quote_vault.to_pubkey(), false),
+                AccountMeta::new(market.event_queue.to_pubkey(), false),
+                AccountMeta::new_readonly(market.oracle.to_pubkey(), false),
+                AccountMeta::new_readonly(spl_token::ID, false),
+                AccountMeta::new_readonly(solana_sdk::system_program::id(), false),
+            ];
+
+            let ix =
+                Instruction::new_with_bytes(self.openbook_pid, place_order_ix.as_slice(), accounts);
+
+            let recent_blockhash = *self.block_hash.read().await;
+
+            // to generate new signature each time
+            let noop_ix = noop::timestamp();
+            // to have higher compute budget limit
+            let cu_limits_ix =
+                compute_budget::ComputeBudgetInstruction::set_compute_unit_limit(1000000);
+
+            let transaction = Transaction::new(
+                &[&user.get_keypair()],
+                Message::new(&[noop_ix, cu_limits_ix, ix], Some(&user.pubkey())),
+                recent_blockhash,
+            );
+
+            match rpc_client.simulate_transaction(&transaction).await {
+                Ok(_) => {
+                    result.requests_completed += 1;
+                    result.bytes_received += 0;
+                }
+                Err(e) => {
+                    result.requests_failed += 1;
+                    result.errors.push(format!("{:?}", e.kind()));
+                }
+            }
+            result.bytes_sent += 0;
+        }
+        Ok(result)
+    }
+}

src/solana_runtime/accounts_fetching.rs

@@ -1,20 +1,16 @@
-use std::{
-    collections::HashSet,
-    str::FromStr,
-    sync::{atomic::AtomicU64, Arc},
-};
-
 use async_trait::async_trait;
 use const_env::from_env;
 use rand::{seq::IteratorRandom, SeedableRng};
 use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::{pubkey::Pubkey, signature::Keypair, signer::Signer};
+use std::{str::FromStr, sync::Arc};
 use tokio::time::Instant;
 
-use crate::{config::Config, test_registry::TestingTask};
-
-#[from_env]
-const NB_ACCOUNT_FETCHING_TASKS: usize = 10;
+use crate::{
+    bencher::{Bencher, Benchmark, Run, Stats},
+    config::Config,
+    test_registry::TestingTask,
+};
 
 #[from_env]
 const NB_OF_ACCOUNTS_FETCHED_PER_TASK: usize = 100;
@@ -29,101 +25,65 @@ impl AccountsFetchingTests {
 
 #[async_trait]
 impl TestingTask for AccountsFetchingTests {
-    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));
-
-        let known_accounts = config
+    async fn test(&self, args: crate::cli::Args, config: Config) -> anyhow::Result<Stats> {
+        let accounts = config
             .known_accounts
             .iter()
             .map(|x| Pubkey::from_str(x.as_str()).unwrap())
             .collect::<Vec<_>>();
         let unknown_accounts: Vec<Pubkey> =
-            AccountsFetchingTests::create_random_address(known_accounts.len());
-        let number_of_fetched_accounts = NB_OF_ACCOUNTS_FETCHED_PER_TASK.min(known_accounts.len());
+            AccountsFetchingTests::create_random_address(accounts.len());
 
-        let hash_set_known = Arc::new(known_accounts.iter().copied().collect::<HashSet<_>>());
-
-        let mut tasks = vec![];
-        for i in 0..NB_ACCOUNT_FETCHING_TASKS {
-            // each new task will fetch (100/NB_ACCOUNT_FETCHING_TASKS) * i percent of unknown accounts
-            // so first task will fetch no unknown accounts and last will fetch almost all unknown accounts
-            let known_accounts = known_accounts.clone();
-            let unknown_accounts = unknown_accounts.clone();
-            let duration = args.get_duration_to_run_test();
-            let rpc_client = rpc_client.clone();
-            let hash_set_known = hash_set_known.clone();
-            let total_fetches = total_fetches.clone();
-            let task = tokio::spawn(async move {
-                let percentage_of_unknown_tasks =
-                    (100.0 / (NB_ACCOUNT_FETCHING_TASKS as f64)) * (i as f64);
-
-                println!("started fetching accounts task #{}", i);
-                let unknown_accounts_to_take = ((number_of_fetched_accounts as f64)
-                    * percentage_of_unknown_tasks
-                    / 100.0) as usize;
-                let known_accounts_to_take =
-                    number_of_fetched_accounts.saturating_sub(unknown_accounts_to_take);
-
-                let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(i as u64);
-
-                let instant = Instant::now();
-                while instant.elapsed() < duration {
-                    let known_accounts = known_accounts
-                        .iter()
-                        .choose_multiple(&mut rng, known_accounts_to_take);
-                    let unknown_accounts = unknown_accounts
-                        .iter()
-                        .choose_multiple(&mut rng, unknown_accounts_to_take);
-                    let accounts_to_fetch = [known_accounts, unknown_accounts]
-                        .concat()
-                        .iter()
-                        .map(|x| *(*x))
-                        .collect::<Vec<_>>();
-                    let res = rpc_client
-                        .get_multiple_accounts(accounts_to_fetch.as_slice())
-                        .await;
-                    total_fetches.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-                    match res {
-                        Ok(res) => {
-                            if res.len() == accounts_to_fetch.len() {
-                                for i in 0..accounts_to_fetch.len() {
-                                    if hash_set_known.contains(&accounts_to_fetch[i]) {
-                                        if res[i].is_none() {
-                                            println!(
-                                                "unable to fetch known account {}",
-                                                accounts_to_fetch[i]
-                                            );
-                                        }
-                                    } else if res[i].is_some() {
-                                        println!("fetched unknown account should not be possible");
-                                    }
-                                }
-                            } else {
-                                println!("fetched accounts results mismatched");
-                            }
-                        }
-                        Err(e) => {
-                            println!("accounts fetching failed because {}", e);
-                        }
-                    }
-                }
-            });
-            tasks.push(task);
-        }
-
-        futures::future::join_all(tasks).await;
-        println!(
-            "Accounts fetching did {} requests of {} accounts in {} tasks",
-            total_fetches.load(std::sync::atomic::Ordering::Relaxed),
-            number_of_fetched_accounts,
-            NB_ACCOUNT_FETCHING_TASKS
-        );
-
-        Ok(())
+        let instant = GetAccountsBench {
+            accounts_list: [accounts, unknown_accounts].concat(),
+        };
+        let metric = Bencher::bench::<GetAccountsBench>(instant, args).await?;
+        log::info!("{} {}", self.get_name(), serde_json::to_string(&metric)?);
+        Ok(metric)
     }
 
     fn get_name(&self) -> String {
         "Accounts Fetching".to_string()
     }
 }
+
+#[derive(Clone)]
+pub struct GetAccountsBench {
+    accounts_list: Vec<Pubkey>,
+}
+
+#[async_trait::async_trait]
+impl Benchmark for GetAccountsBench {
+    async fn run(
+        self,
+        rpc_client: Arc<RpcClient>,
+        duration: std::time::Duration,
+        random_number: u64,
+    ) -> anyhow::Result<crate::bencher::Run> {
+        let mut result = Run::default();
+        let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(random_number);
+        let number_of_fetched_accounts =
+            NB_OF_ACCOUNTS_FETCHED_PER_TASK.min(self.accounts_list.len());
+        let start = Instant::now();
+        while start.elapsed() < duration {
+            let accounts = self
+                .accounts_list
+                .iter()
+                .copied()
+                .choose_multiple(&mut rng, number_of_fetched_accounts);
+
+            match rpc_client.get_multiple_accounts(accounts.as_slice()).await {
+                Ok(_) => {
+                    result.requests_completed += 1;
+                    result.bytes_received += 0;
+                }
+                Err(e) => {
+                    result.requests_failed += 1;
+                    result.errors.push(format!("{:?}", e.kind()));
+                }
+            }
+            result.bytes_sent += 0;
+        }
+        Ok(result)
+    }
+}

src/solana_runtime/get_block.rs

@@ -8,7 +8,7 @@ use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::slot_history::Slot;
 
 use crate::{
-    bencher::{Bencher, Benchmark, Run},
+    bencher::{Bencher, Benchmark, Run, Stats},
     cli::Args,
     config::Config,
     test_registry::TestingTask,
@@ -18,10 +18,12 @@ pub struct GetBlockTest;
 
 #[async_trait::async_trait]
 impl TestingTask for GetBlockTest {
-    async fn test(&self, args: Args, _config: Config) -> anyhow::Result<()> {
-        let metric = Bencher::bench::<GetBlockBench>(args).await?;
-        info!("{}", serde_json::to_string(&metric)?);
-        Ok(())
+    async fn test(&self, args: Args, _: Config) -> anyhow::Result<Stats> {
+        let slot = { args.get_rpc_client().get_slot().await.unwrap() };
+        let instant = GetBlockBench { slot };
+        let metric = Bencher::bench::<GetBlockBench>(instant, args).await?;
+        info!("{} {}", self.get_name(), serde_json::to_string(&metric)?);
+        Ok(metric)
     }
 
     fn get_name(&self) -> String {
@@ -29,19 +31,19 @@ impl TestingTask for GetBlockTest {
     }
 }
 
+#[derive(Clone)]
 pub struct GetBlockBench {
     slot: Slot,
 }
 
 #[async_trait::async_trait]
 impl Benchmark for GetBlockBench {
-    async fn prepare(rpc_client: Arc<RpcClient>) -> anyhow::Result<Self> {
-        Ok(Self {
-            slot: rpc_client.get_slot().await?,
-        })
-    }
-
-    async fn run(&mut self, rpc_client: Arc<RpcClient>, duration: Duration) -> anyhow::Result<Run> {
+    async fn run(
+        self,
+        rpc_client: Arc<RpcClient>,
+        duration: Duration,
+        _: u64,
+    ) -> anyhow::Result<Run> {
         let mut result = Run::default();
 
         let start = Instant::now();

src/solana_runtime/get_slot.rs

@@ -4,17 +4,19 @@ use std::time::{Duration, Instant};
 use log::info;
 use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 
-use crate::bencher::{Bencher, Benchmark, Run};
+use crate::bencher::{Bencher, Benchmark, Run, Stats};
 use crate::{cli::Args, config::Config, test_registry::TestingTask};
 
+#[derive(Clone)]
 pub struct GetSlotTest;
 
 #[async_trait::async_trait]
 impl TestingTask for GetSlotTest {
-    async fn test(&self, args: Args, _config: Config) -> anyhow::Result<()> {
-        let stats = Bencher::bench::<Self>(args).await?;
-        info!("GetSlotTest {}", serde_json::to_string(&stats)?);
-        Ok(())
+    async fn test(&self, args: Args, _: Config) -> anyhow::Result<Stats> {
+        let instant = GetSlotTest;
+        let stats = Bencher::bench::<Self>(instant, args).await?;
+        info!("{} {}", self.get_name(), serde_json::to_string(&stats)?);
+        Ok(stats)
     }
 
     fn get_name(&self) -> String {
@@ -24,11 +26,12 @@ impl TestingTask for GetSlotTest {
 
 #[async_trait::async_trait]
 impl Benchmark for GetSlotTest {
-    async fn prepare(_: Arc<RpcClient>) -> anyhow::Result<Self> {
-        Ok(Self)
-    }
-
-    async fn run(&mut self, rpc_client: Arc<RpcClient>, duration: Duration) -> anyhow::Result<Run> {
+    async fn run(
+        self,
+        rpc_client: Arc<RpcClient>,
+        duration: Duration,
+        _: u64,
+    ) -> anyhow::Result<Run> {
         let mut result = Run::default();
 
         let start = Instant::now();

src/solana_runtime/send_and_get_status_memo.rs

@@ -1,31 +1,19 @@
-use crate::test_registry::TestingTask;
+use crate::{
+    bencher::{Bencher, Benchmark, Run, Stats},
+    test_registry::TestingTask,
+};
 use async_trait::async_trait;
-use const_env::from_env;
-use dashmap::DashSet;
 use rand::{distributions::Alphanumeric, prelude::Distribution, seq::SliceRandom, SeedableRng};
 use solana_rpc_client::nonblocking::rpc_client::RpcClient;
 use solana_sdk::{
     hash::Hash, instruction::Instruction, message::Message, pubkey::Pubkey, signature::Keypair,
     signer::Signer, transaction::Transaction,
 };
-use std::{
-    str::FromStr,
-    sync::Arc,
-    time::{Duration, Instant},
-};
+use std::{str::FromStr, sync::Arc, time::Instant};
 use tokio::sync::RwLock;
 
 const MEMO_PROGRAM_ID: &str = "MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr";
 
-#[from_env]
-const NB_MEMO_TRANSACTIONS_SENT_PER_SECOND: usize = 256;
-
-#[derive(Clone, Debug, Copy)]
-struct Metric {
-    pub number_of_confirmed_txs: usize,
-    pub number_of_unconfirmed_txs: usize,
-}
-
 pub struct SendAndConfrimTesting {
     pub block_hash: Arc<RwLock<Hash>>,
 }
@@ -38,125 +26,67 @@ fn create_memo_tx(msg: &[u8], payer: &Keypair, blockhash: Hash) -> Transaction {
     Transaction::new(&[payer], message, blockhash)
 }
 
-fn generate_random_strings(num_of_txs: usize, random_seed: Option<u64>) -> Vec<Vec<u8>> {
-    let seed = random_seed.map_or(0, |x| x);
-    let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(seed);
-    (0..num_of_txs)
-        .map(|_| Alphanumeric.sample_iter(&mut rng).take(10).collect())
-        .collect()
-}
-
-async fn send_and_confirm_transactions(
-    rpc_client: Arc<RpcClient>,
-    tx_count: usize,
-    funded_payer: Keypair,
-    seed: u64,
-    block_hash: Arc<RwLock<Hash>>,
-) -> Metric {
-    let map_of_txs = Arc::new(DashSet::new());
-    // transaction sender task
-    {
-        let map_of_txs = map_of_txs.clone();
-        let rpc_client = rpc_client.clone();
-        tokio::spawn(async move {
-            let map_of_txs = map_of_txs.clone();
-            let rand_strings = generate_random_strings(tx_count, Some(seed));
-
-            for rand_string in rand_strings {
-                let blockhash = { *block_hash.read().await };
-                let tx = create_memo_tx(&rand_string, &funded_payer, blockhash);
-                if let Ok(signature) = rpc_client.send_transaction(&tx).await {
-                    map_of_txs.insert(signature);
-                }
-            }
-        });
-    }
-
-    let confirmation_time = Instant::now();
-    let mut confirmed_count = 0;
-
-    let mut metric = Metric {
-        number_of_confirmed_txs: 0,
-        number_of_unconfirmed_txs: 0,
-    };
-
-    while confirmation_time.elapsed() < Duration::from_secs(120)
-        && !(map_of_txs.is_empty() && confirmed_count == tx_count)
-    {
-        let signatures = map_of_txs.iter().map(|x| *x.key()).collect::<Vec<_>>();
-        if signatures.is_empty() {
-            tokio::time::sleep(Duration::from_millis(1)).await;
-            continue;
-        }
-
-        if let Ok(res) = rpc_client.get_signature_statuses(&signatures).await {
-            for i in 0..signatures.len() {
-                let tx_status = &res.value[i];
-                if tx_status.is_some() {
-                    let signature = signatures[i];
-                    let tx_data = map_of_txs.get(&signature).unwrap();
-                    metric.number_of_confirmed_txs += 1;
-                    drop(tx_data);
-                    map_of_txs.remove(&signature);
-                    confirmed_count += 1;
-                }
-            }
-        }
-    }
-
-    for _ in map_of_txs.iter() {
-        metric.number_of_unconfirmed_txs += 1;
-    }
-    metric
-}
-
 #[async_trait]
 impl TestingTask for SendAndConfrimTesting {
     async fn test(
         &self,
         args: crate::cli::Args,
         config: crate::config::Config,
-    ) -> anyhow::Result<()> {
-        println!("started sending and confirming memo transactions");
-        let rpc_client = Arc::new(RpcClient::new(args.rpc_addr.clone()));
-
-        let mut run_interval_ms = tokio::time::interval(Duration::from_secs(1));
-        let nb_runs = args.duration_in_seconds;
-        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);
-            let payer = payers.choose(&mut rng).unwrap();
-            let payer = Keypair::from_bytes(payer.to_bytes().as_slice()).unwrap();
-            let block_hash = self.block_hash.clone();
-            tasks.push(tokio::spawn(send_and_confirm_transactions(
-                rpc_client.clone(),
-                NB_MEMO_TRANSACTIONS_SENT_PER_SECOND,
-                payer,
-                seed,
-                block_hash.clone(),
-            )));
-            // wait for an interval
-            run_interval_ms.tick().await;
-        }
-
-        let instant = Instant::now();
-        let tasks_res = futures::future::join_all(tasks).await;
-        let duration = instant.elapsed();
-        let mut total_txs_confirmed = 0;
-        let mut total_txs_unconfirmed = 0;
-
-        for metric in tasks_res.into_iter().flatten() {
-            total_txs_confirmed += metric.number_of_confirmed_txs;
-            total_txs_unconfirmed += metric.number_of_unconfirmed_txs;
-        }
-
-        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(())
+    ) -> anyhow::Result<Stats> {
+        let instant = SendMemoTransactionsBench {
+            block_hash: self.block_hash.clone(),
+            payers: config
+                .users
+                .iter()
+                .map(|x| Arc::new(x.get_keypair()))
+                .collect(),
+        };
+        let metric = Bencher::bench::<SendMemoTransactionsBench>(instant, args).await?;
+        log::info!("{} {}", self.get_name(), serde_json::to_string(&metric)?);
+        Ok(metric)
     }
 
     fn get_name(&self) -> String {
         "Send and confirm memo transaction".to_string()
     }
 }
+
+#[derive(Clone)]
+struct SendMemoTransactionsBench {
+    block_hash: Arc<RwLock<Hash>>,
+    payers: Vec<Arc<Keypair>>,
+}
+
+#[async_trait::async_trait]
+impl Benchmark for SendMemoTransactionsBench {
+    async fn run(
+        self,
+        rpc_client: Arc<RpcClient>,
+        duration: std::time::Duration,
+        random_number: u64,
+    ) -> anyhow::Result<crate::bencher::Run> {
+        let mut result = Run::default();
+
+        let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(random_number);
+        let start = Instant::now();
+        while start.elapsed() < duration {
+            let msg: Vec<u8> = Alphanumeric.sample_iter(&mut rng).take(10).collect();
+            let payer = self.payers.choose(&mut rng).unwrap();
+
+            let blockhash = { *self.block_hash.read().await };
+            let tx = create_memo_tx(&msg, &payer, blockhash);
+            match rpc_client.send_transaction(&tx).await {
+                Ok(_) => {
+                    result.requests_completed += 1;
+                    result.bytes_received += 0;
+                }
+                Err(e) => {
+                    result.requests_failed += 1;
+                    result.errors.push(format!("{:?}", e.kind()));
+                }
+            }
+            result.bytes_sent += 0;
+        }
+        Ok(result)
+    }
+}

src/test_registry.rs

@@ -1,9 +1,12 @@
-use crate::{cli::Args, config::Config};
+use std::{collections::HashMap, sync::Arc};
+
+use crate::{bencher::Stats, cli::Args, config::Config};
 use async_trait::async_trait;
+use tokio::sync::RwLock;
 
 #[async_trait]
 pub trait TestingTask: Send + Sync {
-    async fn test(&self, args: Args, config: Config) -> anyhow::Result<()>;
+    async fn test(&self, args: Args, config: Config) -> anyhow::Result<Stats>;
     fn get_name(&self) -> String;
 }
 
@@ -18,21 +21,34 @@ impl TestRegistry {
     }
 
     pub async fn start_testing(self, args: Args, config: Config) {
+        let results = Arc::new(RwLock::new(HashMap::new()));
         let tasks = self.tests.into_iter().map(|test| {
             let args = args.clone();
             let config = config.clone();
             let name = test.get_name();
+            let results = results.clone();
 
             tokio::spawn(async move {
                 log::info!("test {name}");
 
                 match test.test(args, config).await {
-                    Ok(_) => log::info!("test {name} passed"),
+                    Ok(metric) => {
+                        log::info!("test {name} passed");
+                        let mut lock = results.write().await;
+                        lock.insert(test.get_name(), metric);
+                    }
                     Err(e) => log::info!("test {name} failed with error {e}"),
                 }
             })
         });
 
         futures::future::join_all(tasks).await;
+        let res = results.read().await.clone();
+        if !args.output_file.is_empty() {
+            let result_string = serde_json::to_string(&res);
+            if let Ok(result) = result_string {
+                std::fs::write(args.output_file, result).expect("Could not write output file");
+            }
+        }
     }
 }