Add exception handling in task_worker_loop to prevent worker thread termination (#7658)

Copilot · eddyashton · achamayou · web-flow · commit 4e02b8b35497 · 2026-03-06T11:27:22.000Z
Co-authored-by: copilot-swe-agent[bot] &lt;198982749+Copilot@users.noreply.github.com&gt;
Co-authored-by: eddyashton &lt;6000239+eddyashton@users.noreply.github.com&gt;
Co-authored-by: Amaury Chamayou &lt;amchamay@microsoft.com&gt;
Co-authored-by: Eddy Ashton &lt;edashton@microsoft.com&gt;
Co-authored-by: achamayou &lt;4016369+achamayou@users.noreply.github.com&gt;
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -44,6 +44,12 @@ include(GNUInstallDirs)
 # Use fixed name instead of absolute path for reproducible builds
 add_compile_options("-ffile-prefix-map=${CCF_DIR}=CCF")
 
+# In Debug builds, export symbols to the dynamic symbol table so that
+# backtrace_symbols can resolve function names in stacktraces, and preserve
+# frame pointers so that backtrace() can walk the full call stack.
+add_link_options($<$<CONFIG:Debug>:-rdynamic>)
+add_compile_options($<$<CONFIG:Debug>:-fno-omit-frame-pointer>)
+
 set(CMAKE_MODULE_PATH "${CCF_DIR}/cmake;${CMAKE_MODULE_PATH}")
 
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
@@ -275,7 +281,9 @@ add_ccf_static_library(
        ${CCF_DIR}/src/tasks/ordered_tasks.cpp
        ${CCF_DIR}/src/tasks/fan_in_tasks.cpp
        ${CCF_DIR}/src/tasks/thread_manager.cpp
+       ${CCF_DIR}/src/tasks/worker.cpp
 )
+target_link_libraries(ccf_tasks PRIVATE ${CMAKE_DL_LIBS})
 
 # Common test args for Python scripts starting up CCF networks
 set(WORKER_THREADS
diff --git a/src/tasks/test/basic_tasks.cpp b/src/tasks/test/basic_tasks.cpp
@@ -3,6 +3,7 @@
 
 #include "tasks/basic_task.h"
 #include "tasks/task_system.h"
+#include "tasks/worker.h"
 
 #include <doctest/doctest.h>
 #include <iostream>
@@ -243,3 +244,173 @@ TEST_CASE("Scheduling" * doctest::test_suite("basic_tasks"))
   std::iota(target.begin(), target.end(), 0);
   REQUIRE(count_with_me == target);
 }
+
+// Helper functions at namespace scope to ensure external linkage, so that
+// backtrace_symbols can resolve their names with -rdynamic.
+namespace exception_handling_test
+{
+  void level_3_throws_runtime_error()
+  {
+    throw std::runtime_error("Test exception");
+  }
+
+  void level_2_calls_level_3()
+  {
+    level_3_throws_runtime_error();
+  }
+
+  void level_1_calls_level_2()
+  {
+    level_2_calls_level_3();
+  }
+
+  void level_3_throws_int()
+  {
+    throw 42;
+  }
+
+  void level_2_calls_level_3_int()
+  {
+    level_3_throws_int();
+  }
+
+  void level_1_calls_level_2_int()
+  {
+    level_2_calls_level_3_int();
+  }
+
+  struct ThrowsException : public ccf::tasks::BaseTask
+  {
+    void do_task_implementation() override
+    {
+      level_1_calls_level_2();
+    }
+
+    const std::string& get_name() const override
+    {
+      static const std::string name = "ThrowsException";
+      return name;
+    }
+  };
+
+  struct ThrowsUnknown : public ccf::tasks::BaseTask
+  {
+    void do_task_implementation() override
+    {
+      level_1_calls_level_2_int();
+    }
+
+    const std::string& get_name() const override
+    {
+      static const std::string name = "ThrowsUnknown";
+      return name;
+    }
+  };
+}
+
+TEST_CASE("Exception handling" * doctest::test_suite("basic_tasks"))
+{
+  // Custom logger that captures log messages for assertion
+  struct CapturingLogger : public ccf::logger::AbstractLogger
+  {
+    std::mutex mutex;
+    std::vector<std::string> messages;
+
+    void write(const ccf::logger::LogLine& ll) override
+    {
+      std::lock_guard<std::mutex> lock(mutex);
+      messages.push_back(ll.msg);
+    }
+
+    bool contains(const std::string& substring)
+    {
+      std::lock_guard<std::mutex> lock(mutex);
+      for (const auto& m : messages)
+      {
+        if (m.find(substring) != std::string::npos)
+        {
+          return true;
+        }
+      }
+      return false;
+    }
+
+    void clear()
+    {
+      std::lock_guard<std::mutex> lock(mutex);
+      messages.clear();
+    }
+  };
+
+  auto capturing_logger = std::make_unique<CapturingLogger>();
+  auto* logger_ptr = capturing_logger.get();
+  ccf::logger::config::loggers().push_back(std::move(capturing_logger));
+
+  // Task that runs successfully after exceptions
+  std::atomic<bool> success_task_ran = false;
+  ccf::tasks::Task success_task = ccf::tasks::make_basic_task(
+    [&success_task_ran]() { success_task_ran.store(true); }, "SuccessTask");
+
+  ccf::tasks::JobBoard job_board;
+  std::atomic<bool> stop_signal = false;
+
+  // Queue tasks: two that throw, then one that should still run
+  job_board.add_task(
+    std::make_shared<exception_handling_test::ThrowsException>());
+  job_board.add_task(
+    std::make_shared<exception_handling_test::ThrowsUnknown>());
+  job_board.add_task(success_task);
+
+  std::thread worker([&]() {
+    ccf::tasks::task_worker_loop(
+      job_board, stop_signal, /*abort_on_throw=*/false);
+  });
+
+  // Wait for the success task to run
+  const auto wait_step = std::chrono::milliseconds(10);
+  const auto max_wait = std::chrono::seconds(5);
+  auto waited = std::chrono::milliseconds(0);
+  while (!success_task_ran.load() && waited < max_wait)
+  {
+    std::this_thread::sleep_for(wait_step);
+    waited += wait_step;
+  }
+
+  stop_signal.store(true);
+  worker.join();
+
+  // With CCF_TASK_EXCEPTION_NO_ABORT, the worker loop continues after
+  // exceptions, so the success task should have run
+  REQUIRE(success_task_ran.load());
+
+  // Verify that fatal messages were logged for both exception types
+  REQUIRE(logger_ptr->contains(
+    "ThrowsException task failed with exception: Test exception"));
+  REQUIRE(
+    logger_ptr->contains("ThrowsUnknown task failed with unknown exception"));
+
+  // Verify that stack traces contain demangled function names from the
+  // known call chains. These functions have external linkage and are
+  // exported to the dynamic symbol table via -rdynamic in Debug builds.
+  // Note: very small leaf functions (e.g. level_3_throws_int, which is
+  // just `throw 42;`) may be inlined by the compiler, so we only assert
+  // on the caller frames that reliably appear.
+
+  // ThrowsException call chain
+  REQUIRE(logger_ptr->contains("level_3_throws_runtime_error"));
+  REQUIRE(logger_ptr->contains("level_2_calls_level_3()"));
+  REQUIRE(logger_ptr->contains("level_1_calls_level_2()"));
+
+  // ThrowsUnknown call chain
+  REQUIRE(logger_ptr->contains("level_2_calls_level_3_int"));
+  REQUIRE(logger_ptr->contains("level_1_calls_level_2_int"));
+
+  // Clean up: remove the capturing logger
+  auto& loggers = ccf::logger::config::loggers();
+  loggers.erase(
+    std::remove_if(
+      loggers.begin(),
+      loggers.end(),
+      [logger_ptr](const auto& l) { return l.get() == logger_ptr; }),
+    loggers.end());
+}
diff --git a/src/tasks/thread_manager.cpp b/src/tasks/thread_manager.cpp
@@ -90,7 +90,10 @@ namespace ccf::tasks
           auto& stop_signal = stop_signals[i].value;
           stop_signal.store(false);
           workers[i] = std::thread(
-            task_worker_loop, std::ref(job_board), std::ref(stop_signal));
+            task_worker_loop,
+            std::ref(job_board),
+            std::ref(stop_signal),
+            /*abort_on_throw=*/true);
         }
       }
 
