# KPN++

A C++20 Kahn Process Network (KPN) library. Each node wraps a function and runs in its own thread, communicating with downstream nodes via bounded FIFO channels. Includes Python bindings via nanobind.

---

## Requirements

| Dependency | Version | Notes |
|---|---|---|
| CMake | ≥ 3.21 | |
| C++ compiler | GCC ≥ 11, Clang ≥ 13, MSVC 19.29 | C++20 required |
| Threads | system | `find_package(Threads)` |
| nanobind | ≥ 2.1 | auto-fetched if not installed; Python ≥ 3.8 |
| Catch2 | v3 | auto-fetched for tests |
| Google Test | v1.14 | auto-fetched for tests |
| OpenCV | ≥ 4 | optional; only for example 09 |

---

## Build

```bash
cmake -B build -DKPN_BUILD_PYTHON=OFF   # core + tests + C++ examples
cmake --build build --parallel
ctest --test-dir build
```

Enable Python bindings (requires nanobind and Python dev headers):

```bash
cmake -B build -DKPN_BUILD_PYTHON=ON
cmake --build build --parallel
```

Disable examples:

```bash
cmake -B build -DKPN_BUILD_EXAMPLES=OFF
```

---

## Core Concepts

### Nodes

A node wraps any callable. Its input types are taken from the function's parameter list; its output types from the return type. Multi-output nodes return `std::tuple<...>`.

```cpp
#include <kpn/kpn.hpp>
using namespace kpn;
```

Source, transform, and sink — from [`examples/01_hello_pipeline/main.cpp`](examples/01_hello_pipeline/main.cpp):

<!-- @snippet examples/01_hello_pipeline/main.cpp basic_node_fns -->

Multi-output node returning a tuple — from [`examples/03_multi_output/main.cpp`](examples/03_multi_output/main.cpp):

<!-- @snippet examples/03_multi_output/main.cpp multi_output_fn -->

### Creating Nodes

**Index-only ports** (from [`examples/01_hello_pipeline/main.cpp`](examples/01_hello_pipeline/main.cpp)):

<!-- @snippet examples/01_hello_pipeline/main.cpp index_only_nodes -->

**Named ports** (from [`examples/02_named_ports/main.cpp`](examples/02_named_ports/main.cpp)):

<!-- @snippet examples/02_named_ports/main.cpp named_port_creation -->

**Multi-named output source** (from [`examples/09_opencv_cellshade/main.cpp`](examples/09_opencv_cellshade/main.cpp)):

```cpp
auto src = make_node<capture>(out<"colour","grey">{}, 8);
```

Port names are NTTP `fixed_string` values — resolved entirely at compile time, zero runtime cost.

### Building a Network

`Network` is **non-owning** — declare nodes first, then register them. Nodes must outlive the network.

From [`examples/02_named_ports/main.cpp`](examples/02_named_ports/main.cpp):

<!-- @snippet examples/02_named_ports/main.cpp named_port_network -->

`.build()` runs cycle detection — throws `NetworkCycleError` on cycles.

> **Named port syntax in template context:** when the node variable is `auto`-deduced, use `.template output<"name">()` and `.template input<"name">()` to help the parser.

### Channel Semantics

- **Bounded FIFO**: default capacity 5, configurable per-node at construction.
- **Blocking `pop()`**: consumer blocks until data is available (KPN semantics).
- **Throwing `push()`**: throws `ChannelOverflowError` if the channel is full and accepting.
- **Silent drop on disabled channel**: after `node.stop()`, its input channels are disabled — producers that push into them have the value silently dropped. No exception, no blocking.
- **Source throttling**: source nodes (no inputs) must sleep or yield to avoid overflowing downstream FIFOs. See example 09.

### Storage Policy

Large types (`sizeof > 8` or non-trivially-copyable) are stored as `std::shared_ptr<const T>` inside the channel — no copies, shared immutable ownership. Small trivially-copyable types are stored by value.

