[storage/archive] Add Benchmarks (#845)

Author: patrick-ogrady

storage/Cargo.toml

@@ -45,6 +45,11 @@ name="bmt"
 harness = false
 path = "src/bmt/benches/bench.rs"
 
+[[bench]]
+name="index"
+harness = false
+path = "src/index/benches/bench.rs"
+
 [[bench]]
 name="journal"
 harness = false

storage/src/archive/benches/bench.rs

@@ -1,11 +1,8 @@
 use criterion::criterion_main;
 
-mod hashmap_insert;
-mod hashmap_insert_fixed;
-mod hashmap_iteration;
+mod get;
+mod put;
+mod restart;
+mod utils;
 
-criterion_main!(
-    hashmap_iteration::benches,
-    hashmap_insert_fixed::benches,
-    hashmap_insert::benches,
-);
+criterion_main!(put::benches, get::benches, restart::benches);

storage/src/archive/benches/get.rs

@@ -0,0 +1,153 @@
+//! Random key-lookup benchmark for Archive.
+
+use super::utils::{append_random, get_archive, ArchiveType, Key};
+use commonware_runtime::{
+    benchmarks::{context, tokio},
+    tokio::Config,
+    Runner,
+};
+use commonware_storage::archive::Identifier;
+use criterion::{black_box, criterion_group, Criterion};
+use futures::future::try_join_all;
+use rand::{rngs::StdRng, Rng, SeedableRng};
+use std::time::Instant;
+
+/// Items pre-loaded into the archive.
+const ITEMS: u64 = 1_000_000;
+
+fn select_keys(keys: &[Key], reads: usize) -> Vec<Key> {
+    let mut rng = StdRng::seed_from_u64(42);
+    let mut selected_keys = Vec::with_capacity(reads);
+    for _ in 0..reads {
+        selected_keys.push(keys[rng.gen_range(0..ITEMS as usize)].clone());
+    }
+    selected_keys
+}
+
+fn select_indices(reads: usize) -> Vec<u64> {
+    let mut rng = StdRng::seed_from_u64(42);
+    let mut selected_indices = Vec::with_capacity(reads);
+    for _ in 0..reads {
+        selected_indices.push(rng.gen_range(0..ITEMS));
+    }
+    selected_indices
+}
+
+async fn read_serial_keys(a: &ArchiveType, reads: &[Key]) {
+    for k in reads {
+        black_box(a.get(Identifier::Key(k)).await.unwrap().unwrap());
+    }
+}
+
+async fn read_serial_indices(a: &ArchiveType, indices: &[u64]) {
+    for idx in indices {
+        black_box(a.get(Identifier::Index(*idx)).await.unwrap().unwrap());
+    }
+}
+
+async fn read_concurrent_keys(a: &ArchiveType, reads: Vec<Key>) {
+    let futures = reads.iter().map(|k| a.get(Identifier::Key(k)));
+    black_box(try_join_all(futures).await.unwrap());
+}
+
+async fn read_concurrent_indices(a: &ArchiveType, indices: &[u64]) {
+    let mut futs = Vec::with_capacity(indices.len());
+    for idx in indices {
+        futs.push(a.get(Identifier::Index(*idx)));
+    }
+    black_box(try_join_all(futs).await.unwrap());
+}
+
+fn bench_get(c: &mut Criterion) {
+    // Create a config we can use across all benchmarks (with a fixed `storage_directory`).
+    let cfg = Config::default();
+    for compression in [None, Some(3)] {
+        // Create a shared on-disk archive once so later setup is fast.
+        let builder = commonware_runtime::tokio::Runner::new(cfg.clone());
+        let keys = builder.start(|ctx| async move {
+            let mut a = get_archive(ctx, compression).await;
+            let keys = append_random(&mut a, ITEMS).await;
+            a.close().await.unwrap();
+            keys
+        });
+
+        // Run the benchmarks.
+        let runner = tokio::Runner::new(cfg.clone());
+        for mode in ["serial", "concurrent"] {
+            for pattern in ["key", "index"] {
+                for reads in [1_000, 10_000, 100_000] {
+                    let label = format!(
+                        "{}/mode={} pattern={} comp={} reads={}",
+                        module_path!(),
+                        mode,
+                        pattern,
+                        compression
+                            .map(|l| l.to_string())
+                            .unwrap_or_else(|| "off".into()),
+                        reads
+                    );
+                    c.bench_function(&label, |b| {
+                        let keys = keys.clone();
+                        b.to_async(&runner).iter_custom(move |iters| {
+                            let keys = keys.clone();
+                            async move {
+                                let ctx = context::get::<commonware_runtime::tokio::Context>();
+                                let archive = get_archive(ctx, compression).await;
+                                if pattern == "key" {
+                                    let selected_keys = select_keys(&keys, reads);
+                                    let start = Instant::now();
+                                    for _ in 0..iters {
+                                        match mode {
+                                            "serial" => {
+                                                read_serial_keys(&archive, &selected_keys).await
+                                            }
+                                            "concurrent" => {
+                                                read_concurrent_keys(
+                                                    &archive,
+                                                    selected_keys.clone(),
+                                                )
+                                                .await
+                                            }
+                                            _ => unreachable!(),
+                                        }
+                                    }
+                                    start.elapsed()
+                                } else {
+                                    let selected_indices = select_indices(reads);
+                                    let start = Instant::now();
+                                    for _ in 0..iters {
+                                        match mode {
+                                            "serial" => {
+                                                read_serial_indices(&archive, &selected_indices)
+                                                    .await
+                                            }
+                                            "concurrent" => {
+                                                read_concurrent_indices(&archive, &selected_indices)
+                                                    .await
+                                            }
+                                            _ => unreachable!(),
+                                        }
+                                    }
+                                    start.elapsed()
+                                }
+                            }
+                        });
+                    });
+                }
+            }
+        }
+
+        // Clean up shared artifacts.
+        let cleaner = commonware_runtime::tokio::Runner::new(cfg.clone());
+        cleaner.start(|ctx| async move {
+            let a = get_archive(ctx, compression).await;
+            a.destroy().await.unwrap();
+        });
+    }
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default().sample_size(10);
+    targets = bench_get
+}

storage/src/archive/benches/put.rs

@@ -0,0 +1,42 @@
+use super::utils::{append_random, get_archive};
+use commonware_runtime::benchmarks::{context, tokio};
+use criterion::{criterion_group, Criterion};
+use std::time::{Duration, Instant};
+
+fn bench_put(c: &mut Criterion) {
+    let runner = tokio::Runner::default();
+    for compression in [None, Some(3)] {
+        for items in [10_000, 100_000, 250_000] {
+            let label = format!(
+                "{}/items={} comp={}",
+                module_path!(),
+                items,
+                compression
+                    .map(|l| l.to_string())
+                    .unwrap_or_else(|| "off".into()),
+            );
+            c.bench_function(&label, |b| {
+                b.to_async(&runner).iter_custom(move |iters| async move {
+                    let ctx = context::get::<commonware_runtime::tokio::Context>();
+                    let mut total = Duration::ZERO;
+                    for _ in 0..iters {
+                        let mut archive = get_archive(ctx.clone(), compression).await;
+
+                        let start = Instant::now();
+                        append_random(&mut archive, items).await;
+                        total += start.elapsed();
+
+                        archive.destroy().await.unwrap();
+                    }
+                    total
+                });
+            });
+        }
+    }
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default().sample_size(10);
+    targets = bench_put
+}

storage/src/archive/benches/restart.rs

@@ -0,0 +1,62 @@
+use super::utils::{append_random, get_archive};
+use commonware_runtime::{
+    benchmarks::{context, tokio},
+    tokio::Config,
+    Runner,
+};
+use criterion::{criterion_group, Criterion};
+use std::time::{Duration, Instant};
+
+fn bench_restart(c: &mut Criterion) {
+    // Create a config we can use across all benchmarks (with a fixed `storage_directory`).
+    let cfg = Config::default();
+    for compression in [None, Some(3)] {
+        for items in [10_000, 100_000, 250_000] {
+            let builder = commonware_runtime::tokio::Runner::new(cfg.clone());
+            builder.start(|ctx| async move {
+                let mut a = get_archive(ctx, compression).await;
+                append_random(&mut a, items).await;
+                a.close().await.unwrap();
+            });
+
+            // Run the benchmarks
+            let runner = tokio::Runner::new(cfg.clone());
+            c.bench_function(
+                &format!(
+                    "{}/items={} comp={}",
+                    module_path!(),
+                    items,
+                    compression
+                        .map(|l| l.to_string())
+                        .unwrap_or_else(|| "off".into())
+                ),
+                |b| {
+                    b.to_async(&runner).iter_custom(|iters| async move {
+                        let ctx = context::get::<commonware_runtime::tokio::Context>();
+                        let mut total = Duration::ZERO;
+                        for _ in 0..iters {
+                            let start = Instant::now();
+                            let a = get_archive(ctx.clone(), compression).await; // replay happens inside init
+                            total += start.elapsed();
+                            a.close().await.unwrap();
+                        }
+                        total
+                    });
+                },
+            );
+
+            // Tear down
+            let cleaner = commonware_runtime::tokio::Runner::new(cfg.clone());
+            cleaner.start(|ctx| async move {
+                let a = get_archive(ctx, compression).await;
+                a.destroy().await.unwrap();
+            });
+        }
+    }
+}
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default().sample_size(10);
+    targets = bench_restart
+}

