Skip to content

refactor: collapse duplicated harness and engine implementations #57

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tisonkun merged 8 commits into from
Sep 20, 2026
Merged

refactor: collapse duplicated harness and engine implementations #57

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
559070c
refactor: collapse duplicated harness and engine implementations
tisonkun Sep 20, 2026
70e2366
fix: reject invalid benchmark enum environment values
tisonkun Sep 20, 2026
9465cd4
refactor: borrow I/O runtime state without exposing Arc
tisonkun Sep 20, 2026
4a7d290
refactor: reuse shared field readers for index pages
tisonkun Sep 20, 2026
8454850
test: coordinate shutdown races without an engine wrapper
tisonkun Sep 20, 2026
8c7e856
refactor: share one concrete I/O engine across drivers
tisonkun Sep 20, 2026
be3678e
Merge main into codex/deduplicate-shared-implementations
tisonkun Sep 20, 2026
7090d46
refactor: unify CRC32C computation over byte segments
tisonkun Sep 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,8 @@ The source Region stays pinned until accepted replacement writes finish. Conditi

Reads and writes use independent bounded engine pools. Reclaim has separate read lanes. POSIX uses positioned worker I/O; optional io_uring uses fixed-depth rings. Buffered I/O is the default. Direct mode aligns runtime record I/O and keeps control, recovery, and unavoidable remainder operations buffered. Locks cover bounded in-memory work and release before device I/O.

Each lane uses one concrete `IoEngine` for admission, submission, cancellation, statistics, and shutdown. Driver-specific constructors start POSIX workers or an io_uring driver behind the same bounded command and completion protocol. Callers share the engine through `Arc`; its final owner joins the workers. Submitted requests retain their buffers and capacity until actual completion, independently of the caller's wait deadline.

### Memory

The managed-memory limit covers the index mapping, heat bits, L1, append buffers, reclaim buffers, metadata, cache-owned thread stacks, recovery scratch, and transient reads. Total deployment memory additionally includes allocator metadata, Tokio, process overhead, and the kernel page cache. `CacheConfig::new` rejects invalid or insufficient memory budgets before file access; actual allocation can still fail during open.
Expand Down
177 changes: 26 additions & 151 deletions benchmarks/cache/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,25 +14,31 @@

use std::env;
use std::fmt;
use std::fs;
use std::hint::black_box;
use std::io;
use std::ops::Range;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;

use asyncband::barrier::Barrier;
use benchmarks::config::env_bool;
use benchmarks::config::env_optional_f64;
use benchmarks::config::env_u32;
use benchmarks::config::env_usize;
use benchmarks::config::invalid;
use benchmarks::config::io_engine_from_env;
use benchmarks::config::parse_io_mode;
use benchmarks::config::parse_l1_eviction_policy;
use benchmarks::config::read_max_in_flight;
use benchmarks::harness::BenchFiles;
use benchmarks::harness::put_eventually;
use benchmarks::report::JobReport;
use benchmarks::report::LatencyHistogram;
use benchmarks::report::RunReporter;
use benchmarks::report::emit_cache_report;
use benchmarks::report::should_sample;
use cache2::Cache;
use cache2::CacheConfig;
use cache2::CacheTier;
Expand All @@ -56,7 +62,6 @@ const MAX_VALUE_BYTES: usize = REGION_BYTES - 64;
const READ_RETRY_TIMEOUT: Duration = Duration::from_secs(1);
const WRITE_RETRY_TIMEOUT: Duration = Duration::from_secs(10);
const RETRY_DELAY: Duration = Duration::from_micros(50);
const WRITE_YIELD_RETRIES: usize = 8;

