// Throughput benchmark: items/second vs. graph topology and size.
//
// Topologies:
//   chain   — linear depth D: push → n[0..D-1] → pop
//   wide    — fanout<W>:      push → fanout → W parallel nodes → W pops
//   diamond — push → fanout<2> → 2×2 nodes → 2 pops
//
// Two scheduling modes for each topology:
//   private — each node owns a private ThreadPool(1)  [Node<>]
//   pool    — all nodes share one ThreadPool(T)        [PoolNode<> + shared pool]
//
// Each row is run --reps times (plus discarded warm-up runs); the reported
// figure is the median items/sec, with the inter-quartile spread as a
// reliability indicator. A row whose iqr_pct is above a few percent is not
// measuring what it claims to measure.
//
// Usage: ./bench_pipeline [options] | tee results.csv
//        ./bench_pipeline --help

#include <kpn/kpn.hpp>

#include "bench_env.hpp"

#ifdef KPN_BENCH_TBB
#include <oneapi/tbb/flow_graph.h>
namespace tbb_flow = oneapi::tbb::flow;
#endif

#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cmath>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include <sys/resource.h>

using namespace kpn;
using namespace std::chrono_literals;
using sclock = std::chrono::steady_clock;

// ── configurable work ─────────────────────────────────────────────────────────

static std::atomic<int> g_work_us{0};

static int chain_fn(int x) {
    int us = g_work_us.load(std::memory_order_relaxed);
    if (us > 0) {
        auto end = sclock::now() + std::chrono::microseconds(us);
        while (sclock::now() < end);
    }
    return x;
}

using ChainNode     = Node<chain_fn, in<>, out<>>;
using PoolChainNode = PoolNode<chain_fn, in<>, out<>>;

// ── push helper: yield-spin on overflow (no artificial sleep latency) ─────────

static void push_retry(Channel<int>& ch, int val) {
    while (true) {
        try   { ch.push(val); return; }
        catch (const ChannelOverflowError&) { std::this_thread::yield(); }
        catch (const ChannelClosedError&)   { return; }
    }
}

// ── configuration (M1, M3, M4, M5) ────────────────────────────────────────────

struct Config {
    std::vector<int> work_amts {10, 100, 1000};
    std::vector<int> pool_sizes{1, 2, 4, 8, 16, 20};   // M5
    std::vector<int> depths    {1, 2, 4, 8, 16, 32};
    std::vector<int> widths    {1, 2, 3, 4};
    int    reps        = 5;      // M3: measured repetitions per row
    int    warmup      = 1;      // M4: discarded repetitions per row
    double target_sec  = 0.30;   // aimed-for duration of one repetition
    long   min_items   = 2000;   // M1: floor, independent of work_us and depth
    double max_sec     = 3.0;    // ceiling; only bites where min_items cannot fit
    bool   do_chain = true, do_wide = true, do_diamond = true;
    bool   do_priv  = true, do_pool = true, do_tbb = true;
};

static Config g_cfg;

// M1 — sample size from a time budget with a hard floor, rather than a
// hand-tuned ladder that collapsed to 50–200 items on exactly the rows under
// investigation.
//
// `stages` is the number of node firings per item; `units` the number of
// threads able to run them concurrently. Steady-state throughput of the
// pipeline is bounded by work_us * stages / units, so that is the per-item
// cost the sample size is derived from. Depth beyond `units` costs throughput;
// depth below it costs only latency, which does not scale the run.
static long pick_items(int work_us, int stages, int units) {
    units = std::max(1, std::min(units, bench::hw_units()));
    const double per_item_us =
        std::max(1.0, static_cast<double>(work_us)) *
        std::max(1.0, static_cast<double>(stages) / units);

    long want = static_cast<long>(g_cfg.target_sec * 1e6 / per_item_us);
    long cap  = static_cast<long>(g_cfg.max_sec    * 1e6 / per_item_us);

    want = std::max(want, g_cfg.min_items);
    // The floor wins unless honouring it would blow the time ceiling by more
    // than the ceiling allows; such rows are reported with their true N so the
    // reader can see they are short.
    if (want > cap) want = std::max(cap, 200L);
    return want;
}

// ── one measured repetition ───────────────────────────────────────────────────

struct Sample {
    double items_per_sec = 0;
    double overhead_us   = 0;
    long   nivcsw        = 0;   // involuntary context switches during the run
    long   nvcsw         = 0;   // voluntary   context switches during the run
};

// ── chain ─────────────────────────────────────────────────────────────────────

static Sample bench_chain(int depth, int work_us, long N) {
    const std::size_t CAP = static_cast<std::size_t>(N);

    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= depth; ++i)
        chs.push_back(std::make_shared<Channel<int>>(CAP));

    std::vector<std::unique_ptr<ChainNode>> nodes;
    for (int i = 0; i < depth; ++i) {
        nodes.push_back(std::make_unique<ChainNode>(CAP));
        nodes.back()->set_input_channel<0>(chs[i]);
        nodes.back()->set_output_channel<0>(chs[i + 1].get());
    }

    for (auto& n : nodes) n->start();

    std::atomic<sclock::time_point> t1;
    std::thread reader([&] {
        for (long i = 0; i < N; ++i) chs.back()->pop();
        t1.store(sclock::now(), std::memory_order_release);
    });

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
        for (long i = 0; i < N; ++i) push_retry(*chs[0], static_cast<int>(i));
    });

    pusher.join();
    reader.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    for (auto& n : nodes) n->stop();

    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    // Subtract theoretical pipeline fill cost (depth-1)*W so that overhead
    // reflects only framework latency, not the expected pipeline startup time.
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
    s.items_per_sec = N / elapsed;
    return s;
}

static Sample bench_chain_pool(int depth, int work_us, int pool_threads, long N) {
    const std::size_t CAP = static_cast<std::size_t>(N);

    auto pool = std::make_shared<ThreadPool>(pool_threads);

    std::vector<std::shared_ptr<Channel<int>>> chs;
    for (int i = 0; i <= depth; ++i)
        chs.push_back(std::make_shared<Channel<int>>(CAP));

    std::vector<std::unique_ptr<PoolChainNode>> nodes;
    for (int i = 0; i < depth; ++i) {
        nodes.push_back(std::make_unique<PoolChainNode>(pool, CAP));
        nodes.back()->set_input_channel<0>(chs[i]);
        nodes.back()->set_output_channel<0>(chs[i + 1].get());
    }

    pool->start();
    for (auto& n : nodes) n->start();

    std::atomic<sclock::time_point> t1;
    std::thread reader([&] {
        for (long i = 0; i < N; ++i) chs.back()->pop();
        t1.store(sclock::now(), std::memory_order_release);
    });

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
        for (long i = 0; i < N; ++i) push_retry(*chs[0], static_cast<int>(i));
    });

    pusher.join();
    reader.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    for (auto& n : nodes) n->stop();
    pool->stop();

    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
    s.items_per_sec = N / elapsed;
    return s;
}

// ── wide (fanout<W>) ──────────────────────────────────────────────────────────

template<std::size_t W>
static Sample bench_wide(int work_us, long N) {
    const std::size_t CAP = static_cast<std::size_t>(N);

    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, W>>(CAP);
    fan->template set_input_channel<0>(src_ch);

    std::array<std::unique_ptr<ChainNode>,    W> nodes;
    std::array<std::shared_ptr<Channel<int>>, W> sink_chs;

    for (std::size_t i = 0; i < W; ++i) {
        nodes[i]    = std::make_unique<ChainNode>(CAP);
        sink_chs[i] = std::make_shared<Channel<int>>(CAP);
        nodes[i]->template set_output_channel<0>(sink_chs[i].get());
    }

    [&]<std::size_t... Is>(std::index_sequence<Is...>) {
        (fan->template set_output_channel<Is>(
             &nodes[Is]->template input_channel<0>()), ...);
    }(std::make_index_sequence<W>{});

    fan->start();
    for (auto& n : nodes) n->start();

    std::array<std::thread, W> readers;
    std::atomic<sclock::time_point> t1;
    std::atomic<int> readers_done{0};

    for (std::size_t w = 0; w < W; ++w) {
        readers[w] = std::thread([&, w] {
            for (long i = 0; i < N; ++i) sink_chs[w]->pop();
            if (readers_done.fetch_add(1, std::memory_order_acq_rel) + 1
                    == static_cast<int>(W))
                t1.store(sclock::now(), std::memory_order_release);
        });
    }

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });

    pusher.join();
    for (auto& r : readers) r.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    for (auto& n : nodes) n->stop();

    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
    s.items_per_sec = N / elapsed;
    return s;
}