Override the policy for a specific type (from [`examples/04_storage_policy/main.cpp`](examples/04_storage_policy/main.cpp)):

<!-- @snippet examples/04_storage_policy/main.cpp storage_policy_spec -->

### Diagnostics & Error Handling

Custom diagnostics handler — fires on the watchdog interval (from [`examples/05_error_handling/main.cpp`](examples/05_error_handling/main.cpp)):

<!-- @snippet examples/05_error_handling/main.cpp diagnostics_handler -->

### Shutdown

`node.stop()` / `net.stop()`:
1. Sets `accepting_ = false` on all input channels (drops in-flight pushes silently).
2. Clears any queued items from those channels.
3. Unblocks any thread blocked on `pop()` (throws `ChannelClosedError` inside `run_loop`, which exits cleanly).
4. Joins the node thread.

---

## Named Ports — Design Notes

Port names use C++20 NTTP `fixed_string`. The deduction guide is required:

```cpp
template<std::size_t N>
fixed_string(const char (&)[N]) -> fixed_string<N>;
```

`fixed_string<4>` and `fixed_string<7>` are distinct types — `input<"img">()` and `input<"sigma">()` resolve to different template instantiations at compile time. Wrong names produce a `static_assert` at the call site with a readable message.

---

## Sub-Networks

`Network` implements `INode`, so it can be nested inside a larger `Network`:

```cpp
// Inner sub-network
Network pipe;
pipe.add("pre", pre_node)
    .add("enh", enh_node)
    .connect("pre", pre_node.output<0>(), "enh", enh_node.input<0>())
    .expose_input("img",    pre_node.input<0>())
    .expose_output("result", enh_node.output<0>())
    .build();

// Outer network
Network top;
top.add("pipe", pipe)
   .add("sink", sink_node)
   .connect("pipe", pipe.output<"result">(), "sink", sink_node.input<0>())
   .build();
top.start();
```

---

## Display / GUI Nodes

**Do not wrap `imshow`/`waitKey` as a KPN node.** Qt and Wayland require these to run on the main thread (the thread that owns the event loop). Instead, derive from `MainThreadNode<>` — it owns the input channels, implements `INode`, and exposes a `step()` method to call on the main thread.

`DisplayNode` from [`examples/09_opencv_cellshade/main.cpp`](examples/09_opencv_cellshade/main.cpp):

<!-- @snippet examples/09_opencv_cellshade/main.cpp display_node -->

Wire it into the network and drive it from the main thread:

<!-- @snippet examples/09_opencv_cellshade/main.cpp main_thread_step -->

---

## OpenCV Cell-Shading Example

Real-time cell-shading pipeline from [`examples/09_opencv_cellshade/main.cpp`](examples/09_opencv_cellshade/main.cpp).

**Source node** — returns two frames (colour + grey) as a tuple, routing them to separate downstream branches:

<!-- @snippet examples/09_opencv_cellshade/main.cpp capture_fn -->

**Full network wiring:**

<!-- @snippet examples/09_opencv_cellshade/main.cpp opencv_network -->

---

## Fan-Out (Multi-Output)

From [`examples/03_multi_output/main.cpp`](examples/03_multi_output/main.cpp) — one node fans out to two independent sinks via a tuple return:

<!-- @snippet examples/03_multi_output/main.cpp fanout_network -->

---

## Python Bindings

> Python bindings are scaffolded but not yet fully implemented. See `python/kpn_python.cpp` and `include/kpn/python/bindings.hpp`.

A `PyNetwork` is constructed from a closed list of C++ node types. The variant of all port types is derived at compile time — no runtime type registration needed.

**GIL rules (non-negotiable):**
- Acquire the GIL only for the duration of a Python callable invocation.
- Release the GIL before any blocking channel operation (`pop()`, `push()`, `net.read()`, `net.write()`).

Violating the second rule deadlocks.

---

## Examples

