// Copyright 2026 Jonathan Shook
// SPDX-License-Identifier: Apache-2.0

//! Throughput benchmarks for slabtastic write and read paths.
//!
//! ## Benchmark groups
//!
//! **Single-threaded:**
//! - `write` — end-to-end write throughput (file create → add_record × N → finish)
//! - `sequential_read` — full-file iteration via `SlabReader::iter`
//! - `random_read` — 10 000 point lookups via `SlabReader::get` with
//!   pseudo-random ordinals spread across the file
//!
//! **Concurrent (multi-threaded):**
//! - `concurrent_random_read` — N threads (2/4/8) each open an independent
//!   `SlabReader` on the same 10 M-record file and perform 10 000 random
//!   `get()` calls with different access patterns. Measures OS page-cache
//!   and file-I/O scaling under read contention.
//! - `concurrent_sequential_read` — N threads (2/4/8) each open their own
//!   `SlabReader` on the same 1 M-record file and call `iter()` to read
//!   every record. Measures sequential bandwidth scaling.
//! - `concurrent_write` — N threads (2/4/8) each write an independent
//!   1 M-record slab file in parallel. Measures aggregate write throughput
//!   when multiple pipelines run simultaneously.
//!
//! Single-threaded and concurrent benchmarks use `iter_batched` to open
//! files in the setup phase (outside the timed region) so that only the
//! actual read/write operations are measured. Async benchmarks
//! (`read_to_sink_async`, `write_from_iter_async`) include open/close
//! since the async API takes a path and manages the file internally.
//!
//! ## Reading benchmark output
//!
//! Benchmark IDs encode the workload shape so that the Criterion output
//! is self-documenting. For example:
//!
//! ```text
//! concurrent_random_read/2t_x_10k_in_10M  thrpt: [1.23 Melem/s ...]
//! ```
//!
//! means 2 threads × 10 000 gets each = 20 000 total gets against a
//! 10 M-record file, and the throughput is measured in total gets/s.
//!
//! All benchmarks use deterministic 5–25 byte records generated by
//! [`make_record`]. For 100 M+ scale tests see `tests/large_scale.rs`.

use std::path::PathBuf;
use std::thread;

use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use slabtastic::{SlabReader, SlabWriter, WriterConfig};
use tempfile::NamedTempFile;

/// Format a count as a compact human-readable string (e.g. 1_000_000 → "1M").
fn human(n: u64) -> String {
    if n >= 1_000_000_000 && n.is_multiple_of(1_000_000_000) {
        format!("{}B", n / 1_000_000_000)
    } else if n >= 1_000_000 && n.is_multiple_of(1_000_000) {
        format!("{}M", n / 1_000_000)
    } else if n >= 1_000 && n.is_multiple_of(1_000) {
        format!("{}k", n / 1_000)
    } else {
        n.to_string()
    }
}

/// Generate a deterministic record of 5–25 bytes based on ordinal.
///
/// The length cycles through 5..=25 (mod 21) so that page packing is
/// exercised with non-uniform record sizes, matching realistic short-text
/// workloads.
fn make_record(ordinal: u64) -> Vec<u8> {
    let len = 5 + (ordinal % 21) as usize;
    let base = format!("{ordinal:0>width$}", width = len);
    base.into_bytes()[..len].to_vec()
}

/// Write a test file with `count` records using default config and return
/// the `NamedTempFile` handle (must be kept alive to prevent deletion)
/// together with the file path.
fn prepare_file(count: u64) -> (NamedTempFile, PathBuf) {
    let tmp = NamedTempFile::new().unwrap();
    let path = tmp.path().to_path_buf();
    let config = WriterConfig::default();
    let mut writer = SlabWriter::new(&path, config).unwrap();
    for i in 0..count {
        writer.add_record(&make_record(i)).unwrap();
    }
    writer.finish().unwrap();
    (tmp, path)
}

// ---------------------------------------------------------------------------
// Single-threaded benchmarks
// ---------------------------------------------------------------------------

