improve exception handling and testability

silkworm/infra/concurrency/context_pool.hpp

@@ -76,8 +76,10 @@ std::ostream& operator<<(std::ostream& out, const Context& c);
 //! Pool of \ref Context instances running as separate reactive schedulers.
 template <typename T = Context>
 class ContextPool {
+    using ExceptionHandler = std::function<void(std::exception_ptr)>;
+
   public:
-    explicit ContextPool(std::size_t pool_size) : next_index_{0} {
+    explicit ContextPool(std::size_t pool_size) : next_index_{0}, exception_handler_{termination_handler} {
         if (pool_size == 0) {
             throw std::logic_error("ContextPool::ContextPool pool_size is 0");
         }
@@ -120,7 +122,10 @@ class ContextPool {
                     context.execute_loop();
                 } catch (const std::exception& ex) {
                     SILK_CRIT << "ContextPool context.execute_loop exception: " << ex.what();
-                    std::terminate();
+                    exception_handler_(std::make_exception_ptr(ex));
+                } catch (...) {
+                    SILK_CRIT << "ContextPool context.execute_loop unexpected exception";
+                    exception_handler_(std::current_exception());
                 }
                 SILK_TRACE << "Thread end context[" << i << "] thread_id: " << std::this_thread::get_id();
             });
@@ -179,7 +184,15 @@ class ContextPool {
         return *context.io_context();
     }
 
+    void set_exception_handler(ExceptionHandler exception_handler) {
+        exception_handler_ = exception_handler;
+    }
+
   protected:
+    static void termination_handler(std::exception_ptr) {
+        std::terminate();
+    }
+
     //! The pool of execution contexts.
     std::vector<T> contexts_;
 
@@ -191,6 +204,9 @@ class ContextPool {
 
     //! Flag indicating if pool has been stopped.
     std::atomic_bool stopped_{false};
+
+    //! Exception handler invoked on execution loop abnormal termination
+    ExceptionHandler exception_handler_;
 };
 
 }  // namespace silkworm::concurrency

silkworm/infra/grpc/client/client_context_pool_test.cpp

@@ -17,6 +17,8 @@
 #include "client_context_pool.hpp"
 
 #include <atomic>
+#include <cstring>
+#include <exception>
 #include <stdexcept>
 #include <string>
 #include <thread>
@@ -213,37 +215,20 @@ TEST_CASE("ClientContextPool: cannot restart context pool", "[silkworm][infra][g
     }
 }
 
-// This custom ClientContextPool is required to test how unhandled exceptions are treated by ClientContextPool
-// because in such case ClientContextPool terminates execution using std::terminate, which is not suitable for tests
-class ClientContextPool_ForTest : public ClientContextPool {
-  public:
-    explicit ClientContextPool_ForTest(std::size_t pool_size) : ClientContextPool(pool_size) {}
-    void start() override {
-        for (std::size_t i{0}; i < contexts_.size(); ++i) {
-            auto& context = contexts_[i];
-            context_threads_.create_thread([&]() {
-                try {
-                    context.execute_loop();
-                } catch (const std::exception& ex) {
-                    loop_exception_caught = true;
-                    // In case of any loop exception in any thread, close down the pool
-                    stop();
-                }
-            });
-        }
-    }
-
-    bool loop_exception_caught{false};
-};
-
 TEST_CASE("ClientContextPool: handle loop exception", "[silkworm][infra][grpc][client][client_context]") {
     test_util::SetLogVerbosityGuard guard{log::Level::kNone};
 
-    ClientContextPool_ForTest cp{3};
+    ClientContextPool cp{3};
+    std::exception_ptr run_exception;
+    cp.set_exception_handler([&](std::exception_ptr eptr) {
+        run_exception = eptr;
+        // In case of any loop exception in any thread, close down the pool
+        cp.stop();
+    });
     auto context_pool_thread = std::thread([&]() { cp.run(); });
     boost::asio::post(cp.next_io_context(), [&]() { throw std::logic_error{"unexpected"}; });
     CHECK_NOTHROW(context_pool_thread.join());
-    CHECK(cp.loop_exception_caught);
+    CHECK(bool(run_exception));
 }
 
 #endif  // SILKWORM_SANITIZE

silkworm/infra/grpc/server/server_context_pool_test.cpp

@@ -166,41 +166,24 @@ TEST_CASE("ServerContextPool", "[silkworm][infra][grpc][server][server_context]"
     }
 }
 
-// This custom ServerContextPool is required to test how unhandled exceptions are treated by ServerContextPool
-// because in such case ServerContextPool terminates execution using std::terminate, which is not suitable for tests
-class ServerContextPool_ForTest : public ServerContextPool {
-  public:
-    explicit ServerContextPool_ForTest(std::size_t pool_size) : ServerContextPool(pool_size) {}
-    void start() override {
-        for (std::size_t i{0}; i < contexts_.size(); ++i) {
-            auto& context = contexts_[i];
-            context_threads_.create_thread([&]() {
-                try {
-                    context.execute_loop();
-                } catch (const std::exception& ex) {
-                    loop_exception_caught = true;
-                    // In case of any loop exception in any thread, close down the pool
-                    stop();
-                }
-            });
-        }
-    }
-
-    bool loop_exception_caught{false};
-};
-
 TEST_CASE("ServerContextPool: handle loop exception", "[silkworm][infra][grpc][client][client_context]") {
     test_util::SetLogVerbosityGuard guard{log::Level::kNone};
     grpc::ServerBuilder builder;
 
-    ServerContextPool_ForTest cp{3};
+    ServerContextPool cp{3};
     cp.add_context(builder.AddCompletionQueue(), WaitMode::blocking);
     cp.add_context(builder.AddCompletionQueue(), WaitMode::blocking);
     cp.add_context(builder.AddCompletionQueue(), WaitMode::blocking);
+    std::exception_ptr run_exception;
+    cp.set_exception_handler([&](std::exception_ptr eptr) {
+        run_exception = eptr;
+        // In case of any loop exception in any thread, close down the pool
+        cp.stop();
+    });
     auto context_pool_thread = std::thread([&]() { cp.run(); });
     boost::asio::post(cp.next_io_context(), [&]() { throw std::logic_error{"unexpected"}; });
     CHECK_NOTHROW(context_pool_thread.join());
-    CHECK(cp.loop_exception_caught);
+    CHECK(bool(run_exception));
 }
 #endif  // SILKWORM_SANITIZE