# 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):

```cpp
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'; }
```

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

```cpp
// Multi-output: returns (key, value) as a tuple — KPN++ routes each element
// to its own output port automatically.
static std::tuple<std::string, std::string> parse(std::string kv) {
    auto sep = kv.find('=');
    if (sep == std::string::npos) return {kv, ""};
    return {kv.substr(0, sep), kv.substr(sep + 1)};
}
```

### Creating Nodes

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

```cpp
auto src  = make_node<produce>(5);
auto dbl  = make_node<double_it>(5);
auto sink = make_node<print_it>(5);
```

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

```cpp
// tokenise: no inputs, one named output "words"
auto tok = make_node<tokenise>(out<"words">{}, 4);

// count_words: named input "words", named outputs "count" and "words"
auto cnt = make_node<count_words>(in<"words">{}, out<"count", "words">{}, 4);

// report: two named inputs
auto snk = make_node<report>(in<"count", "words">{}, 4);
```

**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):

```cpp
Network net;
net.add("tok", tok)
   .add("cnt", cnt)
   .add("snk", snk)
   .connect("tok", tok.template output<"words">(), "cnt", cnt.template input<"words">())
   .connect("cnt", cnt.template output<"count">(), "snk", snk.template input<"count">())
   .connect("cnt", cnt.template output<"words">(), "snk", snk.template input<"words">())
   .build();

net.start();
std::this_thread::sleep_for(std::chrono::milliseconds(500));
net.stop();
```

`.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)):

```cpp
// 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<Tag> {
    static constexpr bool by_value = true;
};
```

### Diagnostics & Error Handling

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

```cpp
// 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<NodeSnapshot>& nodes,
                               const std::vector<ChannelSnapshot>& channels) {
    std::cout << "[diag] ";
    for (auto& n : nodes)
        std::cout << n.name << "=" << n.throughput_fps << "fps  ";
    for (auto& c : channels)
        std::cout << "channel fill=" << static_cast<int>(c.fill_pct()) << "% "
                  << "overflows=" << c.overflows;
    std::cout << '\n';
});
```

### 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):

```cpp
class DisplayNode : public kpn::MainThreadNode<DisplayNode,
                                               kpn::in<"composite", "edges">,
                                               cv::Mat, cv::Mat> {
public:
    DisplayNode() : MainThreadNode(8) {
        cv::namedWindow("Cell Shade", cv::WINDOW_NORMAL);
        cv::namedWindow("Edge Mask",  cv::WINDOW_NORMAL);
        cv::resizeWindow("Cell Shade", 1280, 720);
        cv::resizeWindow("Edge Mask",   640, 360);
    }

    ~DisplayNode() { cv::destroyAllWindows(); }

    bool operator()(cv::Mat composite, cv::Mat edges) {
        cv::imshow("Cell Shade", composite);
        cv::Mat edges_bgr;
        cv::cvtColor(edges, edges_bgr, cv::COLOR_GRAY2BGR);
        cv::imshow("Edge Mask", edges_bgr);
        int key = cv::waitKey(1);
        if (key == 'q' || key == 27) return false;
        return window_open("Cell Shade") && window_open("Edge Mask");
    }

private:
    static bool window_open(const char* name) {
        try { return cv::getWindowProperty(name, cv::WND_PROP_VISIBLE) >= 1; }
        catch (const cv::Exception&) { return false; }
    }
};
```

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

```cpp
net.start();

// Main thread drives display — imshow/waitKey stay on the GUI thread.
// step() returns false when operator() returns false (q pressed / window closed).
while (disp.step())
    cv::waitKey(8); // yield event loop when no frame ready

