harmos-runtime 0.1.8

Harmos runtime: managed state, undo, recovery, background work and typed streams.
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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

/* ***
 *
 * Harmos
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * ***
 */
/* === Modules === */

/* === Re-Exports === */

/* === Imports === */

// Standard Library Imports
use std::time::{Duration, Instant, UNIX_EPOCH};

// Third Party Imports
use criterion::{BenchmarkId, Criterion, Throughput};
use harmos_runtime::storage::Query;
use harmos_runtime::storage::query::transform::{
    Transform,
    project::{Project, Projection},
};
use harmos_runtime::storage::query::{QueryArrow, RowSlice, Source, StreamSelection};

// Local Imports
use super::{fixture::Fixture, waveforms};

/* === Constants === */

/* === Macros === */

/* === Type Aliases === */

/* === Structs === */

/* === Enums === */

/* === Traits === */

/* === Implementations === */

/* === Public Functions === */

pub fn run(criterion: &mut Criterion, executor: &tokio::runtime::Runtime) {
    let mut group = criterion.benchmark_group("storage/query");
    group.sample_size(10);
    group.sampling_mode(criterion::SamplingMode::Flat);
    for periods in [1, 6] {
        let samples = waveforms::samples(periods);
        let inputs = waveforms::inputs(&samples);
        for concurrency in [1, 2, 4] {
            let fixture = executor.block_on(Fixture::new());
            executor.block_on(async {
                fixture.storage.write_schema(waveforms::schema(), waveforms::definition()).await.unwrap();
                for inputs in inputs.chunks(192) {
                    let batch = inputs.to_vec();
                    fixture.storage.write_stream(batch).await.unwrap();
                }
                fixture.storage.flush().await.unwrap();
            });
            group.throughput(Throughput::Elements(((samples.len() / 2 - 3) * concurrency) as u64));
            for metric in ["all-first-batches", "completion"] {
                group.bench_function(BenchmarkId::new(format!("{metric}/{concurrency}"), samples.len()), |bencher| {
                    bencher.iter_custom(|iterations| {
                        executor.block_on(async {
                            let mut elapsed = Duration::ZERO;
                            let mut resources = super::metrics::Observed::default();
                            for _ in 0..iterations {
                                let query = Query {
                                    at: fixture.storage.cursor_saved(),
                                    source: Source::Stream(StreamSelection {
                                        schema: waveforms::schema(),
                                        scope: None,
                                        item: None,
                                        stream: None,
                                        source_start: Some(UNIX_EPOCH),
                                        source_end: Some(
                                            UNIX_EPOCH + Duration::from_millis((periods * waveforms::SAMPLES_PER_PERIOD / 2) as u64),
                                        ),
                                        after: 0,
                                    }),
                                    transforms: vec![Transform::Project(Project {
                                        fields: vec![Projection {
                                            field: "payload.value".into(),
                                            name: "value".into(),
                                        }],
                                    })],
                                    slice: RowSlice { offset: 3, limit: None },
                                };
                                let probe = super::metrics::Probe::start(fixture.storage.clone());
                                let openings = (0..concurrency).map(|_| async {
                                    let mut reader = fixture.storage.query(query.clone()).await.unwrap();
                                    assert!(matches!(reader.next().await.unwrap(), Some(QueryArrow::Schema(_))));
                                    let Some(QueryArrow::Batch(first)) = reader.next().await.unwrap() else {
                                        panic!("nonempty query");
                                    };
                                    (reader, first)
                                });

                                let started = Instant::now();
                                let readers = futures_util::future::join_all(openings).await;
                                let first_elapsed = started.elapsed();
                                let completions = readers.into_iter().map(|(mut reader, first)| async move {
                                    let mut rows = first.as_record_batch().num_rows();
                                    drop(first);
                                    while let Some(output) = reader.next().await.unwrap() {
                                        if let QueryArrow::Batch(batch) = output {
                                            rows += batch.as_record_batch().num_rows();
                                        }
                                    }
                                    (rows, reader.completion().unwrap().rows)
                                });
                                let completions = futures_util::future::join_all(completions).await;
                                let completion = started.elapsed();
                                resources.merge(probe.finish());
                                elapsed += if metric == "all-first-batches" { first_elapsed } else { completion };
                                for (rows, completed) in completions {
                                    assert_eq!(rows, samples.len() / 2 - 3);
                                    assert_eq!(completed, rows as u64);
                                }
                            }
                            resources.report(&format!("query/{metric}"), samples.len(), concurrency, Some(&fixture.settings));
                            elapsed
                        })
                    });
                });
            }
            executor.block_on(fixture.finish());
        }
    }
    group.finish();
}
/* === Private Functions === */

/* === Tests === */