Compare commits
3 Commits
76cf386bd9
...
77e31fc504
Author | SHA1 | Date |
---|---|---|
|
77e31fc504 | |
|
7b4b869cf8 | |
|
caffe84b73 |
|
@ -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,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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,
|
||||
}
|
||||
|
|
|
@ -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));
|
||||
|
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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();
|
||||
|
|
|
@ -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();
|
||||
|
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue