path: root/src/document/bench.rs

//! 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.
//!
//! `change_signatures`: what signing every change by its committer's node key would
//! cost (`future-work.md`, Write protection by checking changes): an Ed25519 signature
//! over a change's hash, made on every commit, checked on every change received.
//! Against what the node does anyway on the same input: a local write, a sync message
//! of many changes received (catching up), and a document of many changes loaded
//! (joining, or starting from the file). Checked change by change, and actor by actor:
//! only each actor's latest signed change, as a change's hash covers its dependencies
//! and an actor's change builds on its previous one, so one signature vouches for all
//! the actor's changes up to it (each of the documents' actors writes a hundredth of
//! their changes). And, for scale, the misplaced-author check the node makes on the
//! same input (`changes::misplaced_author`).

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
}

/// `count` changes to `doc`, each writing one key under `data` (from `first` on), made
/// at `at` and after.
pub(crate) fn writes(doc: &mut Document, first: usize, count: usize, at: i64) {
    for i in 0..count {
        let mut transaction = doc.transaction();
        let data = transaction
            .get(ROOT, "data")
            .expect("read")
            .map(|(_, id)| id)
            .unwrap_or_else(|| {
                transaction
                    .put_object(ROOT, "data", automerge::ObjType::Map)
                    .expect("map")
            });
        transaction
            .put(
                &data,
                format!("late{}", first + i),
                ScalarValue::Int(i as i64),
            )
            .expect("write");
        transaction
            .commit_with(automerge::transaction::CommitOptions::default().with_time(at + i as i64));
    }
}

/// A new document holding what `of` holds, as one change at time 0 (a snapshot).
pub(crate) fn snapshot(of: &Document) -> Document {
    let mut new = Document::new();
    let mut transaction = new.transaction();
    store::materialize_into(of, &mut transaction, &|_| false).expect("snapshot");
    transaction.commit_with(automerge::transaction::CommitOptions::default().with_time(0));
    new
}

/// 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);
}

/// What each signature would be made over: a domain, then the change's hash.
fn signed(hash: &ChangeHash) -> Vec<u8> {
    let mut message = b"continuum/change/1\0".to_vec();
    message.extend_from_slice(&hash.0);
    message
}

/// `doc`'s changes' signatures by `key`, by change.
fn signatures(doc: &Document, key: &crate::identity::NodeKey) -> Vec<(ChangeHash, [u8; 64])> {
    doc.get_changes(&[])
        .iter()
        .map(|change| (change.hash(), key.sign(&signed(&change.hash()))))
        .collect()
}

/// Checks every signature of `signatures` against `public`.
fn check(public: &[u8; 32], signatures: &[(ChangeHash, [u8; 64])]) {
    for (hash, signature) in signatures {
        assert!(crate::identity::verify(public, &signed(hash), signature));
    }
}

/// Checks `doc`'s changes actor by actor: each actor's latest change's signature, and
/// then every change against it (verified up to that seq); how many actors.
fn check_by_actor(
    doc: &Document,
    public: &[u8; 32],
    signatures: &std::collections::HashMap<ChangeHash, [u8; 64]>,
) -> usize {
    let metas = doc.get_changes_meta(&[]);
    let mut latest: std::collections::HashMap<&automerge::ActorId, (u64, ChangeHash)> =
        std::collections::HashMap::new();
    for meta in &metas {
        let entry = latest.entry(meta.actor.as_ref()).or_insert((0, meta.hash));
        if meta.seq >= entry.0 {
            *entry = (meta.seq, meta.hash);
        }
    }
    let mut verified_up_to = std::collections::HashMap::new();
    for (actor, (seq, hash)) in &latest {
        assert!(crate::identity::verify(
            public,
            &signed(hash),
            &signatures[hash]
        ));
        verified_up_to.insert(*actor, *seq);
    }
    let verified = metas
        .iter()
        .filter(|meta| meta.seq <= verified_up_to[meta.actor.as_ref()])
        .count();
    assert_eq!(verified, metas.len());
    latest.len()
}

