KPN/tests/test_static_network.cpp

#include <catch2/catch_test_macros.hpp>
#include <kpn/kpn.hpp>
#include <chrono>
#include <thread>
#include <atomic>

using namespace kpn;

static int increment(int x)  { return x + 1; }
static int multiply2(int x)  { return x * 2; }
static int multiply3(int x)  { return x * 3; }
static int add10(int x)      { return x + 10; }
static int negate_val(int x) { return -x; }
static int square(int x)     { return x * x; }

// ── Linear pipeline ───────────────────────────────────────────────────────────

TEST_CASE("static_network: linear pipeline produces correct result", "[static_network]") {
    auto src = make_node<increment>(5);
    auto dst = make_node<multiply2>(5);

    Channel<int> final_out(5);
    dst.set_output_channel<0>(&final_out);

    auto net = make_network(
        edge(src.output<0>(), dst.input<0>())
    );

    net.start();
    src.input_channel<0>().push(5);   // 5 → increment → 6 → multiply2 → 12
    int result = final_out.pop();
    net.stop();

    REQUIRE(result == 12);
}

TEST_CASE("static_network: three-node pipeline", "[static_network]") {
    auto a = make_node<increment>(5);
    auto b = make_node<multiply2>(5);
    auto c = make_node<add10>(5);

    Channel<int> out(5);
    c.set_output_channel<0>(&out);

    auto net = make_network(
        edge(a.output<0>(), b.input<0>()),
        edge(b.output<0>(), c.input<0>())
    );

    net.start();
    a.input_channel<0>().push(3);   // 3 → +1=4 → *2=8 → +10=18
    int result = out.pop();
    net.stop();

    REQUIRE(result == 18);
}

// ── Auto fan-out ──────────────────────────────────────────────────────────────

TEST_CASE("static_network: auto fanout delivers to both consumers", "[static_network]") {
    // Use distinct functions so each node has a distinct type in the graph
    auto src  = make_node<increment>(8);
    auto dstA = make_node<multiply2>(8);
    auto dstB = make_node<multiply3>(8);

    Channel<int> outA(8), outB(8);
    dstA.set_output_channel<0>(&outA);
    dstB.set_output_channel<0>(&outB);

    // Two edges from the same output port — FanoutNode<int,2> is auto-inserted
    auto net = make_network(
        edge(src.output<0>(), dstA.input<0>()),
        edge(src.output<0>(), dstB.input<0>())
    );

    net.start();
    src.input_channel<0>().push(3);   // 3 → +1=4 → *2=8 and *3=12

    int a = outA.pop();
    int b = outB.pop();
    net.stop();

    REQUIRE(a == 8);
    REQUIRE(b == 12);
}

TEST_CASE("static_network: auto fanout preserves ordering across multiple items", "[static_network]") {
    auto src  = make_node<increment>(16);
    auto dstA = make_node<multiply2>(16);
    auto dstB = make_node<negate_val>(16);

    Channel<int> outA(16), outB(16);
    dstA.set_output_channel<0>(&outA);
    dstB.set_output_channel<0>(&outB);

    auto net = make_network(
        edge(src.output<0>(), dstA.input<0>()),
        edge(src.output<0>(), dstB.input<0>())
    );

    net.start();
    for (int i = 0; i < 5; ++i)
        src.input_channel<0>().push(i);   // 0..4 → +1 → *2 or negate

    for (int i = 0; i < 5; ++i) {
        REQUIRE(outA.pop() == (i + 1) * 2);
        REQUIRE(outB.pop() == -(i + 1));
    }
    net.stop();
}

// ── Stop/start lifecycle ──────────────────────────────────────────────────────

TEST_CASE("static_network: stop disables input channel", "[static_network]") {
    auto src = make_node<increment>(5);
    auto dst = make_node<multiply2>(5);

    auto net = make_network(
        edge(src.output<0>(), dst.input<0>())
    );

    net.start();
    net.stop();

    // After stop, input channel disabled — push must not throw
    src.input_channel<0>().push(99);
    REQUIRE(src.input_channel<0>().size() == 0);
}

