infra: fix loop exception handling in client and server context pool

silkworm/infra/concurrency/context_pool.hpp

@@ -26,6 +26,7 @@
 
 #include <boost/asio/io_context.hpp>
 
+#include <silkworm/infra/common/ensure.hpp>
 #include <silkworm/infra/common/log.hpp>
 #include <silkworm/infra/concurrency/context_pool_settings.hpp>
 #include <silkworm/infra/concurrency/idle_strategy.hpp>
@@ -75,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");
         }
@@ -119,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();
             });
@@ -167,6 +173,7 @@ class ContextPool {
 
     //! Use a round-robin scheme to choose the next context to use
     T& next_context() {
+        ensure(contexts_.size() > 0, "ContextPool: no context in pool");
         // Increment the next index first to make sure that different calling threads get different contexts.
         size_t index = next_index_.fetch_add(1) % contexts_.size();
         return contexts_[index];
@@ -177,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_;
 
@@ -189,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.cpp

@@ -16,6 +16,7 @@
 
 #include "client_context_pool.hpp"
 
+#include <exception>
 #include <thread>
 
 #include <magic_enum.hpp>
@@ -78,11 +79,20 @@ void ClientContext::execute_loop_multi_threaded() {
     std::thread grpc_context_thread{[grpc_context = grpc_context_]() {
         grpc_context->run_completion_queue();
     }};
-    io_context_->run();
+    std::exception_ptr run_exception;
+    try {
+        io_context_->run();
+    } catch (...) {
+        run_exception = std::current_exception();
+    }
 
     grpc_context_work_.reset();
     grpc_context_->stop();
     grpc_context_thread.join();
+
+    if (run_exception) {
+        std::rethrow_exception(run_exception);
+    }
     SILK_DEBUG << "Multi-thread execution loop end [" << std::this_thread::get_id() << "]";
 }
 

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,6 +215,22 @@ TEST_CASE("ClientContextPool: cannot restart context pool", "[silkworm][infra][g
     }
 }
 
+TEST_CASE("ClientContextPool: handle loop exception", "[silkworm][infra][grpc][client][client_context]") {
+    test_util::SetLogVerbosityGuard guard{log::Level::kNone};
+
+    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(bool(run_exception));
+}
+
 #endif  // SILKWORM_SANITIZE
 
 }  // namespace silkworm::rpc

silkworm/infra/grpc/server/server_context_pool.cpp

@@ -16,6 +16,7 @@
 
 #include "server_context_pool.hpp"
 
+#include <exception>
 #include <thread>
 #include <utility>
 
@@ -75,11 +76,20 @@ void ServerContext::execute_loop_backoff() {
         client_grpc_context_->run_completion_queue();
         SILK_DEBUG << "Client GrpcContext execution loop end [" << std::this_thread::get_id() << "]";
     }};
-    io_context()->run();
+    std::exception_ptr run_exception;
+    try {
+        io_context()->run();
+    } catch (...) {
+        run_exception = std::current_exception();
+    }
 
     client_grpc_context_work_.reset();
     client_grpc_context_->stop();
     client_grpc_context_thread.join();
+
+    if (run_exception) {
+        std::rethrow_exception(run_exception);
+    }
     SILK_DEBUG << "Back-off execution loop end [" << std::this_thread::get_id() << "]";
 }
 
@@ -109,14 +119,23 @@ void ServerContext::execute_loop_multi_threaded() {
         client_grpc_context_->run_completion_queue();
         SILK_TRACE << "Client GrpcContext execution loop end [" << std::this_thread::get_id() << "]";
     }};
-    io_context()->run();
+    std::exception_ptr run_exception;
+    try {
+        io_context()->run();
+    } catch (...) {
+        run_exception = std::current_exception();
+    }
 
     server_grpc_context_work_.reset();
     client_grpc_context_work_.reset();
     server_grpc_context_->stop();
     client_grpc_context_->stop();
     server_grpc_context_thread.join();
     client_grpc_context_thread.join();
+
+    if (run_exception) {
+        std::rethrow_exception(run_exception);
+    }
     SILK_TRACE << "Multi-thread execution loop end [" << std::this_thread::get_id() << "]";
 }
 

silkworm/infra/grpc/server/server_context_pool_test.cpp

@@ -102,6 +102,7 @@ TEST_CASE("ServerContextPool", "[silkworm][infra][grpc][server][server_context]"
     SECTION("next_context") {
         ServerContextPool server_context_pool{2};
         REQUIRE(server_context_pool.num_contexts() == 0);
+        CHECK_THROWS_AS(server_context_pool.next_context(), std::logic_error);
         auto queue_ptr1 = builder.AddCompletionQueue();
         auto queue_raw_ptr1 = queue_ptr1.get();
         auto queue_ptr2 = builder.AddCompletionQueue();
@@ -120,6 +121,7 @@ TEST_CASE("ServerContextPool", "[silkworm][infra][grpc][server][server_context]"
     SECTION("next_io_context") {
         ServerContextPool server_context_pool{2};
         REQUIRE(server_context_pool.num_contexts() == 0);
+        CHECK_THROWS_AS(server_context_pool.next_io_context(), std::logic_error);
         server_context_pool.add_context(builder.AddCompletionQueue(), WaitMode::blocking);
         server_context_pool.add_context(builder.AddCompletionQueue(), WaitMode::blocking);
         CHECK(server_context_pool.num_contexts() == 2);
@@ -163,6 +165,26 @@ TEST_CASE("ServerContextPool", "[silkworm][infra][grpc][server][server_context]"
         CHECK_NOTHROW(server_context_pool.join());
     }
 }
+
+TEST_CASE("ServerContextPool: handle loop exception", "[silkworm][infra][grpc][client][client_context]") {
+    test_util::SetLogVerbosityGuard guard{log::Level::kNone};
+    grpc::ServerBuilder builder;
+
+    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(bool(run_exception));
+}
 #endif  // SILKWORM_SANITIZE
 
 }  // namespace silkworm::rpc