refactor: Use newer BlockStore trait methods

wnfs-wg · Feb 15, 2024 · 69de1f3 · 69de1f3
1 parent 1d86575
commit 69de1f3
Show file tree

Hide file tree

Showing 11 changed files with 204 additions and 309 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/car-mirror-benches/benches/artificially_slow_blockstore.rs b/car-mirror-benches/benches/artificially_slow_blockstore.rs
@@ -27,9 +27,9 @@ pub fn push_throttled(c: &mut Criterion) {
             },
             |(client_store, root)| {
                 let client_store = &ThrottledBlockStore(client_store);
-                let client_cache = &InMemoryCache::new(10_000, 150_000);
+                let client_cache = &InMemoryCache::new(10_000);
                 let server_store = &ThrottledBlockStore::new();
-                let server_cache = &InMemoryCache::new(10_000, 150_000);
+                let server_cache = &InMemoryCache::new(10_000);
                 let config = &Config::default();
 
                 // Simulate a multi-round protocol run in-memory
@@ -81,9 +81,9 @@ pub fn pull_throttled(c: &mut Criterion) {
             },
             |(server_store, root)| {
                 let server_store = &ThrottledBlockStore(server_store);
-                let server_cache = &InMemoryCache::new(10_000, 150_000);
+                let server_cache = &InMemoryCache::new(10_000);
                 let client_store = &ThrottledBlockStore::new();
-                let client_cache = &InMemoryCache::new(10_000, 150_000);
+                let client_cache = &InMemoryCache::new(10_000);
                 let config = &Config::default();
 
                 // Simulate a multi-round protocol run in-memory
@@ -117,14 +117,22 @@ struct ThrottledBlockStore(MemoryBlockStore);
 
 impl BlockStore for ThrottledBlockStore {
     async fn get_block(&self, cid: &Cid) -> Result<Bytes> {
-        let bytes = self.0.get_block(cid).await?;
         async_std::task::sleep(Duration::from_micros(50)).await; // Block fetching is artifically slowed by 50 microseconds
-        Ok(bytes)
+        self.0.get_block(cid).await
     }
 
     async fn put_block(&self, bytes: impl Into<Bytes> + CondSend, codec: u64) -> Result<Cid> {
         self.0.put_block(bytes, codec).await
     }
+
+    async fn put_block_keyed(&self, cid: Cid, bytes: impl Into<Bytes> + CondSend) -> Result<()> {
+        self.0.put_block_keyed(cid, bytes).await
+    }
+
+    async fn has_block(&self, cid: &Cid) -> Result<bool> {
+        // The idea is that has_block would be faster than `get_block`, as it should be managed closer to CPU memory
+        self.0.has_block(cid).await
+    }
 }
 
 impl ThrottledBlockStore {

diff --git a/car-mirror-benches/benches/in_memory.rs b/car-mirror-benches/benches/in_memory.rs
@@ -22,9 +22,9 @@ pub fn push(c: &mut Criterion) {
                 (store, root)
             },
             |(ref client_store, root)| {
-                let client_cache = &InMemoryCache::new(10_000, 150_000);
+                let client_cache = &InMemoryCache::new(10_000);
                 let server_store = &MemoryBlockStore::new();
-                let server_cache = &InMemoryCache::new(10_000, 150_000);
+                let server_cache = &InMemoryCache::new(10_000);
                 let config = &Config::default();
 
                 // Simulate a multi-round protocol run in-memory
@@ -68,9 +68,9 @@ pub fn pull(c: &mut Criterion) {
                 (store, root)
             },
             |(ref server_store, root)| {
-                let server_cache = &InMemoryCache::new(10_000, 150_000);
+                let server_cache = &InMemoryCache::new(10_000);
                 let client_store = &MemoryBlockStore::new();
-                let client_cache = &InMemoryCache::new(10_000, 150_000);
+                let client_cache = &InMemoryCache::new(10_000);
                 let config = &Config::default();
 
                 // Simulate a multi-round protocol run in-memory

diff --git a/car-mirror-benches/benches/simulated_latency.rs b/car-mirror-benches/benches/simulated_latency.rs
@@ -68,12 +68,12 @@ pub fn pull_with_simulated_latency(
                     links_to_padded_ipld(block_padding),
                 ));
                 let store = async_std::task::block_on(setup_blockstore(blocks)).unwrap();
-                let cache = InMemoryCache::new(10_000, 150_000);
+                let cache = InMemoryCache::new(10_000);
                 (store, cache, root)
             },
             |(ref server_store, ref server_cache, root)| {
                 let client_store = &MemoryBlockStore::new();
-                let client_cache = &InMemoryCache::new(10_000, 150_000);
+                let client_cache = &InMemoryCache::new(10_000);
                 let config = &Config::default();
 
                 // Simulate a multi-round protocol run in-memory
@@ -145,12 +145,12 @@ pub fn push_with_simulated_latency(
                     links_to_padded_ipld(block_padding),
                 ));
                 let store = async_std::task::block_on(setup_blockstore(blocks)).unwrap();
-                let cache = InMemoryCache::new(10_000, 150_000);
+                let cache = InMemoryCache::new(10_000);
                 (store, cache, root)
             },
             |(ref client_store, ref client_cache, root)| {
                 let server_store = &MemoryBlockStore::new();
-                let server_cache = &InMemoryCache::new(10_000, 150_000);
+                let server_cache = &InMemoryCache::new(10_000);
                 let config = &Config::default();
 
                 // Simulate a multi-round protocol run in-memory

diff --git a/car-mirror/src/common.rs b/car-mirror/src/common.rs
@@ -308,6 +308,8 @@ async fn car_frame_from_block(block: (Cid, Bytes)) -> Result<Bytes, Error> {
     Ok(bytes.into())
 }
 
+/// Ensure that any requested subgraph roots are actually part
+/// of the DAG from the root.
 async fn verify_missing_subgraph_roots(
     root: Cid,
     missing_subgraph_roots: &[Cid],
@@ -316,9 +318,10 @@ async fn verify_missing_subgraph_roots(
 ) -> Result<Vec<Cid>, Error> {
     let subgraph_roots: Vec<Cid> = DagWalk::breadth_first([root])
         .stream(store, cache)
-        .try_filter_map(
-            |cid| async move { Ok(missing_subgraph_roots.contains(&cid).then_some(cid)) },
-        )
+        .try_filter_map(|item| async move {
+            let cid = item.to_cid()?;
+            Ok(missing_subgraph_roots.contains(&cid).then_some(cid))
+        })
         .try_collect()
         .await?;
 
@@ -362,7 +365,9 @@ fn stream_blocks_from_roots<'a>(
     Box::pin(async_stream::try_stream! {
         let mut dag_walk = DagWalk::breadth_first(subgraph_roots.clone());
 
-        while let Some(cid) = dag_walk.next(&store, &cache).await? {
+        while let Some(item) = dag_walk.next(&store, &cache).await? {
+            let cid = item.to_cid()?;
+
             if should_block_be_skipped(&cid, &bloom, &subgraph_roots) {
                 continue;
             }