// ── Compile-time cycle detection ──────────────────────────────────────────────
// Cycles fire a static_assert in make_network(), so we can only test the
// no-cycle path at runtime.
//
// To manually verify a cycle error: add
//   auto bad = make_network(edge(a.output<0>(), b.input<0>()),
//                           edge(b.output<0>(), a.input<0>()));
// and confirm: "make_network: graph contains a directed cycle"
//
// To manually verify a duplicate-tag error: add
//   auto x = make_node<increment>(5);
//   auto y = make_node<increment>(5);   // same type as x — UniqueTag=0 for both
//   auto bad = make_network(edge(x.output<0>(), y.input<0>()));
// and confirm: "make_network: two nodes have the same (Func, UniqueTag)"

TEST_CASE("static_network: acyclic graph does not trigger static_assert", "[static_network]") {
    auto a = make_node<increment>(5);
    auto b = make_node<add10>(5);

    Channel<int> out(5);
    b.set_output_channel<0>(&out);

    auto net = make_network(edge(a.output<0>(), b.input<0>()));

    net.start();
    a.input_channel<0>().push(5);
    REQUIRE(out.pop() == 16);   // 5 → +1=6 → +10=16
    net.stop();
}

// ── Label and UniqueTag ───────────────────────────────────────────────────────

TEST_CASE("static_network: same function distinguished by UniqueTag", "[static_network]") {
    // Two nodes wrapping the same function — only possible with distinct UniqueTag
    auto a = make_node<increment, "stage1", 0>(8);
    auto b = make_node<increment, "stage2", 1>(8);  // same func, tag=1 → distinct type

    Channel<int> out(8);
    b.set_output_channel<0>(&out);

    auto net = make_network(edge(a.output<0>(), b.input<0>()));

    net.start();
    a.input_channel<0>().push(10);
    REQUIRE(out.pop() == 12);   // 10 → +1=11 → +1=12
    net.stop();
}

TEST_CASE("static_network: two independent fan-outs of the same element type are wired independently", "[static_network]") {
    // Both src_a and src_b fan out to two consumers each.
    // Without the FanoutId fix, both would produce FanoutNode<int,2> — the same
    // C++ type — and find_node would wire all four consumers to the first instance.
    auto src_a = make_node<add10,     "src_a">(8);
    auto src_b = make_node<negate_val, "src_b">(8);
    auto cA    = make_node<increment,  "cA">(8);
    auto cB    = make_node<multiply2,  "cB">(8);
    auto cC    = make_node<multiply3,  "cC">(8);
    auto cD    = make_node<square,     "cD">(8);

    Channel<int> outA(8), outB(8), outC(8), outD(8);
    cA.set_output_channel<0>(&outA);
    cB.set_output_channel<0>(&outB);
    cC.set_output_channel<0>(&outC);
    cD.set_output_channel<0>(&outD);

    auto net = make_network(
        edge(src_a.output<0>(), cA.input<0>()),   // src_a → FanoutNode<int,2,0> → cA, cB
        edge(src_a.output<0>(), cB.input<0>()),
        edge(src_b.output<0>(), cC.input<0>()),   // src_b → FanoutNode<int,2,1> → cC, cD
        edge(src_b.output<0>(), cD.input<0>())
    );

    net.start();
    src_a.input_channel<0>().push(0);   // 0 → +10=10 → {+1=11, *2=20}
    src_b.input_channel<0>().push(5);   // 5 → negate=-5 → {*3=-15, ^2=25}

    REQUIRE(outA.pop() == 11);
    REQUIRE(outB.pop() == 20);
    REQUIRE(outC.pop() == -15);
    REQUIRE(outD.pop() == 25);
    net.stop();
}