net.stop();
```

---

## 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:

```cpp
static std::tuple<cv::Mat, cv::Mat> capture() {
    constexpr int W = 640, H = 480;
    static cv::VideoCapture cap;
    static bool opened = false;
    if (!opened) {
        opened = true;
        cap.open(0, cv::CAP_V4L2);
        if (cap.isOpened()) {
            cap.set(cv::CAP_PROP_FRAME_WIDTH,  W);
            cap.set(cv::CAP_PROP_FRAME_HEIGHT, H);
        } else {
            std::cerr << "[capture] no webcam — using synthetic animated pattern\n";
        }
    }

    cv::Mat frame;
    if (cap.isOpened()) {
        auto t0 = std::chrono::steady_clock::now();
        cap >> frame;
        auto elapsed = std::chrono::steady_clock::now() - t0;
        if (elapsed < std::chrono::milliseconds(20))
            std::this_thread::sleep_for(std::chrono::milliseconds(33) - elapsed);
        if (frame.empty()) frame = cv::Mat::zeros(H, W, CV_8UC3);
    } else {
        static int tick = 0;
        static cv::Mat grad = make_gradient(W, H);
        ++tick;
        frame = grad.clone();
        int r = 150 + (tick % 80) * 4;
        cv::circle(frame, {W/2,   H/2},   r,     {255, 200,   0}, -1);
        cv::circle(frame, {W/2,   H/2},   r / 2, {  0, 128, 255}, -1);
        cv::circle(frame, {W*2/5, H*2/5}, r / 3, {200,   0, 200}, -1);
        std::this_thread::sleep_for(std::chrono::milliseconds(33));
    }
    return {frame.clone(), frame.clone()};
}
```

**Full network wiring:**

```cpp
auto src       = make_node<capture>  (out<"colour","grey">{},            8);
auto gray_node = make_node<to_gray>  (in<"bgr">{},   out<"gray">{},      8);
auto edge_node = make_node<edges_fn> (in<"gray">{},  out<"edges">{},     8);
auto quant     = make_node<quantise> (in<"bgr">{},   out<"quantised">{}, 8);
auto comp      = make_node<composite>(in<"edges","colour">{}, out<"result","edges">{}, 8);

// DisplayNode: two windows opened in constructor, step() drives main thread.
DisplayNode disp;

Network net;
net.add("src",     src)
   .add("gray",    gray_node)
   .add("edges",   edge_node)
   .add("quant",   quant)
   .add("comp",    comp)
   .add("display", disp)
   .connect("src",   src.template output<"colour">(),      "quant",   quant.template input<"bgr">())
   .connect("quant", quant.template output<"quantised">(), "comp",    comp.template input<"colour">())
   .connect("src",   src.template output<"grey">(),        "gray",    gray_node.template input<"bgr">())
   .connect("gray",  gray_node.template output<"gray">(),  "edges",   edge_node.template input<"gray">())
   .connect("edges", edge_node.template output<"edges">(), "comp",    comp.template input<"edges">())
   .connect("comp",  comp.template output<"result">(),     "display", disp.template input<"composite">())
   .connect("comp",  comp.template output<"edges">(),      "display", disp.template input<"edges">())
   .build();
```

---

## 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:

```cpp
auto gen  = make_node<generate>(out<"kv">{},          4);
auto par  = make_node<parse>   (in<"kv">{},  out<"key", "value">{}, 4);
auto keys = make_node<print_key>  (in<"key">{},   4);
auto vals = make_node<print_value>(in<"value">{}, 4);

Network net;
net.add("gen",  gen)
   .add("par",  par)
   .add("keys", keys)
   .add("vals", vals)
   .connect("gen",  gen.template output<"kv">(),    "par",  par.template input<"kv">())
   .connect("par",  par.template output<"key">(),   "keys", keys.template input<"key">())
   .connect("par",  par.template output<"value">(), "vals", vals.template input<"value">())
   .build();

net.start();
std::this_thread::sleep_for(std::chrono::milliseconds(600));
net.stop();
```

---

## Python Bindings

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

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

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

Violating the second rule deadlocks.

---

## Examples

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

Run the cell-shading example:

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

---

## 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++ | 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++ | 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++ 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
```

---

## Contributing

Contributions are welcome. This project is hosted on a self-hosted Gitea
instance that accepts sign-in and registration with a GitHub account, so you
can log in with your existing GitHub identity to open issues and pull requests.

If you change any code that appears in a README snippet, edit `README.md.in`
(the template) rather than `README.md` directly, then regenerate:

```bash
cmake --build build --target readme   # or: python scripts/render_readme.py
```

---

## Acknowledgments

AI tooling was used heavily throughout the development of this project,
including the design, implementation, tests, and documentation. All output
has been reviewed, but please keep this in mind when reading or building on the
code.

---

## License

Released under the [MIT License](LICENSE). Copyright (c) 2026 Duncan Tourolle.