add document benchmarks

2 files changed, 172 insertions, 0 deletions

diff --git a/src/document/bench.rs b/src/document/bench.rs
new file mode 100644
index 0000000..78f87d8
--- /dev/null
+++ b/src/document/bench.rs
@@ -0,0 +1,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.
+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`.
+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);
+}
diff --git a/src/document/mod.rs b/src/document/mod.rs
index b9a8b6b..d7c2fcb 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -35,6 +35,8 @@ pub(crate) fn actor(doc: &Document) -> ActorId {
 }
 
 pub(crate) mod at;
+#[cfg(test)]
+mod bench;
 pub(crate) mod changes;
 pub(crate) mod file;
 pub(crate) mod lock;