TEST_CASE("static_network: code reuse - same function at corresponding stages of parallel branches", "[static_network]") {
    // Both branches use increment and multiply2 — code reuse via distinct UniqueTag.
    // Topology: src → fanout → { increment(tag=1) → multiply2(tag=1) → outA }
    //                        → { increment(tag=2) → multiply2(tag=2) → outB }
    auto src  = make_node<add10,    "src"         >(8);
    auto incA = make_node<increment, "inc",  1>(8);
    auto mulA = make_node<multiply2, "mul",  1>(8);
    auto incB = make_node<increment, "inc",  2>(8);
    auto mulB = make_node<multiply2, "mul",  2>(8);

    Channel<int> outA(8), outB(8);
    mulA.set_output_channel<0>(&outA);
    mulB.set_output_channel<0>(&outB);

    auto net = make_network(
        edge(src.output<0>(),  incA.input<0>()),
        edge(src.output<0>(),  incB.input<0>()),
        edge(incA.output<0>(), mulA.input<0>()),
        edge(incB.output<0>(), mulB.input<0>())
    );

    net.start();
    src.input_channel<0>().push(0);   // 0 → +10=10 → both: +1=11 → *2=22

    REQUIRE(outA.pop() == 22);
    REQUIRE(outB.pop() == 22);
    net.stop();
}

TEST_CASE("static_network: label is accessible as static member", "[static_network]") {
    using MyNode = decltype(make_node<increment, "my_node">(5));
    REQUIRE(MyNode::label() == "my_node");
    REQUIRE(MyNode::unique_tag == 0);

    using TaggedNode = decltype(make_node<increment, "tagged", 42>(5));
    REQUIRE(TaggedNode::label() == "tagged");
    REQUIRE(TaggedNode::unique_tag == 42);
}

TEST_CASE("static_network: fanout with labelled same-function consumers", "[static_network]") {
    auto src  = make_node<increment, "src"         >(8);
    auto dstA = make_node<increment, "consumer_a", 1>(8);
    auto dstB = make_node<increment, "consumer_b", 2>(8);

    Channel<int> outA(8), outB(8);
    dstA.set_output_channel<0>(&outA);
    dstB.set_output_channel<0>(&outB);

    // Fan-out from src to two increment nodes — only possible because tags differ
    auto net = make_network(
        edge(src.output<0>(), dstA.input<0>()),
        edge(src.output<0>(), dstB.input<0>())
    );

    net.start();
    src.input_channel<0>().push(5);   // 5 → +1=6 → both +1=7

    REQUIRE(outA.pop() == 7);
    REQUIRE(outB.pop() == 7);
    net.stop();
}

