commonware-storage 2026.9.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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

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

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

/// Replay all items in the given `journal`.
async fn bench_run(
    journal: Journal<Context, FixedBytes<ITEM_SIZE>>,
    buffer: usize,
    read_options: ReadOptions,
) -> Journal<Context, FixedBytes<ITEM_SIZE>> {
    let (journal, reader) = journal.snapshot().await.unwrap();
    let stream = reader
        .replay(0, NZUsize!(buffer), read_options)
        .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}"),
        }
    }
    journal
}

/// Benchmark the replaying of items from a variable journal containing exactly that
/// number of items.
fn bench_variable_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());

        // Run DONT_CACHE first because it may not prune pages left resident by
        // cached replays.
        for (read_options, label) in [
            (ReadOptions::DONT_CACHE, "dont_cache"),
            (ReadOptions::default(), "cache"),
        ] {
            for buffer in [16_384, 65_536, 1_048_576] {
                c.bench_function(
                    &format!(
                        "{}/items={} buffer={} size={} read_options={}",
                        module_path!(),
                        items,
                        buffer,
                        ITEM_SIZE,
                        label
                    ),
                    |b| {
                        // Setup: populate journal (once, on first sample).
                        if !initialized {
                            Runner::new(cfg.clone()).start(|ctx| async move {
                                let j = get_variable_journal(ctx, PARTITION, ITEMS_PER_BLOB).await;
                                append_fixed_random_data::<_, ITEM_SIZE>(j, items).await;
                            });
                            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_variable_journal(
                                ctx.child("storage"),
                                PARTITION,
                                ITEMS_PER_BLOB,
                            )
                            .await;
                            let mut duration = Duration::ZERO;
                            for _ in 0..iters {
                                let start = Instant::now();
                                j = bench_run(j, buffer, read_options).await;
                                duration += start.elapsed();
                            }
                            duration
                        });
                    },
                );
            }
        }

        // Cleanup: destroy journal.
        if initialized {
            Runner::new(cfg).start(|context| async move {
                let j = get_variable_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_variable_replay
}