/// @file dpf/party_runner.hpp
/// @brief Run a composed protocol as 2 or 3 parties, in one process or many.
/// @details The layout of edges comes from the plan: a 2PC peer edge between
///          parties 0 and 1, an RSS ring (each party sends to the previous one
///          and receives from the next), and dealer edges from party 2 to
///          parties 0 and 1. Every pair of parties shares one link; each
///          logical edge uses its own lane range on it.
///
///          `run_parties` drives all parties on threads over the transport in
///          `run_config` (in-process async memory, unix sockets, TCP mux,
///          parallel TCP, or SCTP). `run_node` drives one party of a
///          multi-process run from a static `host:port` table
///          (`--party=1 --peers=a:9000,b:9000,c:9000`).
#ifndef LIBDPF_INCLUDE_DPF_PARTY_RUNNER_HPP__
#define LIBDPF_INCLUDE_DPF_PARTY_RUNNER_HPP__

#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <ctime>
#include <exception>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <string>
#include <thread>
#include <utility>
#include <vector>

#if defined(__linux__)
#include <pthread.h>
#include <sched.h>
#endif

#include "dpf/net/asio_ns.hpp"

#include "dpf/compose.hpp"
#include "dpf/compose_async.hpp"
#include "dpf/experiment.hpp"
#include "dpf/log.hpp"
#include "dpf/net/async_round_sink.hpp"
#include "dpf/net/async_stream_array.hpp"
#include "dpf/net/io_pool.hpp"
#include "dpf/net/party_session.hpp"
#include "dpf/prg_count.hpp"
#include "dpf/random.hpp"
#include "dpf/run_config.hpp"
#include "dpf/thread_work.hpp"

namespace dpf
{
namespace app
{

using party_values = std::vector<std::vector<std::uint8_t>>;

/// @brief Lane ranges of each logical edge on a pair's link.
struct edge_layout
{
    std::size_t parties = 2;
    std::vector<std::size_t> peer_slots;
    std::vector<std::size_t> ring_slots;
    std::vector<std::size_t> dealer_slots;
    std::size_t peer_lanes = 0;
    std::size_t ring_lanes = 0;
    std::size_t dealer_lanes = 0;
    std::size_t peer_off = 0;
    std::size_t ring_off = 0;
    std::size_t dealer_off = 0;
    std::size_t pair_lanes = 1;
};

inline edge_layout layout_for(const protocol::plan & p, std::size_t parties,
    const run_config & cfg)
{
    if (parties < 2 || parties > 3)
        throw std::invalid_argument("run_parties: 2 or 3 parties (got "
            + std::to_string(parties) + ")");
    edge_layout L;
    L.parties = parties;
    const auto s = protocol::slot_bytes_by_channel(p);
    L.peer_slots = s.peer;
    L.ring_slots = s.rss_next;
    L.dealer_slots = s.dealer;
    auto lanes = [&](const std::vector<std::size_t> & v) {
        return v.empty() ? std::size_t{0}
                         : net::lane_count_for_rounds(v.size(), cfg.n_lanes);
    };
    L.peer_lanes = lanes(s.peer);
    L.ring_lanes = lanes(s.rss_next);
    L.dealer_lanes = lanes(s.dealer);
    if (L.dealer_lanes != 0 && parties < 3)
        throw std::invalid_argument("run_parties: the plan has dealer waves; "
            "pass a plan for party 2 (the dealer) as well");
    L.ring_off = L.peer_lanes;
    L.dealer_off = L.peer_lanes + L.ring_lanes;
    L.pair_lanes = std::max<std::size_t>(1, L.dealer_off + L.dealer_lanes);
    return L;
}

/// @brief One party's sinks over its links (`to[p]` = link to party `p`).
class party_edges
{
public:
    party_edges(unsigned me, const edge_layout & L,
        const std::vector<net::async_stream_array *> & to, std::size_t instances,
        const net::sink_options & so)
    {
        const unsigned n = static_cast<unsigned>(L.parties);
        auto view = [&](unsigned other, std::size_t off,
                        std::size_t lanes) -> net::async_stream_array & {
            if (other >= to.size() || to[other] == nullptr)
                throw std::logic_error("party " + std::to_string(me)
                    + " has no link to party " + std::to_string(other));
            views_.push_back(
                std::make_unique<net::async_stream_view>(*to[other], off, lanes));
            return *views_.back();
        };
        net::sink_options single = so;
        net::sink_options split = so;
        split.reconnect = nullptr;
        if (L.peer_lanes != 0 && me < 2)
            peer_ = std::make_unique<net::async_round_sink>(
                view(1 - me, L.peer_off, L.peer_lanes), L.peer_slots, instances,
                single);
        if (L.ring_lanes != 0)
        {
            const unsigned next = (me + 1) % n;
            const unsigned prev = (me + n - 1) % n;
            auto & out = view(prev, L.ring_off, L.ring_lanes);
            if (next == prev)
                ring_ = std::make_unique<net::async_round_sink>(out, L.ring_slots,
                    instances, single);
            else
                ring_ = std::make_unique<net::async_round_sink>(out,
                    view(next, L.ring_off, L.ring_lanes), L.ring_slots, instances,
                    split);
        }
        if (L.dealer_lanes != 0)
        {
            if (me < 2)
                dealer_ = std::make_unique<net::async_round_sink>(
                    view(2, L.dealer_off, L.dealer_lanes), L.dealer_slots,
                    instances, single);
            else
            {
                dealer_ = std::make_unique<net::async_round_sink>(
                    view(0, L.dealer_off, L.dealer_lanes), L.dealer_slots,
                    instances, single);
                dealer_p1_ = std::make_unique<net::async_round_sink>(
                    view(1, L.dealer_off, L.dealer_lanes), L.dealer_slots,
                    instances, single);
            }
        }
    }

