diff --git a/examples/15_node_error_handler/main.cpp b/examples/15_node_error_handler/main.cpp new file mode 100644 index 0000000..5cacad1 --- /dev/null +++ b/examples/15_node_error_handler/main.cpp @@ -0,0 +1,73 @@ +// Example 15 — Per-node Error Handler +// +// Demonstrates set_error_handler() for deciding whether a network can +// continue when a node throws an exception. +// +// The "validator" node rejects even numbers by throwing std::runtime_error. +// Its error handler logs the failure and returns true (skip & continue), +// so odd numbers still flow through to the sink. +// +// Compare: a second handler (commented below) returns false instead, +// which stops the node and gracefully shuts the downstream side down. +// +// Pipeline: [source] --int--> [validator] --int--> [sink] + +#include +#include +#include +#include + +static int counter = 0; + +static int source() { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + return ++counter; +} + +static int validate(int x) { + if (x % 2 == 0) + throw std::runtime_error("even number rejected: " + std::to_string(x)); + return x; +} + +static int received = 0; + +static void sink(int x) { + std::cout << " processed: " << x << '\n'; + ++received; +} + +int main() { + using namespace kpn; + + auto src = make_node (); + auto proc = make_node(); + auto snk = make_node (); + + // Return true → skip this invocation, keep the node running. + // Return false → stop the node (downstream drains then also stops). + proc.set_error_handler([](std::string_view name, std::exception_ptr ep) { + try { std::rethrow_exception(ep); } + catch (const std::exception& e) { + std::cerr << "[" << name << "] skipping item — " << e.what() << '\n'; + } + return true; + }); + + Network net; + net.add("source", src) + .add("validator", proc) + .add("sink", snk) + .connect("source", src.output<0>(), "validator", proc.input<0>()) + .connect("validator", proc.output<0>(), "sink", snk.input<0>()) + .build(); + + std::cout << "source emits 1..N; validator rejects even numbers.\n" + << "Error messages on stderr, accepted items on stdout.\n\n"; + + net.start(); + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + net.stop(); + + std::cout << "\nItems accepted by sink: " << received << '\n'; +} diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index b5cebc7..48e5c63 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -11,6 +11,7 @@ kpn_example(03_multi_output) kpn_example(04_storage_policy) kpn_example(05_error_handling) kpn_example(06_watchdog) +kpn_example(15_node_error_handler) kpn_example(10_static_hello_pipeline) kpn_example(11_static_fanout) if(KPN_WEB_DEBUG) diff --git a/include/kpn/node.hpp b/include/kpn/node.hpp index 4776375..6ad1567 100644 --- a/include/kpn/node.hpp +++ b/include/kpn/node.hpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -18,6 +19,10 @@ namespace kpn { +// Called when a node's function throws. Return true to skip the failed +// invocation and keep running, false to stop the node. +using NodeErrorHandler = std::function; + // ── INode — type-erased interface for Network / watchdog ───────────────────── struct INode { @@ -99,6 +104,8 @@ public: void set_name(std::string name) override { name_ = std::move(name); } + void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } + const NodeStats& stats() const override { return stats_; } NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { @@ -230,8 +237,13 @@ private: break; } catch (const ChannelOverflowError& e) { std::cerr << "[kpn] overflow: " << e.what() << "\n"; + } catch (...) { + if (error_handler_ && error_handler_(name_, std::current_exception())) + continue; + break; } } + stop_flag_.store(true, std::memory_order_relaxed); } // Pop all inputs into a tuple of argument values @@ -287,6 +299,7 @@ private: std::atomic stop_flag_{false}; std::jthread thread_; NodeStats stats_; + NodeErrorHandler error_handler_; }; // ── ObjectNode — wraps a callable object (functor / class with operator()) ──── @@ -357,6 +370,8 @@ public: void set_name(std::string name) override { name_ = std::move(name); } + void set_error_handler(NodeErrorHandler h) { error_handler_ = std::move(h); } + const NodeStats& stats() const override { return stats_; } NodeSnapshot node_snapshot(const std::string& name, double elapsed_s) const override { @@ -474,8 +489,13 @@ private: break; } catch (const ChannelOverflowError& e) { std::cerr << "[kpn] overflow: " << e.what() << "\n"; + } catch (...) { + if (error_handler_ && error_handler_(name_, std::current_exception())) + continue; + break; } } + stop_flag_.store(true, std::memory_order_relaxed); } template @@ -523,6 +543,7 @@ private: std::atomic stop_flag_{false}; std::jthread thread_; NodeStats stats_; + NodeErrorHandler error_handler_; }; // ── make_node overloads for callable objects ────────────────────────────────── diff --git a/tests/test_node.cpp b/tests/test_node.cpp index 516e5e0..d0bfaa3 100644 --- a/tests/test_node.cpp +++ b/tests/test_node.cpp @@ -2,6 +2,7 @@ #include #include #include +#include using namespace kpn; @@ -47,3 +48,104 @@ TEST_CASE("node stop unblocks cleanly", "[node]") { node.stop(); REQUIRE_FALSE(node.running()); } + +// ── Error handler tests ─────────────────────────────────────────────────────── + +namespace { + +struct SometimesThrower { + bool throw_next = true; + int operator()(int x) { + if (std::exchange(throw_next, false)) + throw std::runtime_error("deliberate skip"); + return x * 2; + } +}; + +struct AlwaysThrows { + int operator()(int) { throw std::runtime_error("always throws"); return 0; } +}; + +static int always_throws_fn(int) { + throw std::runtime_error("nttp always throws"); + return 0; +} + +} // namespace + +TEST_CASE("node stops cleanly on exception with no error handler", "[node][error_handler]") { + AlwaysThrows obj; + auto node = make_node(obj); + + node.start(); + node.input_channel<0>().push(1); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE_FALSE(node.running()); +} + +TEST_CASE("node continues when error handler returns true", "[node][error_handler]") { + SometimesThrower obj; + auto node = make_node(obj); + + Channel out_ch(5); + node.set_output_channel<0>(&out_ch); + + bool handler_called = false; + node.set_error_handler([&](std::string_view, std::exception_ptr) { + handler_called = true; + return true; + }); + + node.start(); + node.input_channel<0>().push(0); // throws → skipped, no output + node.input_channel<0>().push(21); // succeeds → 42 + int result = out_ch.pop(); // blocking — waits for the second item + node.stop(); + + REQUIRE(handler_called); + REQUIRE(result == 42); +} + +TEST_CASE("node stops when error handler returns false", "[node][error_handler]") { + AlwaysThrows obj; + auto node = make_node(obj); + + node.set_error_handler([](std::string_view, std::exception_ptr) { return false; }); + + node.start(); + node.input_channel<0>().push(1); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE_FALSE(node.running()); +} + +TEST_CASE("error handler receives node name and exception", "[node][error_handler]") { + AlwaysThrows obj; + auto node = make_node(obj); + node.set_name("test_node"); + + std::string captured_name; + std::string captured_msg; + node.set_error_handler([&](std::string_view name, std::exception_ptr ep) { + captured_name = name; + try { std::rethrow_exception(ep); } + catch (const std::exception& e) { captured_msg = e.what(); } + return false; + }); + + node.start(); + node.input_channel<0>().push(1); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + node.stop(); + + REQUIRE(captured_name == "test_node"); + REQUIRE(captured_msg == "always throws"); +} + +TEST_CASE("Node stops cleanly on exception with no error handler", "[node][error_handler]") { + auto node = make_node(); + + node.start(); + node.input_channel<0>().push(1); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + REQUIRE_FALSE(node.running()); +}