[commonware-runtime/commonware-storage benches] reuse large journal across benchmarks, fix Runner naming (#816)

roberto-bayardo · patrick-ogrady · web-flow · commit eaa0c03c3ed2 · 2025-04-24T13:24:34.000-07:00
Co-authored-by: Patrick O'Grady &lt;me@patrickogrady.xyz&gt;
diff --git a/runtime/src/benchmarks/tokio.rs b/runtime/src/benchmarks/tokio.rs
@@ -1,7 +1,7 @@
 //! Implements a [criterion]-compatible executor for the [tokio] runtime.
 
 use super::context;
-use crate::{tokio, Runner};
+use crate::{tokio, Runner as _};
 use criterion::async_executor::AsyncExecutor;
 use futures::Future;
 
@@ -15,7 +15,7 @@ use futures::Future;
 /// use std::time::Duration;
 ///
 /// fn my_benchmark(c: &mut Criterion) {
-///     let executor = tokio::Executor::default();
+///     let executor = tokio::Runner::default();
 ///     c.bench_function("sleep_benchmark", |b| {
 ///         b.to_async(&executor).iter_batched(|| (),
 ///         |_| async {
@@ -28,28 +28,28 @@ use futures::Future;
 /// }
 /// ```
 #[derive(Clone)]
-pub struct Executor {
+pub struct Runner {
     cfg: tokio::Config,
 }
 
-impl Executor {
+impl Runner {
     /// Create a new bencher with the given configuration
     pub fn new(cfg: tokio::Config) -> Self {
         Self { cfg }
     }
 }
 
-impl Default for Executor {
+impl Default for Runner {
     fn default() -> Self {
         Self::new(tokio::Config::default())
     }
 }
 
-impl AsyncExecutor for &Executor {
+impl AsyncExecutor for &Runner {
     fn block_on<T>(&self, future: impl Future<Output = T>) -> T {
-        let executor = tokio::Runner::new(self.cfg.clone());
+        let runner = tokio::Runner::new(self.cfg.clone());
 
-        let result = executor.start(|ctx| {
+        let result = runner.start(|ctx| {
             // Create and store our context
             context::set(ctx);
 
diff --git a/storage/src/journal/benches/bench.rs b/storage/src/journal/benches/bench.rs
@@ -16,21 +16,25 @@ criterion_main!(
     fixed_replay::benches,
 );
 
-/// Append `items_to_write` random items to a journal of items with ITEM_SIZE bytes each. The journal
-/// is configured to use `items_per_blob` items per blob.
-async fn append_random_data<const ITEM_SIZE: usize>(
+/// Open and return a temp journal with the given config parameters and items of size ITEM_SIZE.
+async fn get_journal<const ITEM_SIZE: usize>(
     context: Context,
     partition_name: &str,
     items_per_blob: u64,
-    items_to_write: u64,
 ) -> Journal<Context, FixedBytes<ITEM_SIZE>> {
     // Initialize the journal at the given partition.
     let journal_config = JConfig {
         partition: partition_name.to_string(),
         items_per_blob,
     };
-    let mut journal = Journal::init(context, journal_config).await.unwrap();
+    Journal::init(context, journal_config).await.unwrap()
+}
 
+/// Append `items_to_write` random items to the given journal, syncing the changes before returning.
+async fn append_random_data<const ITEM_SIZE: usize>(
+    journal: &mut Journal<Context, FixedBytes<ITEM_SIZE>>,
+    items_to_write: u64,
+) {
     // Append `items_to_write` random items to the journal.
     let mut rng = StdRng::seed_from_u64(0);
     let mut arr = [0; ITEM_SIZE];
@@ -44,6 +48,4 @@ async fn append_random_data<const ITEM_SIZE: usize>(
 
     // Sync the journal to ensure all data is written to disk.
     journal.sync().await.unwrap();
-
-    journal
 }
diff --git a/storage/src/journal/benches/fixed_append.rs b/storage/src/journal/benches/fixed_append.rs
@@ -1,4 +1,4 @@
-use crate::append_random_data;
+use crate::{append_random_data, get_journal};
 use commonware_runtime::benchmarks::{context, tokio};
 use criterion::{criterion_group, Criterion};
 use std::time::{Duration, Instant};
@@ -13,7 +13,7 @@ const ITEMS_PER_BLOB: u64 = 10_000;
 const ITEM_SIZE: usize = 32;
 
 fn bench_fixed_append(c: &mut Criterion) {
-    let executor = tokio::Executor::default();
+    let runner = tokio::Runner::default();
     for items_to_write in [1_000, 10_000, 100_000, 1_000_000] {
         c.bench_function(
             &format!(
@@ -23,18 +23,17 @@ fn bench_fixed_append(c: &mut Criterion) {
                 ITEM_SIZE
             ),
             |b| {
-                b.to_async(&executor).iter_custom(|iters| async move {
+                b.to_async(&runner).iter_custom(|iters| async move {
                     let ctx = context::get::<commonware_runtime::tokio::Context>();
                     let mut duration = Duration::ZERO;
                     for _ in 0..iters {
+                        // Create a new journal for each iteration
+                        let mut j =
+                            get_journal::<ITEM_SIZE>(ctx.clone(), PARTITION, ITEMS_PER_BLOB).await;
+
+                        // Append random data to the journal
                         let start = Instant::now();
-                        let j = append_random_data::<ITEM_SIZE>(
-                            ctx.clone(),
-                            PARTITION,
-                            ITEMS_PER_BLOB,
-                            items_to_write,
-                        )
-                        .await;
+                        append_random_data(&mut j, items_to_write).await;
                         duration += start.elapsed();
 
                         // Destroy the journal after appending to avoid polluting the next iteration
diff --git a/storage/src/journal/benches/fixed_read_random.rs b/storage/src/journal/benches/fixed_read_random.rs
@@ -1,7 +1,8 @@
-use super::append_random_data;
+use super::{append_random_data, get_journal};
 use commonware_runtime::{
     benchmarks::{context, tokio},
-    tokio::Context,
+    tokio::{Config, Context, Runner},
+    Runner as _,
 };
 use commonware_storage::journal::fixed::Journal;
 use commonware_utils::array::FixedBytes;
@@ -47,7 +48,21 @@ async fn bench_run_concurrent(
 }
 
 fn bench_fixed_read_random(c: &mut Criterion) {
-    let executor = tokio::Executor::default();
+    // Create a config we can use across all benchmarks (with a fixed `storage_directory`), allowing the
+    // same test file to be re-used.
+    let cfg = Config::default();
+
+    // Generate a large temp journal with random data.
+    let runner = Runner::new(cfg.clone());
+    runner.start(|ctx| async move {
+        // Create a large temp journal with random data.
+        let mut j = get_journal(ctx, PARTITION, ITEMS_PER_BLOB).await;
+        append_random_data::<ITEM_SIZE>(&mut j, ITEMS_TO_WRITE).await;
+        j.close().await.unwrap();
+    });
+
+    // Run the benchmarks
+    let runner = tokio::Runner::new(cfg.clone());
     for mode in ["serial", "concurrent"] {
         for items_to_read in [100, 1_000, 10_000, 100_000] {
             c.bench_function(
@@ -59,18 +74,9 @@ fn bench_fixed_read_random(c: &mut Criterion) {
                     ITEM_SIZE
                 ),
                 |b| {
-                    b.to_async(&executor).iter_custom(|iters| async move {
-                        // Append random data to the journal
+                    b.to_async(&runner).iter_custom(|iters| async move {
                         let ctx = context::get::<commonware_runtime::tokio::Context>();
-                        let j = append_random_data(
-                            ctx.clone(),
-                            PARTITION,
-                            ITEMS_PER_BLOB,
-                            ITEMS_TO_WRITE,
-                        )
-                        .await;
-
-                        // Run the benchmark
+                        let j = get_journal(ctx.clone(), PARTITION, ITEMS_PER_BLOB).await;
                         let mut duration = Duration::ZERO;
                         for _ in 0..iters {
                             let start = Instant::now();
@@ -81,16 +87,19 @@ fn bench_fixed_read_random(c: &mut Criterion) {
                             }
                             duration += start.elapsed();
                         }
-
-                        // Destroy the journal after reading to avoid polluting the next iteration
-                        j.destroy().await.unwrap();
-
                         duration
                     });
                 },
             );
         }
     }
+
+    // Clean up the temp journal
+    let runner = Runner::new(cfg);
+    runner.start(|context| async move {
+        let j = get_journal::<ITEM_SIZE>(context, PARTITION, ITEMS_PER_BLOB).await;
+        j.destroy().await.unwrap();
+    });
 }
 
 criterion_group! {
diff --git a/storage/src/journal/benches/fixed_read_sequential.rs b/storage/src/journal/benches/fixed_read_sequential.rs
@@ -1,4 +1,4 @@
-use super::append_random_data;
+use super::{append_random_data, get_journal};
 use commonware_runtime::{
     benchmarks::{context, tokio},
     tokio::Context,
@@ -27,21 +27,16 @@ async fn bench_run(journal: &Journal<Context, FixedBytes<ITEM_SIZE>>, items_to_r
 /// Benchmark the sequential read of items from a journal containing exactly that
 /// number of items.
 fn bench_fixed_read_sequential(c: &mut Criterion) {
-    let executor = tokio::Executor::default();
+    let runner = tokio::Runner::default();
     for items in [1_000, 10_000, 100_000, 500_000] {
         c.bench_function(
             &format!("{}/items={} size={}", module_path!(), items, ITEM_SIZE),
             |b| {
-                b.to_async(&executor).iter_custom(|iters| async move {
+                b.to_async(&runner).iter_custom(|iters| async move {
                     // Append random data to the journal
                     let ctx = context::get::<commonware_runtime::tokio::Context>();
-                    let j = append_random_data::<ITEM_SIZE>(
-                        ctx.clone(),
-                        PARTITION,
-                        ITEMS_PER_BLOB,
-                        items,
-                    )
-                    .await;
+                    let mut j = get_journal::<ITEM_SIZE>(ctx, PARTITION, ITEMS_PER_BLOB).await;
+                    append_random_data::<ITEM_SIZE>(&mut j, items).await;
                     let sz = j.size().await.unwrap();
                     assert_eq!(sz, items);
 
diff --git a/storage/src/journal/benches/fixed_replay.rs b/storage/src/journal/benches/fixed_replay.rs
@@ -1,4 +1,4 @@
-use super::append_random_data;
+use super::{append_random_data, get_journal};
 use commonware_runtime::{
     benchmarks::{context, tokio},
     tokio::Context,
@@ -20,7 +20,7 @@ const ITEM_SIZE: usize = 32;
 
 /// Replay all items in the given `journal`.
 async fn bench_run(journal: &mut Journal<Context, FixedBytes<ITEM_SIZE>>, items_to_read: u64) {
-    let concurrency = (items_to_read / ITEMS_PER_BLOB) as usize;
+    let concurrency = std::cmp::max(1, (items_to_read / ITEMS_PER_BLOB) as usize);
     let stream = journal
         .replay(concurrency)
         .await
@@ -39,21 +39,16 @@ async fn bench_run(journal: &mut Journal<Context, FixedBytes<ITEM_SIZE>>, items_
 /// Benchmark the replaying of items from a journal containing exactly that
 /// number of items.
 fn bench_fixed_replay(c: &mut Criterion) {
-    let executor = tokio::Executor::default();
+    let runner = tokio::Runner::default();
     for items in [1_000, 10_000, 100_000, 500_000] {
         c.bench_function(
             &format!("{}/items={} size={}", module_path!(), items, ITEM_SIZE),
             |b| {
-                b.to_async(&executor).iter_custom(|iters| async move {
+                b.to_async(&runner).iter_custom(|iters| async move {
                     // Append random data to the journal
                     let ctx = context::get::<commonware_runtime::tokio::Context>();
-                    let mut j = append_random_data::<ITEM_SIZE>(
-                        ctx.clone(),
-                        PARTITION,
-                        ITEMS_PER_BLOB,
-                        items,
-                    )
-                    .await;
+                    let mut j = get_journal::<ITEM_SIZE>(ctx, PARTITION, ITEMS_PER_BLOB).await;
+                    append_random_data::<ITEM_SIZE>(&mut j, items).await;
                     let sz = j.size().await.unwrap();
                     assert_eq!(sz, items);
 
diff --git a/storage/src/journal/fixed.rs b/storage/src/journal/fixed.rs
@@ -346,6 +346,7 @@ impl<E: Storage + Metrics, A: Array> Journal<E, A> {
         &mut self,
         concurrency: usize,
     ) -> Result<impl Stream<Item = Result<(u64, A), Error>> + '_, Error> {
+        assert!(concurrency > 0);
         // Collect all blobs to replay
         let mut blobs = Vec::with_capacity(self.blobs.len());
         let (newest_blob_index, _) = self.newest_blob();