    protocol::edge_mesh mesh() const
    {
        protocol::edge_mesh m;
        m.sinks = {peer_.get(), ring_.get(), dealer_.get(), dealer_p1_.get()};
        return m;
    }

    std::vector<net::async_round_sink *> sinks() const
    {
        std::vector<net::async_round_sink *> out;
        for (auto * s : {peer_.get(), ring_.get(), dealer_.get(), dealer_p1_.get()})
            if (s != nullptr)
                out.push_back(s);
        return out;
    }

private:
    std::vector<std::unique_ptr<net::async_stream_view>> views_;
    std::unique_ptr<net::async_round_sink> peer_;
    std::unique_ptr<net::async_round_sink> ring_;
    std::unique_ptr<net::async_round_sink> dealer_;
    std::unique_ptr<net::async_round_sink> dealer_p1_;
};

/// @brief Summed counters over a party's distinct links.
inline net::stream_stats sum_stats(const std::vector<net::async_stream_array *> & to)
{
    net::stream_stats t;
    for (auto * l : to)
    {
        if (l == nullptr)
            continue;
        const auto s = l->stats();
        t.bytes_out += s.bytes_out;
        t.bytes_in += s.bytes_in;
        t.payload_out += s.payload_out;
        t.payload_in += s.payload_in;
        t.frames_out += s.frames_out;
        t.frames_in += s.frames_in;
        t.write_calls += s.write_calls;
    }
    return t;
}

/// @brief Drive party `me`'s plan over its links.
inline void drive_party(unsigned me, const protocol::plan & plan,
    party_values & values, const std::map<std::uint32_t, protocol::kernel_fn> & kernels,
    const std::vector<net::async_stream_array *> & to, const run_config & cfg,
    const protocol::round_probe * probe = nullptr, net::io_pool * pool = nullptr,
    protocol::cell_fn cell = nullptr)
{
    const auto L = layout_for(plan, to.size(), cfg);
    net::sink_options so;
    so.framing = cfg.framing;
    if (cfg.wait_timeout.count() != 0)
        so.drain_timeout = cfg.wait_timeout;
    party_edges edges(me, L, to, cfg.instances, so);
    auto opt = cfg.drive();
    opt.probe = probe;
    opt.cell = cell;
    if (pool != nullptr)
    {
        opt.workers = pool;
        for (auto * l : to)
            if (l != nullptr)
            {
                opt.pump = &l->context();
                break;
            }
    }
    auto mesh = edges.mesh();
    if (edges.sinks().empty())
    {
        protocol::detail::finish_steps(plan, {}, mesh, values, kernels, me,
            cfg.instances, opt);
        return;
    }
    protocol::drive_via_schedule(plan, mesh, values, kernels, me, opt);
}

namespace detail
{

inline void pin_thread(int cpu)
{
#if defined(__linux__)
    if (cpu < 0)
        return;
    cpu_set_t set;
    CPU_ZERO(&set);
    CPU_SET(static_cast<unsigned>(cpu), &set);
    const int rc = pthread_setaffinity_np(pthread_self(), sizeof(set), &set);
    if (rc != 0)
        DPF_LOG(warning, "pin.failed").kv("cpu", cpu).kv("error", std::strerror(rc));
    else
        DPF_LOG(debug, "pin").kv("cpu", cpu);
#else
    if (cpu >= 0)
        DPF_LOG(warning, "pin.failed").kv("cpu", cpu).kv("error", "not supported");
#endif
}

/// @brief Holds every party until all have finished setup, so no party's
///        clock starts while another is still connecting or spawning.
class start_gate
{
public:
    explicit start_gate(std::size_t parties) : left_(parties) {}