template<std::size_t W>
static Sample bench_wide_pool(int work_us, int pool_threads, long N) {
    const std::size_t CAP = static_cast<std::size_t>(N);

    auto pool   = std::make_shared<ThreadPool>(pool_threads);
    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, W>>(CAP);
    fan->template set_input_channel<0>(src_ch);

    std::array<std::unique_ptr<PoolChainNode>, W> nodes;
    std::array<std::shared_ptr<Channel<int>>,  W> sink_chs;

    for (std::size_t i = 0; i < W; ++i) {
        nodes[i]    = std::make_unique<PoolChainNode>(pool, CAP);
        sink_chs[i] = std::make_shared<Channel<int>>(CAP);
        nodes[i]->template set_output_channel<0>(sink_chs[i].get());
    }

    [&]<std::size_t... Is>(std::index_sequence<Is...>) {
        (fan->template set_output_channel<Is>(
             &nodes[Is]->template input_channel<0>()), ...);
    }(std::make_index_sequence<W>{});

    fan->start();
    pool->start();
    for (auto& n : nodes) n->start();

    std::array<std::thread, W> readers;
    std::atomic<sclock::time_point> t1;
    std::atomic<int> readers_done{0};

    for (std::size_t w = 0; w < W; ++w) {
        readers[w] = std::thread([&, w] {
            for (long i = 0; i < N; ++i) sink_chs[w]->pop();
            if (readers_done.fetch_add(1, std::memory_order_acq_rel) + 1
                    == static_cast<int>(W))
                t1.store(sclock::now(), std::memory_order_release);
        });
    }

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });

    pusher.join();
    for (auto& r : readers) r.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    for (auto& n : nodes) n->stop();
    pool->stop();

    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
    s.items_per_sec = N / elapsed;
    return s;
}

// ── diamond ───────────────────────────────────────────────────────────────────

static Sample bench_diamond(int work_us, long N) {
    const std::size_t CAP = static_cast<std::size_t>(N);

    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, 2>>(CAP);
    fan->template set_input_channel<0>(src_ch);

    auto nL  = std::make_unique<ChainNode>(CAP);
    auto nR  = std::make_unique<ChainNode>(CAP);
    auto nL2 = std::make_unique<ChainNode>(CAP);
    auto nR2 = std::make_unique<ChainNode>(CAP);
    auto chL  = std::make_shared<Channel<int>>(CAP);
    auto chR  = std::make_shared<Channel<int>>(CAP);
    auto snkL = std::make_shared<Channel<int>>(CAP);
    auto snkR = std::make_shared<Channel<int>>(CAP);

    fan->template set_output_channel<0>(&nL->template input_channel<0>());
    fan->template set_output_channel<1>(&nR->template input_channel<0>());
    nL->set_output_channel<0>(chL.get());
    nR->set_output_channel<0>(chR.get());
    nL2->set_input_channel<0>(chL);
    nR2->set_input_channel<0>(chR);
    nL2->set_output_channel<0>(snkL.get());
    nR2->set_output_channel<0>(snkR.get());

    fan->start(); nL->start(); nR->start(); nL2->start(); nR2->start();

    std::atomic<sclock::time_point> t1;
    std::atomic<int> done{0};
    auto make_reader = [&](Channel<int>& ch) {
        return std::thread([&] {
            for (long i = 0; i < N; ++i) ch.pop();
            if (done.fetch_add(1, std::memory_order_acq_rel) + 1 == 2)
                t1.store(sclock::now(), std::memory_order_release);
        });
    };
    auto rL = make_reader(*snkL);
    auto rR = make_reader(*snkR);

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });

    pusher.join(); rL.join(); rR.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop(); nL->stop(); nR->stop(); nL2->stop(); nR2->stop();

    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
    s.items_per_sec = N / elapsed;
    return s;
}

