//! 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`). //! //! `change_paths`: what knowing every received change's paths would cost, to judge it by //! rules by path (`future-work.md`, Write protection by checking changes): each //! change's operations decoded, and each object they name followed up to the root as //! the change left it (`store::changed_by`, so every node finds the same paths). Against //! receiving a sync message of many changes and loading a document of many changes, as //! above, each change writing a value three levels deep; and decoding alone, the part //! that grows with the change rather than with its objects. Decoded as the document //! rebuilds them, and from the bytes they came as; and inspected before they are //! applied (`inspect::Inspector`), their objects made among them or looked up in the //! document, each once. //! //! `screening`: what judging every change received costs while catching up //! (`access::screen`), against applying them unjudged, as a node whose document has no //! policy does now: many changes into an empty document, as one delta brings them; //! judged with no policy in them (each passes, read at its dependencies, after its //! paths were found), and with the policy their first change wrote (each judged by the //! rules). 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) -> Vec { 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 = (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 { 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, ) -> 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], 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 = 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> = 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, ); } } /// A document of `changes` changes, each writing `servers///cpu` (4 regions, /// 50 servers each), by `actors` actors in turn. fn nested(changes: usize, actors: usize) -> Document { let mut doc = Document::new(); let per = changes.div_ceil(actors.max(1)); for i in 0..changes { if i % per == 0 { doc.set_actor(automerge::ActorId::random()); } let path = [ "servers".to_owned(), format!("region{}", i % 4), format!("{}", (i / 4) % 50), "cpu".to_owned(), ]; transact(&mut doc, UnixMillis(1), &WRITER, |doc| { store::write_path(doc, &path, &serde_json::json!(i)) }) .expect("write"); } doc } /// Costs of knowing the paths of `doc`'s changes, besides `base`: decoding them as the /// document rebuilds them, and their paths so (`store::changed_by`); and decoding them /// from the bytes they came as, and their paths so (`store::changed_by_decoded`). fn path_costs(workload: &str, base: Duration, doc: &Document) { let hashes: Vec = doc.get_changes(&[]).iter().map(|c| c.hash()).collect(); let raw: Vec<(ChangeHash, Vec)> = doc .get_changes(&[]) .iter() .map(|change| (change.hash(), change.raw_bytes().to_vec())) .collect(); let decode_bytes = median(3, || { let started = Instant::now(); for (_, bytes) in &raw { std::hint::black_box( automerge::Change::from_bytes(bytes.clone()) .expect("change") .decode(), ); } started.elapsed() }); let paths_bytes = median(3, || { let started = Instant::now(); for (hash, bytes) in &raw { let change = automerge::Change::from_bytes(bytes.clone()) .expect("change") .decode(); std::hint::black_box(store::changed_by_decoded(doc, hash, change, InLists::Index)); } started.elapsed() }); let decode = median(3, || { let started = Instant::now(); for hash in &hashes { std::hint::black_box(doc.get_change_by_hash(hash).expect("change").decode()); } started.elapsed() }); let paths = median(3, || { let started = Instant::now(); for hash in &hashes { std::hint::black_box(store::changed_by(doc, hash, InLists::Index)); } started.elapsed() }); // Before they are applied (`inspect::Inspector`): those of a sync message to a node // that has none of them, their objects made among them; those of a document loaded, // looked up in it. let inspected = median(3, || { let empty = Document::new(); let started = Instant::now(); let changes: Vec = raw .iter() .map(|(_, bytes)| { automerge::Change::from_bytes(bytes.clone()) .expect("change") .decode() }) .collect(); std::hint::black_box(super::inspect::Inspector::new(&empty).paths(&changes)); started.elapsed() }); let looked_up = median(3, || { let started = Instant::now(); let changes: Vec = raw .iter() .map(|(_, bytes)| { automerge::Change::from_bytes(bytes.clone()) .expect("change") .decode() }) .collect(); std::hint::black_box(super::inspect::Inspector::new(doc).paths(&changes)); started.elapsed() }); report_signatures( workload, base, &[ ("decoding", decode), ("paths", paths), ("from bytes: decoding", decode_bytes), ("paths", paths_bytes), ("inspected, made among them", inspected), ("looked up", looked_up), ], ); } #[test] #[ignore = "a benchmark: cargo benchmarks"] fn bench_change_paths() { println!(); // Catching up: one sync message of that many changes received. for changes in [1_000, 10_000] { let sender = nested(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() }); path_costs( &format!("one sync message of {changes} changes"), receive, &sender, ); } // Joining, or starting from the file: a document of that many changes loaded. for changes in [10_000, 100_000] { let doc = nested(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() }); path_costs( &format!("a document of {changes} changes loaded"), load, &doc, ); } } /// `changes` changes on `doc`, each writing `servers///cpu`, by `actors` /// actors in turn, as `nested` does. fn nested_on(mut doc: Document, changes: usize, actors: usize) -> Document { let per = changes.div_ceil(actors.max(1)); for i in 0..changes { if i % per == 0 { doc.set_actor(automerge::ActorId::random()); } let path = [ "servers".to_owned(), format!("region{}", i % 4), format!("{}", (i / 4) % 50), "cpu".to_owned(), ]; transact(&mut doc, UnixMillis(1), &WRITER, |doc| { store::write_path(doc, &path, &serde_json::json!(i)) }) .expect("write"); } doc } /// The time to take `doc`'s changes into an empty document: unjudged, as Automerge /// applies them; and screened (`access::screen::apply`). fn screening_costs(workload: &str, doc: &Document) { let raw: Vec> = doc .get_changes(&[]) .iter() .map(|change| change.raw_bytes().to_vec()) .collect(); let changes = || -> Vec { raw.iter() .map(|bytes| automerge::Change::from_bytes(bytes.clone()).expect("change")) .collect() }; let unjudged = median(3, || { let changes = changes(); let mut empty = Document::new(); let started = Instant::now(); empty.apply_changes(changes).expect("apply"); started.elapsed() }); let screened = median(3, || { let changes = changes(); let mut empty = Document::new(); let started = Instant::now(); let screened = crate::access::screen::apply( &mut empty, Epoch::GENESIS, changes, &mut crate::access::screen::Waiting::default(), &mut crate::access::screen::Refusals::default(), None, &mut crate::document::PatchLog::inactive(), ) .expect("screen"); let elapsed = started.elapsed(); assert_eq!(screened.applied, raw.len(), "every change taken"); elapsed }); report_signatures( workload, unjudged, &[("screened", screened.saturating_sub(unjudged))], ); let changes = changes(); screening_parts(&changes, 400); // And with the history a document of that many has: whether a read at its heads // costs more there. if changes.len() > 10_000 { screening_parts(&changes, changes.len() - 1_000); } } /// Where screening `changes` spends its time, per change judged: decoding them, their /// operations placed (`inspect::Inspector`), and judging them, as screening does (what /// is read at the same heads read once, `rules::Reads`) and afresh (`rules::judge`); of /// that, reading the policy (`policy::of`) and, for scale, one read (the committer's /// member entry). As screening judges most: the first `warm` ones in (the maps they make, /// which flush the batch), the rest judged at the heads after them (`screen`'s frontier). fn screening_parts(changes: &[automerge::Change], warm: usize) { use crate::access::{policy, rules}; use crate::document::at::At; use crate::document::inspect::{Inspector, Op, Path}; use std::collections::{BTreeSet, HashMap}; let warm = warm.min(changes.len()); let mut base = Document::new(); base.apply_changes(changes[..warm].to_vec()).expect("apply"); let rest = &changes[warm..]; let seen = base.get_heads(); let mut authors: HashMap = HashMap::new(); let committers: Vec> = rest .iter() .map(|change| { let actor = change.actor_id(); let named = change .author() .or_else(|| base.get_author_for_actor(actor)) .and_then(|author| { std::str::from_utf8(author.as_bytes()) .ok() .map(str::to_owned) }) .or_else(|| authors.get(actor).cloned()); if let Some(named) = &named { authors.insert(actor.clone(), named.clone()); } named }) .collect(); let mut decoded = Vec::new(); let decode = median(3, || { let started = Instant::now(); decoded = rest.iter().map(automerge::Change::decode).collect(); started.elapsed() }); let mut ops: Vec> = Vec::new(); let place = median(3, || { let started = Instant::now(); ops = Inspector::new(&base).ops(&decoded); started.elapsed() }); let paths: Vec> = ops .iter() .map(|ops| ops.iter().map(|op| op.path.clone()).collect()) .collect(); let judge_all = |reads: &mut Option| { let started = Instant::now(); for (index, change) in rest.iter().enumerate() { let change = rules::Change { hash: change.hash(), deps: change.deps(), seen: &seen, committer: committers[index].as_deref(), user: None, paths: &paths[index], ops: &ops[index], }; let judged = match reads { Some(reads) => rules::judge_reading(&base, Epoch::GENESIS, &change, reads), None => rules::judge(&base, Epoch::GENESIS, &change), }; assert_eq!(judged, Ok(()), "every change taken"); } started.elapsed() }; let judge = median(3, || judge_all(&mut Some(rules::Reads::default()))); let afresh = median(3, || judge_all(&mut None)); let read_policy = median(3, || { let started = Instant::now(); for _ in rest { let at = At::new(&base, &seen).expect("heads"); std::hint::black_box(policy::of(&at)); } started.elapsed() }); let entry = crate::swarm::members::member_path(WRITER.node); let one_read = median(3, || { let started = Instant::now(); for _ in rest { let at = At::new(&base, &seen).expect("heads"); std::hint::black_box(store::read_path(&at, &entry).ok()); } started.elapsed() }); let per = |d: Duration| d.as_secs_f64() * 1e6 / rest.len().max(1) as f64; println!( "{:>40} per change: decode {:.2} µs, place {:.2} µs, judge {:.2} µs (afresh {:.2} µs: policy::of {:.2} µs, one read {:.2} µs)", format!("after {warm}"), per(decode), per(place), per(judge), per(afresh), per(read_policy), per(one_read), ); } #[test] #[ignore = "a benchmark: cargo benchmarks"] fn bench_screening() { println!(); for changes in [1_000, 10_000, 100_000] { screening_costs( &format!("{changes} changes, no policy"), &nested(changes, changes / 100), ); // The swarm, its policy and the writer's member entry first: every change after // is judged by the rules (no registry: the writer's node, a member, may). let mut start = Document::new(); transact(&mut start, UnixMillis(1), &WRITER, |doc| { let path = |text: &str| -> Vec { text.split('/').map(str::to_owned).collect() }; store::write_path(doc, &path("_continuum/swarm"), &serde_json::json!("bench"))?; store::write_path( doc, &path("_continuum/access/policy"), &serde_json::json!({"version": 1}), )?; store::write_path( doc, &crate::swarm::members::member_path(WRITER.node), &serde_json::json!({}), ) }) .expect("start"); screening_costs( &format!("{changes} changes, the policy first"), &nested_on(start, changes, changes / 100), ); } }