storage/src/archive/benches/utils.rs

@@ -0,0 +1,59 @@
+//! Helpers shared by the Archive benchmarks.
+
+use commonware_runtime::tokio::Context;
+use commonware_storage::{
+    archive::{Archive, Config},
+    index::translator::TwoCap,
+};
+use commonware_utils::array::FixedBytes;
+use rand::{rngs::StdRng, RngCore, SeedableRng};
+
+/// Partition used across all archive benchmarks.
+pub const PARTITION: &str = "archive_bench_partition";
+
+/// Number of buffered writes before a forced sync.
+const PENDING_WRITES: usize = 1_000;
+
+/// Section-mask that yields reasonably small blobs for local testing.
+const SECTION_MASK: u64 = 0xffff_ffff_ffff_ff00u64;
+
+/// Fixed-length key and value types.
+pub type Key = FixedBytes<64>;
+pub type Val = FixedBytes<32>;
+
+/// Concrete archive type reused by every benchmark.
+pub type ArchiveType = Archive<TwoCap, Context, Key, (), Val>;
+
+/// Open (or create) a fresh archive with optional compression.
+///
+/// The caller is responsible for closing or destroying it.
+pub async fn get_archive(ctx: Context, compression: Option<u8>) -> ArchiveType {
+    let cfg = Config {
+        partition: PARTITION.into(),
+        translator: TwoCap,
+        compression,
+        codec_config: (),
+        section_mask: SECTION_MASK,
+        pending_writes: PENDING_WRITES,
+        replay_concurrency: 1,
+    };
+    Archive::init(ctx, cfg).await.unwrap()
+}
+
+/// Append `count` random (index,key,value) triples and sync once.
+pub async fn append_random(archive: &mut ArchiveType, count: u64) -> Vec<Key> {
+    let mut rng = StdRng::seed_from_u64(0);
+    let mut key_buf = [0u8; 64];
+    let mut val_buf = [0u8; 32];
+
+    let mut keys = Vec::with_capacity(count as usize);
+    for i in 0..count {
+        rng.fill_bytes(&mut key_buf);
+        let key = Key::new(key_buf);
+        keys.push(key.clone());
+        rng.fill_bytes(&mut val_buf);
+        archive.put(i, key, Val::new(val_buf)).await.unwrap();
+    }
+    archive.sync().await.unwrap();
+    keys
+}

storage/src/archive/storage.rs

@@ -417,4 +417,9 @@ impl<T: Translator, E: Storage + Metrics, K: Array, VC: CodecConfig + Copy, V: C
     pub async fn close(self) -> Result<(), Error> {
         self.journal.close().await.map_err(Error::Journal)
     }
+
+    /// Remove all on-disk data created by this `Archive`.
+    pub async fn destroy(self) -> Result<(), Error> {
+        self.journal.destroy().await.map_err(Error::Journal)
+    }
 }

storage/src/index/benches/bench.rs

@@ -0,0 +1,11 @@
+use criterion::criterion_main;
+
+mod hashmap_insert;
+mod hashmap_insert_fixed;
+mod hashmap_iteration;
+
+criterion_main!(
+    hashmap_iteration::benches,
+    hashmap_insert_fixed::benches,
+    hashmap_insert::benches,
+);