static Sample bench_diamond_pool(int work_us, int pool_threads, long N) {
    const std::size_t CAP = static_cast<std::size_t>(N);

    auto pool   = std::make_shared<ThreadPool>(pool_threads);
    auto src_ch = std::make_shared<Channel<int>>(CAP);
    auto fan    = std::make_unique<FanoutNode<int, 2>>(CAP);
    fan->template set_input_channel<0>(src_ch);

    auto nL  = std::make_unique<PoolChainNode>(pool, CAP);
    auto nR  = std::make_unique<PoolChainNode>(pool, CAP);
    auto nL2 = std::make_unique<PoolChainNode>(pool, CAP);
    auto nR2 = std::make_unique<PoolChainNode>(pool, CAP);
    auto chL  = std::make_shared<Channel<int>>(CAP);
    auto chR  = std::make_shared<Channel<int>>(CAP);
    auto snkL = std::make_shared<Channel<int>>(CAP);
    auto snkR = std::make_shared<Channel<int>>(CAP);

    fan->template set_output_channel<0>(&nL->template input_channel<0>());
    fan->template set_output_channel<1>(&nR->template input_channel<0>());
    nL->set_output_channel<0>(chL.get());
    nR->set_output_channel<0>(chR.get());
    nL2->set_input_channel<0>(chL);
    nR2->set_input_channel<0>(chR);
    nL2->set_output_channel<0>(snkL.get());
    nR2->set_output_channel<0>(snkR.get());

    fan->start();
    pool->start();
    nL->start(); nR->start(); nL2->start(); nR2->start();

    std::atomic<sclock::time_point> t1;
    std::atomic<int> done{0};
    auto make_reader = [&](Channel<int>& ch) {
        return std::thread([&] {
            for (long i = 0; i < N; ++i) ch.pop();
            if (done.fetch_add(1, std::memory_order_acq_rel) + 1 == 2)
                t1.store(sclock::now(), std::memory_order_release);
        });
    };
    auto rL = make_reader(*snkL);
    auto rR = make_reader(*snkR);

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    std::thread pusher([&] {
        for (long i = 0; i < N; ++i) push_retry(*src_ch, static_cast<int>(i));
    });

    pusher.join(); rL.join(); rR.join();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);
    fan->stop();
    nL->stop(); nR->stop(); nL2->stop(); nR2->stop();
    pool->stop();

    double elapsed = std::chrono::duration<double>(
        t1.load(std::memory_order_acquire) - t0).count();
    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
    s.items_per_sec = N / elapsed;
    return s;
}

// ── TBB flow graph ────────────────────────────────────────────────────────────
#ifdef KPN_BENCH_TBB

static Sample bench_chain_tbb(int depth, int work_us, long N) {
    tbb_flow::graph g;
    using FN = tbb_flow::function_node<int, int>;
    std::vector<std::unique_ptr<FN>> nodes;
    nodes.reserve(depth);
    for (int i = 0; i < depth; ++i)
        nodes.push_back(std::make_unique<FN>(g, tbb_flow::serial,
            [](int x) -> int { return chain_fn(x); }));
    for (int i = 0; i + 1 < depth; ++i)
        tbb_flow::make_edge(*nodes[i], *nodes[i + 1]);

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < N; ++i) nodes[0]->try_put(static_cast<int>(i));
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);

    double elapsed     = std::chrono::duration<double>(t1 - t0).count();
    double pipeline_us = static_cast<double>(work_us) * (N + depth - 1);
    s.overhead_us   = (elapsed * 1e6 - pipeline_us) / N;
    s.items_per_sec = N / elapsed;
    return s;
}

