commit 5e77dc836bfe02489c8bd86ba9eadfce4a6f3f05 Author: Duncan Tourolle Date: Fri May 8 17:48:16 2026 +0200 First attempt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..09931b4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,27 @@ +# Build output +build/ + +# Python +__pycache__/ +*.py[cod] +*.pyd +*.pyo +*.egg-info/ +dist/ +*.egg +.venv/ +venv/ + +# Editors +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Claude Code local settings +.claude/settings.local.json diff --git a/CMakeLists.txt b/CMakeLists.txt new file mode 100644 index 0000000..a1fe93c --- /dev/null +++ b/CMakeLists.txt @@ -0,0 +1,71 @@ +cmake_minimum_required(VERSION 3.21) +project(kpnpp VERSION 0.1.0 LANGUAGES CXX) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_CXX_EXTENSIONS OFF) + +option(KPN_BUILD_TESTS "Build tests" ON) +option(KPN_BUILD_PYTHON "Build Python bindings (requires nanobind)" ON) +option(KPN_BUILD_EXAMPLES "Build examples" ON) +option(KPN_WEB_DEBUG "Enable web debug UI (cpp-httplib)" OFF) + +# ── Core library (header-only) ──────────────────────────────────────────────── +add_library(kpn INTERFACE) +target_include_directories(kpn INTERFACE + $ + $ +) +target_compile_features(kpn INTERFACE cxx_std_20) + +# Threads required by node/channel implementation +find_package(Threads REQUIRED) +target_link_libraries(kpn INTERFACE Threads::Threads) + +# ── Web debug UI (optional) ─────────────────────────────────────────────────── +if(KPN_WEB_DEBUG) + include(FetchContent) + FetchContent_Declare( + cpp-httplib + GIT_REPOSITORY https://github.com/yhirose/cpp-httplib.git + GIT_TAG v0.18.0 + ) + FetchContent_MakeAvailable(cpp-httplib) + # httplib made available but NOT forced onto kpn interface — targets opt in + # by defining KPN_WEB_DEBUG=1 and linking httplib::httplib themselves. + # This prevents tests and other examples from pulling in the HTTP server. +endif() + +# Convenience function for targets that want web debug +function(kpn_target_enable_web_debug target) + target_compile_definitions(${target} PRIVATE KPN_WEB_DEBUG=1) + target_link_libraries(${target} PRIVATE httplib::httplib) +endfunction() + +# ── Tests ───────────────────────────────────────────────────────────────────── +if(KPN_BUILD_TESTS) + enable_testing() + add_subdirectory(tests) +endif() + +# ── Python bindings ─────────────────────────────────────────────────────────── +if(KPN_BUILD_PYTHON) + find_package(Python 3.8 COMPONENTS Interpreter Development.Module REQUIRED) + find_package(nanobind CONFIG QUIET) + if(NOT nanobind_FOUND) + # Fall back to FetchContent if nanobind not installed system-wide + include(FetchContent) + FetchContent_Declare( + nanobind + GIT_REPOSITORY https://github.com/wjakob/nanobind.git + GIT_TAG v2.12.0 + ) + FetchContent_MakeAvailable(nanobind) + endif() + add_subdirectory(python) +endif() + +# ── Examples ────────────────────────────────────────────────────────────────── +if(KPN_BUILD_EXAMPLES) + add_subdirectory(examples) +endif() diff --git a/README.md b/README.md new file mode 100644 index 0000000..9ab76d3 --- /dev/null +++ b/README.md @@ -0,0 +1,257 @@ +# KPN++ + +A C++20 Kahn Process Network (KPN) library. Each node wraps a function and runs in its own thread, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind. + +--- + +## Requirements + +| Dependency | Version | Notes | +|---|---|---| +| CMake | ≥ 3.21 | | +| C++ compiler | GCC ≥ 11, Clang ≥ 13, MSVC 19.29 | C++20 required | +| Threads | system | `find_package(Threads)` | +| nanobind | ≥ 2.1 | auto-fetched if not installed; Python ≥ 3.8 | +| Catch2 | v3 | auto-fetched for tests | +| Google Test | v1.14 | auto-fetched for tests | +| OpenCV | ≥ 4 | optional; only for example 09 | + +--- + +## Build + +```bash +cmake -B build -DKPN_BUILD_PYTHON=OFF # core + tests + C++ examples +cmake --build build --parallel +ctest --test-dir build +``` + +Enable Python bindings (requires nanobind and Python dev headers): + +```bash +cmake -B build -DKPN_BUILD_PYTHON=ON +cmake --build build --parallel +``` + +Disable examples: + +```bash +cmake -B build -DKPN_BUILD_EXAMPLES=OFF +``` + +--- + +## Core Concepts + +### Nodes + +A node wraps any callable. Its input types are taken from the function's parameter list; its output types from the return type. Multi-output nodes return `std::tuple<...>`. + +```cpp +#include +using namespace kpn; + +// Single input, single output +int double_it(int x) { return x * 2; } + +// Multi-output — must return std::tuple +std::tuple split(cv::Mat frame) { ... } + +// Sink — void return, no output ports +void display(cv::Mat frame) { cv::imshow("out", frame); } +``` + +### Creating Nodes + +```cpp +// No port names (index-only access) +auto node = make_node(/*fifo_capacity=*/5); + +// Named input ports only +auto node = make_node(in<"value">{}, 5); + +// Named input and output ports +auto node = make_node(in<"value">{}, out<"result">{}, 5); + +// Named output ports only (e.g. a source with no inputs) +auto node = make_node(out<"colour","grey">{}, 5); +``` + +Port names are NTTP `fixed_string` values — resolved entirely at compile time, zero runtime cost. + +### Building a Network + +`Network` is **non-owning** — declare nodes first, then register them. Nodes must outlive the network. + +```cpp +auto src = make_node(in<"x">{}, out<"value">{}, 5); +auto proc = make_node(in<"value">{}, out<"result">{}, 5); + +Network net; +net.add("src", src) + .add("proc", proc) + .connect("src", src.output<0>(), "proc", proc.input<0>()) // by index + .connect("src", src.template output<"value">(), "proc", proc.template input<"value">()) // by name + .build(); // runs cycle detection — throws NetworkCycleError on cycles + +net.start(); +// ... do work ... +net.stop(); +``` + +> **Named port syntax in template context:** when the node variable is `auto`-deduced, use `.template output<"name">()` and `.template input<"name">()` to help the parser. + +### Channel Semantics + +- **Bounded FIFO**: default capacity 5, configurable per-node at construction. +- **Blocking `pop()`**: consumer blocks until data is available (KPN semantics). +- **Throwing `push()`**: throws `ChannelOverflowError` if the channel is full and accepting. +- **Silent drop on disabled channel**: after `node.stop()`, its input channels are disabled — producers that push into them have the value silently dropped. No exception, no blocking. +- **Source throttling**: source nodes (no inputs) must sleep or yield to avoid overflowing downstream FIFOs. See example 09. + +### Storage Policy + +Large types (`sizeof > 8` or non-trivially-copyable) are stored as `std::shared_ptr` inside the channel — no copies, shared immutable ownership. Small trivially-copyable types are stored by value. + +Override the policy for a specific type: + +```cpp +template<> struct kpn::channel_storage_policy { + static constexpr bool by_value = true; +}; +``` + +### Shutdown + +`node.stop()` / `net.stop()`: +1. Sets `accepting_ = false` on all input channels (drops in-flight pushes silently). +2. Clears any queued items from those channels. +3. Unblocks any thread blocked on `pop()` (throws `ChannelClosedError` inside `run_loop`, which exits cleanly). +4. Joins the node thread. + +--- + +## Named Ports — Design Notes + +Port names use C++20 NTTP `fixed_string`. The deduction guide is required: + +```cpp +template +fixed_string(const char (&)[N]) -> fixed_string; +``` + +`fixed_string<4>` and `fixed_string<7>` are distinct types — `input<"img">()` and `input<"sigma">()` resolve to different template instantiations at compile time. Wrong names produce a `static_assert` at the call site with a readable message. + +--- + +## Sub-Networks + +`Network` implements `INode`, so it can be nested inside a larger `Network`: + +```cpp +// Inner sub-network +Network pipe; +pipe.add("pre", pre_node) + .add("enh", enh_node) + .connect("pre", pre_node.output<0>(), "enh", enh_node.input<0>()) + .expose_input("img", pre_node.input<0>()) + .expose_output("result", enh_node.output<0>()) + .build(); + +// Outer network +Network top; +top.add("pipe", pipe) + .add("sink", sink_node) + .connect("pipe", pipe.output<"result">(), "sink", sink_node.input<0>()) + .build(); +top.start(); +``` + +--- + +## Display / GUI Nodes + +**Do not wrap `imshow`/`waitKey` as a KPN node.** Qt and Wayland require these to run on the main thread (the thread that owns the event loop). Instead, wire the final output channel to a `Channel` and pop it on the main thread: + +```cpp +Channel result_ch(8); +comp.set_output_channel<0>(&result_ch); +// ... build and start network ... + +// Main thread display loop +while (true) { + cv::Mat frame; + if (!result_ch.try_pop(frame, std::chrono::milliseconds(100))) continue; + cv::imshow("output", frame); + if (cv::waitKey(1) == 'q') break; +} +net.stop(); +``` + +--- + +## Python Bindings + +> Python bindings are scaffolded but not yet fully implemented. See `python/kpn_python.cpp` and `include/kpn/python/bindings.hpp`. + +A `PyNetwork` is constructed from a closed list of C++ node types. The variant of all port types is derived at compile time — no runtime type registration needed. + +**GIL rules (non-negotiable):** +- Acquire the GIL only for the duration of a Python callable invocation. +- Release the GIL before any blocking channel operation (`pop()`, `push()`, `net.read()`, `net.write()`). + +Violating the second rule deadlocks. + +--- + +## Examples + +| Example | What it shows | +|---|---| +| `01_hello_pipeline` | Linear pipeline, index-based port wiring | +| `02_named_ports` | `in<>`/`out<>` name tags, named port access | +| `03_multi_output` | Tuple-returning node, per-element sub-port routing | +| `04_storage_policy` | `channel_storage_policy` default and specialisation | +| `05_error_handling` | `ChannelOverflowError`, `ErrorHandler` | +| `06_watchdog` | Watchdog interval, stall detection | +| `07_python_network` | PyNetwork, pure Python node *(pending)* | +| `08_python_subport` | `net.read`, `net.write`, sub-port tap *(pending)* | +| `09_opencv_cellshade` | Real-time cell-shading on webcam/pattern; requires OpenCV ≥ 4 | + +Run the cell-shading example: + +```bash +./build/examples/09_opencv_cellshade +# Press 'q' or close the window to stop. +# Falls back to an animated synthetic pattern if no webcam is found. +``` + +--- + +## Project Structure + +``` +include/kpn/ + fixed_string.hpp — NTTP string, in<>/out<> tags, index_of + traits.hpp — function_traits, normalised_return_t, output_count_v + channel.hpp — Channel, channel_storage_policy, exceptions + port.hpp — InputPort, OutputPort + node.hpp — Node,out<...>>, make_node, INode + network.hpp — Network (builder, cycle detection, watchdog) + variant_node.hpp — VariantNode, PythonConverter, unique_types (Python layer) + python/ + bindings.hpp — nanobind helpers, GIL rule documentation + kpn.hpp — umbrella header +src/ + network.cpp — non-template Network implementation +tests/ + test_fixed_string.cpp + test_traits.cpp + test_channel.cpp + test_node.cpp + test_network.cpp +python/ + kpn_python.cpp — nanobind module entry point +examples/ + 01_hello_pipeline/ … 09_opencv_cellshade/ +``` diff --git a/SPEC.md b/SPEC.md new file mode 100644 index 0000000..0d98d5f --- /dev/null +++ b/SPEC.md @@ -0,0 +1,999 @@ +# KPN++ — Kahn Process Network Library Specification + +## Overview + +A C++20 template-metaprogramming library for building Kahn Process Networks, where each node +wraps a function/method, runs in its own thread, and communicates via bounded FIFO queues. +Includes nanobind bindings for Python graph construction and prototyping. + +--- + +## Project Structure + +``` +kpn++/ +├── CMakeLists.txt +├── include/kpn/ +│ ├── fixed_string.hpp # NTTP string type for named ports +│ ├── traits.hpp # Function signature introspection +│ ├── channel.hpp # Bounded FIFO channel + storage policy +│ ├── node.hpp # Node wrapper + thread management +│ ├── port.hpp # Input/Output port handles +│ ├── network.hpp # Graph builder + orchestrator/watchdog +│ ├── variant_node.hpp # Runtime-typed node for Python graphs +│ └── python/ +│ └── bindings.hpp # Nanobind binding helpers +├── src/ +│ └── network.cpp # Orchestrator thread impl +├── tests/ +├── examples/ +│ ├── 01_hello_pipeline/ +│ ├── 02_named_ports/ +│ ├── 03_multi_output/ +│ ├── 04_storage_policy/ +│ ├── 05_error_handling/ +│ ├── 06_watchdog/ +│ ├── 07_python_network/ +│ ├── 08_python_subport/ +│ └── 09_opencv_cellshade/ # optional, requires OpenCV +└── python/ + └── kpn_python.cpp # Nanobind module definition +``` + +--- + +## Component 0 — `fixed_string.hpp`: NTTP String + +Named ports use C++20 non-type template parameters (NTTPs). `std::string_view` and +`const char*` are not valid NTTPs because they are not structurally comparable. The standard +solution is a `fixed_string` literal type with `constexpr` internal storage. + +```cpp +template +struct fixed_string { + char data[N]{}; + constexpr fixed_string(const char (&s)[N]) { std::copy_n(s, N, data); } + constexpr bool operator==(const fixed_string&) const = default; + constexpr std::string_view view() const { return {data, N - 1}; } +}; + +// Deduction guide — required so fixed_string("img") works as an NTTP. +// Without it the compiler cannot infer N and the named-port API does not compile. +template +fixed_string(const char (&)[N]) -> fixed_string; +``` + +`fixed_string<3>` and `fixed_string<6>` are distinct types, so `input<"img">()` and +`input<"sigma">()` produce different template instantiations — this is intentional and +enables zero-overhead compile-time port dispatch. + +Named-port lookup uses a `constexpr` function over the name pack. It returns a sentinel +`npos` on miss rather than `static_assert`-ing internally, so the assertion fires at the +`input<"img">()` call site — giving the user a readable error at the point of use instead +of deep in template instantiation: + +```cpp +inline constexpr std::size_t npos = std::size_t(-1); + +template +constexpr std::size_t index_of() { + std::size_t i = 0; + bool found = false; + ((Name == Names ? (found = true) : (found ? 0 : ++i)), ...); + return found ? i : npos; +} + +// Used at the call site: +template +auto input() { + constexpr std::size_t idx = index_of(); + static_assert(idx != npos, "unknown input port name"); + return input(); +} +``` + +--- + +## Component 1 — `traits.hpp`: Function Introspection + +Extracts parameter types and return type from any callable at compile time. + +```cpp +// For: Image blur(Image in, float sigma) +// function_traits::args == std::tuple +// function_traits::return_t == Image + +// For multi-output: std::tuple detect(Image in) +// return_t == std::tuple → 2 output ports +// return_t == Image → 1 output port (normalised to tuple internally) +// return_t == void → 0 output ports (sink node) +``` + +Handles: free functions, lambdas, `std::function`, member function pointers. + +A helper alias normalises the return type to always be a tuple for uniform handling in +`run_loop`: + +```cpp +template +using normalised_return_t = + std::conditional_t, T, std::tuple>; +// void return → std::tuple<> (empty tuple, zero output ports) +``` + +--- + +## Component 2 — `channel.hpp`: Bounded FIFO + Storage Policy + +### Storage Policy + +The type stored in a channel depends on a specialisable trait. Users can override it for +any type: + +```cpp +template +struct channel_storage_policy { + static constexpr bool by_value = + std::is_trivially_copyable_v && sizeof(T) <= 8; +}; + +// User opt-in to value semantics for a small struct: +template<> struct channel_storage_policy { + static constexpr bool by_value = true; +}; + +// Derived storage type: +template +using channel_storage_t = std::conditional_t< + channel_storage_policy::by_value, + T, + std::shared_ptr +>; +``` + +### Channel + +`Channel` stores `channel_storage_t` internally. The producer calls `push(T value)` +and the channel transparently wraps it in `make_shared` when needed. All consumers +of the same channel receive the same `shared_ptr` — no copies of large objects. + +`run_loop` dereferences `shared_ptr` before passing to the wrapped function, so a +function declared `void f(const Image& img)` works naturally and the compiler enforces +immutability — no policy enforcement or `const_cast` needed. + +```cpp +template +class Channel { +public: + using storage_type = channel_storage_t; + + explicit Channel(std::size_t capacity = 5); + + void push(T value); // wraps in shared_ptr if needed; throws on overflow + T pop(); // blocks (KPN semantics); unwraps shared_ptr if needed + bool try_pop(T& out, std::chrono::milliseconds timeout); + + std::size_t size() const; + std::size_t capacity() const; +}; + +class ChannelOverflowError : public std::runtime_error {}; +``` + +### Ownership + +A `Channel` is **owned by its consumer node** — it lives as a member of the destination +node. The producer node holds a non-owning raw pointer to push into it. The channel is +destroyed when its consumer is destroyed, which is the correct lifetime. + +The `Network` itself is **non-owning** — nodes are declared by the user and outlive the +network. `net.add("name", node)` registers a raw pointer; the user is responsible for keeping +nodes alive for the network's lifetime. This avoids type-erasure ownership complexity and +keeps node construction explicit. + +### Backpressure and Shutdown — `accepting_` Flag + +Each channel carries a single `std::atomic accepting_` (default `true`). This is the +**sole shutdown mechanism** — no `try_pop` polling, no sentinel values, no drain logic. + +```cpp +template +class Channel { + std::atomic accepting_{true}; +public: + void push(T value) { + if (!accepting_.load(std::memory_order_relaxed)) return; // silently drop + // normal push — throws ChannelOverflowError if full + } + + void enable() { accepting_.store(true, std::memory_order_relaxed); } + void disable() { + accepting_.store(false, std::memory_order_relaxed); + clear(); // drop all queued items immediately + cv_.notify_all(); // unblock any waiting pop() + } +}; +``` + +**Who flips the flag:** the **consumer node** — it's the channel owner. `node.stop()` calls +`disable()` on all its input channels. `node.start()` calls `enable()`. The producer never +touches the flag; it calls `push()` and if the channel is disabled the value is silently +dropped and the producer continues. + +**Overflow** (`push()` on a full, accepting channel) still throws `ChannelOverflowError` — +this signals a design error (undersized FIFO) and is unchanged. + +**Blocking `pop()`** unblocks immediately when `disable()` is called (via `cv_.notify_all()`), +and throws `ChannelClosedError` if the queue is empty and the channel is disabled. + +### `try_pop` Purpose + +`try_pop` exists for **watchdog polling only** — not for shutdown (the `accepting_` flag +handles that) and not for normal processing. The watchdog uses it to probe whether a node +is making progress without blocking the watchdog thread. + +> **Note (C++ compile-time graphs):** In a fully compiled C++ graph the variant never +> appears. The compiler wires `Channel` to `Channel` directly. The variant is +> a pure compile-time construct used only for type-checking and generates zero runtime +> overhead. + +--- + +## Component 3 — `port.hpp`: Port Handles + +```cpp +template +struct InputPort { NodeT& node; }; + +template +struct OutputPort { NodeT& node; }; +``` + +Nodes expose port handles via: + +```cpp +// By index — always available +node_a.input<0>() // returns InputPort +node_b.output<1>() // returns OutputPort + +// By name — only valid when names were provided at make_node time +node_a.input<"img">() +node_b.output<"edges">() +``` + +Named access resolves to an index at compile time via `index_of` (see `fixed_string.hpp`). +Zero runtime cost — the name dispatch is fully eliminated by the compiler. + +--- + +## Component 4 — `node.hpp`: Node Wrapper + +```cpp +template< + auto Func, + fixed_string... InputNames, // optional; count must match arity or be 0 + fixed_string... OutputNames // optional; count must match output count or be 0 +> +class Node { +public: + explicit Node(std::size_t fifo_capacity = 5); + + void start(); + void stop(); // signals thread to finish current item then exit + + // Port access — by index + template auto input(); + template auto output(); + + // Port access — by name (compile error if names were not provided) + template auto input(); + template auto output(); + + static constexpr std::size_t input_count; + static constexpr std::size_t output_count; + +private: + void run_loop(); + // Pops each input channel, dereferences shared_ptr if needed, + // calls Func, unpacks the normalised tuple return, + // pushes each element to its output channel. + + std::thread thread_; + // Input channels owned here (one per input port). + // Output channel pointers (non-owning) set at connect time. +}; +``` + +**Factory syntax — `in<>` / `out<>` tag structs:** + +A flat name pack `make_node` is ambiguous (where do inputs end?). +Option chosen: `in<...>` and `out<...>` tag types that wrap the name packs unambiguously. +Both are optional; omitting either means those ports are index-only. + +```cpp +// Tag types (trivial, no data): +template struct in {}; +template struct out {}; + +// Factory: +// No names +auto node = make_node(/*fifo_capacity=*/10); + +// Input names only +auto node = make_node>(10); + +// Both input and output names +auto node = make_node, out<"blurred","mask">>(10); +``` + +**Wrong name count is a compile error.** The `Node` class `static_assert`s that +`sizeof...(InputNames) == 0 || sizeof...(InputNames) == input_count` (and same for outputs). +Without this, a mismatch between name count and arity produces an unreadable template error. + +```cpp +static_assert( + sizeof...(InputNames) == 0 || sizeof...(InputNames) == input_count, + "make_node: number of input names must match function arity, or provide none" +); +``` + +Multi-output functions must return `std::tuple<...>`. Single return accepted as-is. +`void` return = sink node (no output ports). + +--- + +## Component 5 — `network.hpp`: Graph Builder + Orchestrator + +`Network` is **non-owning** — nodes are declared by the user and must outlive the network. +`add()` registers a raw pointer. Graph construction uses a builder pattern so the full +topology is known before `build()`, enabling cycle detection and topological ordering. + +```cpp +class Network : public INode { // Network is itself an INode — enables sub-networks +public: + // Register a node by name. NodeT must satisfy INode. Network holds a raw pointer. + template + Network& add(std::string name, NodeT& node); + + // Connect output port of src to input port of dst. + // Type mismatch → static_assert at compile time. + template + Network& connect(const std::string& src_name, OutputPort, + const std::string& dst_name, InputPort); + + // Expose an internal node's input/output as a boundary port of this (sub-)network. + // Allows a Network to be connected into a larger Network like a single node. + template + Network& expose_input(std::string boundary_name, InputPort); + + template + Network& expose_output(std::string boundary_name, OutputPort); + + // DFS cycle check + topological sort. Throws NetworkCycleError on cycles. + Network& build(); + + void start() override; // starts all internal nodes in topological order + void stop() override; // stops all internal nodes in reverse order; disables channels + bool running() const override; + + void set_watchdog_interval(std::chrono::milliseconds); + + using ErrorHandler = std::function; + void set_error_handler(ErrorHandler); + +private: + std::map nodes_; // non-owning + std::map> adj_; + std::vector topo_; + std::jthread watchdog_; + std::chrono::milliseconds watchdog_interval_{500}; + ErrorHandler error_handler_; +}; +``` + +**Node lifetime contract:** nodes must outlive the `Network`. The typical pattern is to +declare nodes and the network in the same scope: + +```cpp +// Nodes declared first — they own their input channels +auto blur = make_node>(10); +auto detect = make_node>(10); + +Network net; +net.add("blur", blur) + .add("detect", detect) + .connect("blur", blur.output<0>(), "detect", detect.input<0>()) + .connect("blur", blur.output<"blurred">(), "detect", detect.input<"img">()) + .build(); +net.start(); +``` + +**Sub-networks** — because `Network` implements `INode`, it can be registered inside a +larger `Network` as a named node. Boundary ports declared via `expose_input` / +`expose_output` make the internal nodes' ports available to the outer graph: + +```cpp +// Inner sub-network +auto stage1 = make_node(5); +auto stage2 = make_node(5); +Network pipe; +pipe.add("pre", stage1).add("enh", stage2) + .connect("pre", stage1.output<0>(), "enh", stage2.input<0>()) + .expose_input("img", stage1.input<0>()) + .expose_output("result", stage2.output<0>()) + .build(); + +// Outer network treats `pipe` as a single node +auto sink = make_node(5); +Network top; +top.add("pipe", pipe).add("sink", sink) + .connect("pipe", pipe.output<"result">(), "sink", sink.input<0>()) + .build(); +top.start(); +``` + +`NetworkCycleError` is thrown by `build()` if the graph contains a directed cycle. + +--- + +## Component 6 — `variant_node.hpp`: Runtime-typed Node (Python graphs) + +### Motivation + +Python graphs cannot use compile-time type resolution. A `PyNetwork` is constructed with a +**closed list of C++ node types** known at binding time. The library derives a deduplicated +`std::variant` from all port types across those nodes. Type safety is enforced at +`connect()` time via string signatures. + +### Variant Deduplication + +All port types from the registered nodes are collected into a flat pack, duplicates are +removed via a `unique_types` TMP metafunction, then the variant is instantiated once: + +```cpp +template +using py_variant_t = std::variant>>; +``` + +This is pure TMP and runs entirely at compile time. The resulting variant has no redundant +alternatives at runtime. + +### PyNetwork Construction + +`make_py_network` is a **pure C++ template** — no CMake code-gen step. The variant is +derived entirely at compile time from the registered node type list. The nanobind module +definition is the single place where node types are listed; recompiling the extension is +the "registration" step. + +```cpp +// In kpn_python.cpp — list all node types that may appear in Python graphs: +auto py_net = make_py_network(); +// VariantValue = std::variant< /* deduplicated port types from A, B, C */ > +// Registers to_python / from_python converters for each alternative. +``` + +### VariantChannel + +```cpp +using VariantValue = py_variant_t; + +class VariantChannel { +public: + explicit VariantChannel(std::size_t capacity = 5); + void push(VariantValue v); // throws ChannelOverflowError if full + VariantValue pop(); // blocks (KPN semantics) +}; +``` + +### VariantNode + +Wraps a registered C++ node type. Its `run_loop` uses `std::visit` to extract the concrete +type from a `VariantValue`, calls the underlying function, then wraps the result back into a +`VariantValue` for the output channel. + +```cpp +class VariantNode { +public: + std::string input_type_sig(std::size_t idx) const; + std::string output_type_sig(std::size_t idx) const; + + void connect_input (std::size_t port, std::shared_ptr); + void connect_output(std::size_t port, std::shared_ptr); + + void start(); + void stop(); +}; +``` + +### PythonConverter — Crossing the C++/Python Boundary + +Every type in the variant must provide a `PythonConverter` specialisation. This is the +single mechanism used for all data crossing into or out of Python (PyNodes, `net.read`, +`net.write`): + +```cpp +template +struct PythonConverter { + static nanobind::object to_python(const T&); + static T from_python(nanobind::object); +}; +``` + +### PyNode — Pure Python Processing Node + +A `PyNode` holds a `nanobind::object` as its function. Its `run_loop`: + +1. Pops `VariantValue` from each input channel +2. `std::visit` → calls `PythonConverter::to_python` for each → **acquires GIL** +3. Calls the Python callable +4. **Releases GIL** → calls `PythonConverter::from_python` on the return value +5. Pushes result as `VariantValue` to output channel + +### Sub-value Extraction and Injection + +A C++ node returning `std::tuple` exposes three independent output ports. Each +element is pushed to its own `VariantChannel` — sub-indexing is a first-class concept at the +channel level, not an afterthought. + +**Python tap — read one output port into Python:** + +```python +value = net.read("detect", output=2) +# Pops from output channel 2, calls PythonConverter::to_python. +# GIL released while blocking on pop(), re-acquired before to_python call. +``` + +**Python inject — write a Python value into a specific input port:** + +```python +net.write("blur", input=1, value=my_sigma) +# Calls PythonConverter::from_python(my_sigma), pushes to input channel 1. +# GIL released while blocking on push() if channel is full. +``` + +**Python splitter node:** + +```python +def split(packed): + img, mask, score = packed + return img, mask + +net.add_node("split", split, inputs=["packed"], outputs=["img", "mask"]) +net.connect("detect", 0, "split", 0) +net.connect("split", 0, "show", 0) +net.connect("split", 1, "save", 0) +``` + +**Direct C++ sub-output to Python node input:** + +```python +net.connect("detect", 1, "py_thresh", 0) +# Type sig of detect:output[1] must match py_thresh:input[0] — checked at connect(). +``` + +### Type-check at connect time (Python) + +```python +# Raises kpn.TypeError if signatures don't match +net.connect("blur", 0, "thresh", 0) # (src_name, out_idx, dst_name, in_idx) +``` + +--- + +## Component 7 — Orchestrator / Watchdog + +Runs in its own dedicated thread inside `Network` / `PyNetwork`. Responsibilities: + +- Starts nodes in topological order; stops them in reverse order +- Tracks per-node execution time (exponential moving average) +- Emits warning via logger callback (default: `stderr`) when a node stalls beyond threshold +- Catches exceptions from node threads and routes them to `ErrorHandler` +- Graceful shutdown: signals all nodes, joins with timeout, reports any that fail to stop + +--- + +## GIL Rules (non-negotiable constraints on binding implementation) + +Two rules govern all interaction between node threads and the Python interpreter: + +1. **Acquire for callback** — a node thread must hold the GIL only for the duration of a + Python callable invocation (`nb::gil_scoped_acquire` wrapping the call site). + +2. **Release while blocking** — any blocking operation on a channel (`pop()`, `push()`, + `net.read()`, `net.write()`) must release the GIL before blocking + (`nb::gil_scoped_release` wrapping the call site), then re-acquire after. + +Violating rule 2 deadlocks: a PyNode thread waiting to acquire the GIL cannot proceed while +another thread holds the GIL and blocks on a channel waiting for that PyNode to produce. + +--- + +## Error Handling Contract + +| Situation | Behaviour | +|---|---| +| FIFO overflow | `ChannelOverflowError` thrown in producer thread → `ErrorHandler` | +| Node function throws | Exception pointer captured → `ErrorHandler` | +| Type mismatch (C++) | `static_assert` at `connect()` compile time | +| Type mismatch (Python) | `kpn.TypeError` raised at `net.connect()` call | +| Cycle in graph | `NetworkCycleError` thrown at `build()` time | +| Thread fails to stop | Watchdog warning after configurable timeout | +| `from_python` / `to_python` fails | Exception propagated to `ErrorHandler` | + +--- + +## Future Extension Points (Heterogeneous Execution) + +Not implemented now, but the design must not close these doors: + +- **`IChannel` abstract interface** — `Channel` and a future `RemoteChannel` (wrapping + a socket/queue) would share the same `push`/`pop` interface. Nodes never know whether + their channel is in-process or remote. + +- **`Serializer` trait** — parallel to `PythonConverter` and `channel_storage_policy`, + a specialisable trait for cross-device serialisation (MessagePack for ESP32, pinned memory + for GPU zero-copy, etc.). + +- **`NodeKind` tag** — `enum class NodeKind { Local, Gpu, Remote }` on the `INode` + interface, letting the watchdog apply different health-check and timeout strategies per + device type. + +These three extension points are sufficient to support GPU and embedded/network targets +without redesigning the core. + +--- + +## Thread Model + +**v1: one `std::thread` per node.** This maps directly to KPN theory and is simple to reason +about. It does not scale to networks with hundreds of nodes but is appropriate for the +typical use case (tens of nodes, each doing non-trivial work). + +`std::jthread` (C++20) is preferred over `std::thread` where available, as it provides a +built-in `stop_token` that simplifies the `stop()` / `try_pop` shutdown pattern. + +A future executor/thread-pool model (where multiple nodes share a pool of threads and are +scheduled cooperatively) is a possible v2 extension. The `INode` interface is designed to +not assume a 1:1 thread mapping. + +--- + +## Platform and Compiler Requirements + +C++20 is required. Specific features used: + +| Feature | Header / Standard | Min compiler | +|---|---|---| +| NTTP structural types (`fixed_string`) | language | GCC 11, Clang 13, MSVC 19.29 | +| `std::is_trivially_copyable_v` | `` | C++17+ | +| `std::jthread` + `stop_token` | `` | GCC 11, Clang 14, MSVC 19.29 | +| `if constexpr`, fold expressions | language | C++17+ | +| `auto` NTTPs | language | C++20 | +| Concepts (`requires`) | language | GCC 10, Clang 10 | + +**Minimum supported compilers:** GCC 11, Clang 13, MSVC 19.29 (VS 2022). +nanobind requires Python 3.8+ and a C++17-capable compiler (satisfied by the above). + +--- + +## Testing Strategy + +Test frameworks: **Catch2 v3** (header-friendly, good async/threading support via +`REQUIRE_NOTHROW` + thread join patterns) and **Google Test** (for death tests and +parameterised test suites). Both are included; use Catch2 for integration/behaviour tests +and GTest for unit tests where `ASSERT_*` / `EXPECT_*` macros and death tests are +preferable. + +### Hard cases to cover explicitly: + +| Case | What to test | +|---|---| +| Channel blocking | `pop()` blocks until a producer pushes; unblocks exactly once per push | +| Channel overflow | `push()` beyond capacity throws `ChannelOverflowError` | +| Shutdown race | `stop()` called while a node is blocked on `pop()` — thread must exit cleanly | +| Multi-consumer | Two nodes connected to the same output channel each receive every item (fan-out) | +| Tuple unpacking | Multi-output node pushes correct type to each sub-channel | +| Cycle detection | `build()` throws `NetworkCycleError` for a graph with a cycle | +| Named port lookup | `input<"wrong">()` fires `static_assert`; `input<"right">()` resolves correctly | +| Wrong name count | `make_node` with mismatched name count fires readable `static_assert` | +| GIL deadlock | PyNode + blocking `net.read()` from Python do not deadlock | +| `from_python` failure | Exception propagates to `ErrorHandler`, network continues | +| `channel_storage_policy` | Large type is stored as `shared_ptr`; small type by value | + +--- + +## Examples + +Each example is a self-contained program under `examples/`. They are built as part of the +CMake build and serve as both documentation and smoke tests. + +### `examples/01_hello_pipeline` — Basic linear pipeline + +Two nodes connected in sequence. Demonstrates `make_node`, `Network` builder, +index-based port connection, `start_all` / `stop_all`. + +```cpp +// producer → transform → sink +int produce() { return 42; } +int double_it(int x) { return x * 2; } +void print_it(int x) { std::cout << x << '\n'; } + +auto src = make_node(5); +auto dbl = make_node(5); +auto sink = make_node(5); + +Network net; +net.add("src", src) + .add("dbl", dbl) + .add("sink", sink) + .connect("src", src.output<0>(), "dbl", dbl.input<0>()) + .connect("dbl", dbl.output<0>(), "sink", sink.input<0>()) + .build(); +net.start_all(); +``` + +### `examples/02_named_ports` — Named port access + +Same pipeline but using `in<>` / `out<>` name tags and named port access. Demonstrates +`fixed_string` NTTP dispatch and the `static_assert` on wrong names. + +```cpp +auto dbl = make_node, out<"result">>(5); +// ... +.connect("src", src.output<0>(), "dbl", dbl.input<"value">()) +.connect("dbl", dbl.output<"result">(), "sink", sink.input<0>()) +``` + +### `examples/03_multi_output` — Tuple-returning node / sub-port routing + +A single node returns `std::tuple`. Each output is routed to a different +downstream node. Demonstrates tuple normalisation, per-element channel push, and +`output<1>()` sub-indexing. + +```cpp +std::tuple detect(Image in) { ... } +void show_image(const Image& img) { ... } +void save_mask(const Mask& m) { ... } + +// detect:output<0> → show_image, detect:output<1> → save_mask +``` + +### `examples/04_storage_policy` — `channel_storage_policy` specialisation + +Shows the default behaviour (large struct stored as `shared_ptr`, small int by +value) and a user specialisation that overrides the default for a custom type. + +```cpp +struct BigFrame { uint8_t pixels[1920*1080*3]; }; +// stored as shared_ptr automatically + +struct Tiny { float x, y; }; // 8 bytes — by value by default +template<> struct channel_storage_policy { static constexpr bool by_value = true; }; +``` + +### `examples/05_error_handling` — Overflow and node exceptions + +Demonstrates `ChannelOverflowError` (producer faster than consumer, tiny FIFO), custom +`ErrorHandler`, and a node that throws mid-execution. + +```cpp +net.set_error_handler([](std::string_view name, std::exception_ptr ep) { + try { std::rethrow_exception(ep); } + catch (const std::exception& e) { + std::cerr << "[" << name << "] " << e.what() << '\n'; + } +}); +``` + +### `examples/06_watchdog` — Orchestrator / watchdog + +A node that artificially stalls. Shows watchdog warning emission, configurable interval, +and graceful shutdown after a timeout. + +```cpp +net.set_watchdog_interval(std::chrono::milliseconds(200)); +// stall_node sleeps for 2s per item — watchdog fires warning after 200ms +``` + +### `examples/07_python_network` — PyNetwork with C++ and Python nodes + +Python script that imports `kpn`, registers C++ node types via `make_py_network`, adds a +pure Python processing node, connects them, and runs the graph. + +```python +import kpn + +net = kpn.make_network([kpn.BlurNode, kpn.DetectNode]) + +def py_filter(img): + return img[::2, ::2] # downsample in Python + +net.add_node("blur", kpn.BlurNode, inputs=["img"]) +net.add_node("downsample",py_filter, inputs=["img"], outputs=["img"]) +net.add_node("detect", kpn.DetectNode, inputs=["img"]) +net.connect("blur", 0, "downsample", 0) +net.connect("downsample", 0, "detect", 0) +net.start() +``` + +### `examples/09_opencv_cellshade` — Real-time cell-shading with OpenCV (optional) + +Captures live video from a system camera and applies a cell-shading effect entirely inside +a KPN++ graph. Built only when OpenCV is found at CMake time; skipped silently otherwise. + +**Graph topology:** + +``` +[capture] ──Mat──> [split] ──B──> [median_b] ──B──┐ + ├──G──> [median_g] ──G──┤ + └──R──> [median_r] ──R──┴──> [merge] ──Mat──> [combine] ──> [display] +[capture] ──Mat──────────────────────> [detect_edges] ──mask──────────> [combine] +``` + +Effect steps: +1. **`split_channels`** — `cv::split` into three single-channel `cv::Mat` planes. +2. **`median_b/g/r`** — independent `cv::medianBlur(kernel=15)` per channel; large kernel + posterises colours into flat cartoon-like regions and runs in parallel across channels. +3. **`merge_channels`** — `cv::merge` back to BGR. +4. **`detect_edges`** — greyscale, `cv::Canny`, then `cv::dilate` to produce thick outlines. +5. **`combine`** — zeros out BGR pixels wherever the edge mask is non-zero → black outlines + drawn over the flat-colour image. +6. **`display`** — `cv::imshow`; ESC key signals shutdown via `g_running` atomic. + +Demonstrates: named ports, fan-out from a single node to two downstream paths, parallel +per-channel processing, multi-input `combine` node, and error handler driving graceful stop. + +```cpp +// Build only if OpenCV is present: +// cmake .. -DKPN_BUILD_EXAMPLES=ON +// ./09_opencv_cellshade [camera_index] # default: 0 +``` + +### `examples/08_python_subport` — Python sub-value tap and inject + +Shows `net.read("node", output=N)` and `net.write("node", input=N, value=v)` from Python, +plus connecting a C++ tuple output sub-port directly to a Python node input. + +```python +# Tap only output<1> (Mask) of a C++ detect node into Python +net.connect("detect", 1, "py_thresh", 0) +val = net.read("detect", output=0) # blocks until Image is available +net.write("blur", input=1, value=1.5) # inject sigma +``` + +--- + +## Component 8 — Web Debug UI (optional, compile-time toggle) + +An optional in-process HTTP server that serves a live graph visualisation of the running +network. Zero cost when disabled — no symbols compiled in, no headers pulled. + +### Toggle + +```cpp +// Before any kpn include — enables the web debug server +#define KPN_WEB_DEBUG 1 +#include +``` + +CMake projects that want it globally: + +```cmake +option(KPN_WEB_DEBUG "Enable KPN++ web debug UI" OFF) +if(KPN_WEB_DEBUG) + target_compile_definitions(my_app PRIVATE KPN_WEB_DEBUG=1) + # cpp-httplib is fetched automatically by CMake when this flag is ON +endif() +``` + +### Implementation + +`include/kpn/web_debug.hpp` — included by `network.hpp` only when `KPN_WEB_DEBUG` is defined. + +Depends on **cpp-httplib** (single-header, no external process, no Python required). +Served on `localhost:9090` by default (configurable via `net.set_web_debug_port(uint16_t)`). + +When enabled, `Network` gains: + +```cpp +#ifdef KPN_WEB_DEBUG +void set_web_debug_port(uint16_t port); // default 9090 +void start_web_debug(); // called internally by start() +void stop_web_debug(); // called internally by stop() +#endif +``` + +`start()` automatically calls `start_web_debug()` when `KPN_WEB_DEBUG` is defined. + +### Endpoints + +| Endpoint | Method | Description | +|---|---|---| +| `/` | GET | Serves the single-page HTML app (inline, no files needed) | +| `/api/snapshot` | GET | Returns a JSON snapshot of all node and channel stats | + +The HTML page is embedded as a C++ string literal — no asset files to deploy. + +### JSON Snapshot Format + +```json +{ + "nodes": [ + { "id": "src", "frames": 120, "ema_exec_ms": 33.2, "max_exec_ms": 45.1, + "blocked_ms": 0.1, "fps": 29.8 }, + { "id": "quant", "frames": 120, "ema_exec_ms": 4.1, ... } + ], + "edges": [ + { "source": "src", "target": "quant", "label": "colour", + "fill_pct": 12.5, "peak_pct": 87.5, "capacity": 8, "current": 1, + "pushes": 120, "drops": 0, "overflows": 0 } + ] +} +``` + +Node `id` comes from the name registered via `net.add("name", node)`. +Edge `label` comes from the channel name registered via `connect()` (format: `"src:N → dst:M"`). + +### Browser UI + +The page polls `/api/snapshot` every **500 ms** and renders a **D3.js v7 force-directed +graph**: + +- **Nodes** — circles labelled with node name; colour encodes exec load: + - green (`ema_exec_ms` < 10ms), yellow (10–50ms), orange (50–100ms), red (>100ms) + - hover tooltip shows: frames, ema_exec_ms, max_exec_ms, blocked_ms, fps +- **Edges** — directed arrows labelled with the channel name and fill%; colour: + - green (fill < 50%), yellow (50–80%), red (≥80%) — matches the `<<<` flag in the text report + - hover tooltip shows: pushes, drops, overflows, capacity + +D3 is loaded from CDN (`d3js.org`). The entire UI is a single inline HTML string in +`web_debug.hpp` — no file serving, no build step for assets. + +### Thread model + +`start_web_debug()` launches a `std::jthread` running `httplib::Server::listen()`. +The server is stopped via `httplib::Server::stop()` called from `stop_web_debug()`. +`/api/snapshot` calls `collect_snapshots()` (already thread-safe — reads atomics with +relaxed ordering) and serialises to JSON using a minimal hand-rolled serialiser +(no third-party JSON library required). + +### Example usage + +```cpp +#define KPN_WEB_DEBUG 1 +#include + +// ... build network as normal ... +net.set_web_debug_port(9090); // optional, 9090 is the default +net.start(); +// Open http://localhost:9090 in a browser +``` + +--- + +## CMake Layout + +| Target | Type | Notes | +|---|---|---| +| `kpn` | header-only interface library | C++20, no external deps | +| `kpn_python` | nanobind shared library | links `kpn`, requires Python 3.8+ | +| `kpn_tests` | executable | Catch2 v3 + Google Test | +| `kpn_examples` | executables (one per example) | built by default, off with `-DKPN_EXAMPLES=OFF` | +| `kpn_web_debug` | compile-time option | `#define KPN_WEB_DEBUG 1`; fetches cpp-httplib via CMake FetchContent | + +--- + +## Resolved Design Decisions + +All major design questions are now closed: + +| Question | Decision | +|---|---| +| Shutdown mechanism | `accepting_` flag per channel; `disable()` clears queue and unblocks `pop()` | +| Overflow behaviour | `ChannelOverflowError` thrown on full accepting channel; silently dropped on disabled channel | +| Network ownership | Non-owning; user declares nodes, network holds raw pointers | +| Node lifetime contract | Nodes must outlive their `Network`; declare in same scope | +| Sub-networks | `Network` implements `INode`; `expose_input`/`expose_output` define boundary ports | +| `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 | diff --git a/examples/01_hello_pipeline/main.cpp b/examples/01_hello_pipeline/main.cpp new file mode 100644 index 0000000..d79972e --- /dev/null +++ b/examples/01_hello_pipeline/main.cpp @@ -0,0 +1,38 @@ +#include +#include +#include +#include + +// A minimal linear pipeline: producer → double → print +// +// [produce] --int--> [double_it] --int--> [print_it] + +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); + + // Wire channels + auto& dbl_in = dbl.input_channel<0>(); + auto& sink_in = sink.input_channel<0>(); + src.set_output_channel<0>(&dbl_in); + dbl.set_output_channel<0>(&sink_in); + + Network net; + net.add("src", src) + .add("dbl", dbl) + .add("sink", sink) + .connect("src", src.output<0>(), "dbl", dbl.input<0>()) + .connect("dbl", dbl.output<0>(), "sink", sink.input<0>()) + .build(); + + net.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + net.stop(); +} diff --git a/examples/02_named_ports/main.cpp b/examples/02_named_ports/main.cpp new file mode 100644 index 0000000..8f089b2 --- /dev/null +++ b/examples/02_named_ports/main.cpp @@ -0,0 +1,3 @@ +#include +// TODO: implement example +int main() { return 0; } diff --git a/examples/03_multi_output/main.cpp b/examples/03_multi_output/main.cpp new file mode 100644 index 0000000..8f089b2 --- /dev/null +++ b/examples/03_multi_output/main.cpp @@ -0,0 +1,3 @@ +#include +// TODO: implement example +int main() { return 0; } diff --git a/examples/04_storage_policy/main.cpp b/examples/04_storage_policy/main.cpp new file mode 100644 index 0000000..8f089b2 --- /dev/null +++ b/examples/04_storage_policy/main.cpp @@ -0,0 +1,3 @@ +#include +// TODO: implement example +int main() { return 0; } diff --git a/examples/05_error_handling/main.cpp b/examples/05_error_handling/main.cpp new file mode 100644 index 0000000..8f089b2 --- /dev/null +++ b/examples/05_error_handling/main.cpp @@ -0,0 +1,3 @@ +#include +// TODO: implement example +int main() { return 0; } diff --git a/examples/06_watchdog/main.cpp b/examples/06_watchdog/main.cpp new file mode 100644 index 0000000..8f089b2 --- /dev/null +++ b/examples/06_watchdog/main.cpp @@ -0,0 +1,3 @@ +#include +// TODO: implement example +int main() { return 0; } diff --git a/examples/07_python_network/example.py b/examples/07_python_network/example.py new file mode 100644 index 0000000..0eee84e --- /dev/null +++ b/examples/07_python_network/example.py @@ -0,0 +1,32 @@ +""" +07_python_network — hello pipeline with a Python node in the middle. + +Graph: + [ProduceNode] --int--> [py_double] --int--> [PrintItNode] + +ProduceNode and PrintItNode are C++ nodes wrapped in VariantNodeWrapper. +py_double is a pure Python callable — doubles its input using Python arithmetic. +""" + +import sys +import time +sys.path.insert(0, "build/python") + +import kpn_python as kpn + +def py_double(x: int) -> int: + return x * 2 + +net = kpn.Network() + +net.add("src", kpn.make_produce()) +net.add_node("dbl", py_double, inputs=["int"], outputs=["int"]) +net.add("sink", kpn.make_print_it()) + +net.connect("src", 0, "dbl", 0) +net.connect("dbl", 0, "sink", 0) + +net.build() +net.start() +time.sleep(0.1) +net.stop() diff --git a/examples/08_python_subport/example.py b/examples/08_python_subport/example.py new file mode 100644 index 0000000..1351b5e --- /dev/null +++ b/examples/08_python_subport/example.py @@ -0,0 +1,38 @@ +""" +08_python_subport — tap a C++ node's output from Python using net.read(). + +Graph: + [ProduceNode] --int--> [DoubleItNode] --int--> (tapped by net.read()) + +The sink is Python: instead of connecting a PrintItNode, we call net.read() +to pull values out of DoubleItNode's output directly into Python. +We also demonstrate net.write() by injecting a value into DoubleItNode's input. +""" + +import sys +import time +import threading +sys.path.insert(0, "build/python") + +import kpn_python as kpn + +net = kpn.Network() + +net.add("src", kpn.make_produce()) +net.add("dbl", kpn.make_double_it()) + +net.connect("src", 0, "dbl", 0) +net.build() +net.start() + +# Collect a few values from DoubleItNode's output via Python tap +results = [] +for _ in range(5): + val = net.read("dbl", 0) + results.append(val) + +net.stop() + +print("values read from C++ DoubleItNode output:", results) +assert all(v == 84 for v in results), f"expected all 84, got {results}" +print("all correct (42 * 2 = 84)") diff --git a/examples/09_opencv_cellshade/bench_capture.cpp b/examples/09_opencv_cellshade/bench_capture.cpp new file mode 100644 index 0000000..538fd48 --- /dev/null +++ b/examples/09_opencv_cellshade/bench_capture.cpp @@ -0,0 +1,53 @@ +#include +#include +#include +#include +#include +#include + +static void bench(const char* name, int device, int backend, int W, int H, int frames) { + cv::VideoCapture cap(device, backend); + if (!cap.isOpened()) { + std::printf("%-12s failed to open /dev/video%d\n", name, device); + return; + } + cap.set(cv::CAP_PROP_FRAME_WIDTH, W); + cap.set(cv::CAP_PROP_FRAME_HEIGHT, H); + + // Warm up + for (int i = 0; i < 5; ++i) cap.grab(); + + std::vector times; + times.reserve(frames); + for (int i = 0; i < frames; ++i) { + auto t0 = std::chrono::steady_clock::now(); + bool ok = cap.grab(); + double ms = std::chrono::duration( + std::chrono::steady_clock::now() - t0).count(); + if (ok) times.push_back(ms); + } + cap.release(); + + if (times.empty()) { + std::printf("%-12s no frames captured\n", name); + return; + } + std::sort(times.begin(), times.end()); + double avg = std::accumulate(times.begin(), times.end(), 0.0) / times.size(); + double mn = times.front(); + double mx = times.back(); + double p95 = times[times.size() * 95 / 100]; + std::printf("%-12s avg=%6.1fms min=%5.1fms p95=%6.1fms max=%6.1fms (%zu frames)\n", + name, avg, mn, p95, mx, times.size()); +} + +int main(int argc, char** argv) { + int device = argc > 1 ? std::atoi(argv[1]) : 0; + int frames = argc > 2 ? std::atoi(argv[2]) : 60; + int W = 1920, H = 1080; + + std::printf("Benchmarking /dev/video%d at %dx%d, %d frames each\n\n", device, W, H, frames); + bench("V4L2", device, cv::CAP_V4L2, W, H, frames); + bench("GStreamer", device, cv::CAP_GSTREAMER, W, H, frames); + return 0; +} diff --git a/examples/09_opencv_cellshade/main.cpp b/examples/09_opencv_cellshade/main.cpp new file mode 100644 index 0000000..5884f5b --- /dev/null +++ b/examples/09_opencv_cellshade/main.cpp @@ -0,0 +1,190 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// ── Cell-shading pipeline ───────────────────────────────────────────────────── +// +// [capture] --"colour"--> [quantise] ──────────────────────────┐ +// ├──> [composite] ──"result"──┐ +// [capture] --"grey"---> [to_gray] --> [edges] ──"edges"───────┘ │ +// └──────────────────────────────"edges"──────────┴──> [display] +// +// DisplayNode derives from MainThreadNode<> — two inputs (composite + raw edges), +// two windows opened in the constructor. step() is called on the main thread. + +// ── 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; +} + +// Returns both the composite frame and the original edge mask so downstream +// nodes (display) can receive both without fan-out on the edges channel. +static std::tuple composite(cv::Mat edge_mask, cv::Mat colour) { + cv::Mat result = colour.clone(); + result.setTo(cv::Scalar(0, 0, 0), edge_mask); + return {result, edge_mask}; +} + +// ── DisplayNode ─────────────────────────────────────────────────────────────── +// +// Two inputs: the cell-shaded composite frame and the raw edge mask. +// MainThreadNode<> provides channel ownership, INode boilerplate, and stats. +// The constructor opens both windows on the main thread (Wayland requirement). +// operator() is called by step() whenever both channels have a frame ready. + +class DisplayNode : public kpn::MainThreadNode, + cv::Mat, cv::Mat> { +public: + DisplayNode() : MainThreadNode(8) { + cv::namedWindow("Cell Shade", cv::WINDOW_NORMAL); + cv::namedWindow("Edge Mask", cv::WINDOW_NORMAL); + cv::resizeWindow("Cell Shade", 1280, 720); + cv::resizeWindow("Edge Mask", 640, 360); + } + + ~DisplayNode() { cv::destroyAllWindows(); } + + bool operator()(cv::Mat composite, cv::Mat edges) { + cv::imshow("Cell Shade", composite); + cv::Mat edges_bgr; + cv::cvtColor(edges, edges_bgr, cv::COLOR_GRAY2BGR); + cv::imshow("Edge Mask", edges_bgr); + int key = cv::waitKey(1); + if (key == 'q' || key == 27) return false; + return window_open("Cell Shade") && window_open("Edge Mask"); + } + +private: + static bool window_open(const char* name) { + try { return cv::getWindowProperty(name, cv::WND_PROP_VISIBLE) >= 1; } + catch (const cv::Exception&) { return false; } + } +}; + +// ───────────────────────────────────────────────────────────────────────────── + +int main() { + using namespace kpn; + + auto src = make_node (out<"colour","grey">{}, 8); + auto gray_node = make_node (in<"bgr">{}, out<"gray">{}, 8); + auto edge_node = make_node (in<"gray">{}, out<"edges">{}, 8); + auto quant = make_node (in<"bgr">{}, out<"quantised">{}, 8); + auto comp = make_node(in<"edges","colour">{}, out<"result","edges">{}, 8); + + // DisplayNode: two windows opened in constructor, step() drives main thread. + DisplayNode disp; + + Network net; + net.add("src", src) + .add("gray", gray_node) + .add("edges", edge_node) + .add("quant", quant) + .add("comp", comp) + .add("display", disp) + .connect("src", src.template output<"colour">(), "quant", quant.template input<"bgr">()) + .connect("quant", quant.template output<"quantised">(), "comp", comp.template input<"colour">()) + .connect("src", src.template output<"grey">(), "gray", gray_node.template input<"bgr">()) + .connect("gray", gray_node.template output<"gray">(), "edges", edge_node.template input<"gray">()) + .connect("edges", edge_node.template output<"edges">(), "comp", comp.template input<"edges">()) + .connect("comp", comp.template output<"result">(), "display", disp.template input<"composite">()) + .connect("comp", comp.template output<"edges">(), "display", disp.template input<"edges">()) + .build(); + + net.set_watchdog_interval(std::chrono::milliseconds(5000)); + net.set_web_debug_port(9090); + + std::cout << "Cell-shading pipeline running. Press 'q' to stop.\n"; + std::cout << "Web debug UI: http://localhost:9090\n"; + + net.start(); + + // Main thread drives display — imshow/waitKey stay on the GUI thread. + // step() returns false when operator() returns false (q pressed / window closed). + while (disp.step()) + cv::waitKey(8); // yield event loop when no frame ready + + net.stop(); + return 0; +} diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt new file mode 100644 index 0000000..287bb64 --- /dev/null +++ b/examples/CMakeLists.txt @@ -0,0 +1,30 @@ +cmake_minimum_required(VERSION 3.21) + +function(kpn_example name) + add_executable(${name} ${name}/main.cpp) + target_link_libraries(${name} PRIVATE kpn) +endfunction() + +kpn_example(01_hello_pipeline) +kpn_example(02_named_ports) +kpn_example(03_multi_output) +kpn_example(04_storage_policy) +kpn_example(05_error_handling) +kpn_example(06_watchdog) +# 07 and 08 require the Python bindings — only add if built +if(KPN_BUILD_PYTHON) + # These are Python scripts, not compiled targets — installed alongside kpn_python +endif() + +# 09 requires OpenCV — only build if found +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}) + if(KPN_WEB_DEBUG) + kpn_target_enable_web_debug(09_opencv_cellshade) + endif() + message(STATUS "KPN++ example 09_opencv_cellshade: OpenCV ${OpenCV_VERSION} found — building") +else() + message(STATUS "KPN++ example 09_opencv_cellshade: OpenCV not found — skipping") +endif() diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp new file mode 100644 index 0000000..23f57f1 --- /dev/null +++ b/include/kpn/channel.hpp @@ -0,0 +1,175 @@ +#pragma once +#include "diagnostics.hpp" +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace kpn { + +// ── Storage policy ──────────────────────────────────────────────────────────── + +template +struct channel_storage_policy { + static constexpr bool by_value = + std::is_trivially_copyable_v && sizeof(T) <= 8; +}; + +template +using channel_storage_t = std::conditional_t< + channel_storage_policy::by_value, + T, + std::shared_ptr +>; + +// ── Exceptions ──────────────────────────────────────────────────────────────── + +class ChannelOverflowError : public std::runtime_error { +public: + explicit ChannelOverflowError(std::size_t capacity) + : std::runtime_error("channel overflow: capacity " + std::to_string(capacity) + + " exceeded") {} + ChannelOverflowError(std::size_t capacity, std::string context) + : std::runtime_error(std::move(context) + ": capacity " + std::to_string(capacity) + + " exceeded") {} +}; + +class ChannelClosedError : public std::runtime_error { +public: + ChannelClosedError() : std::runtime_error("channel closed") {} +}; + +// ── Channel ─────────────────────────────────────────────────────────────────── + +template +class Channel { +public: + using storage_type = channel_storage_t; + + explicit Channel(std::size_t capacity = 5) : capacity_(capacity) {} + + Channel(const Channel&) = delete; + Channel& operator=(const Channel&) = delete; + + // Push a value. + // - If channel is disabled (accepting_ == false): silently drop, return immediately. + // - If channel is full: throw ChannelOverflowError. + void push(T value) { + if (!accepting_.load(std::memory_order_relaxed)) { + stats_.record_drop(); + return; + } + std::unique_lock lock(mutex_); + if (!accepting_.load(std::memory_order_relaxed)) { + stats_.record_drop(); + return; + } + if (queue_.size() >= capacity_) { + stats_.record_overflow(); + throw ChannelOverflowError(capacity_); + } + queue_.push(make_storage(std::move(value))); + stats_.record_push(queue_.size()); + lock.unlock(); + cv_.notify_one(); + } + + // Blocking pop. Unblocks when an item is available or the channel is disabled. + // Throws ChannelClosedError if disabled and queue is empty. + T pop() { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { + return !queue_.empty() || !accepting_.load(std::memory_order_relaxed); + }); + if (queue_.empty()) + throw ChannelClosedError{}; + T value = extract(std::move(queue_.front())); + queue_.pop(); + stats_.record_pop(); + return value; + } + + // Non-blocking pop with timeout. For watchdog/display use only — not used in run_loop. + bool try_pop(T& out, std::chrono::milliseconds timeout) { + std::unique_lock lock(mutex_); + if (!cv_.wait_for(lock, timeout, [this] { + return !queue_.empty() || !accepting_.load(std::memory_order_relaxed); + })) + return false; + if (queue_.empty()) + return false; + out = extract(std::move(queue_.front())); + queue_.pop(); + stats_.record_pop(); + return true; + } + + // Enable the channel (called by consumer node on start()). + void enable() { + accepting_.store(true, std::memory_order_relaxed); + } + + // Disable the channel: drop all queued items, unblock any waiting pop(). + // Called by consumer node on stop(). Producer push() will silently drop after this. + void disable() { + accepting_.store(false, std::memory_order_relaxed); + { + std::lock_guard lock(mutex_); + while (!queue_.empty()) queue_.pop(); + } + cv_.notify_all(); + } + + std::size_t size() const { + std::lock_guard lock(mutex_); + return queue_.size(); + } + + std::size_t capacity() const { return capacity_; } + bool is_accepting() const { return accepting_.load(std::memory_order_relaxed); } + const ChannelStats& stats() const { return stats_; } + + ChannelSnapshot snapshot(const std::string& name) const { + std::lock_guard lock(mutex_); + return { + name, + capacity_, + queue_.size(), + stats_.peak_fill.load(std::memory_order_relaxed), + stats_.pushes.load(std::memory_order_relaxed), + stats_.drops.load(std::memory_order_relaxed), + stats_.overflows.load(std::memory_order_relaxed), + stats_.pops.load(std::memory_order_relaxed), + sizeof(T), // payload bytes — sizeof(T) regardless of storage policy + }; + } + +private: + static storage_type make_storage(T&& v) { + if constexpr (channel_storage_policy::by_value) + return std::move(v); + else + return std::make_shared(std::move(v)); + } + + static T extract(storage_type&& s) { + if constexpr (channel_storage_policy::by_value) + return std::move(s); + else + return *s; + } + + const std::size_t capacity_; + std::queue queue_; + std::atomic accepting_{true}; + mutable std::mutex mutex_; + std::condition_variable cv_; + ChannelStats stats_; +}; + +} // namespace kpn diff --git a/include/kpn/diagnostics.hpp b/include/kpn/diagnostics.hpp new file mode 100644 index 0000000..e95fe16 --- /dev/null +++ b/include/kpn/diagnostics.hpp @@ -0,0 +1,137 @@ +#pragma once +#include +#include +#include +#include +#include +#include // clock_gettime, CLOCK_THREAD_CPUTIME_ID + +namespace kpn { + +using clock_t = std::chrono::steady_clock; +using duration_t = std::chrono::duration; // milliseconds + +// ── Per-channel statistics ──────────────────────────────────────────────────── + +struct ChannelStats { + std::atomic pushes{0}; + std::atomic drops{0}; + std::atomic overflows{0}; + std::atomic pops{0}; + std::atomic peak_fill{0}; + + ChannelStats() = default; + ChannelStats(const ChannelStats&) = delete; + ChannelStats& operator=(const ChannelStats&) = delete; + + void record_push(std::size_t current_fill) { + pushes.fetch_add(1, std::memory_order_relaxed); + std::size_t prev = peak_fill.load(std::memory_order_relaxed); + while (current_fill > prev && + !peak_fill.compare_exchange_weak(prev, current_fill, + std::memory_order_relaxed, std::memory_order_relaxed)) + ; + } + void record_drop() { drops.fetch_add(1, std::memory_order_relaxed); } + void record_overflow() { overflows.fetch_add(1, std::memory_order_relaxed); } + void record_pop() { pops.fetch_add(1, std::memory_order_relaxed); } +}; + +// ── Per-node statistics ─────────────────────────────────────────────────────── + +struct NodeStats { + std::atomic frames_processed{0}; + + // Wall-clock execution time EMA — warmup mean for first WARMUP_FRAMES, + // then EMA alpha=0.1. Stored as integer microseconds for atomic updates. + static constexpr int WARMUP_FRAMES = 5; + std::atomic ema_exec_us{0}; + std::atomic max_exec_us{0}; + std::atomic total_blocked_us{0}; + + // Thread CPU time — actual CPU consumed by this node's thread, + // measured via CLOCK_THREAD_CPUTIME_ID. Excludes time sleeping or + // blocked on mutexes/channels. Sampled once per frame. + std::atomic total_cpu_us{0}; // cumulative CPU µs consumed + + NodeStats() = default; + NodeStats(const NodeStats&) = delete; + NodeStats& operator=(const NodeStats&) = delete; + + // Call at the start of run_loop to capture thread CPU baseline. + // Returns the raw timespec for use in record_exec. + static struct timespec cpu_now() { + struct timespec ts{}; + clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts); + return ts; + } + + static int64_t timespec_us(const struct timespec& ts) { + return static_cast(ts.tv_sec) * 1'000'000 + + static_cast(ts.tv_nsec) / 1'000; + } + + void record_exec(duration_t exec_time, duration_t blocked_time, + const struct timespec& cpu_before, const struct timespec& cpu_after) { + frames_processed.fetch_add(1, std::memory_order_relaxed); + + int64_t us = static_cast(exec_time.count() * 1000.0); + + uint64_t n = frames_processed.load(std::memory_order_relaxed); + int64_t prev = ema_exec_us.load(std::memory_order_relaxed); + int64_t next = (n <= static_cast(WARMUP_FRAMES)) + ? prev + (us - prev) / static_cast(n) + : prev + (us - prev) / 10; + ema_exec_us.store(next, std::memory_order_relaxed); + + int64_t cur_max = max_exec_us.load(std::memory_order_relaxed); + if (us > cur_max) + max_exec_us.store(us, std::memory_order_relaxed); + + int64_t blocked_us = static_cast(blocked_time.count() * 1000.0); + total_blocked_us.fetch_add(blocked_us, std::memory_order_relaxed); + + int64_t cpu_delta = timespec_us(cpu_after) - timespec_us(cpu_before); + if (cpu_delta > 0) + total_cpu_us.fetch_add(cpu_delta, std::memory_order_relaxed); + } +}; + +// ── Snapshot for reporting (copyable, taken by watchdog) ───────────────────── + +struct ChannelSnapshot { + std::string name; + std::size_t capacity; + std::size_t current_fill; + std::size_t peak_fill; + uint64_t pushes; + uint64_t drops; + uint64_t overflows; + uint64_t pops; + std::size_t item_bytes; // sizeof(T) for the stored type — set by Channel + + double fill_pct() const { + return capacity ? 100.0 * current_fill / capacity : 0.0; + } + double peak_pct() const { + return capacity ? 100.0 * peak_fill / capacity : 0.0; + } + // Bandwidth in MB/s: bytes transferred / elapsed seconds + double bandwidth_mbs(double elapsed_s) const { + if (elapsed_s <= 0.0 || item_bytes == 0) return 0.0; + return static_cast(pushes * item_bytes) / elapsed_s / 1e6; + } +}; + +struct NodeSnapshot { + std::string name; + uint64_t frames_processed; + double ema_exec_ms; + double max_exec_ms; + double total_blocked_ms; + double throughput_fps; + double total_cpu_ms; // cumulative CPU time consumed by this node's thread + double cpu_util_pct; // exec_ms / (exec_ms + blocked_ms) * 100 +}; + +} // namespace kpn diff --git a/include/kpn/fixed_string.hpp b/include/kpn/fixed_string.hpp new file mode 100644 index 0000000..d6c90b5 --- /dev/null +++ b/include/kpn/fixed_string.hpp @@ -0,0 +1,46 @@ +#pragma once +#include +#include +#include + +namespace kpn { + +template +struct fixed_string { + char data[N]{}; + + constexpr fixed_string(const char (&s)[N]) { std::copy_n(s, N, data); } + constexpr bool operator==(const fixed_string&) const = default; + + template + constexpr bool operator==(const fixed_string&) const { return false; } + constexpr std::string_view view() const { return {data, N - 1}; } +}; + +template +fixed_string(const char (&)[N]) -> fixed_string; + +// ── Port name pack lookup ───────────────────────────────────────────────────── + +inline constexpr std::size_t npos = std::size_t(-1); + +template +constexpr std::size_t index_of() { + std::size_t i = 0; + bool found = false; + auto check = [&](auto n) { + if (!found) { + if (Name == n) found = true; + else ++i; + } + }; + (check(Names), ...); + return found ? i : npos; +} + +// ── in<> / out<> name tag structs ───────────────────────────────────────────── + +template struct in {}; +template struct out {}; + +} // namespace kpn diff --git a/include/kpn/kpn.hpp b/include/kpn/kpn.hpp new file mode 100644 index 0000000..6ac3309 --- /dev/null +++ b/include/kpn/kpn.hpp @@ -0,0 +1,10 @@ +#pragma once +// Convenience umbrella header — include this to get the full C++ API. + +#include "fixed_string.hpp" +#include "traits.hpp" +#include "channel.hpp" +#include "port.hpp" +#include "node.hpp" +#include "main_thread_node.hpp" +#include "network.hpp" diff --git a/include/kpn/main_thread_node.hpp b/include/kpn/main_thread_node.hpp new file mode 100644 index 0000000..852340c --- /dev/null +++ b/include/kpn/main_thread_node.hpp @@ -0,0 +1,186 @@ +#pragma once +#include "channel.hpp" +#include "diagnostics.hpp" +#include "fixed_string.hpp" +#include "node.hpp" // INode +#include "port.hpp" + +#include +#include +#include +#include +#include +#include + +namespace kpn { + +// ── MainThreadNode ──────────────────────────────────────────────────────────── +// +// Base class for nodes that must run on the main thread (e.g. OpenCV display +// on Wayland/Qt). Registered as a normal INode in the Network so it appears +// in diagnostics and the web UI, but spawns no thread. +// +// Usage: +// class MyDisplay : public kpn::MainThreadNode, TypeA, TypeB> { +// public: +// MyDisplay(...) { /* constructor runs on main thread */ } +// bool operator()(TypeA a, TypeB b) { ...; return true; /* false = stop */ } +// }; +// +// MyDisplay disp(...); +// net.add("display", disp).connect(...).build(); +// net.start(); +// while (disp.step()) ; // drives the event loop on the main thread +// net.stop(); +// +// step() behaviour: +// - try_pop on every input channel with zero timeout +// - if all inputs have data: calls operator(), records stats, returns its result +// - if any input is missing: returns true immediately (caller should yield/waitKey) + +template +class MainThreadNode; + +template +class MainThreadNode, Args...> : public INode { +public: + static constexpr std::size_t input_count = sizeof...(Args); + + static_assert( + sizeof...(InNames) == 0 || sizeof...(InNames) == input_count, + "MainThreadNode: name count must match input type count, or provide none" + ); + + using args_tuple = std::tuple; // required by Network::connect type check + + explicit MainThreadNode(std::size_t fifo_capacity = 8) { + init_channels(std::make_index_sequence{}, fifo_capacity); + } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { + enable_channels(std::make_index_sequence{}); + running_.store(true, std::memory_order_relaxed); + } + + void stop() override { + running_.store(false, std::memory_order_relaxed); + disable_channels(std::make_index_sequence{}); + } + + bool running() const override { + return running_.load(std::memory_order_relaxed); + } + + void set_name(std::string) override {} + + 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 (for Network::connect) ─────────────────────────────────── + + template + Channel>>& input_channel() { + return *std::get(channels_); + } + + template + InputPort input() { + static_assert(I < input_count, "input index out of range"); + return {*this}; + } + + template + auto input() { + constexpr std::size_t idx = index_of(); + static_assert(idx != npos, "unknown input port name"); + return input(); + } + + // ── Main-thread driver ──────────────────────────────────────────────────── + + // Call this in a loop on the main thread instead of net.start()'s thread. + // Returns false when operator() returns false or all channels are closed. + bool step() { + if (!running_.load(std::memory_order_relaxed)) return false; + + auto t0 = clock_t::now(); + auto inputs = try_pop_all(std::make_index_sequence{}); + auto t1 = clock_t::now(); + + if (!inputs.has_value()) return true; // not all inputs ready — yield + + auto cpu0 = NodeStats::cpu_now(); + bool cont = std::apply( + [this](Args&&... a) { + return static_cast(this)->operator()(std::forward(a)...); + }, + std::move(*inputs)); + auto cpu1 = NodeStats::cpu_now(); + auto t2 = clock_t::now(); + + stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); + return cont; + } + +private: + template + void init_channels(std::index_sequence, std::size_t cap) { + ((std::get(channels_) = + std::make_unique>>(cap)), ...); + } + + template + void enable_channels(std::index_sequence) { + (std::get(channels_)->enable(), ...); + } + + template + void disable_channels(std::index_sequence) { + (std::get(channels_)->disable(), ...); + } + + // Try to pop one item from every channel with zero timeout. + // Returns nullopt if any channel has no data ready. + template + std::optional try_pop_all(std::index_sequence) { + args_tuple result; + bool all_ready = true; + // Use a fold that short-circuits on first missing item + ((all_ready = all_ready && + std::get(channels_)->try_pop( + std::get(result), std::chrono::milliseconds(0))), ...); + if (!all_ready) return std::nullopt; + return result; + } + + // Build the channel tuple type + template + static auto make_channel_tuple(std::index_sequence) + -> std::tuple>>...>; + + using channels_t = decltype(make_channel_tuple( + std::make_index_sequence{})); + + channels_t channels_; + std::atomic running_{false}; + NodeStats stats_; +}; + +} // namespace kpn diff --git a/include/kpn/network.hpp b/include/kpn/network.hpp new file mode 100644 index 0000000..252dabb --- /dev/null +++ b/include/kpn/network.hpp @@ -0,0 +1,336 @@ +#pragma once +#include "diagnostics.hpp" +#include "node.hpp" +#include "port.hpp" + +#ifdef KPN_WEB_DEBUG +#include "web_debug.hpp" +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace kpn { + +// ── Exceptions ──────────────────────────────────────────────────────────────── + +class NetworkCycleError : public std::runtime_error { +public: + NetworkCycleError() : std::runtime_error("network graph contains a directed cycle") {} +}; + +class NetworkBuildError : public std::runtime_error { +public: + explicit NetworkBuildError(std::string msg) + : std::runtime_error(std::move(msg)) {} +}; + +// ── Channel snapshot accessor — type-erased so Network can collect them ─────── + +struct IChannelProbe { + virtual ~IChannelProbe() = default; + virtual ChannelSnapshot snapshot() const = 0; +}; + +template +struct ChannelProbe : IChannelProbe { + const Channel& ch; + std::string name; + ChannelProbe(const Channel& c, std::string n) : ch(c), name(std::move(n)) {} + ChannelSnapshot snapshot() const override { return ch.snapshot(name); } +}; + +// ── Network ─────────────────────────────────────────────────────────────────── + +class Network : public INode { +public: + using ErrorHandler = + std::function; + using DiagnosticsHandler = + std::function&, + const std::vector&)>; + + // ── Builder API ─────────────────────────────────────────────────────────── + + template + Network& add(std::string name, NodeT& node) { + if (nodes_.count(name)) + throw NetworkBuildError("duplicate node name: " + name); + node.set_name(name); + nodes_.emplace(name, &node); + adj_[name]; + return *this; + } + + template + Network& connect(const std::string& src_name, + OutputPort, + const std::string& dst_name, + InputPort) { + using out_t = std::tuple_element_t; + using dst_in_t = std::tuple_element_t; + static_assert(std::is_same_v, + "connect: output type does not match input type"); + + auto* src = dynamic_cast(nodes_.at(src_name)); + auto* dst = dynamic_cast(nodes_.at(dst_name)); + if (!src) throw NetworkBuildError("node '" + src_name + "' type mismatch"); + if (!dst) throw NetworkBuildError("node '" + dst_name + "' type mismatch"); + + auto& in_ch = dst->template input_channel(); + src->template set_output_channel(&in_ch); + + // Register channel probe for diagnostics + std::string ch_name = src_name + ":" + std::to_string(SrcIdx) + + " → " + dst_name + ":" + std::to_string(DstIdx); + channel_probes_.push_back( + std::make_unique>(in_ch, ch_name)); + + adj_[src_name].push_back(dst_name); + return *this; + } + + template + Network& expose_input(std::string boundary_name, InputPort) { + exposed_inputs_[boundary_name] = boundary_name; + return *this; + } + + template + Network& expose_output(std::string boundary_name, OutputPort) { + exposed_outputs_[boundary_name] = boundary_name; + return *this; + } + + Network& build() { + topo_.clear(); + std::map color; + for (auto& [name, _] : nodes_) + if (color[name] == 0) + dfs(name, color); + return *this; + } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { + start_time_ = clock_t::now(); + for (auto& name : topo_) + nodes_.at(name)->start(); + start_watchdog(); +#ifdef KPN_WEB_DEBUG + web_server_ = std::make_unique( + web_debug_port_, + [this]() { + auto s = collect_snapshots(); + return web_debug::to_json(s.nodes, s.channels, s.elapsed_s); + }); + web_server_->start(); + std::cerr << "[kpn] web debug UI: http://localhost:" << web_debug_port_ << "\n"; +#endif + } + + void stop() override { +#ifdef KPN_WEB_DEBUG + if (web_server_) web_server_->stop(); +#endif + stop_watchdog(); + for (auto it = topo_.rbegin(); it != topo_.rend(); ++it) + nodes_.at(*it)->stop(); + } + + bool running() const override { return watchdog_.joinable(); } + void set_name(std::string) override {} + + const NodeStats& stats() const override { + static NodeStats dummy; + return dummy; + } + + NodeSnapshot node_snapshot(const std::string& name, double) const override { + return {name, 0, 0, 0, 0, 0, 0, 0}; + } + + // ── Configuration ───────────────────────────────────────────────────────── + + void set_watchdog_interval(std::chrono::milliseconds interval) { + watchdog_interval_ = interval; + } + + void set_error_handler(ErrorHandler h) { error_handler_ = std::move(h); } + void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_ = std::move(h); } + +#ifdef KPN_WEB_DEBUG + void set_web_debug_port(uint16_t port) { web_debug_port_ = port; } +#endif + + // Print a diagnostics report to a stream (default: stderr). + // Can be called at any time; thread-safe (reads atomics with relaxed ordering). + void print_diagnostics(std::ostream& os = std::cerr) const { + auto s = collect_snapshots(); + os << format_report(s.nodes, s.channels, s.elapsed_s); + } + +private: + // ── Diagnostics collection ──────────────────────────────────────────────── + + struct Snapshots { + std::vector nodes; + std::vector channels; + double elapsed_s; + }; + + Snapshots collect_snapshots() const { + double elapsed_s = std::chrono::duration( + clock_t::now() - start_time_).count(); + + std::vector nodes; + for (auto& name : topo_) + nodes.push_back(nodes_.at(name)->node_snapshot(name, elapsed_s)); + + std::vector channels; + for (auto& probe : channel_probes_) + channels.push_back(probe->snapshot()); + + return {std::move(nodes), std::move(channels), elapsed_s}; + } + + static std::string format_report(const std::vector& nodes, + const std::vector& channels, + double elapsed_s = 0.0) { + std::ostringstream os; + os << std::fixed << std::setprecision(1); + + os << "\n┌─ KPN++ Diagnostics ────────────────────────────────────────────────────────\n"; + + // Node table + os << "│ Nodes:\n"; + os << "│ " << std::left + << std::setw(16) << "name" + << std::setw(10) << "frames" + << std::setw(12) << "exec ms" + << std::setw(12) << "max ms" + << std::setw(14) << "blocked ms" + << std::setw(10) << "fps" + << std::setw(12) << "cpu ms" + << std::setw(10) << "util%" + << "\n│ " << std::string(92, '-') << "\n"; + for (auto& n : nodes) { + os << "│ " << std::left + << std::setw(16) << n.name + << std::setw(10) << n.frames_processed + << std::setw(12) << n.ema_exec_ms + << std::setw(12) << n.max_exec_ms + << std::setw(14) << n.total_blocked_ms + << std::setw(10) << n.throughput_fps + << std::setw(12) << n.total_cpu_ms + << std::setw(10) << n.cpu_util_pct + << "\n"; + } + + // Channel table + os << "│\n│ Channels:\n"; + os << "│ " << std::left + << std::setw(40) << "edge" + << std::setw(8) << "fill%" + << std::setw(8) << "peak%" + << std::setw(8) << "pushes" + << std::setw(8) << "drops" + << std::setw(8) << "oflow" + << std::setw(12) << "MB/s" + << std::setw(10) << "item B" + << "\n│ " << std::string(102, '-') << "\n"; + for (auto& c : channels) { + std::string flag = c.fill_pct() >= 80.0 ? " <<<" : + c.peak_pct() >= 80.0 ? " (peak)" : ""; + os << "│ " << std::left + << std::setw(40) << c.name + << std::setw(8) << c.fill_pct() + << std::setw(8) << c.peak_pct() + << std::setw(8) << c.pushes + << std::setw(8) << c.drops + << std::setw(8) << c.overflows + << std::setw(12) << c.bandwidth_mbs(elapsed_s) + << std::setw(10) << c.item_bytes + << flag << "\n"; + } + + // Bottleneck hint: node with highest ema_exec_ms + if (!nodes.empty()) { + auto it = std::max_element(nodes.begin(), nodes.end(), + [](const NodeSnapshot& a, const NodeSnapshot& b) { + return a.ema_exec_ms < b.ema_exec_ms; + }); + os << "│\n│ Bottleneck hint: '" << it->name + << "' (avg exec " << it->ema_exec_ms << " ms)\n"; + } + os << "└────────────────────────────────────────────────────────────────\n"; + return os.str(); + } + + // ── Cycle detection / topological sort ─────────────────────────────────── + + void dfs(const std::string& name, std::map& color) { + color[name] = 1; + for (auto& nbr : adj_[name]) { + if (color[nbr] == 1) throw NetworkCycleError{}; + if (color[nbr] == 0) dfs(nbr, color); + } + color[name] = 2; + topo_.insert(topo_.begin(), name); + } + + // ── Watchdog ────────────────────────────────────────────────────────────── + + void start_watchdog() { + watchdog_ = std::jthread([this](std::stop_token tok) { + while (!tok.stop_requested()) { + std::this_thread::sleep_for(watchdog_interval_); + if (tok.stop_requested()) break; + + auto s = collect_snapshots(); + + if (diag_handler_) { + diag_handler_(s.nodes, s.channels); + } else { + std::cerr << format_report(s.nodes, s.channels, s.elapsed_s); + } + } + }); + } + + void stop_watchdog() { + if (watchdog_.joinable()) + watchdog_.request_stop(), watchdog_.join(); + } + + // ── State ───────────────────────────────────────────────────────────────── + + std::map nodes_; + std::map> adj_; + std::vector topo_; + std::map exposed_inputs_; + std::map exposed_outputs_; + std::vector> channel_probes_; + ErrorHandler error_handler_; + DiagnosticsHandler diag_handler_; + std::chrono::milliseconds watchdog_interval_{3000}; + std::jthread watchdog_; + clock_t::time_point start_time_; +#ifdef KPN_WEB_DEBUG + uint16_t web_debug_port_{9090}; + std::unique_ptr web_server_; +#endif +}; + +} // namespace kpn diff --git a/include/kpn/node.hpp b/include/kpn/node.hpp new file mode 100644 index 0000000..5213bf6 --- /dev/null +++ b/include/kpn/node.hpp @@ -0,0 +1,565 @@ +#pragma once +#include "channel.hpp" +#include "diagnostics.hpp" +#include "fixed_string.hpp" +#include "port.hpp" +#include "traits.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace kpn { + +// ── INode — type-erased interface for Network / watchdog ───────────────────── + +struct INode { + virtual ~INode() = default; + virtual void start() = 0; + virtual void stop() = 0; + virtual bool running() const = 0; + virtual const NodeStats& stats() const = 0; + virtual NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const = 0; + virtual void set_name(std::string name) = 0; +}; + +// ── Node ───────────────────────────────────────────────────────────────────── +// +// Template parameters: +// Func — the wrapped function (auto NTTP, deduced as a function pointer) +// 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<>> +class Node; + +// Specialisation that unpacks the in<>/out<> tag packs +template +class Node, out> : public INode { +public: + using F = decltype(Func); + using args_tuple = args_t; + using return_raw = return_t; + using return_tuple = normalised_return_t; + + static constexpr std::size_t input_count = arity_v; + static constexpr std::size_t output_count = std::tuple_size_v; + + static_assert( + sizeof...(InNames) == 0 || sizeof...(InNames) == input_count, + "make_node: number of input names must match function arity, or provide none" + ); + static_assert( + sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count, + "make_node: number of output names must match return tuple size, or provide none" + ); + + explicit Node(std::size_t fifo_capacity = 5) + : fifo_capacity_(fifo_capacity) + { + init_input_channels(std::make_index_sequence{}); + } + + ~Node() override { stop(); } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { + enable_inputs(std::make_index_sequence{}); + 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); + // Disable all input channels: drops queued items and unblocks waiting pop() + disable_inputs(std::make_index_sequence{}); + 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 — by index ──────────────────────────────────────────────── + + template + InputPort input() { + static_assert(I < input_count, "input index out of range"); + return {*this}; + } + + template + OutputPort output() { + static_assert(I < output_count, "output index out of range"); + return {*this}; + } + + // ── Port access — by name ───────────────────────────────────────────────── + + template + auto input() { + constexpr std::size_t idx = index_of(); + static_assert(idx != npos, "unknown input port name"); + return input(); + } + + template + auto output() { + constexpr std::size_t idx = index_of(); + static_assert(idx != npos, "unknown output port name"); + return output(); + } + + // ── Internal channel accessors (used by Network at connect time) ────────── + + template + Channel>& input_channel() { + return *std::get(input_channels_); + } + + // Replace the owned input channel with an externally provided one. + // Used by VariantNodeWrapper to share a Channel with a VariantChannel adapter. + template + void set_input_channel( + std::shared_ptr>> ch) { + std::get(input_channels_) = std::move(ch); + } + + template + void set_output_channel( + Channel>* ch) { + std::get(output_channels_) = ch; + } + +private: + // ── Channel storage ─────────────────────────────────────────────────────── + + // Input channels — shared ownership so VariantChannel adapters can share them + template + void init_input_channels(std::index_sequence) { + ((std::get(input_channels_) = + std::make_shared>>(fifo_capacity_)), + ...); + } + + template + void enable_inputs(std::index_sequence) { + (std::get(input_channels_)->enable(), ...); + } + + template + void disable_inputs(std::index_sequence) { + (std::get(input_channels_)->disable(), ...); + } + + template + static auto make_input_channel_tuple(std::index_sequence) + -> std::tuple>>...>; + + using input_channels_t = decltype(make_input_channel_tuple( + std::make_index_sequence{})); + + // Output channels — non-owning pointers, set at connect time + template + static auto make_output_channel_tuple(std::index_sequence) + -> std::tuple>*...>; + + using output_channels_t = decltype(make_output_channel_tuple( + std::make_index_sequence{})); + + // ── run_loop ────────────────────────────────────────────────────────────── + + void run_loop() { + while (!stop_flag_.load(std::memory_order_relaxed)) { + try { + auto t0 = clock_t::now(); + auto args = pop_inputs(std::make_index_sequence{}); + auto t1 = clock_t::now(); + auto cpu0 = NodeStats::cpu_now(); + + if constexpr (std::is_void_v) { + std::apply(Func, args); + } else { + auto result = std::apply(Func, args); + push_outputs(normalise(std::move(result)), + std::make_index_sequence{}); + } + + 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] overflow: " << e.what() << "\n"; + } + } + } + + // Pop all inputs into a tuple of argument values + template + args_tuple pop_inputs(std::index_sequence) { + return {std::get(input_channels_)->pop()...}; + } + + // Normalise return value to tuple (handles void and single-value returns) + template + static return_tuple normalise(R&& r) { + if constexpr (is_tuple_v) + return std::move(r); + else + return std::make_tuple(std::move(r)); + } + + static return_tuple normalise_void() { return {}; } + + // Push each output element to its connected channel (if connected) + template + void push_outputs(return_tuple&& result, std::index_sequence) { + (push_one(std::get(std::move(result))), ...); + } + + template + void push_one(std::tuple_element_t&& val) { + auto* ch = std::get(output_channels_); + if (!ch) return; + try { + ch->push(std::move(val)); + } catch (const ChannelOverflowError&) { + throw ChannelOverflowError(ch->capacity(), "node '" + name_ + "' " + output_port_label()); + } + } + + template + static std::string output_port_label() { + if constexpr (sizeof...(OutNames) > 0) { + constexpr std::array names{OutNames.view()...}; + return std::string("output['") + std::string(names[I]) + "']"; + } else { + return "output[" + std::to_string(I) + "]"; + } + } + + // ── State ───────────────────────────────────────────────────────────────── + + std::string name_; + std::size_t fifo_capacity_; + input_channels_t input_channels_; + output_channels_t output_channels_{}; + std::atomic stop_flag_{false}; + std::jthread thread_; + NodeStats stats_; +}; + +// ── ObjectNode — wraps a callable object (functor / class with operator()) ──── +// +// Use this when the node needs state initialised in a constructor. +// The object must outlive the ObjectNode (stored by reference). +// +// Usage: +// MyFunctor obj(...); +// auto node = make_node(obj, in<"x">{}, out<"y">{}, capacity); + +template, typename OutputTag = out<>> +class ObjectNode; + +template +class ObjectNode, out> : 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::size_t input_count = arity_v; + static constexpr std::size_t output_count = std::tuple_size_v; + + static_assert( + sizeof...(InNames) == 0 || sizeof...(InNames) == input_count, + "make_node: number of input names must match operator() arity, or provide none" + ); + static_assert( + sizeof...(OutNames) == 0 || sizeof...(OutNames) == output_count, + "make_node: number of output names must match return tuple size, or provide none" + ); + + explicit ObjectNode(Obj& obj, std::size_t fifo_capacity = 5) + : obj_(obj), fifo_capacity_(fifo_capacity) + { + init_input_channels(std::make_index_sequence{}); + } + + ~ObjectNode() override { stop(); } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { + enable_inputs(std::make_index_sequence{}); + 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); + disable_inputs(std::make_index_sequence{}); + 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 < input_count, "input index out of range"); + return {*this}; + } + + template + OutputPort output() { + static_assert(I < output_count, "output index out of range"); + return {*this}; + } + + template + auto input() { + constexpr std::size_t idx = index_of(); + static_assert(idx != npos, "unknown input port name"); + return input(); + } + + template + auto output() { + constexpr std::size_t idx = index_of(); + static_assert(idx != npos, "unknown output port name"); + return output(); + } + + template + Channel>& input_channel() { + return *std::get(input_channels_); + } + + template + void set_input_channel( + std::shared_ptr>> ch) { + std::get(input_channels_) = std::move(ch); + } + + template + void set_output_channel(Channel>* ch) { + std::get(output_channels_) = ch; + } + +private: + template + void init_input_channels(std::index_sequence) { + ((std::get(input_channels_) = + std::make_shared>>(fifo_capacity_)), + ...); + } + + template + void enable_inputs(std::index_sequence) { + (std::get(input_channels_)->enable(), ...); + } + + template + void disable_inputs(std::index_sequence) { + (std::get(input_channels_)->disable(), ...); + } + + template + static auto make_input_channel_tuple(std::index_sequence) + -> std::tuple>>...>; + + using input_channels_t = decltype(make_input_channel_tuple( + std::make_index_sequence{})); + + template + static auto make_output_channel_tuple(std::index_sequence) + -> std::tuple>*...>; + + using output_channels_t = decltype(make_output_channel_tuple( + std::make_index_sequence{})); + + void run_loop() { + while (!stop_flag_.load(std::memory_order_relaxed)) { + try { + auto t0 = clock_t::now(); + auto args = pop_inputs(std::make_index_sequence{}); + auto t1 = clock_t::now(); + auto cpu0 = NodeStats::cpu_now(); + + if constexpr (std::is_void_v) { + std::apply([this](auto&&... a) { obj_(std::forward(a)...); }, args); + } else { + auto result = std::apply([this](auto&&... a) { return obj_(std::forward(a)...); }, args); + push_outputs(normalise(std::move(result)), + std::make_index_sequence{}); + } + + 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] overflow: " << e.what() << "\n"; + } + } + } + + template + args_tuple pop_inputs(std::index_sequence) { + return {std::get(input_channels_)->pop()...}; + } + + template + static return_tuple normalise(R&& r) { + if constexpr (is_tuple_v) return std::move(r); + else return std::make_tuple(std::move(r)); + } + + template + void push_outputs(return_tuple&& result, std::index_sequence) { + (push_one(std::get(std::move(result))), ...); + } + + template + void push_one(std::tuple_element_t&& val) { + auto* ch = std::get(output_channels_); + if (!ch) return; + try { + ch->push(std::move(val)); + } catch (const ChannelOverflowError&) { + throw ChannelOverflowError(ch->capacity(), "node '" + name_ + "' " + output_port_label()); + } + } + + template + static std::string output_port_label() { + if constexpr (sizeof...(OutNames) > 0) { + constexpr std::array names{OutNames.view()...}; + return std::string("output['") + std::string(names[I]) + "']"; + } else { + return "output[" + std::to_string(I) + "]"; + } + } + + Obj& obj_; + std::string name_; + std::size_t fifo_capacity_; + input_channels_t input_channels_; + output_channels_t output_channels_{}; + std::atomic stop_flag_{false}; + std::jthread thread_; + NodeStats stats_; +}; + +// ── make_node overloads for callable objects ────────────────────────────────── + +template +auto make_node(Obj& obj, std::size_t fifo_capacity = 5) { + return ObjectNode, out<>>(obj, fifo_capacity); +} + +template +auto make_node(Obj& obj, in, std::size_t fifo_capacity = 5) { + return ObjectNode, out<>>(obj, fifo_capacity); +} + +template +auto make_node(Obj& obj, out, std::size_t fifo_capacity = 5) { + return ObjectNode, out>(obj, fifo_capacity); +} + +template +auto make_node(Obj& obj, in, out, std::size_t fifo_capacity = 5) { + return ObjectNode, out>(obj, fifo_capacity); +} + +// ── make_node factory (NTTP) ────────────────────────────────────────────────── +// +// Usage: +// make_node(capacity) +// make_node(in<"a","b">{}, capacity) +// make_node(out<"x">{}, capacity) +// make_node(in<"a","b">{}, out<"x">{}, capacity) + +// No names +template +auto make_node(std::size_t fifo_capacity = 5) { + return Node, out<>>(fifo_capacity); +} + +// in<> only +template +auto make_node(in, std::size_t fifo_capacity = 5) { + return Node, out<>>(fifo_capacity); +} + +// out<> only (no input names) +template +auto make_node(out, std::size_t fifo_capacity = 5) { + return Node, out>(fifo_capacity); +} + +// in<> and out<> +template +auto make_node(in, out, std::size_t fifo_capacity = 5) { + return Node, out>(fifo_capacity); +} + +} // namespace kpn diff --git a/include/kpn/port.hpp b/include/kpn/port.hpp new file mode 100644 index 0000000..321d93b --- /dev/null +++ b/include/kpn/port.hpp @@ -0,0 +1,17 @@ +#pragma once +#include + +namespace kpn { + +// Forward-declared so port handles can reference a node without including node.hpp +template +struct InputPort { + NodeT& node; +}; + +template +struct OutputPort { + NodeT& node; +}; + +} // namespace kpn diff --git a/include/kpn/python/bindings.hpp b/include/kpn/python/bindings.hpp new file mode 100644 index 0000000..6a324f9 --- /dev/null +++ b/include/kpn/python/bindings.hpp @@ -0,0 +1,412 @@ +#pragma once +// Nanobind binding helpers for KPN++ Python interface. +// Included only by python/kpn_python.cpp — do not include from core headers. + +#include "../variant_node.hpp" +#include "../network.hpp" + +#ifdef KPN_BUILD_PYTHON +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace kpn::python { + +namespace nb = nanobind; + +// ── PyNetwork ──────────────────────────────────────────────────────── +// Runtime graph builder for Python. Holds IVariantNode instances and connects +// them via IVariantChannel adapters. The variant only lives at the boundary; +// each node's internal Channel stores raw T values. + +template +class PyNetwork { +public: + using VNode = IVariantNode; + using VChannel = IVariantChannel; + + // ── Builder API ─────────────────────────────────────────────────────────── + + void add(std::string name, std::shared_ptr node) { + if (nodes_.count(name)) + throw std::runtime_error("duplicate node name: " + name); + node->set_name(name); + nodes_.emplace(name, std::move(node)); + adj_[name]; + } + + // connect(src_name, out_idx, dst_name, in_idx) + // Wires src's output port out_idx to dst's input port in_idx. + // Type check: both sides must carry the same T. + void connect(const std::string& src_name, std::size_t out_idx, + const std::string& dst_name, std::size_t in_idx) + { + auto& src = node_at(src_name); + auto& dst = node_at(dst_name); + + if (out_idx >= src.output_count()) + throw std::out_of_range(src_name + ": output index " + + std::to_string(out_idx) + " out of range"); + if (in_idx >= dst.input_count()) + throw std::out_of_range(dst_name + ": input index " + + std::to_string(in_idx) + " out of range"); + + if (src.output_type(out_idx) != dst.input_type(in_idx)) + throw std::runtime_error( + "type mismatch: " + src_name + ".output[" + std::to_string(out_idx) + + "] (" + src.output_type(out_idx).name() + ") → " + + dst_name + ".input[" + std::to_string(in_idx) + + "] (" + dst.input_type(in_idx).name() + ")"); + + // The destination node owns the input channel — get it, then tell src to use it. + auto ch = dst.input_channel(in_idx); + src.set_output_channel(out_idx, std::move(ch)); + adj_[src_name].push_back(dst_name); + } + + void build() { + topo_.clear(); + std::map color; + for (auto& [name, _] : nodes_) + if (color[name] == 0) dfs(name, color); + } + + // ── Lifecycle ───────────────────────────────────────────────────────────── + + void start() { + for (auto& name : topo_) + nodes_.at(name)->start(); + } + + void stop() { + for (auto it = topo_.rbegin(); it != topo_.rend(); ++it) + nodes_.at(*it)->stop(); + } + + // ── Python tap/inject ───────────────────────────────────────────────────── + + // Read one value from node's output port. Releases GIL while blocking. + nb::object read(const std::string& node_name, std::size_t out_idx) { + // We need a channel that sits on the output of this node. + // read() installs a tap channel if not already present. + auto key = tap_key(node_name, out_idx); + if (!taps_.count(key)) { + auto& src = node_at(node_name); + if (out_idx >= src.output_count()) + throw std::out_of_range(node_name + ": output index out of range"); + // Create a tap channel matching the output type and wire it + auto tap = make_tap_channel(src.output_type(out_idx)); + src.set_output_channel(out_idx, tap); + taps_[key] = std::move(tap); + } + Variant v; + { + nb::gil_scoped_release release; + v = taps_.at(key)->pop(); + } + return variant_to_python(std::move(v)); + } + + // Write a Python value into node's input port. Releases GIL while blocking. + void write(const std::string& node_name, std::size_t in_idx, nb::object value) { + auto& dst = node_at(node_name); + if (in_idx >= dst.input_count()) + throw std::out_of_range(node_name + ": input index out of range"); + + auto ch = dst.input_channel(in_idx); + Variant v = python_to_variant(ch->type_index(), std::move(value)); + { + nb::gil_scoped_release release; + ch->push(std::move(v)); + } + } + + // ── Converter registration ──────────────────────────────────────────────── + // Called once per type at module init time to register to/from Python converters. + + template + void register_type( + std::function to_py, + std::function from_py) + { + auto idx = std::type_index(typeid(T)); + to_python_[idx] = [to_py](const Variant& v) { return to_py(std::get(v)); }; + from_python_[idx] = [from_py](nb::object o) -> Variant { + return Variant{ from_py(std::move(o)) }; + }; + } + +private: + VNode& node_at(const std::string& name) { + auto it = nodes_.find(name); + if (it == nodes_.end()) + throw std::runtime_error("unknown node: " + name); + return *it->second; + } + + void dfs(const std::string& name, std::map& color) { + color[name] = 1; + for (auto& nbr : adj_[name]) { + if (color[nbr] == 1) + throw std::runtime_error("cycle detected in graph"); + if (color[nbr] == 0) dfs(nbr, color); + } + color[name] = 2; + topo_.insert(topo_.begin(), name); + } + + std::string tap_key(const std::string& node, std::size_t idx) { + return node + ":" + std::to_string(idx); + } + + std::shared_ptr make_tap_channel(std::type_index type) { + // Create the right VariantChannel based on the registered type index. + // We need a factory registered per type — stored in tap_factories_. + auto it = tap_factories_.find(type); + if (it == tap_factories_.end()) + throw std::runtime_error( + "no tap factory for type: " + std::string(type.name()) + + " — was register_type() called for this type?"); + return it->second(); + } + + nb::object variant_to_python(Variant v) { + auto idx = std::visit([](auto& x) { + return std::type_index(typeid(x)); + }, v); + auto it = to_python_.find(idx); + if (it == to_python_.end()) + throw std::runtime_error("no to_python converter for type"); + return it->second(v); + } + + Variant python_to_variant(std::type_index idx, nb::object obj) { + auto it = from_python_.find(idx); + if (it == from_python_.end()) + throw std::runtime_error("no from_python converter for type"); + return it->second(std::move(obj)); + } + +public: + // Called by register_type to also register a tap channel factory. + template + void register_tap_factory(std::size_t capacity = 5) { + auto idx = std::type_index(typeid(T)); + tap_factories_[idx] = [capacity]() -> std::shared_ptr { + auto ch = std::make_shared>(capacity); + return std::make_shared>(std::move(ch)); + }; + } + +private: + std::map> nodes_; + std::map> adj_; + std::vector topo_; + std::map> taps_; + + std::map> to_python_; + std::map> from_python_; + std::map()>> tap_factories_; +}; + +// ── PyNode ─────────────────────────────────────────────────────────── +// A pure-Python processing node. Holds a nanobind callable. +// run_loop: pop inputs (release GIL), call Python (acquire GIL), push outputs (release GIL). + +template +class PyNode : public IVariantNode { +public: + using VChannel = IVariantChannel; + + using ChannelFactory = std::function(std::size_t capacity)>; + + PyNode(nb::object callable, + std::vector in_types, + std::vector out_types, + std::map> to_py, + std::map> from_py, + std::map ch_factories, + std::size_t capacity = 5) + : callable_(std::move(callable)) + , in_types_(std::move(in_types)) + , out_types_(std::move(out_types)) + , to_python_(std::move(to_py)) + , from_python_(std::move(from_py)) + , ch_factories_(std::move(ch_factories)) + , in_channels_(in_types_.size()) + , out_channels_(out_types_.size()) + { + for (std::size_t i = 0; i < in_types_.size(); ++i) { + auto it = ch_factories_.find(in_types_[i]); + if (it == ch_factories_.end()) + throw std::runtime_error("PyNode: no channel factory for input type"); + in_channels_[i] = it->second(capacity); + } + } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { + for (auto& ch : in_channels_) 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); + for (auto& ch : in_channels_) ch->disable(); + if (thread_.joinable()) { + thread_.request_stop(); + // Release GIL while joining — run_loop may be waiting to acquire it. + nb::gil_scoped_release release; + thread_.join(); + } + } + + bool running() const override { + return thread_.joinable() && !stop_flag_.load(std::memory_order_relaxed); + } + + void set_name(std::string name) override { + IVariantNode::set_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 blk_ms = stats_.total_blocked_us.load(std::memory_order_relaxed) / 1000.0; + double total_ms = exec_ms + blk_ms; + return { name, frames, exec_ms, + stats_.max_exec_us.load(std::memory_order_relaxed) / 1000.0, + blk_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 }; + } + + // ── IVariantNode ────────────────────────────────────────────────────────── + + std::size_t input_count() const override { return in_types_.size(); } + std::size_t output_count() const override { return out_types_.size(); } + + std::type_index input_type(std::size_t i) const override { return in_types_[i]; } + std::type_index output_type(std::size_t i) const override { return out_types_[i]; } + + std::shared_ptr input_channel(std::size_t i) override { + return in_channels_[i]; + } + + void set_output_channel(std::size_t i, + std::shared_ptr ch) override { + out_channels_[i] = std::move(ch); + } + +private: + void run_loop() { + // This thread does not hold the GIL. It acquires it only for Python calls. + while (!stop_flag_.load(std::memory_order_relaxed)) { + try { + auto t0 = clock_t::now(); + + // Pop all inputs — no GIL needed, these are pure C++ channel ops + std::vector inputs(in_channels_.size()); + for (std::size_t i = 0; i < in_channels_.size(); ++i) + inputs[i] = in_channels_[i]->pop(); + + auto t1 = clock_t::now(); + auto cpu0 = NodeStats::cpu_now(); + + // Acquire GIL only for the Python call and type conversion + std::vector outputs; + { + nb::gil_scoped_acquire acquire; + nb::list py_args; + for (auto& v : inputs) + py_args.append(variant_to_python(v)); + + nb::object result = callable_(*py_args); + + if (out_channels_.size() == 1) { + outputs.push_back(python_to_variant(out_types_[0], result)); + } else { + nb::tuple tup = nb::cast(result); + for (std::size_t i = 0; i < out_channels_.size(); ++i) + outputs.push_back(python_to_variant(out_types_[i], tup[i])); + } + } + + auto cpu1 = NodeStats::cpu_now(); + auto t2 = clock_t::now(); + stats_.record_exec(duration_t(t2 - t1), duration_t(t1 - t0), cpu0, cpu1); + + // Push outputs — no GIL needed + for (std::size_t i = 0; i < out_channels_.size(); ++i) { + if (out_channels_[i]) + out_channels_[i]->push(std::move(outputs[i])); + } + + } catch (const ChannelClosedError&) { + break; + } catch (const ChannelOverflowError&) { + // drop and continue + } + } + } + + nb::object variant_to_python(const Variant& v) { + auto idx = std::visit([](const auto& x) { + return std::type_index(typeid(x)); + }, v); + return to_python_.at(idx)(v); + } + + Variant python_to_variant(std::type_index idx, nb::object obj) { + return from_python_.at(idx)(std::move(obj)); + } + + nb::object callable_; + std::vector in_types_; + std::vector out_types_; + std::map> to_python_; + std::map> from_python_; + std::map ch_factories_; + std::vector> in_channels_; + std::vector> out_channels_; + std::atomic stop_flag_{false}; + std::jthread thread_; + NodeStats stats_; +}; + +// ── register_py_network ─────────────────────────────────────────────────────── +// Registers PyNetwork and PyNode with the given nanobind module. +// Call once per module, passing the Variant type derived from your registered node types. + +template +void register_py_network(nb::module_& m, const char* class_name = "Network") { + using Net = PyNetwork; + + nb::class_(m, class_name) + .def(nb::init<>()) + .def("connect", &Net::connect, + nb::arg("src"), nb::arg("out_idx"), + nb::arg("dst"), nb::arg("in_idx")) + .def("build", &Net::build) + .def("start", &Net::start) + .def("stop", &Net::stop) + .def("read", &Net::read, + nb::arg("node"), nb::arg("out_idx") = 0) + .def("write", &Net::write, + nb::arg("node"), nb::arg("in_idx"), nb::arg("value")); +} + +} // namespace kpn::python + +#endif // KPN_BUILD_PYTHON diff --git a/include/kpn/traits.hpp b/include/kpn/traits.hpp new file mode 100644 index 0000000..c09d54f --- /dev/null +++ b/include/kpn/traits.hpp @@ -0,0 +1,86 @@ +#pragma once +#include +#include + +namespace kpn { + +// ── Primary template — not defined; only specialisations match ──────────────── +template +struct function_traits; + +// Free function +template +struct function_traits { + using return_t = R; + using args = std::tuple; + static constexpr std::size_t arity = sizeof...(Args); +}; + +// Function pointer +template +struct function_traits : function_traits {}; + +// Member function pointer (const) +template +struct function_traits : function_traits {}; + +// Member function pointer (non-const) +template +struct function_traits : function_traits {}; + +// Callable (lambda / std::function) — delegate to operator() +template +struct function_traits : function_traits {}; + +// ── Helpers ─────────────────────────────────────────────────────────────────── + +template +using return_t = typename function_traits>::return_t; + +template +using args_t = typename function_traits>::args; + +template +inline constexpr std::size_t arity_v = function_traits>::arity; + +// ── Tuple detection ─────────────────────────────────────────────────────────── + +template +struct is_tuple : std::false_type {}; + +template +struct is_tuple> : std::true_type {}; + +template +inline constexpr bool is_tuple_v = is_tuple::value; + +// ── Normalise return type to always be a tuple ──────────────────────────────── +// void → std::tuple<> +// T (non-tup) → std::tuple +// tuple<...> → tuple<...> (unchanged) + +template +struct normalise_return { + using type = std::tuple; +}; + +template<> +struct normalise_return { + using type = std::tuple<>; +}; + +template +struct normalise_return> { + using type = std::tuple; +}; + +template +using normalised_return_t = typename normalise_return::type; + +// ── Output count from a function type ──────────────────────────────────────── + +template +inline constexpr std::size_t output_count_v = + std::tuple_size_v>>; + +} // namespace kpn diff --git a/include/kpn/variant_node.hpp b/include/kpn/variant_node.hpp new file mode 100644 index 0000000..cacaa6e --- /dev/null +++ b/include/kpn/variant_node.hpp @@ -0,0 +1,249 @@ +#pragma once +#include "channel.hpp" +#include "node.hpp" +#include "traits.hpp" + +#include +#include +#include +#include +#include +#include + +namespace kpn { + +// ── unique_types TMP helper ─────────────────────────────────────────────────── + +namespace detail { + +template +struct unique_types_impl; + +template +struct unique_types_impl> { + using type = std::tuple; +}; + +template +struct unique_types_impl, Head, Tail...> { + using type = std::conditional_t< + (std::is_same_v || ...), + typename unique_types_impl, Tail...>::type, + typename unique_types_impl, Tail...>::type + >; +}; + +template +using unique_types_t = typename unique_types_impl, Ts...>::type; + +template +struct tuple_to_variant; + +template +struct tuple_to_variant> { + using type = std::variant; +}; + +} // namespace detail + +// ── IVariantChannel ─────────────────────────────────────────────────────────── +// Type-erased channel surface used by PyNetwork. +// The underlying FIFO stores raw T — variant conversion happens only at push/pop. + +template +class IVariantChannel { +public: + virtual ~IVariantChannel() = default; + virtual void push(Variant v) = 0; + virtual Variant pop() = 0; + virtual std::type_index type_index() const = 0; + virtual std::string type_name() const = 0; + virtual void enable() = 0; + virtual void disable() = 0; +}; + +// ── VariantChannel ──────────────────────────────────────────────── +// Shares a Channel with the node that owns the input slot. +// push(): std::get from Variant → raw T into the queue. +// pop(): raw T from queue → wrapped into Variant. + +template +class VariantChannel : public IVariantChannel { +public: + explicit VariantChannel(std::shared_ptr> ch) + : channel_(std::move(ch)) {} + + void push(Variant v) override { + channel_->push(std::get(std::move(v))); + } + Variant pop() override { + return Variant{ channel_->pop() }; + } + std::type_index type_index() const override { return std::type_index(typeid(T)); } + std::string type_name() const override { return typeid(T).name(); } + void enable() override { channel_->enable(); } + void disable() override { channel_->disable(); } + + Channel* raw_ptr() { return channel_.get(); } + +private: + std::shared_ptr> channel_; +}; + +// ── IVariantNode ────────────────────────────────────────────────────────────── + +template +class IVariantNode : public INode, + public std::enable_shared_from_this> { +public: + void set_name(std::string name) override { name_ = std::move(name); } + const std::string& name() const { return name_; } + + virtual std::size_t input_count() const = 0; + virtual std::size_t output_count() const = 0; + virtual std::type_index input_type(std::size_t i) const = 0; + virtual std::type_index output_type(std::size_t i) const = 0; + + // Returns the IVariantChannel wrapping this node's input slot i. + // PyNetwork calls this on the *destination* node to get the channel to wire upstream into. + virtual std::shared_ptr> input_channel(std::size_t i) = 0; + + // Wires this node's output slot i into ch (ch belongs to the downstream node's input). + virtual void set_output_channel(std::size_t i, + std::shared_ptr> ch) = 0; +private: + std::string name_; +}; + +// ── VariantNodeWrapper ────────────────────────── +// Wraps a Node for use inside a PyNetwork. +// +// At construction: for each input port I, builds a shared_ptr> and +// installs it in both the wrapped node (via set_input_channel) and a +// VariantChannel adapter. PyNetwork passes the adapter to the upstream +// node's set_output_channel so the upstream raw pointer points at the same Channel. +// +// At connect (output side): downcasts the provided IVariantChannel to +// VariantChannel, then calls node_.set_output_channel with the raw ptr. + +template, + typename OutputTag = out<>> +class VariantNodeWrapper; + +template +class VariantNodeWrapper, out> + : public IVariantNode +{ + using NodeT = Node, out>; + +public: + using args_tuple = typename NodeT::args_tuple; + using return_tuple = typename NodeT::return_tuple; + + static constexpr std::size_t n_in = NodeT::input_count; + static constexpr std::size_t n_out = NodeT::output_count; + + explicit VariantNodeWrapper(std::size_t fifo_capacity = 5) + : node_(fifo_capacity) + , in_channels_(n_in) + , out_channels_(n_out) + , out_type_indices_(n_out, std::type_index(typeid(void))) + { + init_inputs(std::make_index_sequence{}, fifo_capacity); + init_out_types(std::make_index_sequence{}); + } + + // ── INode ───────────────────────────────────────────────────────────────── + + void start() override { node_.start(); } + void stop() override { node_.stop(); } + bool running() const override { return node_.running(); } + const NodeStats& stats() const override { return node_.stats(); } + void set_name(std::string name) override { node_.set_name(std::move(name)); } + NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { + return node_.node_snapshot(name, elapsed_s); + } + + // ── IVariantNode ────────────────────────────────────────────────────────── + + std::size_t input_count() const override { return n_in; } + std::size_t output_count() const override { return n_out; } + + std::type_index input_type(std::size_t i) const override { + return in_channels_[i]->type_index(); + } + std::type_index output_type(std::size_t i) const override { + return out_type_indices_[i]; + } + + std::shared_ptr> input_channel(std::size_t i) override { + return in_channels_[i]; + } + + void set_output_channel(std::size_t i, + std::shared_ptr> ch) override { + set_output_impl(i, std::move(ch), std::make_index_sequence{}); + } + +private: + template + void init_inputs(std::index_sequence, std::size_t cap) { + ((init_one_input(cap)), ...); + } + + template + void init_one_input(std::size_t cap) { + using T = std::tuple_element_t; + auto shared_ch = std::make_shared>(cap); + // Install the shared Channel in the node's input slot + node_.template set_input_channel(shared_ch); + // Wrap it in a VariantChannel for PyNetwork to use + in_channels_[I] = std::make_shared>(std::move(shared_ch)); + } + + template + void init_out_types(std::index_sequence) { + ((out_type_indices_[Is] = + std::type_index(typeid(std::tuple_element_t))), ...); + } + + template + void set_output_impl(std::size_t port, + std::shared_ptr> ch, + std::index_sequence) { + bool matched = false; + ((Is == port && (set_output_at(std::move(ch)), matched = true)), ...); + if (!matched) + throw std::out_of_range("set_output_channel: port index out of range"); + } + + template + void set_output_at(std::shared_ptr> ch) { + using T = std::tuple_element_t; + auto* typed = dynamic_cast*>(ch.get()); + if (!typed) + throw std::runtime_error( + "set_output_channel: type mismatch at output port " + std::to_string(I)); + // The downstream VariantChannel holds a shared Channel; point our output at it. + // We need a raw ptr — extract it via a getter. + node_.template set_output_channel(typed->raw_ptr()); + out_channels_[I] = std::move(ch); + } + + NodeT node_; + std::vector>> in_channels_; + std::vector>> out_channels_; + std::vector out_type_indices_; +}; + +// ── PythonConverter ─────────────────────────────────────────────────────────── + +template +struct PythonConverter { + static_assert(sizeof(T) == 0, + "PythonConverter must be specialised for each type used in a PyNetwork"); +}; + +} // namespace kpn diff --git a/include/kpn/web_debug.hpp b/include/kpn/web_debug.hpp new file mode 100644 index 0000000..4721082 --- /dev/null +++ b/include/kpn/web_debug.hpp @@ -0,0 +1,345 @@ +#pragma once +// Only compiled when KPN_WEB_DEBUG is defined. network.hpp includes this conditionally. + +#include "diagnostics.hpp" +#include + +#include +#include +#include +#include +#include + +namespace kpn::web_debug { + +// ── Minimal JSON serialiser ─────────────────────────────────────────────────── + +static std::string escape_json(const std::string& s) { + std::string out; + out.reserve(s.size()); + for (char c : s) { + if (c == '"') out += "\\\""; + else if (c == '\\') out += "\\\\"; + else if (c == '\n') out += "\\n"; + else if (c == '\r') out += "\\r"; + else if (c == '\t') out += "\\t"; + else out += c; + } + return out; +} + +// Parse "src_name:N → dst_name:M" into {src_name, dst_name}. +// Stored separately so the browser doesn't need to regex-parse a UTF-8 arrow. +static std::pair parse_edge_name(const std::string& name) { + // Format: ": → :" + auto arrow = name.find(" \xe2\x86\x92 "); // UTF-8 for → + if (arrow == std::string::npos) return {name, name}; + std::string src_part = name.substr(0, arrow); + std::string dst_part = name.substr(arrow + 5); // " → " = 5 bytes (space + 3-byte arrow + space) + // Strip ":N" index suffix + auto sc1 = src_part.rfind(':'); + auto sc2 = dst_part.rfind(':'); + if (sc1 != std::string::npos) src_part = src_part.substr(0, sc1); + if (sc2 != std::string::npos) dst_part = dst_part.substr(0, sc2); + return {src_part, dst_part}; +} + +static std::string to_json(const std::vector& nodes, + const std::vector& channels, + double elapsed_s = 0.0) { + std::ostringstream o; + o << std::fixed; + o.precision(2); + + o << "{\"nodes\":["; + for (std::size_t i = 0; i < nodes.size(); ++i) { + const auto& n = nodes[i]; + if (i) o << ','; + o << "{\"id\":\"" << escape_json(n.name) << "\"" + << ",\"frames\":" << n.frames_processed + << ",\"ema_exec_ms\":" << n.ema_exec_ms + << ",\"max_exec_ms\":" << n.max_exec_ms + << ",\"blocked_ms\":" << n.total_blocked_ms + << ",\"fps\":" << n.throughput_fps + << ",\"total_cpu_ms\":" << n.total_cpu_ms + << ",\"cpu_util_pct\":" << n.cpu_util_pct + << "}"; + } + o << "],\"edges\":["; + for (std::size_t i = 0; i < channels.size(); ++i) { + const auto& c = channels[i]; + if (i) o << ','; + auto [src, dst] = parse_edge_name(c.name); + o << "{\"name\":\"" << escape_json(c.name) << "\"" + << ",\"source\":\"" << escape_json(src) << "\"" + << ",\"target\":\"" << escape_json(dst) << "\"" + << ",\"capacity\":" << c.capacity + << ",\"current\":" << c.current_fill + << ",\"fill_pct\":" << c.fill_pct() + << ",\"peak_pct\":" << c.peak_pct() + << ",\"pushes\":" << c.pushes + << ",\"drops\":" << c.drops + << ",\"overflows\":" << c.overflows + << ",\"item_bytes\":" << c.item_bytes + << ",\"bw_mbs\":" << c.bandwidth_mbs(elapsed_s) + << "}"; + } + o << "]}"; + return o.str(); +} + +// ── Embedded single-page HTML ───────────────────────────────────────────────── + +static const char* HTML = R"html( + + + +KPN++ Web Debug + + + + + +
+ + + + +)html"; + +// ── WebDebugServer ──────────────────────────────────────────────────────────── + +class WebDebugServer { +public: + using SnapshotFn = std::function; + + explicit WebDebugServer(uint16_t port, SnapshotFn fn) + : port_(port), snapshot_fn_(std::move(fn)) {} + + void start() { + svr_.Get("/", [](const httplib::Request&, httplib::Response& res) { + res.set_content(HTML, "text/html"); + }); + svr_.Get("/api/snapshot", [this](const httplib::Request&, httplib::Response& res) { + res.set_content(snapshot_fn_(), "application/json"); + }); + thread_ = std::thread([this] { + svr_.listen("0.0.0.0", static_cast(port_)); + }); + } + + void stop() { + svr_.stop(); + if (thread_.joinable()) thread_.join(); + } + + ~WebDebugServer() { stop(); } + + WebDebugServer(const WebDebugServer&) = delete; + WebDebugServer& operator=(const WebDebugServer&) = delete; + +private: + uint16_t port_; + SnapshotFn snapshot_fn_; + httplib::Server svr_; + std::thread thread_; +}; + +} // namespace kpn::web_debug diff --git a/python/CMakeLists.txt b/python/CMakeLists.txt new file mode 100644 index 0000000..3e2e237 --- /dev/null +++ b/python/CMakeLists.txt @@ -0,0 +1,5 @@ +cmake_minimum_required(VERSION 3.21) + +nanobind_add_module(kpn_python kpn_python.cpp) +target_link_libraries(kpn_python PRIVATE kpn) +target_compile_definitions(kpn_python PRIVATE KPN_BUILD_PYTHON) diff --git a/python/kpn_python.cpp b/python/kpn_python.cpp new file mode 100644 index 0000000..583d9cb --- /dev/null +++ b/python/kpn_python.cpp @@ -0,0 +1,164 @@ +#define KPN_BUILD_PYTHON +#include +#include +#include +#include +#include + +#include +#include + +namespace nb = nanobind; +using namespace kpn; +using namespace kpn::python; + +// ── Node functions for the hello-pipeline examples ──────────────────────────── + +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'; } + +// ── Variant type ────────────────────────────────────────────────────────────── +// Deduplicated port types across all registered node functions. +// produce: () → int | double_it: int → int | print_it: int → void +// Unique types: int + +using KpnVariant = std::variant; +using Net = PyNetwork; +using ProduceNode = VariantNodeWrapper; +using DoubleItNode= VariantNodeWrapper; +using PrintItNode = VariantNodeWrapper; + +// ── Converter helpers for int ───────────────────────────────────────────────── + +static nb::object int_to_py(const KpnVariant& v) { + return nb::int_(std::get(v)); +} + +static KpnVariant int_from_py(nb::object o) { + return KpnVariant{ nb::cast(o) }; +} + +// ── Type name resolver (extensible) ────────────────────────────────────────── + +static std::type_index resolve_type(const std::string& name) { + if (name == "int") return std::type_index(typeid(int)); + throw std::runtime_error("unknown type name '" + name + + "' — only 'int' is registered in this module"); +} + +// ── Ensure converters are registered on a Net instance ─────────────────────── + +static void ensure_converters(Net& net) { + net.register_type( + [](const int& v) -> nb::object { return nb::int_(v); }, + [](nb::object o) -> int { return nb::cast(o); } + ); + net.register_tap_factory(); +} + +NB_MODULE(kpn_python, m) { + m.doc() = "KPN++ Python bindings — Kahn Process Network library"; + + // ── IVariantNode base ───────────────────────────────────────────────────── + nb::class_>(m, "INode"); + + // ── Concrete C++ node wrappers ───────────────────────────────────────────── + // Factories return shared_ptr so Python can pass them to net.add(). + nb::class_>(m, "ProduceNode") + .def("__init__", [](ProduceNode* self, std::size_t cap) { + new (self) ProduceNode(cap); + }, nb::arg("capacity") = 5); + + nb::class_>(m, "DoubleItNode") + .def("__init__", [](DoubleItNode* self, std::size_t cap) { + new (self) DoubleItNode(cap); + }, nb::arg("capacity") = 5); + + nb::class_>(m, "PrintItNode") + .def("__init__", [](PrintItNode* self, std::size_t cap) { + new (self) PrintItNode(cap); + }, nb::arg("capacity") = 5); + + // Expose factory functions that return shared_ptr — these are what net.add() accepts. + m.def("make_produce", [](std::size_t cap) -> std::shared_ptr> { + return std::make_shared(cap); + }, nb::arg("capacity") = 5); + m.def("make_double_it", [](std::size_t cap) -> std::shared_ptr> { + return std::make_shared(cap); + }, nb::arg("capacity") = 5); + m.def("make_print_it", [](std::size_t cap) -> std::shared_ptr> { + return std::make_shared(cap); + }, nb::arg("capacity") = 5); + + // ── Network ─────────────────────────────────────────────────────────────── + nb::class_(m, "Network") + .def(nb::init<>()) + + // add(name, c++_node) — for pre-constructed C++ nodes + .def("add", [](Net& self, std::string name, + std::shared_ptr> node) { + self.add(std::move(name), std::move(node)); + }, nb::arg("name"), nb::arg("node")) + + // add_node(name, callable, inputs=[type_names], outputs=[type_names]) + // Creates a pure Python processing node. + .def("add_node", [](Net& self, + std::string name, + nb::object callable, + std::vector in_names, + std::vector out_names, + std::size_t capacity) + { + std::vector in_types, out_types; + for (auto& s : in_names) in_types.push_back(resolve_type(s)); + for (auto& s : out_names) out_types.push_back(resolve_type(s)); + + std::map> to_py; + std::map> from_py; + std::map::ChannelFactory> ch_factories; + + auto int_idx = std::type_index(typeid(int)); + to_py[int_idx] = int_to_py; + from_py[int_idx] = int_from_py; + ch_factories[int_idx] = [](std::size_t cap) -> std::shared_ptr> { + auto ch = std::make_shared>(cap); + return std::make_shared>(std::move(ch)); + }; + + auto node = std::make_shared>( + std::move(callable), + std::move(in_types), + std::move(out_types), + std::move(to_py), + std::move(from_py), + std::move(ch_factories), + capacity + ); + self.add(std::move(name), std::move(node)); + }, + nb::arg("name"), nb::arg("callable"), + nb::arg("inputs") = std::vector{}, + nb::arg("outputs") = std::vector{}, + nb::arg("capacity") = 5) + + .def("connect", &Net::connect, + nb::arg("src"), nb::arg("out_idx"), + nb::arg("dst"), nb::arg("in_idx")) + .def("build", &Net::build) + .def("start", &Net::start) + .def("stop", &Net::stop) + + // read(node, out_idx=0) — blocking pop from a C++ node's output port into Python + .def("read", [](Net& self, const std::string& node, std::size_t out_idx) { + ensure_converters(self); + return self.read(node, out_idx); + }, nb::arg("node"), nb::arg("out_idx") = 0) + + // write(node, in_idx, value) — push a Python value into a node's input port + .def("write", [](Net& self, const std::string& node, + std::size_t in_idx, nb::object value) { + ensure_converters(self); + self.write(node, in_idx, std::move(value)); + }, nb::arg("node"), nb::arg("in_idx"), nb::arg("value")); +} diff --git a/src/network.cpp b/src/network.cpp new file mode 100644 index 0000000..063d15b --- /dev/null +++ b/src/network.cpp @@ -0,0 +1,6 @@ +// network.cpp — orchestrator/watchdog implementation details. +// Most of the Network class is header-only (template-heavy). +// Non-template implementation lives here once the watchdog grows +// beyond the stub in network.hpp. + +#include diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt new file mode 100644 index 0000000..e8e282d --- /dev/null +++ b/tests/CMakeLists.txt @@ -0,0 +1,44 @@ +cmake_minimum_required(VERSION 3.21) + +# ── Catch2 ──────────────────────────────────────────────────────────────────── +find_package(Catch2 3 QUIET) +if(NOT Catch2_FOUND) + include(FetchContent) + FetchContent_Declare( + Catch2 + GIT_REPOSITORY https://github.com/catchorg/Catch2.git + GIT_TAG v3.5.3 + ) + FetchContent_MakeAvailable(Catch2) +endif() + +# ── Google Test ─────────────────────────────────────────────────────────────── +find_package(GTest QUIET) +if(NOT GTest_FOUND) + include(FetchContent) + FetchContent_Declare( + googletest + GIT_REPOSITORY https://github.com/google/googletest.git + GIT_TAG v1.14.0 + ) + FetchContent_MakeAvailable(googletest) +endif() + +# ── Test executable ─────────────────────────────────────────────────────────── +add_executable(kpn_tests + test_fixed_string.cpp + test_traits.cpp + test_channel.cpp + test_node.cpp + test_network.cpp +) + +target_link_libraries(kpn_tests PRIVATE + kpn + Catch2::Catch2WithMain + GTest::gtest +) + +include(CTest) +include(Catch) +catch_discover_tests(kpn_tests) diff --git a/tests/test_channel.cpp b/tests/test_channel.cpp new file mode 100644 index 0000000..a45c203 --- /dev/null +++ b/tests/test_channel.cpp @@ -0,0 +1,99 @@ +#include +#include +#include + +using namespace kpn; + +TEST_CASE("channel storage policy: small trivial type by value", "[channel]") { + STATIC_REQUIRE(channel_storage_policy::by_value); + STATIC_REQUIRE(channel_storage_policy::by_value); +} + +TEST_CASE("channel storage policy: large type by shared_ptr", "[channel]") { + struct Big { char data[64]; }; + STATIC_REQUIRE(!channel_storage_policy::by_value); +} + +TEST_CASE("push and pop single value", "[channel]") { + Channel ch(5); + ch.push(42); + REQUIRE(ch.pop() == 42); +} + +TEST_CASE("channel respects capacity", "[channel]") { + Channel ch(2); + ch.push(1); + ch.push(2); + REQUIRE_THROWS_AS(ch.push(3), ChannelOverflowError); +} + +TEST_CASE("pop blocks until value available", "[channel]") { + Channel ch(5); + int result = 0; + std::thread producer([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + ch.push(99); + }); + result = ch.pop(); + producer.join(); + REQUIRE(result == 99); +} + +TEST_CASE("try_pop returns false on timeout", "[channel]") { + Channel ch(5); + int out = 0; + REQUIRE_FALSE(ch.try_pop(out, std::chrono::milliseconds(10))); +} + +TEST_CASE("try_pop succeeds when value present", "[channel]") { + Channel ch(5); + ch.push(7); + int out = 0; + REQUIRE(ch.try_pop(out, std::chrono::milliseconds(10))); + REQUIRE(out == 7); +} + +TEST_CASE("disable unblocks waiting pop", "[channel]") { + Channel ch(5); + std::thread disabler([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + ch.disable(); + }); + REQUIRE_THROWS_AS(ch.pop(), ChannelClosedError); + disabler.join(); +} + +TEST_CASE("push to disabled channel is silently dropped", "[channel]") { + Channel ch(5); + ch.disable(); + ch.push(99); // must not throw, must not enqueue + REQUIRE(ch.size() == 0); +} + +TEST_CASE("disable clears existing queue contents", "[channel]") { + Channel ch(5); + ch.push(1); + ch.push(2); + REQUIRE(ch.size() == 2); + ch.disable(); + REQUIRE(ch.size() == 0); +} + +TEST_CASE("enable re-accepts pushes after disable", "[channel]") { + Channel ch(5); + ch.disable(); + ch.push(1); + REQUIRE(ch.size() == 0); + ch.enable(); + ch.push(42); + REQUIRE(ch.pop() == 42); +} + +TEST_CASE("large type stored as shared_ptr — no copy on pop", "[channel]") { + struct Big { char data[64]; int tag; }; + Channel ch(5); + Big b{}; b.tag = 123; + ch.push(b); + auto out = ch.pop(); + REQUIRE(out.tag == 123); +} diff --git a/tests/test_fixed_string.cpp b/tests/test_fixed_string.cpp new file mode 100644 index 0000000..22b10f7 --- /dev/null +++ b/tests/test_fixed_string.cpp @@ -0,0 +1,27 @@ +#include +#include + +using namespace kpn; + +TEST_CASE("fixed_string equality", "[fixed_string]") { + constexpr fixed_string a("img"); + constexpr fixed_string b("img"); + constexpr fixed_string c("sigma"); + STATIC_REQUIRE(a == b); + STATIC_REQUIRE(!(a == c)); +} + +TEST_CASE("fixed_string view", "[fixed_string]") { + constexpr fixed_string s("hello"); + REQUIRE(s.view() == "hello"); +} + +TEST_CASE("index_of hit", "[fixed_string]") { + constexpr auto idx = index_of(); + STATIC_REQUIRE(idx == 1); +} + +TEST_CASE("index_of miss returns npos", "[fixed_string]") { + constexpr auto idx = index_of(); + STATIC_REQUIRE(idx == npos); +} diff --git a/tests/test_network.cpp b/tests/test_network.cpp new file mode 100644 index 0000000..9d75f64 --- /dev/null +++ b/tests/test_network.cpp @@ -0,0 +1,56 @@ +#include +#include +#include +#include + +using namespace kpn; + +static int increment(int x) { return x + 1; } +static int multiply2(int x) { return x * 2; } + +TEST_CASE("network build and run: linear pipeline", "[network]") { + // Nodes declared first — they own their input channels and must outlive the network + auto src = make_node(5); + auto dst = make_node(5); + + auto& src_in = src.input_channel<0>(); + Channel final_out(5); + dst.set_output_channel<0>(&final_out); + + Network net; + net.add("src", src) + .add("dst", dst) + .connect("src", src.output<0>(), "dst", dst.input<0>()) + .build(); + + net.start(); + src_in.push(5); // 5 → increment → 6 → multiply2 → 12 + int result = final_out.pop(); + net.stop(); + + REQUIRE(result == 12); +} + +TEST_CASE("network detects cycle", "[network]") { + auto a = make_node(5); + auto b = make_node(5); + + Network net; + net.add("a", a).add("b", b); + net.connect("a", a.output<0>(), "b", b.input<0>()); + net.connect("b", b.output<0>(), "a", a.input<0>()); + + REQUIRE_THROWS_AS(net.build(), NetworkCycleError); +} + +TEST_CASE("stop disables input channels — producer push is silently dropped", "[network]") { + auto node = make_node(5); + auto& in_ch = node.input_channel<0>(); + + node.start(); + node.stop(); + + // After stop, channel is disabled — push must not throw + in_ch.push(99); + REQUIRE(in_ch.size() == 0); +} diff --git a/tests/test_node.cpp b/tests/test_node.cpp new file mode 100644 index 0000000..516e5e0 --- /dev/null +++ b/tests/test_node.cpp @@ -0,0 +1,49 @@ +#include +#include +#include +#include + +using namespace kpn; + +static int double_it(int x) { return x * 2; } +static std::tuple split_it(int x) { return {x, float(x) * 0.5f}; } +static void consume_it(int x) { (void)x; } + +TEST_CASE("node input/output counts", "[node]") { + STATIC_REQUIRE(Node::input_count == 1); + STATIC_REQUIRE(Node::output_count == 1); + STATIC_REQUIRE(Node::output_count == 2); + STATIC_REQUIRE(Node::output_count == 0); +} + +TEST_CASE("node named port index resolution", "[node]") { + using N = Node, out<"result">>; + STATIC_REQUIRE(index_of() == 0); + STATIC_REQUIRE(index_of() == 0); +} + +TEST_CASE("node processes a single item end-to-end", "[node]") { + Node src_node(5); // used only as input source placeholder + Node node(5); + + // Manually wire: push to input channel, connect a downstream channel, run one item + auto& in_ch = node.input_channel<0>(); + + Channel out_ch(5); + node.set_output_channel<0>(&out_ch); + + node.start(); + in_ch.push(21); + int result = out_ch.pop(); + node.stop(); + + REQUIRE(result == 42); +} + +TEST_CASE("node stop unblocks cleanly", "[node]") { + Node node(5); + node.start(); + // Node is blocked waiting for input — stop() must return without deadlock + node.stop(); + REQUIRE_FALSE(node.running()); +} diff --git a/tests/test_traits.cpp b/tests/test_traits.cpp new file mode 100644 index 0000000..69b1970 --- /dev/null +++ b/tests/test_traits.cpp @@ -0,0 +1,42 @@ +#include +#include +#include + +using namespace kpn; + +static int free_func(int a, float b) { return a; } +static void sink_func(int a) {} +static std::tuple multi_func(int a) { return {a, 1.f}; } + +TEST_CASE("arity of free function", "[traits]") { + STATIC_REQUIRE(arity_v == 2); +} + +TEST_CASE("return type of free function", "[traits]") { + STATIC_REQUIRE(std::is_same_v, int>); +} + +TEST_CASE("normalised return: single value", "[traits]") { + STATIC_REQUIRE(std::is_same_v, std::tuple>); +} + +TEST_CASE("normalised return: void", "[traits]") { + STATIC_REQUIRE(std::is_same_v, std::tuple<>>); +} + +TEST_CASE("normalised return: tuple passthrough", "[traits]") { + using T = std::tuple; + STATIC_REQUIRE(std::is_same_v, T>); +} + +TEST_CASE("output_count: single return", "[traits]") { + STATIC_REQUIRE(output_count_v == 1); +} + +TEST_CASE("output_count: void return", "[traits]") { + STATIC_REQUIRE(output_count_v == 0); +} + +TEST_CASE("output_count: tuple return", "[traits]") { + STATIC_REQUIRE(output_count_v == 2); +}