| Example | What it shows |
|---|---|
| `01_hello_pipeline` | Linear pipeline, index-based port wiring |
| `02_named_ports` | `in<>`/`out<>` name tags, named port access |
| `03_multi_output` | Tuple-returning node, per-element sub-port routing |
| `04_storage_policy` | `channel_storage_policy` default and specialisation |
| `05_error_handling` | `ChannelOverflowError`, `ErrorHandler` |
| `06_watchdog` | Watchdog interval, stall detection |
| `07_python_network` | PyNetwork, pure Python node *(pending)* |
| `08_python_subport` | `net.read`, `net.write`, sub-port tap *(pending)* |
| `09_opencv_cellshade` | Real-time cell-shading on webcam/pattern; requires OpenCV ≥ 4 |

Run the cell-shading example:

```bash
./build/examples/09_opencv_cellshade
# Press 'q' or close the window to stop.
# Falls back to an animated synthetic pattern if no webcam is found.
```

---

## Performance

Measured on Linux (x86-64, `-O3 -march=native`) with `benchmarks/bench_pipeline`.
Each topology pushes N items through the graph; `overhead_us/item` strips out the
per-node compute time to isolate framework cost.

Overhead formula: `(elapsed − (N + depth − 1) × work_us) / N` removes the expected
pipeline-fill cost so the number reflects pure framework latency.

### Baseline overhead (private pools, 100 µs/node)

| Topology | items/sec | overhead µs/item |
|---|---|---|
| chain depth-1 | 9 797 | ~2 |
| chain depth-4 | 9 448 | ~4 |
| chain depth-8 | 9 078 | ~7 |
| chain depth-16 | 7 004 | ~13 ← oversubscription |
| chain depth-32 | 4 179 | ~77 ← oversubscription |
| wide fanout-1 | 9 751 | ~3 |
| wide fanout-4 | 9 668 | ~3 |
| diamond (2×2) | 9 607 | ~4 |

Chain overhead is flat at **~2–7 µs/hop** for depths within the machine's core count,
then rises once threads compete for CPU. Wide and diamond topologies add no measurable
overhead as fanout increases — all branches run in parallel.

### Scheduling modes

`Node<>` gives each node a private `ThreadPool(1)`. `PoolNode<>` lets multiple
nodes share one pool. The right choice depends on the graph shape:

| Scenario | Recommended |
|---|---|
| Work per node < 100 µs, deep chain | Private pools — lower per-hop latency |
| Work per node ≥ 100 µs, wide/diamond | Shared pool, `threads = hardware_concurrency` |
| Any graph, bounded thread count required | Shared pool, `threads ≥ max parallel nodes` |

A shared single-thread pool (`threads=1`) fully serialises the graph — throughput
divides by depth for chains and by width for fanout topologies. A shared pool with
`threads ≥ max_concurrent_nodes` matches private-pool throughput while keeping the
OS thread count bounded.

### vs. TBB flow graph

Benchmarked against `tbb::flow::function_node<int,int>` (serial concurrency) with
`tbb::flow::broadcast_node<int>` for fanout. Run with `cmake -DKPN_BUILD_BENCHMARKS=ON`
— TBB benchmarks are included automatically when `find_package(TBB)` succeeds.

**Channel implementation:** lock-free SPSC ring buffer with `std::atomic::wait/notify_one`
(C++20 portable futex) plus a configurable spin-before-sleep window (default ~4 µs).
Large types are stored as `shared_ptr<const T>` — fanout copies reference counts,
not data.

Overhead µs/item at **work_us = 10** (framework overhead dominates):

| Topology | KPN private | TBB |
|---|---|---|
| chain depth-1 | 1.7 | **1.4** |
| chain depth-4 | 2.5 | **2.2** |
| chain depth-8 | **3.0** | 3.6 |
| chain depth-16 | **9.3** | 13.0 |
| chain depth-32 | 23.2 | **14.2** |
| wide fanout-4 | 2.5 | **1.4** |
| diamond (2×2) | 3.4 | **1.9** |