template<std::size_t W>
static Sample bench_wide_tbb(int work_us, long N) {
    tbb_flow::graph g;
    tbb_flow::broadcast_node<int> fan(g);
    using FN = tbb_flow::function_node<int, int>;
    std::array<std::unique_ptr<FN>, W> nodes;
    for (auto& n : nodes) {
        n = std::make_unique<FN>(g, tbb_flow::serial,
            [](int x) -> int { return chain_fn(x); });
        tbb_flow::make_edge(fan, *n);
    }

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < N; ++i) fan.try_put(static_cast<int>(i));
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);

    double elapsed = std::chrono::duration<double>(t1 - t0).count();
    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
    s.items_per_sec = N / elapsed;
    return s;
}

static Sample bench_diamond_tbb(int work_us, long N) {
    tbb_flow::graph g;
    tbb_flow::broadcast_node<int> fan(g);
    using FN = tbb_flow::function_node<int, int>;
    auto fn = [](int x) -> int { return chain_fn(x); };
    FN nL(g, tbb_flow::serial, fn), nR(g, tbb_flow::serial, fn);
    FN nL2(g, tbb_flow::serial, fn), nR2(g, tbb_flow::serial, fn);
    tbb_flow::make_edge(fan, nL);  tbb_flow::make_edge(fan, nR);
    tbb_flow::make_edge(nL, nL2);  tbb_flow::make_edge(nR, nR2);

    bench::RusageDelta ru; ru.start();
    auto t0 = sclock::now();
    for (long i = 0; i < N; ++i) fan.try_put(static_cast<int>(i));
    g.wait_for_all();
    auto t1 = sclock::now();
    Sample s;
    ru.finish(s.nivcsw, s.nvcsw);

    double elapsed = std::chrono::duration<double>(t1 - t0).count();
    s.overhead_us   = (elapsed * 1e6) / N - static_cast<double>(work_us);
    s.items_per_sec = N / elapsed;
    return s;
}
#endif // KPN_BENCH_TBB

// ── repetition driver (M2, M3, M4) ────────────────────────────────────────────

using bench::percentile;

// A row: median of `reps` repetitions, after `warmup` discarded ones.
// M2 — items/sec is the primary figure; derived overhead is secondary,
// because it is a difference of large numbers and magnifies noise ~10×.
template<class Fn>
static void run_row(const char* topology, int size, int work_us, int sched,
                    long N, Fn&& one_rep) {
    for (int i = 0; i < g_cfg.warmup; ++i) (void)one_rep();   // M4

    std::vector<double> ips, ovh;
    long ivcsw = 0, vcsw = 0;
    for (int i = 0; i < g_cfg.reps; ++i) {
        Sample s = one_rep();
        ips.push_back(s.items_per_sec);
        ovh.push_back(s.overhead_us);
        ivcsw += s.nivcsw;
        vcsw  += s.nvcsw;
    }

    const double med  = percentile(ips, 0.5);
    const double q1   = percentile(ips, 0.25);
    const double q3   = percentile(ips, 0.75);
    const double iqr  = med > 0 ? 100.0 * (q3 - q1) / med : 0.0;
    const double lo   = *std::min_element(ips.begin(), ips.end());
    const double hi   = *std::max_element(ips.begin(), ips.end());
    const double spread = med > 0 ? 100.0 * (hi - lo) / med : 0.0;
    const double ivcsw_per_item = static_cast<double>(ivcsw) / (double(N) * g_cfg.reps);
    const double vcsw_per_item  = static_cast<double>(vcsw)  / (double(N) * g_cfg.reps);

    const std::string s = sched  < 0 ? "tbb"
                        : sched == 0 ? "priv"
                                     : std::to_string(sched);

    std::fprintf(stderr, "%-10s %-5d %-8d %-6s %-8ld %-12.0f %-7.1f %-7.1f %-9.1f %-8.2f %-8.2f\n",
                 topology, size, work_us, s.c_str(), N,
                 med, iqr, spread, percentile(ovh, 0.5), ivcsw_per_item, vcsw_per_item);
    std::printf("%s,%d,%d,%s,%ld,%d,%.0f,%.0f,%.0f,%.2f,%.2f,%.2f,%.3f,%.3f\n",
                topology, size, work_us, s.c_str(), N, g_cfg.reps,
                med, lo, hi, iqr, spread, percentile(ovh, 0.5),
                ivcsw_per_item, vcsw_per_item);
    std::fflush(stdout);
}

