From 4b6e498ba7e70a34cc0b57638f9e56a43b7f41ae Mon Sep 17 00:00:00 2001 From: Duncan Tourolle Date: Sun, 19 Jul 2026 16:55:36 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20persistent-pipeline=20reuse=20=E2=80=94?= =?UTF-8?q?=20push=5Fblocking,=20node=20introspection,=20stateful=20wrappe?= =?UTF-8?q?r?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds three pieces needed to build one KPN network and reuse it across many replays/configs instead of tearing down and rebuilding per run: - Channel::push_blocking (+ IVariantChannel/VariantChannel forwarding): lossless backpressure push that waits for space instead of dropping when the ring is full. PyNode's run_loop now uses it so a downstream consumer lagging behind never silently drops a frame. - PyNetwork::node_ptr / node_stats: raw node handle by name (for a binding to dynamic_cast to a concrete wrapper and call functor-specific runtime setters) and a per-node timing snapshot for profiling. - ObjectVariantNodeWrapper: variant-node adapter for functors that need runtime-constructed state (a Config, a loaded gallery), mirroring VariantNodeWrapper's channel plumbing but backed by ObjectNode. Built and used downstream in scene-actor-extraction's sae_kpn Python replay bindings for repeated threshold-sweep evaluation of the same pipeline. --- include/kpn/channel.hpp | 29 ++++ include/kpn/python/bindings.hpp | 25 +++- include/kpn/python/object_variant_node.hpp | 149 +++++++++++++++++++++ include/kpn/variant_node.hpp | 5 + 4 files changed, 205 insertions(+), 3 deletions(-) create mode 100644 include/kpn/python/object_variant_node.hpp diff --git a/include/kpn/channel.hpp b/include/kpn/channel.hpp index 99ebe51..f8842d4 100644 --- a/include/kpn/channel.hpp +++ b/include/kpn/channel.hpp @@ -136,6 +136,35 @@ public: push_callback_(); } + // Lossless push with BACKPRESSURE: if the ring is full, wait for the consumer to + // drain instead of dropping (the throwing push()) — the producer just runs slower. + // Use when every value must be delivered (e.g. replaying a dump for scoring, where + // a dropped frame silently corrupts the result). SPSC: only the sole producer may + // call it. Returns false if the channel was disabled while waiting. + bool push_blocking(T value) { + for (;;) { + if (!accepting_.load(std::memory_order_acquire)) { + stats_.record_drop(); + return false; + } + const std::size_t t = tail_.load(std::memory_order_relaxed); + const std::size_t h = head_.load(std::memory_order_acquire); + if (t - h < capacity_) { // space available → normal push + const std::size_t data_bytes = ChannelDataSize::bytes(value); + const bool was_empty = (t == h); + buf_[t & ring_mask_] = make_storage(std::move(value)); + tail_.store(t + 1, std::memory_order_release); + stats_.record_push(t - h + 1, data_bytes); + wake_.fetch_add(1, std::memory_order_release); + wake_.notify_one(); + if (was_empty && push_callback_) push_callback_(); + return true; + } + // full: yield briefly and retry (consumer will drain) + std::this_thread::sleep_for(std::chrono::microseconds(50)); + } + } + // Lossless, non-blocking delivery for a must-deliver control token (EOF). // // A sentinel is stored out-of-band — in a dedicated slot that does NOT diff --git a/include/kpn/python/bindings.hpp b/include/kpn/python/bindings.hpp index 0a31bfe..2491bf8 100644 --- a/include/kpn/python/bindings.hpp +++ b/include/kpn/python/bindings.hpp @@ -217,6 +217,20 @@ public: return it->second; } + // Raw node handle by name — lets a binding dynamic_cast to a concrete wrapper + // type and call its functor's runtime setters (persistent-pipeline reuse). + VNode* node_ptr(const std::string& name) { return &node_at(name); } + + // Per-node timing snapshot for profiling where a replay spends its time. + std::map node_stats(const std::string& name) { + auto& n = node_at(name); + NodeSnapshot s = n.node_snapshot(name, 0.0); + return {{"frames", double(s.frames_processed)}, + {"exec_ms", s.ema_exec_ms}, {"max_ms", s.max_exec_ms}, + {"blocked_ms", s.total_blocked_ms}, {"fps", s.throughput_fps}, + {"cpu_ms", s.total_cpu_ms}, {"cpu_util_pct", s.cpu_util_pct}}; + } + private: VNode& node_at(const std::string& name) { auto it = nodes_.find(name); @@ -422,13 +436,17 @@ private: for (std::size_t i = 0; i < out_channels_.size(); ++i) { if (out_channels_[i]) - out_channels_[i]->push(std::move(outputs[i])); + // Lossless: wait for space rather than drop. A dropped frame + // silently corrupts a replay's score; backpressure just slows + // the producer. (Was push() + "drop on overflow".) + out_channels_[i]->push_blocking(std::move(outputs[i])); } } catch (const ChannelClosedError&) { break; } catch (const ChannelOverflowError&) { - // drop and continue + // no longer reachable with push_blocking, kept for safety + break; } } } @@ -530,7 +548,8 @@ void register_py_network(nb::module_& m, const char* class_name = "Network") { .def("read", &Net::read, nb::arg("node"), nb::arg("out_idx") = std::size_t(0)) .def("write", &Net::write, - nb::arg("node"), nb::arg("in_idx"), nb::arg("value")); + nb::arg("node"), nb::arg("in_idx"), nb::arg("value")) + .def("node_stats", &Net::node_stats, nb::arg("node")); } } // namespace kpn::python diff --git a/include/kpn/python/object_variant_node.hpp b/include/kpn/python/object_variant_node.hpp new file mode 100644 index 0000000..82ff149 --- /dev/null +++ b/include/kpn/python/object_variant_node.hpp @@ -0,0 +1,149 @@ +#pragma once +// ObjectVariantNodeWrapper — variant-node adapter for *stateful* functors. +// +// VariantNodeWrapper (variant_node.hpp) wraps Node, where Func is a +// default-constructible NTTP callable. That doesn't fit nodes whose functor must +// be constructed with runtime state (a Config, a loaded gallery, etc.) — those use +// ObjectNode, which takes `Obj& obj` at construction. +// +// This wrapper owns an Obj instance and exposes the same IVariantNode surface so a +// stateful C++ node can live inside a PyNetwork. Build one via a factory that +// constructs the functor from Python-supplied config, e.g.: +// +// auto n = std::make_shared, out<"matched">>>( +// fifo_cap, gallery, cfg); // Obj ctor args forwarded +// net.add("identity_matcher", n); +// +// The wrapper mirrors VariantNodeWrapper's channel plumbing exactly; only the +// underlying node type (PoolObjectNode, holding Obj&) differs. + +#include "../channel.hpp" +#include "../node.hpp" +#include "../variant_node.hpp" + +#include +#include +#include +#include +#include +#include +#include + +namespace kpn { + +template, + typename OutputTag = out<>> +class ObjectVariantNodeWrapper; + +template +class ObjectVariantNodeWrapper, out> + : public IVariantNode +{ + using NodeT = ObjectNode, out>; + +public: + using args_tuple = typename NodeT::args_tuple; + using return_tuple = typename NodeT::return_tuple; + + static constexpr std::size_t n_in = NodeT::input_count; + static constexpr std::size_t n_out = NodeT::output_count; + + // Owns the functor; forwards remaining args to Obj's constructor. + template + explicit ObjectVariantNodeWrapper(std::size_t fifo_capacity, ObjArgs&&... obj_args) + : obj_(std::forward(obj_args)...) + , node_(obj_, fifo_capacity) + , in_channels_(n_in) + , out_channels_(n_out) + , out_type_indices_(n_out, std::type_index(typeid(void))) + { + init_inputs(std::make_index_sequence{}, fifo_capacity); + init_out_types(std::make_index_sequence{}); + } + + // Access the owned functor so callers can invoke its runtime setters (e.g. to + // change a threshold on a persistent pipeline without rebuilding the node). + Obj& functor() { return obj_; } + + // ── INode ───────────────────────────────────────────────────────────────── + void start() override { node_.start(); } + void stop() override { node_.stop(); } + bool running() const override { return node_.running(); } + const NodeStats& stats() const override { return node_.stats(); } + void set_name(std::string name) override { node_.set_name(std::move(name)); } + NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { + return node_.node_snapshot(name, elapsed_s); + } + + // ── IVariantNode ────────────────────────────────────────────────────────── + std::size_t input_count() const override { return n_in; } + std::size_t output_count() const override { return n_out; } + + std::type_index input_type(std::size_t i) const override { + return in_channels_[i]->type_index(); + } + std::type_index output_type(std::size_t i) const override { + return out_type_indices_[i]; + } + + std::shared_ptr> input_channel(std::size_t i) override { + return in_channels_[i]; + } + + void set_output_channel(std::size_t i, + std::shared_ptr> ch) override { + set_output_impl(i, std::move(ch), std::make_index_sequence{}); + } + +private: + template + void init_inputs(std::index_sequence, std::size_t cap) { + ((init_one_input(cap)), ...); + } + + template + void init_one_input(std::size_t cap) { + using T = std::tuple_element_t; + auto shared_ch = std::make_shared>(cap); + node_.template set_input_channel(shared_ch); + in_channels_[I] = std::make_shared>(std::move(shared_ch)); + } + + template + void init_out_types(std::index_sequence) { + ((out_type_indices_[Is] = + std::type_index(typeid(std::tuple_element_t))), ...); + } + + template + void set_output_impl(std::size_t port, + std::shared_ptr> ch, + std::index_sequence) { + bool matched = false; + ((Is == port && (set_output_at(std::move(ch)), matched = true)), ...); + if (!matched) + throw std::out_of_range("set_output_channel: port index out of range"); + } + + template + void set_output_at(std::shared_ptr> ch) { + using T = std::tuple_element_t; + auto* typed = dynamic_cast*>(ch.get()); + if (!typed) + throw std::runtime_error( + "set_output_channel: type mismatch at output port " + std::to_string(I)); + node_.template set_output_channel(typed->raw_ptr()); + out_channels_[I] = std::move(ch); + } + + Obj obj_; // owned; node_ holds Obj& — declaration order keeps obj_ alive first + NodeT node_; + std::vector>> in_channels_; + std::vector>> out_channels_; + std::vector out_type_indices_; +}; + +} // namespace kpn diff --git a/include/kpn/variant_node.hpp b/include/kpn/variant_node.hpp index 0af32e7..9a98599 100644 --- a/include/kpn/variant_node.hpp +++ b/include/kpn/variant_node.hpp @@ -55,6 +55,8 @@ class IVariantChannel { public: virtual ~IVariantChannel() = default; virtual void push(Variant v) = 0; + // Lossless push with backpressure (waits instead of dropping when full). + virtual void push_blocking(Variant v) = 0; virtual Variant pop() = 0; virtual std::type_index type_index() const = 0; virtual std::string type_name() const = 0; @@ -76,6 +78,9 @@ public: void push(Variant v) override { channel_->push(std::get(std::move(v))); } + void push_blocking(Variant v) override { + channel_->push_blocking(std::get(std::move(v))); + } Variant pop() override { return Variant{ channel_->pop() }; }