Implement benchmarks

car-mirror-benches/Cargo.toml

@@ -7,10 +7,13 @@ authors = ["Stephen Akinyemi <[email protected]>"]
 
 [dependencies]
 car-mirror = { path = "../car-mirror", version = "0.1", features = ["test_utils"] }
+wnfs-common = "0.1.23"
+async-std = { version = "1.11", features = ["attributes"] }
+anyhow = "1.0"
 
 [dev-dependencies]
 criterion = { version = "0.4", default-features = false }
 
 [[bench]]
-name = "a_benchmark"
+name = "in_memory"
 harness = false

car-mirror-benches/benches/in_memory.rs

@@ -0,0 +1,88 @@
+use car_mirror::{
+    common::Config,
+    pull, push,
+    test_utils::{arb_ipld_dag, links_to_padded_ipld, setup_blockstore},
+};
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+use wnfs_common::MemoryBlockStore;
+
+pub fn push(c: &mut Criterion) {
+    let mut rvg = car_mirror::test_utils::Rvg::deterministic();
+
+    c.bench_function("push cold", |b| {
+        b.iter_batched(
+            || {
+                let (blocks, root) = rvg.sample(&arb_ipld_dag(
+                    250..256,
+                    0.9, // Very highly connected
+                    links_to_padded_ipld(10 * 1024),
+                ));
+                let store = async_std::task::block_on(setup_blockstore(blocks)).unwrap();
+                (store, root)
+            },
+            |(ref client_store, root)| {
+                let server_store = &MemoryBlockStore::new();
+                let config = &Config::default();
+
+                // Simulate a multi-round protocol run in-memory
+                async_std::task::block_on(async move {
+                    let mut request = push::request(root, None, config, client_store).await?;
+                    loop {
+                        let response = push::response(root, request, config, server_store).await?;
+
+                        if response.indicates_finished() {
+                            break;
+                        }
+                        request = push::request(root, Some(response), config, client_store).await?;
+                    }
+
+                    Ok::<(), anyhow::Error>(())
+                })
+                .unwrap();
+            },
+            BatchSize::LargeInput,
+        )
+    });
+}
+
+pub fn pull(c: &mut Criterion) {
+    let mut rvg = car_mirror::test_utils::Rvg::deterministic();
+
+    c.bench_function("pull cold", |b| {
+        b.iter_batched(
+            || {
+                let (blocks, root) = rvg.sample(&arb_ipld_dag(
+                    250..256,
+                    0.9,                             // Very highly connected
+                    links_to_padded_ipld(10 * 1024), // 10KiB random data per block
+                ));
+                let store = async_std::task::block_on(setup_blockstore(blocks)).unwrap();
+                (store, root)
+            },
+            |(ref server_store, root)| {
+                let client_store = &MemoryBlockStore::new();
+                let config = &Config::default();
+
+                // Simulate a multi-round protocol run in-memory
+                async_std::task::block_on(async move {
+                    let mut request = pull::request(root, None, config, client_store).await?;
+                    loop {
+                        let response = pull::response(root, request, config, server_store).await?;
+                        request = pull::request(root, Some(response), config, client_store).await?;
+
+                        if request.indicates_finished() {
+                            break;
+                        }
+                    }
+
+                    Ok::<(), anyhow::Error>(())
+                })
+                .unwrap();
+            },
+            BatchSize::LargeInput,
+        )
+    });
+}
+
+criterion_group!(benches, push, pull);
+criterion_main!(benches);

car-mirror/src/dag_walk.rs