Overhead µs/item at **work_us = 100** (moderate compute, KPN wins):

| Topology | KPN private | TBB |
|---|---|---|
| chain depth-1 | **2.1** | 3.5 |
| chain depth-4 | **4.3** | 5.2 |
| chain depth-8 | **6.7** | 8.5 |
| chain depth-16 | **12.8** | 17.4 |
| chain depth-32 | **77** | 81 |
| wide fanout-4 | 3.4 | **1.9** |
| diamond (2×2) | **4.1** | 6.1 |

KPN private pools beat TBB for every chain and diamond topology at 100 µs/node, and
match TBB within ~20% at 10 µs/node for shallow chains. TBB retains an edge on wide
fanout (serial dispatch loop vs. work-stealing pool) and at extreme oversubscription
depths (chain-32 at 10 µs). The remaining gap at light work is the cost of
`atomic::wait` vs. TBB's continuously-spinning worker threads.

### vs. TBB — API

The function signature is the node. KPN infers input and output types automatically;
there is no graph object to manage.

**Single-output node:**
```cpp
// KPN — 1 line
int scale(int x) { return x * 2; }

// TBB — must state types, concurrency policy, and carry a graph reference
tbb::flow::function_node<int,int> n(g, tbb::flow::serial, [](int x){ return x*2; });
```

**Multi-output node:**
```cpp
// KPN — return a tuple
std::tuple<cv::Mat,cv::Mat> split(cv::Mat f) { return {f, f}; }

// TBB — multifunction_node + explicit try_put per port
tbb::flow::multifunction_node<cv::Mat, std::tuple<cv::Mat,cv::Mat>> n(
    g, tbb::flow::serial,
    [](cv::Mat f, auto& ports) {
        std::get<0>(ports).try_put(f);
        std::get<1>(ports).try_put(f);
    });
```

**Named ports** — compile-time checked, zero runtime cost, not available in TBB:
```cpp
auto node = make_node<split>(in<"frame">{}, out<"colour","grey">{}, 5);
net.connect("cam", cam.output<"frame">(), "split", node.input<"frame">());
//                                                         ^^^^^^^ typo → compile error
```

| | KPN | TBB |
|---|---|---|
| Node definition | plain function | `function_node<In,Out>` + explicit types |
| Multi-output | `return std::tuple<A,B>` | `multifunction_node` + `try_put` × N |
| Named ports | `in<"name">` / `out<"name">` compile-time | none |
| Graph lifetime | none | `graph g` must outlive all nodes |
| Shutdown | `net.stop()` | `g.wait_for_all()` + manual |
| Python bindings | designed-in | none |

Build the benchmarks with:

```bash
cmake -B build -DKPN_BUILD_BENCHMARKS=ON
cmake --build build --target bench_pipeline
./build/benchmarks/bench_pipeline | tee results.csv
```

---

## Project Structure

```
include/kpn/
  fixed_string.hpp   — NTTP string, in<>/out<> tags, index_of
  traits.hpp         — function_traits, normalised_return_t, output_count_v
  channel.hpp        — Channel<T>, channel_storage_policy, exceptions
  port.hpp           — InputPort<N,I>, OutputPort<N,I>
  node.hpp           — Node<Func,in<...>,out<...>>, make_node, INode
  network.hpp        — Network (builder, cycle detection, watchdog)
  variant_node.hpp   — VariantNode, PythonConverter<T>, unique_types (Python layer)
  python/
    bindings.hpp     — nanobind helpers, GIL rule documentation
  kpn.hpp            — umbrella header
src/
  network.cpp        — non-template Network implementation
tests/
  test_fixed_string.cpp
  test_traits.cpp
  test_channel.cpp
  test_node.cpp
  test_network.cpp
python/
  kpn_python.cpp     — nanobind module entry point
examples/
  01_hello_pipeline/ … 09_opencv_cellshade/
scripts/
  render_readme.py   — regenerates README.md from README.md.in
```