struct BenchConfig {
entries: usize,
Expand Down Expand Up @@ -105,24 +110,8 @@ impl BenchConfig {
env_usize("CACHE_BENCH_READ_LATENCY_SAMPLE_INTERVAL", 16)?;
let clients = env_usize("CACHE_BENCH_CLIENTS", 8)?;
let write_clients = env_usize("CACHE_BENCH_WRITE_CLIENTS", 4)?;
let io_mode = match env::var("CACHE_BENCH_IO_MODE")
.unwrap_or_else(|_| "buffered".to_owned())
.as_str()
{
"buffered" => IoMode::Buffered,
"direct" => IoMode::Direct,
value => return Err(invalid(format!("unsupported I/O mode: {value}"))),
};
let l1_eviction_policy = match env::var("CACHE_BENCH_L1_EVICTION")
.unwrap_or_else(|_| "clock".to_owned())
.as_str()
{
"clock" => L1EvictionPolicy::Clock,
"s3-fifo" => L1EvictionPolicy::S3Fifo,
value => {
return Err(invalid(format!("unsupported L1 eviction policy: {value}")));
}
};
let io_mode = parse_io_mode("CACHE_BENCH_IO_MODE")?;
let l1_eviction_policy = parse_l1_eviction_policy("CACHE_BENCH_L1_EVICTION")?;
let latency = |name| -> io::Result<cache2::LatencyMode> {
Ok(match env_u32(name, 0)? {
0 => cache2::LatencyMode::Off,
Expand Down Expand Up @@ -284,38 +273,6 @@ impl BenchConfig {
}
}

struct BenchFiles {
data: PathBuf,
}

impl BenchFiles {
fn new(directory: &Path) -> Self {
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
Self {
data: directory.join(format!(
"cache2-bench-{}-{timestamp}.cache",
std::process::id()
)),
}
}
}

impl Drop for BenchFiles {
fn drop(&mut self) {
for path in [
self.data.clone(),
sidecar(&self.data, ".state"),
sidecar(&self.data, ".image"),
sidecar(&self.data, ".image.next"),
] {
let _ = fs::remove_file(path);
}
}
}

struct Measurement {
elapsed: Duration,
operations: usize,
Expand Down Expand Up @@ -375,7 +332,7 @@ fn run_benchmark() -> io::Result<()> {
}

async fn run(config: BenchConfig) -> io::Result<()> {
let files = BenchFiles::new(&config.directory);
let files = BenchFiles::new(&config.directory, "bench");
let l1_entry_eligible = benchmark_entry_is_l1_eligible(config.value_bytes);
let cache_config =
CacheConfig::new(config.storage_options().build()?, config.runtime_options())?;
Expand Down Expand Up @@ -415,11 +372,11 @@ async fn run(config: BenchConfig) -> io::Result<()> {
config.io_mode,
config.stats.activity_counters,
);
println!("file={}", files.data.display());
println!("file={}", files.data().display());

let cache = Arc::new(
Cache::open(
&files.data,
files.data(),
if initial_l1_bytes == config.l1_capacity_bytes {
cache_config.clone()
} else {
Expand Down Expand Up @@ -477,7 +434,7 @@ async fn run(config: BenchConfig) -> io::Result<()> {
let warm_close = started.elapsed();
report_latency("warm_close", "warm close", warm_close);

let cache = Arc::new(Cache::open(&files.data, cache_config.clone()).await?);
let cache = Arc::new(Cache::open(files.data(), cache_config.clone()).await?);
if cache.startup_mode() != StartupMode::Warm {
return Err(io::Error::other(
"benchmark did not reopen from a clean image",
Expand Down Expand Up @@ -659,7 +616,7 @@ async fn run(config: BenchConfig) -> io::Result<()> {
drop(cache);

let resident = if l1_entry_eligible {
let cache = Arc::new(Cache::open(&files.data, cache_config.clone()).await?);
let cache = Arc::new(Cache::open(files.data(), cache_config.clone()).await?);
if cache.startup_mode() != StartupMode::Cold {
return Err(io::Error::other(
"fast-closed benchmark did not reopen empty",
Expand Down Expand Up @@ -741,8 +698,12 @@ fn concurrent_writes(
for ordinal in (client..entries).step_by(clients) {
let key = benchmark_key(ordinal);
value[..8].copy_from_slice(&(ordinal as u64).to_le_bytes());
let (receipt, operation_attempts) =
put_eventually(&cache, black_box(&key), black_box(&value))?;
let (receipt, operation_attempts) = put_eventually(
&cache,
black_box(&key),
black_box(&value),
WRITE_RETRY_TIMEOUT,
)?;
attempts = attempts.saturating_add(operation_attempts);
throttled_writes =
throttled_writes.saturating_add(usize::from(operation_attempts > 1));
Expand Down Expand Up @@ -804,17 +765,14 @@ async fn concurrent_reads(
let mut latency = LatencyHistogram::default();
for ordinal in (client..operations).step_by(clients) {
let key_ordinal = first_key + ordinal % key_count;
let started = (latency_sample_interval != 0
&& ordinal.is_multiple_of(latency_sample_interval))
.then(Instant::now);
let started =
should_sample(ordinal as u64, latency_sample_interval).then(Instant::now);
let value = if expected_tier == CacheTier::L2 {
read_l2_once(&cache, key_ordinal).await?
} else {
Some(read_l1_eventually(&cache, key_ordinal, client).await?)
};
if let Some(started) = started {
latency.record(started.elapsed());
}
latency.record_sampled(started);
primary.operations += 1;
if let Some(value) = value {
primary.bytes += value.len() as u128;
Expand Down Expand Up @@ -953,31 +911,6 @@ async fn read_l1_eventually(cache: &Cache, key_ordinal: usize, client: usize) ->
}
}

fn put_eventually(cache: &Cache, key: &[u8], value: &[u8]) -> io::Result<(u64, usize)> {
let deadline = Instant::now() + WRITE_RETRY_TIMEOUT;
let mut attempts = 0_usize;
loop {
attempts = attempts.saturating_add(1);
match cache.put(key, value) {
Ok(receipt) => return Ok((receipt, attempts)),
Err(error) if error.kind() == cache2::ErrorKind::Overloaded => {
if Instant::now() >= deadline {
return Err(io::Error::new(
io::ErrorKind::TimedOut,
"benchmark write did not enter bounded staging",
));
}
if attempts <= WRITE_YIELD_RETRIES {
std::thread::yield_now();
} else {
std::thread::sleep(RETRY_DELAY);
}
}
Err(error) => return Err(error.into()),
}
}
}

fn benchmark_key(ordinal: usize) -> [u8; 16] {
let mut key = [0_u8; 16];
key[..8].copy_from_slice(b"cache2::");
Expand Down Expand Up @@ -1138,61 +1071,3 @@ fn require_minimum_rate(name: &str, measurement: &Measurement) -> io::Result<()>
}
Ok(())
}

fn env_optional_f64(name: &str) -> io::Result<Option<f64>> {
match env::var(name) {
Ok(value) => {
let parsed = value
.parse::<f64>()
.map_err(|_| invalid(format!("{name} must be a finite non-negative number")))?;
if !parsed.is_finite() || parsed < 0.0 {
return Err(invalid(format!(
"{name} must be a finite non-negative number"
)));
}
Ok(Some(parsed))
}
Err(env::VarError::NotPresent) => Ok(None),
Err(error) => Err(invalid(format!("cannot read {name}: {error}"))),
}
}

fn env_usize(name: &str, default: usize) -> io::Result<usize> {
match env::var(name) {
Ok(value) => value
.parse()
.map_err(|_| invalid(format!("{name} must be an unsigned integer"))),
Err(env::VarError::NotPresent) => Ok(default),
Err(error) => Err(invalid(format!("cannot read {name}: {error}"))),
}
}

fn env_u32(name: &str, default: u32) -> io::Result<u32> {
match env::var(name) {
Ok(value) => value
.parse()
.map_err(|_| invalid(format!("{name} must be an unsigned integer"))),
Err(env::VarError::NotPresent) => Ok(default),
Err(error) => Err(invalid(format!("cannot read {name}: {error}"))),
}
}

fn env_bool(name: &str, default: bool) -> io::Result<bool> {
match env::var(name) {
Ok(value) if value == "true" || value == "1" => Ok(true),
Ok(value) if value == "false" || value == "0" => Ok(false),
Ok(_) => Err(invalid(format!("{name} must be true, false, 1, or 0"))),
Err(env::VarError::NotPresent) => Ok(default),
Err(error) => Err(invalid(format!("cannot read {name}: {error}"))),
}
}

fn sidecar(path: &Path, suffix: &str) -> PathBuf {
let mut value = path.as_os_str().to_owned();
value.push(suffix);
PathBuf::from(value)
}

fn invalid(message: impl Into<String>) -> io::Error {
io::Error::new(io::ErrorKind::InvalidInput, message.into())
}
Loading