diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index 5bdb9ac..75ad6e6 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -374,29 +374,6 @@ auto make_network(Edges&&... edges) { // 4. Construct owned fanout storage on the heap (FanoutNode has jthread — not moveable) auto fanout_storage = std::make_unique(); - // 5. Collect unique user node pointers + their display names, in edge-declaration order - std::vector user_node_ptrs; - std::vector user_node_names; - auto collect = [&](auto& e) { - using SrcT = std::decay_t; - using DstT = std::decay_t; - auto* s = static_cast(&e.src); - auto* d = static_cast(&e.dst); - if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), s) == user_node_ptrs.end()) { - auto sname = node_display_name(); - user_node_ptrs.push_back(s); - user_node_names.push_back(sname); - s->set_name(sname); - } - if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) { - auto dname = node_display_name(); - user_node_ptrs.push_back(d); - user_node_names.push_back(dname); - d->set_name(dname); - } - }; - (collect(edges), ...); - // 5. Wire all expanded SimpleEdges. // find_node: searches fanout storage then user edge pack, returns NodeT*. // Uses if constexpr in a fold so mismatched types never reach assignment. @@ -421,6 +398,38 @@ auto make_network(Edges&&... edges) { return ptr; }; + // 5. Collect user node pointers + display names in *topological* order. + // + // Topo is computed above for the cycle check and used to be discarded, + // while this vector was filled in edge-declaration order — and then named + // user_nodes_topo_ and relied upon as if it were sorted. halt() stops in + // its reverse, and shutdown() walks it forwards stopping each node and + // draining its outputs before the next, which is only a graceful drain if + // the order really is sources-first. It held for every network in the tree + // because edges happen to be declared in pipeline order, and would have + // broken silently for one that was not. + // + // Fanout nodes appear in Topo too; they are skipped here because they are + // owned separately, in fanout_storage. + std::vector user_node_ptrs; + std::vector user_node_names; + [&](tmp::TypeList) { + ([&]() { + if constexpr (!requires { NodeT::is_fanout_node; }) { + if (auto* p = find_node.template operator()()) { + auto* n = static_cast(p); + if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), n) + == user_node_ptrs.end()) { + auto nm = node_display_name(); + user_node_ptrs.push_back(n); + user_node_names.push_back(nm); + n->set_name(nm); + } + } + } + }.template operator()(), ...); + }(typename Topo::topo{}); + // Pre-pass: build fanout_id → source display name map so fanout nodes // can be named after the node feeding them (e.g. "capture_fanout"). std::map fanout_src_name; diff --git a/tests/test_static_network.cpp b/tests/test_static_network.cpp index 4b1e780..188de84 100644 --- a/tests/test_static_network.cpp +++ b/tests/test_static_network.cpp @@ -349,3 +349,86 @@ TEST_CASE("shutdown returns when a consumer has wedged", "[static_network][shutd 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* 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* 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 seen{0}, made{0}; + + OrderSource src_fn{&made}; + OrderRelay relay_fn; + OrderSink sink_fn{&seen}; + + kpn::ObjectNode, kpn::out<"v">, "order_source", 0> s(src_fn, 8); + kpn::ObjectNode, kpn::out<"w">, "order_relay", 0> r(relay_fn, 64); + kpn::ObjectNode, 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); +}