@@ -162,7 +162,7 @@ mod tests {
 #[cfg(test)]
 mod proptests {
     use super::*;
-    use crate::test_utils::{encode, generate_dag};
+    use crate::test_utils::{arb_ipld_dag, encode};
     use futures::TryStreamExt;
     use libipld::{
         multihash::{Code, MultihashDigest},
@@ -174,7 +174,7 @@ mod proptests {
     use wnfs_common::{BlockStore, MemoryBlockStore};
 
     fn ipld_dags() -> impl Strategy<Value = (Vec<(Cid, Ipld)>, Cid)> {
-        generate_dag(256, |cids, _| {
+        arb_ipld_dag(1..256, 0.5, |cids, _| {
             let ipld = Ipld::List(cids.into_iter().map(Ipld::Link).collect());
             let cid = Cid::new_v1(
                 IpldCodec::DagCbor.into(),

car-mirror/src/test_utils/blockstore_utils.rs

@@ -0,0 +1,26 @@
+use anyhow::Result;
+use bytes::Bytes;
+use libipld::{Cid, Ipld, IpldCodec};
+use libipld_core::codec::Encode;
+use wnfs_common::{BlockStore, MemoryBlockStore};
+
+/// Take a list of dag-cbor IPLD blocks and store all of them as dag-cbor in a
+/// MemoryBlockStore & return it.
+pub async fn setup_blockstore(blocks: Vec<(Cid, Ipld)>) -> Result<MemoryBlockStore> {
+    let store = MemoryBlockStore::new();
+    for (cid, ipld) in blocks.into_iter() {
+        let cid_store = store
+            .put_block(encode(&ipld), IpldCodec::DagCbor.into())
+            .await?;
+        debug_assert_eq!(cid, cid_store);
+    }
+
+    Ok(store)
+}
+
+/// Encode some IPLD as dag-cbor.
+pub fn encode(ipld: &Ipld) -> Bytes {
+    let mut vec = Vec::new();
+    ipld.encode(IpldCodec::DagCbor, &mut vec).unwrap();
+    Bytes::from(vec)
+}

car-mirror/src/test_utils/dag_strategy.rs

@@ -1,20 +1,71 @@
-use std::{collections::HashSet, fmt::Debug};
-
-use libipld::Cid;
-use proptest::{strategy::Strategy, test_runner::TestRng};
+use super::encode;
+use bytes::Bytes;
+use libipld::{Cid, Ipld, IpldCodec};
+use libipld_core::multihash::{Code, MultihashDigest};
+use proptest::{prelude::Rng, strategy::Strategy, test_runner::TestRng};
 use roaring_graphs::{arb_dag, DirectedAcyclicGraph, Vertex};
+use std::{
+    collections::{BTreeMap, HashSet},
+    fmt::Debug,
+    ops::Range,
+};
 
 /// A strategy for use with proptest to generate random DAGs (directed acyclic graphs).
 /// The strategy generates a list of blocks of type T and their CIDs, as well as
 /// the root block's CID.
-pub fn generate_dag<T: Debug + Clone>(
-    max_nodes: u16,
+pub fn arb_ipld_dag<T: Debug + Clone>(
+    vertex_count: impl Into<Range<Vertex>>,
+    edge_probability: f64,
     generate_block: impl Fn(Vec<Cid>, &mut TestRng) -> (Cid, T) + Clone,
 ) -> impl Strategy<Value = (Vec<(Cid, T)>, Cid)> {
-    arb_dag(1..max_nodes, 0.5)
+    arb_dag(vertex_count, edge_probability)
         .prop_perturb(move |dag, mut rng| dag_to_nodes(&dag, &mut rng, generate_block.clone()))
 }
 
+/// A block-generating function for use with `arb_ipld_dag`.
+pub fn links_to_ipld(cids: Vec<Cid>, _: &mut TestRng) -> (Cid, Ipld) {
+    let ipld = Ipld::List(cids.into_iter().map(Ipld::Link).collect());
+    let cid = Cid::new_v1(
+        IpldCodec::DagCbor.into(),
+        Code::Blake3_256.digest(&encode(&ipld)),
+    );
+    (cid, ipld)
+}
+
+/// A block-generating function for use with `arb_ipld_dag`.
+pub fn links_to_dag_cbor(cids: Vec<Cid>, _: &mut TestRng) -> (Cid, Bytes) {
+    let ipld = Ipld::List(cids.into_iter().map(Ipld::Link).collect());
+    let bytes = encode(&ipld);
+    let cid = Cid::new_v1(IpldCodec::DagCbor.into(), Code::Blake3_256.digest(&bytes));
+    (cid, bytes)
+}
+
+/// A block-generating function for use with `arb_ipld_dag`.
+///
+/// Creates (a function that creates) an IPLD block with given links & some
+/// random `padding_bytes` bytes attached.
+pub fn links_to_padded_ipld(
+    padding_bytes: usize,
+) -> impl Fn(Vec<Cid>, &mut TestRng) -> (Cid, Ipld) + Clone {
+    move |cids, rng| {
+        let mut padding = Vec::with_capacity(padding_bytes);
+        for _ in 0..padding_bytes {
+            padding.push(rng.gen::<u8>());
+        }
+
+        let ipld = Ipld::Map(BTreeMap::from([
+            ("data".into(), Ipld::Bytes(padding)),
+            (
+                "links".into(),
+                Ipld::List(cids.into_iter().map(Ipld::Link).collect()),
+            ),
+        ]));
+        let bytes = encode(&ipld);
+        let cid = Cid::new_v1(IpldCodec::DagCbor.into(), Code::Blake3_256.digest(&bytes));
+        (cid, ipld)
+    }
+}
+
 /// Turn a directed acyclic graph into a list of nodes (with their CID) and a root CID.
 /// This will select only the DAG that's reachable from the root.
 pub fn dag_to_nodes<T>(

car-mirror/src/test_utils/local_utils.rs

@@ -1,16 +1,10 @@
 ///! Crate-local test utilities
-use super::{generate_dag, Rvg};
+use super::{arb_ipld_dag, links_to_padded_ipld, setup_blockstore, Rvg};
 use crate::{common::references, dag_walk::DagWalk};
 use anyhow::Result;
-use bytes::Bytes;
 use futures::TryStreamExt;
-use libipld::{Cid, Ipld, IpldCodec};
-use libipld_core::{
-    codec::Encode,
-    multihash::{Code, MultihashDigest},
-};
-use proptest::{prelude::Rng, strategy::Strategy};
-use std::collections::BTreeMap;
+use libipld::{Cid, Ipld};
+use proptest::strategy::Strategy;
 use wnfs_common::{BlockStore, MemoryBlockStore};
 
 #[derive(Clone, Debug)]
@@ -44,23 +38,7 @@ pub(crate) fn padded_dag_strategy(
     dag_size: u16,
     block_padding: usize,
 ) -> impl Strategy<Value = (Vec<(Cid, Ipld)>, Cid)> {
-    generate_dag(dag_size, move |cids, rng| {
-        let mut padding = Vec::with_capacity(block_padding);
-        for _ in 0..block_padding {
-            padding.push(rng.gen::<u8>());
-        }
-
-        let ipld = Ipld::Map(BTreeMap::from([
-            ("data".into(), Ipld::Bytes(padding)),
-            (
-                "links".into(),
-                Ipld::List(cids.into_iter().map(Ipld::Link).collect()),
-            ),
-        ]));
-        let bytes = encode(&ipld);
-        let cid = Cid::new_v1(IpldCodec::DagCbor.into(), Code::Blake3_256.digest(&bytes));
-        (cid, ipld)
-    })
+    arb_ipld_dag(1..dag_size, 0.5, links_to_padded_ipld(block_padding))
 }
 
 pub(crate) fn variable_blocksize_dag() -> impl Strategy<Value = (Vec<(Cid, Ipld)>, Cid)> {
@@ -76,20 +54,9 @@ pub(crate) fn variable_blocksize_dag() -> impl Strategy<Value = (Vec<(Cid, Ipld)
     const MAX_BLOCK_SIZE: usize = 256 * 1024;
     const MAX_BLOCK_PADDING: usize = MAX_BLOCK_SIZE - EST_OVERHEAD - MAX_LINK_BYTES;
 
-    (32..MAX_BLOCK_PADDING)
-        .prop_ind_flat_map(move |block_padding| padded_dag_strategy(MAX_DAG_NODES, block_padding))
-}
-
-pub(crate) async fn setup_blockstore(blocks: Vec<(Cid, Ipld)>) -> Result<MemoryBlockStore> {
-    let store = MemoryBlockStore::new();
-    for (cid, ipld) in blocks.into_iter() {
-        let cid_store = store
-            .put_block(encode(&ipld), IpldCodec::DagCbor.into())
-            .await?;
-        debug_assert_eq!(cid, cid_store);
-    }
-
-    Ok(store)
+    (32..MAX_BLOCK_PADDING).prop_ind_flat_map(move |block_padding| {
+        arb_ipld_dag(1..MAX_DAG_NODES, 0.5, links_to_padded_ipld(block_padding))
+    })
 }
 
 pub(crate) async fn setup_random_dag(
@@ -119,10 +86,3 @@ pub(crate) async fn total_dag_blocks(root: Cid, store: &impl BlockStore) -> Resu
         .await?
         .len())
 }
-
-/// Encode some IPLD as dag-cbor
-pub(crate) fn encode(ipld: &Ipld) -> Bytes {
-    let mut vec = Vec::new();
-    ipld.encode(IpldCodec::DagCbor, &mut vec).unwrap();
-    Bytes::from(vec)
-}

car-mirror/src/test_utils/mod.rs

@@ -7,6 +7,10 @@ mod rvg;
 pub use dag_strategy::*;
 #[cfg(feature = "test_utils")]
 pub use rvg::*;
+#[cfg(feature = "test_utils")]
+mod blockstore_utils;
+#[cfg(feature = "test_utils")]
+pub use blockstore_utils::*;
 
 #[cfg(test)]
 mod local_utils;