Improve error handling in ThreadPool (#6011)

* Bug fix: clear all errors after rethrowing
* Semantics improvement - rethrow the actual exception instead of just keeping the message. It also helps with errors not derived from `std::exception`
* Performance: remove unnecessary mutex guarding the per-thread error lists.

Signed-off-by: Michal Zientkiewicz <[email protected]>

Author: mzient

dali/pipeline/util/thread_pool.cc

@@ -90,15 +90,16 @@ void ThreadPool::WaitForWork(bool checkForErrors) {
   started_ = false;
   if (checkForErrors) {
     // Check for errors
-    std::lock_guard lock(error_mutex_);
+    std::exception_ptr err;
     for (size_t i = 0; i < threads_.size(); ++i) {
-      if (!tl_errors_[i].empty()) {
+      if (!err && !tl_errors_[i].empty()) {
         // Throw the first error that occurred
-        string error = make_string("Error in thread ", i, ": ", tl_errors_[i].front());
-        tl_errors_[i].pop();
-        throw std::runtime_error(error);
+        err = std::move(tl_errors_[i].front());
       }
+      tl_errors_[i] = {};
     }
+    if (err)
+      std::rethrow_exception(err);
   }
 }
 
@@ -150,12 +151,8 @@ void ThreadPool::ThreadMain(int thread_id, int device_id, bool set_affinity,
       nvml::SetCPUAffinity(core);
     }
 #endif
-  } catch (std::exception &e) {
-    std::lock_guard lock(error_mutex_);
-    tl_errors_[thread_id].push(e.what());
   } catch (...) {
-    std::lock_guard lock(error_mutex_);
-    tl_errors_[thread_id].push("Caught unknown exception");
+    tl_errors_[thread_id].push(std::current_exception());
   }
 
   while (running_) {
@@ -179,12 +176,8 @@ void ThreadPool::ThreadMain(int thread_id, int device_id, bool set_affinity,
     // in the threads and return an error if one occured.
     try {
       work(thread_id);
-    } catch (std::exception &e) {
-      std::lock_guard lock(error_mutex_);
-      tl_errors_[thread_id].push(e.what());
     } catch (...) {
-      std::lock_guard lock(error_mutex_);
-      tl_errors_[thread_id].push("Caught unknown exception");
+      tl_errors_[thread_id].push(std::current_exception());
     }
 
     // The task is now complete - we can atomically decrement the number of outstanding work.

dali/pipeline/util/thread_pool.h

@@ -24,6 +24,7 @@
 #include <queue>
 #include <thread>
 #include <vector>
+#include <stdexcept>
 #include <string>
 #include "dali/core/common.h"
 #if NVML_ENABLED
@@ -94,11 +95,11 @@ class DLL_PUBLIC ThreadPool {
   bool running_ = true;
   bool started_ = false;
   alignas(64) std::atomic_int outstanding_work_{0};
-  std::mutex error_mutex_, completed_mutex_;
+  std::mutex completed_mutex_;
   std::condition_variable completed_;
 
-  //  Stored error strings for each thread
-  vector<std::queue<string>> tl_errors_;
+  // Stored errors for each thread
+  vector<std::queue<std::exception_ptr>> tl_errors_;
 #if NVML_ENABLED
   nvml::NvmlInstance nvml_handle_;
 #endif