This commit is contained in:
parent
127ffb3849
commit
3c683c821d
@ -1,25 +0,0 @@
|
||||
name: '🐳 Build Builder Image'
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
paths:
|
||||
- 'Dockerfile.builder'
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
build-and-push:
|
||||
runs-on: linux/amd64
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Log in to Gitea registry
|
||||
run: echo "${{ secrets.REGISTRY_PASSWORD }}" | docker login gitea.tourolle.paris -u "${{ secrets.REGISTRY_USER }}" --password-stdin
|
||||
|
||||
- name: Build image
|
||||
run: docker build -f Dockerfile.builder -t gitea.tourolle.paris/dtourolle/kpnpp-builder:latest .
|
||||
|
||||
- name: Push image
|
||||
run: docker push gitea.tourolle.paris/dtourolle/kpnpp-builder:latest
|
||||
88
SPEC.md
88
SPEC.md
@ -304,15 +304,16 @@ private:
|
||||
};
|
||||
```
|
||||
|
||||
**Factory syntax — `in<>` / `out<>` tag structs:**
|
||||
**Factory syntax — `in<>` / `latch<>` / `out<>` tag structs:**
|
||||
|
||||
A flat name pack `make_node<f, "a", "b", "c">` is ambiguous (where do inputs end?).
|
||||
Option chosen: `in<...>` and `out<...>` tag types that wrap the name packs unambiguously.
|
||||
Both are optional; omitting either means those ports are index-only.
|
||||
Option chosen: `in<...>`, `latch<...>`, and `out<...>` tag types that wrap the name packs unambiguously.
|
||||
All are optional; omitting either means those ports are index-only.
|
||||
|
||||
```cpp
|
||||
// Tag types (trivial, no data):
|
||||
template<fixed_string... Names> struct in {};
|
||||
template<fixed_string... Names> struct latch {};
|
||||
template<fixed_string... Names> struct out {};
|
||||
|
||||
// Factory:
|
||||
@ -324,6 +325,9 @@ auto node = make_node<my_func, in<"img","sigma">>(10);
|
||||
|
||||
// Both input and output names
|
||||
auto node = make_node<my_func, in<"img","sigma">, out<"blurred","mask">>(10);
|
||||
|
||||
// Mixed synchronous and latched inputs
|
||||
auto node = make_node<my_func, in<"error">, latch<"setpoint">, out<"output">>(10);
|
||||
```
|
||||
|
||||
**Wrong name count is a compile error.** The `Node` class `static_assert`s that
|
||||
@ -342,6 +346,83 @@ Multi-output functions must return `std::tuple<...>`. Single return accepted as-
|
||||
|
||||
---
|
||||
|
||||
## Component 4a — Latched Input Ports
|
||||
|
||||
### Motivation
|
||||
|
||||
Control and robotics applications naturally have two kinds of inputs at different update rates:
|
||||
|
||||
- **Synchronous inputs** (`in<>`) — the node must have fresh data on every fire. Typical for sensor readings that drive the computation (e.g. encoder RPM).
|
||||
- **Latched inputs** (`latch<>`) — the node uses the most recently received value, and does not block if no new value has arrived. Typical for setpoints or parameters that change infrequently relative to the control loop (e.g. bearing from a CV pipeline, PID gains).
|
||||
|
||||
Without latched ports, a node must block on all inputs simultaneously. This forces the control loop to run at the rate of the slowest input — unacceptable when a 1kHz encoder loop must wait for a 30Hz vision update.
|
||||
|
||||
### Semantics
|
||||
|
||||
A `latch<>` port:
|
||||
|
||||
1. **Does not block** if its channel is empty — it reuses the last successfully popped value.
|
||||
2. **Does block on first fire** — there is no meaningful "default" value, so the node waits until at least one value has arrived on each latched port before firing for the first time.
|
||||
3. **Consumes the value** when one is available (standard `pop()`), then holds it until the next value arrives.
|
||||
|
||||
The node fires whenever all `in<>` ports have data, using the last known value for each `latch<>` port.
|
||||
|
||||
### Implementation in `run_loop`
|
||||
|
||||
`run_loop` maintains a `std::tuple` of cached values, one slot per latched port. On each iteration:
|
||||
|
||||
```cpp
|
||||
// Synchronous ports — blocking pop (existing behaviour)
|
||||
auto sync_args = std::make_tuple(input<0>().pop(), input<1>().pop(), ...);
|
||||
|
||||
// Latched ports — non-blocking try_pop; keep cached value on miss
|
||||
try_pop(latch_cache_<I>, latch_channel_<I>); // updates cache if data available
|
||||
|
||||
// Call wrapped function with merged argument tuple
|
||||
auto result = std::apply(Func, merge(sync_args, latch_cache_));
|
||||
```
|
||||
|
||||
Latched channels are otherwise identical to synchronous channels: bounded FIFO, `shared_ptr` storage policy, same shutdown behaviour.
|
||||
|
||||
### Example — PID with live setpoint
|
||||
|
||||
```cpp
|
||||
// bearing arrives at ~30 Hz from CV; rpm arrives at ~1 kHz from encoder
|
||||
double pid_compute(double rpm, double bearing) { ... }
|
||||
|
||||
auto pid = make_node<pid_compute,
|
||||
in<"rpm">, // synchronous — blocks until fresh encoder tick
|
||||
latch<"bearing">, // latched — uses last known bearing from CV
|
||||
out<"pwm">
|
||||
>(8);
|
||||
|
||||
Network net;
|
||||
net.add("tacho", tacho_node)
|
||||
.add("tracker", tracker_node)
|
||||
.add("pid", pid)
|
||||
.connect("tacho", tacho_node.output<"rpm">(), "pid", pid.input<"rpm">())
|
||||
.connect("tracker", tracker_node.output<"bearing">(),"pid", pid.input<"bearing">())
|
||||
.build();
|
||||
```
|
||||
|
||||
The PID node fires at encoder rate. If no new bearing has arrived since the last tick, it reuses the previous one — correct behaviour for a control loop.
|
||||
|
||||
### Port Ordering Contract
|
||||
|
||||
`in<>` and `latch<>` ports together must cover all function parameters in declaration order. The `static_assert` on name count is extended to cover both tags jointly:
|
||||
|
||||
```cpp
|
||||
static_assert(
|
||||
sizeof...(InNames) + sizeof...(LatchNames) == input_count ||
|
||||
(sizeof...(InNames) == 0 && sizeof...(LatchNames) == 0),
|
||||
"make_node: in<> and latch<> names together must match function arity, or provide none"
|
||||
);
|
||||
```
|
||||
|
||||
The function parameter at position `i` is synchronous if `i` is in the `in<>` pack, latched if in the `latch<>` pack. Mixed ordering is allowed — the tag packs define which positions are latched, not a contiguous suffix.
|
||||
|
||||
---
|
||||
|
||||
## Component 5 — `network.hpp`: Graph Builder + Orchestrator
|
||||
|
||||
`Network` is **non-owning** — nodes are declared by the user and must outlive the network.
|
||||
@ -997,3 +1078,4 @@ All major design questions are now closed:
|
||||
| Sub-networks | `Network` implements `INode`; `expose_input`/`expose_output` define boundary ports |
|
||||
| `make_py_network` | Pure C++ template; nanobind module recompilation is the registration step |
|
||||
| GIL strategy | Acquire per Python callback; release while blocking on channel ops |
|
||||
| Mixed-rate inputs | `latch<>` tag for ports that reuse last-seen value; blocks only on first fire; node fires at rate of `in<>` ports |
|
||||
|
||||
@ -1,3 +1,80 @@
|
||||
// Example 02 — Named Ports
|
||||
//
|
||||
// A three-stage text pipeline where port names carry semantic meaning:
|
||||
//
|
||||
// [tokenise] --"words"--> [count_words] --"count"--> [report]
|
||||
// --"words"--> [report]
|
||||
//
|
||||
// Named ports let the connect() call read like documentation:
|
||||
// connect("tok", tok.output<"words">(), "cnt", cnt.input<"words">())
|
||||
// A typo in the name is a compile-time error, not a runtime surprise.
|
||||
|
||||
#include <kpn/kpn.hpp>
|
||||
// TODO: implement example
|
||||
int main() { return 0; }
|
||||
#include <chrono>
|
||||
#include <iostream>
|
||||
#include <sstream>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <tuple>
|
||||
#include <vector>
|
||||
|
||||
// ── Node functions ────────────────────────────────────────────────────────────
|
||||
|
||||
static int sentence_index = 0;
|
||||
|
||||
static std::vector<std::string> tokenise() {
|
||||
static const char* sentences[] = {
|
||||
"the quick brown fox jumps over the lazy dog",
|
||||
"kahn process networks are a model of concurrent computation",
|
||||
"each node runs in its own thread communicating via channels",
|
||||
"named ports catch wiring mistakes at compile time",
|
||||
};
|
||||
std::istringstream ss(sentences[sentence_index++ % 4]);
|
||||
std::vector<std::string> words;
|
||||
std::string w;
|
||||
while (ss >> w) words.push_back(w);
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
return words;
|
||||
}
|
||||
|
||||
// Returns (word_count, original_words) — two outputs via tuple
|
||||
static std::tuple<int, std::vector<std::string>>
|
||||
count_words(std::vector<std::string> words) {
|
||||
return {static_cast<int>(words.size()), std::move(words)};
|
||||
}
|
||||
|
||||
static void report(int count, std::vector<std::string> words) {
|
||||
std::cout << "[" << count << " words] ";
|
||||
for (auto& w : words) std::cout << w << ' ';
|
||||
std::cout << '\n';
|
||||
}
|
||||
|
||||
// ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
int main() {
|
||||
using namespace kpn;
|
||||
|
||||
// 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 — note the function takes (int, vector<string>)
|
||||
// so we need two separate input ports wired independently
|
||||
// For a two-input sink we wire each output of cnt to a different input of report
|
||||
auto snk = make_node<report>(in<"count", "words">{}, 4);
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
@ -1,3 +1,75 @@
|
||||
// Example 03 — Multi-Output (Fan-Out)
|
||||
//
|
||||
// A single "parse" node reads "KEY=VALUE" strings and fans out to two
|
||||
// independent downstream sinks — one for keys, one for values.
|
||||
//
|
||||
// +--> [print_key]
|
||||
// [generate] --string--> [parse]
|
||||
// +--> [print_value]
|
||||
//
|
||||
// The parser returns std::tuple<std::string, std::string>.
|
||||
// Network::connect() routes each element of the tuple to a different node.
|
||||
|
||||
#include <kpn/kpn.hpp>
|
||||
// TODO: implement example
|
||||
int main() { return 0; }
|
||||
#include <chrono>
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <tuple>
|
||||
|
||||
// ── Node functions ────────────────────────────────────────────────────────────
|
||||
|
||||
static int gen_index = 0;
|
||||
|
||||
static std::string generate() {
|
||||
static const char* pairs[] = {
|
||||
"host=localhost",
|
||||
"port=8080",
|
||||
"timeout=30s",
|
||||
"retries=3",
|
||||
"protocol=http2",
|
||||
};
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(60));
|
||||
return pairs[gen_index++ % 5];
|
||||
}
|
||||
|
||||
// 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)};
|
||||
}
|
||||
|
||||
static void print_key(std::string key) {
|
||||
std::cout << "KEY → " << key << '\n';
|
||||
}
|
||||
|
||||
static void print_value(std::string value) {
|
||||
std::cout << "VALUE → " << value << '\n';
|
||||
}
|
||||
|
||||
// ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
int main() {
|
||||
using namespace kpn;
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
@ -1,3 +1,118 @@
|
||||
// Example 04 — Storage Policy
|
||||
//
|
||||
// Demonstrates how KPN++ chooses between by-value and shared_ptr storage
|
||||
// inside channels depending on the type.
|
||||
//
|
||||
// Small trivially-copyable types (int, double, etc.) are stored by value.
|
||||
// Large or non-trivially-copyable types are stored as shared_ptr<const T>
|
||||
// — zero copies even across multiple downstream consumers.
|
||||
//
|
||||
// This example also shows how to override the policy for a specific type
|
||||
// with a template specialisation.
|
||||
//
|
||||
// [produce_frame] --Frame--> [process_frame] --Frame--> [consume_frame]
|
||||
// [produce_small] --int----> [double_it] --int----> [print_int]
|
||||
|
||||
#include <kpn/kpn.hpp>
|
||||
// TODO: implement example
|
||||
int main() { return 0; }
|
||||
#include <array>
|
||||
#include <chrono>
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <type_traits>
|
||||
|
||||
// ── Types ─────────────────────────────────────────────────────────────────────
|
||||
|
||||
// Large frame type — will be stored as shared_ptr<const Frame> by default
|
||||
struct Frame {
|
||||
std::array<uint8_t, 4096> pixels{};
|
||||
int id = 0;
|
||||
};
|
||||
|
||||
// A small struct we force to be stored by value via policy specialisation
|
||||
struct Tag {
|
||||
int value = 0;
|
||||
};
|
||||
|
||||
// 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;
|
||||
};
|
||||
|
||||
// ── Node functions ────────────────────────────────────────────────────────────
|
||||
|
||||
static int frame_id = 0;
|
||||
static int tag_id = 0;
|
||||
|
||||
static Frame produce_frame() {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(40));
|
||||
Frame f;
|
||||
f.id = frame_id++;
|
||||
f.pixels.fill(static_cast<uint8_t>(f.id & 0xFF));
|
||||
return f;
|
||||
}
|
||||
|
||||
static Frame process_frame(Frame f) {
|
||||
// Simulate some work
|
||||
for (auto& p : f.pixels) p = static_cast<uint8_t>(255 - p);
|
||||
return f;
|
||||
}
|
||||
|
||||
static void consume_frame(Frame f) {
|
||||
std::cout << "[frame] id=" << f.id
|
||||
<< " first_px=" << static_cast<int>(f.pixels[0])
|
||||
<< " storage=shared_ptr (sizeof Frame = " << sizeof(Frame) << " B)\n";
|
||||
}
|
||||
|
||||
static Tag produce_tag() {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(40));
|
||||
return {tag_id++};
|
||||
}
|
||||
|
||||
static Tag double_tag(Tag t) { return {t.value * 2}; }
|
||||
|
||||
static void print_tag(Tag t) {
|
||||
std::cout << "[tag] value=" << t.value
|
||||
<< " storage=by_value (sizeof Tag = " << sizeof(Tag) << " B)\n";
|
||||
}
|
||||
|
||||
// ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
int main() {
|
||||
using namespace kpn;
|
||||
|
||||
// Verify storage decisions at compile time
|
||||
static_assert(!channel_storage_policy<Frame>::by_value,
|
||||
"Frame should use shared_ptr storage");
|
||||
static_assert(channel_storage_policy<Tag>::by_value,
|
||||
"Tag should use by_value storage (via specialisation)");
|
||||
static_assert(channel_storage_policy<int>::by_value,
|
||||
"int should use by_value storage");
|
||||
|
||||
auto prod_f = make_node<produce_frame> (out<"frame">{}, 4);
|
||||
auto proc_f = make_node<process_frame> (in<"frame">{}, out<"frame">{}, 4);
|
||||
auto cons_f = make_node<consume_frame> (in<"frame">{}, 4);
|
||||
|
||||
auto prod_t = make_node<produce_tag> (out<"tag">{}, 4);
|
||||
auto dbl_t = make_node<double_tag> (in<"tag">{}, out<"tag">{}, 4);
|
||||
auto print_t = make_node<print_tag> (in<"tag">{}, 4);
|
||||
|
||||
Network net;
|
||||
net.add("prod_f", prod_f)
|
||||
.add("proc_f", proc_f)
|
||||
.add("cons_f", cons_f)
|
||||
.add("prod_t", prod_t)
|
||||
.add("dbl_t", dbl_t)
|
||||
.add("print_t", print_t)
|
||||
.connect("prod_f", prod_f.template output<"frame">(), "proc_f", proc_f.template input<"frame">())
|
||||
.connect("proc_f", proc_f.template output<"frame">(), "cons_f", cons_f.template input<"frame">())
|
||||
.connect("prod_t", prod_t.template output<"tag">(), "dbl_t", dbl_t.template input<"tag">())
|
||||
.connect("dbl_t", dbl_t.template output<"tag">(), "print_t", print_t.template input<"tag">())
|
||||
.build();
|
||||
|
||||
net.start();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(400));
|
||||
net.stop();
|
||||
}
|
||||
|
||||
@ -1,3 +1,78 @@
|
||||
// Example 05 — Error Handling & Diagnostics
|
||||
//
|
||||
// Demonstrates observable failure modes and the diagnostics system:
|
||||
//
|
||||
// 1. ChannelOverflowError — a fast producer saturates a slow consumer.
|
||||
// When the channel is full, push() throws ChannelOverflowError.
|
||||
// The node's run_loop catches it and prints to stderr.
|
||||
// Channel statistics (overflows, peak fill) accumulate for the report.
|
||||
//
|
||||
// 2. Custom diagnostics handler — instead of the default periodic table,
|
||||
// install a handler that surfaces only the metrics you care about.
|
||||
//
|
||||
// 3. net.print_diagnostics() — print a full report at any time.
|
||||
//
|
||||
// Pipeline: [producer] --int--> [slow_consumer]
|
||||
|
||||
#include <kpn/kpn.hpp>
|
||||
// TODO: implement example
|
||||
int main() { return 0; }
|
||||
#include <chrono>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
// ── Node functions ────────────────────────────────────────────────────────────
|
||||
|
||||
// Produces at ~100/s — faster than the consumer can keep up (50 ms each)
|
||||
static int producer() {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||||
static int n = 0;
|
||||
return ++n;
|
||||
}
|
||||
|
||||
// Slow consumer: 50 ms per item — will cause channel to fill and overflow
|
||||
static void slow_consumer(int x) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
std::cout << "[consumed] " << x << '\n';
|
||||
}
|
||||
|
||||
// ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
int main() {
|
||||
using namespace kpn;
|
||||
|
||||
// Capacity=4: fills up quickly when producer outpaces consumer 5:1
|
||||
auto prod = make_node<producer> (out<"v">{}, /*capacity=*/4);
|
||||
auto cons = make_node<slow_consumer> (in<"v">{}, /*capacity=*/4);
|
||||
|
||||
Network net;
|
||||
|
||||
// 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';
|
||||
});
|
||||
|
||||
net.set_watchdog_interval(std::chrono::milliseconds(200));
|
||||
|
||||
net.add("prod", prod)
|
||||
.add("cons", cons)
|
||||
.connect("prod", prod.template output<"v">(), "cons", cons.template input<"v">())
|
||||
.build();
|
||||
|
||||
std::cout << "producer: 10ms/item, consumer: 50ms/item, capacity=4\n"
|
||||
<< "Overflow messages appear on stderr; diagnostics on stdout.\n\n";
|
||||
|
||||
net.start();
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(800));
|
||||
net.stop();
|
||||
|
||||
// Full report after shutdown — overflow and drop counts are preserved
|
||||
std::cout << '\n';
|
||||
net.print_diagnostics();
|
||||
}
|
||||
|
||||
@ -1,3 +1,90 @@
|
||||
// Example 06 — Watchdog & Diagnostics (+ optional web debug UI)
|
||||
//
|
||||
// A four-stage pipeline with deliberately uneven node speeds:
|
||||
//
|
||||
// [source] --int--> [fast_filter] --int--> [slow_transform] --int--> [sink]
|
||||
//
|
||||
// The watchdog fires every second and prints a full diagnostics report:
|
||||
// - frames processed, throughput fps, exec time, blocked time per node
|
||||
// - channel fill %, peak fill %, pushes, overflows, bandwidth
|
||||
// - bottleneck hint (node with highest avg exec time)
|
||||
//
|
||||
// "slow_transform" sleeps 30 ms per item, making it the obvious bottleneck.
|
||||
// Watch the channel upstream of it saturate and the fps converge to ~33.
|
||||
//
|
||||
// Build with -DKPN_WEB_DEBUG=ON to also get a live D3 graph at localhost:9090.
|
||||
|
||||
#include <kpn/kpn.hpp>
|
||||
// TODO: implement example
|
||||
int main() { return 0; }
|
||||
#include <chrono>
|
||||
#include <cmath>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
// ── Node functions ────────────────────────────────────────────────────────────
|
||||
|
||||
static int source() {
|
||||
// Produces at ~50 fps — faster than slow_transform (33 fps) so the
|
||||
// channel between filter and slow gradually fills, but not catastrophically
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
static int n = 0;
|
||||
return ++n;
|
||||
}
|
||||
|
||||
static int fast_filter(int x) {
|
||||
// Trivial work: ~0.1 ms
|
||||
return (x % 2 == 0) ? x : x + 1;
|
||||
}
|
||||
|
||||
static int slow_transform(int x) {
|
||||
// Simulates expensive processing (e.g. a neural net inference step)
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(30));
|
||||
return x * x;
|
||||
}
|
||||
|
||||
static void sink(int x) {
|
||||
// Print every 10th result to avoid flooding the terminal
|
||||
if (x % 100 < 4)
|
||||
std::cout << "[sink] " << x << '\n';
|
||||
}
|
||||
|
||||
// ── main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
int main() {
|
||||
using namespace kpn;
|
||||
|
||||
// Larger capacity buffers so the fast nodes don't immediately overflow
|
||||
auto src = make_node<source> (out<"v">{}, 16);
|
||||
auto filt = make_node<fast_filter> (in<"v">{}, out<"v">{}, 16);
|
||||
auto slow = make_node<slow_transform> (in<"v">{}, out<"v">{}, 8);
|
||||
auto snk = make_node<sink> (in<"v">{}, 8);
|
||||
|
||||
Network net;
|
||||
|
||||
// Watchdog fires every 1 second — prints the built-in diagnostics table
|
||||
// including the "Bottleneck hint" line
|
||||
net.set_watchdog_interval(std::chrono::milliseconds(1000));
|
||||
|
||||
net.add("source", src)
|
||||
.add("filter", filt)
|
||||
.add("slow", slow)
|
||||
.add("sink", snk)
|
||||
.connect("source", src.template output<"v">(), "filter", filt.template input<"v">())
|
||||
.connect("filter", filt.template output<"v">(), "slow", slow.template input<"v">())
|
||||
.connect("slow", slow.template output<"v">(), "sink", snk.template input<"v">())
|
||||
.build();
|
||||
|
||||
std::cout << "Running for 30 seconds — watch 'slow' become the bottleneck.\n"
|
||||
<< "Watchdog diagnostics print every 1 second.\n";
|
||||
#ifdef KPN_WEB_DEBUG
|
||||
net.set_web_debug_port(9090);
|
||||
std::cout << "Web debug UI: http://localhost:9090 (live graph, auto-updates every 500 ms)\n";
|
||||
#endif
|
||||
std::cout << '\n';
|
||||
|
||||
net.start();
|
||||
std::this_thread::sleep_for(std::chrono::seconds(30));
|
||||
net.stop();
|
||||
|
||||
std::cout << "\n=== Final diagnostics ===\n";
|
||||
net.print_diagnostics();
|
||||
}
|
||||
|
||||
@ -11,6 +11,9 @@ kpn_example(03_multi_output)
|
||||
kpn_example(04_storage_policy)
|
||||
kpn_example(05_error_handling)
|
||||
kpn_example(06_watchdog)
|
||||
if(KPN_WEB_DEBUG)
|
||||
kpn_target_enable_web_debug(06_watchdog)
|
||||
endif()
|
||||
# 07 and 08 require the Python bindings — only add if built
|
||||
if(KPN_BUILD_PYTHON)
|
||||
# These are Python scripts, not compiled targets — installed alongside kpn_python
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user