diff --git a/.gitignore b/.gitignore index ce8b0c8..53dce5e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ /target *.cache +__pycache__/ diff --git a/BENCHMARK.md b/BENCHMARK.md index 9319c62..876c911 100644 --- a/BENCHMARK.md +++ b/BENCHMARK.md @@ -2,26 +2,63 @@ Compare revisions only on the same host, filesystem, device, toolchain, and workload. Keep raw output with the tested commit; this repository intentionally does not publish machine-specific historical numbers. +## Ablation experiments + +Start from a saved source tree and binary, including any uncommitted changes. Change one mechanism per experimental build, record its patch and binary checksum, and keep routing, workload, memory limits, admission ceilings, statistics, and I/O topology fixed. Run builds and timed workloads separately. Retain every sample, including regressions and failed runs; a correctness or memory-bound failure disqualifies a candidate regardless of throughput. + +Use isolated measurements to explain a cost before changing the architecture. For L1 accounting, compare same-key replacement with new-key eviction at a full fixed byte budget: replacement has no eviction-counter increment, while both paths transfer an exclusive memory charge. Measure each change separately and together to expose interactions. Check small and 16 KiB values, and retain data-validation and exact memory-accounting checks. An isolated speedup does not establish a whole-cache speedup. + +Before editing the measured production path, qualify the measurement with byte-identical A/A runs. Use the comparison procedure below, and stop if its uncertainty or drift gate fails. Qualify the chosen implementation with balanced baseline/candidate runs using the batch-plus-drain and mixed-turnover workloads below. Compare effective hit throughput, accepted writes, rejections, latency, CPU, and RSS together. Include a single-writer case and a longer sampled Adaptive run when the changed mechanism affects contention or allocation. Treat variation in an unaffected path as a noise control; repeat or lengthen measurements before attributing small differences to a change. Keep experimental switches and rejected candidates out of the production request path. + +## Noise control and paired comparisons + +Build and save each binary before starting measurements. Save its source snapshot or patch, toolchain, and workload environment alongside the output. Do not compile, run tests, or profile concurrently with a timed comparison. Keep the host, filesystem, I/O topology, statistics, memory limits, and power state fixed. On macOS laptops, connected AC power alone is insufficient: a charger can be connected while the battery continues discharging. + +The request benchmark retains the cold single-batch `put_drain` result. For steadier write measurements, set `CACHE_BENCH_WRITE_ROUNDS` to a positive fixed count and optionally set `CACHE_BENCH_WRITE_WARMUP_ROUNDS` (default 2). The additional `put_steady` phase overwrites the same key set after the cold batch, reuses the same writer threads and payload buffers, assigns work in fixed 64-operation chunks, and drains every round. Chunk assignment prevents one slowly scheduled worker from holding an entire static share of a short batch. Only the configured measurement rounds contribute to its elapsed time, operations, retries, and bytes. Thread creation, warmup, final payload checks, and close are outside that interval. Dispatch, admission, and drain are inside it. This measures repeated overwrite and, when enough bytes are written, Region turnover; it is a different workload from cold population. Both phases start their clock before releasing any writer. `CACHE_BENCH_WRITE_ONLY=true` skips the subsequent read/recovery phases so comparisons can focus on writing; use the full default benchmark and soak warm verification separately for qualification. + +Calibrate the round count in a pilot so the selected measured phase lasts at least five seconds, then freeze it across all builds. Increase warmup until startup allocation and page faults no longer dominate; do not silently substitute longer single batches for repeated short batches. A longer interval alone does not establish stability. Keep the original cold results, but do not infer small improvements from a single millisecond-scale batch. + +`scripts/compare-benchmarks.py` requires Python 3 and an environment JSON object with string keys and values. It runs saved binaries serially in alternating ABBA and BAAB blocks, saves every raw log, binary hash, power sample, and parsed result, and reports paired geometric changes with a deterministic 95% bootstrap interval over whole blocks. The default is eight blocks, or sixteen runs of each binary. The point estimate and interval use the paired blocks; marginal medians are also reported and need not imply exactly the same change. This interval describes repeated runs on the tested host, not uncertainty across hardware or workloads. + +```sh +python3 scripts/test_compare_benchmarks.py +python3 scripts/compare-benchmarks.py \ + --baseline /tmp/before/cache --candidate /tmp/before/cache \ + --env /tmp/write-environment.json --output /tmp/write-aa \ + --metric put_steady --aa +python3 scripts/compare-benchmarks.py \ + --baseline /tmp/before/cache --candidate /tmp/after/cache \ + --env /tmp/write-environment.json --output /tmp/write-ab \ + --metric put_steady +``` + +A/A requires byte-identical binaries. By default its entire interval must fit within ±3%, and both the spread of the ABBA block centers and the spread between same-binary repeats within each block must stay within 5%; either spread also fails A/B. Checking the repeats prevents alternating fast and slow runs from passing merely because pairing cancels their noise. Selected phases shorter than five seconds, benchmark failures, and missing metrics stop the run and retain evidence. On macOS, the runner records power state before and after each process and stops on battery power, discharging, or battery below 20%. Desktop Macs do not need a battery. Explicit thresholds are recorded in the manifest; select them before measuring and do not relax them after seeing a candidate's result. Run the same A/A control for each relevant workload, and repeat controls if the environment changes. Do not discard outliers or rerun only unfavorable samples to manufacture a pass. + +An external affinity or scheduling launcher can be supplied with repeated `--prefix` arguments, one argument at a time, including `--prefix=-a` for an option beginning with a dash. The runner applies the same prefix to both binaries and records it in the manifest. A launcher does not establish stability; its runs still need the same A/A qualification. + +For mixed turnover, pass `cache_soak` binaries and select `--metric get --metric hit --metric put`. `hit` counts validated application-visible hits per second, independently of library statistics, so fast misses cannot masquerade as useful read throughput. Set `CACHE_SOAK_ACTIVITY_COUNTERS=false` to measure the default library path; the harness still validates every hit, counts outcomes, samples latency, checks memory bounds, and can perform final warm verification. Keep those observer settings constant. Compare CPU, retries/rejections, latency, hit population, and RSS from the saved logs alongside throughput. Retain an optimization only after fresh confirmation, with a gain larger than the measured noise and no unexplained loss in the other qualification workloads. + ## Unified report format All benchmark targets emit a common fio-style report in addition to their existing `result ...` records. The human-readable section groups work into jobs and reports completed-operation rate, attempts, payload bandwidth, bytes, runtime, worker count, errors or overloads, and sampled latency. Every run ends with process CPU time, context switches, page faults, peak RSS, and pass or error status. Lines beginning with `report version=2` are the stable machine-readable form. They are whitespace-separated `key=value` records with these types: -| Type | Contents | -| ------------- | ------------------------------------------------------------------------------------------- | -| `header` | Benchmark identity, scenario, operating system, and architecture | -| `job` | IOPS, bandwidth, bytes, runtime, workers, errors, and latency distribution | -| `io` | Read/write submissions, completions, failures, cancellation, depth, timing, and I/O path | -| `cache` | Request outcomes, hits/misses, overload, promotion, rotation, and reclaim | -| `resources` | Managed memory, peak RSS, logical disk use, and L1/index/Region occupancy | -| `run` | Overall status, runtime, CPU, scheduler, faults, and peak RSS | +| Type | Contents | +| ---------------- | ---------------------------------------------------------------------------------------- | +| `header` | Benchmark identity, scenario, operating system, and architecture | +| `job` | IOPS, bandwidth, bytes, runtime, workers, errors, and latency distribution | +| `io` | Read/write submissions, completions, failures, cancellation, depth, timing, and I/O path | +| `cache` | Request outcomes, hits/misses, overload, promotion, rotation, and reclaim | +| `resources` | Managed memory, peak RSS, logical disk use, and L1/index/Region occupancy | +| `process_memory` | Soak phase, current RSS, and process peak RSS | +| `run` | Overall status, runtime, CPU, scheduler, faults, and peak RSS | -`mixed_workloads` and `cache_soak` emit the cache-specific records by default; `cache` emits them when `CACHE_BENCH_ACTIVITY_COUNTERS=true`. The `phase` field distinguishes independently reset cache instances, such as the initial write and warm-reopened L2 phases. Existing `result ...` lines are kept for qualification scripts and compatibility. +`mixed_workloads` and `cache_soak` emit the cache-specific records by default; soak records remain available with activity counters disabled, but disabled activity fields are zero; `cache` emits them when `CACHE_BENCH_ACTIVITY_COUNTERS=true`. The `phase` field distinguishes independently reset cache instances, such as the initial write and warm-reopened L2 phases. Existing `result ...` lines are kept for qualification scripts and compatibility. The `io` records count logical engine requests and positive runtime file operations. They distinguish buffered and direct paths, but they are not physical-device I/O counters; use OS or device telemetry when that distinction matters. -Latency uses a fixed 65-bucket log2 histogram, so measurement memory is bounded and independent of operation count. Percentiles are bucket upper bounds; `avg~` and `stdev~` are estimates from bucket midpoints. The default sampling interval is 16 operations. Set the target's latency sampling interval to zero to disable it. Non-soak workers update thread-local histograms; soak workers use three relaxed atomic updates per sampled operation. Sampling otherwise adds one timer start and one elapsed-time read to each selected operation. +Latency uses a fixed 65-bucket log2 histogram, so measurement memory is bounded and independent of operation count. Percentiles are bucket upper bounds; `avg~` and `stdev~` are estimates from bucket midpoints. The default sampling interval is 16 operations. Set the target's latency sampling interval to zero to disable it. Non-soak workers update thread-local histograms; each soak worker owns separate counters and histograms, with three uncontended relaxed atomic updates per sampled operation. Periodic snapshots merge the bounded worker records, and final bandwidth uses payload bytes accumulated locally by each worker, independently of library statistics. Sampling otherwise adds one timer start and one elapsed-time read to each selected operation. ## Request-path benchmark @@ -29,7 +66,7 @@ Latency uses a fixed 65-bucket log2 histogram, so measurement memory is bounded cargo +1.98.0 x bench --locked --bench cache ``` -The benchmark measures accepted puts plus drain, resident L1 reads, warm close, and successful L2 reads with best-effort promotion. Values encode their key ordinal, so a wrong-key result fails the run. Harness retries of `ErrorKind::Overloaded` are workload setup, not library behavior. +The benchmark measures accepted puts plus drain, resident L1 reads, warm close, and successful L2 reads with best-effort promotion. The `put_admission` result separately reports `admission_elapsed_ns` and `drain_elapsed_ns`, while `put_drain` retains their combined throughput. Values encode their key ordinal, so a wrong-key result fails the run. Harness retries of `ErrorKind::Overloaded` are workload setup, not library behavior. The main controls are grouped below. See `benchmarks/cache/main.rs` for defaults and validation rules. @@ -40,12 +77,13 @@ The main controls are grouped below. See `benchmarks/cache/main.rs` for defaults | Concurrency | `CACHE_BENCH_CLIENTS`, `CACHE_BENCH_WRITE_CLIENTS`, `CACHE_BENCH_APPEND_SHARDS`, `CACHE_BENCH_POSIX_READ_WORKERS`, `CACHE_BENCH_READ_IO_WAIT_CAPACITY`, `CACHE_BENCH_READ_IO_WAIT_TIMEOUT_US`, `CACHE_BENCH_POSIX_WRITE_WORKERS`, `CACHE_BENCH_POSIX_RECLAIM_WORKERS` | | I/O path | `CACHE_BENCH_IO_ENGINE`, `CACHE_BENCH_IO_MODE`, `CACHE_BENCH_DIR` | | Cache policy | `CACHE_BENCH_L1_EVICTION=clock\|s3-fifo`, `CACHE_BENCH_HOT_ENTRIES`, `CACHE_BENCH_HOT_READ_INTERVAL` | -| Measurement | `CACHE_BENCH_READ_LATENCY_SAMPLE_INTERVAL` (default 16; zero disables), `CACHE_BENCH_ACTIVITY_COUNTERS` (default false; true adds cache/I/O/resource records) | +| Batching | `CACHE_BENCH_WRITE_FLUSH_THRESHOLD_BYTES` (defaults to the library threshold; the effective byte value is printed) | +| Measurement | `CACHE_BENCH_READ_LATENCY_SAMPLE_INTERVAL` (default 16; zero disables), `CACHE_BENCH_ACTIVITY_COUNTERS` (default false), `CACHE_BENCH_WRITE_ROUNDS` (default 0), `CACHE_BENCH_WRITE_WARMUP_ROUNDS` (default 2), `CACHE_BENCH_WRITE_ONLY` (default false) | | Gates | `CACHE_BENCH_MIN_PUT_OPS`, `CACHE_BENCH_MIN_RESIDENT_L1_OPS`, `CACHE_BENCH_MIN_L2_OPS`, `CACHE_BENCH_MAX_WARM_CLOSE_MS` | Request and latency collection are controlled independently by `CACHE_BENCH_REQUEST_STATS` (complete request counters, default false), `CACHE_BENCH_L1_LATENCY_SAMPLE_INTERVAL`, `CACHE_BENCH_L2_LATENCY_SAMPLE_INTERVAL`, and `CACHE_BENCH_MUTATION_LATENCY_SAMPLE_INTERVAL` (independent: 0 off, 1 full, greater values sample with that mean interval; default 0), `CACHE_BENCH_IO_LATENCY` (full engine latency, default false), and `CACHE_BENCH_STATS_SHARDS` (default 16). These instrument the library itself. `CACHE_BENCH_READ_LATENCY_SAMPLE_INTERVAL` remains the independent benchmark observer and should be held constant across comparisons. The effective statistics settings are printed with each run. Compare disabled, counters-only, full and sampled modes on the same workload, keeping actual hit/overload populations in view. -`CACHE_BENCH_ACTIVITY_COUNTERS=true` adds cache accounting on the measured request path. Use the same setting for baseline and candidate runs. The `result phase=read_io` line then also reports `in_flight_peak`, `configured_in_flight`, `slot_wait_ns`, `payload_bytes`, and `read_amp` (engine read bytes divided by served payload bytes). +`CACHE_BENCH_ACTIVITY_COUNTERS=true` adds cache accounting on the measured request path. Use the same setting for baseline and candidate runs. The `result phase=read_io` line then also reports `in_flight_peak`, `configured_in_flight`, `slot_wait_ns`, `payload_bytes`, and `read_amp` (engine read bytes divided by served payload bytes). The cache report also includes `write_buffer_rejections` alongside total `write_rejections`. For device measurements, use a data set larger than host RAM and no larger than half of L2 capacity. Run baseline and candidate in alternating order at least five times, compare medians, and retain every sample. Throughput does not replace correctness, overload, memory, or latency checks. @@ -85,7 +123,7 @@ Request throughput includes generation, value population, cache calls, and hit v cargo +1.98.0 x bench --locked --bench cache_soak ``` -The default short run mixes reads, writes, deletes, Region rotation, reclaim, and 256 B, 4 KiB, 16 KiB, and 256 KiB values. Each hit is checked for key, version, length, and payload. Older valid values are counted as stale; future, wrong-key, malformed, I/O-error, or managed-memory/RSS bound violations fail the run. +The default short run mixes reads, writes, deletes, Region rotation, reclaim, and 256 B, 4 KiB, 16 KiB, and 256 KiB values. The read stream retains its shared ordinal and cyclic key permutation. Reader and writer ordinals are isolated from each other and from the stop flag, and each worker owns its outcome counters and latency histogram. Compare revisions using the same harness version, since observer overhead also consumes CPU. Each hit is checked for key, version, length, and payload. Older valid values are counted as stale; future, wrong-key, malformed, I/O-error, or managed-memory/RSS bound violations fail the run. Use `CACHE_SOAK_*` to select duration, sample interval, capacity, L1 and managed memory, key count, mixed value sizes, client counts, worker topology, I/O path, and output directory. Important qualification switches are: @@ -95,8 +133,14 @@ Use `CACHE_SOAK_*` to select duration, sample interval, capacity, L1 and managed - `CACHE_SOAK_REQUIRE_PATH_COVERAGE=true` to require writes, deletes, L2 reads, rotation, reclaim, and warm recovery; - `CACHE_SOAK_REQUIRE_REINSERT_COVERAGE=true` for a focused read-heavy run that must both reinsert a hot record and exhaust the bounded reinsert budget. +`CACHE_SOAK_ACTIVITY_COUNTERS` defaults to true for diagnostic coverage. Set it to false for the library default path. `CACHE_SOAK_REQUIRE_PATH_COVERAGE` and `CACHE_SOAK_REQUIRE_REINSERT_COVERAGE` require activity counters and are rejected when counters are disabled; value validation, outcome counting, final warm verification, and memory checks remain active. Workers use a common start gate, with timing beginning before they are released. + `CACHE_SOAK_LATENCY_SAMPLE_INTERVAL` controls put, get, and delete latency sampling. It defaults to 16; zero disables latency sampling. +The soak samples current RSS on Linux and macOS and enforces the existing memory allowance during the workload. Other platforms report zero when current RSS is unavailable. `process_memory` records expose `current_rss_bytes` and `peak_rss_bytes` at open, periodic samples, and warm-close/drop/reopen boundaries. Reopen explicitly drops the closed cache before allocating its replacement, so the reported peak does not include two simultaneously retained cache instances. Allocator-retained pages still count toward RSS even after the cache's managed charges are released. + +For concurrent 16 KiB turnover, compare eight writers and four readers with `CACHE_SOAK_VALUE_BYTES=16384`, `CACHE_SOAK_KEYS=16384`, `CACHE_SOAK_CAPACITY_MIB=256`, `CACHE_SOAK_L1_CAPACITY_MIB=64`, `CACHE_SOAK_MANAGED_MEMORY_LIMIT_MIB=768`, and `CACHE_SOAK_APPEND_SHARDS=4`. Hold the duration, sampling interval, I/O topology, and final warm verification constant. Run Disabled and Adaptive admission separately in alternating revision order at least five times. Compare get throughput together with L1 hits, L2 hits, misses, accepted puts, overloads, latency, CPU, managed peak, and phase RSS: faster fills can change the read-hit population, and fast misses alone are not a throughput improvement. Pair this with `cache` runs using 8,192 and 65,536 entries, 16 KiB values, eight write clients, four append shards, and the default flush threshold to check that batch-write-plus-drain gains remain. A longer sampled Adaptive run should distinguish a stable resident footprint from continuing allocator retention. + The I/O topology uses the shared backend-specific controls below, with `CACHE_SOAK` as the prefix. Repeat sizes in `CACHE_SOAK_VALUE_BYTES` to weight a production distribution. Use a short matrix rather than one oversized run: small-capacity turnover, high-cardinality mixed sizes, read-heavy reinsertion, and CLOCK/S3-FIFO A/B. Longer soaks are optional follow-up evidence, not a default gate. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7643719..c43f2d9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,6 +15,8 @@ rustup toolchain install nightly --profile minimal --component rustfmt,clippy cargo install --locked cargo-deny hawkeye taplo-cli typos-cli ``` +Python 3 is required for the benchmark comparison tools and their tests, which are included in `cargo x test`. + Before submitting a pull request, run: ```sh @@ -23,7 +25,7 @@ cargo x test cargo x lint ``` -`check` covers the workspace and optional features, `test` includes the extended library tests, and `lint` checks formatting, code, documentation, packaging, and dependencies. Use `cargo x lint --fix` to apply supported automatic fixes, then review the diff. +`check` covers the workspace and optional features, `test` includes the extended library and benchmark-tool tests, and `lint` checks formatting, code, documentation, packaging, and dependencies. Use `cargo x lint --fix` to apply supported automatic fixes, then review the diff. Cover observable behavior changes with tests. See [BENCHMARK.md](BENCHMARK.md) for performance workloads and qualification. diff --git a/benchmarks/cache/main.rs b/benchmarks/cache/main.rs index 98bce7e..a915d43 100644 --- a/benchmarks/cache/main.rs +++ b/benchmarks/cache/main.rs @@ -51,6 +51,8 @@ use cache2::StartupMode; use cache2::StorageOptions; use cache2::Value; +mod steady; + const MIB: usize = 1024 * 1024; const REGION_BYTES: usize = 32 * MIB; const L1_ENTRY_BYTES: usize = 256 * 1024; @@ -74,10 +76,14 @@ struct BenchConfig { l1_capacity_bytes: usize, managed_memory_limit_bytes: usize, append_shards: u32, + write_flush_threshold_bytes: usize, read_io_wait_capacity: usize, read_io_wait_timeout: Duration, read_latency_sample_interval: usize, write_clients: usize, + write_rounds: usize, + write_warmup_rounds: usize, + write_only: bool, clients: usize, io_engine: IoEngineOptions, fill_control: cache2::FillControlOptions, @@ -111,6 +117,13 @@ 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 write_rounds = env_usize("CACHE_BENCH_WRITE_ROUNDS", 0)?; + let write_warmup_rounds = env_usize("CACHE_BENCH_WRITE_WARMUP_ROUNDS", 2)?; + let write_only = env_bool("CACHE_BENCH_WRITE_ONLY", false)?; + entries + .checked_mul(write_rounds) + .and_then(|_| write_rounds.checked_add(write_warmup_rounds)) + .ok_or_else(|| invalid("benchmark write round count is too large"))?; 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 { @@ -224,10 +237,17 @@ impl BenchConfig { l1_capacity_bytes, managed_memory_limit_bytes, append_shards, + write_flush_threshold_bytes: env_usize( + "CACHE_BENCH_WRITE_FLUSH_THRESHOLD_BYTES", + RuntimeOptions::default().write_flush_threshold_bytes, + )?, read_io_wait_capacity, read_io_wait_timeout, read_latency_sample_interval, write_clients, + write_rounds, + write_warmup_rounds, + write_only, clients, io_engine, fill_control: benchmarks::config::fill_control_from_env("CACHE_BENCH")?, @@ -252,6 +272,7 @@ impl BenchConfig { println!("fill_control={:?}", self.fill_control); options.io_mode = self.io_mode; options.append_shards = self.append_shards; + options.write_flush_threshold_bytes = self.write_flush_threshold_bytes; options.l1_capacity_bytes = self.l1_capacity_bytes; options.l1_eviction_policy = self.l1_eviction_policy; options.managed_memory_limit_bytes = self.managed_memory_limit_bytes; @@ -284,6 +305,14 @@ struct Measurement { checksum: u64, } +#[derive(Default)] +struct WriteTotals { + checksum: u64, + attempts: usize, + throttled: usize, + operations: usize, +} + struct WriteAdmission { measurement: Measurement, attempts: usize, @@ -350,6 +379,10 @@ async fn run(config: BenchConfig) -> io::Result<()> { println!("C² cache benchmark"); println!("stats={:?}", config.stats); + println!( + "write_flush_threshold_bytes={}", + config.write_flush_threshold_bytes + ); println!( "entries={} index_slots={} index_load={:.1}% resident_entries={} hot_entries={} hot_read_interval={} value={} B data={:.1} MiB l1_capacity={:.1} MiB initial_l1={:.1} MiB managed_memory_limit={:.1} MiB append_shards={} configured_read_in_flight={} read_wait_capacity={} read_wait_timeout_us={} read_latency_sample_interval={} write_clients={} read_clients={} l2_clients={} l1_entry_eligible={} l1_eviction={:?} engine={:?} mode={:?} activity_counters={}", config.entries, @@ -398,9 +431,11 @@ async fn run(config: BenchConfig) -> io::Result<()> { config.value_bytes, config.write_clients, )?; + let admission_elapsed = write.measurement.elapsed; let drain_started = Instant::now(); cache.drain().await?; - write.measurement.elapsed += drain_started.elapsed(); + let drain_elapsed = drain_started.elapsed(); + write.measurement.elapsed += drain_elapsed; let write_detailed = config .stats .activity_counters @@ -417,11 +452,13 @@ async fn run(config: BenchConfig) -> io::Result<()> { 0, ); println!( - "result phase=put_admission clients={} attempts={} retries={} throttled={}", + "result phase=put_admission clients={} attempts={} retries={} throttled={} admission_elapsed_ns={} drain_elapsed_ns={}", config.write_clients, write.attempts, write.attempts.saturating_sub(config.entries), write.throttled_writes, + admission_elapsed.as_nanos(), + drain_elapsed.as_nanos(), ); if let Some(detailed) = &write_detailed { emit_cache_report( @@ -433,6 +470,15 @@ async fn run(config: BenchConfig) -> io::Result<()> { ); } + if config.write_rounds != 0 { + steady::run(Arc::clone(&cache), &config).await?; + } + if config.write_only { + require_minimum_rate("CACHE_BENCH_MIN_PUT_OPS", &write.measurement)?; + cache.close_fast().await?; + return Ok(()); + } + let started = Instant::now(); cache.close_warm().await?; drop(cache); @@ -706,46 +752,33 @@ fn concurrent_writes( clients: usize, ) -> io::Result { let barrier = Arc::new(std::sync::Barrier::new(clients + 1)); + let start = Arc::new(std::sync::Barrier::new(clients + 1)); std::thread::scope(|scope| { let mut handles = Vec::with_capacity(clients); for client in 0..clients { let cache = Arc::clone(&cache); let barrier = Arc::clone(&barrier); + let start = Arc::clone(&start); handles.push(scope.spawn(move || { let mut value = vec![0xa5; value_bytes]; barrier.wait(); - let mut checksum = 0_u64; - let mut attempts = 0_usize; - let mut throttled_writes = 0_usize; - 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), - WRITE_RETRY_TIMEOUT, - )?; - attempts = attempts.saturating_add(operation_attempts); - throttled_writes = - throttled_writes.saturating_add(usize::from(operation_attempts > 1)); - checksum = checksum.wrapping_add(receipt.rotate_left((ordinal % 64) as u32)); - } - Ok::<_, io::Error>((checksum, attempts, throttled_writes)) + start.wait(); + write_ordinals(&cache, &mut value, (client..entries).step_by(clients)) })); } barrier.wait(); let started = Instant::now(); + start.wait(); let mut checksum = 0_u64; let mut attempts = 0_usize; let mut throttled_writes = 0_usize; for handle in handles { - let (client_checksum, client_attempts, client_throttled_writes) = handle + let totals = handle .join() .map_err(|_| io::Error::other("benchmark writer panicked"))??; - checksum = checksum.wrapping_add(client_checksum); - attempts = attempts.saturating_add(client_attempts); - throttled_writes = throttled_writes.saturating_add(client_throttled_writes); + checksum = checksum.wrapping_add(totals.checksum); + attempts = attempts.saturating_add(totals.attempts); + throttled_writes = throttled_writes.saturating_add(totals.throttled); } Ok(WriteAdmission { measurement: Measurement { @@ -760,6 +793,31 @@ fn concurrent_writes( }) } +fn write_ordinals( + cache: &Cache, + value: &mut [u8], + ordinals: impl Iterator, +) -> io::Result { + let mut totals = WriteTotals::default(); + for ordinal in ordinals { + let key = benchmark_key(ordinal); + value[..8].copy_from_slice(&(ordinal as u64).to_le_bytes()); + let (receipt, attempts) = put_eventually( + cache, + black_box(&key), + black_box(&*value), + WRITE_RETRY_TIMEOUT, + )?; + totals.operations += 1; + totals.attempts = totals.attempts.saturating_add(attempts); + totals.throttled += usize::from(attempts > 1); + totals.checksum = totals + .checksum + .wrapping_add(receipt.rotate_left((ordinal % 64) as u32)); + } + Ok(totals) +} + async fn concurrent_reads( cache: Arc, key_range: Range, @@ -776,12 +834,15 @@ async fn concurrent_reads( .and_then(|count| u32::try_from(count).ok()) .ok_or_else(|| io::Error::other("benchmark client count exceeds async barrier capacity"))?; let barrier = Arc::new(Barrier::new(participants)); + let start = Arc::new(Barrier::new(participants)); let mut handles = Vec::with_capacity(clients); for client in 0..clients { let cache = Arc::clone(&cache); let barrier = Arc::clone(&barrier); + let start = Arc::clone(&start); handles.push(tokio::spawn(async move { barrier.wait().await; + start.wait().await; let mut primary = TierCounts::default(); let mut sampled = TierCounts::default(); let mut latency = LatencyHistogram::default(); @@ -822,6 +883,7 @@ async fn concurrent_reads( } barrier.wait().await; let started = Instant::now(); + start.wait().await; let mut primary = TierCounts::default(); let mut sampled = TierCounts::default(); let mut latency = LatencyHistogram::default(); diff --git a/benchmarks/cache/steady.rs b/benchmarks/cache/steady.rs new file mode 100644 index 0000000..48f0350 --- /dev/null +++ b/benchmarks/cache/steady.rs @@ -0,0 +1,158 @@ +// Copyright 2026 ScopeDB, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; +use std::sync::mpsc; + +use super::*; + +// Dropping command senders also stops workers on an early error. Unlike a +// reusable barrier, a disconnected channel cannot strand the other writers. +#[derive(Default)] +struct Writers { + commands: Vec>, + results: Vec>>, + threads: Vec>, +} + +impl Drop for Writers { + fn drop(&mut self) { + self.commands.clear(); + for thread in self.threads.drain(..) { + let _ = thread.join(); + } + } +} + +pub async fn run(cache: Arc, config: &BenchConfig) -> io::Result<()> { + let mut writers = Writers::default(); + let next = Arc::new(AtomicUsize::new(0)); + for _ in 0..config.write_clients { + let (command, commands) = mpsc::sync_channel(1); + let cache = Arc::clone(&cache); + let (completed, results) = mpsc::sync_channel(1); + let entries = config.entries; + let next = Arc::clone(&next); + let value_bytes = config.value_bytes; + writers.threads.push(std::thread::spawn(move || { + let mut value = vec![0xa5; value_bytes]; + let _ = completed.send(Ok(WriteTotals::default())); + while commands.recv().is_ok() { + // Fixed small chunks let faster workers keep doing useful work + // instead of waiting for one slow core's entire static share. + let ordinals = std::iter::from_fn(|| { + let first = next.fetch_add(64, Ordering::Relaxed); + (first < entries).then_some(first..first.saturating_add(64).min(entries)) + }) + .flatten(); + let result = write_ordinals(&cache, &mut value, ordinals); + let failed = result.is_err(); + if completed.send(result).is_err() || failed { + break; + } + } + })); + writers.commands.push(command); + writers.results.push(results); + } + for results in &writers.results { + results + .recv() + .map_err(|_| io::Error::other("steady writer failed to start"))??; + } + let mut measurement = Measurement { + elapsed: Duration::ZERO, + operations: config.entries * config.write_rounds, + bytes: (config.entries as u128) + * (config.write_rounds as u128) + * (config.value_bytes as u128), + checksum: 0, + }; + let mut attempts = 0_usize; + let mut throttled = 0_usize; + let mut drain_elapsed = Duration::ZERO; + for round in 0..config.write_warmup_rounds + config.write_rounds { + // Timestamp before dispatch: no accepted operation can precede it. + next.store(0, Ordering::Relaxed); + let started = Instant::now(); + for command in &writers.commands { + command + .send(()) + .map_err(|_| io::Error::other("steady writer stopped"))?; + } + let mut operations = 0_usize; + for results in &writers.results { + let totals = results + .recv() + .map_err(|_| io::Error::other("steady writer panicked"))??; + operations += totals.operations; + if round >= config.write_warmup_rounds { + attempts = attempts.saturating_add(totals.attempts); + throttled += totals.throttled; + measurement.checksum = measurement.checksum.wrapping_add(totals.checksum); + } + } + if operations != config.entries { + return Err(io::Error::other( + "steady round did not complete its fixed operation count", + )); + } + let drain_started = Instant::now(); + cache.drain().await?; + if round >= config.write_warmup_rounds { + drain_elapsed += drain_started.elapsed(); + measurement.elapsed += started.elapsed(); + } + } + report( + "put_steady", + "steady put + drain", + "write", + config.write_clients, + &measurement, + throttled as u64, + None, + 0, + ); + println!( + "result phase=put_steady_admission rounds={} warmup_rounds={} attempts={} retries={} throttled={} admission_elapsed_ns={} drain_elapsed_ns={}", + config.write_rounds, + config.write_warmup_rounds, + attempts, + attempts.saturating_sub(measurement.operations), + throttled, + measurement.elapsed.saturating_sub(drain_elapsed).as_nanos(), + drain_elapsed.as_nanos() + ); + // Validation is outside the timed interval and remains active with all + // library statistics disabled. Misses are allowed by the cache contract. + let mut verified = 0; + for ordinal in (0..config.entries).step_by((config.entries / 1024).max(1)) { + if let Some(value) = cache.get(benchmark_key(ordinal)).await? { + verify_value(ordinal, &value)?; + if value.len() != config.value_bytes || value[8..].iter().any(|byte| *byte != 0xa5) { + return Err(io::Error::other("steady write payload mismatch")); + } + verified += 1; + } + } + if verified == 0 { + return Err(io::Error::other( + "steady write verification found no values", + )); + } + println!("result phase=put_steady_verify hits={verified}"); + Ok(()) +} diff --git a/benchmarks/cache_soak/main.rs b/benchmarks/cache_soak/main.rs index 48e9ff8..ce1aff4 100644 --- a/benchmarks/cache_soak/main.rs +++ b/benchmarks/cache_soak/main.rs @@ -86,6 +86,7 @@ struct SoakConfig { latency_sample_interval: usize, warm_reopen: bool, final_warm_verify: bool, + activity_counters: bool, require_path_coverage: bool, require_reinsert_coverage: bool, io_engine: IoEngineOptions, @@ -150,14 +151,20 @@ impl SoakConfig { } else { 0 }; + let activity_counters = env_bool("CACHE_SOAK_ACTIVITY_COUNTERS", true)?; let require_path_coverage = env_bool("CACHE_SOAK_REQUIRE_PATH_COVERAGE", false)?; let require_reinsert_coverage = env_bool("CACHE_SOAK_REQUIRE_REINSERT_COVERAGE", false)?; + if !activity_counters && (require_path_coverage || require_reinsert_coverage) { + return Err(invalid("soak path coverage requires activity counters")); + } let io_mode = parse_io_mode("CACHE_SOAK_IO_MODE")?; let l1_eviction_policy = parse_l1_eviction_policy("CACHE_SOAK_L1_EVICTION")?; let directory = env::var_os("CACHE_SOAK_DIR") .map(PathBuf::from) .unwrap_or_else(env::temp_dir); - if duration.is_zero() + if Instant::now().checked_add(duration).is_none() + || Instant::now().checked_add(sample_period).is_none() + || duration.is_zero() || sample_period.is_zero() || value_bytes.is_empty() || value_bytes @@ -191,6 +198,7 @@ impl SoakConfig { latency_sample_interval, warm_reopen, final_warm_verify, + activity_counters, require_path_coverage, require_reinsert_coverage, io_engine, @@ -218,12 +226,13 @@ impl SoakConfig { options.l1_capacity_bytes = self.l1_capacity_bytes; options.l1_eviction_policy = self.l1_eviction_policy; options.managed_memory_limit_bytes = self.managed_memory_limit_bytes; - options.stats.activity_counters = true; + options.stats.activity_counters = self.activity_counters; options } } #[derive(Default)] +#[repr(align(128))] struct SoakCounters { writes: AtomicU64, deletes: AtomicU64, @@ -235,9 +244,13 @@ struct SoakCounters { put_latency: AtomicLatencyHistogram, delete_latency: AtomicLatencyHistogram, get_latency: AtomicLatencyHistogram, + // Workers accumulate payload bytes locally and publish once at exit. Job + // bandwidth must not depend on optional library activity counters. + written_bytes: AtomicU64, + read_bytes: AtomicU64, } -#[derive(Clone)] +#[derive(Clone, Default)] struct CounterSnapshot { writes: u64, deletes: u64, @@ -249,6 +262,8 @@ struct CounterSnapshot { put_latency: LatencyHistogram, delete_latency: LatencyHistogram, get_latency: LatencyHistogram, + written_bytes: u64, + read_bytes: u64, } #[derive(Default)] @@ -258,23 +273,48 @@ struct WarmVerification { misses: u64, } -impl SoakCounters { - fn snapshot(&self) -> CounterSnapshot { - CounterSnapshot { - writes: self.writes.load(Ordering::Relaxed), - deletes: self.deletes.load(Ordering::Relaxed), - write_rejections: self.write_rejections.load(Ordering::Relaxed), - delete_rejections: self.delete_rejections.load(Ordering::Relaxed), - hits: self.hits.load(Ordering::Relaxed), - stale_hits: self.stale_hits.load(Ordering::Relaxed), - misses: self.misses.load(Ordering::Relaxed), - put_latency: self.put_latency.snapshot(), - delete_latency: self.delete_latency.snapshot(), - get_latency: self.get_latency.snapshot(), - } +fn counter_snapshot(workers: &[SoakCounters]) -> CounterSnapshot { + let mut total = CounterSnapshot::default(); + for worker in workers { + total.writes = total + .writes + .saturating_add(worker.writes.load(Ordering::Relaxed)); + total.deletes = total + .deletes + .saturating_add(worker.deletes.load(Ordering::Relaxed)); + total.write_rejections = total + .write_rejections + .saturating_add(worker.write_rejections.load(Ordering::Relaxed)); + total.delete_rejections = total + .delete_rejections + .saturating_add(worker.delete_rejections.load(Ordering::Relaxed)); + total.hits = total + .hits + .saturating_add(worker.hits.load(Ordering::Relaxed)); + total.stale_hits = total + .stale_hits + .saturating_add(worker.stale_hits.load(Ordering::Relaxed)); + total.misses = total + .misses + .saturating_add(worker.misses.load(Ordering::Relaxed)); + total.written_bytes = total + .written_bytes + .saturating_add(worker.written_bytes.load(Ordering::Relaxed)); + total.read_bytes = total + .read_bytes + .saturating_add(worker.read_bytes.load(Ordering::Relaxed)); + total.put_latency.merge(worker.put_latency.snapshot()); + total.delete_latency.merge(worker.delete_latency.snapshot()); + total.get_latency.merge(worker.get_latency.snapshot()); } + total } +// Keep the shared writer ticket away from the frequently read stop flag and +// other clients' read-only state on 64-byte and 128-byte cache-line hosts. +#[repr(align(128))] +struct WorkOrdinal(AtomicU64); + struct ResourceSample { detailed: DetailedCacheSnapshot, logical_bytes: u64, @@ -307,6 +347,7 @@ fn run_benchmark() -> io::Result<()> { let peak_disk_bytes = storage.peak_disk_bytes(); let cache_config = CacheConfig::new(storage, config.runtime_options())?; let mut cache = open_cache(&runtime, &files, &cache_config)?; + report_process_memory("cold_open")?; let key_count = u64::try_from(config.key_count).map_err(|_| invalid("soak key count exceeds u64"))?; let value_size_count = u64::try_from(config.value_bytes.len()) @@ -317,7 +358,11 @@ fn run_benchmark() -> io::Result<()> { populate_for_warm_reopen(&cache, &config, &expected, key_count, value_size_count)?; runtime.block_on(cache.drain())?; runtime.block_on(cache.close_warm())?; + report_process_memory("prepare_close")?; + drop(cache); + report_process_memory("prepare_drop")?; cache = open_cache(&runtime, &files, &cache_config)?; + report_process_memory("prepare_open")?; if cache.startup_mode() != StartupMode::Warm { return Err(io::Error::other( "soak warm-reopen preparation did not recover a clean image", @@ -327,22 +372,21 @@ fn run_benchmark() -> io::Result<()> { } else { 0 }; - let next_write = AtomicU64::new(first_write); - let next_read = AtomicU64::new(0); + let next_write = WorkOrdinal(AtomicU64::new(first_write)); + let next_read = WorkOrdinal(AtomicU64::new(0)); let stop = AtomicBool::new(false); let client_count = config .writers .checked_add(config.readers) .ok_or_else(|| invalid("soak client count is too large"))?; - let counters = SoakCounters::default(); - let started = Instant::now(); - let deadline = started - .checked_add(config.duration) - .ok_or_else(|| invalid("soak deadline is too far in the future"))?; + let counters: Vec<_> = (0..client_count).map(|_| SoakCounters::default()).collect(); + let ready = std::sync::Barrier::new(client_count + 1); + let start = std::sync::Barrier::new(client_count + 1); + let mut max_managed_memory = 0_usize; println!( - "C² soak duration={}s capacity={:.1}MiB l1_capacity={:.1}MiB managed_memory_limit={:.1}MiB values={} keys={} append_shards={} writers={} readers={} operation_interval_us={} latency_sample_interval={} warm_reopen={} final_warm_verify={} require_path_coverage={} require_reinsert_coverage={} delete_interval={} l1_eviction={:?} engine={:?} mode={:?} peak_disk={} rss_slack={} rss_reopen_allowance={} data={}", + "C² soak duration={}s capacity={:.1}MiB l1_capacity={:.1}MiB managed_memory_limit={:.1}MiB values={} keys={} append_shards={} writers={} readers={} operation_interval_us={} latency_sample_interval={} warm_reopen={} final_warm_verify={} activity_counters={} require_path_coverage={} require_reinsert_coverage={} delete_interval={} l1_eviction={:?} engine={:?} mode={:?} peak_disk={} rss_slack={} rss_reopen_allowance={} data={}", config.duration.as_secs(), config.capacity_bytes as f64 / MIB as f64, config.l1_capacity_bytes as f64 / MIB as f64, @@ -361,6 +405,7 @@ fn run_benchmark() -> io::Result<()> { config.latency_sample_interval, config.warm_reopen, config.final_warm_verify, + config.activity_counters, config.require_path_coverage, config.require_reinsert_coverage, DELETE_INTERVAL, @@ -373,19 +418,28 @@ fn run_benchmark() -> io::Result<()> { files.data().display(), ); - std::thread::scope(|scope| -> io::Result<()> { + let started = std::thread::scope(|scope| -> io::Result { let mut workers = Vec::with_capacity(client_count); - for _ in 0..config.writers { - workers.push(scope.spawn(|| { + for counters in &counters[..config.writers] { + let next_write = &next_write.0; + let ready = &ready; + let start = &start; + let cache = &cache; + let config = &config; + let expected = &expected; + let stop = &stop; + workers.push(scope.spawn(move || { + ready.wait(); + start.wait(); let result = run_writer( - &cache, - &config, - &expected, + cache, + config, + expected, key_count, value_size_count, - &next_write, - &stop, - &counters, + next_write, + stop, + counters, ); if result.is_err() { stop.store(true, Ordering::Release); @@ -397,11 +451,15 @@ fn run_benchmark() -> io::Result<()> { let cache = &cache; let config = &config; let expected = &expected; - let next_read = &next_read; let stop = &stop; - let counters = &counters; + let next_read = &next_read.0; + let counters = &counters[config.writers + reader_id]; let runtime = runtime.handle().clone(); + let ready = &ready; + let start = &start; workers.push(scope.spawn(move || { + ready.wait(); + start.wait(); let result = run_reader( cache, config, @@ -421,9 +479,19 @@ fn run_benchmark() -> io::Result<()> { })); } + ready.wait(); + let started = Instant::now(); + // Every exit after readiness must release the start gate, including + // an unrepresentable deadline, so scoped workers cannot be stranded. + let Some(deadline) = started.checked_add(config.duration) else { + stop.store(true, Ordering::Release); + start.wait(); + return Err(invalid("soak deadline is too far in the future")); + }; + start.wait(); let mut next_sample = started .checked_add(config.sample_period) - .ok_or_else(|| invalid("soak sample deadline is too far in the future"))?; + .unwrap_or(deadline); let mut sample_error = None; while Instant::now() < deadline && !stop.load(Ordering::Acquire) { let wake_at = min(next_sample, deadline); @@ -440,13 +508,18 @@ fn run_benchmark() -> io::Result<()> { config.rss_reopen_allowance_bytes, ) { Ok(sample) => { + if let Err(error) = report_process_memory("workload_sample") { + sample_error = Some(error); + stop.store(true, Ordering::Release); + break; + } max_managed_memory = max_managed_memory .max(sample.detailed.summary.managed_memory_peak_bytes); report_sample( false, started.elapsed(), Duration::ZERO, - counters.snapshot(), + counter_snapshot(&counters), &sample, max_managed_memory, &config, @@ -476,7 +549,7 @@ fn run_benchmark() -> io::Result<()> { } match sample_error.or(worker_error) { Some(error) => Err(error), - None => Ok(()), + None => Ok(started), } })?; @@ -492,7 +565,7 @@ fn run_benchmark() -> io::Result<()> { config.rss_reopen_allowance_bytes, )?; max_managed_memory = max_managed_memory.max(sample.detailed.summary.managed_memory_peak_bytes); - let final_counters = counters.snapshot(); + let final_counters = counter_snapshot(&counters); if config.require_path_coverage { validate_measured_path_coverage(&final_counters, &sample)?; } @@ -500,8 +573,15 @@ fn run_benchmark() -> io::Result<()> { validate_reinsert_coverage(&sample)?; } if config.final_warm_verify { + report_process_memory("workload_complete")?; runtime.block_on(cache.close_warm())?; + report_process_memory("warm_close")?; + // Release the old instance's staging and L1 before allocating the + // replacement. Assignment alone keeps both alive during open_cache. + drop(cache); + report_process_memory("warm_drop")?; cache = open_cache(&runtime, &files, &cache_config)?; + report_process_memory("warm_open")?; if cache.startup_mode() != StartupMode::Warm { return Err(io::Error::other( "soak final verification did not recover a clean image", @@ -528,6 +608,7 @@ fn run_benchmark() -> io::Result<()> { max_managed_memory = max_managed_memory.max(verification_resources.managed_memory_peak_bytes); runtime.block_on(cache.close_fast())?; + report_process_memory("warm_verify")?; println!( "warm_verification keys={} hits={} stale_hits={} misses={} l2_hits={} l2_misses={} managed={} managed_peak={} errors=0", config.key_count, @@ -627,6 +708,7 @@ fn run_writer( ) -> io::Result<()> { let maximum_value_bytes = config.value_bytes.iter().copied().max().unwrap_or(0); let mut value = vec![0_u8; maximum_value_bytes]; + let mut written_bytes = 0_u64; while !stop.load(Ordering::Acquire) { pace(config.operation_interval); if stop.load(Ordering::Acquire) { @@ -655,6 +737,7 @@ fn run_writer( Ok(_) => { counters.put_latency.record_sampled(put_started); counters.writes.fetch_add(1, Ordering::Relaxed); + written_bytes = written_bytes.saturating_add(value_bytes as u64); } Err(error) if error.kind() == cache2::ErrorKind::Overloaded => { counters.put_latency.record_sampled(put_started); @@ -683,6 +766,9 @@ fn run_writer( } } } + counters + .written_bytes + .fetch_add(written_bytes, Ordering::Relaxed); Ok(()) } @@ -700,6 +786,7 @@ fn run_reader( runtime: &tokio::runtime::Handle, ) -> io::Result<()> { let reader_id = u64::try_from(reader_id).map_err(|_| invalid("reader id exceeds u64"))?; + let mut read_bytes = 0_u64; while !stop.load(Ordering::Acquire) { pace(config.operation_interval); if stop.load(Ordering::Acquire) { @@ -722,6 +809,7 @@ fn run_reader( &config.value_bytes, )?; counters.hits.fetch_add(1, Ordering::Relaxed); + read_bytes = read_bytes.saturating_add(observed.len() as u64); if stale { counters.stale_hits.fetch_add(1, Ordering::Relaxed); } @@ -732,6 +820,7 @@ fn run_reader( } } } + counters.read_bytes.fetch_add(read_bytes, Ordering::Relaxed); Ok(()) } @@ -938,7 +1027,7 @@ fn report_sample( if config.writers != 0 { JobReport::new("cache_soak", None, "put", "write", elapsed, counters.writes) .workers(config.writers) - .bytes(u128::from(resources.written_bytes)) + .bytes(u128::from(counters.written_bytes)) .errors(counters.write_rejections) .latency(&counters.put_latency, config.latency_sample_interval) .emit(); @@ -952,7 +1041,7 @@ fn report_sample( counters.hits.saturating_add(counters.misses), ) .workers(config.readers) - .bytes(u128::from(resources.served_bytes)) + .bytes(u128::from(counters.read_bytes)) .latency(&counters.get_latency, config.latency_sample_interval) .emit(); if config.writers != 0 { @@ -1048,3 +1137,12 @@ fn pace(interval: Duration) { std::thread::sleep(interval); } } + +fn report_process_memory(phase: &str) -> io::Result<()> { + println!( + "report version=2 type=process_memory benchmark=cache_soak phase={phase} current_rss_bytes={} peak_rss_bytes={}", + current_rss_bytes()?, + peak_rss_bytes(), + ); + Ok(()) +} diff --git a/benchmarks/src/harness.rs b/benchmarks/src/harness.rs index 66682b0..8933fac 100644 --- a/benchmarks/src/harness.rs +++ b/benchmarks/src/harness.rs @@ -220,8 +220,34 @@ pub fn current_rss_bytes() -> io::Result { .ok_or_else(|| io::Error::other("current RSS byte count overflow")) } -/// Returns the current resident set size in bytes, or 0 on platforms without `/proc`. -#[cfg(not(target_os = "linux"))] +/// Returns the current resident set size in bytes from macOS process accounting. +#[cfg(target_os = "macos")] +pub fn current_rss_bytes() -> io::Result { + let mut info = MaybeUninit::::uninit(); + let length = i32::try_from(size_of::()) + .map_err(|_| invalid("process memory info size exceeds i32"))?; + // SAFETY: query the current process into writable storage of the exact + // type and size required by PROC_PIDTASKINFO. No reference is exposed yet. + let written = unsafe { + libc::proc_pidinfo( + libc::getpid(), + libc::PROC_PIDTASKINFO, + 0, + info.as_mut_ptr().cast(), + length, + ) + }; + if written != length { + return Err(io::Error::other( + "cannot read current RSS from proc_pidinfo", + )); + } + // SAFETY: the complete proc_taskinfo was initialized by the successful call. + Ok(unsafe { info.assume_init() }.pti_resident_size) +} + +/// Returns 0 on platforms without current resident-memory accounting. +#[cfg(not(any(target_os = "linux", target_os = "macos")))] pub fn current_rss_bytes() -> io::Result { Ok(0) } diff --git a/benchmarks/src/report.rs b/benchmarks/src/report.rs index 3df5609..3133aab 100644 --- a/benchmarks/src/report.rs +++ b/benchmarks/src/report.rs @@ -498,7 +498,7 @@ pub fn emit_cache_report( fill.estimated_drain_ns, ); println!( - "report version=2 type=cache benchmark={} scenario={} phase={} health={:?} activity_counters_enabled={} puts={} deletes={} written_bytes={} served_bytes={} l1_hits={} l1_misses={} l2_hits={} l2_misses={} l2_read_memory_misses={} l2_read_busy_misses={} l2_read_overloads={} l2_read_wait_ns={} promotions={} l1_evictions={} l1_bypasses={} write_rejections={} io_failures={} rotations={} reclaimed_regions={} reclaim_bytes={} reclaim_records={} reinsert_records={} reinsert_bytes={} reinsert_skipped={} reinsert_budget_skipped={}", + "report version=2 type=cache benchmark={} scenario={} phase={} health={:?} activity_counters_enabled={} puts={} deletes={} written_bytes={} served_bytes={} l1_hits={} l1_misses={} l2_hits={} l2_misses={} l2_read_memory_misses={} l2_read_busy_misses={} l2_read_overloads={} l2_read_wait_ns={} promotions={} l1_evictions={} l1_bypasses={} write_rejections={} write_buffer_rejections={} io_failures={} rotations={} reclaimed_regions={} reclaim_bytes={} reclaim_records={} reinsert_records={} reinsert_bytes={} reinsert_skipped={} reinsert_budget_skipped={}", benchmark, scenario, phase, @@ -520,6 +520,7 @@ pub fn emit_cache_report( cache.l1_evictions, cache.l1_bypasses, cache.write_rejections, + detailed.write_buffer_rejections, cache.io_failures, cache.region_rotations, cache.reclaim.regions, diff --git a/scripts/compare-benchmarks.py b/scripts/compare-benchmarks.py new file mode 100644 index 0000000..e7ea8bf --- /dev/null +++ b/scripts/compare-benchmarks.py @@ -0,0 +1,231 @@ +#!/usr/bin/env python3 +# Copyright 2026 ScopeDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Serial ABBA comparisons of saved cache/cache_soak benchmark binaries.""" + +import argparse +import hashlib +import json +import math +import os +from pathlib import Path +import platform +import random +import re +import statistics +import subprocess + + +def parse_measurement(output, metrics, minimum_seconds): + rows = [ + dict(part.split("=", 1) for part in line.split() if "=" in part) + for line in output.splitlines() + if line.startswith(("result ", "report version=2 ", "complete ")) + ] + run = next(row for row in rows if row.get("type") == "run") + if run["status"] != "pass": + raise ValueError("benchmark status was not pass") + rates, durations, details = {}, {}, {"run": run} + for row in rows: + if "ops_per_sec" in row: + rates[row["phase"]] = float(row["ops_per_sec"]) + durations[row["phase"]] = int(row["elapsed_ns"]) / 1e9 + if row.get("type") == "job" and row.get("benchmark") == "cache_soak": + rates[row["job"]] = float(row["iops"]) + durations[row["job"]] = int(row["runtime_ns"]) / 1e9 + details[row["job"]] = row + if row.get("phase") in ("put_admission", "put_steady_admission"): + details[row["phase"]] = row + if "hits" in row and "writes" in row: + rates["hit"] = int(row["hits"]) / durations["get"] + durations["hit"] = durations["get"] + details["complete"] = row + for metric in metrics: + if not math.isfinite(rates.get(metric, 0)) or rates.get(metric, 0) <= 0: + raise ValueError(f"missing or nonpositive metric: {metric}") + if durations[metric] < minimum_seconds: + raise ValueError(f"{metric} too short: {durations[metric]:.3f}s") + return {"rates": rates, "seconds": durations, "details": details} + + +def summarize(samples, metrics, block_count): + result = {} + for metric in metrics: + changes, centers, repeat_spreads = [], [], [] + for block in range(block_count): + groups = { + version: [ + math.log(sample["rates"][metric]) + for sample in samples + if sample["block"] == block and sample["version"] == version + ] + for version in "AB" + } + if any(len(group) != 2 for group in groups.values()): + raise ValueError("each block requires two samples per binary") + # Pairing can cancel large alternating run-to-run noise exactly. + # Check repeated runs of the same binary before averaging them. + repeat_spreads.extend(max(group) - min(group) for group in groups.values()) + means = {version: statistics.mean(group) for version, group in groups.items()} + changes.append(means["B"] - means["A"]) + centers.append(statistics.mean(means.values())) + # The four serial runs form one paired unit. Resampling individual runs + # would pretend the within-block order and common drift were independent. + rng = random.Random(20261027) + bootstrap = sorted( + statistics.mean(rng.choices(changes, k=block_count)) for _ in range(10_000) + ) + result[metric] = { + "median_A": statistics.median( + sample["rates"][metric] for sample in samples if sample["version"] == "A" + ), + "median_B": statistics.median( + sample["rates"][metric] for sample in samples if sample["version"] == "B" + ), + "paired_change_percent": 100 * math.expm1(statistics.mean(changes)), + "paired_bootstrap_95_percent": [ + 100 * math.expm1(bootstrap[249]), + 100 * math.expm1(bootstrap[9749]), + ], + "block_change_percent": [100 * math.expm1(change) for change in changes], + "block_center_spread_percent": 100 * math.expm1(max(centers) - min(centers)), + "max_within_block_spread_percent": 100 * math.expm1(max(repeat_spreads)), + } + return result + + +def stable_power(output, minimum_battery_percent): + # Desktop Macs have no battery line. A connected but underpowered charger + # can still discharge a laptop; AC Power by itself is insufficient. + battery = re.search(r"(\d+)%;\s*([^;]+);", output) + return "'AC Power'" in output and ( + battery is None + or (int(battery[1]) >= minimum_battery_percent and battery[2] in ("charging", "charged", "AC attached", "finishing charge")) + ) + + +def record_power(output_directory, label, minimum_battery_percent): + if platform.system() != "Darwin": + return + output = subprocess.check_output(["/usr/bin/pmset", "-g", "batt"], text=True) + (output_directory / f"{label}-power.txt").write_text(output) + if not stable_power(output, minimum_battery_percent): + raise RuntimeError("unstable power or low battery; retained logs, stopped comparison") + + +def write_json(path, value): + path.write_text(json.dumps(value, indent=2) + "\n") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--baseline", type=Path, required=True) + parser.add_argument("--candidate", type=Path, required=True) + parser.add_argument("--env", type=Path, required=True, help="JSON object of workload environment variables") + parser.add_argument("--output", type=Path, required=True, help="new directory for all raw logs and results") + parser.add_argument("--blocks", type=int, default=8) + parser.add_argument("--metric", action="append", required=True, help="put_steady, or soak get/hit/put; repeatable") + parser.add_argument("--prefix", action="append", default=[], help="launcher argument before the binary; repeat for taskset/taskpolicy arguments") + parser.add_argument("--aa", action="store_true", help="require identical binaries and a narrow interval around zero") + parser.add_argument("--tolerance", type=float, default=3.0, help="A/A confidence interval bound, percent") + parser.add_argument("--max-drift", type=float, default=5.0, help="maximum spread of block centers or same-binary repeats within a block, percent") + parser.add_argument("--min-seconds", type=float, default=5.0, help="minimum measured duration for each selected metric") + parser.add_argument("--min-battery-percent", type=int, default=20, help="macOS battery floor; AC without discharging is always required") + args = parser.parse_args() + if args.blocks < 4: + parser.error("at least four blocks are required") + if any(not math.isfinite(value) or value <= 0 for value in (args.tolerance, args.max_drift, args.min_seconds)): + parser.error("tolerances and duration must be finite and positive") + if not 0 <= args.min_battery_percent <= 100: + parser.error("battery floor must be in 0..100") + paths = {"A": args.baseline.resolve(), "B": args.candidate.resolve()} + hashes = {version: hashlib.sha256(path.read_bytes()).hexdigest() for version, path in paths.items()} + if args.aa and hashes["A"] != hashes["B"]: + parser.error("A/A requires byte-identical binaries") + settings = json.loads(args.env.read_text()) + if not isinstance(settings, dict) or any( + not isinstance(key, str) or not isinstance(value, str) for key, value in settings.items() + ): + parser.error("environment must be an object of string keys and string values") + environment = { + key: value for key, value in os.environ.items() + if not key.startswith(("CACHE_BENCH_", "CACHE_WORKLOAD_", "CACHE_SOAK_")) + } + environment.update(settings) + args.output.mkdir(parents=True, exist_ok=False) + write_json(args.output / "manifest.json", { + "binaries": {version: str(path) for version, path in paths.items()}, + "sha256": hashes, + "runner_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + "environment": settings, + "platform": platform.platform(), + "command_prefix": args.prefix, + "blocks": args.blocks, + "metrics": args.metric, + "aa": args.aa, + "tolerance_percent": args.tolerance, + "max_drift_percent": args.max_drift, + "min_seconds": args.min_seconds, + "min_battery_percent": args.min_battery_percent, + }) + samples = [] + try: + for block in range(args.blocks): + order = "ABBA" if block % 2 == 0 else "BAAB" + for slot, version in enumerate(order): + label = f"{block + 1:02}-{slot + 1}-{version}" + record_power(args.output, label + "-before", args.min_battery_percent) + path = args.output / f"{label}.log" + with path.open("w") as log: + process = subprocess.run( + [*args.prefix, str(paths[version])], env=environment, + stdout=log, stderr=subprocess.STDOUT, timeout=300, + ) + if process.returncode: + raise RuntimeError(f"{path}: exit {process.returncode}") + sample = { + "block": block, "slot": slot, "version": version, + **parse_measurement(path.read_text(), args.metric, args.min_seconds), + } + samples.append(sample) + write_json(args.output / "samples.json", samples) + record_power(args.output, label + "-after", args.min_battery_percent) + print(json.dumps({ + "block": block + 1, "slot": slot + 1, "version": version, + "rates": {metric: sample["rates"][metric] for metric in args.metric}, + "seconds": {metric: sample["seconds"][metric] for metric in args.metric}, + }), flush=True) + except (OSError, RuntimeError, ValueError, StopIteration, subprocess.SubprocessError) as error: + write_json(args.output / "failure.json", {"error": str(error), "completed_samples": len(samples)}) + raise + metrics = summarize(samples, args.metric, args.blocks) + stable = all( + max(value["block_center_spread_percent"], value["max_within_block_spread_percent"]) <= args.max_drift + for value in metrics.values() + ) + aa_pass = stable and all( + -args.tolerance <= value["paired_bootstrap_95_percent"][0] + and value["paired_bootstrap_95_percent"][1] <= args.tolerance + for value in metrics.values() + ) + result = {"metrics": metrics, "stable": stable, "aa_pass": aa_pass if args.aa else None} + write_json(args.output / "summary.json", result) + print(json.dumps(result), flush=True) + if not stable or (args.aa and not aa_pass): + raise SystemExit(2) + + +if __name__ == "__main__": + main() diff --git a/scripts/test_compare_benchmarks.py b/scripts/test_compare_benchmarks.py new file mode 100644 index 0000000..3d2404c --- /dev/null +++ b/scripts/test_compare_benchmarks.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +# Copyright 2026 ScopeDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import contextlib +import importlib.util +import io +import json +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +spec = importlib.util.spec_from_file_location( + "compare_benchmarks", Path(__file__).with_name("compare-benchmarks.py") +) +compare = importlib.util.module_from_spec(spec) +spec.loader.exec_module(compare) + + +class ComparisonTests(unittest.TestCase): + def run_comparison_fixture(self, rates, *, aa=False, expected_exit=0): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + binary = root / "benchmark" + binary.write_bytes(b"fixture benchmark") + settings = root / "environment.json" + settings.write_text(json.dumps({"CACHE_SOAK_KEYS": "123"})) + output = root / "results" + arguments = [ + "compare-benchmarks.py", "--baseline", str(binary), + "--candidate", str(binary), "--env", str(settings), + "--output", str(output), "--blocks", "4", "--metric", "put_steady", + ] + if aa: + arguments.append("--aa") + remaining = iter(rates) + + def run_fixture(command, *, env, stdout, **kwargs): + self.assertEqual(command, [str(binary)]) + self.assertEqual(env["CACHE_SOAK_KEYS"], "123") + self.assertNotIn("CACHE_BENCH_TEST_INHERITED", env) + stdout.write( + f"result phase=put_steady ops_per_sec={next(remaining)} elapsed_ns=5000000000\n" + "report version=2 type=run status=pass\n" + ) + return compare.subprocess.CompletedProcess(command, 0) + + with ( + mock.patch("sys.argv", arguments), + mock.patch.dict(compare.os.environ, {"CACHE_BENCH_TEST_INHERITED": "1"}), + mock.patch.object(compare, "record_power"), + mock.patch.object(compare.platform, "platform", return_value="test host"), + mock.patch.object(compare.subprocess, "run", side_effect=run_fixture), + contextlib.redirect_stdout(io.StringIO()), + ): + if expected_exit: + with self.assertRaises(SystemExit) as error: + compare.main() + self.assertEqual(error.exception.code, expected_exit) + else: + compare.main() + self.assertEqual(len(list(output.glob("*.log"))), 16) + samples = json.loads((output / "samples.json").read_text()) + self.assertEqual("".join(sample["version"] for sample in samples), "ABBABAAB" * 2) + manifest = json.loads((output / "manifest.json").read_text()) + self.assertEqual(manifest["sha256"]["A"], manifest["sha256"]["B"]) + return json.loads((output / "summary.json").read_text()) + + def test_stable_aa_retains_all_samples_and_passes(self): + summary = self.run_comparison_fixture([1000] * 16, aa=True) + self.assertTrue(summary["stable"]) + self.assertTrue(summary["aa_pass"]) + + def test_pairing_does_not_hide_alternating_run_noise(self): + for aa in (False, True): + with self.subTest(aa=aa): + summary = self.run_comparison_fixture([1000, 500] * 8, aa=aa, expected_exit=2) + metric = summary["metrics"]["put_steady"] + self.assertEqual(metric["paired_bootstrap_95_percent"], [0, 0]) + self.assertEqual(metric["block_center_spread_percent"], 0) + self.assertAlmostEqual(metric["max_within_block_spread_percent"], 100) + self.assertFalse(summary["stable"]) + self.assertEqual(summary["aa_pass"], False if aa else None) + + def test_block_drift_fails_comparison_even_when_repeats_are_stable(self): + summary = self.run_comparison_fixture( + [1000 * 1.1 ** block for block in range(4) for _ in range(4)], expected_exit=2, + ) + metric = summary["metrics"]["put_steady"] + self.assertEqual(metric["max_within_block_spread_percent"], 0) + self.assertGreater(metric["block_center_spread_percent"], 33) + self.assertFalse(summary["stable"]) + + def test_parse_counts_hits_independently_of_library_statistics(self): + output = """report version=2 type=job benchmark=cache_soak job=get iops=100 runtime_ns=20000000000 +complete hits=600 writes=1000 cache_puts=0 l1_hits=0 l2_hits=0 +report version=2 type=run status=pass +""" + result = compare.parse_measurement(output, ["get", "hit"], 5) + self.assertEqual(result["rates"]["hit"], 30) + self.assertEqual(result["rates"]["get"], 100) + + def test_rejects_failed_short_and_missing_measurements(self): + output = """result phase=put_steady ops_per_sec=1000 elapsed_ns=2000000000 +report version=2 type=run status=pass +""" + for text, metrics, minimum in [ + (output.replace("status=pass", "status=error"), ["put_steady"], 1), + (output, ["put_steady"], 5), + (output, ["get"], 1), + ]: + with self.assertRaises(ValueError): + compare.parse_measurement(text, metrics, minimum) + + def test_abba_cancels_linear_log_drift_without_hiding_absolute_drift(self): + samples = [] + for block in range(4): + for slot, version in enumerate("ABBA" if block % 2 == 0 else "BAAB"): + samples.append({ + "block": block, "version": version, + "rates": {"put": 1000 * 1.01 ** (4 * block + slot) * (1.1 if version == "B" else 1)}, + }) + result = compare.summarize(samples, ["put"], 4)["put"] + self.assertAlmostEqual(result["paired_change_percent"], 10) + for bound in result["paired_bootstrap_95_percent"]: + self.assertAlmostEqual(bound, 10) + self.assertGreater(result["block_center_spread_percent"], 12) + + def test_noisy_pairs_keep_their_uncertainty(self): + samples = [] + for block, ratio in enumerate([0.8, 1.2, 0.8, 1.2]): + for version in "ABBA": + samples.append({ + "block": block, "version": version, + "rates": {"put": 1000 * (ratio if version == "B" else 1)}, + }) + interval = compare.summarize(samples, ["put"], 4)["put"]["paired_bootstrap_95_percent"] + self.assertLess(interval[0], -3) + self.assertGreater(interval[1], 3) + + def test_rejects_incomplete_blocks(self): + with self.assertRaises(ValueError): + compare.summarize([], ["put"], 4) + + def test_ac_power_alone_does_not_establish_stable_power(self): + for state in ["3%; discharging;", "85%; discharging;", "3%; charging;"]: + self.assertFalse(compare.stable_power("Now drawing from 'AC Power'\n" + state, 20)) + self.assertFalse(compare.stable_power("Now drawing from 'Battery Power'\n80%; discharging;", 20)) + for state in ["80%; charging;", "100%; charged;", "80%; AC attached; not charging", ""]: + self.assertTrue(compare.stable_power("Now drawing from 'AC Power'\n" + state, 20)) + + +if __name__ == "__main__": + unittest.main() diff --git a/xtask/src/main.rs b/xtask/src/main.rs index a19fcba..3102d92 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -60,7 +60,7 @@ enum SubCommand { about = "Run nightly Rust lints, formatting, spelling, documentation, and policy checks" )] Lint(CommandLint), - #[command(about = "Run workspace tests and extended library tests")] + #[command(about = "Run workspace, extended library, and benchmark-tool tests")] Test(CommandTest), } @@ -229,6 +229,7 @@ impl CommandTest { "--", "--ignored", ]); + command_run("python3", ["scripts/test_compare_benchmarks.py"]); } }