    /// @brief False when another party failed before arriving.
    bool arrive_and_wait()
    {
        std::unique_lock<std::mutex> lock(mu_);
        if (broken_)
            return false;
        if (--left_ == 0)
        {
            cv_.notify_all();
            return true;
        }
        cv_.wait(lock, [this] { return left_ == 0 || broken_; });
        return !broken_;
    }

    /// @brief Release waiters; they return false.
    void fail()
    {
        std::lock_guard<std::mutex> lock(mu_);
        if (left_ != 0)
            broken_ = true;
        cv_.notify_all();
    }

private:
    std::mutex mu_;
    std::condition_variable cv_;
    std::size_t left_ = 0;
    bool broken_ = false;
};

/// @brief Thrown by a party released from the gate because another failed.
struct gate_broken : std::runtime_error
{
    gate_broken()
      : std::runtime_error("run_parties: another party failed during setup")
    {
    }
};

inline std::string error_text(const std::exception_ptr & e)
{
    try
    {
        std::rethrow_exception(e);
    }
    catch (const std::exception & x)
    {
        return x.what();
    }
    catch (...)
    {
        return "non-standard exception";
    }
}

/// @brief This thread's wall time, CPU time, symmetric-key blocks, and random
///        bytes from construction to `done`. Other threads are not included.
struct party_meter
{
    std::chrono::steady_clock::time_point t0 = std::chrono::steady_clock::now();
    std::uint64_t cpu0 = dpf::thread_cpu_ns();
    prg::counts sym0 = prg::snapshot();
    std::uint64_t rnd0 = random_bytes_count();
    std::uint64_t wall_ns = 0;
    std::uint64_t cpu_ns = 0;
    prg::counts sym{};
    std::uint64_t random_bytes = 0;

