commonware-storage 2026.7.0

Persist and retrieve data from an abstract store.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

use crate::{append_fixed_random_data, get_fixed_journal, ITEMS_PER_BLOB, ITEM_SIZE};
use commonware_runtime::{
    benchmarks::{context, tokio},
    tokio::{Config, Context, Runner},
    Runner as _, Supervisor as _,
};
use commonware_storage::journal::contiguous::{fixed::Journal, Contiguous as _};
use commonware_utils::{sequence::FixedBytes, NZUsize};
use criterion::{criterion_group, Criterion};
use futures::{pin_mut, StreamExt};
use std::{
    hint::black_box,
    time::{Duration, Instant},
};

/// Partition name to use in the journal config.
const PARTITION: &str = "test-partition";

/// Replay all items in the given `journal`.
async fn bench_run(journal: &mut Journal<Context, FixedBytes<ITEM_SIZE>>, buffer: usize) {
    let reader = journal.snapshot().await.unwrap();
    let stream = reader
        .replay(0, NZUsize!(buffer))
        .await
        .expect("failed to replay journal");
    pin_mut!(stream);
    while let Some(result) = stream.next().await {
        match result {
            Ok(item) => {
                black_box(item);
            }
            Err(err) => panic!("Failed to read item: {err}"),
        }
    }
}

/// Benchmark the replaying of items from a journal containing exactly that
/// number of items.
fn bench_fixed_replay(c: &mut Criterion) {
    for items in [1_000, 10_000, 100_000, 500_000] {
        let cfg = Config::default();
        let mut initialized = false;
        let runner = tokio::Runner::new(cfg.clone());
        for buffer in [16_384, 65_536, 1_048_576] {
            c.bench_function(
                &format!(
                    "{}/items={} buffer={} size={}",
                    module_path!(),
                    items,
                    buffer,
                    ITEM_SIZE
                ),
                |b| {
                    // Setup: populate journal (once, on first sample).
                    if !initialized {
                        Runner::new(cfg.clone()).start(|ctx| async move {
                            let mut j = get_fixed_journal(ctx, PARTITION, ITEMS_PER_BLOB).await;
                            append_fixed_random_data::<_, ITEM_SIZE>(&mut j, items).await;
                            j.sync().await.unwrap();
                        });
                        initialized = true;
                    }

                    // Benchmark: measure replay time.
                    b.to_async(&runner).iter_custom(|iters| async move {
                        let ctx = context::get::<commonware_runtime::tokio::Context>();
                        let mut j =
                            get_fixed_journal(ctx.child("storage"), PARTITION, ITEMS_PER_BLOB)
                                .await;
                        let mut duration = Duration::ZERO;
                        for _ in 0..iters {
                            let start = Instant::now();
                            bench_run(&mut j, buffer).await;
                            duration += start.elapsed();
                        }
                        duration
                    });
                },
            );
        }

        // Cleanup: destroy journal.
        if initialized {
            Runner::new(cfg).start(|context| async move {
                let j = get_fixed_journal::<ITEM_SIZE>(context, PARTITION, ITEMS_PER_BLOB).await;
                j.destroy().await.unwrap();
            });
        }
    }
}

criterion_group! {
    name = benches;
    config = Criterion::default().sample_size(10);
    targets = bench_fixed_replay
}