/// As `document`, its changes made by `actors` actors in turn (as many by each).
fn document_by(keys: usize, actors: usize) -> Document {
    let mut doc = Document::new();
    let mut map = None;
    let per = keys.div_ceil(actors.max(1));
    for key in 0..keys {
        if key % per == 0 {
            doc.set_actor(automerge::ActorId::random());
        }
        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
}

fn report_signatures(workload: &str, base: Duration, costs: &[(&str, Duration)]) {
    let ms = |d: Duration| d.as_secs_f64() * 1e3;
    let mut line = format!("{workload:<40} {:>10.3} ms", ms(base));
    for (what, cost) in costs {
        line.push_str(&format!(
            "   {what} +{:.3} ms ({:+.0}%)",
            ms(*cost),
            100.0 * cost.as_secs_f64() / base.as_secs_f64().max(f64::MIN_POSITIVE)
        ));
    }
    println!("{line}");
}

/// Costs of checking `doc`'s changes as they come, besides `base`: the signatures change
/// by change and actor by actor, and the misplaced-author check on the bytes they came
/// as (`raw`).
fn signature_costs(
    workload: &str,
    base: Duration,
    doc: &Document,
    raw: &[Vec<u8>],
    key: &crate::identity::NodeKey,
) {
    let public = key.public();
    let signatures = signatures(doc, key);
    let each = median(5, || {
        let started = Instant::now();
        check(&public, &signatures);
        started.elapsed()
    });
    let table: std::collections::HashMap<ChangeHash, [u8; 64]> =
        signatures.iter().copied().collect();
    let mut actors = 0;
    let by_actor = median(5, || {
        let started = Instant::now();
        actors = check_by_actor(doc, &public, &table);
        started.elapsed()
    });
    let misplaced = median(5, || {
        let started = Instant::now();
        for bytes in raw {
            std::hint::black_box(super::changes::misplaced_author(bytes));
        }
        started.elapsed()
    });
    report_signatures(
        workload,
        base,
        &[
            ("signatures", each),
            (&format!("by actor ({actors})"), by_actor),
            ("misplaced authors", misplaced),
        ],
    );
}

#[test]
#[ignore = "a benchmark: cargo benchmarks"]
fn bench_change_signatures() {
    println!();
    let key = crate::identity::NodeKey::from_seed([7; 32]);
    // A local write, and signing it.
    let mut doc = document(1_000);
    let writes = 1_000;
    let started = Instant::now();
    for i in 0..writes {
        transact(&mut doc, UnixMillis(2), &WRITER, |doc| {
            let data = doc.get(ROOT, "data")?.map(|(_, id)| id).expect("data");
            doc.put(&data, format!("k{}", i % 1_000), ScalarValue::Int(-i))
        })
        .expect("write");
    }
    let write = started.elapsed() / writes as u32;
    let heads = doc.get_heads();
    let started = Instant::now();
    for _ in 0..writes {
        std::hint::black_box(key.sign(&signed(&heads[0])));
    }
    let sign = started.elapsed() / writes as u32;
    report_signatures("a local write (each)", write, &[("signing it", sign)]);

    // Catching up: one sync message of that many changes received.
    for changes in [1_000, 10_000] {
        let sender = document_by(changes, changes / 100);
        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 receive = 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();
            doc.receive_sync_message(&mut state, message)
                .expect("receive");
            started.elapsed()
        });
        // Checked change by change, as a sync message's are.
        let raw: Vec<Vec<u8>> = sender
            .get_changes(&[])
            .iter()
            .map(|change| change.raw_bytes().to_vec())
            .collect();
        signature_costs(
            &format!("one sync message of {changes} changes"),
            receive,
            &sender,
            &raw,
            &key,
        );
    }

    // Joining, or starting from the file: a document of that many changes loaded.
    for changes in [10_000, 100_000] {
        let doc = document_by(changes, changes / 100);
        let bytes = super::save(&doc);
        let load = median(3, || {
            let started = Instant::now();
            std::hint::black_box(super::load(&bytes).expect("load"));
            started.elapsed()
        });
        // Checked as the whole document, as a file or a download is.
        signature_costs(
            &format!("a document of {changes} changes loaded"),
            load,
            &doc,
            std::slice::from_ref(&bytes),
            &key,
        );
    }
}