// Regression: shutdown() must return even when a consumer stopped consuming.
//
// The drain step was an unbounded `while (anything anywhere is non-empty)` poll
// over *every* channel in the graph. Two defects in one loop: it waited for the
// whole network to be idle before stopping each successive layer rather than
// just the node it had stopped — the dynamic Network's version even took a node
// name and ignored it — and it had no deadline, so anything wedged downstream
// turned a graceful shutdown into the hang it exists to avoid.
//
// It could also fail to terminate with nothing wedged at all. current_fill came
// from a snapshot that loaded tail_ before head_; a concurrent pop between the
// two reads yields a head_ past the sampled tail_, and the unsigned difference
// wraps to ~2^64. Any poll for "is it empty yet" against that value runs
// forever. Both indices only ever increase, so loading head_ first can at worst
// under-report a push, which this loop tolerates and a wrap does not.
//
// Here the sink never takes anything, so its input cannot drain and the only
// correct outcome is to give up and say so. The bound asserted is deliberately
// loose: the point is that it terminates, not how fast.
namespace {

struct DrainSource {
    static constexpr std::string_view label() { return "drain_source"; }
    int n{0};
    int operator()() {
        std::this_thread::sleep_for(std::chrono::microseconds(100));
        return n++;
    }
};

struct NeverConsumes {
    static constexpr std::string_view label() { return "never_consumes"; }
    std::atomic<bool>* wedged;
    void operator()(int) {
        // Blocks for the duration of the test: the input channel behind it
        // fills and stays full.
        while (!wedged->load(std::memory_order_acquire))
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
};

}  // namespace

TEST_CASE("shutdown returns when a consumer has wedged", "[static_network][shutdown]") {
    std::atomic<bool> release{false};

    DrainSource   src_fn;
    NeverConsumes sink_fn{&release};

    kpn::ObjectNode<DrainSource,   kpn::in<>,    kpn::out<"v">, "drain_source",   0> s(src_fn,  4);
    kpn::ObjectNode<NeverConsumes, kpn::in<"v">, kpn::out<>,    "never_consumes", 0> k(sink_fn, 4);

    auto net = kpn::make_network(kpn::edge(s.output<"v">(), k.input<"v">()));
    net.set_drain_timeout(std::chrono::milliseconds(100));
    net.start();

    // Let the channel fill and the sink jam.
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

    // Unjam the sink well after the drain timeout should have expired. Stopping
    // a node joins its worker, so a sink blocked forever would hang the test in
    // stop() rather than in the drain loop this case is about.
    std::thread unjam([&] {
        std::this_thread::sleep_for(std::chrono::milliseconds(800));
        release.store(true, std::memory_order_release);
    });

    const auto t0 = std::chrono::steady_clock::now();
    net.shutdown();
    const auto elapsed = std::chrono::steady_clock::now() - t0;

    unjam.join();

    const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
    INFO("shutdown took " << ms << " ms");
    CHECK(ms < 3000);   // unbounded before; one 100 ms drain timeout after
}

// Regression: node order must come from the topological sort, not from the
// order the edges happened to be written in.
//
// make_network computes Topo for the cycle check and then dropped it, filling
// the node vector in edge-declaration order — and named it user_nodes_topo_.
// halt() stops in its reverse, and shutdown() walks it forwards stopping each
// node and draining its outputs before moving to the next, which is a graceful
// drain only if the order really is sources-first.
//
// Every network in this tree declares edges in pipeline order, so the two
// coincided and nothing failed. This case declares them backwards, which is
// legal and which make_network otherwise accepts silently.
//
// Asserted through shutdown() rather than by reading the order back, because
// the order is private and the ordering is not the point — what it buys is.
// A sources-first shutdown lets the values already in flight reach the sink;
// stopping the sink first strands them, and the drain step then has nobody
// left to take them.
namespace {

struct OrderSource {
    static constexpr std::string_view label() { return "order_source"; }
    std::atomic<int>* made;
    int operator()() {
        std::this_thread::sleep_for(std::chrono::microseconds(20));
        return made->fetch_add(1, std::memory_order_relaxed);
    }
};

// Deliberately slower than the source, so a deep backlog builds up in its input
// channel. That backlog is what a sources-first shutdown preserves and a
// sink-first one throws away, and it needs to be big enough that the difference
// cannot be mistaken for one value in flight.
struct OrderRelay {
    static constexpr std::string_view label() { return "order_relay"; }
    int operator()(int v) {
        std::this_thread::sleep_for(std::chrono::microseconds(300));
        return v;
    }
};

struct OrderSink {
    static constexpr std::string_view label() { return "order_sink"; }
    std::atomic<int>* seen;
    void operator()(int) { seen->fetch_add(1, std::memory_order_relaxed); }
};

}  // namespace

TEST_CASE("edges declared out of order still start and stop sources-first",
          "[static_network][shutdown]") {
    std::atomic<int> seen{0}, made{0};

    OrderSource src_fn{&made};
    OrderRelay  relay_fn;
    OrderSink   sink_fn{&seen};

    kpn::ObjectNode<OrderSource, kpn::in<>,    kpn::out<"v">, "order_source", 0> s(src_fn,   8);
    kpn::ObjectNode<OrderRelay,  kpn::in<"v">, kpn::out<"w">, "order_relay",  0> r(relay_fn, 64);
    kpn::ObjectNode<OrderSink,   kpn::in<"w">, kpn::out<>,    "order_sink",   0> k(sink_fn,  64);

    // Sink edge first, source edge last — the reverse of pipeline order.
    auto net = kpn::make_network(
        kpn::edge(r.output<"w">(), k.input<"w">()),
        kpn::edge(s.output<"v">(), r.input<"v">())
    );
    net.start();
    std::this_thread::sleep_for(std::chrono::milliseconds(300));

    const int before = seen.load(std::memory_order_relaxed);
    REQUIRE(before > 0);              // the pipeline ran at all

    net.shutdown();

    // Sources stop first and each layer drains before the next stops, so the
    // backlog queued in front of the relay still reaches the sink. Stopping in
    // declaration order stops the relay first and discards all of it.
    const int after = seen.load(std::memory_order_relaxed);
    INFO("made " << made.load() << ", delivered " << before
         << " before shutdown, " << after << " after");
    CHECK(after - before >= 20);
}