    void done()
    {
        wall_ns = static_cast<std::uint64_t>(std::chrono::duration_cast<
            std::chrono::nanoseconds>(std::chrono::steady_clock::now() - t0).count());
        cpu_ns = dpf::thread_cpu_ns() - cpu0;
        const auto now = prg::snapshot();
        for (std::size_t k = 0; k < now.size(); ++k)
            sym[k] = now[k] - sym0[k];
        random_bytes = random_bytes_count() - rnd0;
    }
};

inline void log_plan(unsigned me, const protocol::plan & p, const edge_layout & L,
    const run_config & cfg)
{
    std::size_t bytes = 0;
    for (auto b : p.slot_bytes_all())
        bytes += b;
    DPF_LOG(info, "plan").kv("party", me).kv("parties", L.parties)
        .kv("rounds", p.rounds()).kv("waves", p.waves()).kv("slot_bytes", bytes)
        .kv("peer_rounds", L.peer_slots.size()).kv("ring_rounds", L.ring_slots.size())
        .kv("dealer_rounds", L.dealer_slots.size()).kv("peer_lanes", L.peer_lanes)
        .kv("ring_lanes", L.ring_lanes).kv("dealer_lanes", L.dealer_lanes)
        .kv("pair_lanes", L.pair_lanes).kv("instances", cfg.instances)
        .kv("framing", net::framing_name(cfg.framing))
        .kv("pipeline", cfg.pipeline_credit);
}

inline void log_party_done(unsigned me, const party_meter & m,
    const net::stream_stats & w, std::size_t compute_threads)
{
    if (!log::enabled(log::level::info))
        return;
    log::record rec(log::level::info, "party.done");
    std::uint64_t protocol_blocks = 0;
    for (std::size_t k = 0; k < 2 * prg::primitive_count; ++k)
        protocol_blocks += m.sym[k];
    rec.kv("party", me).kv("wall_ns", m.wall_ns).kv("cpu_ns", m.cpu_ns)
        .kv("counted_threads", compute_threads == 0 ? "party" : "party+pool")
        .kv("sym_blocks", protocol_blocks);
    for (std::size_t u = 0; u < prg::purpose_count; ++u)
        for (std::size_t q = 0; q < prg::primitive_count; ++q)
        {
            const auto n = m.sym[u * prg::primitive_count + q];
            if (n == 0)
                continue;
            const std::string key = std::string("sym.")
                + prg::purpose_name(static_cast<prg::purpose>(u)) + "."
                + prg::primitive_name(static_cast<prg::primitive>(q));
            rec.kv(key.c_str(), n);
        }
    rec.kv("random_bytes", m.random_bytes).kv("wire_out", w.bytes_out)
        .kv("wire_in", w.bytes_in).kv("payload_out", w.payload_out)
        .kv("payload_in", w.payload_in).kv("frames_out", w.frames_out)
        .kv("frames_in", w.frames_in).kv("write_calls", w.write_calls);
}

}  // namespace detail

/// @brief What one `run_parties` call measured.
struct parties_result
{
    std::vector<net::stream_stats> wire;  ///< per party, summed over links
    std::uint64_t party0_wall_ns = 0;
    /// Each party's drive time on its own thread (party 0 is `party0_wall_ns`).
    std::vector<std::uint64_t> party_wall_ns;
};

/// @brief Drive every party's plan on its own thread over `cfg.kind`.
/// @details `plans[i]` and `values[i]` belong to party `i` (2 or 3 parties).
///          Link setup is outside the timed region, and every party waits at
///          a start gate until all have finished setup. A party that fails
///          closes its links so the others fail fast. When `ex` is set, party
///          0's thread records its round probe and wall/CPU/PRG timing there.
///          When `seeds` (default `ex`) is set, party `i` draws from
///          `seeds->derive_party(i)`, so the run replays from that master;
///          the parties' noted seeds are folded into `ex`.
inline parties_result run_parties(const std::vector<protocol::plan> & plans,
    std::vector<party_values> & values,
    const std::map<std::uint32_t, protocol::kernel_fn> & kernels = {},
    const run_config & cfg = {}, experiment * ex = nullptr,
    protocol::cell_fn cell = nullptr, const experiment * seeds = nullptr)
{
    if (seeds == nullptr)
        seeds = ex;
    const std::size_t n = plans.size();
    if (n < 2 || n > 3)
        throw std::invalid_argument("run_parties: 2 or 3 plans");
    if (values.size() < n)
        values.resize(n);
    std::size_t pair_lanes = 1;
    for (std::size_t i = 0; i < n; ++i)
        pair_lanes = std::max(pair_lanes, layout_for(plans[i], n, cfg).pair_lanes);

    std::vector<std::unique_ptr<asio::io_context>> ios;
    for (std::size_t i = 0; i < n; ++i)
        ios.push_back(std::make_unique<asio::io_context>());
    std::vector<std::vector<net::async_stream_array *>> to(n,
        std::vector<net::async_stream_array *>(n, nullptr));
    std::vector<std::unique_ptr<net::async_stream_array>> owned;

