tidaldb/tidal/benches/wal.rs

#![allow(clippy::unwrap_used, clippy::cast_precision_loss)]

//! Criterion benchmarks for the WAL append / group-commit / fsync hot path.
//!
//! Before this bench existed the entire WAL write path was optimized blind: the
//! signals bench wires a `NoopWalWriter`, and `recovery.rs` only exercises the
//! read/replay side. This is the instrument the perf sweep (2026-06-13, finding
//! rank 8) called for — it makes every later WAL change (dedup double-hash,
//! encode double-copy, per-flush allocations) provable and guards
//! committed-events/s against silent regression.
//!
//! ## Benchmarks
//!
//! - **`wal_append_throughput/writers8_batch{1,10,100}`** — 8 concurrent stagers
//!   each submit 250 distinct `EventRecord`s via `append_record_staged` and then
//!   block on every `PendingAppend`. This drives the *real* group-commit funnel
//!   against a real on-disk WAL (real `fdatasync`/`F_FULLFSYNC`), so the
//!   throughput number reflects how well the writer thread coalesces concurrent
//!   stagers into shared fsyncs. Sweeping `batch_size` shows the coalescing curve.
//!
//! - **`wal_encode_batch/events{1,256}`** — the pure-CPU encode cost
//!   (`encode_batch`: BLAKE3 checksum + 32-byte v3 packing) with no I/O, at a
//!   single event and at a full `MAX_EVENTS_PER_BATCH` batch. Isolates the
//!   serialization kernel from the fsync so a CPU regression there is visible.
//!
//! Run:
//! ```bash
//! cargo bench -p tidaldb --bench wal
//! ```

use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use criterion::{Criterion, Throughput, black_box, criterion_group, criterion_main};
use tidaldb::wal::{
    WalConfig, WalHandle,
    format::{EventRecord, MAX_EVENTS_PER_BATCH, encode_batch},
};

/// Monotonic entity-id source so no two records across the whole bench (or
/// across criterion iterations) ever collide in the dedup window — every staged
/// event must reach the writer and be counted, otherwise throughput is a lie.
static NEXT_ID: AtomicU64 = AtomicU64::new(1);

fn open_wal(dir: &std::path::Path, batch_size: usize) -> WalHandle {
    let config = WalConfig {
        dir: dir.to_path_buf(),
        batch_size,
        // Short timeout: a solo stager still commits promptly, but concurrent
        // stagers fill the batch before the timer fires (the path we measure).
        batch_timeout: Duration::from_millis(5),
        ..WalConfig::default()
    };
    let (handle, _replayed, _blobs, _sessions) = WalHandle::open(config).expect("open wal");
    handle
}

fn append_throughput(c: &mut Criterion) {
    const WRITERS: u64 = 8;
    const PER_WRITER: u64 = 250;
    let total = WRITERS * PER_WRITER;

    let mut group = c.benchmark_group("wal_append_throughput");
    // Each iteration fsyncs thousands of events to a real disk — keep sample
    // count modest and give the wall-clock room.
    group.sample_size(10);
    group.measurement_time(Duration::from_secs(20));
    group.throughput(Throughput::Elements(total));

    for batch_size in [1usize, 10, 100] {
        // One WAL reused across iterations (segments rotate as in production).
        let dir = tempfile::tempdir().expect("tempdir");
        let handle = open_wal(dir.path(), batch_size);
        let sender = handle.sender();

        group.bench_function(format!("writers8_batch{batch_size}"), |b| {
            b.iter_custom(|iters| {
                let mut elapsed = Duration::ZERO;
                for _ in 0..iters {
                    // Reserve a unique, non-overlapping id range for this iter.
                    let base = NEXT_ID.fetch_add(total, Ordering::Relaxed);
                    let start = Instant::now();
                    let threads: Vec<_> = (0..WRITERS)
                        .map(|w| {
                            let s = sender.clone();
                            std::thread::spawn(move || {
                                let mut pending = Vec::with_capacity(PER_WRITER as usize);
                                for i in 0..PER_WRITER {
                                    let id = base + w * PER_WRITER + i;
                                    let rec = EventRecord::signal(id, 0, 1.0, id.max(1) * 1_000);
                                    pending.push(s.append_record_staged(rec).unwrap());
                                }
                                // Block on durability for every staged append.
                                for p in pending {
                                    p.wait().unwrap();
                                }
                            })
                        })
                        .collect();
                    for t in threads {
                        t.join().unwrap();
                    }
                    elapsed += start.elapsed();
                }
                elapsed
            });
        });

        handle.shutdown().expect("shutdown wal");
    }
    group.finish();
}

fn encode_cpu(c: &mut Criterion) {
    let mut group = c.benchmark_group("wal_encode_batch");
    for n in [1usize, usize::from(MAX_EVENTS_PER_BATCH)] {
        let events: Vec<EventRecord> = (0..n as u64)
            .map(|i| EventRecord::signal(i + 1, 0, 1.0, (i + 1) * 1_000))
            .collect();
        group.throughput(Throughput::Elements(n as u64));
        group.bench_function(format!("events{n}"), |b| {
            b.iter(|| {
                let bytes = encode_batch(black_box(&events), 1, 1_000).unwrap();
                black_box(bytes);
            });
        });
    }
    group.finish();
}

criterion_group!(benches, append_throughput, encode_cpu);
criterion_main!(benches);