/// Measure single-threaded write throughput.
///
/// The file is created and the writer is opened in the setup phase
/// (outside the timed region) so that only the `add_record` loop is
/// measured. `finish()` (which writes the pages page and flushes) is
/// likewise excluded by not calling it inside the timed closure — the
/// writer is simply dropped after the record loop.
fn bench_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("write");
    group.sample_size(10);

    for count in [1_000_000u64, 10_000_000] {
        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new("records", human(count)),
            &count,
            |b, &count| {
                b.iter_batched(
                    || {
                        let tmp = NamedTempFile::new().unwrap();
                        let path = tmp.path().to_path_buf();
                        let writer = SlabWriter::new(&path, WriterConfig::default()).unwrap();
                        (tmp, writer)
                    },
                    |(_tmp, mut writer)| {
                        for i in 0..count {
                            writer.add_record(&make_record(i)).unwrap();
                        }
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }
    group.finish();
}

/// Measure single-threaded sequential read throughput via `iter()`.
///
/// The file is opened in the setup phase (outside the timed region) so
/// that only the page-by-page deserialization and record extraction are
/// measured. Uses `iter_batched` with `LargeInput` batch size since each
/// setup allocates a `SlabReader` with its file descriptor and in-memory
/// pages-page index.
fn bench_sequential_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("sequential_read");
    group.sample_size(10);

    for count in [1_000_000u64, 10_000_000] {
        let (_tmp, path) = prepare_file(count);

        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new("records", human(count)),
            &count,
            |b, &_count| {
                b.iter_batched(
                    || SlabReader::open(&path).unwrap(),
                    |reader| {
                        let records = reader.iter().unwrap();
                        assert!(!records.is_empty());
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

/// Measure single-threaded random-access read throughput via `get()`.
///
/// 10 000 pseudo-random ordinals (deterministic, using prime-modular
/// scatter) are looked up against a pre-built file. The file is opened
/// in the setup phase so that only the binary-search index lookup plus
/// single-page deserialization cost is measured per `get()` call.
fn bench_random_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("random_read");
    group.sample_size(10);

    for count in [1_000_000u64, 10_000_000] {
        let (_tmp, path) = prepare_file(count);

        let sample_count = 10_000u64;
        let ordinals: Vec<i64> = (0..sample_count)
            .map(|i| ((i * 7919) % count) as i64)
            .collect();

        group.throughput(Throughput::Elements(sample_count));
        group.bench_with_input(
            BenchmarkId::new(
                format!("{}_gets_in", human(sample_count)),
                human(count),
            ),
            &ordinals,
            |b, ordinals| {
                b.iter_batched(
                    || SlabReader::open(&path).unwrap(),
                    |reader| {
                        for &ord in ordinals {
                            reader.get_ref(ord).unwrap();
                        }
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

// ---------------------------------------------------------------------------
// Concurrent benchmarks
// ---------------------------------------------------------------------------

/// Measure concurrent random-access read scaling (1/2/4/8/16 threads).
///
/// All threads share the same pre-built 10 M-record file, each with its
/// own `SlabReader` and file descriptor. Readers are pre-opened in the
/// setup phase so that only the `get()` calls are timed.
///
/// Each thread performs 100 000 random `get()` calls using a
/// deterministic prime-modular scatter (each thread gets a distinct seed
/// so access patterns don't overlap). The 1-thread case establishes a
/// single-threaded baseline; comparing it against the multi-thread cases
/// shows scaling behavior.
///
/// Throughput is reported as aggregate gets/s across all threads.
///
/// **Note:** each `get()` reads the containing page and extracts the
/// single record via zero-copy offset lookup (no full page
/// deserialization). Per-get cost is dominated by the seek + read
/// syscalls rather than the index lookup.
fn bench_concurrent_random_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("concurrent_random_read");
    group.sample_size(10);

    let record_count = 10_000_000u64;
    let (_tmp, path) = prepare_file(record_count);
    let reads_per_thread = 100_000u64;

    for n_threads in [1, 2, 4, 8, 16] {
        let total_reads = reads_per_thread * n_threads as u64;
        group.throughput(Throughput::Elements(total_reads));
        group.bench_with_input(
            BenchmarkId::new(
                format!(
                    "{}t_x_{}_gets_in_{}",
                    n_threads,
                    human(reads_per_thread),
                    human(record_count),
                ),
                human(total_reads),
            ),
            &n_threads,
            |b, &n_threads| {
                b.iter_batched(
                    || {
                        (0..n_threads)
                            .map(|_| SlabReader::open(&path).unwrap())
                            .collect::<Vec<_>>()
                    },
                    |readers| {
                        if readers.len() == 1 {
                            // Single-thread: avoid thread-spawn overhead entirely
                            let reader = readers.into_iter().next().unwrap();
                            for i in 0..reads_per_thread {
                                let ordinal =
                                    (i.wrapping_mul(7919) % record_count) as i64;
                                reader.get_ref(ordinal).unwrap();
                            }
                        } else {
                            let handles: Vec<_> = readers
                                .into_iter()
                                .enumerate()
                                .map(|(t, reader)| {
                                    thread::spawn(move || {
                                        let seed = (t as u64).wrapping_mul(6271);
                                        for i in 0..reads_per_thread {
                                            let ordinal =
                                                ((i.wrapping_mul(7919).wrapping_add(seed))
                                                    % record_count)
                                                    as i64;
                                            reader.get_ref(ordinal).unwrap();
                                        }
                                    })
                                })
                                .collect();
                            for h in handles {
                                h.join().unwrap();
                            }
                        }
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

/// Measure concurrent sequential read scaling (2/4/8 threads).
///
/// All threads share the same pre-built 1 M-record file and each
/// performs a full `iter()`. Readers are pre-opened in the setup phase
/// so that only the page-by-page iteration is timed. Throughput is
/// reported as the aggregate record count across all threads per second.
fn bench_concurrent_sequential_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("concurrent_sequential_read");
    group.sample_size(10);

    let record_count = 1_000_000u64;
    let (_tmp, path) = prepare_file(record_count);

    for n_threads in [2, 4, 8] {
        let total = record_count * n_threads as u64;
        group.throughput(Throughput::Elements(total));
        group.bench_with_input(
            BenchmarkId::new(
                format!(
                    "{}t_x_{}_records",
                    n_threads,
                    human(record_count),
                ),
                human(total),
            ),
            &n_threads,
            |b, &n_threads| {
                b.iter_batched(
                    || {
                        (0..n_threads)
                            .map(|_| SlabReader::open(&path).unwrap())
                            .collect::<Vec<_>>()
                    },
                    |readers| {
                        let handles: Vec<_> = readers
                            .into_iter()
                            .map(|reader| {
                                thread::spawn(move || {
                                    let records = reader.iter().unwrap();
                                    assert!(!records.is_empty());
                                })
                            })
                            .collect();
                        for h in handles {
                            h.join().unwrap();
                        }
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

/// Measure concurrent independent-file write scaling (2/4/8 threads).
///
/// Each thread writes to its own pre-created temp file with a
/// pre-opened writer. Writers are created in the setup phase so that
/// only the `add_record` loops are timed. `finish()` is excluded — the
/// writers are dropped after the record loop. Throughput is reported as
/// the aggregate record count across all threads per second.
fn bench_concurrent_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("concurrent_write");
    group.sample_size(10);

    let records_per_thread = 1_000_000u64;

    for n_threads in [2, 4, 8] {
        let total = records_per_thread * n_threads as u64;
        group.throughput(Throughput::Elements(total));
        group.bench_with_input(
            BenchmarkId::new(
                format!(
                    "{}t_x_{}_records",
                    n_threads,
                    human(records_per_thread),
                ),
                human(total),
            ),
            &n_threads,
            |b, &n_threads| {
                b.iter_batched(
                    || {
                        (0..n_threads)
                            .map(|_| {
                                let tmp = NamedTempFile::new().unwrap();
                                let path = tmp.path().to_path_buf();
                                let writer =
                                    SlabWriter::new(&path, WriterConfig::default()).unwrap();
                                (tmp, writer)
                            })
                            .collect::<Vec<_>>()
                    },
                    |writers| {
                        let handles: Vec<_> = writers
                            .into_iter()
                            .enumerate()
                            .map(|(t, (_tmp, mut writer))| {
                                thread::spawn(move || {
                                    let base = t as u64 * records_per_thread;
                                    for i in 0..records_per_thread {
                                        writer.add_record(&make_record(base + i)).unwrap();
                                    }
                                    // Keep _tmp alive until writes are done
                                    drop(_tmp);
                                })
                            })
                            .collect();
                        for h in handles {
                            h.join().unwrap();
                        }
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }
    group.finish();
}

// ---------------------------------------------------------------------------
// Streaming / callback-mode benchmarks
// ---------------------------------------------------------------------------

/// Measure batched-iteration throughput via `SlabReader::batch_iter`.
///
/// The file is pre-built with 1 M records. The reader is opened and
/// converted to a `SlabBatchIter` in the setup phase (outside the timed
/// region) so that only the batch-draining loop is measured.
fn bench_batch_iter(c: &mut Criterion) {
    let mut group = c.benchmark_group("batch_iter");
    group.sample_size(10);

    let count = 1_000_000u64;
    let (_tmp, path) = prepare_file(count);

    for batch_size in [64, 1024, 16384] {
        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new(
                format!("{}_records_batch{}", human(count), batch_size),
                batch_size,
            ),
            &batch_size,
            |b, &batch_size| {
                b.iter_batched(
                    || {
                        let reader = SlabReader::open(&path).unwrap();
                        reader.batch_iter(batch_size)
                    },
                    |mut iter| {
                        let mut total = 0u64;
                        loop {
                            let batch = iter.next_batch().unwrap();
                            if batch.is_empty() {
                                break;
                            }
                            total += batch.len() as u64;
                        }
                        assert_eq!(total, count);
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

/// Measure synchronous sink-read throughput via `read_all_to_sink`.
///
/// The reader is opened in the setup phase (outside the timed region)
/// so that only the page-by-page record streaming is measured.
/// Throughput is records per second.
fn bench_read_all_to_sink(c: &mut Criterion) {
    let mut group = c.benchmark_group("read_all_to_sink");
    group.sample_size(10);

    for count in [1_000_000u64, 10_000_000] {
        let (_tmp, path) = prepare_file(count);

        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new("records", human(count)),
            &count,
            |b, &_count| {
                b.iter_batched(
                    || SlabReader::open(&path).unwrap(),
                    |reader| {
                        let mut sink = Vec::new();
                        let n = reader.read_all_to_sink(&mut sink).unwrap();
                        assert!(n > 0);
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

/// Measure async sink-read throughput via `read_to_sink_async`.
///
/// Spawns a background thread to read all records from a pre-built file
/// into a `Vec<u8>` sink, then waits for completion. This measures the
/// full callback-mode lifecycle: thread spawn, file open on the
/// background thread, sequential iteration with progress updates, and
/// join.
fn bench_read_to_sink_async(c: &mut Criterion) {
    let mut group = c.benchmark_group("read_to_sink_async");
    group.sample_size(10);

    for count in [1_000_000u64, 10_000_000] {
        let (_tmp, path) = prepare_file(count);

        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new("records", human(count)),
            &count,
            |b, &_count| {
                b.iter(|| {
                    let sink: Vec<u8> = Vec::new();
                    let task = SlabReader::read_to_sink_async(
                        path.clone(),
                        sink,
                        |_| {},
                    );
                    let n = task.wait().unwrap();
                    assert!(n > 0);
                });
            },
        );
    }
    group.finish();
}

/// Measure async iterator-write throughput via `write_from_iter_async`.
///
/// Spawns a background thread that consumes an iterator of `count`
/// records and writes them to a fresh slab file. Measures the full
/// callback-mode write lifecycle including thread spawn, file creation,
/// record iteration with progress updates, finish, and join.
fn bench_write_from_iter_async(c: &mut Criterion) {
    let mut group = c.benchmark_group("write_from_iter_async");
    group.sample_size(10);

    for count in [1_000_000u64, 10_000_000] {
        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new("records", human(count)),
            &count,
            |b, &count| {
                b.iter(|| {
                    let tmp = NamedTempFile::new().unwrap();
                    let path = tmp.path().to_path_buf();
                    let iter = (0..count).map(make_record);
                    let task = SlabWriter::write_from_iter_async(
                        path,
                        WriterConfig::default(),
                        iter,
                        |_| {},
                    );
                    let n = task.wait().unwrap();
                    assert_eq!(n, count);
                });
            },
        );
    }
    group.finish();
}

/// Measure bulk-write throughput via `add_records`.
///
/// Pre-generates all records, then writes them in a single
/// `add_records` call. The writer is created in the setup phase and
/// `finish()` is excluded from the timed region — only the bulk
/// `add_records` call is measured. Contrasts with the single-record
/// `bench_write` to show the overhead (or lack thereof) of the bulk API.
fn bench_bulk_write(c: &mut Criterion) {
    let mut group = c.benchmark_group("bulk_write");
    group.sample_size(10);

    {
        let count = 1_000_000u64;
        let records: Vec<Vec<u8>> = (0..count).map(make_record).collect();
        let refs: Vec<&[u8]> = records.iter().map(|r| r.as_slice()).collect();

        group.throughput(Throughput::Elements(count));
        group.bench_with_input(
            BenchmarkId::new("records", human(count)),
            &count,
            |b, &_count| {
                b.iter_batched(
                    || {
                        let tmp = NamedTempFile::new().unwrap();
                        let path = tmp.path().to_path_buf();
                        let writer = SlabWriter::new(&path, WriterConfig::default()).unwrap();
                        (tmp, writer)
                    },
                    |(_tmp, mut writer)| {
                        writer.add_records(&refs).unwrap();
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }
    group.finish();
}

/// Measure multi-batch concurrent read throughput via `multi_batch_get`.
///
/// Pre-builds a 1 M-record file, then submits 2/4/8/16 concurrent
/// batches of 10 000 ordinals each. Throughput is reported as total
/// ordinals across all batches per second. Within each iteration, the
/// benchmark asserts that result indices match batch indices (order
/// sensitivity test).
fn bench_multi_batch_read(c: &mut Criterion) {
    let mut group = c.benchmark_group("multi_batch_read");
    group.sample_size(10);

    let record_count = 1_000_000u64;
    let (_tmp, path) = prepare_file(record_count);
    let ordinals_per_batch = 10_000u64;

    for n_batches in [2, 4, 8, 16] {
        let total_ordinals = ordinals_per_batch * n_batches as u64;
        group.throughput(Throughput::Elements(total_ordinals));

        // Pre-build batches with distinct ordinal patterns per batch
        let batches: Vec<Vec<i64>> = (0..n_batches)
            .map(|b| {
                let seed = (b as u64).wrapping_mul(6271);
                (0..ordinals_per_batch)
                    .map(|i| {
                        ((i.wrapping_mul(7919).wrapping_add(seed)) % record_count) as i64
                    })
                    .collect()
            })
            .collect();

        group.bench_with_input(
            BenchmarkId::new(
                format!(
                    "{}b_x_{}_in_{}",
                    n_batches,
                    human(ordinals_per_batch),
                    human(record_count),
                ),
                human(total_ordinals),
            ),
            &n_batches,
            |b, &_n_batches| {
                b.iter_batched(
                    || SlabReader::open(&path).unwrap(),
                    |reader| {
                        let batch_refs: Vec<&[i64]> =
                            batches.iter().map(|v| v.as_slice()).collect();
                        let results = reader.multi_batch_get(&batch_refs);

                        // Order sensitivity: result count matches batch count
                        // and each result has the expected number of ordinals
                        assert_eq!(results.len(), batches.len());
                        for (i, result) in results.iter().enumerate() {
                            assert_eq!(
                                result.records.len(),
                                batches[i].len(),
                                "result {i} record count mismatch"
                            );
                            // Verify ordinals match submission order
                            for (j, (ord, _)) in result.records.iter().enumerate() {
                                assert_eq!(
                                    *ord, batches[i][j],
                                    "result {i} ordinal {j} mismatch"
                                );
                            }
                        }
                    },
                    BatchSize::LargeInput,
                );
            },
        );
    }
    group.finish();
}

criterion_group!(
    benches,
    bench_write,
    bench_sequential_read,
    bench_random_read,
    bench_concurrent_random_read,
    bench_concurrent_sequential_read,
    bench_concurrent_write,
    bench_batch_iter,
    bench_read_all_to_sink,
    bench_read_to_sink_async,
    bench_write_from_iter_async,
    bench_bulk_write,
    bench_multi_batch_read,
);
criterion_main!(benches);
