diff --git a/SPEC.md b/SPEC.md index 3fa5383..5e422ef 100644 --- a/SPEC.md +++ b/SPEC.md @@ -344,6 +344,70 @@ static_assert( Multi-output functions must return `std::tuple<...>`. Single return accepted as-is. `void` return = sink node (no output ports). +### Node identity: `Label` and `UniqueTag` NTTPs + +Two problems arise when using `Node` as a compile-time graph vertex: + +1. **Debug names** — without a label the web UI and `print_diagnostics` fall back to + `"node[0]"` placeholders. `set_name()` provides a runtime name for the runtime + `Network`, but `StaticNetwork` has no `add("name", node)` call to attach one. + +2. **Same-function collision** — two nodes wrapping the same function (e.g. two + `make_node`) have the same type. The `StaticNetwork` topo sort and fanout + detection use node types as graph vertices, so it cannot distinguish them, causing + infinite recursion in the DFS. + +Both are solved by two additional NTTPs on `Node`: + +```cpp +template< + auto Func, + typename InputTag = in<>, + typename LatchTag = latch<>, + typename OutputTag = out<>, + fixed_string Label = "", // human-readable name; shown in diagnostics/web UI + std::size_t UniqueTag = 0 // collision-breaker; must differ between any two +> // same-Func nodes in one make_network() call +class Node : public INode { ... }; +``` + +Both have defaults so all existing code (`make_node(5)`) compiles unchanged. + +**Factory syntax extension:** + +```cpp +// Label only (UniqueTag defaults to 0) +auto blur = make_node(8); + +// Both label and unique tag — required when the same function is used twice +auto preblur = make_node(8); +auto postblur = make_node(8); +``` + +**Duplicate-tag detection** — `make_network()` checks at compile time that no two nodes +in the edge pack share the same `(Func, UniqueTag)` pair. This fires a readable +`static_assert` at the call site before any thread is started: + +``` +static_assert(!has_duplicate_tags_v, + "make_network: two nodes have identical (Func, UniqueTag) — increment UniqueTag " + "on one of them to make them distinct"); +``` + +**Label availability** — `Node` exposes the label as a `static constexpr` so +`StaticNetwork` can read it at compile time for diagnostics: + +```cpp +static constexpr std::string_view label() { return Label.view(); } +``` + +The web debug UI and `print_diagnostics` use `label()` when non-empty, falling back to +`"node[]"` for unlabelled nodes. The runtime `Network` continues to use the +string passed to `add("name", node)` — `Label` is orthogonal to that mechanism. + +**`ObjectNode`** gains the same two NTTPs with the same defaults. The `make_node(obj, ...)` +overloads are extended identically. + --- ## Component 4a — Latched Input Ports @@ -1065,6 +1129,231 @@ net.start(); --- +## Component 9 — `static_network.hpp`: Compile-time Graph Builder + +### Motivation + +The runtime `Network` builder has two limitations that only a compile-time graph can fix: + +1. **Fan-out `N` is unknowable at the first `connect()` call.** A `FanoutNode` requires + `N` as a template parameter. With runtime `connect()`, the network has seen only one edge + when the first call arrives; it cannot know how many more will follow for that port. Auto- + inserting the right `FanoutNode` requires seeing the complete edge list at once — + which is only possible if the edge list is a type. + +2. **Start/stop goes through virtual dispatch.** `Network` stores `INode*` and calls virtual + `start()`/`stop()`. With a typed node tuple the compiler sees the concrete types and can + inline or devirtualise. This matters at startup/shutdown, not in the hot path — but it is + avoidable overhead. + +The runtime `Network` is **not removed**. It remains the right choice for Python graphs, +sub-networks embedded in dynamic topologies, and any case where the graph shape is not known +until runtime. `StaticNetwork` is an additional builder for the common case where the full +C++ topology is known at compile time. + +### API + +```cpp +// edge() constructs a typed edge descriptor from two port handles. +// All type information (SrcNode, SrcIdx, DstNode, DstIdx) is in the return type. +template +auto edge(OutputPort, InputPort) + -> Edge; + +// make_network() accepts all edges as a variadic pack. +// It deduces the full topology, auto-inserts FanoutNodes where needed, +// wires all channels, and returns a StaticNetwork owning the fanout nodes. +// User nodes are held by reference (non-owning), same lifetime contract as Network. +template +auto make_network(Edges&&... edges) -> StaticNetwork<...>; +``` + +Usage: + +```cpp +auto src = make_node(8); +auto blur = make_node>(8); +auto detect = make_node>(8); +auto sink = make_node(8); + +// src:output<0> feeds both blur and detect — fan-out is auto-inserted +auto net = make_network( + edge(src.output<0>(), blur.input<0>()), + edge(src.output<0>(), detect.input<0>()), // same source port + edge(blur.output<0>(), sink.input<0>()), + edge(detect.output<0>(), sink.input<1>()) +); +net.start(); +// ... +net.stop(); +``` + +No `add()`, no `build()`, no string names. The graph is fully wired in the `make_network` +call. `start()` and `stop()` are non-virtual tuple traversals. + +### Edge type + +```cpp +// Carries references to the two endpoint nodes. Stores no data beyond that. +template +struct Edge { + SrcNode& src; + DstNode& dst; +}; +``` + +### Fan-out detection metafunction + +`make_network` receives `Edge<...>` types as a pack. Before wiring, a metafunction scans +the pack for output ports with more than one downstream edge: + +``` +fanout_groups +``` + +This is a compile-time multimap: keys are `(SrcNode type, SrcIdx)`, values are the list of +destination `(DstNode type, DstIdx)` pairs sharing that key. + +For each key with N > 1 destinations: +- Compute `T = std::tuple_element_t` +- Synthesise a `FanoutNode` — call it `F` +- Replace the N original edges with: + - one edge: `src:SrcIdx → F:input<0>` + - N edges: `F:output<0..N-1> → original dst:DstIdx` + +For keys with N == 1 the edge is kept as-is. + +The result is an expanded edge list with all fan-outs made explicit, and a list of +`FanoutNode` types that need to be instantiated. + +### `StaticNetwork` structure + +```cpp +template owned by the network + typename TopoOrder> // index_sequence encoding start/stop order +class StaticNetwork : public INode { +public: + void start(); // std::apply over TopoOrder — no virtual dispatch, no map lookup + void stop(); // reverse of TopoOrder + + bool running() const; + + // Diagnostics — iterates typed tuples; same NodeSnapshot / ChannelSnapshot output + // as Network, compatible with print_diagnostics and the web debug UI. + void print_diagnostics(std::ostream& = std::cerr) const; + + // StaticNetwork is itself an INode, so it can be embedded in a runtime Network + // via net.add("stage", static_net) exactly like any other node. + void set_name(std::string) override; + const NodeStats& stats() const override; + NodeSnapshot node_snapshot(const std::string&, double) const override; + +private: + FanoutStorage fanouts_; // owns the auto-generated FanoutNode instances + // User nodes held by reference — same non-owning contract as Network +}; +``` + +`FanoutStorage` is a `std::tuple, FanoutNode, ...>` with one +element per auto-inserted fanout. It is owned by the `StaticNetwork` and lives as long as +the network does — which satisfies the channel lifetime contract (channels are owned by their +consumer, and the fanout node is the consumer of the upstream output). + +### Cycle detection + +With the full edge list as a type pack, cycle detection is a `static_assert` rather than a +runtime exception. A compile-time DFS over the expanded edge list fires a readable assertion +at the `make_network` call site: + +``` +static_assert(!has_cycle_v, + "make_network: graph contains a directed cycle"); +``` + +`NetworkCycleError` is no longer needed for `StaticNetwork` — the cycle is caught before any +object is constructed. + +### Topological order + +The same compile-time DFS produces a topological ordering as an `std::index_sequence` over +the node tuple. `start()` iterates it forward, `stop()` iterates it in reverse. No runtime +sort, no `std::vector`. + +### Node labels for diagnostics / web debug + +Labels come directly from the `Label` NTTP on each `Node` type — no separate annotation +on `edge()` is needed. `StaticNetwork` reads `NodeType::label()` at compile time for each +vertex in the topo order and stores the result as a `std::string_view` array at +construction time. Zero runtime overhead: the label is a compile-time string literal. + +```cpp +auto src = make_node(8); +auto blur = make_node(8); +auto detect = make_node(8); + +auto net = make_network( + edge(src.output<0>(), blur.input<0>()), + edge(src.output<0>(), detect.input<0>()) +); +// web UI shows nodes named "src", "blur", "detect" +// auto-inserted FanoutNode is labelled "fanout[src:0]" +``` + +Unlabelled nodes (`Label == ""`) fall back to `"node[]"` in diagnostics. +Auto-inserted fanout nodes are labelled `"fanout[:]"` automatically. + +### Wiring sequence in `make_network` + +All wiring happens in the `make_network` constructor body — no `build()` call needed: + +1. Instantiate `FanoutStorage` (default-construct each `FanoutNode`). +2. For each expanded edge (in topological order): + - Call `src.set_output_channel(&dst.input_channel())`. +3. Return the `StaticNetwork`. + +Channel pointers are set once and never changed. No dynamic allocation after construction. + +### What is eliminated vs `Network` + +| `Network` (runtime) | `StaticNetwork` (compile-time) | +|---|---| +| `std::map` | typed `std::tuple` of references | +| Runtime DFS + `NetworkCycleError` | `static_assert` at `make_network` call site | +| Virtual `start()`/`stop()` per node | `std::apply` over typed tuple | +| Explicit `make_fanout` | auto-inserted from edge pack | +| `connected_outputs_` duplicate check | structural impossibility — no duplicate edge can produce two `set_output_channel` calls | +| `build()` step | no build step — wired in constructor | + +The hot path (per-item `pop` → `push` in each node thread) is identical in both cases. + +### Compatibility + +- `StaticNetwork` implements `INode`, so it can be registered inside a runtime `Network` + via `net.add("name", static_net)` — enabling mixed static/dynamic graphs. +- All existing node types (`Node`, `ObjectNode`, `FanoutNode`, `MainThreadNode`) work + unchanged as vertices in a `StaticNetwork`. +- The Python `PyNetwork` is unaffected — it remains runtime-only. + +### File layout addition + +``` +include/kpn/ + static_network.hpp # Edge<>, make_network(), StaticNetwork<> + tmp/ + fanout_groups.hpp # fanout_groups metafunction + topo_sort.hpp # compile-time DFS + cycle check + repeat_tuple.hpp # repeat_tuple_t (moved from fanout.hpp) +``` + +`fanout.hpp` keeps `FanoutNode` and `make_fanout` for users who want to wire +fanouts explicitly in a runtime `Network`. `static_network.hpp` uses `FanoutNode` internally +but the user never calls `make_fanout` when using `make_network`. + +--- + ## Resolved Design Decisions All major design questions are now closed: @@ -1079,3 +1368,6 @@ All major design questions are now closed: | `make_py_network` | Pure C++ template; nanobind module recompilation is the registration step | | GIL strategy | Acquire per Python callback; release while blocking on channel ops | | Mixed-rate inputs | `latch<>` tag for ports that reuse last-seen value; blocks only on first fire; node fires at rate of `in<>` ports | +| Fan-out | Explicit `FanoutNode` for runtime `Network`; auto-inserted by `make_network()` for `StaticNetwork` | +| Static vs runtime graph | Both coexist; `StaticNetwork` for C++ graphs known at compile time, `Network` for Python/dynamic graphs; `StaticNetwork` implements `INode` so it embeds in `Network` | +| Node identity in static graphs | `Label` NTTP (human name for diagnostics) + `UniqueTag` NTTP (collision-breaker for same-Func nodes); both default to `""` / `0` so existing code is unaffected | diff --git a/examples/10_static_hello_pipeline/main.cpp b/examples/10_static_hello_pipeline/main.cpp new file mode 100644 index 0000000..e8f19df --- /dev/null +++ b/examples/10_static_hello_pipeline/main.cpp @@ -0,0 +1,34 @@ +// Example 10 — Static Hello Pipeline +// +// The same linear pipeline as example 01, built with make_network() instead +// of the runtime Network builder. The topology is fully known at compile time: +// cycle detection is a static_assert, no build() step is needed, and start/stop +// require no virtual dispatch through a string-keyed node map. +// +// [produce] --int--> [double_it] --int--> [print_it] + +#include +#include +#include +#include + +static int produce() { return 42; } +static int double_it(int x) { return x * 2; } +static void print_it(int x) { std::cout << "result: " << x << '\n'; } + +int main() { + using namespace kpn; + + auto src = make_node(5); + auto dbl = make_node(5); + auto sink = make_node(5); + + auto net = make_network( + edge(src.output<0>(), dbl.input<0>()), + edge(dbl.output<0>(), sink.input<0>()) + ); + + net.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + net.stop(); +} diff --git a/examples/11_static_fanout/main.cpp b/examples/11_static_fanout/main.cpp new file mode 100644 index 0000000..234a5bd --- /dev/null +++ b/examples/11_static_fanout/main.cpp @@ -0,0 +1,55 @@ +// Example 11 — Static Fan-Out (automatic FanoutNode insertion) +// +// Two consumers read from the same output port of [generate]. With the runtime +// Network builder this would require an explicit make_fanout<>. With make_network() +// the duplicate source port is detected at compile time and a FanoutNode +// is inserted automatically — the user just writes two edges from the same port. +// +// +--> [print_key] +// [generate] --string--> [fan] +// +--> [print_upper] +// +// The FanoutNode is owned by the StaticNetwork and invisible to the user. + +#include +#include +#include +#include +#include +#include + +static int gen_index = 0; + +static std::string generate() { + static const char* words[] = {"hello", "kpn", "fanout", "static", "network"}; + std::this_thread::sleep_for(std::chrono::milliseconds(80)); + return words[gen_index++ % 5]; +} + +static void print_lower(std::string s) { + std::cout << "lower: " << s << '\n'; +} + +static void print_upper(std::string s) { + std::transform(s.begin(), s.end(), s.begin(), ::toupper); + std::cout << "upper: " << s << '\n'; +} + +int main() { + using namespace kpn; + + auto gen = make_node(5); + auto lower = make_node(5); + auto upper = make_node(5); + + // Two edges from gen.output<0>() — make_network() detects the fan-out + // and inserts FanoutNode automatically. + auto net = make_network( + edge(gen.output<0>(), lower.input<0>()), + edge(gen.output<0>(), upper.input<0>()) + ); + + net.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + net.stop(); +} diff --git a/examples/12_static_cellshade/main.cpp b/examples/12_static_cellshade/main.cpp new file mode 100644 index 0000000..0357db0 --- /dev/null +++ b/examples/12_static_cellshade/main.cpp @@ -0,0 +1,217 @@ +// Example 12 — Static Cell-Shading Pipeline with Auto Fan-Out +// +// The same cell-shading effect as example 09, rebuilt with make_network(). +// +// Key differences from example 09: +// +// 1. Fan-out is automatic. The edge detector output feeds both the compositing +// node and the debug display window. In example 09 this required composite() +// to re-output the edge mask as a second return value (a workaround). Here +// make_network() detects the duplicate source port and inserts +// FanoutNode automatically — composite() is a clean single-output +// node. +// +// 2. No add()/connect()/build() ceremony. The full topology is expressed once +// in the make_network() call. Cycle detection and duplicate-tag checking are +// compile-time static_asserts. +// +// 3. Every node has a Label NTTP so the web debug UI shows real names. +// +// Topology: +// +// [capture] --colour--> [quant] ──────────────────────────────> [comp] --> [display_composite] +// [capture] --grey----> [to_gray] --> [edges] --edges--(fan)--> [comp] +// --edges----------> [display_edges] ← auto-fanout +// +// Note: capture returns std::tuple (colour, grey). +// The two outputs are separate ports routed independently. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// ── Gradient base for synthetic pattern ─────────────────────────────────────── + +static cv::Mat make_gradient(int W, int H) { + cv::Mat xr(H, W, CV_8UC1), yg(H, W, CV_8UC1), b(H, W, CV_8UC1, cv::Scalar(128)); + for (int x = 0; x < W; ++x) xr.col(x).setTo(x * 255 / W); + for (int y = 0; y < H; ++y) yg.row(y).setTo(y * 255 / H); + cv::Mat channels[3] = {b, yg, xr}; + cv::Mat grad; + cv::merge(channels, 3, grad); + return grad; +} + +// ── Pipeline functions ──────────────────────────────────────────────────────── + +static std::tuple capture() { + constexpr int W = 640, H = 480; + static cv::VideoCapture cap; + static bool opened = false; + if (!opened) { + opened = true; + cap.open(0, cv::CAP_V4L2); + if (cap.isOpened()) { + cap.set(cv::CAP_PROP_FRAME_WIDTH, W); + cap.set(cv::CAP_PROP_FRAME_HEIGHT, H); + } else { + std::cerr << "[capture] no webcam — using synthetic animated pattern\n"; + } + } + + cv::Mat frame; + if (cap.isOpened()) { + auto t0 = std::chrono::steady_clock::now(); + cap >> frame; + auto elapsed = std::chrono::steady_clock::now() - t0; + if (elapsed < std::chrono::milliseconds(20)) + std::this_thread::sleep_for(std::chrono::milliseconds(33) - elapsed); + if (frame.empty()) frame = cv::Mat::zeros(H, W, CV_8UC3); + } else { + static int tick = 0; + static cv::Mat grad = make_gradient(W, H); + ++tick; + frame = grad.clone(); + int r = 150 + (tick % 80) * 4; + cv::circle(frame, {W/2, H/2}, r, {255, 200, 0}, -1); + cv::circle(frame, {W/2, H/2}, r / 2, { 0, 128, 255}, -1); + cv::circle(frame, {W*2/5, H*2/5}, r / 3, {200, 0, 200}, -1); + std::this_thread::sleep_for(std::chrono::milliseconds(33)); + } + return {frame.clone(), frame.clone()}; +} + +static cv::Mat to_gray(cv::Mat bgr) { + cv::Mat gray; + cv::cvtColor(bgr, gray, cv::COLOR_BGR2GRAY); + return gray; +} + +static cv::Mat edges_fn(cv::Mat gray) { + cv::Mat blurred, mask; + cv::GaussianBlur(gray, blurred, {5, 5}, 0); + cv::Canny(blurred, mask, 50, 150); + return mask; +} + +static cv::Mat quantise(cv::Mat bgr) { + constexpr int levels = 4; + constexpr double step = 256.0 / levels; + static const cv::Mat lut = []() { + cv::Mat l(1, 256, CV_8UC1); + for (int i = 0; i < 256; ++i) + l.at(i) = cv::saturate_cast( + std::floor(i / step) * step + step / 2.0); + return l; + }(); + cv::Mat out; + cv::LUT(bgr, lut, out); + return out; +} + +// Clean single-output composite — no longer needs to pass edges through. +static cv::Mat composite(cv::Mat edge_mask, cv::Mat colour) { + cv::Mat result = colour.clone(); + result.setTo(cv::Scalar(0, 0, 0), edge_mask); + return result; +} + +// ── Display nodes ───────────────────────────────────────────────────────────── +// +// Two separate MainThreadNode subclasses — one for the composited result, +// one for the raw edge mask. Each runs on the main thread via step(). +// The fan-out from [edges] to both consumers is inserted automatically by +// make_network(). + +class DisplayComposite : public kpn::MainThreadNode, + cv::Mat> { +public: + // Label and unique_tag for StaticNetwork identity + static constexpr std::string_view label() { return "display_composite"; } + static constexpr std::size_t unique_tag = 0; + + DisplayComposite() : MainThreadNode(8) { + cv::namedWindow("Cell Shade", cv::WINDOW_NORMAL); + cv::resizeWindow("Cell Shade", 1280, 720); + } + ~DisplayComposite() { cv::destroyWindow("Cell Shade"); } + + bool operator()(cv::Mat frame) { + cv::imshow("Cell Shade", frame); + int key = cv::waitKey(1); + if (key == 'q' || key == 27) return false; + try { return cv::getWindowProperty("Cell Shade", cv::WND_PROP_VISIBLE) >= 1; } + catch (const cv::Exception&) { return false; } + } +}; + +class DisplayEdges : public kpn::MainThreadNode, + cv::Mat> { +public: + static constexpr std::string_view label() { return "display_edges"; } + static constexpr std::size_t unique_tag = 1; + + DisplayEdges() : MainThreadNode(8) { + cv::namedWindow("Edge Mask", cv::WINDOW_NORMAL); + cv::resizeWindow("Edge Mask", 640, 360); + } + ~DisplayEdges() { cv::destroyWindow("Edge Mask"); } + + bool operator()(cv::Mat mask) { + cv::Mat bgr; + cv::cvtColor(mask, bgr, cv::COLOR_GRAY2BGR); + cv::imshow("Edge Mask", bgr); + cv::waitKey(1); + try { return cv::getWindowProperty("Edge Mask", cv::WND_PROP_VISIBLE) >= 1; } + catch (const cv::Exception&) { return false; } + } +}; + +// ───────────────────────────────────────────────────────────────────────────── + +int main() { + using namespace kpn; + + // Nodes — all labelled for web debug UI + auto src = make_node(8); + auto gray_node = make_node(8); + auto edge_node = make_node(8); + auto quant = make_node(8); + auto comp = make_node(8); + + // DisplayNodes live on the main thread — registered as sinks + DisplayComposite disp_comp; + DisplayEdges disp_edges; + + // make_network() detects that edge_node.output<0>() feeds two consumers + // (comp and disp_edges) and inserts FanoutNode automatically. + auto net = make_network( + edge(src.output<0>(), quant.input<0>()), // colour → quant + edge(src.output<1>(), gray_node.input<0>()), // grey → to_gray + edge(gray_node.output<0>(), edge_node.input<0>()), // gray → edges + edge(edge_node.output<0>(), comp.input<0>()), // edges → comp (fan-out src) + edge(edge_node.output<0>(), disp_edges.input<0>()), // edges → display_edges (auto fanout) + edge(quant.output<0>(), comp.input<1>()), // quantised → comp + edge(comp.output<0>(), disp_comp.input<0>()) // result → display_composite + ); + + std::cout << "Cell-shading pipeline (static) running. Press 'q' to stop.\n"; + + net.start(); + + // Main thread drives both display nodes — step() returns false when + // operator() returns false (q pressed or window closed). + while (disp_comp.step() && disp_edges.step()) + cv::waitKey(8); + + net.stop(); + return 0; +} diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index e3e7e9e..25641b0 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -11,6 +11,8 @@ kpn_example(03_multi_output) kpn_example(04_storage_policy) kpn_example(05_error_handling) kpn_example(06_watchdog) +kpn_example(10_static_hello_pipeline) +kpn_example(11_static_fanout) if(KPN_WEB_DEBUG) kpn_target_enable_web_debug(06_watchdog) endif() @@ -24,6 +26,9 @@ find_package(OpenCV QUIET COMPONENTS core imgproc highgui videoio) if(OpenCV_FOUND) add_executable(09_opencv_cellshade 09_opencv_cellshade/main.cpp) target_link_libraries(09_opencv_cellshade PRIVATE kpn ${OpenCV_LIBS}) + + add_executable(12_static_cellshade 12_static_cellshade/main.cpp) + target_link_libraries(12_static_cellshade PRIVATE kpn ${OpenCV_LIBS}) if(KPN_WEB_DEBUG) kpn_target_enable_web_debug(09_opencv_cellshade) endif() diff --git a/include/kpn/fanout.hpp b/include/kpn/fanout.hpp new file mode 100644 index 0000000..25a655b --- /dev/null +++ b/include/kpn/fanout.hpp @@ -0,0 +1,175 @@ +#pragma once +#include "channel.hpp" +#include "diagnostics.hpp" +#include "node.hpp" +#include "port.hpp" +#include "traits.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace kpn { + +namespace detail { + +// Produces std::tuple with N repetitions — used so that +// Network::connect can do its normal return_tuple type-check against FanoutNode. +template> +struct repeat_tuple; + +template +struct repeat_tuple> { + template using always_T = T; + using type = std::tuple...>; +}; + +template +using repeat_tuple_t = typename repeat_tuple::type; + +} // namespace detail + +// ── FanoutNode ──────────────────────────────────────────────────────────────── +// +// Reads one item from its single input channel and pushes a copy to each of +// N output channels. All N downstream nodes receive every item. +// +// Usage: +// auto fan = make_fanout(/*capacity=*/8); +// net.connect("src", src.output<0>(), "fan", fan.input<0>()) +// .connect("fan", fan.output<0>(), "nodeA", nodeA.input<0>()) +// .connect("fan", fan.output<1>(), "nodeB", nodeB.input<0>()) + +template +class FanoutNode : public INode { +public: + using args_tuple = std::tuple; + using return_tuple = detail::repeat_tuple_t; + using return_raw = return_tuple; + + static constexpr std::size_t input_count = 1; + static constexpr std::size_t output_count = N; + + explicit FanoutNode(std::size_t fifo_capacity = 5) + : fifo_capacity_(fifo_capacity) + { + input_ch_ = std::make_shared>(fifo_capacity); + } + + ~FanoutNode() override { stop(); } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { + input_ch_->enable(); + stop_flag_.store(false, std::memory_order_relaxed); + thread_ = std::jthread([this](std::stop_token) { run_loop(); }); + } + + void stop() override { + stop_flag_.store(true, std::memory_order_relaxed); + input_ch_->disable(); + if (thread_.joinable()) thread_.request_stop(), thread_.join(); + } + + bool running() const override { + return thread_.joinable() && !stop_flag_.load(std::memory_order_relaxed); + } + + void set_name(std::string name) override { name_ = std::move(name); } + + const NodeStats& stats() const override { return stats_; } + + NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { + uint64_t frames = stats_.frames_processed.load(std::memory_order_relaxed); + double exec_ms = stats_.ema_exec_us.load(std::memory_order_relaxed) / 1000.0; + double blocked_ms = stats_.total_blocked_us.load(std::memory_order_relaxed) / 1000.0; + double total_ms = exec_ms + blocked_ms; + return {name, frames, exec_ms, + stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0, + blocked_ms, + elapsed_s > 0 ? frames / elapsed_s : 0.0, + stats_.total_cpu_us.load(std::memory_order_relaxed) / 1000.0, + total_ms > 0 ? 100.0 * exec_ms / total_ms : 0.0}; + } + + // ── Port access ─────────────────────────────────────────────────────────── + + template + InputPort input() { + static_assert(I == 0, "FanoutNode has exactly one input"); + return {*this}; + } + + template + OutputPort output() { + static_assert(I < N, "FanoutNode output index out of range"); + return {*this}; + } + + // ── Internal channel accessors (called by Network::connect) ─────────────── + + template + Channel& input_channel() { + static_assert(I == 0); + return *input_ch_; + } + + template + void set_input_channel(std::shared_ptr> ch) { + static_assert(I == 0); + input_ch_ = std::move(ch); + } + + template + void set_output_channel(Channel* ch) { + static_assert(I < N); + out_channels_[I] = ch; + } + +private: + void run_loop() { + while (!stop_flag_.load(std::memory_order_relaxed)) { + try { + auto t0 = clock_t::now(); + T val = input_ch_->pop(); + auto t1 = clock_t::now(); + auto cpu0 = NodeStats::cpu_now(); + + for (std::size_t i = 0; i < N; ++i) { + if (out_channels_[i]) + out_channels_[i]->push(val); + } + + auto cpu1 = NodeStats::cpu_now(); + auto t2 = clock_t::now(); + stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); + } catch (const ChannelClosedError&) { + break; + } catch (const ChannelOverflowError& e) { + std::cerr << "[kpn] fanout overflow: " << e.what() << "\n"; + } + } + } + + std::string name_; + std::size_t fifo_capacity_; + std::shared_ptr> input_ch_; + std::array*, N> out_channels_{}; + std::atomic stop_flag_{false}; + std::jthread thread_; + NodeStats stats_; +}; + +// ── Factory ─────────────────────────────────────────────────────────────────── + +template +FanoutNode make_fanout(std::size_t fifo_capacity = 5) { + return FanoutNode(fifo_capacity); +} + +} // namespace kpn diff --git a/include/kpn/kpn.hpp b/include/kpn/kpn.hpp index 6ac3309..66a2376 100644 --- a/include/kpn/kpn.hpp +++ b/include/kpn/kpn.hpp @@ -6,5 +6,7 @@ #include "channel.hpp" #include "port.hpp" #include "node.hpp" +#include "fanout.hpp" +#include "static_network.hpp" #include "main_thread_node.hpp" #include "network.hpp" diff --git a/include/kpn/network.hpp b/include/kpn/network.hpp index 252dabb..7e563ef 100644 --- a/include/kpn/network.hpp +++ b/include/kpn/network.hpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -87,6 +88,13 @@ public: if (!src) throw NetworkBuildError("node '" + src_name + "' type mismatch"); if (!dst) throw NetworkBuildError("node '" + dst_name + "' type mismatch"); + auto port_key = std::make_pair(src_name, SrcIdx); + if (connected_outputs_.count(port_key)) + throw NetworkBuildError( + "connect: output port '" + src_name + "':" + std::to_string(SrcIdx) + + " is already connected — use make_fanout for fan-out"); + connected_outputs_.insert(port_key); + auto& in_ch = dst->template input_channel(); src->template set_output_channel(&in_ch); @@ -321,6 +329,7 @@ private: std::vector topo_; std::map exposed_inputs_; std::map exposed_outputs_; + std::set> connected_outputs_; std::vector> channel_probes_; ErrorHandler error_handler_; DiagnosticsHandler diag_handler_; diff --git a/include/kpn/node.hpp b/include/kpn/node.hpp index 5213bf6..4776375 100644 --- a/include/kpn/node.hpp +++ b/include/kpn/node.hpp @@ -37,18 +37,27 @@ struct INode { // InputNames — optional kpn::in<"a","b"> tag type (at most one) // OutputNames — optional kpn::out<"x","y"> tag type (at most one) -template, typename OutputTag = out<>> +template, + typename OutputTag = out<>, + fixed_string Label = "", + std::size_t UniqueTag = 0> class Node; // Specialisation that unpacks the in<>/out<> tag packs -template -class Node, out> : public INode { +template +class Node, out, Label, UniqueTag> : public INode { public: using F = decltype(Func); using args_tuple = args_t; using return_raw = return_t; using return_tuple = normalised_return_t; + // Identity accessors — used by StaticNetwork for diagnostics and type-level uniqueness + static constexpr std::string_view label() { return Label.view(); } + static constexpr std::size_t unique_tag = UniqueTag; + static constexpr std::size_t input_count = arity_v; static constexpr std::size_t output_count = std::tuple_size_v; @@ -289,17 +298,25 @@ private: // MyFunctor obj(...); // auto node = make_node(obj, in<"x">{}, out<"y">{}, capacity); -template, typename OutputTag = out<>> +template, + typename OutputTag = out<>, + fixed_string Label = "", + std::size_t UniqueTag = 0> class ObjectNode; -template -class ObjectNode, out> : public INode { +template +class ObjectNode, out, Label, UniqueTag> : public INode { public: using F = decltype(&Obj::operator()); using args_tuple = args_t; using return_raw = return_t; using return_tuple = normalised_return_t; + static constexpr std::string_view label() { return Label.view(); } + static constexpr std::size_t unique_tag = UniqueTag; + static constexpr std::size_t input_count = arity_v; static constexpr std::size_t output_count = std::tuple_size_v; @@ -534,32 +551,36 @@ auto make_node(Obj& obj, in, out, std::size_t fifo_capa // // Usage: // make_node(capacity) +// make_node(capacity) +// make_node(capacity) // UniqueTag=1 // make_node(in<"a","b">{}, capacity) -// make_node(out<"x">{}, capacity) -// make_node(in<"a","b">{}, out<"x">{}, capacity) +// make_node(in<"a","b">{}, out<"x">{}, capacity) -// No names -template +// No port names +template auto make_node(std::size_t fifo_capacity = 5) { - return Node, out<>>(fifo_capacity); + return Node, out<>, Label, UniqueTag>(fifo_capacity); } // in<> only -template +template auto make_node(in, std::size_t fifo_capacity = 5) { - return Node, out<>>(fifo_capacity); + return Node, out<>, Label, UniqueTag>(fifo_capacity); } -// out<> only (no input names) -template +// out<> only +template auto make_node(out, std::size_t fifo_capacity = 5) { - return Node, out>(fifo_capacity); + return Node, out, Label, UniqueTag>(fifo_capacity); } // in<> and out<> -template +template auto make_node(in, out, std::size_t fifo_capacity = 5) { - return Node, out>(fifo_capacity); + return Node, out, Label, UniqueTag>(fifo_capacity); } } // namespace kpn diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp new file mode 100644 index 0000000..4db1c13 --- /dev/null +++ b/include/kpn/static_network.hpp @@ -0,0 +1,237 @@ +#pragma once +#include "channel.hpp" +#include "diagnostics.hpp" +#include "fanout.hpp" +#include "node.hpp" +#include "port.hpp" +#include "tmp/fanout_groups.hpp" +#include "tmp/topo_sort.hpp" + +#include +#include +#include +#include +#include + +namespace kpn { + +// ── Edge descriptor ─────────────────────────────────────────────────────────── +// +// Carries compile-time type info and runtime references to the two endpoints. +// Constructed by the edge() factory; consumed by make_network(). + +template +struct Edge { + using src_node_t = SrcNode; + using dst_node_t = DstNode; + static constexpr std::size_t src_idx = SrcIdx; + static constexpr std::size_t dst_idx = DstIdx; + + SrcNode& src; + DstNode& dst; +}; + +template +auto edge(OutputPort, InputPort in) + -> Edge; // deduction only; defined below + +template +Edge +edge(OutputPort out, InputPort in) { + return {out.node, in.node}; +} + +// ── StaticNetwork ───────────────────────────────────────────────────────────── + +// node_label: returns Label NTTP as string_view if present, else empty. +template +struct node_label_helper { + static constexpr std::string_view value = ""; +}; +template +struct node_label_helper> { + static constexpr std::string_view value = NodeT::label(); +}; + +template +inline constexpr std::string_view node_label_v = node_label_helper::value; + +// node_display_name: Label if non-empty, else "node[UniqueTag]" +// Returned as std::string at runtime (called once at StaticNetwork construction). +template +std::string node_display_name() { + constexpr std::string_view lbl = node_label_v; + if constexpr (!lbl.empty()) { + return std::string(lbl); + } else if constexpr (requires { NodeT::unique_tag; }) { + return "node[" + std::to_string(NodeT::unique_tag) + "]"; + } else { + return "node[?]"; + } +} + +template +class StaticNetwork : public INode { +public: + // FanoutStorage = std::tuple, ...> (owned, heap-allocated) + // TopoNodeList = tmp::TypeList sources-first + + StaticNetwork(std::unique_ptr fanouts, + std::vector user_nodes_topo, + std::vector fanout_ptrs, + std::vector user_node_names) + : fanouts_(std::move(fanouts)) + , user_nodes_topo_(std::move(user_nodes_topo)) + , fanout_nodes_ptr_(std::move(fanout_ptrs)) + , user_node_names_(std::move(user_node_names)) + {} + + ~StaticNetwork() override { stop(); } + + void start() override { + stop_flag_ = false; + for (auto* n : user_nodes_topo_) n->start(); + for (auto* n : fanout_nodes_ptr_) n->start(); + } + + void stop() override { + stop_flag_ = true; + for (auto it = fanout_nodes_ptr_.rbegin(); it != fanout_nodes_ptr_.rend(); ++it) + (*it)->stop(); + for (auto it = user_nodes_topo_.rbegin(); it != user_nodes_topo_.rend(); ++it) + (*it)->stop(); + } + + bool running() const override { return !stop_flag_; } + + void set_name(std::string name) override { name_ = std::move(name); } + const NodeStats& stats() const override { static NodeStats dummy; return dummy; } + NodeSnapshot node_snapshot(const std::string& n, double) const override { + return {n, 0, 0, 0, 0, 0, 0, 0}; + } + + // Print diagnostics using compile-time node labels + void print_diagnostics(std::ostream& os = std::cerr) const { + os << "\n┌─ KPN++ StaticNetwork diagnostics ─────────────────────────────\n"; + for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) { + auto snap = user_nodes_topo_[i]->node_snapshot(user_node_names_[i], 0.0); + os << "│ " << snap.name + << " frames=" << snap.frames_processed + << " ema=" << snap.ema_exec_ms << "ms\n"; + } + os << "└────────────────────────────────────────────────────────────────\n"; + } + + FanoutStorage& fanouts_storage() { return *fanouts_; } + +private: + std::string name_; + bool stop_flag_{false}; + std::unique_ptr fanouts_; + std::vector user_nodes_topo_; + std::vector fanout_nodes_ptr_; + std::vector user_node_names_; // parallel to user_nodes_topo_ +}; + +// ── make_network ────────────────────────────────────────────────────────────── + +template +auto make_network(Edges&&... edges) { + // 1. Expand edges — detect fan-outs, splice FanoutNodes + using FanoutSto = tmp::fanout_storage_t...>; + using ExpandedEdges = tmp::expanded_edges_t...>; + + // 2. Duplicate-tag check — fires before cycle check for a cleaner error message + using UserNodes = typename tmp::all_node_types...>>::type; + static_assert(!tmp::has_duplicate_tags_v...>, + "make_network: two nodes have the same (Func, UniqueTag) — they are " + "indistinguishable as graph vertices. Add a UniqueTag: " + "make_node(capacity)"); + + // 3. Cycle check + using Topo = tmp::topo_sort; + constexpr bool has_cycle = Topo::has_cycle; + static_assert(!has_cycle, + "make_network: graph contains a directed cycle"); + + // 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()) { + user_node_ptrs.push_back(s); + user_node_names.push_back(node_display_name()); + } + if (std::find(user_node_ptrs.begin(), user_node_ptrs.end(), d) == user_node_ptrs.end()) { + user_node_ptrs.push_back(d); + user_node_names.push_back(node_display_name()); + } + }; + (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. + auto find_node = [&]() -> NodeT* { + NodeT* ptr = nullptr; + std::apply([&](auto&... fn) { + ([&](auto& node) { + if constexpr (std::is_same_v, NodeT>) + if (!ptr) ptr = &node; + }(fn), ...); + }, *fanout_storage); + if (!ptr) { + ([&](auto& e) { + if (!ptr) { + if constexpr (std::is_same_v, NodeT>) + ptr = &e.src; + else if constexpr (std::is_same_v, NodeT>) + ptr = &e.dst; + } + }(edges), ...); + } + return ptr; + }; + + auto wire_one = [&](SE) { + using SrcNode = typename SE::src_node_t; + using DstNode = typename SE::dst_node_t; + constexpr std::size_t SrcIdx = SE::src_idx; + constexpr std::size_t DstIdx = SE::dst_idx; + + auto* src = find_node.template operator()(); + auto* dst = find_node.template operator()(); + if (src && dst) + src->template set_output_channel( + &dst->template input_channel()); + }; + + [&](tmp::TypeList) { + (wire_one(SEs{}), ...); + }(ExpandedEdges{}); + + // 6. Collect fanout node pointers + std::vector fanout_ptrs; + std::apply([&](auto&... fn) { + (fanout_ptrs.push_back(static_cast(&fn)), ...); + }, *fanout_storage); + + // 7. Construct and return the StaticNetwork + using Net = StaticNetwork; + return Net(std::move(fanout_storage), + std::move(user_node_ptrs), + std::move(fanout_ptrs), + std::move(user_node_names)); +} + +} // namespace kpn diff --git a/include/kpn/tmp/fanout_groups.hpp b/include/kpn/tmp/fanout_groups.hpp new file mode 100644 index 0000000..f045aa9 --- /dev/null +++ b/include/kpn/tmp/fanout_groups.hpp @@ -0,0 +1,358 @@ +#pragma once +#include +#include +#include +#include + +// Metafunctions that scan a pack of Edge<> types, group edges by source port, +// auto-insert FanoutNode where N>1, and produce an expanded edge list +// plus a tuple type for the owned fanout nodes. +// +// Key types produced: +// expanded_edges_t — type list of SimpleEdge after fanout insertion +// fanout_storage_t — std::tuple, ...> to own + +namespace kpn::tmp { + +// ── Type list ───────────────────────────────────────────────────────────────── + +template +struct TypeList {}; + +template +struct append; +template +struct append, T> { using type = TypeList; }; +template +using append_t = typename append::type; + +template +struct concat; +template +struct concat, TypeList> { using type = TypeList; }; +template +using concat_t = typename concat::type; + +// ── Source port identity (used as compile-time map key) ─────────────────────── + +// Two edges share a source port when their SrcNode type AND SrcIdx are identical. +template +struct SrcPort {}; + +template +using src_port_of = SrcPort; + +// ── Count occurrences of a SrcPort in an edge list ─────────────────────────── + +template +struct count_port; + +template +struct count_port> : std::integral_constant {}; + +template +struct count_port> + : std::integral_constant> ? 1 : 0) + + count_port>::value> {}; + +// ── Collect all destination (DstNode&, DstIdx) for a given SrcPort ──────────── + +// A destination descriptor — just type tags, no references (references go in +// the runtime SimpleEdge structs produced after wiring). +template +struct DstDesc {}; + +template +struct collect_dsts; + +template +struct collect_dsts> { using type = TypeList<>; }; + +template +struct collect_dsts> { + using rest = typename collect_dsts>::type; + using type = std::conditional_t< + std::is_same_v>, + append_t>, + rest>; +}; + +// ── SimpleEdge: a resolved edge after fanout expansion ──────────────────────── +// +// At runtime, StaticNetwork wires edges using SimpleEdge descriptors. +// Each SimpleEdge is just a pair of (SrcNode&, SrcIdx, DstNode&, DstIdx) stored +// as a type — the actual references come from the node tuple at wire time. + +template +struct SimpleEdge { + using src_node_t = SrcNode; + using dst_node_t = DstNode; + static constexpr std::size_t src_idx = SrcIdx; + static constexpr std::size_t dst_idx = DstIdx; +}; + +// ── FanoutPlaceholder: a fanout node that will be owned by StaticNetwork ─────── + +template +struct FanoutPlaceholder { + using value_type = T; + static constexpr std::size_t fan_n = N; + static constexpr std::size_t fan_id = FanoutId; +}; + +// ── expand_edge: for one original edge, produce the replacement SimpleEdge(s) ── +// +// If the source port has N>1 consumers: the edge from src→fanout and the +// fanout→dst edges are synthesised elsewhere (see expand_all). Here we only +// need to emit the fanout→dst edge for this particular destination. +// +// For N==1 edges we emit the edge unchanged. +// +// This is called after the fanout node type has already been determined. + +} // namespace kpn::tmp + +namespace kpn { +// Forward declaration — FanoutNode is defined in fanout.hpp. +template class FanoutNode; +} // namespace kpn + +namespace kpn::tmp { + +// ── Master expansion: iterate edges, build expanded list + storage tuple ─────── +// +// Strategy: +// 1. First pass: for each unique SrcPort with N>1, record a FanoutPlaceholder. +// 2. Second pass: rewrite each edge. +// - N==1 edges become a single SimpleEdge unchanged. +// - N>1 edges: on first encounter emit src→fanout SimpleEdge + N fanout→dst +// SimpleEdges; on subsequent encounters for the same src port emit nothing +// (already handled). +// +// To implement "first encounter" tracking we carry a list of already-processed +// SrcPorts through the fold. + +template> + typename EdgeList, // TypeList> — accumulated output + typename RemainingEdges> // TypeList still to process +struct expand_impl; + +// Base case — no more edges +template +struct expand_impl> { + using fanouts = FanoutList; + using edges = EdgeList; +}; + +// Helper: is Port in ProcessedPorts? +template +struct already_processed : std::false_type {}; +template +struct already_processed> + : std::conditional_t, + std::true_type, + already_processed>> {}; + +// Helper: given DstDesc list + FanoutPlaceholder id, produce SimpleEdge list +// FanoutNode::output → DstNode::input +template +struct fanout_to_dst_edges; + +template +struct fanout_to_dst_edges, I> { + using type = TypeList<>; +}; + +template +struct fanout_to_dst_edges, DstTail...>, I> { + using head_edge = SimpleEdge, I, DstNodeT, DstI>; + using rest = typename fanout_to_dst_edges, I+1>::type; + using type = append_t; +}; + +// Recursive case — process head edge +template +struct expand_impl> { + + using AllEdges = TypeList; + using Port = src_port_of; + using SrcNode = typename Head::src_node_t; + using T = std::tuple_element_t; + static constexpr std::size_t N = count_port::value + + count_port::value + // recount against full original list approximation: + // simpler: recount in remaining + already done + ; + + // Recount properly against the complete original edge list is not possible here + // without passing it along. Instead we pre-compute N before entering the fold. + // See expand_all below which pre-computes per-port counts. + // + // This struct is not used directly — expand_all drives the logic with pre-computed N. +}; + +// ── expand_all: top-level entry point ───────────────────────────────────────── +// +// Pre-computes per-source-port counts, then runs a fold that processes edges +// one by one. + +// Port count map entry +template +struct PortCount {}; + +// Build port count list from full edge list +template +struct build_port_counts; + +template +struct build_port_counts> { + using type = TypeList<>; +}; + +template +struct build_port_counts> { + static constexpr std::size_t cnt = count_port::value; + using rest = typename build_port_counts>::type; + using type = append_t>; +}; + +// Collect unique source ports from edge list +template> +struct unique_src_ports; + +template +struct unique_src_ports, SeenSoFar> { using type = SeenSoFar; }; + +template +struct unique_src_ports, SeenSoFar> { + using Port = src_port_of; + using next_seen = std::conditional_t< + already_processed::value, + SeenSoFar, + append_t>; + using type = typename unique_src_ports, next_seen>::type; +}; + +// Look up count for a port +template +struct lookup_count : std::integral_constant {}; +template +struct lookup_count, Rest...>> + : std::integral_constant {}; +template +struct lookup_count> + : lookup_count> {}; + +// Fold state for the wiring pass +template +struct FoldState { + using processed = ProcessedPorts; + static constexpr std::size_t next_id = NextFanoutId; + using fanouts = FanoutList; + using edges = EdgeList; +}; + +// Process one edge given pre-computed port counts +template +struct process_edge { + using Port = src_port_of; + using SrcNode = typename Edge::src_node_t; + using T = std::tuple_element_t; + static constexpr std::size_t N = lookup_count::value; + + // N==1: pass through unchanged + using passthrough_edges = append_t>; + + // N>1, first encounter: emit src→fanout + all fanout→dst edges + using DstDescs = typename collect_dsts::type; + static constexpr std::size_t fid = State::next_id; + using fanout_type = FanoutPlaceholder; + using src_to_fan = SimpleEdge, 0>; + using fan_to_dsts = typename fanout_to_dst_edges::type; + using fanout_edges = concat_t, fan_to_dsts>; + using new_fanouts = append_t; + + static constexpr bool seen = already_processed::value; + + using type = std::conditional_t< + (N == 1), + FoldState, + std::conditional_t< + !seen, + FoldState, + State::next_id + 1, + new_fanouts, fanout_edges>, + // already processed — skip (fanout edges already emitted) + State>>; +}; + +// Fold over all edges +template +struct fold_edges; + +template +struct fold_edges, CountList, AllEdges> { using type = State; }; + +template +struct fold_edges, CountList, AllEdges> { + using next = typename process_edge::type; + using type = typename fold_edges, CountList, AllEdges>::type; +}; + +// ── Public interface ─────────────────────────────────────────────────────────── + +template +struct expand_all { + using AllEdges = TypeList; + using UniquePorts = typename unique_src_ports::type; + using CountList = typename build_port_counts::type; + using InitState = FoldState, 0, TypeList<>, TypeList<>>; + using FinalState = typename fold_edges::type; + + using fanout_placeholders = typename FinalState::fanouts; // TypeList> + using expanded_edges = typename FinalState::edges; // TypeList> +}; + +// Convert TypeList...> to std::tuple...> +template +struct to_fanout_tuple; + +template<> +struct to_fanout_tuple> { using type = std::tuple<>; }; + +template +struct to_fanout_tuple, Rest...>> { + using rest = typename to_fanout_tuple>::type; + // prepend FanoutNode + template struct prepend; + template struct prepend> { + using type = std::tuple, Ts...>; + }; + using type = typename prepend::type; +}; + +template +using fanout_storage_t = typename to_fanout_tuple< + typename expand_all::fanout_placeholders>::type; + +template +using expanded_edges_t = typename expand_all::expanded_edges; + +} // namespace kpn::tmp diff --git a/include/kpn/tmp/repeat_tuple.hpp b/include/kpn/tmp/repeat_tuple.hpp new file mode 100644 index 0000000..8f3d70c --- /dev/null +++ b/include/kpn/tmp/repeat_tuple.hpp @@ -0,0 +1,23 @@ +#pragma once +#include +#include +#include + +namespace kpn::tmp { + +// repeat_tuple_t — std::tuple with N elements. +// Used by FanoutNode and fanout_groups to express N homogeneous outputs. + +template> +struct repeat_tuple; + +template +struct repeat_tuple> { + template using always_T = T; + using type = std::tuple...>; +}; + +template +using repeat_tuple_t = typename repeat_tuple::type; + +} // namespace kpn::tmp diff --git a/include/kpn/tmp/topo_sort.hpp b/include/kpn/tmp/topo_sort.hpp new file mode 100644 index 0000000..c288cc3 --- /dev/null +++ b/include/kpn/tmp/topo_sort.hpp @@ -0,0 +1,223 @@ +#pragma once +#include "fanout_groups.hpp" +#include +#include +#include + +// Compile-time topological sort + cycle detection over a TypeList>. +// +// Nodes are identified by type (not by index), since two different node types +// are always distinct vertices. The sort produces a TypeList of node types in +// topological order (sources first). +// +// cycle detection: if DFS revisits a node currently on the stack → static_assert. + +namespace kpn::tmp { + +// ── Collect all unique node types from an expanded edge list ────────────────── + +template> +struct all_node_types; + +template +struct all_node_types, Seen> { using type = Seen; }; + +template +struct all_node_types, Seen> { + using S1 = std::conditional_t< + already_processed::value, + Seen, append_t>; + using S2 = std::conditional_t< + already_processed::value, + S1, append_t>; + using type = typename all_node_types, S2>::type; +}; + +// ── Collect successors (dst node types) of a given src node type ────────────── + +template +struct successors; + +template +struct successors> { using type = TypeList<>; }; + +template +struct successors> { + using rest = typename successors>::type; + using type = std::conditional_t< + std::is_same_v, + std::conditional_t< + already_processed::value, + rest, + append_t>, + rest>; +}; + +// ── DFS state ───────────────────────────────────────────────────────────────── + +// Color: 0=white(unseen), 1=grey(on stack), 2=black(done) +// We track grey nodes as a TypeList to detect back-edges. + +template +struct is_grey : already_processed {}; + +// DFS result: has_cycle flag + topo order (black nodes appended post-visit) +template +struct DfsResult { static constexpr bool has_cycle = Cycle; using topo = TopoList; }; + +template +struct dfs_node; + +// Iterate successors +template +struct dfs_successors; + +template +struct dfs_successors, EdgeList, GreyList, BlackList, TopoList, CycleSoFar> { + static constexpr bool has_cycle = CycleSoFar; + using black = BlackList; + using topo = TopoList; +}; + +template +struct dfs_successors, EdgeList, + GreyList, BlackList, TopoList, CycleSoFar> { + // Visit Head + using visit = dfs_node; + static constexpr bool cycle1 = CycleSoFar || visit::has_cycle; + // Continue with remaining successors using updated black/topo + using rest = dfs_successors, EdgeList, + GreyList, typename visit::black, + typename visit::topo, cycle1>; + static constexpr bool has_cycle = rest::has_cycle; + using black = typename rest::black; + using topo = typename rest::topo; +}; + +template +struct dfs_node { + // Already black — skip + static constexpr bool already_done = already_processed::value; + // On stack — cycle + static constexpr bool on_stack = is_grey::value; + + // Visit successors (only if not already done / on stack) + using succs = typename successors::type; + using new_grey = append_t; + + using visit_succs = dfs_successors; + + static constexpr bool has_cycle = already_done ? false : + on_stack ? true : + visit_succs::has_cycle; + + // Append node to topo after all successors (post-order = reverse topo) + using topo_after = std::conditional_t< + already_done || on_stack, + TopoList, + append_t>; + + using black = std::conditional_t< + already_done || on_stack, + BlackList, + append_t>; + + using topo = topo_after; +}; + +// ── Top-level DFS over all nodes ────────────────────────────────────────────── + +template +struct dfs_all; + +template +struct dfs_all, EdgeList, BlackList, TopoList, CycleSoFar> { + static constexpr bool has_cycle = CycleSoFar; + using topo = TopoList; +}; + +template +struct dfs_all, EdgeList, BlackList, TopoList, CycleSoFar> { + using visit = dfs_node, BlackList, TopoList>; + static constexpr bool cycle1 = CycleSoFar || visit::has_cycle; + using rest = dfs_all, EdgeList, + typename visit::black, typename visit::topo, cycle1>; + static constexpr bool has_cycle = rest::has_cycle; + using topo = typename rest::topo; +}; + +// ── Public interface ─────────────────────────────────────────────────────────── + +// topo_sort: +// ::topo — TypeList of node types, sources first (start order) +// ::has_cycle — true if a cycle was detected +template +struct topo_sort { + using Nodes = typename all_node_types::type; + using result = dfs_all, TypeList<>, false>; + // DFS post-order gives reverse topo; reverse the list for sources-first order. + // Reversing a TypeList: + template> + struct reverse_list; + template + struct reverse_list, Acc> { using type = Acc; }; + template + struct reverse_list, Acc> + : reverse_list, append_t> {}; // wrong direction intentionally + + // Post-order appends children before parent, so the list is already + // reverse-topo (sinks first). Reverse it to get sources first. + template> + struct rev; + template + struct rev, Acc> { using type = Acc; }; + template + struct rev, Acc> : rev, TypeList> { + // prepend H to Acc: TypeList + }; + // Simpler: just reverse by prepending + template> + struct rev2 { using type = A; }; + template + struct rev2, TypeList> + : rev2, TypeList> {}; + + static constexpr bool has_cycle = result::has_cycle; + using topo = typename rev2::type; +}; + +// ── Duplicate-tag detection ─────────────────────────────────────────────────── +// +// Two user nodes share a (Func, UniqueTag) pair iff they have the same type. +// has_duplicate_tags_v is true if any two user node types are identical. + +template +struct type_in_list : std::false_type {}; +template +struct type_in_list> + : std::conditional_t, + std::true_type, + type_in_list>> {}; + +template +struct has_duplicate_node_types : std::false_type {}; +template +struct has_duplicate_node_types> + : std::conditional_t>::value, + std::true_type, + has_duplicate_node_types>> {}; + +template +inline constexpr bool has_duplicate_tags_v = + has_duplicate_node_types< + typename all_node_types>::type>::value; + +} // namespace kpn::tmp diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index e8e282d..2c14ed3 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -31,6 +31,7 @@ add_executable(kpn_tests test_channel.cpp test_node.cpp test_network.cpp + test_static_network.cpp ) target_link_libraries(kpn_tests PRIVATE diff --git a/tests/test_static_network.cpp b/tests/test_static_network.cpp new file mode 100644 index 0000000..b5f82a5 --- /dev/null +++ b/tests/test_static_network.cpp @@ -0,0 +1,209 @@ +#include +#include +#include +#include +#include + +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(5); + auto dst = make_node(5); + + Channel 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(5); + auto b = make_node(5); + auto c = make_node(5); + + Channel 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(8); + auto dstA = make_node(8); + auto dstB = make_node(8); + + Channel outA(8), outB(8); + dstA.set_output_channel<0>(&outA); + dstB.set_output_channel<0>(&outB); + + // Two edges from the same output port — FanoutNode 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(16); + auto dstA = make_node(16); + auto dstB = make_node(16); + + Channel 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(5); + auto dst = make_node(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(5); +// auto y = make_node(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(5); + auto b = make_node(5); + + Channel 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(8); + auto b = make_node(8); // same func, tag=1 → distinct type + + Channel 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: label is accessible as static member", "[static_network]") { + using MyNode = decltype(make_node(5)); + REQUIRE(MyNode::label() == "my_node"); + REQUIRE(MyNode::unique_tag == 0); + + using TaggedNode = decltype(make_node(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(8); + auto dstA = make_node(8); + auto dstB = make_node(8); + + Channel 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(); +}