diff --git a/external/KPN b/external/KPN index 7c6a8be..949c813 160000 --- a/external/KPN +++ b/external/KPN @@ -1 +1 @@ -Subproject commit 7c6a8be2b7ce8f5744259ad326059a04ddf86920 +Subproject commit 949c8134efb38030b47387044f5929e64745be39 diff --git a/src/main.cpp b/src/main.cpp index 409731c..af9fca0 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -48,8 +48,11 @@ #include #include #include +#include +#include #include #include +#include #include // ── CLI parsing ─────────────────────────────────────────────────────────────── @@ -192,17 +195,53 @@ int main(int argc, char** argv) { ); #endif + // ── Pipeline observability (KPN event handler) ──────────────────────────── + // Tally dropped frames per node (channel overflow) and detect a node that + // stops unexpectedly. A Closed event from any node other than result_sink at + // EOF means a stage crashed — without this the main loop below would hang on + // `done` forever, so we trip a flag to unblock it and exit non-zero. + std::mutex event_mtx; + std::map overflow_counts; + std::atomic node_crashed{false}; + + net.set_event_handler( + [&](std::string_view node_name, kpn::NodeEvent ev, + std::chrono::steady_clock::time_point) { + if (ev == kpn::NodeEvent::Overflow) { + std::lock_guard lk(event_mtx); + ++overflow_counts[std::string(node_name)]; + } else { // NodeEvent::Closed + // result_sink closing once EOF has been signalled is the normal + // shutdown path, not a crash. + if (node_name == "result_sink" && done.load(std::memory_order_acquire)) + return; + std::cerr << "[main] node '" << node_name + << "' stopped unexpectedly — aborting pipeline\n"; + node_crashed.store(true, std::memory_order_release); + } + }); + // ── Run ─────────────────────────────────────────────────────────────────── - // Note: StaticNetwork does not expose set_error_handler; node exceptions - // are printed to stderr by the KPN run_loop and the network continues. std::cerr << "[main] starting pipeline…\n"; net.start(); - // Main thread waits until ResultSinkFunc signals EOF completion - while (!done.load(std::memory_order_acquire)) + // Main thread waits until ResultSinkFunc signals EOF completion, or a node + // crash trips node_crashed. + while (!done.load(std::memory_order_acquire) && + !node_crashed.load(std::memory_order_acquire)) std::this_thread::sleep_for(std::chrono::milliseconds(100)); net.stop(); net.print_diagnostics(); - return 0; + + { + std::lock_guard lk(event_mtx); + if (!overflow_counts.empty()) { + std::cerr << "[main] dropped frames (channel overflow):\n"; + for (const auto& [name, count] : overflow_counts) + std::cerr << " " << name << ": " << count << "\n"; + } + } + + return node_crashed.load(std::memory_order_acquire) ? 1 : 0; } diff --git a/src/scene_preview.cpp b/src/scene_preview.cpp index c313570..7019854 100644 --- a/src/scene_preview.cpp +++ b/src/scene_preview.cpp @@ -35,8 +35,11 @@ #include #include #include +#include +#include #include #include +#include #include static Config parse_args(int argc, char** argv) { @@ -143,12 +146,36 @@ int main(int argc, char** argv) { kpn::edge(tracker.output<"annotation">(), sink.input<"annotation">()) ); + // ── Pipeline observability (KPN event handler) ──────────────────────────── + // Tally dropped frames per node (channel overflow) and detect a node that + // stops unexpectedly so the preview loop below can bail out instead of + // spinning on a dead pipeline. + std::mutex event_mtx; + std::map overflow_counts; + std::atomic node_crashed{false}; + + net.set_event_handler( + [&](std::string_view node_name, kpn::NodeEvent ev, + std::chrono::steady_clock::time_point) { + if (ev == kpn::NodeEvent::Overflow) { + std::lock_guard lk(event_mtx); + ++overflow_counts[std::string(node_name)]; + } else { // NodeEvent::Closed + if (node_name == "result_sink" && done.load(std::memory_order_acquire)) + return; + std::cerr << "[main] node '" << node_name + << "' stopped unexpectedly — aborting pipeline\n"; + node_crashed.store(true, std::memory_order_release); + } + }); + // ── Run ─────────────────────────────────────────────────────────────────── std::cerr << "[main] starting pipeline — press q or Esc to quit early\n"; net.start(); - // Main thread drives the display window; returns false on EOF or q/Esc - while (preview.step()) { + // Main thread drives the display window; returns false on EOF or q/Esc. + // Bail out early if a node crashes. + while (!node_crashed.load(std::memory_order_acquire) && preview.step()) { cv::waitKey(1); // pump OS events between frames } @@ -156,5 +183,15 @@ int main(int argc, char** argv) { sink_fn.flush(); // write whatever was accumulated (no-op if EOF already flushed) cv::destroyAllWindows(); net.print_diagnostics(); - return 0; + + { + std::lock_guard lk(event_mtx); + if (!overflow_counts.empty()) { + std::cerr << "[main] dropped frames (channel overflow):\n"; + for (const auto& [name, count] : overflow_counts) + std::cerr << " " << name << ": " << count << "\n"; + } + } + + return node_crashed.load(std::memory_order_acquire) ? 1 : 0; }