-
Notifications
You must be signed in to change notification settings - Fork 68
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add a program to submit random transactions #907
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,161 @@ | ||
//! Utility program to submit random transactions to an Espresso Sequencer. | ||
|
||
use async_compatibility_layer::logging::{setup_backtrace, setup_logging}; | ||
use async_std::task::{sleep, spawn}; | ||
use bytesize::ByteSize; | ||
use clap::Parser; | ||
use commit::{Commitment, Committable}; | ||
use derive_more::From; | ||
use futures::{ | ||
channel::mpsc::{self, Sender}, | ||
sink::SinkExt, | ||
stream::StreamExt, | ||
}; | ||
use hotshot_query_service::{ | ||
availability::{BlockQueryData, QueryablePayload}, | ||
Error, | ||
}; | ||
use rand::{thread_rng, Rng, RngCore}; | ||
use sequencer::{options::parse_duration, SeqTypes, Transaction}; | ||
use snafu::Snafu; | ||
use std::{collections::HashSet, time::Duration}; | ||
use surf_disco::{Client, Url}; | ||
|
||
/// Submit random transactions to an Espresso Sequencer. | ||
#[derive(Clone, Debug, Parser)] | ||
struct Options { | ||
/// Minimum size of transaction to submit. | ||
#[clap(long, default_value = "1", value_parser = parse_size)] | ||
min_size: usize, | ||
|
||
/// Maximum size of transaction to submit. | ||
#[clap(long, default_value = "1kb", value_parser = parse_size)] | ||
max_size: usize, | ||
|
||
/// Optional delay between submitting transactions. | ||
/// | ||
/// Can be used to moderate the rate of submission. | ||
#[clap(long, value_parser = parse_duration)] | ||
delay: Option<Duration>, | ||
|
||
/// Maximum number of unprocessed transaction submissions. | ||
/// | ||
/// This can be used to apply backpressure so that the tasks submitting transactions do not get | ||
/// too far ahead of the task processing results. | ||
#[clap(long, default_value = "1000")] | ||
channel_bound: usize, | ||
|
||
/// Number of parallel tasks to run. | ||
#[clap(short, long, default_value = "1")] | ||
jobs: usize, | ||
|
||
/// URL of the query service. | ||
url: Url, | ||
} | ||
|
||
#[derive(Clone, Debug, From, Snafu)] | ||
struct ParseSizeError { | ||
msg: String, | ||
} | ||
|
||
fn parse_size(s: &str) -> Result<usize, ParseSizeError> { | ||
Ok(s.parse::<ByteSize>()?.0 as usize) | ||
} | ||
|
||
#[async_std::main] | ||
async fn main() { | ||
setup_backtrace(); | ||
setup_logging(); | ||
|
||
let opt = Options::parse(); | ||
let (sender, mut receiver) = mpsc::channel(opt.channel_bound); | ||
|
||
// Subscribe to block stream so we can check that our transactions are getting sequenced. | ||
let client = Client::<Error>::new(opt.url.clone()); | ||
let block_height: usize = client | ||
.get("status/latest_block_height") | ||
.send() | ||
.await | ||
.unwrap(); | ||
let mut blocks = client | ||
.socket(&format!("availability/stream/blocks/{block_height}")) | ||
.subscribe() | ||
.await | ||
.unwrap(); | ||
tracing::info!("listening for blocks starting at {block_height}"); | ||
|
||
// Spawn tasks to submit transactions. | ||
for _ in 0..opt.jobs { | ||
spawn(submit_transactions(opt.clone(), sender.clone())); | ||
} | ||
|
||
// Keep track of the results. | ||
let mut pending = HashSet::new(); | ||
while let Some(block) = blocks.next().await { | ||
let block: BlockQueryData<SeqTypes> = match block { | ||
Ok(block) => block, | ||
Err(err) => { | ||
tracing::warn!("error getting block: {err}"); | ||
continue; | ||
} | ||
}; | ||
tracing::info!("got block {}", block.height()); | ||
|
||
// Get all transactions which were submitted before this block. | ||
while let Ok(Some(tx)) = receiver.try_next() { | ||
pending.insert(tx); | ||
} | ||
|
||
// Clear pending transactions from the block. | ||
for (_, tx) in block.payload().enumerate() { | ||
if pending.remove(&tx.commit()) { | ||
tracing::debug!("got transaction {}", tx.commit()); | ||
} | ||
} | ||
|
||
tracing::info!("{} transactions still pending", pending.len()); | ||
} | ||
tracing::info!( | ||
"block stream ended with {} transactions still pending", | ||
pending.len() | ||
); | ||
} | ||
|
||
async fn submit_transactions(opt: Options, mut sender: Sender<Commitment<Transaction>>) { | ||
let client = Client::<Error>::new(opt.url); | ||
loop { | ||
let tx = random_transaction(opt.min_size, opt.max_size); | ||
let hash = tx.commit(); | ||
tracing::debug!( | ||
"submitting transaction {hash} for namespace {} of size {}", | ||
tx.vm(), | ||
tx.payload().len() | ||
); | ||
if let Err(err) = client | ||
.post::<()>("submit/submit") | ||
.body_binary(&tx) | ||
.unwrap() | ||
.send() | ||
.await | ||
{ | ||
tracing::error!("failed to submit transaction: {err}"); | ||
} | ||
sender.send(hash).await.ok(); | ||
|
||
if let Some(delay) = opt.delay { | ||
sleep(delay).await; | ||
} | ||
} | ||
} | ||
|
||
fn random_transaction(min_size: usize, max_size: usize) -> Transaction { | ||
let mut rng = thread_rng(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. How does this work with multithreading? It might be nice to allow passing in a seed to make the generated transactions deterministic, but we can also add that later. edit: first question answered here: https://rust-random.github.io/book/guide-parallel.html |
||
|
||
let vm = rng.next_u64(); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Wondering if we want to narrow this down to a few VM IDs instead of sampling from the whole range which means we will probably never use the same VM ID twice. |
||
|
||
let len = rng.gen_range(min_size..=max_size); | ||
let mut payload = vec![0; len]; | ||
rng.fill_bytes(&mut payload); | ||
|
||
Transaction::new(vm.into(), payload) | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This doesn't have the clap derive. What does that do?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Makes it a (required) positional argument