[runtime/utils/buffer/paged/append] zero-copy writes (#3434)

Co-authored-by: Patrick O'Grady <me@patrickogrady.xyz>

Author: andresilva

runtime/src/utils/buffer/benches/append.rs

@@ -1,19 +1,26 @@
 //! Benchmark sequential append performance.
 
 use super::{create_append, destroy_append, CACHE_SIZE, PAGE_SIZE};
-use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner as _};
+use commonware_runtime::{
+    buffer::paged::CacheRef, deterministic, tokio, BufferPooler, Runner, Storage,
+};
 use commonware_utils::NZUsize;
 use criterion::Criterion;
 use std::time::Instant;
 
-pub fn bench(c: &mut Criterion) {
-    for chunk_size in [64, 256, 1024, 4096] {
-        c.bench_function(&format!("{}/chunk={}", module_path!(), chunk_size), |b| {
+fn bench_backend<R>(c: &mut Criterion, backend: &str, chunk_size: usize)
+where
+    R: Runner + Default,
+    R::Context: Storage + BufferPooler,
+{
+    c.bench_function(
+        &format!("{}/backend={backend} chunk={chunk_size}", module_path!()),
+        |b| {
             b.iter_custom(|iters| {
-                let name = format!("append_seq_{chunk_size}").into_bytes();
+                let name = format!("append_seq_{backend}_{chunk_size}").into_bytes();
                 let data = vec![0xABu8; chunk_size];
 
-                let executor = deterministic::Runner::default();
+                let executor = R::default();
                 executor.start(|ctx| async move {
                     let cache_ref = CacheRef::from_pooler(&ctx, PAGE_SIZE, NZUsize!(CACHE_SIZE));
                     let append = create_append(&ctx, &name, cache_ref).await;
@@ -32,6 +39,13 @@ pub fn bench(c: &mut Criterion) {
                     elapsed
                 })
             });
-        });
+        },
+    );
+}
+
+pub fn bench(c: &mut Criterion) {
+    for chunk_size in [64, 256, 1024, 4096] {
+        bench_backend::<deterministic::Runner>(c, "deterministic", chunk_size);
+        bench_backend::<tokio::Runner>(c, "tokio", chunk_size);
     }
 }

runtime/src/utils/buffer/benches/bench.rs

@@ -1,14 +1,13 @@
 //! Benchmarks for the page cache via the Append blob wrapper.
 //!
-//! Uses memory-based storage (deterministic runtime) to isolate
-//! page cache performance from disk I/O.
+//! Includes both deterministic in-memory runs to isolate buffer/page-cache work
+//! and Tokio-backed runs to exercise real storage I/O paths.
 //!
 //! Run with: `cargo bench --bench buffer_paged -p commonware-runtime`
 
 use commonware_runtime::{
     buffer::paged::{Append, CacheRef},
-    deterministic::Context,
-    Storage as _,
+    Storage,
 };
 use commonware_utils::NZU16;
 use criterion::{criterion_group, criterion_main};
@@ -22,17 +21,15 @@ const PAGE_SIZE_USIZE: usize = PAGE_SIZE.get() as usize;
 const WRITE_BUFFER_SIZE: usize = PAGE_SIZE_USIZE * 4;
 const CACHE_SIZE: usize = 10_000;
 
-type MemBlob = <Context as commonware_runtime::Storage>::Blob;
-
 /// Create a new Append wrapper for benchmarking.
-async fn create_append(ctx: &Context, name: &[u8], cache_ref: CacheRef) -> Append<MemBlob> {
+async fn create_append<C: Storage>(ctx: &C, name: &[u8], cache_ref: CacheRef) -> Append<C::Blob> {
     let (blob, size) = ctx.open("bench_partition", name).await.unwrap();
     Append::new(blob, size, WRITE_BUFFER_SIZE, cache_ref)
         .await
         .unwrap()
 }
 
-async fn destroy_append(ctx: &Context, append: Append<MemBlob>, name: &[u8]) {
+async fn destroy_append<C: Storage>(ctx: &C, append: Append<C::Blob>, name: &[u8]) {
     drop(append);
     ctx.remove("bench_partition", Some(name)).await.unwrap();
 }

runtime/src/utils/buffer/benches/read.rs

@@ -1,7 +1,9 @@
 //! Benchmark read performance at random offsets.
 
 use super::{create_append, destroy_append, CACHE_SIZE, PAGE_SIZE, PAGE_SIZE_USIZE};
-use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner as _};
+use commonware_runtime::{
+    buffer::paged::CacheRef, deterministic, tokio, BufferPooler, Runner, Storage,
+};
 use commonware_utils::NZUsize;
 use criterion::Criterion;
 use rand::{rngs::StdRng, Rng, SeedableRng};
@@ -11,17 +13,21 @@ use std::time::Instant;
 const TOTAL_PAGES: usize = CACHE_SIZE;
 const TOTAL_SIZE: usize = PAGE_SIZE_USIZE * TOTAL_PAGES;
 
-pub fn bench(c: &mut Criterion) {
-    for read_size in [64, 256, 1024, 4096] {
-        let name = format!("read_{read_size}").into_bytes();
-
-        c.bench_function(&format!("{}/size={}", module_path!(), read_size), |b| {
+fn bench_backend<R>(c: &mut Criterion, backend: &str, read_size: usize)
+where
+    R: Runner + Default,
+    R::Context: Storage + BufferPooler,
+{
+    c.bench_function(
+        &format!("{}/backend={backend} size={read_size}", module_path!()),
+        |b| {
             b.iter_custom(|iters| {
-                let name = name.clone();
+                let name = format!("read_{backend}_{read_size}").into_bytes();
 
-                let executor = deterministic::Runner::default();
+                let executor = R::default();
                 executor.start(|ctx| async move {
                     let cache_ref = CacheRef::from_pooler(&ctx, PAGE_SIZE, NZUsize!(CACHE_SIZE));
+
                     // Setup: populate the blob
                     let append = create_append(&ctx, &name, cache_ref.clone()).await;
                     let data = vec![0xABu8; TOTAL_SIZE];
@@ -30,7 +36,7 @@ pub fn bench(c: &mut Criterion) {
                     drop(append);
 
                     // Benchmark: random reads
-                    let append = create_append(&ctx, &name, cache_ref.clone()).await;
+                    let append = create_append(&ctx, &name, cache_ref).await;
                     let mut buf = vec![0u8; read_size];
                     let max_offset = TOTAL_SIZE - read_size;
                     let mut rng = StdRng::seed_from_u64(42);
@@ -51,6 +57,13 @@ pub fn bench(c: &mut Criterion) {
                     elapsed
                 })
             });
-        });
+        },
+    );
+}
+
+pub fn bench(c: &mut Criterion) {
+    for read_size in [64, 256, 1024, 4096] {
+        bench_backend::<deterministic::Runner>(c, "deterministic", read_size);
+        bench_backend::<tokio::Runner>(c, "tokio", read_size);
     }
 }