//! Benchmark gate for П-9 (review 26-06-10): does the tree-global `write_lock`
//! cap multi-threaded insert throughput on a single SkipList / ConstTree?
//!
//! Run: cargo run --release -p armdb --features bench-internals --example insert_scaling
//!
//! Phases:
//!   A. raw SkipList insert scaling (index only, no disk) — write_lock isolated
//!   B. ConstTree insert scaling (end-to-end: shard mutex + write buffer + index)
//!   C. ConstTree update scaling (SeqLock in-place path, bypasses write_lock) — control
//!
//! Each phase inserts a fixed TOTAL of new unique keys split across T threads,
//! so perfect scaling shows wall time dropping ~1/T. CPU ratio = process CPU
//! time / wall time (≈ number of effectively running threads).

use std::time::Instant;

use armdb::skiplist::node::{ConstNode, random_height};
use armdb::skiplist::{InsertResult, SkipList};
use armdb::{Config, ConstTree, DiskLoc};
use tempfile::TempDir;

const PREFILL_LIST: u64 = 8_000_000;
const PREFILL_TREE: u64 = 4_000_000;
const TOTAL_OPS: u64 = 1_000_000;
const THREADS: &[usize] = &[1, 2, 4, 8, 16];

/// splitmix64 — bijection on u64, gives unique pseudo-random keys per index.
fn key64(i: u64) -> u64 {
    let mut z = i.wrapping_add(0x9E3779B97F4A7C15);
    z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
    z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
    z ^ (z >> 31)
}

fn key_bytes(i: u64) -> [u8; 8] {
    key64(i).to_be_bytes()
}

/// Process CPU time (user + sys) in seconds, from /proc/self/stat.
/// Linux-only; returns 0.0 elsewhere — the "CPU x" column is then meaningless.
fn cpu_seconds() -> f64 {
    #[cfg(target_os = "linux")]
    {
        let stat = std::fs::read_to_string("/proc/self/stat").unwrap();
        let rest = stat.rsplit(')').next().unwrap();
        let fields: Vec<&str> = rest.split_whitespace().collect();
        let utime: f64 = fields[11].parse().unwrap();
        let stime: f64 = fields[12].parse().unwrap();
        let ticks = 100.0; // CLK_TCK on Linux
        (utime + stime) / ticks
    }
    #[cfg(not(target_os = "linux"))]
    {
        0.0
    }
}

struct PhaseResult {
    threads: usize,
    wall_s: f64,
    mops: f64,
    cpu_ratio: f64,
}

fn print_table(title: &str, results: &[PhaseResult]) {
    println!("\n=== {title} ===");
    println!(
        "{:>8} {:>10} {:>10} {:>10} {:>9}",
        "threads", "wall s", "Mops/s", "speedup", "CPU x"
    );
    let base = results[0].mops;
    for r in results {
        println!(
            "{:>8} {:>10.2} {:>10.2} {:>9.2}x {:>9.1}",
            r.threads,
            r.wall_s,
            r.mops,
            r.mops / base,
            r.cpu_ratio
        );
    }
}

fn run_phase(threads: usize, total: u64, f: impl Fn(u64, u64) + Sync) -> PhaseResult {
    let per_thread = total / threads as u64;
    let cpu0 = cpu_seconds();
    let t0 = Instant::now();
    std::thread::scope(|s| {
        for t in 0..threads {
            let f = &f;
            let start = t as u64 * per_thread;
            s.spawn(move || f(start, start + per_thread));
        }
    });
    let wall_s = t0.elapsed().as_secs_f64();
    let cpu_s = cpu_seconds() - cpu0;
    PhaseResult {
        threads,
        wall_s,
        mops: (per_thread * threads as u64) as f64 / wall_s / 1e6,
        cpu_ratio: cpu_s / wall_s,
    }
}

// ---------------------------------------------------------------------------
// Phase A: raw SkipList
// ---------------------------------------------------------------------------

type Node = ConstNode<[u8; 8], 8>;

fn phase_a() {
    println!("\n[A] raw SkipList: prefill {PREFILL_LIST} keys…");
    let list: SkipList<Node> = SkipList::new(false);
    let t0 = Instant::now();
    {
        let guard = list.collector().enter();
        for i in 0..PREFILL_LIST {
            let node = Node::alloc(
                key_bytes(i),
                [0u8; 8],
                DiskLoc::new(0, 0, 0),
                random_height(),
            );
            match list.insert(node, &guard) {
                InsertResult::Inserted => {}
                InsertResult::Exists(_) => panic!("duplicate key in prefill"),
            }
        }
    }
    println!("    prefill done in {:.1}s", t0.elapsed().as_secs_f64());

    let mut next_base = PREFILL_LIST;
    let mut results = Vec::new();
    for &t in THREADS {
        let base = next_base;
        next_base += TOTAL_OPS;
        let r = run_phase(t, TOTAL_OPS, |start, end| {
            let guard = list.collector().enter();
            for i in start..end {
                let node = Node::alloc(
                    key_bytes(base + i),
                    [0u8; 8],
                    DiskLoc::new(0, 0, 0),
                    random_height(),
                );
                match list.insert(node, &guard) {
                    InsertResult::Inserted => {}
                    InsertResult::Exists(_) => panic!("duplicate key"),
                }
            }
        });
        results.push(r);
    }
    print_table(
        "Phase A: raw SkipList insert (new keys, write_lock path)",
        &results,
    );
}

// ---------------------------------------------------------------------------
// Phases B & C: ConstTree end-to-end
// ---------------------------------------------------------------------------

fn phase_bc() {
    let dir = TempDir::new().unwrap();
    let mut config = Config::balanced().build();
    config.shard_count = 16; // shard mutexes must not be the suspect
    let tree = ConstTree::<[u8; 8], 16>::open(dir.path(), config).unwrap();

    println!("\n[B] ConstTree: prefill {PREFILL_TREE} keys…");
    let t0 = Instant::now();
    for i in 0..PREFILL_TREE {
        tree.insert(&key_bytes(i), &[0u8; 16]).unwrap();
    }
    println!("    prefill done in {:.1}s", t0.elapsed().as_secs_f64());

    let mut next_base = PREFILL_TREE;
    let mut results = Vec::new();
    for &t in THREADS {
        let base = next_base;
        next_base += TOTAL_OPS;
        let r = run_phase(t, TOTAL_OPS, |start, end| {
            for i in start..end {
                tree.insert(&key_bytes(base + i), &[0u8; 16]).unwrap();
            }
        });
        results.push(r);
    }
    print_table("Phase B: ConstTree insert (new keys, end-to-end)", &results);

    // Phase C: updates over existing keys — SeqLock in-place, no write_lock.
    let mut results = Vec::new();
    for &t in THREADS {
        let r = run_phase(t, TOTAL_OPS, |start, end| {
            for i in start..end {
                // threads touch disjoint ranges inside the prefill keyspace
                let k = key_bytes(i % PREFILL_TREE);
                tree.put(&k, &[1u8; 16]).unwrap();
            }
        });
        results.push(r);
    }
    print_table(
        "Phase C: ConstTree put over existing keys (SeqLock path) — control",
        &results,
    );
}

fn main() {
    println!(
        "insert_scaling gate: {} logical cores",
        std::thread::available_parallelism().unwrap()
    );
    phase_a();
    phase_bc();
}
