From 6f384dc4b5eab0413b49a82297f191d6375542ee Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Fri, 19 Jun 2026 22:26:39 +0200 Subject: [PATCH] Added callbacks for node errors and fifo overflow Add new doc system which should/might deploy to pages. --- .gitea/workflows/docs.yaml | 41 +++++ .gitea/workflows/test.yaml | 16 +- .gitignore | 1 + docs/channels.md | 55 ++++++ docs/error-handling.md | 84 +++++++++ docs/examples.md | 40 +++++ docs/fanout.md | 44 +++++ docs/getting-started.md | 76 ++++++++ docs/index.md | 39 +++++ docs/network.md | 67 +++++++ docs/nodes.md | 81 +++++++++ docs/requirements.txt | 3 + docs/shared-resource.md | 42 +++++ docs/static-network.md | 46 +++++ examples/01_hello_pipeline/main.cpp | 12 +- examples/02_named_ports/main.cpp | 8 +- examples/03_multi_output/main.cpp | 8 +- examples/04_storage_policy/main.cpp | 4 +- examples/05_error_handling/main.cpp | 4 +- examples/09_opencv_cellshade/main.cpp | 16 +- examples/15_node_error_handler/main.cpp | 2 + examples/16_event_callbacks/main.cpp | 83 +++++++++ examples/CMakeLists.txt | 12 +- include/kpn/inode.hpp | 13 ++ include/kpn/interrupt_node.hpp | 42 +++-- include/kpn/network.hpp | 20 ++- include/kpn/pool_node.hpp | 125 ++++++++----- include/kpn/static_network.hpp | 17 ++ mkdocs.yml | 50 ++++++ tests/test_pool_node.cpp | 223 ++++++++++++++++++++++++ 30 files changed, 1192 insertions(+), 82 deletions(-) create mode 100644 .gitea/workflows/docs.yaml create mode 100644 docs/channels.md create mode 100644 docs/error-handling.md create mode 100644 docs/examples.md create mode 100644 docs/fanout.md create mode 100644 docs/getting-started.md create mode 100644 docs/index.md create mode 100644 docs/network.md create mode 100644 docs/nodes.md create mode 100644 docs/requirements.txt create mode 100644 docs/shared-resource.md create mode 100644 docs/static-network.md create mode 100644 examples/16_event_callbacks/main.cpp create mode 100644 mkdocs.yml diff --git a/.gitea/workflows/docs.yaml b/.gitea/workflows/docs.yaml new file mode 100644 index 0000000..f7ffedb --- /dev/null +++ b/.gitea/workflows/docs.yaml @@ -0,0 +1,41 @@ +name: '๐Ÿ“š Docs' + +on: + push: + branches: + - master + paths: + - 'docs/**' + - 'mkdocs.yml' + - 'examples/**/*.cpp' # snippet sources + workflow_dispatch: + +jobs: + deploy: + runs-on: linux/amd64 + container: + image: python:3.12-slim + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + with: + fetch-depth: 0 # full history needed for mkdocs gh-deploy + + - name: Install MkDocs dependencies + run: pip install --quiet -r docs/requirements.txt + + - name: Configure git identity + run: | + git config user.name "Gitea Actions" + git config user.email "actions@gitea.tourolle.paris" + + - name: Build and deploy to gitea-pages branch + env: + GH_TOKEN: ${{ github.token }} + run: | + mkdocs gh-deploy \ + --force \ + --remote-branch gitea-pages \ + --remote-name origin \ + --message "docs: deploy from ${{ github.sha }}" diff --git a/.gitea/workflows/test.yaml b/.gitea/workflows/test.yaml index a371806..1deea37 100644 --- a/.gitea/workflows/test.yaml +++ b/.gitea/workflows/test.yaml @@ -41,7 +41,7 @@ jobs: -G Ninja \ -DCMAKE_BUILD_TYPE=Debug \ -DKPN_BUILD_TESTS=ON \ - -DKPN_BUILD_EXAMPLES=OFF \ + -DKPN_BUILD_EXAMPLES=ON \ -DKPN_BUILD_PYTHON=ON \ -DFETCHCONTENT_BASE_DIR=$HOME/.cmake/fetchcontent @@ -49,18 +49,26 @@ jobs: working-directory: test-${{ github.run_id }} run: cmake --build build --parallel - - name: Run tests + - name: Run unit tests working-directory: test-${{ github.run_id }} run: | cd build - ctest --output-on-failure --output-junit test-results.xml + ctest --output-on-failure --output-junit test-results.xml --label-exclude examples + + - name: Run example smoke tests + working-directory: test-${{ github.run_id }} + run: | + cd build + ctest --output-on-failure --output-junit example-results.xml -L examples - name: Upload test results if: always() uses: actions/upload-artifact@v3 with: name: test-results - path: test-${{ github.run_id }}/build/test-results.xml + path: | + test-${{ github.run_id }}/build/test-results.xml + test-${{ github.run_id }}/build/example-results.xml retention-days: 7 - name: Cleanup diff --git a/.gitignore b/.gitignore index 174ad4c..910b5d0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ # Build output build/ build_debug/ +site/ # Python __pycache__/ *.py[cod] diff --git a/docs/channels.md b/docs/channels.md new file mode 100644 index 0000000..f606883 --- /dev/null +++ b/docs/channels.md @@ -0,0 +1,55 @@ +# Channels + +A `Channel` is a lock-free SPSC (single-producer, single-consumer) ring buffer with atomic wait/notify. + +## Semantics + +- **Bounded**: fixed capacity set at construction. Default is 5 items. +- **Backpressure**: when full, `push()` throws `ChannelOverflowError` immediately โ€” no blocking, no spin. +- **Blocking consumer**: `pop()` blocks until an item is available or the channel is disabled. +- **Disable**: `channel.disable()` stops accepting pushes and unblocks any waiting `pop()` with `ChannelClosedError`. + +## Storage policy + +Small trivially-copyable types (โ‰ค 8 bytes) are stored by value. Larger types are heap-allocated and passed via `shared_ptr` โ€” one allocation per push, zero-copy fan-out: + +```cpp +--8<-- "examples/04_storage_policy/main.cpp:storage_policy_spec" +``` + +Specialize `kpn::ChannelDataSize` for accurate bandwidth reporting on heap-owning types: + +```cpp +template<> +struct kpn::ChannelDataSize { + static std::size_t bytes(const cv::Mat& m) { return m.total() * m.elemSize(); } +}; +``` + +## Named ports + +`in<"name">` and `out<"name">` tag nodes for readable wiring: + +```cpp +--8<-- "examples/02_named_ports/main.cpp:named_port_creation" +``` + +Named ports are checked at compile time โ€” a typo in a port name is a compile error. + +## Capacity tuning + +Set capacity per node at construction: + +```cpp +auto node = make_node(/*capacity=*/20); +``` + +Capacity is rounded up internally to the next power of two. Monitor fill levels via diagnostics to tune for your workload โ€” a too-small capacity causes overflows; a too-large one wastes memory and hides producer/consumer speed mismatches. + +## Spin count + +`Channel` spins for up to ~4 ยตs (200 `pause` hints at ~20 ns each on x86) before sleeping on a futex. Set to 0 for power-constrained or predominantly-idle pipelines: + +```cpp +Channel ch(/*capacity=*/5, /*spin_count=*/0); +``` diff --git a/docs/error-handling.md b/docs/error-handling.md new file mode 100644 index 0000000..784495f --- /dev/null +++ b/docs/error-handling.md @@ -0,0 +1,84 @@ +# Error Handling & Events + +KPN++ provides three complementary layers for observing and reacting to failures. + +--- + +## 1. Per-node error handler + +Called when a node's function throws an unhandled exception. Return `true` to skip the failed invocation and keep running; `false` to stop the node. + +```cpp +--8<-- "examples/15_node_error_handler/main.cpp:error_handler" +``` + +When a node stops (either from `false` return or no handler installed), it: + +1. Disables its **input** channels โ€” upstream stops pushing into dead queues. +2. Disables its **output** channels โ€” downstream nodes receive `ChannelClosedError` on their next pop, propagating the shutdown naturally through the graph. + +--- + +## 2. Per-node overflow callback + +Fired with a timestamp each time an output push is dropped because the channel is full. The node name is known at registration so it is not included โ€” keeping the callback zero-overhead when unused. + +```cpp +--8<-- "examples/16_event_callbacks/main.cpp:per_node_callback" +``` + +!!! note + The callback is purely informational โ€” the node always continues after an overflow. To stop the node on overflow, call `node.stop()` from inside the callback. + +A matching `set_closed_callback()` fires (also with just a timestamp) when the node stops due to a closed upstream channel: + +```cpp +node.set_closed_callback([](std::chrono::steady_clock::time_point ts) { + std::cerr << "node stopped at t=" << ts.time_since_epoch().count() << '\n'; +}); +``` + +Each node holds two callback slots per event type โ€” one user-set (registered above) and one injected by the network (see below). Both fire independently. + +--- + +## 3. Network-level event handler + +One callback for the whole network. Receives the node name (captured in a closure by the network at `build()` / `start()`), a `NodeEvent`, and a timestamp: + +```cpp +--8<-- "examples/16_event_callbacks/main.cpp:network_event_handler" +``` + +`NodeEvent` values: + +| Value | Meaning | +|---|---| +| `NodeEvent::Overflow` | An output push was dropped (channel full) | +| `NodeEvent::Closed` | The node stopped (crash or upstream close cascade) | + +The network handler and any per-node callbacks are **independent** โ€” both fire when set. + +--- + +## Complete example + +`examples/16_event_callbacks/main.cpp` shows a fast producer overflowing a slow consumer, with both a per-node overflow callback and a network-level event handler active simultaneously. + +Node functions: + +```cpp +--8<-- "examples/16_event_callbacks/main.cpp:node_fns" +``` + +Per-node overflow callback: + +```cpp +--8<-- "examples/16_event_callbacks/main.cpp:per_node_callback" +``` + +Network-level event handler: + +```cpp +--8<-- "examples/16_event_callbacks/main.cpp:network_event_handler" +``` diff --git a/docs/examples.md b/docs/examples.md new file mode 100644 index 0000000..d2ed9ec --- /dev/null +++ b/docs/examples.md @@ -0,0 +1,40 @@ +# Examples + +All C++ examples are built by default and registered as CTest smoke tests. Run them all with: + +```bash +ctest --test-dir build -L examples +``` + +## Index + +| Example | What it shows | +|---|---| +| `01_hello_pipeline` | Linear pipeline, index-based port wiring | +| `02_named_ports` | `in<>`/`out<>` name tags, named port access | +| `03_multi_output` | Tuple-returning node, per-element routing | +| `04_storage_policy` | `channel_storage_policy` specialisation | +| `05_error_handling` | Diagnostics handler, overflow channel stats | +| `06_watchdog` | Watchdog interval, stall detection | +| `10_static_hello_pipeline` | `StaticNetwork` + `make_network()` | +| `11_static_fanout` | `StaticNetwork` with `FanoutNode` | +| `15_node_error_handler` | `set_error_handler()` โ€” skip or stop on exception | +| `16_event_callbacks` | `set_overflow_callback()`, `set_event_handler()` | + +## OpenCV examples (optional) + +Built only when OpenCV โ‰ฅ 4 is found: + +| Example | What it shows | +|---|---| +| `09_opencv_cellshade` | Real-time cell-shading on webcam; `MainThreadNode` for display | +| `12_static_cellshade` | Same pipeline as a `StaticNetwork` | +| `13_debug_cellshade` | Web debug UI overlay on the cell-shading pipeline | + +Run the cell-shading example: + +```bash +./build/examples/09_opencv_cellshade +# Press 'q' or close the window to stop. +# Falls back to an animated synthetic pattern if no webcam is found. +``` diff --git a/docs/fanout.md b/docs/fanout.md new file mode 100644 index 0000000..54a3185 --- /dev/null +++ b/docs/fanout.md @@ -0,0 +1,44 @@ +# Fan-out & Routing + +## FanoutNode + +Reads one item and pushes a copy to each of N output channels. All downstream nodes receive every item. + +```cpp +auto fan = make_fanout(/*capacity=*/8); + +net.connect("src", src.output<0>(), "fan", fan.input<0>()) + .connect("fan", fan.output<0>(), "nodeA", nodeA.input<0>()) + .connect("fan", fan.output<1>(), "nodeB", nodeB.input<0>()); +``` + +If one downstream channel overflows, that output drops the item independently โ€” the other outputs are unaffected. + +See `examples/11_static_fanout`. + +## RouterNode + +Reads one item and pushes it to exactly one of N outputs, chosen by a selector function: + +```cpp +auto router = make_router( + [](const Frame& f) -> std::size_t { return f.stream_id % 3; }); + +net.connect("src", src.output<0>(), "router", router.input<0>()) + .connect("router", router.output<0>(), "nodeA", nodeA.input<0>()) + .connect("router", router.output<1>(), "nodeB", nodeB.input<0>()) + .connect("router", router.output<2>(), "nodeC", nodeC.input<0>()); +``` + +If the selector returns `>= N` the item is silently dropped. + +## FilterNode + +Reads one item and passes it downstream only when a predicate returns `true`: + +```cpp +auto filt = make_filter([](const Frame& f) { return f.valid; }); + +net.connect("src", src.output<0>(), "filt", filt.input<0>()) + .connect("filt", filt.output<0>(), "dst", dst.input<0>()); +``` diff --git a/docs/getting-started.md b/docs/getting-started.md new file mode 100644 index 0000000..cbaf94c --- /dev/null +++ b/docs/getting-started.md @@ -0,0 +1,76 @@ +# Getting Started + +## Requirements + +| Dependency | Version | Notes | +|---|---|---| +| CMake | โ‰ฅ 3.21 | | +| C++ compiler | GCC โ‰ฅ 11, Clang โ‰ฅ 13 | C++20 required | +| nanobind | โ‰ฅ 2.1 | auto-fetched; Python โ‰ฅ 3.8 | +| Catch2 | v3 | auto-fetched for tests | +| OpenCV | โ‰ฅ 4 | optional; only for examples 09/12/13 | + +## Build + +```bash +cmake -B build # core + tests + C++ examples +cmake --build build --parallel +ctest --test-dir build # run all tests including example smoke tests +``` + +Enable Python bindings: + +```bash +cmake -B build -DKPN_BUILD_PYTHON=ON +cmake --build build --parallel +``` + +Skip examples: + +```bash +cmake -B build -DKPN_BUILD_EXAMPLES=OFF +``` + +## Your first pipeline + +Three functions โ€” source, transform, sink โ€” wired into a `Network`: + +```cpp +--8<-- "examples/01_hello_pipeline/main.cpp:basic_node_fns" +``` + +Create nodes, connect them, build and run: + +```cpp +--8<-- "examples/01_hello_pipeline/main.cpp:network_build" +``` + +That's it. Types are inferred from function signatures. The channel between `src` and `dbl` carries `int`; the channel between `dbl` and `prn` also carries `int`. A type mismatch is a compile error. + +## Named ports + +For nodes with multiple inputs or outputs, name the ports for clarity: + +```cpp +--8<-- "examples/02_named_ports/main.cpp:named_port_creation" +``` + +Wire by name instead of index: + +```cpp +--8<-- "examples/02_named_ports/main.cpp:named_port_network" +``` + +## Multi-output nodes + +Return a `std::tuple` to fan out to multiple downstream nodes: + +```cpp +--8<-- "examples/03_multi_output/main.cpp:multi_output_fn" +``` + +Wire each tuple element to its own downstream node: + +```cpp +--8<-- "examples/03_multi_output/main.cpp:fanout_network" +``` diff --git a/docs/index.md b/docs/index.md new file mode 100644 index 0000000..7f9004c --- /dev/null +++ b/docs/index.md @@ -0,0 +1,39 @@ +# KPN++ + +A C++20 [Kahn Process Network](https://en.wikipedia.org/wiki/Kahn_process_networks) library. Each node wraps a plain function and runs concurrently, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind. + +--- + +## Why KPN++? + +- **Zero boilerplate** โ€” wrap any callable as a node; types flow automatically from the function signature +- **Bounded channels** โ€” backpressure is structural, not bolted on +- **Observable** โ€” per-node and network-level callbacks for overflow and stop events; diagnostics snapshots; optional web UI +- **Composable** โ€” `Network` for runtime wiring, `StaticNetwork` for compile-time topology with zero overhead + +--- + +## Quick example + +```cpp +#include +using namespace kpn; + +--8<-- "examples/01_hello_pipeline/main.cpp:basic_node_fns" + +int main() { +--8<-- "examples/01_hello_pipeline/main.cpp:network_build" +} +``` + +--- + +## Install & build + +```bash +cmake -B build +cmake --build build --parallel +ctest --test-dir build # unit tests + example smoke tests +``` + +See [Getting Started](getting-started.md) for full build options. diff --git a/docs/network.md b/docs/network.md new file mode 100644 index 0000000..a9c81d3 --- /dev/null +++ b/docs/network.md @@ -0,0 +1,67 @@ +# Networks + +A `Network` wires nodes together at runtime using a builder chain. + +## Building a network + +```cpp +--8<-- "examples/01_hello_pipeline/main.cpp:network_build" +``` + +The builder chain: + +| Method | Purpose | +|---|---| +| `.add(name, node)` | Register a node; assigns its name | +| `.connect(src, port, dst, port)` | Wire one output port to one input port | +| `.build()` | Compute topological order; inject network callbacks | +| `.start()` | Start nodes in topological order | +| `.stop()` | Stop all nodes immediately | +| `.shutdown()` | Graceful drain: stop sources first, wait for channels to empty, then stop downstream | + +## Port access + +Ports are accessed by index or by name: + +```cpp +// By index +net.connect("src", src.output<0>(), "dst", dst.input<0>()); + +// By name (requires named ports) +--8<-- "examples/02_named_ports/main.cpp:named_port_network" +``` + +## Diagnostics + +Install a diagnostics handler to receive periodic snapshots of every node and channel: + +```cpp +--8<-- "examples/05_error_handling/main.cpp:diagnostics_handler" +``` + +Or print a full report at any time: + +```cpp +net.print_diagnostics(); // writes to stderr by default +net.print_diagnostics(std::cout); +``` + +## Network-level event handler + +Observe overflow and node-stop events across the entire network in one place: + +```cpp +--8<-- "examples/16_event_callbacks/main.cpp:network_event_handler" +``` + +`NodeEvent` is either `NodeEvent::Overflow` (item dropped on full channel) or `NodeEvent::Closed` (node stopped due to crash or closed upstream channel). See [Error Handling & Events](error-handling.md). + +## Shutdown + +`net.stop()` halts immediately โ€” all nodes stop in reverse topological order. + +`net.shutdown()` drains gracefully: source nodes stop first; their output channels are polled until empty; then the next layer stops, and so on. This ensures no items are lost if downstream nodes are still consuming. + +## StaticNetwork + +For zero-overhead compile-time topology, see [Static Networks](static-network.md). diff --git a/docs/nodes.md b/docs/nodes.md new file mode 100644 index 0000000..06149cd --- /dev/null +++ b/docs/nodes.md @@ -0,0 +1,81 @@ +# Nodes + +A node wraps any callable. Its input types are inferred from the function's parameter list; its output types from the return type. + +## Node types + +| Type | Thread model | Use case | +|---|---|---| +| `Node` | Dedicated thread per node | Default โ€” simplest, most isolated | +| `PoolNode` | Shared `ThreadPool` | Many nodes, resource-bounded execution | +| `InterruptNode` | Event-driven, no thread | Camera frame ready, timer tick, socket | +| `FanoutNode` | Dedicated thread | Broadcast one item to N outputs | +| `RouterNode` | Dedicated thread | Route one item to one of N outputs | +| `FilterNode` | Dedicated thread | Pass items matching a predicate | + +## Creating nodes + +All node types are created via factory functions that infer types from the callable: + +```cpp +// Free function โ€” simplest case +auto node = make_node(); + +// Stateful functor (operator() is the function) +MyProcessor proc; +auto node = make_node(proc); + +// Pool node โ€” shares a ThreadPool with other nodes +auto pool = std::make_shared(4); +auto node = make_pool_node(pool); + +// Interrupt node โ€” triggered externally +auto sched = std::make_shared(2); +auto node = make_interrupt_node(sched, out<"frame">{}); +camera_sdk.on_frame_ready(node.get_trigger()); +``` + +## Channel capacity + +Each node's input FIFO has a configurable capacity (default 5): + +```cpp +auto node = make_node(/*capacity=*/20); +auto node = make_pool_node(pool, /*capacity=*/20); +``` + +When an upstream push would exceed capacity, `ChannelOverflowError` is thrown and the item is dropped. See [Error Handling & Events](error-handling.md) to observe and react to this. + +## Source nodes + +A node with no inputs is a source. It self-submits immediately on `start()` and re-submits after each execution: + +```cpp +static int produce() { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + return ++counter; +} +auto src = make_node(); +``` + +!!! tip + Source nodes must sleep or yield to avoid overflowing their output channel. The channel capacity provides the only bound. + +## Sink nodes + +A node with a `void` return is a sink โ€” it consumes items without producing output: + +```cpp +static void print_it(int x) { std::cout << x << '\n'; } +auto snk = make_node(); +``` + +## Error handler + +When a node's function throws an unhandled exception, the default behaviour is to stop the node (disabling its channels so the shutdown cascades downstream). Install a handler to override: + +```cpp +--8<-- "examples/15_node_error_handler/main.cpp:error_handler" +``` + +See [Error Handling & Events](error-handling.md) for the full picture. diff --git a/docs/requirements.txt b/docs/requirements.txt new file mode 100644 index 0000000..4834720 --- /dev/null +++ b/docs/requirements.txt @@ -0,0 +1,3 @@ +mkdocs>=1.5 +mkdocs-material>=9.5 +pymdown-extensions>=10.0 diff --git a/docs/shared-resource.md b/docs/shared-resource.md new file mode 100644 index 0000000..64a14d4 --- /dev/null +++ b/docs/shared-resource.md @@ -0,0 +1,42 @@ +# Shared Resources + +`SharedResource` arbitrates exclusive access to a resource (ONNX session, CUDA stream, serial port) across multiple nodes using a priority-based waiter queue with starvation prevention. + +## Usage + +```cpp +#include +using namespace kpn; + +SharedResource model(session_args...); + +static cv::Mat run_inference(cv::Mat frame) { + // Acquires the model; releases automatically on scope exit. + auto guard = model.acquire_balanced(in_channel, out_channel); + return guard->Run(frame); +} +``` + +## Acquire modes + +| Method | Priority | +|---|---| +| `acquire()` | Equal (fair FIFO) | +| `acquire(fn)` | Custom โ€” `fn()` returns `float` in `[0, 1]` | +| `acquire_balanced(in_ch, out_ch)` | `input_fill ร— output_headroom` โ€” highest urgency wins | + +`acquire_balanced` favours nodes with full input queues and empty output queues โ€” the node that has the most work to do and nowhere to stall wins the resource next. + +## Starvation prevention + +Each waiter's effective score grows with elapsed wait time (`0.05` per second by default), ensuring a low-priority node eventually gets served regardless of how frequently higher-priority nodes compete. + +## Diagnostics + +Register with the network for snapshot reporting: + +```cpp +net.register_resource("model", &model); +``` + +The diagnostics table then shows acquisition count, mean wait time, and current waiter count. diff --git a/docs/static-network.md b/docs/static-network.md new file mode 100644 index 0000000..34cdf2f --- /dev/null +++ b/docs/static-network.md @@ -0,0 +1,46 @@ +# Static Networks + +`StaticNetwork` encodes the entire topology at compile time using a `make_network()` builder. Nodes and channel types are verified statically with zero runtime overhead. + +## Usage + +```cpp +#include +using namespace kpn; + +static int produce() { return 42; } +static int double_it(int x) { return x * 2; } +static void print_it(int x) { std::cout << x << '\n'; } + +int main() { + auto src = make_node (); + auto dbl = make_node(); + auto prn = make_node (); + + auto net = make_network( + edge(src, src.output<0>(), dbl, dbl.input<0>()), + edge(dbl, dbl.output<0>(), prn, prn.input<0>()) + ); + + net.set_event_handler([](std::string_view name, NodeEvent ev, auto ts) { + // same API as Network + }); + + net.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + net.stop(); +} +``` + +See `examples/10_static_hello_pipeline` and `examples/11_static_fanout`. + +## When to use + +| | `Network` | `StaticNetwork` | +|---|---|---| +| Topology known at | Runtime | Compile time | +| Type checking | Runtime (`dynamic_cast`) | Compile time | +| Overhead | Minimal | Zero | +| Flexibility | Add nodes dynamically | Fixed at compile time | + +For most applications `Network` is sufficient. Use `StaticNetwork` when you need the absolute minimum overhead or want compile-time topology verification. diff --git a/examples/01_hello_pipeline/main.cpp b/examples/01_hello_pipeline/main.cpp index a86580c..cfba5a9 100644 --- a/examples/01_hello_pipeline/main.cpp +++ b/examples/01_hello_pipeline/main.cpp @@ -7,20 +7,20 @@ // // [produce] --int--> [double_it] --int--> [print_it] -// [snippet: basic_node_fns] +// --8<-- [start:basic_node_fns] static int produce() { return 42; } static int double_it(int x) { return x * 2; } static void print_it(int x) { std::cout << "result: " << x << '\n'; } -// [/snippet: basic_node_fns] +// --8<-- [end:basic_node_fns] int main() { using namespace kpn; - // [snippet: index_only_nodes] + // --8<-- [start:index_only_nodes] auto src = make_node(5); auto dbl = make_node(5); auto sink = make_node(5); - // [/snippet: index_only_nodes] + // --8<-- [end:index_only_nodes] // Wire channels auto& dbl_in = dbl.input_channel<0>(); @@ -28,7 +28,7 @@ int main() { src.set_output_channel<0>(&dbl_in); dbl.set_output_channel<0>(&sink_in); - // [snippet: network_build] + // --8<-- [start:network_build] Network net; net.add("src", src) .add("dbl", dbl) @@ -40,5 +40,5 @@ int main() { net.start(); std::this_thread::sleep_for(std::chrono::milliseconds(100)); net.stop(); - // [/snippet: network_build] + // --8<-- [end:network_build] } diff --git a/examples/02_named_ports/main.cpp b/examples/02_named_ports/main.cpp index 752e316..a25e1c8 100644 --- a/examples/02_named_ports/main.cpp +++ b/examples/02_named_ports/main.cpp @@ -54,7 +54,7 @@ static void report(int count, std::vector words) { int main() { using namespace kpn; - // [snippet: named_port_creation] + // --8<-- [start:named_port_creation] // tokenise: no inputs, one named output "words" auto tok = make_node(out<"words">{}, 4); @@ -63,9 +63,9 @@ int main() { // report: two named inputs auto snk = make_node(in<"count", "words">{}, 4); - // [/snippet: named_port_creation] + // --8<-- [end:named_port_creation] - // [snippet: named_port_network] + // --8<-- [start:named_port_network] Network net; net.add("tok", tok) .add("cnt", cnt) @@ -78,5 +78,5 @@ int main() { net.start(); std::this_thread::sleep_for(std::chrono::milliseconds(500)); net.stop(); - // [/snippet: named_port_network] + // --8<-- [end:named_port_network] } diff --git a/examples/03_multi_output/main.cpp b/examples/03_multi_output/main.cpp index a783064..4e064d2 100644 --- a/examples/03_multi_output/main.cpp +++ b/examples/03_multi_output/main.cpp @@ -33,7 +33,7 @@ static std::string generate() { return pairs[gen_index++ % 5]; } -// [snippet: multi_output_fn] +// --8<-- [start:multi_output_fn] // Multi-output: returns (key, value) as a tuple โ€” KPN++ routes each element // to its own output port automatically. static std::tuple parse(std::string kv) { @@ -41,7 +41,7 @@ static std::tuple parse(std::string kv) { if (sep == std::string::npos) return {kv, ""}; return {kv.substr(0, sep), kv.substr(sep + 1)}; } -// [/snippet: multi_output_fn] +// --8<-- [end:multi_output_fn] static void print_key(std::string key) { std::cout << "KEY โ†’ " << key << '\n'; @@ -56,7 +56,7 @@ static void print_value(std::string value) { int main() { using namespace kpn; - // [snippet: fanout_network] + // --8<-- [start:fanout_network] auto gen = make_node(out<"kv">{}, 4); auto par = make_node (in<"kv">{}, out<"key", "value">{}, 4); auto keys = make_node (in<"key">{}, 4); @@ -75,5 +75,5 @@ int main() { net.start(); std::this_thread::sleep_for(std::chrono::milliseconds(600)); net.stop(); - // [/snippet: fanout_network] + // --8<-- [end:fanout_network] } diff --git a/examples/04_storage_policy/main.cpp b/examples/04_storage_policy/main.cpp index c46b73d..cff792e 100644 --- a/examples/04_storage_policy/main.cpp +++ b/examples/04_storage_policy/main.cpp @@ -34,14 +34,14 @@ struct Tag { int value = 0; }; -// [snippet: storage_policy_spec] +// --8<-- [start:storage_policy_spec] // Override: store Tag by value despite being a struct // (it's trivially copyable and small โ€” this just makes the policy explicit) template<> struct kpn::channel_storage_policy { static constexpr bool by_value = true; }; -// [/snippet: storage_policy_spec] +// --8<-- [end:storage_policy_spec] // โ”€โ”€ Node functions โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ diff --git a/examples/05_error_handling/main.cpp b/examples/05_error_handling/main.cpp index 395606b..3656561 100644 --- a/examples/05_error_handling/main.cpp +++ b/examples/05_error_handling/main.cpp @@ -45,7 +45,7 @@ int main() { Network net; - // [snippet: diagnostics_handler] + // --8<-- [start:diagnostics_handler] // Custom diagnostics handler โ€” fires on the watchdog interval. // Print a concise one-liner rather than the full table. net.set_diagnostics_handler([](const std::vector& nodes, @@ -58,7 +58,7 @@ int main() { << "overflows=" << c.overflows; std::cout << '\n'; }); - // [/snippet: diagnostics_handler] + // --8<-- [end:diagnostics_handler] net.set_watchdog_interval(std::chrono::milliseconds(200)); diff --git a/examples/09_opencv_cellshade/main.cpp b/examples/09_opencv_cellshade/main.cpp index c715107..bb250c3 100644 --- a/examples/09_opencv_cellshade/main.cpp +++ b/examples/09_opencv_cellshade/main.cpp @@ -38,7 +38,7 @@ static cv::Mat make_gradient(int W, int H) { // โ”€โ”€ Pipeline functions โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ -// [snippet: capture_fn] +// --8<-- [start:capture_fn] static std::tuple capture() { constexpr int W = 640, H = 480; static cv::VideoCapture cap; @@ -75,7 +75,7 @@ static std::tuple capture() { } return {frame.clone(), frame.clone()}; } -// [/snippet: capture_fn] +// --8<-- [end:capture_fn] static cv::Mat to_gray(cv::Mat bgr) { cv::Mat gray; @@ -120,7 +120,7 @@ static std::tuple composite(cv::Mat edge_mask, cv::Mat colour) // The constructor opens both windows on the main thread (Wayland requirement). // operator() is called by step() whenever both channels have a frame ready. -// [snippet: display_node] +// --8<-- [start:display_node] class DisplayNode : public kpn::MainThreadNode, cv::Mat, cv::Mat> { @@ -150,14 +150,14 @@ private: catch (const cv::Exception&) { return false; } } }; -// [/snippet: display_node] +// --8<-- [end:display_node] // โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ int main() { using namespace kpn; - // [snippet: opencv_network] + // --8<-- [start:opencv_network] auto src = make_node (out<"colour","grey">{}, 8); auto gray_node = make_node (in<"bgr">{}, out<"gray">{}, 8); auto edge_node = make_node (in<"gray">{}, out<"edges">{}, 8); @@ -182,7 +182,7 @@ int main() { .connect("comp", comp.template output<"result">(), "display", disp.template input<"composite">()) .connect("comp", comp.template output<"edges">(), "display", disp.template input<"edges">()) .build(); - // [/snippet: opencv_network] + // --8<-- [end:opencv_network] net.set_watchdog_interval(std::chrono::milliseconds(5000)); #ifdef KPN_WEB_DEBUG @@ -192,7 +192,7 @@ int main() { std::cout << "Cell-shading pipeline running. Press 'q' to stop.\n"; std::cout << "Web debug UI: http://localhost:9090\n"; - // [snippet: main_thread_step] + // --8<-- [start:main_thread_step] net.start(); // Main thread drives display โ€” imshow/waitKey stay on the GUI thread. @@ -201,6 +201,6 @@ int main() { cv::waitKey(8); // yield event loop when no frame ready net.stop(); - // [/snippet: main_thread_step] + // --8<-- [end:main_thread_step] return 0; } diff --git a/examples/15_node_error_handler/main.cpp b/examples/15_node_error_handler/main.cpp index 5cacad1..a5bc9a4 100644 --- a/examples/15_node_error_handler/main.cpp +++ b/examples/15_node_error_handler/main.cpp @@ -44,6 +44,7 @@ int main() { auto proc = make_node(); auto snk = make_node (); + // --8<-- [start:error_handler] // Return true โ†’ skip this invocation, keep the node running. // Return false โ†’ stop the node (downstream drains then also stops). proc.set_error_handler([](std::string_view name, std::exception_ptr ep) { @@ -53,6 +54,7 @@ int main() { } return true; }); + // --8<-- [end:error_handler] Network net; net.add("source", src) diff --git a/examples/16_event_callbacks/main.cpp b/examples/16_event_callbacks/main.cpp new file mode 100644 index 0000000..efa2c1b --- /dev/null +++ b/examples/16_event_callbacks/main.cpp @@ -0,0 +1,83 @@ +// Example 16 โ€” Event Callbacks: overflow and node-stopped signals +// +// Two complementary observation mechanisms: +// +// 1. Per-node overflow callback set_overflow_callback() +// Fired (with a timestamp) when a node's output channel is full and an +// item is dropped. Useful for targeted monitoring of a specific node. +// +// 2. Network-level event handler net.set_event_handler() +// Aggregate callback covering every node: receives the node name, a +// NodeEvent (Overflow or Closed), and a timestamp. Register once and +// observe the whole network. +// +// Pipeline: [fast_source] --int--> [slow_sink] +// +// fast_source produces at ~500 items/s; slow_sink consumes at ~20 items/s. +// The channel capacity is 3, so overflows appear within milliseconds. + +#include +#include +#include +#include +#include + +using namespace kpn; +using namespace std::chrono; + +// โ”€โ”€ Node functions โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +// --8<-- [start:node_fns] +static std::atomic g_seq{0}; + +static int fast_source() { + std::this_thread::sleep_for(milliseconds(2)); // ~500/s + return g_seq.fetch_add(1); +} + +static void slow_sink(int x) { + std::this_thread::sleep_for(milliseconds(50)); // ~20/s + std::cout << " consumed: " << x << '\n'; +} +// --8<-- [end:node_fns] + +// โ”€โ”€ main โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +int main() { + auto src = make_node(/*capacity=*/3); + auto snk = make_node (/*capacity=*/3); + + // --8<-- [start:per_node_callback] + // Per-node overflow callback โ€” no node name needed, known at registration. + std::atomic overflow_count{0}; + src.set_overflow_callback([&](steady_clock::time_point ts) { + auto ms = duration_cast(ts.time_since_epoch()).count(); + std::cerr << "[overflow] fast_source at t=" << ms << "ms\n"; + overflow_count.fetch_add(1); + }); + // --8<-- [end:per_node_callback] + + Network net; + + // --8<-- [start:network_event_handler] + // Network-level aggregate handler โ€” covers every node, includes node name. + net.set_event_handler([](std::string_view name, NodeEvent ev, + steady_clock::time_point ts) { + auto ms = duration_cast(ts.time_since_epoch()).count(); + std::string_view kind = (ev == NodeEvent::Overflow) ? "overflow" : "closed"; + std::cerr << "[net:" << kind << "] node=" << name << " t=" << ms << "ms\n"; + }); + // --8<-- [end:network_event_handler] + + net.add("source", src) + .add("sink", snk) + .connect("source", src.output<0>(), "sink", snk.input<0>()) + .build() + .start(); + + std::this_thread::sleep_for(milliseconds(300)); + net.stop(); + + std::cout << "\nTotal overflows observed by per-node callback: " + << overflow_count.load() << '\n'; +} diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index 0e8ad1a..261f5dd 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -1,8 +1,16 @@ cmake_minimum_required(VERSION 3.21) +# Build an example and register it as a CTest smoke test. +# Examples that are self-terminating (fixed sleep โ†’ net.stop()) pass when +# they exit 0 within TIMEOUT seconds. OpenCV/UI examples are excluded. function(kpn_example name) add_executable(${name} ${name}/main.cpp) target_link_libraries(${name} PRIVATE kpn) + add_test(NAME example_${name} COMMAND ${name}) + set_tests_properties(example_${name} PROPERTIES + TIMEOUT 15 + LABELS examples + ) endfunction() kpn_example(01_hello_pipeline) @@ -11,9 +19,11 @@ kpn_example(03_multi_output) kpn_example(04_storage_policy) kpn_example(05_error_handling) kpn_example(06_watchdog) -kpn_example(15_node_error_handler) +set_tests_properties(example_06_watchdog PROPERTIES TIMEOUT 40) kpn_example(10_static_hello_pipeline) kpn_example(11_static_fanout) +kpn_example(15_node_error_handler) +kpn_example(16_event_callbacks) if(KPN_WEB_DEBUG) kpn_target_enable_web_debug(06_watchdog) diff --git a/include/kpn/inode.hpp b/include/kpn/inode.hpp index 1b346bb..56f4b89 100644 --- a/include/kpn/inode.hpp +++ b/include/kpn/inode.hpp @@ -1,5 +1,6 @@ #pragma once #include "diagnostics.hpp" +#include #include #include #include @@ -10,6 +11,13 @@ namespace kpn { // invocation and keep running, false to stop the node. using NodeErrorHandler = std::function; +// Lightweight timestamp-only callback fired on per-node events. +// The node name is known at registration time so it is not included here. +using NodeEventCallback = std::function; + +// Event types reported to the network-level aggregate callback. +enum class NodeEvent { Overflow, Closed }; + // โ”€โ”€ INode โ€” type-erased interface for Network / watchdog โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ struct INode { @@ -21,6 +29,11 @@ struct INode { virtual NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const = 0; virtual void set_name(std::string name) = 0; + // Network-injected callbacks (slot 1 of each node's callback array). + // Default no-ops; overridden by PoolNode, PoolObjectNode, InterruptNode. + virtual void set_network_overflow_callback(NodeEventCallback) {} + virtual void set_network_closed_callback(NodeEventCallback) {} + // halt(): alias for stop() โ€” immediate, discards in-flight work. virtual void halt() { stop(); } diff --git a/include/kpn/interrupt_node.hpp b/include/kpn/interrupt_node.hpp index 8838be9..6e58c24 100644 --- a/include/kpn/interrupt_node.hpp +++ b/include/kpn/interrupt_node.hpp @@ -84,6 +84,11 @@ public: void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } + void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } + void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); } + void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } + void set_network_closed_callback(NodeEventCallback cb) override { closed_callbacks_[1] = std::move(cb); } + const NodeStats& stats() const override { return stats_; } NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { @@ -170,8 +175,8 @@ private: auto cpu1 = NodeStats::cpu_now(); auto t2 = clock_t::now(); stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1); - } catch (const ChannelOverflowError& e) { - std::cerr << "[kpn] interrupt node overflow: " << e.what() << "\n"; + } catch (const ChannelOverflowError&) { + fire_callbacks(event_callbacks_); } catch (...) { if (!error_handler_ || !error_handler_(name_, std::current_exception())) fatal = true; @@ -180,6 +185,8 @@ private: stats_.exec_start_us.store(0, std::memory_order_relaxed); if (fatal) { + fire_callbacks(closed_callbacks_); + disable_outputs(std::make_index_sequence{}); pending_.store(0, std::memory_order_release); stop_flag_.store(true, std::memory_order_relaxed); return; @@ -230,15 +237,28 @@ private: using output_channels_t = decltype(make_output_channel_tuple( std::make_index_sequence{})); - std::shared_ptr scheduler_; - std::string name_; - std::size_t fifo_capacity_; - output_channels_t output_channels_{}; - std::atomic stop_flag_{true}; - std::atomic pending_{0}; // triggers awaiting execution - NodeStats stats_; - NodeErrorHandler error_handler_; - std::chrono::milliseconds max_exec_time_{0}; + template + void disable_outputs(std::index_sequence) { + auto disable_one = [](auto* ch) { if (ch) ch->disable(); }; + (disable_one(std::get(output_channels_)), ...); + } + + static void fire_callbacks(const std::array& cbs) { + const auto ts = std::chrono::steady_clock::now(); + for (auto& cb : cbs) if (cb) cb(ts); + } + + std::shared_ptr scheduler_; + std::string name_; + std::size_t fifo_capacity_; + output_channels_t output_channels_{}; + std::atomic stop_flag_{true}; + std::atomic pending_{0}; + NodeStats stats_; + NodeErrorHandler error_handler_; + std::chrono::milliseconds max_exec_time_{0}; + std::array event_callbacks_{}; // [0]=user [1]=network + std::array closed_callbacks_{}; }; // โ”€โ”€ make_interrupt_node factory โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ diff --git a/include/kpn/network.hpp b/include/kpn/network.hpp index 3c9b745..239ad88 100644 --- a/include/kpn/network.hpp +++ b/include/kpn/network.hpp @@ -44,6 +44,9 @@ public: using DiagnosticsHandler = std::function&, const std::vector&)>; + using EventHandler = + std::function; // โ”€โ”€ Builder API โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ @@ -111,6 +114,19 @@ public: for (auto& [name, _] : nodes_) if (color[name] == 0) dfs(name, color); + if (event_handler_) { + for (auto& name : topo_) { + auto* node = nodes_.at(name); + node->set_network_overflow_callback( + [this, n = name](auto ts) { + event_handler_(n, NodeEvent::Overflow, ts); + }); + node->set_network_closed_callback( + [this, n = name](auto ts) { + event_handler_(n, NodeEvent::Closed, ts); + }); + } + } return *this; } @@ -194,8 +210,9 @@ public: watchdog_interval_ = interval; } - void set_error_handler(ErrorHandler h) { error_handler_ = std::move(h); } + void set_error_handler(ErrorHandler h) { error_handler_ = std::move(h); } void set_diagnostics_handler(DiagnosticsHandler h) { diag_handler_ = std::move(h); } + void set_event_handler(EventHandler h) { event_handler_ = std::move(h); } void register_pool(const std::string& name, IPoolProbe* probe) { pool_probes_.emplace_back(name, probe); @@ -430,6 +447,7 @@ private: std::vector> pool_probes_; ErrorHandler error_handler_; DiagnosticsHandler diag_handler_; + EventHandler event_handler_; std::chrono::milliseconds watchdog_interval_{3000}; std::jthread watchdog_; clock_t::time_point start_time_; diff --git a/include/kpn/pool_node.hpp b/include/kpn/pool_node.hpp index 0900da8..e4ceb54 100644 --- a/include/kpn/pool_node.hpp +++ b/include/kpn/pool_node.hpp @@ -101,6 +101,11 @@ public: void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } + void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } + void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); } + void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } + void set_network_closed_callback(NodeEventCallback cb) override { closed_callbacks_[1] = std::move(cb); } + const NodeStats& stats() const override { return stats_; } NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { @@ -189,12 +194,31 @@ private: (std::get(input_channels_)->disable(), ...); } + template + void disable_outputs(std::index_sequence) { + auto disable_one = [](auto* ch) { if (ch) ch->disable(); }; + (disable_one(std::get(output_channels_)), ...); + } + template void register_callbacks(std::index_sequence) { (std::get(input_channels_)->set_push_callback( [this] { on_input_ready(); }), ...); } + static void fire_callbacks(const std::array& cbs) { + const auto ts = std::chrono::steady_clock::now(); + for (auto& cb : cbs) if (cb) cb(ts); + } + + void self_stop() { + disable_inputs(std::make_index_sequence{}); + disable_outputs(std::make_index_sequence{}); + stats_.exec_start_us.store(0, std::memory_order_relaxed); + queued_.store(false, std::memory_order_release); + stop_flag_.store(true, std::memory_order_relaxed); + } + template static auto make_input_channel_tuple(std::index_sequence) -> std::tuple>>...>; @@ -278,19 +302,17 @@ private: // blocked_time = 0 for pool nodes (we don't block waiting for inputs) stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1); } catch (const ChannelClosedError&) { - stats_.exec_start_us.store(0, std::memory_order_relaxed); - queued_.store(false, std::memory_order_release); - stop_flag_.store(true, std::memory_order_relaxed); + fire_callbacks(closed_callbacks_); + self_stop(); return; - } catch (const ChannelOverflowError& e) { - std::cerr << "[kpn] pool overflow: " << e.what() << "\n"; + } catch (const ChannelOverflowError&) { + fire_callbacks(event_callbacks_); } catch (...) { if (error_handler_ && error_handler_(name_, std::current_exception())) { // continue โ€” fall through to resubmit check } else { - stats_.exec_start_us.store(0, std::memory_order_relaxed); - queued_.store(false, std::memory_order_release); - stop_flag_.store(true, std::memory_order_relaxed); + fire_callbacks(closed_callbacks_); + self_stop(); return; } } @@ -359,16 +381,18 @@ private: // โ”€โ”€ State โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ - std::shared_ptr scheduler_; - std::string name_; - std::size_t fifo_capacity_; - input_channels_t input_channels_; - output_channels_t output_channels_{}; - std::atomic stop_flag_{true}; - std::atomic queued_{false}; - NodeStats stats_; - NodeErrorHandler error_handler_; - std::chrono::milliseconds max_exec_time_{0}; + std::shared_ptr scheduler_; + std::string name_; + std::size_t fifo_capacity_; + input_channels_t input_channels_; + output_channels_t output_channels_{}; + std::atomic stop_flag_{true}; + std::atomic queued_{false}; + NodeStats stats_; + NodeErrorHandler error_handler_; + std::chrono::milliseconds max_exec_time_{0}; + std::array event_callbacks_{}; // [0]=user [1]=network + std::array closed_callbacks_{}; }; // โ”€โ”€ PoolObjectNode โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ @@ -435,6 +459,11 @@ public: void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } void set_max_exec_time(std::chrono::milliseconds t) { max_exec_time_ = t; } + void set_overflow_callback(NodeEventCallback cb) { event_callbacks_[0] = std::move(cb); } + void set_network_overflow_callback(NodeEventCallback cb) override { event_callbacks_[1] = std::move(cb); } + void set_closed_callback(NodeEventCallback cb) { closed_callbacks_[0] = std::move(cb); } + void set_network_closed_callback(NodeEventCallback cb) override { closed_callbacks_[1] = std::move(cb); } + const NodeStats& stats() const override { return stats_; } NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { @@ -490,13 +519,31 @@ private: std::make_shared>>(fifo_capacity_)), ...); } - template void enable_inputs(std::index_sequence) { (std::get(input_channels_)->enable(), ...); } - template void disable_inputs(std::index_sequence) { (std::get(input_channels_)->disable(), ...); } + template void enable_inputs(std::index_sequence) { (std::get(input_channels_)->enable(), ...); } + template void disable_inputs(std::index_sequence) { (std::get(input_channels_)->disable(), ...); } + template + void disable_outputs(std::index_sequence) { + auto disable_one = [](auto* ch) { if (ch) ch->disable(); }; + (disable_one(std::get(output_channels_)), ...); + } template void register_callbacks(std::index_sequence) { (std::get(input_channels_)->set_push_callback([this] { on_input_ready(); }), ...); } + static void fire_callbacks(const std::array& cbs) { + const auto ts = std::chrono::steady_clock::now(); + for (auto& cb : cbs) if (cb) cb(ts); + } + + void self_stop() { + disable_inputs(std::make_index_sequence{}); + disable_outputs(std::make_index_sequence{}); + stats_.exec_start_us.store(0, std::memory_order_relaxed); + queued_.store(false, std::memory_order_release); + stop_flag_.store(true, std::memory_order_relaxed); + } + template static auto make_input_channel_tuple(std::index_sequence) -> std::tuple>>...>; @@ -567,18 +614,16 @@ private: auto t2 = clock_t::now(); stats_.record_exec(duration_t(t2 - t1), duration_t::zero(), cpu0, cpu1); } catch (const ChannelClosedError&) { - stats_.exec_start_us.store(0, std::memory_order_relaxed); - queued_.store(false, std::memory_order_release); - stop_flag_.store(true, std::memory_order_relaxed); + fire_callbacks(closed_callbacks_); + self_stop(); return; - } catch (const ChannelOverflowError& e) { - std::cerr << "[kpn] pool overflow: " << e.what() << "\n"; + } catch (const ChannelOverflowError&) { + fire_callbacks(event_callbacks_); } catch (...) { if (error_handler_ && error_handler_(name_, std::current_exception())) { } else { - stats_.exec_start_us.store(0, std::memory_order_relaxed); - queued_.store(false, std::memory_order_release); - stop_flag_.store(true, std::memory_order_relaxed); + fire_callbacks(closed_callbacks_); + self_stop(); return; } } @@ -623,17 +668,19 @@ private: } } - Obj& obj_; - std::shared_ptr scheduler_; - std::string name_; - std::size_t fifo_capacity_; - input_channels_t input_channels_; - output_channels_t output_channels_{}; - std::atomic stop_flag_{true}; - std::atomic queued_{false}; - NodeStats stats_; - NodeErrorHandler error_handler_; - std::chrono::milliseconds max_exec_time_{0}; + Obj& obj_; + std::shared_ptr scheduler_; + std::string name_; + std::size_t fifo_capacity_; + input_channels_t input_channels_; + output_channels_t output_channels_{}; + std::atomic stop_flag_{true}; + std::atomic queued_{false}; + NodeStats stats_; + NodeErrorHandler error_handler_; + std::chrono::milliseconds max_exec_time_{0}; + std::array event_callbacks_{}; // [0]=user [1]=network + std::array closed_callbacks_{}; }; // โ”€โ”€ make_pool_node factory (NTTP) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ diff --git a/include/kpn/static_network.hpp b/include/kpn/static_network.hpp index 2750c0d..eeb5304 100644 --- a/include/kpn/static_network.hpp +++ b/include/kpn/static_network.hpp @@ -112,6 +112,16 @@ public: void start() override { stop_flag_ = false; start_time_ = clock_t::now(); + if (event_handler_) { + for (std::size_t i = 0; i < user_nodes_topo_.size(); ++i) { + auto* node = user_nodes_topo_[i]; + const auto& n = user_node_names_[i]; + node->set_network_overflow_callback( + [this, n](auto ts) { event_handler_(n, NodeEvent::Overflow, ts); }); + node->set_network_closed_callback( + [this, n](auto ts) { event_handler_(n, NodeEvent::Closed, ts); }); + } + } for (auto* n : user_nodes_topo_) n->start(); for (auto* n : fanout_nodes_ptr_) n->start(); #ifdef KPN_WEB_DEBUG @@ -165,6 +175,12 @@ public: return {n, 0, 0, 0, 0, 0, 0, 0}; } + using EventHandler = + std::function; + + void set_event_handler(EventHandler h) { event_handler_ = std::move(h); } + #ifdef KPN_WEB_DEBUG void set_web_debug_port(uint16_t port) { web_debug_port_ = port; } // Called by DebugHub::register_network() so the hub owns the debug server. @@ -259,6 +275,7 @@ private: std::vector> channel_probes_; std::vector> resource_probes_; std::vector> pool_probes_; + EventHandler event_handler_; clock_t::time_point start_time_; #ifdef KPN_WEB_DEBUG uint16_t web_debug_port_{9090}; diff --git a/mkdocs.yml b/mkdocs.yml new file mode 100644 index 0000000..597795d --- /dev/null +++ b/mkdocs.yml @@ -0,0 +1,50 @@ +site_name: KPN++ +site_description: A C++20 Kahn Process Network library +repo_url: https://github.com/yourusername/kpn +repo_name: kpn + +theme: + name: material + palette: + - scheme: slate + primary: indigo + accent: indigo + features: + - navigation.tabs + - navigation.sections + - navigation.top + - content.code.copy + - content.code.annotate + +nav: + - Home: index.md + - Getting Started: getting-started.md + - Concepts: + - Nodes: nodes.md + - Networks: network.md + - Channels: channels.md + - Error Handling & Events: error-handling.md + - Advanced: + - Static Networks: static-network.md + - Shared Resources: shared-resource.md + - Fan-out & Routing: fanout.md + - Examples: examples.md + +markdown_extensions: + - admonition + - toc: + permalink: true + - pymdownx.highlight: + anchor_linenums: true + line_spans: __span + pygments_lang_class: true + - pymdownx.inlinehilite + - pymdownx.superfences + - pymdownx.tabbed: + alternate_style: true + - pymdownx.snippets: + base_path: ['.'] + check_paths: true + - pymdownx.details + - attr_list + - md_in_html diff --git a/tests/test_pool_node.cpp b/tests/test_pool_node.cpp index 1d90b82..72d60ba 100644 --- a/tests/test_pool_node.cpp +++ b/tests/test_pool_node.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include using namespace kpn; @@ -239,3 +240,225 @@ TEST_CASE("interrupt node: trigger after stop is ignored", "[interrupt_node]") { REQUIRE(out_ch.approx_size() == 0); pool->stop(); } + +// โ”€โ”€ Overflow callback โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +TEST_CASE("pool node overflow callback fires on full output channel", "[pool_node][overflow]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool); + // Pre-fill a tiny channel so every node push overflows. + Channel full_ch(1); + full_ch.push(99); + node.set_output_channel<0>(&full_ch); + + std::atomic overflow_count{0}; + node.set_overflow_callback([&](auto) { overflow_count.fetch_add(1); }); + + node.start(); + node.input_channel<0>().push(1); + node.input_channel<0>().push(2); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + node.stop(); + pool->stop(); + + REQUIRE(overflow_count.load() > 0); +} + +TEST_CASE("pool node overflow callback is independent per instance", "[pool_node][overflow]") { + auto pool = std::make_shared(2); + pool->start(); + + auto nodeA = make_pool_node(pool); + auto nodeB = make_pool_node(pool); + + std::atomic a_overflows{0}, b_overflows{0}; + nodeA.set_overflow_callback([&](auto) { a_overflows.fetch_add(1); }); + + Channel full_ch(1); + full_ch.push(0); + nodeA.set_output_channel<0>(&full_ch); + + Channel ok_ch(20); + nodeB.set_output_channel<0>(&ok_ch); + + nodeA.start(); + nodeB.start(); + + nodeA.input_channel<0>().push(1); + nodeA.input_channel<0>().push(2); + nodeB.input_channel<0>().push(10); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + nodeA.stop(); + nodeB.stop(); + pool->stop(); + + REQUIRE(a_overflows.load() > 0); + REQUIRE(b_overflows.load() == 0); +} + +TEST_CASE("interrupt node overflow callback fires on full output", "[interrupt_node][overflow]") { + auto pool = std::make_shared(2); + pool->start(); + + g_interrupt_counter.store(0); + auto node = make_interrupt_node(pool, out<>{}); + + Channel full_ch(1); + full_ch.push(99); + node.set_output_channel<0>(&full_ch); + + std::atomic overflow_count{0}; + node.set_overflow_callback([&](auto) { overflow_count.fetch_add(1); }); + + node.start(); + auto trigger = node.get_trigger(); + trigger(); trigger(); trigger(); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + node.stop(); + pool->stop(); + + REQUIRE(overflow_count.load() > 0); +} + +// โ”€โ”€ self_stop: disable inputs + outputs on crash โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +static int always_throw(int) { throw std::runtime_error("node crashed"); return 0; } + +TEST_CASE("pool node self_stop disables output on crash so downstream sees closed", "[pool_node][self_stop]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool, 5); + Channel out_ch(10); + node.set_output_channel<0>(&out_ch); + + node.start(); + node.input_channel<0>().push(1); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + REQUIRE_FALSE(out_ch.is_accepting()); + + node.stop(); + pool->stop(); +} + +TEST_CASE("pool node self_stop disables input on crash", "[pool_node][self_stop]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool, 5); + Channel out_ch(5); + node.set_output_channel<0>(&out_ch); + + node.start(); + node.input_channel<0>().push(1); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + REQUIRE_FALSE(node.input_channel<0>().is_accepting()); + + node.stop(); + pool->stop(); +} + +TEST_CASE("pool node closed callback fires on self_stop from crash", "[pool_node][self_stop]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool, 5); + Channel out_ch(5); + node.set_output_channel<0>(&out_ch); + + std::atomic closed_fired{false}; + node.set_closed_callback([&](auto) { closed_fired.store(true); }); + + node.start(); + node.input_channel<0>().push(1); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + REQUIRE(closed_fired.load()); + + node.stop(); + pool->stop(); +} + +// โ”€โ”€ Network-level event callbacks โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +TEST_CASE("network_overflow_callback fires on overflow", "[pool_node][network]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool); + + Channel full_ch(1); + full_ch.push(0); + node.set_output_channel<0>(&full_ch); + + std::atomic net_overflows{0}; + node.set_network_overflow_callback([&](auto) { net_overflows.fetch_add(1); }); + + node.start(); + node.input_channel<0>().push(1); + node.input_channel<0>().push(2); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + node.stop(); + pool->stop(); + + REQUIRE(net_overflows.load() > 0); +} + +TEST_CASE("network_closed_callback fires on crash", "[pool_node][network]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool); + Channel out_ch(5); + node.set_output_channel<0>(&out_ch); + + std::atomic net_closed{false}; + node.set_network_closed_callback([&](auto) { net_closed.store(true); }); + + node.start(); + node.input_channel<0>().push(1); + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + REQUIRE(net_closed.load()); + + node.stop(); + pool->stop(); +} + +TEST_CASE("per-node and network overflow callbacks both fire independently", "[pool_node][network]") { + auto pool = std::make_shared(2); + pool->start(); + + auto node = make_pool_node(pool); + + Channel full_ch(1); + full_ch.push(0); + node.set_output_channel<0>(&full_ch); + + std::atomic per_node{0}, network{0}; + node.set_overflow_callback([&](auto) { per_node.fetch_add(1); }); + node.set_network_overflow_callback([&](auto) { network.fetch_add(1); }); + + node.start(); + node.input_channel<0>().push(1); + node.input_channel<0>().push(2); + + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + node.stop(); + pool->stop(); + + REQUIRE(per_node.load() > 0); + REQUIRE(network.load() > 0); +}