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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
//! Benchmarks, as ignored tests (`cargo benchmarks`, `docs/development.md`): each times
//! what it measures in release builds and prints it; none asserts anything.
//!
//! `change_publishing`: what a change costs until the change publisher knows the paths
//! it touched (`swarm::changes`). Recorded as it is applied (`Replica::logged`, the
//! patch log), against diffing after it (`store::changed`, what the publisher did
//! before, and still does where the log doesn't cover it). Each workload is timed both
//! ways on the same input.
use std::time::{Duration, Instant};
use automerge::sync::{self, SyncDoc};
use automerge::transaction::Transactable;
use automerge::{ROOT, ReadDoc, ScalarValue};
use super::replica::{Replica, Writer, transact};
use super::store::{self, InLists};
use super::{ChangeHash, Document};
use crate::clock::UnixMillis;
use crate::document::replica::Epoch;
const WRITER: Writer = Writer {
node: "continuum:bench",
user: None,
};
fn sorted(mut heads: Vec<ChangeHash>) -> Vec<ChangeHash> {
heads.sort_unstable();
heads
}
/// A document with `keys` keys under `data`, each written by a change of its own.
pub(crate) fn document(keys: usize) -> Document {
let mut doc = Document::new();
let mut map = None;
for key in 0..keys {
transact(&mut doc, UnixMillis(1), &WRITER, |doc| {
let data = match &map {
Some(data) => data,
None => map.insert(doc.put_object(ROOT, "data", automerge::ObjType::Map)?),
};
doc.put(data, format!("k{key}"), ScalarValue::Int(key as i64))
})
.expect("write");
}
doc
}
/// The median of `runs` timings of `f`.
pub(crate) fn median(runs: usize, mut f: impl FnMut() -> Duration) -> Duration {
let mut times: Vec<Duration> = (0..runs).map(|_| f()).collect();
times.sort_unstable();
times[runs / 2]
}
fn report(workload: &str, logged: Duration, diffed: Duration) {
println!(
"{workload:<44} logged {:>10.3} ms diffed {:>10.3} ms ({:.2}×)",
logged.as_secs_f64() * 1e3,
diffed.as_secs_f64() * 1e3,
diffed.as_secs_f64() / logged.as_secs_f64().max(f64::MIN_POSITIVE),
);
}
/// `writes` local writes of one key each into a document of `keys` keys, the changed
/// paths taken after each (as the publisher does after every change it is told of).
fn local_writes(keys: usize, writes: usize) {
let base = document(keys);
let write = |doc: &mut automerge::transaction::Transaction<'_>, i: usize| {
let data = doc.get(ROOT, "data")?.map(|(_, id)| id).expect("data");
doc.put(
&data,
format!("k{}", i % keys),
ScalarValue::Int(-(i as i64)),
)
};
let logged = median(5, || {
let mut replica = Replica::new(Epoch::GENESIS, base.clone());
let started = Instant::now();
for i in 0..writes {
let before = sorted(replica.doc.get_heads());
replica
.transact(UnixMillis(2), &WRITER, |doc| write(doc, i))
.expect("write");
std::hint::black_box(replica.changed_since(&before, InLists::Index));
}
started.elapsed()
});
let diffed = median(5, || {
let mut doc = base.clone();
let started = Instant::now();
for i in 0..writes {
let before = sorted(doc.get_heads());
transact(&mut doc, UnixMillis(2), &WRITER, |doc| write(doc, i)).expect("write");
let after = sorted(doc.get_heads());
std::hint::black_box(store::changed(&doc, &before, &after, InLists::Index));
}
started.elapsed()
});
report(
&format!("{writes} local writes, {keys} keys"),
logged,
diffed,
);
}
/// One sync message carrying `changes` changes to a node that has none of them, and
/// the changed paths after it.
fn sync_message(changes: usize) {
let sender = document(changes);
// The receiver opens; the sender answers with everything it lacks.
let mut receiver_state = sync::State::new();
let opening = Document::new()
.generate_sync_message(&mut receiver_state)
.expect("an opening");
let mut sender_doc = sender.clone();
let mut sender_state = sync::State::new();
sender_doc
.receive_sync_message(&mut sender_state, opening)
.expect("opening");
let message = sender_doc
.generate_sync_message(&mut sender_state)
.expect("the changes")
.encode();
let logged = median(5, || {
let mut replica = Replica::new(Epoch::GENESIS, Document::new());
let mut state = receiver_state.clone();
let message = sync::Message::decode(&message).expect("decode");
let started = Instant::now();
let before = sorted(replica.doc.get_heads());
replica.logged(|doc, log| {
(
doc.receive_sync_message_log_patches(&mut state, message, log)
.expect("receive"),
true,
)
});
std::hint::black_box(replica.changed_since(&before, InLists::Index));
started.elapsed()
});
let diffed = median(5, || {
let mut doc = Document::new();
let mut state = receiver_state.clone();
let message = sync::Message::decode(&message).expect("decode");
let started = Instant::now();
let before = sorted(doc.get_heads());
doc.receive_sync_message(&mut state, message)
.expect("receive");
let after = sorted(doc.get_heads());
std::hint::black_box(store::changed(&doc, &before, &after, InLists::Index));
started.elapsed()
});
report(
&format!("one sync message of {changes} changes"),
logged,
diffed,
);
}
#[test]
#[ignore = "a benchmark: cargo benchmarks"]
fn bench_change_publishing() {
println!(""); // put reports on new line
local_writes(1_000, 1_000);
local_writes(100_000, 1_000);
sync_message(1_000);
sync_message(10_000);
}
|
