Io uring runtime

benchmark/Cargo.toml

@@ -12,6 +12,7 @@ liquid-cache-storage = { workspace = true }
 liquid-cache-common = { workspace = true }
 liquid-cache-local = { workspace = true }
 datafusion = { workspace = true }
+futures = { workspace = true }
 tokio = { workspace = true }
 log = { workspace = true }
 arrow-flight = { workspace = true }
@@ -67,3 +68,7 @@ path = "bench_server.rs"
 [[bin]]
 name = "in_process"
 path = "in_process.rs"
+
+[[bin]]
+name = "storage_runner"
+path = "src/storage_runner.rs"

benchmark/bench_server.rs

@@ -51,6 +51,9 @@ struct CliArgs {
     /// IO mode, available options: uring, uring-direct, std-blocking, tokio, std-spawn-blocking
     #[arg(long = "io-mode", default_value = "uring-multi-async")]
     io_mode: IoMode,
+
+    #[arg(long = "fixed-buffer-pool-size-mb", default_value = "0")]
+    fixed_buffer_pool_size_mb: usize,
 }
 
 #[tokio::main]
@@ -81,6 +84,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         squeeze_policy,
         Box::new(NoHydration::new()),
         Some(args.io_mode),
+        args.fixed_buffer_pool_size_mb,
     )?;
 
     let liquid_cache_server = Arc::new(liquid_cache_server);

benchmark/in_process.rs

@@ -4,7 +4,7 @@ use fastrace::prelude::*;
 use liquid_cache_benchmarks::{
     BenchmarkManifest, InProcessBenchmarkMode, InProcessBenchmarkRunner, setup_observability,
 };
-use liquid_cache_common::IoMode;
+use liquid_cache_common::{IoMode, memory::pool::FixedBufferPool};
 use mimalloc::MiMalloc;
 use serde::Serialize;
 use std::path::PathBuf;