diff --git a/src/tasks/worker.cpp b/src/tasks/worker.cpp
@@ -0,0 +1,158 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the Apache 2.0 License.
+
+#include "tasks/worker.h"
+
+#include <cstdlib>
+#include <cxxabi.h>
+#include <dlfcn.h>
+#include <execinfo.h>
+#include <memory>
+#include <sstream>
+
+namespace ccf::tasks
+{
+  // Maximum number of frames to capture at the throw-point
+  static constexpr int throw_trace_max_frames = 128;
+
+  struct ThrowTrace
+  {
+    void* frames[throw_trace_max_frames] = {};
+    size_t num_frames = 0;
+  };
+
+  namespace
+  {
+    thread_local ThrowTrace current_throw_trace = {};
+
+    struct FreeDeleter
+    {
+      void operator()(char* p) const
+      {
+        // NOLINTNEXTLINE(cppcoreguidelines-no-malloc,cppcoreguidelines-owning-memory)
+        free(p);
+      }
+    };
+
+    struct FreePtrArrayDeleter
+    {
+      void operator()(char** p) const
+      {
+        // NOLINTNEXTLINE(cppcoreguidelines-no-malloc,cppcoreguidelines-owning-memory,bugprone-multi-level-implicit-pointer-conversion)
+        free(p);
+      }
+    };
+
+    std::string demangle_symbol(const char* raw)
+    {
+      // backtrace_symbols format: "binary(mangled+0xoffset) [0xaddr]"
+      // Try to extract and demangle the symbol name between '(' and '+'/')'
+      std::string entry(raw);
+      auto open = entry.find('(');
+      auto plus = entry.find('+', open != std::string::npos ? open : 0);
+      auto close = entry.find(')', open != std::string::npos ? open : 0);
+
+      if (
+        open != std::string::npos && close != std::string::npos &&
+        close > open + 1)
+      {
+        auto end = (plus != std::string::npos && plus < close) ? plus : close;
+        std::string mangled = entry.substr(open + 1, end - open - 1);
+
+        if (!mangled.empty())
+        {
+          int status = 0;
+          std::unique_ptr<char, FreeDeleter> demangled(
+            abi::__cxa_demangle(mangled.c_str(), nullptr, nullptr, &status));
+          if (status == 0 && demangled != nullptr)
+          {
+            std::string rest = entry.substr(end);
+            entry = entry.substr(0, open + 1) + demangled.get() + rest;
+          }
+        }
+      }
+
+      return entry;
+    }
+
+    // Format a demangled stack trace as a string. Note: backtrace_symbols only
+    // resolves symbols exported to the dynamic symbol table (e.g. via
+    // -rdynamic). Static/internal functions will appear as raw addresses. For
+    // broader coverage, consider integrating libbacktrace (reads DWARF
+    // directly) or invoking addr2line at runtime.
+    std::string format_stacktrace(void** frames, int num_frames)
+    {
+      std::ostringstream oss;
+      // NOLINTNEXTLINE(cppcoreguidelines-no-malloc,cppcoreguidelines-owning-memory,bugprone-multi-level-implicit-pointer-conversion)
+      std::unique_ptr<char*, FreePtrArrayDeleter> symbols(
+        backtrace_symbols(frames, num_frames));
+      if (symbols == nullptr)
+      {
+        // If memory allocation fails, return a message indicating the issue
+        return "  (failed to allocate memory for backtrace symbols)\n";
+      }
+      for (int i = 0; i < num_frames; ++i)
+      {
+        oss << "  #" << i << ": " << demangle_symbol(symbols.get()[i]) << "\n";
+      }
+      return oss.str();
+    }
+  }
+
+  void dump_stacktrace(const std::string& msg)
+  {
+    LOG_FATAL_FMT("{}", msg);
+
+    auto& throw_trace = current_throw_trace;
+    if (throw_trace.num_frames > 0)
+    {
+      LOG_FATAL_FMT(
+        "Stack trace:\n{}",
+        format_stacktrace(throw_trace.frames, throw_trace.num_frames));
+
+      // Reset so that a subsequent dump does not re-use a stale trace
+      // (e.g. if an earlier throw was caught internally and a later
+      // throw; / re-throw escapes without calling __cxa_throw).
+      throw_trace.num_frames = 0;
+    }
+    else
+    {
+      LOG_FATAL_FMT("No throw-point stack trace available");
+    }
+  }
+}
+
+// Interpose __cxa_throw to capture a backtrace at each throw-point.
+// This is called by the C++ runtime whenever `throw` is executed.
+extern "C"
+{
+  using CxaThrowFn = void (*)(void*, std::type_info*, void (*)(void*));
+
+  void __cxa_throw(
+    void* thrown_exception, std::type_info* tinfo, void (*dest)(void*))
+  {
+    // Capture the backtrace at the throw site
+    auto& trace = ccf::tasks::current_throw_trace;
+    trace.num_frames =
+      backtrace(trace.frames, ccf::tasks::throw_trace_max_frames);
+
+    // Forward to the real __cxa_throw
+    static auto real_cxa_throw =
+      reinterpret_cast<CxaThrowFn>(dlsym(RTLD_NEXT, "__cxa_throw"));
+    if (real_cxa_throw != nullptr)
+    {
+      real_cxa_throw(thrown_exception, tinfo, dest);
+      // real_cxa_throw is [[noreturn]], so we never reach here
+    }
+    else
+    {
+      // If dlsym failed, we cannot safely proceed. Abort to prevent undefined
+      // behavior.
+      std::abort();
+    }
+    // Both real_cxa_throw and std::abort() are [[noreturn]], but the compiler
+    // may not recognize that for function pointers. This satisfies the compiler
+    // that we never return from this function.
+    __builtin_unreachable();
+  }
+}
diff --git a/src/tasks/worker.h b/src/tasks/worker.h

Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,10 @@ namespace ccf::tasks`
`90`	`90`	`auto& stop_signal = stop_signals[i].value;`
`91`	`91`	`stop_signal.store(false);`
`92`	`92`	`workers[i] = std::thread(`
`93`		`- task_worker_loop, std::ref(job_board), std::ref(stop_signal));`
	`93`	`+ task_worker_loop,`
	`94`	`+ std::ref(job_board),`
	`95`	`+ std::ref(stop_signal),`
	`96`	`+ /abort_on_throw=/true);`
`94`	`97`	`}`
`95`	`98`	`}`
`96`	`99`