Function run_benchmark

src/persist-cli/src/open_loop.rs:169–502

(
    args: Args,
    metrics_registry: MetricsRegistry,
    metrics: Arc<Metrics>,
    writers: Vec<W>,
    readers: Vec<R>,
)

}

async fn run_benchmark<W, R>(
    args: Args,
    metrics_registry: MetricsRegistry,
    metrics: Arc<Metrics>,
    writers: Vec<W>,
    readers: Vec<R>,
) -> Result<(), anyhow::Error>
where
    W: BenchmarkWriter + Send + Sync + 'static,
    R: BenchmarkReader + Send + Sync + 'static,
{
    let num_records_total = args.records_per_second * usize::cast_from(args.runtime.as_secs());
    let data_generator =
        DataGenerator::new(num_records_total, args.record_size_bytes, args.batch_size);

    let benchmark_description = format!(
        "num-readers={} num-writers={} runtime={:?} num_records_total={} records-per-second={} record-size-bytes={} batch-size={}",
        args.num_readers,
        args.num_writers,
        args.runtime,
        num_records_total,
        args.records_per_second,
        args.record_size_bytes,
        args.batch_size
    );

    info!("starting benchmark: {}", benchmark_description);
    let mut generator_handles: Vec<JoinHandle<Result<String, anyhow::Error>>> = vec![];
    let mut write_handles: Vec<JoinHandle<Result<String, anyhow::Error>>> = vec![];
    let mut read_handles: Vec<JoinHandle<Result<(String, R), anyhow::Error>>> = vec![];

    // All workers should have the starting time (so they can consistently track progress
    // and reason about lag independently).
    let start = Instant::now();
    // Use a barrier to start all threads at the same time. We need 2x the number of
    // writers because we start 2 distinct tasks per writer.
    let barrier = Arc::new(Barrier::new(2 * args.num_writers + args.num_readers));

    // The batch interarrival time. We'll use this quantity to rate limit the
    // data generation.
    // No other known way to convert `usize` to `f64`.
    #[allow(clippy::as_conversions)]
    let time_per_batch = {
        let records_per_second_f64 = args.records_per_second as f64;
        let batch_size_f64 = args.batch_size as f64;

        let batches_per_second = records_per_second_f64 / batch_size_f64;
        Duration::from_secs(1).div_f64(batches_per_second)
    };

    for (idx, mut writer) in writers.into_iter().enumerate() {
        let b = Arc::clone(&barrier);
        let data_generator = data_generator.clone();
        let start = start.clone();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

        // Intentionally create the span outside the task to set the parent.
        let generator_span = info_span!("generator", idx);