@@ -70,6 +70,9 @@ struct InProcessBenchmark {
     /// IO mode, available options: uring, uring-direct, std-blocking, tokio, std-spawn-blocking
     #[arg(long = "io-mode", default_value = "uring-multi-async")]
     io_mode: IoMode,
+
+    #[arg(long = "fixed-buffer-pool-size-mb", default_value = "0")]
+    fixed_buffer_pool_size_mb: usize,
 }
 
 impl InProcessBenchmark {
@@ -88,7 +91,8 @@ impl InProcessBenchmark {
             .with_cache_dir(self.cache_dir.clone())
             .with_query_filter(self.query_index)
             .with_io_mode(self.io_mode)
-            .with_output_dir(self.output_dir.clone());
+            .with_output_dir(self.output_dir.clone())
+            .with_fixed_buffer_pool_size_mb(self.fixed_buffer_pool_size_mb);
         runner.run(manifest, self, output).await?;
         Ok(())
     }
@@ -102,6 +106,7 @@ async fn main() -> Result<()> {
     let _guard = root.set_local_parent();
 
     benchmark.run().await?;
+    FixedBufferPool::print_stats();
     fastrace::flush();
     Ok(())
 }

benchmark/src/inprocess_runner.rs

@@ -202,6 +202,7 @@ pub struct InProcessBenchmarkRunner {
     pub io_mode: IoMode,
     pub output_dir: Option<PathBuf>,
     pub collect_perf_events: bool,
+    pub fixed_buffer_pool_size_mb: usize,
 }
 
 impl Default for InProcessBenchmarkRunner {
@@ -224,6 +225,7 @@ impl InProcessBenchmarkRunner {
             io_mode: IoMode::default(),
             output_dir: None,
             collect_perf_events: false,
+            fixed_buffer_pool_size_mb: 0,
         }
     }
 
@@ -282,6 +284,11 @@ impl InProcessBenchmarkRunner {
         self
     }
 
+    pub fn with_fixed_buffer_pool_size_mb(mut self, fixed_buffer_pool_size_mb: usize) -> Self {
+        self.fixed_buffer_pool_size_mb = fixed_buffer_pool_size_mb;
+        self
+    }
+
     #[fastrace::trace]
     async fn setup_context(
         &self,
@@ -348,6 +355,7 @@ impl InProcessBenchmarkRunner {
                     .with_squeeze_policy(Box::new(TranscodeSqueezeEvict))
                     .with_io_mode(self.io_mode)
                     .with_eager_shredding(true)
+                    .with_fixed_buffer_pool_size_mb(self.fixed_buffer_pool_size_mb)
                     .build(session_config)?;
                 (v.0, Some(v.1))
             }
@@ -359,6 +367,7 @@ impl InProcessBenchmarkRunner {
                     .with_hydration_policy(Box::new(NoHydration::new()))
                     .with_squeeze_policy(Box::new(TranscodeEvict))
                     .with_io_mode(self.io_mode)
+                    .with_fixed_buffer_pool_size_mb(self.fixed_buffer_pool_size_mb)
                     .build(session_config)?;
                 (v.0, Some(v.1))
             }

benchmark/src/lib.rs

@@ -23,7 +23,7 @@ pub mod client_runner;
 pub mod inprocess_runner;
 mod manifest;
 mod observability;
-mod tracepoints;
+pub mod tracepoints;
 pub mod utils;
 
 pub use client_runner::*;
@@ -487,6 +487,81 @@ impl Display for IterationResult {
     }
 }
 
+/// Table layout matching [`IterationResult`]'s [`Display`] (borders, row style, disk formatting).
+/// When `uring_runnable` is `Some`, includes work-stealing executor `Runnable::run` timing (see storage runner).
+pub fn format_storage_iteration_metrics(
+    iteration: usize,
+    iteration_wall: Duration,
+    disk_read: u64,
+    disk_written: u64,
+    uring_runnable: Option<(f64, f64)>,
+) -> String {
+    struct StorageIterationTable {
+        iteration: usize,
+        iteration_wall_ms: u64,
+        uring_runnable: Option<(f64, f64)>,
+        disk_read: u64,
+        disk_written: u64,
+    }
+
+    impl std::fmt::Display for StorageIterationTable {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            const INNER: usize = 50;
+            write_border_top(f, INNER)?;
+            write_kv_row(
+                f,
+                INNER,
+                "Iteration:",
+                &format!("{}", self.iteration),
+            )?;
+            write_kv_row(
+                f,
+                INNER,
+                "Iteration wall:",
+                &format!("{} ms", format_number(self.iteration_wall_ms)),
+            )?;
+            if let Some((runnable_wall_ms, wall_minus_runnable_ms)) = self.uring_runnable {
+                write_kv_row(
+                    f,
+                    INNER,
+                    "Runnable wall (sum):",
+                    &format!("{:.3} ms", runnable_wall_ms),
+                )?;
+                write_kv_row(
+                    f,
+                    INNER,
+                    "Wall minus runnable:",
+                    &format!("{:.3} ms", wall_minus_runnable_ms),
+                )?;
+            }
+            write_border_sep(f, INNER)?;
+            write_kv_row(
+                f,
+                INNER,
+                "Disk (Read/Write):",
+                &format!(
+                    "{} / {}",
+                    format_bytes(self.disk_read),
+                    format_bytes(self.disk_written)
+                ),
+            )?;
+            write_border_bottom(f, INNER)
+        }
+    }
+
+    let iteration_wall_ms = (iteration_wall.as_secs_f64() * 1000.0).round() as u64;
+    format!(
+        "{}",
+        StorageIterationTable {
+            iteration,
+            iteration_wall_ms,
+            uring_runnable,
+            disk_read,
+            disk_written,
+        }
+    )
+}
+
 fn format_number(n: u64) -> String {
     let s = n.to_string();
     let chars: Vec<char> = s.chars().collect();