    const bool sockets = cfg.kind == net::transport::mux
        || cfg.kind == net::transport::parallel || cfg.kind == net::transport::sctp;
    if (cfg.kind == net::transport::async_memory)
    {
        for (std::size_t a = 0; a < n; ++a)
            for (std::size_t b = a + 1; b < n; ++b)
            {
                auto pr = net::make_async_dual_memory_stream_pair(*ios[a], *ios[b],
                    pair_lanes, cfg.policy.window_bytes);
                owned.push_back(std::make_unique<net::async_dual_memory_stream_array>(
                    std::move(pr.first)));
                to[a][b] = owned.back().get();
                owned.push_back(std::make_unique<net::async_dual_memory_stream_array>(
                    std::move(pr.second)));
                to[b][a] = owned.back().get();
            }
    }
    else if (cfg.kind == net::transport::local)
    {
        for (std::size_t a = 0; a < n; ++a)
            for (std::size_t b = a + 1; b < n; ++b)
            {
                auto pr = net::make_local_socket_pairs(*ios[a], *ios[b], pair_lanes);
                owned.push_back(std::make_unique<net::async_local_parallel_stream_array>(
                    *ios[a], std::move(pr.first), cfg.policy));
                to[a][b] = owned.back().get();
                owned.push_back(std::make_unique<net::async_local_parallel_stream_array>(
                    *ios[b], std::move(pr.second), cfg.policy));
                to[b][a] = owned.back().get();
            }
    }
    else if (!sockets)
        throw std::invalid_argument(std::string("run_parties: transport ")
            + net::transport_name(cfg.kind)
            + " is not an async link (use async|local|mux|parallel|sctp)");

    DPF_LOG(info, "parties.start").kv("parties", n)
        .kv("transport", net::transport_name(cfg.kind))
        .kv("host", sockets ? cfg.host : std::string("in-process"))
        .kv("pair_lanes", pair_lanes).kv("instances", cfg.instances)
        .kv("compute_threads", cfg.compute_threads)
        .kv("cpu0", cfg.cpu[0]).kv("cpu1", cfg.cpu[1]).kv("cpu2", cfg.cpu[2])
        .kv("experiment", ex != nullptr ? ex->name() : std::string("none"))
        .kv("draws", seeds != nullptr ? "derived from the master" : "os entropy");