// ── argument parsing ──────────────────────────────────────────────────────────

static std::vector<int> parse_int_list(const char* s) {
    std::vector<int> out;
    const char* p = s;
    while (*p) {
        char* end = nullptr;
        long v = std::strtol(p, &end, 10);
        if (end == p) break;
        out.push_back(static_cast<int>(v));
        p = end;
        while (*p == ',' || *p == ' ') ++p;
    }
    return out;
}

static bool has_word(const std::string& csv, const char* word) {
    return csv.find(word) != std::string::npos;
}

static void usage() {
    std::fprintf(stderr,
        "usage: bench_pipeline [options]\n"
        "  --work=10,100,1000     per-node busy-work, microseconds\n"
        "  --depths=1,2,4,8,16,32 chain depths\n"
        "  --widths=1,2,3,4       fanout widths\n"
        "  --pools=1,2,4,8,16,20  shared-pool thread counts\n"
        "  --topos=chain,wide,diamond\n"
        "  --modes=priv,pool,tbb\n"
        "  --reps=5               measured repetitions per row\n"
        "  --warmup=1             discarded repetitions per row\n"
        "  --target-sec=0.30      aimed-for duration of one repetition\n"
        "  --min-items=2000       sample-size floor\n"
        "  --max-sec=3.0          per-repetition ceiling (overrides the floor)\n");
}

static bool parse_args(int argc, char** argv) {
    for (int i = 1; i < argc; ++i) {
        std::string a = argv[i];
        auto eq = a.find('=');
        std::string key = a.substr(0, eq);
        std::string val = eq == std::string::npos ? "" : a.substr(eq + 1);

        if (key == "--help" || key == "-h")   { usage(); std::exit(0); }
        else if (key == "--work")       g_cfg.work_amts  = parse_int_list(val.c_str());
        else if (key == "--depths")     g_cfg.depths     = parse_int_list(val.c_str());
        else if (key == "--widths")     g_cfg.widths     = parse_int_list(val.c_str());
        else if (key == "--pools")      g_cfg.pool_sizes = parse_int_list(val.c_str());
        else if (key == "--reps")       g_cfg.reps       = std::atoi(val.c_str());
        else if (key == "--warmup")     g_cfg.warmup     = std::atoi(val.c_str());
        else if (key == "--target-sec") g_cfg.target_sec = std::atof(val.c_str());
        else if (key == "--min-items")  g_cfg.min_items  = std::atol(val.c_str());
        else if (key == "--max-sec")    g_cfg.max_sec    = std::atof(val.c_str());
        else if (key == "--topos") {
            g_cfg.do_chain   = has_word(val, "chain");
            g_cfg.do_wide    = has_word(val, "wide");
            g_cfg.do_diamond = has_word(val, "diamond");
        }
        else if (key == "--modes") {
            g_cfg.do_priv = has_word(val, "priv");
            g_cfg.do_pool = has_word(val, "pool");
            g_cfg.do_tbb  = has_word(val, "tbb");
        }
        else { std::fprintf(stderr, "unknown option: %s\n", a.c_str()); usage(); return false; }
    }
    if (g_cfg.reps < 1) g_cfg.reps = 1;
    if (g_cfg.warmup < 0) g_cfg.warmup = 0;
    return true;
}

// `wide` is templated on W, so dispatch the runtime width through a switch.
template<class F>
static void with_width(int w, F&& f) {
    switch (w) {
        case 1: f(std::integral_constant<std::size_t, 1>{}); break;
        case 2: f(std::integral_constant<std::size_t, 2>{}); break;
        case 3: f(std::integral_constant<std::size_t, 3>{}); break;
        case 4: f(std::integral_constant<std::size_t, 4>{}); break;
        default:
            std::fprintf(stderr, "width %d not instantiated (1..4 only)\n", w);
    }
}

// ── main ──────────────────────────────────────────────────────────────────────

