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