    auto ports = net::make_mesh_ports(static_cast<unsigned>(n));
    parties_result result;
    result.wire.resize(n);
    result.party_wall_ns.assign(n, 0);
    std::mutex err_mu;
    std::exception_ptr err;
    std::size_t failures = 0;
    detail::start_gate gate(n);
    std::vector<std::vector<experiment::noted_seed>> party_seeds(n);
    std::vector<std::thread> ts;
    for (std::size_t i = 0; i < n; ++i)
    {
        ts.emplace_back([&, i] {
            const log::role_scope role("p" + std::to_string(i));
            detail::pin_thread(cfg.cpu[i]);
            auto work = asio::make_work_guard(*ios[i]);
            std::unique_ptr<net::party_session> session;
            auto links = to[i];
            try
            {
                if (sockets)
                {
                    net::session_options so;
                    so.n_lanes = pair_lanes;
                    so.kind = cfg.kind;
                    so.policy = cfg.policy;
                    so.limits = cfg.limits;
                    so.security = cfg.security;
                    session = std::make_unique<net::party_session>(*ios[i],
                        static_cast<unsigned>(i), static_cast<unsigned>(n), so);
                    session->join(cfg.host, ports);
                    for (std::size_t p = 0; p < n; ++p)
                        if (p != i)
                            links[p] = &session->peer(static_cast<unsigned>(p));
                }
                std::unique_ptr<net::io_pool> pool;
                if (cfg.compute_threads != 0)
                    pool = std::make_unique<net::io_pool>(1, cfg.compute_threads);
                if (log::enabled(log::level::info))
                    detail::log_plan(static_cast<unsigned>(i), plans[i],
                        layout_for(plans[i], n, cfg), cfg);
                std::optional<experiment> stream;
                if (seeds != nullptr)
                    stream.emplace(seeds->derive_party(static_cast<unsigned>(i)));
                if (!gate.arrive_and_wait())
                    throw detail::gate_broken();
                experiment * mine = i == 0 ? ex : nullptr;
                protocol::round_probe probe{};
                if (mine != nullptr)
                {
                    probe = mine->probe();
                    mine->begin_timing();
                }
                detail::party_meter meter;
                const auto t0 = std::chrono::steady_clock::now();
                drive_party(static_cast<unsigned>(i), plans[i], values[i], kernels,
                    links, cfg, mine != nullptr ? &probe : nullptr, pool.get(),
                    cell);
                const auto t1 = std::chrono::steady_clock::now();
                meter.done();
                if (mine != nullptr)
                    mine->end_timing();
                result.party_wall_ns[i] = static_cast<std::uint64_t>(
                    std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0)
                        .count());
                if (i == 0)
                    result.party0_wall_ns = result.party_wall_ns[i];
                result.wire[i] = sum_stats(links);
                detail::log_party_done(static_cast<unsigned>(i), meter, result.wire[i],
                    cfg.compute_threads);
                if (stream)
                    party_seeds[i] = stream->seeds();
            }
            catch (...)
            {
                const auto e = std::current_exception();
                bool released = false;
                try
                {
                    std::rethrow_exception(e);
                }
                catch (const detail::gate_broken &)
                {
                    released = true;
                }
                catch (...)
                {
                }
                gate.fail();
                if (!released)
                    DPF_LOG(error, "party.failed").kv("party", i)
                        .kv("what", detail::error_text(e));
                {
                    std::lock_guard<std::mutex> lock(err_mu);
                    if (!released)
                    {
                        ++failures;
                        if (!err)
                            err = e;
                    }
                }
                for (auto * l : links)
                    if (l != nullptr)
                        l->close();
            }
            work.reset();
        });
    }
    for (auto & t : ts)
        t.join();
    if (err)
    {
        DPF_LOG(error, "parties.failed").kv("parties", n).kv("failed", failures)
            .kv("rethrown", detail::error_text(err));
        std::rethrow_exception(err);
    }
    if (ex != nullptr && seeds != nullptr)
        for (std::size_t i = 0; i < n; ++i)
            ex->fold_seeds("p" + std::to_string(i), party_seeds[i]);
    if (log::enabled(log::level::info))
    {
        const auto slowest = *std::max_element(result.party_wall_ns.begin(),
            result.party_wall_ns.end());
        DPF_LOG(info, "parties.done").kv("parties", n)
            .kv("p0_wall_ns", result.party0_wall_ns).kv("max_wall_ns", slowest);
    }
    return result;
}

/// @brief One node of a multi-process run.
struct node_args
{
    unsigned party = 0;
    std::vector<net::peer_address> peers;
    run_config cfg;
    /// The transport that was asked for when it was not a network socket and
    /// `mux` ran instead (empty otherwise).
    std::string replaced_transport;
};

namespace detail
{

inline unsigned long parse_count(const std::string & what, const std::string & v)
{
    std::size_t pos = 0;
    unsigned long x = 0;
    try
    {
        x = std::stoul(v, &pos, 10);
    }
    catch (const std::exception &)
    {
        pos = 0;
    }
    if (v.empty() || pos != v.size())
        throw std::invalid_argument(what + " needs a number, got '" + v + "'");
    return x;
}

}  // namespace detail

/// @brief Parse `--party=i --peers=host:port,host:port[,host:port]` plus any
///        `run_config` keys (`--transport=mux --lanes=4 ...`).
inline node_args parse_node_args(int argc, char ** argv,
    run_config base = run_config::from_env())
{
    node_args a;
    a.cfg = base;
    bool have_party = false;
    for (const auto & arg : a.cfg.apply_args(argc, argv))
    {
        if (arg.rfind("--party=", 0) == 0)
        {
            a.party = static_cast<unsigned>(detail::parse_count("--party", arg.substr(8)));
            have_party = true;
        }
        else if (arg.rfind("--peers=", 0) == 0)
        {
            std::string list = arg.substr(8);
            std::size_t start = 0;
            while (start <= list.size())
            {
                const auto comma = list.find(',', start);
                const std::string item = list.substr(start,
                    comma == std::string::npos ? std::string::npos : comma - start);
                const auto colon = item.rfind(':');
                if (colon == std::string::npos)
                    throw std::invalid_argument("--peers entry '" + item
                        + "' needs host:port");
                const auto port = detail::parse_count("--peers port", item.substr(colon + 1));
                if (port > 65535)
                    throw std::invalid_argument("--peers entry '" + item
                        + "' has a port above 65535");
                a.peers.push_back(net::peer_address{item.substr(0, colon),
                    static_cast<unsigned short>(port)});
                if (comma == std::string::npos)
                    break;
                start = comma + 1;
            }
        }
        else
            throw std::invalid_argument("unknown argument '" + arg + "'");
    }
    if (!have_party || a.peers.size() < 2)
        throw std::invalid_argument(
            "usage: --party=i --peers=host:port,host:port[,host:port] [--key=value]");
    if (a.party >= a.peers.size())
        throw std::invalid_argument("--party is outside --peers");
    if (!net::is_socket_transport(a.cfg.kind) || a.cfg.kind == net::transport::local)
    {
        a.replaced_transport = net::transport_name(a.cfg.kind);
        a.cfg.kind = net::transport::mux;
    }
    return a;
}

/// @brief Join the static table and drive this node's party.
inline net::stream_stats run_node(const node_args & a, const protocol::plan & plan,
    party_values & values,
    const std::map<std::uint32_t, protocol::kernel_fn> & kernels = {})
{
    const std::size_t n = a.peers.size();
    const auto L = layout_for(plan, n, a.cfg);
    const log::role_scope role("p" + std::to_string(a.party));
    if (!a.replaced_transport.empty())
        DPF_LOG(warning, "config.override").kv("key", "transport")
            .kv("requested", a.replaced_transport).kv("used", "mux")
            .kv("detail", "a node needs a network transport");
    if (log::enabled(log::level::info))
    {
        std::string table;
        for (std::size_t p = 0; p < n; ++p)
            table += (p == 0 ? "" : ",") + a.peers[p].host + ":"
                + std::to_string(a.peers[p].port);
        DPF_LOG(info, "node.start").kv("party", a.party).kv("parties", n)
            .kv("transport", net::transport_name(a.cfg.kind)).kv("peers", table)
            .kv("compute_threads", a.cfg.compute_threads);
        detail::log_plan(a.party, plan, L, a.cfg);
    }
    asio::io_context io;
    auto work = asio::make_work_guard(io);
    net::session_options so;
    so.n_lanes = L.pair_lanes;
    so.kind = a.cfg.kind;
    so.policy = a.cfg.policy;
    so.limits = a.cfg.limits;
    so.security = a.cfg.security;
    net::party_session session(io, a.party, static_cast<unsigned>(n), so);
    session.join(a.peers);
    std::vector<net::async_stream_array *> links(n, nullptr);
    for (std::size_t p = 0; p < n; ++p)
        if (p != a.party)
            links[p] = &session.peer(static_cast<unsigned>(p));
    std::unique_ptr<net::io_pool> pool;
    if (a.cfg.compute_threads != 0)
        pool = std::make_unique<net::io_pool>(1, a.cfg.compute_threads);
    detail::party_meter meter;
    try
    {
        drive_party(a.party, plan, values, kernels, links, a.cfg, nullptr, pool.get());
    }
    catch (...)
    {
        DPF_LOG(error, "party.failed").kv("party", a.party)
            .kv("what", detail::error_text(std::current_exception()));
        throw;
    }
    meter.done();
    const auto wire = sum_stats(links);
    detail::log_party_done(a.party, meter, wire, a.cfg.compute_threads);
    return wire;
}

}  // namespace app
}  // namespace dpf

#endif  // LIBDPF_INCLUDE_DPF_PARTY_RUNNER_HPP__