feat(pipeline): detect node crashes and tally dropped frames
Register a KPN event handler in both scene_analyze and scene_preview:
- Overflow events accumulate per-node dropped-frame counts, printed on exit.
- A Closed event from any node other than result_sink at EOF means a stage
died; trip an atomic so the main loop bails out instead of hanging on
'done' forever, and exit non-zero.
Bumps external/KPN to the commit that exposes set_event_handler / NodeEvent.
This commit is contained in:
+44
-5
@@ -48,8 +48,11 @@
|
||||
#include <chrono>
|
||||
#include <cstring>
|
||||
#include <iostream>
|
||||
#include <map>
|
||||
#include <mutex>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <string_view>
|
||||
#include <thread>
|
||||
|
||||
// ── 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<std::string, long> overflow_counts;
|
||||
std::atomic<bool> 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<std::mutex> 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<std::mutex> 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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user