int main(int argc, char** argv) {
    // A rejected option must fail loudly: a harness driver that silently got
    // no CSV back is worse than one that stops.
    if (!parse_args(argc, argv)) return 2;

    char cfg[192];
    std::snprintf(cfg, sizeof cfg,
                  "reps=%d warmup=%d target_sec=%.2f min_items=%ld max_sec=%.1f",
                  g_cfg.reps, g_cfg.warmup, g_cfg.target_sec,
                  g_cfg.min_items, g_cfg.max_sec);
    bench::print_environment(cfg);

    std::fprintf(stderr, "\n%-10s %-5s %-8s %-6s %-8s %-12s %-7s %-7s %-9s %-8s %-8s\n",
                 "topology", "size", "work_us", "sched", "items", "items/sec",
                 "iqr%", "range%", "ovh_us", "ivcsw/it", "vcsw/it");
    std::fprintf(stderr, "%s\n", std::string(104, '-').c_str());
    std::printf("topology,size,work_us,threads,items,reps,items_per_sec,"
                "items_per_sec_min,items_per_sec_max,iqr_pct,range_pct,"
                "overhead_us_per_item,ivcsw_per_item,vcsw_per_item\n");

    for (int w : g_cfg.work_amts) {
        g_work_us.store(w, std::memory_order_relaxed);

        if (g_cfg.do_priv) {
            std::fprintf(stderr, "\n── work_us=%-4d private pools ──────────────────────\n", w);
            if (g_cfg.do_chain)
                for (int d : g_cfg.depths) {
                    long N = pick_items(w, d, d);
                    run_row("chain", d, w, 0, N, [&] { return bench_chain(d, w, N); });
                }
            if (g_cfg.do_wide)
                for (int wd : g_cfg.widths)
                    with_width(wd, [&](auto W) {
                        long N = pick_items(w, W.value, W.value);
                        run_row("wide", static_cast<int>(W.value), w, 0, N,
                                [&] { return bench_wide<W.value>(w, N); });
                    });
            if (g_cfg.do_diamond) {
                long N = pick_items(w, 4, 4);
                run_row("diamond", 4, w, 0, N, [&] { return bench_diamond(w, N); });
            }
        }

        if (g_cfg.do_pool) {
            for (int pt : g_cfg.pool_sizes) {
                std::fprintf(stderr, "\n── work_us=%-4d shared pool (%d thread%s) ───────────\n",
                             w, pt, pt == 1 ? "" : "s");
                if (g_cfg.do_chain)
                    for (int d : g_cfg.depths) {
                        long N = pick_items(w, d, pt);
                        run_row("chain", d, w, pt, N,
                                [&] { return bench_chain_pool(d, w, pt, N); });
                    }
                if (g_cfg.do_wide)
                    for (int wd : g_cfg.widths)
                        with_width(wd, [&](auto W) {
                            long N = pick_items(w, W.value, pt);
                            run_row("wide", static_cast<int>(W.value), w, pt, N,
                                    [&] { return bench_wide_pool<W.value>(w, pt, N); });
                        });
                if (g_cfg.do_diamond) {
                    long N = pick_items(w, 4, pt);
                    run_row("diamond", 4, w, pt, N,
                            [&] { return bench_diamond_pool(w, pt, N); });
                }
            }
        }

#ifdef KPN_BENCH_TBB
        if (g_cfg.do_tbb) {
            std::fprintf(stderr, "\n── work_us=%-4d TBB flow graph ─────────────────────\n", w);
            if (g_cfg.do_chain)
                for (int d : g_cfg.depths) {
                    long N = pick_items(w, d, d);
                    run_row("chain_tbb", d, w, -1, N,
                            [&] { return bench_chain_tbb(d, w, N); });
                }
            if (g_cfg.do_wide)
                for (int wd : g_cfg.widths)
                    with_width(wd, [&](auto W) {
                        long N = pick_items(w, W.value, W.value);
                        run_row("wide_tbb", static_cast<int>(W.value), w, -1, N,
                                [&] { return bench_wide_tbb<W.value>(w, N); });
                    });
            if (g_cfg.do_diamond) {
                long N = pick_items(w, 4, 4);
                run_row("diamond_tbb", 4, w, -1, N,
                        [&] { return bench_diamond_tbb(w, N); });
            }
        }
#endif
    }
}