From 399ee4cf9b6d900860404723ddda05315d4d092b Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sat, 4 Jul 2026 20:40:35 +0200 Subject: [PATCH 1/4] test: add ThreadSanitizer verification for lock-free Channel Add a contended SPSC stress suite (tests/test_channel_stress.cpp) that actually exercises the ring's memory-ordering pairing and spin/futex/ lost-wakeup logic, plus the CMake and CI plumbing to run it under TSan: - KPN_SANITIZER cache var + kpn_sanitizer_flags() helper (no-op when unset) - kpn_tests_stress executable, labelled "stress" for CTest - reusable tsan.yaml workflow (gcc:14 builder image, already ships libtsan) - ci.yaml gains a tsan job on the same code/dockerfile triggers as test Co-Authored-By: Claude Opus 4.8 --- .gitea/workflows/ci.yaml | 7 + .gitea/workflows/tsan.yaml | 67 ++++++++ CMakeLists.txt | 24 +++ tests/CMakeLists.txt | 31 +++- tests/test_channel_stress.cpp | 288 ++++++++++++++++++++++++++++++++++ 5 files changed, 416 insertions(+), 1 deletion(-) create mode 100644 .gitea/workflows/tsan.yaml create mode 100644 tests/test_channel_stress.cpp diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index 3373cec..e962340 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -73,6 +73,13 @@ jobs: if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }} uses: ./.gitea/workflows/test.yaml + # ThreadSanitizer run for the lock-free Channel. Same trigger conditions as + # test (code or image changed); runs in parallel with test. + tsan: + needs: [changes, docker] + if: ${{ !failure() && !cancelled() && (needs.changes.outputs.code == 'true' || needs.changes.outputs.dockerfile == 'true') }} + uses: ./.gitea/workflows/tsan.yaml + docs: needs: [changes, docker] if: ${{ !failure() && !cancelled() && github.ref == 'refs/heads/master' && (needs.changes.outputs.docs == 'true' || needs.changes.outputs.dockerfile == 'true') }} diff --git a/.gitea/workflows/tsan.yaml b/.gitea/workflows/tsan.yaml new file mode 100644 index 0000000..53f81ad --- /dev/null +++ b/.gitea/workflows/tsan.yaml @@ -0,0 +1,67 @@ +name: '🧡 ThreadSanitizer' + +# Reusable workflow: builds the channel stress suite with ThreadSanitizer and +# runs it. This is the dynamic half of verifying the lock-free SPSC Channel +# (the static half is the CDSChecker model-check harness in verify/). +# +# Triggering and path filtering are owned by ci.yaml (the orchestrator), which +# calls this only when code changed. workflow_dispatch is kept for manual runs. +# +# Runs in the prebuilt builder image (gcc:14), which already ships libtsan β€” no +# package installs at job time. +on: + workflow_call: + workflow_dispatch: + +jobs: + tsan: + runs-on: linux/amd64 + container: + image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + with: + path: tsan-${{ github.run_id }} + + - name: Cache FetchContent dependencies + uses: actions/cache@v3 + with: + path: ~/.cmake/fetchcontent + key: cmake-fetchcontent-${{ hashFiles('**/CMakeLists.txt') }} + restore-keys: cmake-fetchcontent- + + - name: Configure (TSan) + working-directory: tsan-${{ github.run_id }} + run: | + cmake -S . -B build \ + -G Ninja \ + -DCMAKE_BUILD_TYPE=Debug \ + -DKPN_SANITIZER=thread \ + -DKPN_BUILD_TESTS=ON \ + -DKPN_BUILD_EXAMPLES=OFF \ + -DKPN_BUILD_PYTHON=OFF \ + -DFETCHCONTENT_BASE_DIR=$HOME/.cmake/fetchcontent + + - name: Build (TSan) + working-directory: tsan-${{ github.run_id }} + run: cmake --build build --parallel --target kpn_tests kpn_tests_stress + + - name: Run stress suite under TSan + working-directory: tsan-${{ github.run_id }} + # halt_on_error=1 makes the first detected race fail the job; the report + # (with both stacks) is printed to the log. second_deadlock_stack gives + # the full picture for lock-order issues. + env: + TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1" + run: ./build/tests/kpn_tests_stress + + - name: Run unit tests under TSan + working-directory: tsan-${{ github.run_id }} + env: + TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1" + run: ./build/tests/kpn_tests + + - name: Cleanup + if: always() + run: rm -rf tsan-${{ github.run_id }} diff --git a/CMakeLists.txt b/CMakeLists.txt index dd147b5..63037fd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -10,6 +10,30 @@ 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) +# Sanitizer build. Empty = off. Accepts "thread", "address", "undefined", +# or a combination like "address,undefined". Applied to all kpn targets via +# the kpn_sanitizer_flags() helper below. +# +# The lock-free SPSC Channel (include/kpn/channel.hpp) has hand-reasoned +# acquire/release ordering; -DKPN_SANITIZER=thread + the channel stress test +# (tests/test_channel_stress.cpp) is the dynamic half of verifying it. The +# static half is the CDSChecker model-check harness (see verify/). +set(KPN_SANITIZER "" CACHE STRING + "Build with sanitizer: thread | address | undefined | (empty = off)") + +# Translate KPN_SANITIZER into compile/link flags. No-op when empty. +function(kpn_sanitizer_flags out_var) + if(KPN_SANITIZER) + set(${out_var} + -fsanitize=${KPN_SANITIZER} + -fno-omit-frame-pointer + -g + PARENT_SCOPE) + else() + set(${out_var} "" PARENT_SCOPE) + endif() +endfunction() + # ── Core library (header-only) ──────────────────────────────────────────────── add_library(kpn INTERFACE) target_include_directories(kpn INTERFACE diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index a8f7583..fda6711 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -43,6 +43,35 @@ target_link_libraries(kpn_tests PRIVATE GTest::gtest ) +# ── Channel stress suite (separate executable) ──────────────────────────────── +# Contended SPSC tests for the lock-free Channel. Kept out of kpn_tests +# because each case runs many reps / tens of thousands of items and is slow. +# Most valuable under -DKPN_SANITIZER=thread, but correct (and run) without it. +add_executable(kpn_tests_stress test_channel_stress.cpp) +target_link_libraries(kpn_tests_stress PRIVATE kpn Catch2::Catch2WithMain) + +# ── Sanitizer flags ─────────────────────────────────────────────────────────── +# kpn_sanitizer_flags() is defined in the top-level CMakeLists and is a no-op +# unless -DKPN_SANITIZER=... is set. Sanitizer must be on both compile and link. +kpn_sanitizer_flags(_kpn_san) +if(_kpn_san) + foreach(_t kpn_tests kpn_tests_stress) + target_compile_options(${_t} PRIVATE ${_kpn_san}) + target_link_options(${_t} PRIVATE ${_kpn_san}) + endforeach() +endif() + include(CTest) include(Catch) -catch_discover_tests(kpn_tests) + +# DISCOVERY_MODE PRE_TEST defers test enumeration to `ctest` run time. The +# default (POST_BUILD) runs each test binary during the build to list its +# cases β€” which fails a sanitizer build: a TSan/ASan binary needs a fixed +# address-space layout and aborts on startup ("unexpected memory mapping") +# under the container's ASLR, breaking the build before any test runs. The +# tsan.yaml job invokes the binaries directly (not via ctest), so deferring +# discovery costs nothing there and keeps `ctest` working for normal builds. +catch_discover_tests(kpn_tests DISCOVERY_MODE PRE_TEST) +# Register the stress suite under its own label so CI can run / time it +# separately from the fast unit tests. +catch_discover_tests(kpn_tests_stress DISCOVERY_MODE PRE_TEST PROPERTIES LABELS "stress") diff --git a/tests/test_channel_stress.cpp b/tests/test_channel_stress.cpp new file mode 100644 index 0000000..bd113d0 --- /dev/null +++ b/tests/test_channel_stress.cpp @@ -0,0 +1,288 @@ +// Contended stress tests for the lock-free SPSC Channel. +// +// The other channel tests (test_channel.cpp) are single-threaded or use a +// single 20 ms sleep to order two threads β€” they never actually contend on the +// ring, so they exercise neither the memory-ordering pairing nor the +// spin/futex/lost-wakeup logic in pop(). +// +// These tests are written to be run under ThreadSanitizer: +// +// cmake -B build -DKPN_SANITIZER=thread -DKPN_BUILD_EXAMPLES=OFF -DKPN_BUILD_PYTHON=OFF +// cmake --build build --target kpn_tests_tsan +// ./build/tests/kpn_tests_tsan +// +// They are also valid (and meaningful) without a sanitizer: the value/sequence +// assertions catch lost or duplicated items regardless of build flags. TSan +// adds detection of the underlying data race even on runs where the race did +// not corrupt observable state. +// +// Channel is SPSC: exactly one producer thread and one consumer thread per +// channel. Every scenario below honours that contract. + +#include +#include +#include +#include +#include +#include + +using namespace kpn; +using namespace std::chrono_literals; + +namespace { + +// Repeat each scenario enough times that rare interleavings (spin window just +// missing / just catching the next push, disable landing inside the futex +// wait) actually occur across a run. Kept modest so a TSan run stays minutes, +// not hours. +constexpr int kReps = 200; + +} // namespace + +TEST_CASE("SPSC: every pushed item is popped exactly once, in order", "[channel][stress]") { + // Small capacity forces frequent full/empty transitions, so both the + // producer's overflow-retry and the consumer's spin->futex path are hit + // many times. The producer retries on overflow rather than dropping, so + // the consumer must observe a strictly contiguous 0..N-1 sequence. + constexpr int N = 50'000; + Channel ch(/*capacity=*/4, /*spin_count=*/16); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + }); + + int expected = 0; + bool in_order = true; + for (int i = 0; i < N; ++i) { + int v = ch.pop(); + if (v != expected) in_order = false; + ++expected; + } + producer.join(); + + REQUIRE(in_order); + REQUIRE(expected == N); + REQUIRE(ch.size() == 0); +} + +TEST_CASE("SPSC: tight empty<->non-empty transitions exercise spin/futex boundary", + "[channel][stress]") { + // spin_count=0 forces every empty pop() straight into atomic::wait, so this + // hammers the lost-wakeup guard (snapshot wake_, re-check tail_, then wait). + // The producer pushes one item then waits to go empty again, maximising the + // number of empty->non-empty edges relative to item count. + constexpr int N = 20'000; + Channel ch(/*capacity=*/2, /*spin_count=*/0); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + }); + + long sum = 0; + for (int i = 0; i < N; ++i) sum += ch.pop(); + producer.join(); + + // Sum of 0..N-1 β€” detects any lost or duplicated item. + REQUIRE(sum == static_cast(N) * (N - 1) / 2); +} + +TEST_CASE("SPSC: disable() while consumer is blocked in pop() unblocks cleanly", + "[channel][stress]") { + // The data race of record: consumer blocked in pop() (spinning or parked in + // the futex) while the owner thread calls disable(). pop() must observe the + // close and throw ChannelClosedError β€” it must not hang and must not read + // past the ring. Repeated so disable() lands at many points in pop()'s loop. + for (int rep = 0; rep < kReps; ++rep) { + Channel ch(/*capacity=*/4, /*spin_count=*/8); + std::atomic threw{false}; + std::atomic finished{false}; + + std::thread consumer([&] { + try { + ch.pop(); // empty channel: will block + } catch (const ChannelClosedError&) { + threw.store(true, std::memory_order_relaxed); + } + finished.store(true, std::memory_order_relaxed); + }); + + // Give the consumer a chance to reach the wait, then close. + std::this_thread::sleep_for(50us); + ch.disable(); + + consumer.join(); + REQUIRE(finished.load()); + REQUIRE(threw.load()); + } +} + +TEST_CASE("SPSC: producer racing a disable() never throws and never hangs", + "[channel][stress]") { + // Mirror of the above from the producer side: push() racing disable() must + // either enqueue or silently drop, never throw ChannelClosedError and never + // wedge. Overflow is still a legal outcome (full accepting channel) and is + // tolerated here. + for (int rep = 0; rep < kReps; ++rep) { + Channel ch(/*capacity=*/8, /*spin_count=*/8); + std::atomic bad{false}; + + std::thread producer([&] { + for (int i = 0; i < 1000; ++i) { + try { ch.push(i); } + catch (const ChannelOverflowError&) { /* legal: full */ } + catch (...) { bad.store(true, std::memory_order_relaxed); break; } + } + }); + + std::this_thread::sleep_for(20us); + ch.disable(); // owner closes mid-stream + producer.join(); + + REQUIRE_FALSE(bad.load()); + } +} + +TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition", + "[channel][stress]") { + // The empty->non-empty callback ([channel.hpp] was_empty branch) is read by + // the consumer-side notification path. Run it under contention to make sure + // the was_empty detection isn't torn by a concurrent pop(). + Channel ch(/*capacity=*/4, /*spin_count=*/4); + std::atomic callbacks{0}; + ch.set_push_callback([&] { callbacks.fetch_add(1, std::memory_order_relaxed); }); + + constexpr int N = 10'000; + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + }); + + for (int i = 0; i < N; ++i) (void)ch.pop(); + producer.join(); + + // At least one transition, at most one per item; mainly we assert the run + // completed without TSan flagging a race on push_callback_/was_empty. + REQUIRE(callbacks.load() >= 1); + REQUIRE(callbacks.load() <= N); +} + +// What the out-of-band sentinel guarantees under contention β€” and what it does +// not. push_sentinel() publishes has_eof_ (release) after the producer's N ring +// pushes; a consumer that observes has_eof_ (acquire) therefore also observes +// every value pushed before it. What these tests assert: +// +// * Losslessness β€” every value 0..N-1 is delivered exactly once (contiguous, +// no gaps, no duplicates) and the sentinel is delivered exactly once. This +// is the invariant that must hold on every run; a broken acquire/release +// pairing would surface as a lost/duplicated value or (under TSan) a data +// race on has_eof_/eof_value_. +// +// What they deliberately do NOT assert is that the sentinel is the *strictly +// last* item popped. pop() checks emptiness (h == t) using a tail_ snapshot +// taken at the top of its loop; the producer can push more values *and* the +// sentinel in the window before take_sentinel() runs, so the consumer may +// surface the sentinel with a few real values still queued behind it. Those +// values are not lost β€” a consumer that keeps draining still receives them β€” +// but "sentinel arrives dead last" is not a property the channel promises, so +// asserting it would be flaky. We track how many values trailed the sentinel +// for visibility without failing on it. + +TEST_CASE("SPSC: sentinel and all values survive contention (blocking pop)", + "[channel][stress]") { + constexpr int N = 20'000; + constexpr int SENTINEL = -1; + + for (int rep = 0; rep < kReps; ++rep) { + // Small ring + tiny spin window so the ring is frequently empty exactly + // when the sentinel is published β€” the interleaving under test. + Channel ch(/*capacity=*/4, /*spin_count=*/8); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + ch.push_sentinel(SENTINEL); // must-deliver, never overflows/blocks + }); + + std::vector seen(N, false); + int values = 0; + int sentinels = 0; + bool duplicate = false; + // Drain until the sentinel AND all N values have been received; the + // sentinel may arrive before the last few values (see note above). + while (values < N || sentinels == 0) { + int v = ch.pop(); + if (v == SENTINEL) { ++sentinels; continue; } + if (seen[v]) duplicate = true; else seen[v] = true; + ++values; + } + producer.join(); + + REQUIRE_FALSE(duplicate); + REQUIRE(values == N); // every value delivered exactly once + REQUIRE(sentinels == 1); // sentinel delivered exactly once + REQUIRE(ch.size() == 0); + REQUIRE(ch.approx_size() == 0); + } +} + +TEST_CASE("SPSC: sentinel and all values survive contention (try_pop_now)", + "[channel][stress]") { + // The pool-node consume path is try_pop_now(), not pop(): it must surface + // the out-of-band sentinel once the ring is observed empty. The consumer + // spins with no sleeps, racing the producer at full tilt across the + // empty-ring boundary where take_sentinel() is reached. + constexpr int N = 20'000; + constexpr int SENTINEL = -1; + + for (int rep = 0; rep < kReps; ++rep) { + Channel ch(/*capacity=*/4, /*spin_count=*/0); + + std::thread producer([&] { + for (int i = 0; i < N; ++i) { + for (;;) { + try { ch.push(i); break; } + catch (const ChannelOverflowError&) { std::this_thread::yield(); } + } + } + ch.push_sentinel(SENTINEL); + }); + + std::vector seen(N, false); + int values = 0; + int sentinels = 0; + bool duplicate = false; + int v; + while (values < N || sentinels == 0) { + if (!ch.try_pop_now(v)) { std::this_thread::yield(); continue; } + if (v == SENTINEL) { ++sentinels; continue; } + if (seen[v]) duplicate = true; else seen[v] = true; + ++values; + } + producer.join(); + + REQUIRE_FALSE(duplicate); + REQUIRE(values == N); + REQUIRE(sentinels == 1); + // Sentinel held no ring slot; once drained the channel is fully empty. + REQUIRE(ch.size() == 0); + REQUIRE(ch.approx_size() == 0); + } +} -- 2.39.5 From 3ac2242df140e2067d7f85f94976f01ef49c62be Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sat, 4 Jul 2026 20:40:56 +0200 Subject: [PATCH 2/4] docs: rewrite SPEC.md to match the implemented library The spec had drifted far from the code. Key corrections: - Execution model is reactive (PoolNode submits fire_once() to a ThreadPool when inputs are ready), not one blocking thread per node - Channel is a lock-free SPSC ring buffer (atomic wait/notify + spin-before-sleep), not a mutex+CV queue - Remove latch<> ports (never implemented) - NodeErrorHandler returns bool (skip vs stop); per-node - Document new subsystems: scheduler, InterruptNode, Router/FilterNode, MainThreadNode, SharedResource, DebugHub, diagnostics/stats layer - Update StaticNetwork, Python auto_bind layer, examples 01-16 Co-Authored-By: Claude Opus 4.8 --- SPEC.md | 1726 ++++++++++++++++++------------------------------------- 1 file changed, 571 insertions(+), 1155 deletions(-) diff --git a/SPEC.md b/SPEC.md index 5e422ef..8a7d048 100644 --- a/SPEC.md +++ b/SPEC.md @@ -2,9 +2,24 @@ ## 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. +A header-only C++20 template-metaprogramming library for building Kahn Process Networks. Each +node wraps a function (or callable object); its input types are inferred from the parameter +list and its output types from the return type. Nodes communicate over bounded, lock-free +SPSC FIFO channels. + +Unlike a naive "one blocking thread per node" model, KPN++ is **reactive**: a node is +scheduled onto a thread pool whenever all of its input channels have data. A node that wraps a +function with `Node<>` owns a private single-thread pool and behaves exactly like an +independent worker; multiple nodes can instead share one `ThreadPool` for bounded-thread +execution. Source nodes self-resubmit; event-driven sources (`InterruptNode`) fire on an +external trigger. + +The library ships rich runtime diagnostics (per-node exec/CPU/throughput stats, per-channel +fill/bandwidth/overflow counters, pool and shared-resource utilisation), an optional in-process +web debug UI, and nanobind-based Python bindings (partially implemented). + +> **Note on accuracy.** This document describes the code as it exists in `include/kpn/`. Where +> a behaviour is subtle the relevant header is named so the source remains the ground truth. --- @@ -14,39 +29,50 @@ Includes nanobind bindings for Python graph construction and prototyping. 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 +β”‚ β”œβ”€β”€ fixed_string.hpp # NTTP string + in<>/out<> tags + index_of +β”‚ β”œβ”€β”€ traits.hpp # function signature introspection, normalised_return_t, repeat_tuple +β”‚ β”œβ”€β”€ diagnostics.hpp # NodeStats, ChannelStats, *Snapshot, IPoolProbe, IResourceProbe +β”‚ β”œβ”€β”€ channel.hpp # lock-free SPSC ring-buffer Channel + storage policy +β”‚ β”œβ”€β”€ port.hpp # InputPort / OutputPort handles +β”‚ β”œβ”€β”€ inode.hpp # INode interface, NodeErrorHandler, NodeEvent +β”‚ β”œβ”€β”€ scheduler.hpp # IScheduler + work-stealing ThreadPool +β”‚ β”œβ”€β”€ pool_node.hpp # PoolNode / PoolObjectNode (reactive, scheduler-driven) +β”‚ β”œβ”€β”€ interrupt_node.hpp # InterruptNode (external-trigger source) +β”‚ β”œβ”€β”€ node.hpp # Node / ObjectNode (PoolNode + private 1-thread pool) + make_node +β”‚ β”œβ”€β”€ fanout.hpp # FanoutNode + make_fanout +β”‚ β”œβ”€β”€ branch.hpp # RouterNode + FilterNode + make_router / make_filter +β”‚ β”œβ”€β”€ shared_resource.hpp # SharedResource priority-arbitrated exclusive resource +β”‚ β”œβ”€β”€ main_thread_node.hpp # MainThreadNode<> (GUI / main-thread-bound nodes) +β”‚ β”œβ”€β”€ static_network.hpp # Edge<>, make_network(), StaticNetwork<> +β”‚ β”œβ”€β”€ network.hpp # runtime Network builder + watchdog + diagnostics +β”‚ β”œβ”€β”€ debug_hub.hpp # DebugHub multi-network web UI (KPN_WEB_DEBUG only) +β”‚ β”œβ”€β”€ web_debug.hpp # single-network web debug server (KPN_WEB_DEBUG only) +β”‚ β”œβ”€β”€ variant_node.hpp # runtime-typed nodes/channels for Python graphs +β”‚ β”œβ”€β”€ tmp/ +β”‚ β”‚ β”œβ”€β”€ fanout_groups.hpp # compile-time fan-out detection + edge expansion +β”‚ β”‚ β”œβ”€β”€ topo_sort.hpp # compile-time DFS cycle check + topological order +β”‚ β”‚ └── repeat_tuple.hpp # repeat_tuple_t +β”‚ β”œβ”€β”€ python/ +β”‚ β”‚ β”œβ”€β”€ bindings.hpp # PyNetwork / PyNode nanobind helpers +β”‚ β”‚ └── auto_bind.hpp # NodeRegistry / Entry / bind_network / bind_debug +β”‚ └── kpn.hpp # umbrella header +β”œβ”€β”€ src/network.cpp +β”œβ”€β”€ tests/ # Catch2 v3 + GoogleTest +β”œβ”€β”€ examples/ # 01–16 (see Examples) +β”œβ”€β”€ benchmarks/ # bench_pipeline (optional, KPN_BUILD_BENCHMARKS) +└── python/kpn_python.cpp # nanobind module definition ``` +`kpn.hpp` is the umbrella header; including it pulls in the full C++ API (the Python layer is +included only by the binding TU). + --- -## Component 0 β€” `fixed_string.hpp`: NTTP String +## Component 0 β€” `fixed_string.hpp`: NTTP String + Port Tags 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. +`const char*` are not valid NTTPs, so a `fixed_string` literal type provides `constexpr` +internal storage. ```cpp template @@ -57,78 +83,105 @@ struct fixed_string { 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(const char (&)[N]) -> fixed_string; // deduction guide (required) ``` -`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. +`fixed_string<4>` and `fixed_string<7>` are distinct types, so `input<"img">()` and +`input<"sigma">()` produce different instantiations β€” enabling 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: +Named-port lookup uses a `constexpr` `index_of` over the name pack; it returns the sentinel +`npos` on a miss so the `static_assert` fires at the `input<"img">()` **call site**, giving a +readable error at the point of use: ```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(); -} +constexpr std::size_t index_of(); // returns position or npos ``` +### Port tags + +`in<...>` and `out<...>` tag types disambiguate input vs. output name packs in the factory +API. Both are trivial empty structs; both are optional (omit to get index-only ports). + +```cpp +template struct in {}; +template struct out {}; +``` + +> There is **no `latch<>` tag.** An earlier design sketched latched (most-recent-value) +> input ports; this was not implemented and the only input kind is the synchronous one. + --- ## Component 1 β€” `traits.hpp`: Function Introspection -Extracts parameter types and return type from any callable at compile time. +Extracts parameter and return types from any callable at compile time, for free functions, +function pointers, member function pointers (const and non-const), lambdas and `std::function`. ```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) +// function_traits::return_t, ::args (std::tuple<...>), ::arity +template using return_t = ...; // return type +template using args_t = ...; // std::tuple of parameters +template inline constexpr std::size_t arity_v = ...; ``` -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`: +The return type is normalised to a tuple so every node has a uniform output-tuple shape: ```cpp -template -using normalised_return_t = - std::conditional_t, T, std::tuple>; -// void return β†’ std::tuple<> (empty tuple, zero output ports) +// void β†’ std::tuple<> (sink node, 0 outputs) +// T (non-tup) β†’ std::tuple (1 output) +// tuple<...> β†’ tuple<...> (one output port per element) +template using normalised_return_t = ...; +template inline constexpr std::size_t output_count_v = ...; +``` + +`repeat_tuple_t` (also surfaced via `tmp/repeat_tuple.hpp`) builds `std::tuple` +with `N` repetitions β€” used by `FanoutNode` and `RouterNode` to describe their N identical +output ports. + +--- + +## Component 2 β€” `diagnostics.hpp`: Statistics and Snapshots + +Shared timing types: `clock_t = std::chrono::steady_clock`, `duration_t` is a +`double`-millisecond duration. + +- **`NodeStats`** β€” atomic counters updated per fire: `frames_processed`, an EMA of wall-clock + exec time (`ema_exec_us`, warmup-mean for the first 5 frames then Ξ±=0.1), `max_exec_us`, + `total_blocked_us`, thread CPU time (`total_cpu_us` via `CLOCK_THREAD_CPUTIME_ID`), + `queue_wait_us` (pool queue latency), and `exec_start_us` (non-zero while executing; used by + the watchdog to detect hung nodes). +- **`ChannelStats`** β€” `pushes`, `bytes_pushed`, `drops`, `overflows`, `pops`, `peak_fill`. +- **Snapshots** β€” copyable plain structs taken by the watchdog / UI: `NodeSnapshot`, + `ChannelSnapshot` (with `fill_pct()`, `peak_pct()`, `bandwidth_mbs()`), `PoolSnapshot`, + `ResourceSnapshot`, and `NetworkSnapshot` (used by the `DebugHub`). +- **Probe interfaces** β€” `IPoolProbe` and `IResourceProbe` expose a `snapshot(name)` method so + pools and shared resources can be registered with a network for reporting. + +### `ChannelDataSize` trait + +`bytes_pushed` is computed from a specialisable trait, defaulting to `sizeof(T)`. Specialise it +for heap-owning payloads to get accurate bandwidth: + +```cpp +template<> struct kpn::ChannelDataSize { + static std::size_t bytes(const cv::Mat& m) { return m.total() * m.elemSize(); } +}; ``` --- -## Component 2 β€” `channel.hpp`: Bounded FIFO + Storage Policy +## Component 3 β€” `channel.hpp`: Lock-free Bounded FIFO + Storage Policy -### Storage Policy +### Storage policy -The type stored in a channel depends on a specialisable trait. Users can override it for -any type: +The type stored inside a channel is chosen by a specialisable trait. Small trivially-copyable +types are stored by value; everything else as `std::shared_ptr` so fan-out copies a +refcount, not data: ```cpp template @@ -137,29 +190,20 @@ struct channel_storage_policy { 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_storage_policy::by_value, T, std::shared_ptr>; ``` -### Channel +Override it to force value semantics for a custom small type. Push wraps a value in +`make_shared` when needed; pop dereferences it transparently, so a function taking +`const T&` works naturally and immutability is compiler-enforced. -`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. +### Channel β€” SPSC ring buffer -`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. +`Channel` is a single-producer/single-consumer ring buffer (capacity rounded up to a power +of two). It uses C++20 `std::atomic::wait/notify_one` (portable futex) with a configurable +**spin-before-sleep** window so the common case never touches the kernel. ```cpp template @@ -167,590 +211,446 @@ class Channel { public: using storage_type = channel_storage_t; - explicit Channel(std::size_t capacity = 5); + explicit Channel(std::size_t capacity = 5, std::size_t spin_count = 200); - 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); + void push(T value); // drops if disabled; throws ChannelOverflowError if full + bool push_sentinel(T value); // out-of-band, non-blocking must-deliver token (EOF) + T pop(); // blocks (spin then futex); throws ChannelClosedError if disabled+empty + bool try_pop(T& out, std::chrono::milliseconds timeout); // polling (watchdog/display) + bool try_pop_now(T& out); // immediate, non-blocking - std::size_t size() const; + void enable(); // accept pushes + void disable(); // stop accepting + unblock any waiting pop() + void set_push_callback(std::function); // emptyβ†’non-empty notification + + std::size_t size() const; // ring occupancy (excludes any pending sentinel) + std::size_t approx_size() const; // size() + 1 if a sentinel is pending (readiness checks) std::size_t capacity() const; + bool is_accepting() const; + const ChannelStats& stats() const; + ChannelSnapshot snapshot(const std::string& name) const; }; -class ChannelOverflowError : public std::runtime_error {}; +class ChannelOverflowError : public std::runtime_error { /* capacity + optional context */ }; +class ChannelClosedError : public std::runtime_error {}; ``` +`head_` and `tail_`/`wake_` live on separate cache lines (`alignas(64)`) to avoid false +sharing between producer and consumer. `spin_hint()` issues a `pause`/`yield` instruction (or a +compiler fence on other ISAs). + +### The `push_callback` β€” how reactivity works + +`set_push_callback` registers a callback fired when a channel transitions emptyβ†’non-empty. A +consuming `PoolNode` installs this on each of its input channels; when an input becomes ready it +re-evaluates whether **all** inputs have data and, if so, submits itself to the scheduler. This +is the mechanism that replaces a dedicated blocking thread per node. + +### The out-of-band EOF sentinel β€” `push_sentinel` + +`push_sentinel(T value)` delivers a **must-deliver control token** (a graceful-EOF marker) that +cannot be dropped by backpressure. The value is stored in a dedicated slot **outside** the ring, +so it consumes no capacity, never throws `ChannelOverflowError`, and never blocks the producer. + +This matters because a node's worker cannot afford to block on a downstream push: parking that +thread would stop it draining its own input, cascading into a hold-and-wait deadlock under +backpressure. `push_sentinel` sets a published flag (`has_eof_`) and returns immediately, keeping +the worker free to keep popping. + +Ordering is preserved: the consumer's `pop()` / `try_pop_now()` drain the ring **first** and only +surface the sentinel once the ring is observed empty β€” so EOF always arrives after every value +pushed before it. `approx_size()` (used by node readiness checks) counts a pending sentinel as one +consumable item, so a channel carrying *only* a sentinel still schedules its consumer's next fire +and the token is never stranded. Same SPSC contract as `push()` (sole producer); returns `false` +if the channel is already disabled (teardown in progress β†’ the token is moot). + +### Backpressure and shutdown β€” `accepting_` flag + +Each channel carries `std::atomic accepting_` (default `true`). It is the primary shutdown +mechanism; the only additional signal is the out-of-band EOF sentinel above, used for *graceful* +drain rather than an abrupt close. + +- **`push()`** on a disabled channel silently drops the value (recorded as a `drop`). On a + full accepting channel it throws `ChannelOverflowError` (a sizing error). +- **`pop()`** blocks while empty and accepting; `disable()` wakes it and it throws + `ChannelClosedError`. + +The **consumer node** owns its input channels and flips the flag: `start()` calls `enable()`, +`stop()` calls `disable()`. Producers never touch it. + ### 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. +Input channels are owned by their **consumer node** (held as `shared_ptr>`). A +producer node holds a non-owning raw `Channel*` to push into. `Network`/`StaticNetwork` are +otherwise non-owning of user nodes β€” see Components 8–9. -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 +## Component 4 β€” `inode.hpp`: The Node Interface -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. +Every node implements `INode`: ```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 - } +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) = 0; - 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() - } + virtual void set_network_overflow_callback(NodeEventCallback) {} // network-injected + virtual void set_network_closed_callback(NodeEventCallback) {} + + virtual void halt() { stop(); } // immediate, discard in-flight work + virtual void shutdown() { stop(); } // graceful topo-ordered drain (overridden by networks) }; ``` -**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. +Supporting types: -**Overflow** (`push()` on a full, accepting channel) still throws `ChannelOverflowError` β€” -this signals a design error (undersized FIFO) and is unchanged. +```cpp +// Per-node error policy: return true to skip the failed fire and keep running, +// false to stop the node (and signal closed downstream). +using NodeErrorHandler = std::function; -**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. +using NodeEventCallback = std::function; +enum class NodeEvent { Overflow, Closed }; +``` --- -## Component 3 β€” `port.hpp`: Port Handles +## Component 5 β€” `scheduler.hpp`: Thread Pool ```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. +struct IScheduler { + virtual void submit(std::function task, float priority = 0.5f) = 0; + virtual void start() = 0; + virtual void stop() = 0; // join workers, discard pending tasks + virtual void drain() = 0; // block until in-flight tasks complete (workers keep running) }; ``` -**Factory syntax β€” `in<>` / `latch<>` / `out<>` tag structs:** - -A flat name pack `make_node` is ambiguous (where do inputs end?). -Option chosen: `in<...>`, `latch<...>`, and `out<...>` tag types that wrap the name packs unambiguously. -All are optional; omitting either means those ports are index-only. - -```cpp -// Tag types (trivial, no data): -template struct in {}; -template struct latch {}; -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); - -// Mixed synchronous and latched inputs -auto node = make_node, latch<"setpoint">, out<"output">>(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). - -### Node identity: `Label` and `UniqueTag` NTTPs - -Two problems arise when using `Node` as a compile-time graph vertex: - -1. **Debug names** β€” without a label the web UI and `print_diagnostics` fall back to - `"node[0]"` placeholders. `set_name()` provides a runtime name for the runtime - `Network`, but `StaticNetwork` has no `add("name", node)` call to attach one. - -2. **Same-function collision** β€” two nodes wrapping the same function (e.g. two - `make_node`) have the same type. The `StaticNetwork` topo sort and fanout - detection use node types as graph vertices, so it cannot distinguish them, causing - infinite recursion in the DFS. - -Both are solved by two additional NTTPs on `Node`: - -```cpp -template< - auto Func, - typename InputTag = in<>, - typename LatchTag = latch<>, - typename OutputTag = out<>, - fixed_string Label = "", // human-readable name; shown in diagnostics/web UI - std::size_t UniqueTag = 0 // collision-breaker; must differ between any two -> // same-Func nodes in one make_network() call -class Node : public INode { ... }; -``` - -Both have defaults so all existing code (`make_node(5)`) compiles unchanged. - -**Factory syntax extension:** - -```cpp -// Label only (UniqueTag defaults to 0) -auto blur = make_node(8); - -// Both label and unique tag β€” required when the same function is used twice -auto preblur = make_node(8); -auto postblur = make_node(8); -``` - -**Duplicate-tag detection** β€” `make_network()` checks at compile time that no two nodes -in the edge pack share the same `(Func, UniqueTag)` pair. This fires a readable -`static_assert` at the call site before any thread is started: - -``` -static_assert(!has_duplicate_tags_v, - "make_network: two nodes have identical (Func, UniqueTag) β€” increment UniqueTag " - "on one of them to make them distinct"); -``` - -**Label availability** β€” `Node` exposes the label as a `static constexpr` so -`StaticNetwork` can read it at compile time for diagnostics: - -```cpp -static constexpr std::string_view label() { return Label.view(); } -``` - -The web debug UI and `print_diagnostics` use `label()` when non-empty, falling back to -`"node[]"` for unlabelled nodes. The runtime `Network` continues to use the -string passed to `add("name", node)` β€” `Label` is orthogonal to that mechanism. - -**`ObjectNode`** gains the same two NTTPs with the same defaults. The `make_node(obj, ...)` -overloads are extended identically. +`ThreadPool` is a **work-stealing** pool implementing both `IScheduler` and `IPoolProbe`. Each +worker owns a priority queue (max-heap by `priority`, FIFO within equal priority via a sequence +counter). `submit()` distributes round-robin; idle workers steal from the most-loaded peer +using `try_lock`, then sleep on a shared condition variable. The submit/notify path takes the CV +mutex around `notify` to close the lost-wakeup window; `drain()` waits on a separate counter of +in-flight tasks. `priority` lets a hot node (full input, empty output) be scheduled ahead of +others β€” see `PoolNode::compute_priority`. --- -## Component 4a β€” Latched Input Ports +## Component 6 β€” Node Types -### Motivation +All processing nodes share the same shape: typed input channels they own, raw output-channel +pointers set at wiring time, `args_tuple` / `return_tuple` aliases used by the connect-time +type check, and `static constexpr` `label()` / `unique_tag` / `input_count` / `output_count`. -Control and robotics applications naturally have two kinds of inputs at different update rates: +### `PoolNode` / `PoolObjectNode` β€” reactive, scheduler-driven (`pool_node.hpp`) -- **Synchronous inputs** (`in<>`) β€” the node must have fresh data on every fire. Typical for sensor readings that drive the computation (e.g. encoder RPM). -- **Latched inputs** (`latch<>`) β€” the node uses the most recently received value, and does not block if no new value has arrived. Typical for setpoints or parameters that change infrequently relative to the control loop (e.g. bearing from a CV pipeline, PID gains). - -Without latched ports, a node must block on all inputs simultaneously. This forces the control loop to run at the rate of the slowest input β€” unacceptable when a 1kHz encoder loop must wait for a 30Hz vision update. - -### Semantics - -A `latch<>` port: - -1. **Does not block** if its channel is empty β€” it reuses the last successfully popped value. -2. **Does block on first fire** β€” there is no meaningful "default" value, so the node waits until at least one value has arrived on each latched port before firing for the first time. -3. **Consumes the value** when one is available (standard `pop()`), then holds it until the next value arrives. - -The node fires whenever all `in<>` ports have data, using the last known value for each `latch<>` port. - -### Implementation in `run_loop` - -`run_loop` maintains a `std::tuple` of cached values, one slot per latched port. On each iteration: +The core node. Instead of a blocked thread, it submits a `fire_once()` to a shared +`IScheduler` whenever all inputs are ready; `queued_` ensures at most one `fire_once()` is +in flight. `fire_once()` pops every input (`try_pop_now`), runs the function, pushes each +normalised output, records stats, then resubmits if inputs remain ready. Source nodes +(`input_count == 0`) self-submit on `start()` and after each fire. ```cpp -// Synchronous ports β€” blocking pop (existing behaviour) -auto sync_args = std::make_tuple(input<0>().pop(), input<1>().pop(), ...); +template, + typename OutputTag = out<>, + fixed_string Label = "", + std::size_t UniqueTag = 0> +class PoolNode : public INode { ... }; -// Latched ports β€” non-blocking try_pop; keep cached value on miss -try_pop(latch_cache_, latch_channel_); // updates cache if data available - -// Call wrapped function with merged argument tuple -auto result = std::apply(Func, merge(sync_args, latch_cache_)); +auto n = make_pool_node(scheduler, fifo_capacity); // index ports +auto n = make_pool_node(scheduler, in<"a">{}, out<"b">{}, cap); ``` -Latched channels are otherwise identical to synchronous channels: bounded FIFO, `shared_ptr` storage policy, same shutdown behaviour. +`PoolObjectNode` is the same for a stateful callable object (introspected via +`&Obj::operator()`); the object must outlive the node. -### Example β€” PID with live setpoint +Per-node configuration: `set_error_handler(NodeErrorHandler)`, `set_overflow_callback`, +`set_closed_callback`, `set_max_exec_time`. Inside `fire_once()`: +`ChannelOverflowError` fires the overflow callbacks; `ChannelClosedError` (or an error handler +returning `false`) fires the closed callbacks and self-stops; any other exception consults the +error handler. + +**Name-count contract** β€” a `static_assert` requires that the number of input names is `0` or +equals arity (same for outputs): ```cpp -// bearing arrives at ~30 Hz from CV; rpm arrives at ~1 kHz from encoder -double pid_compute(double rpm, double bearing) { ... } - -auto pid = make_node, // synchronous β€” blocks until fresh encoder tick - latch<"bearing">, // latched β€” uses last known bearing from CV - out<"pwm"> ->(8); - -Network net; -net.add("tacho", tacho_node) - .add("tracker", tracker_node) - .add("pid", pid) - .connect("tacho", tacho_node.output<"rpm">(), "pid", pid.input<"rpm">()) - .connect("tracker", tracker_node.output<"bearing">(),"pid", pid.input<"bearing">()) - .build(); +static_assert(sizeof...(InNames) == 0 || sizeof...(InNames) == input_count, + "make_pool_node: number of input names must match function arity, or provide none"); ``` -The PID node fires at encoder rate. If no new bearing has arrived since the last tick, it reuses the previous one β€” correct behaviour for a control loop. +### `Node` / `ObjectNode` β€” convenience wrappers (`node.hpp`) -### Port Ordering Contract - -`in<>` and `latch<>` ports together must cover all function parameters in declaration order. The `static_assert` on name count is extended to cover both tags jointly: +`Node<>` privately owns a `ThreadPool(1)` and derives from `PoolNode<>` with the **same** +template signature, so each `Node` is a self-contained worker with no external scheduler. Its +`start()`/`stop()` start and stop the private pool around the base. This keeps the simple API β€” +`make_node(5)` β€” while routing all execution through the one `fire_once()` code path. ```cpp -static_assert( - sizeof...(InNames) + sizeof...(LatchNames) == input_count || - (sizeof...(InNames) == 0 && sizeof...(LatchNames) == 0), - "make_node: in<> and latch<> names together must match function arity, or provide none" -); +template, typename OutputTag = out<>, + fixed_string Label = "", std::size_t UniqueTag = 0> +class Node : public PoolNode<...> { ... }; + +auto src = make_node(5); +auto dbl = make_node(5); +auto cnt = make_node(in<"words">{}, out<"count","words">{}, 4); ``` -The function parameter at position `i` is synchronous if `i` is in the `in<>` pack, latched if in the `latch<>` pack. Mixed ordering is allowed β€” the tag packs define which positions are latched, not a contiguous suffix. +The `Label` NTTP gives a human-readable name for diagnostics; `UniqueTag` is a collision-breaker +required when the **same function** is used as two distinct vertices in a `StaticNetwork` (two +`make_node` would otherwise be the same type). Both default so existing code is unaffected. +To share one pool across many nodes for bounded-thread execution, use `make_pool_node` directly. + +### `InterruptNode` β€” external-trigger source (`interrupt_node.hpp`) + +A zero-input source driven by an external event (camera frame, timer, socket) instead of +self-resubmission. `get_trigger()` returns a thread-safe callable to hand to the event source; +each call increments a `pending_` counter and submits `fire_once()` on the 0β†’1 transition, +guaranteeing one execution per trigger even under bursts. It does not busy-loop. + +```cpp +auto cam = make_interrupt_node(scheduler, out<"frame">{}); +camera_sdk.on_frame_ready(cam.get_trigger()); +``` + +### `FanoutNode` β€” explicit fan-out (`fanout.hpp`) + +Reads one item and pushes a copy to each of N outputs (per-output overflow drops +independently). Runs on its own `std::jthread` blocking on `pop()`. Used directly in a runtime +`Network` via `make_fanout`, and auto-inserted by `make_network()` for `StaticNetwork`. + +### `RouterNode` / `FilterNode` β€” branching (`branch.hpp`) + +Both run on a dedicated `jthread`. `RouterNode` pushes each item to exactly **one** of N +outputs chosen by a `selector(item) -> size_t` (out-of-range index drops). `FilterNode` forwards +an item only when `pred(item)` is true. Factories: `make_router(sel)`, `make_filter(pred)`. + +### `MainThreadNode, Args…>` β€” GUI / main-thread nodes (`main_thread_node.hpp`) + +For work that *must* run on the thread owning a GUI event loop (OpenCV `imshow`/`waitKey` on +Wayland/Qt). It owns input channels and is registered as a normal `INode` (appears in +diagnostics) but spawns **no** thread. The application drives it by calling `step()` in a loop on +the main thread: `step()` does a zero-timeout `try_pop` on every input, and when all are ready +invokes the derived `operator()(Args…)` (returning `false` to stop). CRTP; the derived class +supplies the operator. --- -## Component 5 β€” `network.hpp`: Graph Builder + Orchestrator +## Component 7 β€” `shared_resource.hpp`: Priority-arbitrated Exclusive Resource -`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. +`SharedResource` wraps a singleton-like resource (an ONNX session, a CUDA stream) shared by +nodes across one or more networks, and arbitrates access with a **priority + aging** waiter +queue. Priority is re-evaluated at every release (so it reflects current queue state), and each +waiter's effective score grows with wait time (`kAgingPerSecond`) to prevent starvation. ```cpp -class Network : public INode { // Network is itself an INode β€” enables sub-networks +SharedResource res(session_args...); + +// inside a node functor: +auto guard = res.acquire_balanced(in_channel, out_channel); // RAII; releases on scope exit +guard->Run(...); +``` + +`acquire_balanced(in, out)` scores a waiter by `input_fill Γ— output_headroom` β€” a node with a +full input queue and empty output is most urgent. `acquire(fn)` takes any `()->float` priority; +`acquire()` treats all waiters equally. Implements `IResourceProbe` so it shows up in +diagnostics and the debug hub. The factory is `make_shared_resource(args…)`. + +--- + +## Component 8 β€” `network.hpp`: Runtime Graph Builder + Watchdog + +`Network` is **non-owning** (nodes outlive it; `add()` stores `INode*`). A builder collects the +full topology before `build()`, enabling cycle detection and topological ordering. + +```cpp +class Network : public INode { public: - // Register a node by name. NodeT must satisfy INode. Network holds a raw pointer. - template - Network& add(std::string name, NodeT& node); + 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); + template + Network& connect(const std::string& src, OutputPort, + const std::string& dst, 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_input (std::string boundary_name, InputPort); // sub-network port Network& expose_output(std::string boundary_name, OutputPort); - // DFS cycle check + topological sort. Throws NetworkCycleError on cycles. - Network& build(); + Network& build(); // DFS cycle check (throws NetworkCycleError) + topo sort - 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 start() override; // start nodes in topo order; launch watchdog (+ web UI) + void stop() override; // == halt() + void halt() override; // immediate: stop nodes in reverse topo order + void shutdown() override; // graceful: stop source layers, drain channels, descend void set_watchdog_interval(std::chrono::milliseconds); + void set_error_handler(ErrorHandler); // void(node_name, exception_ptr) + void set_diagnostics_handler(DiagnosticsHandler); // fired each watchdog tick + void set_event_handler(EventHandler); // void(name, NodeEvent, timestamp) + void register_pool(const std::string&, IPoolProbe*); - 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_; + void print_diagnostics(std::ostream& = std::cerr) const; // formatted table }; ``` -**Node lifetime contract:** nodes must outlive the `Network`. The typical pattern is to -declare nodes and the network in the same scope: +- **`connect`** static-asserts that the source output type equals the destination input type + (via the nodes' `return_tuple` / `args_tuple`), sets the consumer's input channel as the + producer's output pointer, registers a `ChannelProbe` for diagnostics, and rejects a second + connection from the same output port (use `make_fanout`). +- **`build`** colours the graph DFS; a back-edge throws `NetworkCycleError`. It also wires each + node's network-level overflow/closed callbacks to the `EventHandler` if one is set. +- **`halt` vs `shutdown`** β€” `halt()` disables channels and stops nodes in reverse order + immediately; `shutdown()` walks source layers first, polling channel probes until they drain + before stopping the next layer. +- **Watchdog** β€” a `std::jthread` that wakes on `watchdog_interval_` (default 3 s), collects + snapshots, warns about nodes whose `exec_start_us` indicates an execution running > 5 s, and + either calls the diagnostics handler or prints the formatted report. +- **`expose_input`/`expose_output`** record boundary names (sub-network support is scaffolded; + `Network` is itself an `INode` and can be `add()`ed to an outer `Network`). -```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. +The formatted report includes node (frames, exec ms, max ms, blocked ms, fps, cpu ms, util%), +channel (fill%, peak%, pushes, drops, overflow, MB/s, item bytes), and pool tables, plus a +bottleneck hint (highest `ema_exec_ms`). --- -## Component 6 β€” `variant_node.hpp`: Runtime-typed Node (Python graphs) +## Component 9 β€” `static_network.hpp`: Compile-time Graph Builder -### 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: +For C++ graphs whose full topology is known at compile time. The complete edge list is a type +pack, so fan-out arity is known up front, cycle detection is a `static_assert`, and start/stop +are pointer-vector traversals rather than string-map + virtual dispatch. ```cpp -template -using py_variant_t = std::variant>>; +// edge() builds a typed Edge descriptor from two port handles. +template +Edge +edge(OutputPort, InputPort); + +// make_network() takes all edges, expands fan-outs, wires channels, returns a StaticNetwork. +template auto make_network(Edges&&... edges); ``` -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. +Usage β€” no `add`/`connect`/`build`/string names; one source port feeding two destinations +auto-inserts a `FanoutNode`: ```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. +auto src = make_node(8); +auto blur = make_node(8); +auto detect = make_node(8); +auto sink = make_node(8); + +auto net = make_network( + edge(src.output<0>(), blur.input<0>()), + edge(src.output<0>(), detect.input<0>()), // same source port β†’ FanoutNode inserted + edge(blur.output<0>(), sink.input<0>()), + edge(detect.output<0>(), sink.input<1>())); +net.start(); /* … */ net.stop(); ``` -### VariantChannel +`make_network` performs, at compile time: fan-out detection and edge expansion +(`tmp/fanout_groups.hpp`), a duplicate-`(Func, UniqueTag)` check +(`static_assert` β€” "add a UniqueTag"), and a cycle check + topological order +(`tmp/topo_sort.hpp`, `static_assert` β€” "graph contains a directed cycle"). At run time it +heap-allocates owned `FanoutNode` storage, collects user-node pointers in edge order, sets each +node's display name (`Label`, else `node[UniqueTag]`; fan-outs become `"_fanout"`), wires +every expanded edge, and builds channel probes. -```cpp -using VariantValue = py_variant_t; +`StaticNetwork` implements `INode` (so it can be embedded in a +runtime `Network`). It owns the fan-out nodes, holds user nodes by pointer, and provides +`start`/`halt`/`shutdown`, an `EventHandler`, `register_resource` / `register_pool`, +`print_diagnostics`, and `network_snapshot()` (consumed by the `DebugHub`). Compile-time labels +are read from each `NodeType::label()`. -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) -``` +> `make_fanout` remains for explicit fan-out in a runtime `Network`; `make_network` users +> never call it. --- -## Component 7 β€” Orchestrator / Watchdog +## Component 10 β€” Web Debugging (optional, `KPN_WEB_DEBUG`) -Runs in its own dedicated thread inside `Network` / `PyNetwork`. Responsibilities: +Zero cost when disabled β€” guarded headers, no symbols, no dependency. Depends on **cpp-httplib** +(single-header, fetched by CMake when the option is on) and loads **D3.js v7** from CDN. Enable +per-target: -- 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 +```cpp +#define KPN_WEB_DEBUG 1 +#include +``` + +### Single-network server (`web_debug.hpp`) + +When enabled, `Network` / `StaticNetwork` gain `set_web_debug_port(uint16_t)` (default 9090) and +auto-start an in-process HTTP server in `start()`. It serves an inline single-page app at `/` and +a JSON snapshot at `/api/snapshot` (nodes, channels/edges, pools, resources, elapsed). The page +renders a force-directed graph: node colour encodes `ema_exec_ms`, edge colour encodes fill%, +with hover tooltips for the full stat set; it polls every 500 ms. + +### `DebugHub` β€” multi-network UI (`debug_hub.hpp`) + +A standalone server aggregating several networks under one endpoint: + +```cpp +DebugHub hub(9090); +hub.register_network("detect", detect_net); // disables that net's own server +hub.register_network("classify", classify_net); +hub.register_resource("gpu", &gpu_resource); // shows utilisation cards +hub.start(); +``` + +The hub UI has one tab per registered network plus an "All Networks" tab with shared-resource +cards and a cross-network node table. `register_network` calls `net.disable_web_server()` so the +hub is the single debug endpoint; call it before `net.start()`. --- -## GIL Rules (non-negotiable constraints on binding implementation) +## Component 11 β€” Python Bindings (partial) -Two rules govern all interaction between node threads and the Python interpreter: +> Status: scaffolded and partially implemented. The variant machinery, `PyNetwork`/`PyNode`, and +> the auto-binding layer exist; the demo module wires a hello-pipeline. Full sub-port read/write +> and mixed C++/Python graphs are still in progress. -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). +Python graphs cannot resolve types at compile time, so a `PyNetwork` is parameterised by a +`std::variant` derived (at compile time, via `unique_types`) from the port types of a **closed +list of registered C++ node types**. The variant only appears at the C++/Python boundary; each +node's internal `Channel` still stores raw `T` (`variant_node.hpp`: `IVariantChannel`, +`VariantChannel`, `IVariantNode`, `VariantNodeWrapper`). -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. +### Auto-binding (`python/auto_bind.hpp`) -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. +The node list is declared once with a `NodeRegistry` of `Entry`. `bind_network` +registers the `PyNetwork` class, a `make_(capacity)` factory and a `Node` class per +entry, and auto-registers `PythonConverter` for each port type. `bind_debug` additionally exposes +each raw C++ function as a free Python callable for testing without a network. Recompiling the +extension is the registration step β€” there is no CMake code-gen. + +```cpp +using DemoNodes = kpn::python::NodeRegistry< + kpn::python::Entry, + kpn::python::Entry, + kpn::python::Entry>; // variant auto-deduced as std::variant + +NB_MODULE(kpn_python, m) { + bind_network(m); + bind_debug(m); +} +``` + +Custom types are supported by specialising `kpn::PythonConverter` (`to_python` / `from_python`, +optional `type_name`) before `bind_network`. + +### GIL rules (non-negotiable) + +1. **Acquire for callback** β€” hold the GIL only for the duration of a Python callable + invocation (`nb::gil_scoped_acquire` around the call site). +2. **Release while blocking** β€” release the GIL before any blocking channel op + (`nb::gil_scoped_release`), then re-acquire. Violating this deadlocks: a PyNode thread + waiting for the GIL cannot proceed while another thread holds it and blocks on a channel + waiting for that PyNode. --- @@ -758,616 +658,132 @@ another thread holds the GIL and blocks on a channel waiting for that PyNode to | 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` | +| FIFO overflow (full, accepting) | `ChannelOverflowError` thrown in producer; node overflow callbacks fire | +| Push to a disabled channel | Value silently dropped (counted as a `drop`) | +| Node function throws | Routed to the node's `NodeErrorHandler` β†’ `true` skips & continues, `false` stops the node | +| Node stopped / channel closed | `ChannelClosedError` β†’ node fires closed callbacks and self-stops | +| Type mismatch (C++) | `static_assert` at `connect()` / `make_network()` | +| Cycle in graph (runtime) | `NetworkCycleError` thrown at `build()` | +| Cycle in graph (static) | `static_assert` at `make_network()` | +| Duplicate `(Func, UniqueTag)` (static) | `static_assert` at `make_network()` β€” add a `UniqueTag` | +| Hung node | Watchdog warning after threshold | ---- - -## 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. +`Network` additionally exposes an aggregate `EventHandler(name, NodeEvent, timestamp)` for +overflow/closed events across all nodes. --- ## 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). +KPN++ is **reactive**, not one-thread-per-node: -`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 `PoolNode` owns no thread. It registers a push-callback on each input channel; when all + inputs are ready it submits `fire_once()` to a shared `IScheduler` (a `ThreadPool`). +- `Node<>` wraps a `PoolNode` plus a **private `ThreadPool(1)`**, recovering "independent + worker" semantics with the simple `make_node` API. Many nodes can instead share one pool + (`make_pool_node`) for a bounded OS thread count. +- `FanoutNode`, `RouterNode`, and `FilterNode` do run a dedicated `std::jthread` blocking on + `pop()` (they are simple, latency-sensitive routers). +- `InterruptNode` fires on an external trigger; `MainThreadNode` runs on the caller's main + thread via `step()`. -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. +`std::jthread` (C++20) and its `stop_token` are used where a thread is owned, simplifying +cooperative shutdown. Benchmarks (`benchmarks/bench_pipeline`) show ~2–7 Β΅s/hop framework +overhead for chains within the core count, rising under oversubscription. --- ## Platform and Compiler Requirements -C++20 is required. Specific features used: +C++20 is required. -| 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 | +| Feature | Min compiler | +|---|---| +| NTTP structural types (`fixed_string`) | GCC 11, Clang 13, MSVC 19.29 | +| `std::atomic::wait/notify` (channel futex) | GCC 11, Clang 13, MSVC 19.29 | +| `std::jthread` + `stop_token` | GCC 11, Clang 14, MSVC 19.29 | +| `auto` NTTPs, fold expressions, `if constexpr`, concepts | C++20 / C++17 baseline | -**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). +`CLOCK_THREAD_CPUTIME_ID` (per-thread CPU stats in `diagnostics.hpp`) is POSIX. nanobind +requires Python 3.8+ (auto-fetched when `KPN_BUILD_PYTHON=ON`). --- ## 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. +**Catch2 v3** for behaviour/integration tests and **GoogleTest** for unit and death tests; both +are auto-fetched. Existing suites: `test_fixed_string`, `test_traits`, `test_channel`, +`test_node`, `test_network`, `test_static_network`, `test_scheduler`, `test_pool_node`, +`test_shared_resource`. -### 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 | +Cases covered explicitly include: channel blocking/unblocking and overflow; shutdown races +(`stop()` while blocked on `pop()`); `try_pop_now`; fan-out delivery; tuple unpacking to +sub-channels; runtime cycle detection and static cycle/duplicate-tag `static_assert`s; named +port lookup and wrong-name-count `static_assert`s; storage-policy by-value vs `shared_ptr`; +scheduler submit/steal/drain; `PoolNode` reactive scheduling; and `SharedResource` priority + +aging. --- ## 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. +Self-contained programs under `examples/`, built by default (`-DKPN_BUILD_EXAMPLES=OFF` to +skip). They double as 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 | - ---- - -## Component 9 β€” `static_network.hpp`: Compile-time Graph Builder - -### Motivation - -The runtime `Network` builder has two limitations that only a compile-time graph can fix: - -1. **Fan-out `N` is unknowable at the first `connect()` call.** A `FanoutNode` requires - `N` as a template parameter. With runtime `connect()`, the network has seen only one edge - when the first call arrives; it cannot know how many more will follow for that port. Auto- - inserting the right `FanoutNode` requires seeing the complete edge list at once β€” - which is only possible if the edge list is a type. - -2. **Start/stop goes through virtual dispatch.** `Network` stores `INode*` and calls virtual - `start()`/`stop()`. With a typed node tuple the compiler sees the concrete types and can - inline or devirtualise. This matters at startup/shutdown, not in the hot path β€” but it is - avoidable overhead. - -The runtime `Network` is **not removed**. It remains the right choice for Python graphs, -sub-networks embedded in dynamic topologies, and any case where the graph shape is not known -until runtime. `StaticNetwork` is an additional builder for the common case where the full -C++ topology is known at compile time. - -### API - -```cpp -// edge() constructs a typed edge descriptor from two port handles. -// All type information (SrcNode, SrcIdx, DstNode, DstIdx) is in the return type. -template -auto edge(OutputPort, InputPort) - -> Edge; - -// make_network() accepts all edges as a variadic pack. -// It deduces the full topology, auto-inserts FanoutNodes where needed, -// wires all channels, and returns a StaticNetwork owning the fanout nodes. -// User nodes are held by reference (non-owning), same lifetime contract as Network. -template -auto make_network(Edges&&... edges) -> StaticNetwork<...>; -``` - -Usage: - -```cpp -auto src = make_node(8); -auto blur = make_node>(8); -auto detect = make_node>(8); -auto sink = make_node(8); - -// src:output<0> feeds both blur and detect β€” fan-out is auto-inserted -auto net = make_network( - edge(src.output<0>(), blur.input<0>()), - edge(src.output<0>(), detect.input<0>()), // same source port - edge(blur.output<0>(), sink.input<0>()), - edge(detect.output<0>(), sink.input<1>()) -); -net.start(); -// ... -net.stop(); -``` - -No `add()`, no `build()`, no string names. The graph is fully wired in the `make_network` -call. `start()` and `stop()` are non-virtual tuple traversals. - -### Edge type - -```cpp -// Carries references to the two endpoint nodes. Stores no data beyond that. -template -struct Edge { - SrcNode& src; - DstNode& dst; -}; -``` - -### Fan-out detection metafunction - -`make_network` receives `Edge<...>` types as a pack. Before wiring, a metafunction scans -the pack for output ports with more than one downstream edge: - -``` -fanout_groups -``` - -This is a compile-time multimap: keys are `(SrcNode type, SrcIdx)`, values are the list of -destination `(DstNode type, DstIdx)` pairs sharing that key. - -For each key with N > 1 destinations: -- Compute `T = std::tuple_element_t` -- Synthesise a `FanoutNode` β€” call it `F` -- Replace the N original edges with: - - one edge: `src:SrcIdx β†’ F:input<0>` - - N edges: `F:output<0..N-1> β†’ original dst:DstIdx` - -For keys with N == 1 the edge is kept as-is. - -The result is an expanded edge list with all fan-outs made explicit, and a list of -`FanoutNode` types that need to be instantiated. - -### `StaticNetwork` structure - -```cpp -template owned by the network - typename TopoOrder> // index_sequence encoding start/stop order -class StaticNetwork : public INode { -public: - void start(); // std::apply over TopoOrder β€” no virtual dispatch, no map lookup - void stop(); // reverse of TopoOrder - - bool running() const; - - // Diagnostics β€” iterates typed tuples; same NodeSnapshot / ChannelSnapshot output - // as Network, compatible with print_diagnostics and the web debug UI. - void print_diagnostics(std::ostream& = std::cerr) const; - - // StaticNetwork is itself an INode, so it can be embedded in a runtime Network - // via net.add("stage", static_net) exactly like any other node. - void set_name(std::string) override; - const NodeStats& stats() const override; - NodeSnapshot node_snapshot(const std::string&, double) const override; - -private: - FanoutStorage fanouts_; // owns the auto-generated FanoutNode instances - // User nodes held by reference β€” same non-owning contract as Network -}; -``` - -`FanoutStorage` is a `std::tuple, FanoutNode, ...>` with one -element per auto-inserted fanout. It is owned by the `StaticNetwork` and lives as long as -the network does β€” which satisfies the channel lifetime contract (channels are owned by their -consumer, and the fanout node is the consumer of the upstream output). - -### Cycle detection - -With the full edge list as a type pack, cycle detection is a `static_assert` rather than a -runtime exception. A compile-time DFS over the expanded edge list fires a readable assertion -at the `make_network` call site: - -``` -static_assert(!has_cycle_v, - "make_network: graph contains a directed cycle"); -``` - -`NetworkCycleError` is no longer needed for `StaticNetwork` β€” the cycle is caught before any -object is constructed. - -### Topological order - -The same compile-time DFS produces a topological ordering as an `std::index_sequence` over -the node tuple. `start()` iterates it forward, `stop()` iterates it in reverse. No runtime -sort, no `std::vector`. - -### Node labels for diagnostics / web debug - -Labels come directly from the `Label` NTTP on each `Node` type β€” no separate annotation -on `edge()` is needed. `StaticNetwork` reads `NodeType::label()` at compile time for each -vertex in the topo order and stores the result as a `std::string_view` array at -construction time. Zero runtime overhead: the label is a compile-time string literal. - -```cpp -auto src = make_node(8); -auto blur = make_node(8); -auto detect = make_node(8); - -auto net = make_network( - edge(src.output<0>(), blur.input<0>()), - edge(src.output<0>(), detect.input<0>()) -); -// web UI shows nodes named "src", "blur", "detect" -// auto-inserted FanoutNode is labelled "fanout[src:0]" -``` - -Unlabelled nodes (`Label == ""`) fall back to `"node[]"` in diagnostics. -Auto-inserted fanout nodes are labelled `"fanout[:]"` automatically. - -### Wiring sequence in `make_network` - -All wiring happens in the `make_network` constructor body β€” no `build()` call needed: - -1. Instantiate `FanoutStorage` (default-construct each `FanoutNode`). -2. For each expanded edge (in topological order): - - Call `src.set_output_channel(&dst.input_channel())`. -3. Return the `StaticNetwork`. - -Channel pointers are set once and never changed. No dynamic allocation after construction. - -### What is eliminated vs `Network` - -| `Network` (runtime) | `StaticNetwork` (compile-time) | +| Example | What it shows | |---|---| -| `std::map` | typed `std::tuple` of references | -| Runtime DFS + `NetworkCycleError` | `static_assert` at `make_network` call site | -| Virtual `start()`/`stop()` per node | `std::apply` over typed tuple | -| Explicit `make_fanout` | auto-inserted from edge pack | -| `connected_outputs_` duplicate check | structural impossibility β€” no duplicate edge can produce two `set_output_channel` calls | -| `build()` step | no build step β€” wired in constructor | +| `01_hello_pipeline` | Linear pipeline, index-based wiring, `Network` builder | +| `02_named_ports` | `in<>`/`out<>` tags, named port access, wrong-name `static_assert` | +| `03_multi_output` | Tuple-returning node, per-element sub-port routing | +| `04_storage_policy` | `channel_storage_policy` default + specialisation | +| `05_error_handling` | `ChannelOverflowError`, diagnostics handler | +| `06_watchdog` | Watchdog interval, stall detection | +| `07_python_network` | `PyNetwork` with a pure-Python node *(pending)* | +| `08_python_subport` | `net.read` / `net.write`, sub-port tap *(pending)* | +| `09_opencv_cellshade` | Real-time cell-shading on webcam; named ports, fan-out, `MainThreadNode` display (requires OpenCV) | +| `10_static_hello_pipeline` | `make_network()` version of 01 β€” compile-time topology | +| `11_static_fanout` | Auto-inserted `FanoutNode` from a duplicated source port | +| `12_static_cellshade` | Static cell-shading with auto fan-out and `Label` NTTPs | +| `13_debug_cellshade` | One-op-per-node pipeline + variadic `DebugCanvas` tiling node | +| `14_debug_hub` | Two networks sharing a `SharedResource` via `DebugHub` | +| `15_node_error_handler` | Per-node `set_error_handler` (skip-and-continue vs stop) | +| `16_event_callbacks` | `set_overflow_callback` + network `set_event_handler` | -The hot path (per-item `pop` β†’ `push` in each node thread) is identical in both cases. +--- -### Compatibility +## Future Extension Points (Heterogeneous Execution) -- `StaticNetwork` implements `INode`, so it can be registered inside a runtime `Network` - via `net.add("name", static_net)` β€” enabling mixed static/dynamic graphs. -- All existing node types (`Node`, `ObjectNode`, `FanoutNode`, `MainThreadNode`) work - unchanged as vertices in a `StaticNetwork`. -- The Python `PyNetwork` is unaffected β€” it remains runtime-only. +Not implemented, but the design keeps these doors open: -### File layout addition +- **`IChannel` abstract interface** β€” `Channel` and a future `RemoteChannel` (socket / + shared-memory) sharing one `push`/`pop` surface so nodes are agnostic to channel location. +- **`Serializer` trait** β€” parallel to `channel_storage_policy` / `PythonConverter`, for + cross-device serialisation (MessagePack for embedded, pinned memory for GPU zero-copy). +- **`NodeKind` tag** β€” e.g. `{ Local, Gpu, Remote }` on `INode`, letting the watchdog apply + per-device health-check and timeout strategies. -``` -include/kpn/ - static_network.hpp # Edge<>, make_network(), StaticNetwork<> - tmp/ - fanout_groups.hpp # fanout_groups metafunction - topo_sort.hpp # compile-time DFS + cycle check - repeat_tuple.hpp # repeat_tuple_t (moved from fanout.hpp) -``` - -`fanout.hpp` keeps `FanoutNode` and `make_fanout` for users who want to wire -fanouts explicitly in a runtime `Network`. `static_network.hpp` uses `FanoutNode` internally -but the user never calls `make_fanout` when using `make_network`. +The `IScheduler` abstraction already decouples node execution from any specific thread model, +making a cooperative or device-specific executor a drop-in. --- ## 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 | -| Mixed-rate inputs | `latch<>` tag for ports that reuse last-seen value; blocks only on first fire; node fires at rate of `in<>` ports | -| Fan-out | Explicit `FanoutNode` for runtime `Network`; auto-inserted by `make_network()` for `StaticNetwork` | -| Static vs runtime graph | Both coexist; `StaticNetwork` for C++ graphs known at compile time, `Network` for Python/dynamic graphs; `StaticNetwork` implements `INode` so it embeds in `Network` | -| Node identity in static graphs | `Label` NTTP (human name for diagnostics) + `UniqueTag` NTTP (collision-breaker for same-Func nodes); both default to `""` / `0` so existing code is unaffected | +| Execution model | Reactive: nodes submit `fire_once()` to an `IScheduler` when inputs are ready, not one blocking thread per node | +| `Node<>` vs `PoolNode<>` | `Node<>` owns a private `ThreadPool(1)`; `PoolNode<>` shares a pool for bounded threads | +| Channel | Lock-free SPSC ring buffer, `atomic::wait/notify` + spin-before-sleep | +| Shutdown | Per-channel `accepting_` flag; `disable()` unblocks `pop()` (β†’ `ChannelClosedError`) | +| Overflow | `ChannelOverflowError` on full accepting channel; silent drop on disabled channel | +| Node error policy | Per-node `NodeErrorHandler` returning bool (skip vs stop) | +| Network ownership | Non-owning; user declares nodes, network stores `INode*` | +| Fan-out | Explicit `FanoutNode` for runtime `Network`; auto-inserted by `make_network()` | +| Branching | `RouterNode` (select one of N) and `FilterNode` (predicate gate) | +| Static vs runtime graph | Both; `StaticNetwork` for compile-time C++ topology, `Network` for dynamic/Python; `StaticNetwork` is an `INode` so it embeds in `Network` | +| Node identity (static graphs) | `Label` NTTP (name) + `UniqueTag` NTTP (collision-breaker); both default | +| Shared device resource | `SharedResource` with priority + aging arbitration | +| Main-thread / GUI work | `MainThreadNode<>` driven by `step()` on the main thread | +| External-event sources | `InterruptNode` with a thread-safe `get_trigger()` | +| Web debugging | Per-network server + multi-network `DebugHub`, behind `KPN_WEB_DEBUG` | +| Mixed-rate latched inputs | **Not implemented** β€” no `latch<>` ports | -- 2.39.5 From a0c4bf580eb88ae460051bd4712bf427c119156c Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Tue, 14 Jul 2026 19:35:31 +0200 Subject: [PATCH 3/4] fix: deliver EOF sentinel only when the ring is freshly empty MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Channel::pop() surfaced the out-of-band sentinel from its empty branch using the tail_ snapshot taken at the top of the loop. Under contention the producer can push more values *and* the sentinel in the window between that snapshot and take_sentinel(), so pop() could return the sentinel while real values still sat in the ring β€” the sentinel jumping ahead of values pushed before it. No value was lost (a consumer that keeps draining still receives them, and approx_size() keeps counting them so a PoolNode reschedules), but a consumer treating the sentinel as a hard "last message" barrier would act on EOF early. Re-confirm emptiness against a fresh tail_ load before taking the sentinel. Costs one acquire-load on the empty-ring path only; never runs in steady state. The spin and post-spin takes already reload tail_ on the line above them; try_pop_now() already reads tail_ fresh in the same branch β€” both were correct and are unchanged. The two sentinel stress cases now assert the strict "sentinel is last, after every value" ordering (previously relaxed to avoid the flake this fixes). Verified TSan-clean (2606 assertions, no data races) over repeated runs. Co-Authored-By: Claude Opus 4.8 --- include/kpn/channel.hpp | 11 ++++- tests/test_channel_stress.cpp | 89 ++++++++++++++++------------------- 2 files changed, 51 insertions(+), 49 deletions(-) diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index 4466fd8..99ebe51 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -180,7 +180,16 @@ public: if (h == t) { // Ring drained β€” deliver any pending out-of-band sentinel (EOF) // now, so it always arrives after the data pushed before it. - { T s; if (take_sentinel(s)) return s; } + // + // Re-confirm emptiness against a fresh tail_ first: the snapshot + // at the top of the loop may be stale (the producer can push more + // values *and* the sentinel in the window since), and the sentinel + // must never jump ahead of ring values pushed before it. The spin + // and post-spin takes below already reload tail_ on the line above + // them; this is the one take that used the loop-top snapshot. + if (h == tail_.load(std::memory_order_acquire)) { + T s; if (take_sentinel(s)) return s; + } if (!accepting_.load(std::memory_order_acquire)) throw ChannelClosedError{}; diff --git a/tests/test_channel_stress.cpp b/tests/test_channel_stress.cpp index bd113d0..eb58704 100644 --- a/tests/test_channel_stress.cpp +++ b/tests/test_channel_stress.cpp @@ -180,28 +180,24 @@ TEST_CASE("SPSC: push_callback fires on each empty->non-empty transition", REQUIRE(callbacks.load() <= N); } -// What the out-of-band sentinel guarantees under contention β€” and what it does -// not. push_sentinel() publishes has_eof_ (release) after the producer's N ring +// Ordering contract of the out-of-band sentinel under contention. +// +// push_sentinel() publishes has_eof_ (release) after the producer's N ring // pushes; a consumer that observes has_eof_ (acquire) therefore also observes -// every value pushed before it. What these tests assert: +// every value pushed before it. Both pop() and try_pop_now() only surface the +// sentinel once the ring is *freshly* observed empty, so the sentinel is the +// strictly last item received β€” it never jumps ahead of a ring value pushed +// before it. These tests treat the sentinel as a hard "last message" barrier +// (the consumer stops draining the moment it sees it) and assert that all N +// values arrived, in a contiguous 0..N-1 sequence, before it. // -// * Losslessness β€” every value 0..N-1 is delivered exactly once (contiguous, -// no gaps, no duplicates) and the sentinel is delivered exactly once. This -// is the invariant that must hold on every run; a broken acquire/release -// pairing would surface as a lost/duplicated value or (under TSan) a data -// race on has_eof_/eof_value_. -// -// What they deliberately do NOT assert is that the sentinel is the *strictly -// last* item popped. pop() checks emptiness (h == t) using a tail_ snapshot -// taken at the top of its loop; the producer can push more values *and* the -// sentinel in the window before take_sentinel() runs, so the consumer may -// surface the sentinel with a few real values still queued behind it. Those -// values are not lost β€” a consumer that keeps draining still receives them β€” -// but "sentinel arrives dead last" is not a property the channel promises, so -// asserting it would be flaky. We track how many values trailed the sentinel -// for visibility without failing on it. +// Regression guard: an earlier version of pop() checked emptiness against a +// stale tail_ snapshot from the top of its loop, so under load the sentinel +// could surface with a few real values still queued β€” breaking in_order / +// values==N here. Under TSan these also cover the has_eof_/eof_value_ +// acquire/release handshake and the spin/futex wakeup on push_sentinel(). -TEST_CASE("SPSC: sentinel and all values survive contention (blocking pop)", +TEST_CASE("SPSC: sentinel is strictly last, after every value (blocking pop)", "[channel][stress]") { constexpr int N = 20'000; constexpr int SENTINEL = -1; @@ -221,34 +217,32 @@ TEST_CASE("SPSC: sentinel and all values survive contention (blocking pop)", ch.push_sentinel(SENTINEL); // must-deliver, never overflows/blocks }); - std::vector seen(N, false); - int values = 0; - int sentinels = 0; - bool duplicate = false; - // Drain until the sentinel AND all N values have been received; the - // sentinel may arrive before the last few values (see note above). - while (values < N || sentinels == 0) { + int expected = 0; + bool in_order = true; + bool saw_sentinel = false; + // Treat the sentinel as EOF: stop draining the instant it appears. + for (;;) { int v = ch.pop(); - if (v == SENTINEL) { ++sentinels; continue; } - if (seen[v]) duplicate = true; else seen[v] = true; - ++values; + if (v == SENTINEL) { saw_sentinel = true; break; } + if (v != expected) in_order = false; + ++expected; } producer.join(); - REQUIRE_FALSE(duplicate); - REQUIRE(values == N); // every value delivered exactly once - REQUIRE(sentinels == 1); // sentinel delivered exactly once + REQUIRE(saw_sentinel); + REQUIRE(in_order); + REQUIRE(expected == N); // all N values received before the sentinel REQUIRE(ch.size() == 0); REQUIRE(ch.approx_size() == 0); } } -TEST_CASE("SPSC: sentinel and all values survive contention (try_pop_now)", +TEST_CASE("SPSC: sentinel is strictly last, after every value (try_pop_now)", "[channel][stress]") { // The pool-node consume path is try_pop_now(), not pop(): it must surface - // the out-of-band sentinel once the ring is observed empty. The consumer - // spins with no sleeps, racing the producer at full tilt across the - // empty-ring boundary where take_sentinel() is reached. + // the out-of-band sentinel only once the ring is freshly observed empty. + // The consumer spins with no sleeps, racing the producer at full tilt + // across the empty-ring boundary where take_sentinel() is reached. constexpr int N = 20'000; constexpr int SENTINEL = -1; @@ -265,23 +259,22 @@ TEST_CASE("SPSC: sentinel and all values survive contention (try_pop_now)", ch.push_sentinel(SENTINEL); }); - std::vector seen(N, false); - int values = 0; - int sentinels = 0; - bool duplicate = false; + int expected = 0; + bool in_order = true; + bool saw_sentinel = false; int v; - while (values < N || sentinels == 0) { + for (;;) { if (!ch.try_pop_now(v)) { std::this_thread::yield(); continue; } - if (v == SENTINEL) { ++sentinels; continue; } - if (seen[v]) duplicate = true; else seen[v] = true; - ++values; + if (v == SENTINEL) { saw_sentinel = true; break; } + if (v != expected) in_order = false; + ++expected; } producer.join(); - REQUIRE_FALSE(duplicate); - REQUIRE(values == N); - REQUIRE(sentinels == 1); - // Sentinel held no ring slot; once drained the channel is fully empty. + REQUIRE(saw_sentinel); + REQUIRE(in_order); + REQUIRE(expected == N); + // Sentinel held no ring slot; once taken the channel is fully empty. REQUIRE(ch.size() == 0); REQUIRE(ch.approx_size() == 0); } -- 2.39.5 From ec19137ed96fc2d3c922575da4ac1bc26b5f90b8 Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Fri, 17 Jul 2026 20:01:57 +0200 Subject: [PATCH 4/4] ci: fix TSan aborting at init on the nested-LXC runner The ThreadSanitizer job runs on Docker nested in an unprivileged LXC container, whose kernel randomizes mmap addresses beyond the range TSan's fixed shadow mapping expects. TSan aborted at init with "unexpected memory mapping" before any test ran. Disable ASLR per-process with `setarch -R`, which needs the personality(2) syscall that Docker's default seccomp profile blocks; seccomp=unconfined on the container permits it. Verified on the runner that both are required: setarch -R alone gets EPERM, seccomp alone still aborts, both together run clean. Scoped to the tsan job, which runs only our own test binaries. Co-Authored-By: Claude Opus 4.8 --- .gitea/workflows/tsan.yaml | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/.gitea/workflows/tsan.yaml b/.gitea/workflows/tsan.yaml index 53f81ad..7911d7d 100644 --- a/.gitea/workflows/tsan.yaml +++ b/.gitea/workflows/tsan.yaml @@ -18,6 +18,15 @@ jobs: runs-on: linux/amd64 container: image: gitea.tourolle.paris/dtourolle/kpnpp-builder:latest + # This runner is Docker nested in an unprivileged LXC container, whose + # kernel randomizes mmap addresses beyond the range TSan's fixed shadow + # mapping expects, so TSan aborts at init with "unexpected memory + # mapping". The fix is to disable ASLR per-process with `setarch -R` + # (below), which needs the personality(2) syscall that Docker's default + # seccomp profile blocks. seccomp=unconfined permits it. Verified on the + # runner: setarch -R alone gets EPERM, seccomp alone still aborts, both + # together run clean. Scoped to this job, which runs only our own tests. + options: --security-opt seccomp=unconfined steps: - name: Checkout repository uses: actions/checkout@v4 @@ -54,13 +63,14 @@ jobs: # the full picture for lock-order issues. env: TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1" - run: ./build/tests/kpn_tests_stress + # setarch -R disables ASLR for this process; see the container comment. + run: setarch -R ./build/tests/kpn_tests_stress - name: Run unit tests under TSan working-directory: tsan-${{ github.run_id }} env: TSAN_OPTIONS: "halt_on_error=1 second_deadlock_stack=1" - run: ./build/tests/kpn_tests + run: setarch -R ./build/tests/kpn_tests - name: Cleanup if: always() -- 2.39.5