feat: Add before/after fork functions for executor fork safety

Author: tux3

conanfile.py

@@ -70,8 +70,7 @@ def build(self):
         cmake = CMake(self)
         cmake.configure()
         cmake.build()
-        if self.should_install and self.develop:
-            cmake.install()
+        cmake.install()
 
     def package_info(self):
         self.cpp_info.libs = ["tconcurrent"]

include/tconcurrent/executor.hpp

@@ -51,6 +51,16 @@ class executor
     return _p->signal_error(e);
   }
 
+  void stop_before_fork()
+  {
+    return _p->stop_before_fork();
+  }
+
+  void resume_after_fork()
+  {
+    return _p->resume_after_fork();
+  }
+
   explicit operator bool() const
   {
     return !!_p;
@@ -65,6 +75,8 @@ class executor
     virtual bool is_single_threaded() const = 0;
     virtual bool is_in_this_context() const = 0;
     virtual void signal_error(std::exception_ptr const& e) = 0;
+    virtual void stop_before_fork() = 0;
+    virtual void resume_after_fork() = 0;
   };
 
   template <typename T>
@@ -100,6 +112,16 @@ class executor
       return _context.signal_error(e);
     }
 
+    void stop_before_fork() override
+    {
+      return _context.stop_before_fork();
+    }
+
+    void resume_after_fork() override
+    {
+      return _context.resume_after_fork();
+    }
+
   private:
     T& _context;
   };

include/tconcurrent/thread_pool.hpp

@@ -43,6 +43,9 @@ class TCONCURRENT_EXPORT thread_pool
   void start(unsigned int thread_count);
   void stop(bool cancel_work = false);
 
+  void stop_before_fork();
+  void resume_after_fork();
+
   bool is_running() const;
 
   bool is_in_this_context() const;

src/thread_pool.cpp

@@ -25,6 +25,10 @@ struct thread_pool::impl
   std::atomic<unsigned> _num_running_threads{0};
   std::atomic<bool> _dead{false};
 
+  // We need to be fork-safe, which means stopping all our threads before
+  // a fork and restoring them after, so that they restart from a clean state
+  std::atomic<unsigned> _num_threads_before_fork{0};
+
   error_handler_cb _error_cb{detail::default_error_cb};
   task_trace_handler_cb _task_trace_handler;
 };
@@ -144,6 +148,29 @@ void thread_pool::stop(bool cancel_work)
   }
 }
 
+void thread_pool::stop_before_fork()
+{
+  assert(!_p->_num_threads_before_fork);
+  _p->_num_threads_before_fork.store(_p->_num_running_threads);
+
+  // Note that this can't _p.release(), that only happens on Windows during
+  // process exit, so _p will still be valid in resume_after_fork
+  stop(true);
+}
+
+void thread_pool::resume_after_fork()
+{
+  unsigned num_threads = _p->_num_threads_before_fork.load();
+  auto error_cb = std::move(_p->_error_cb);
+  auto task_trace_handler_cb = std::move(_p->_task_trace_handler);
+
+  // Fresh start
+  _p.reset(new impl);
+  _p->_error_cb = std::move(error_cb);
+  _p->_task_trace_handler = std::move(task_trace_handler_cb);
+  this->start(num_threads);
+}
+
 bool thread_pool::is_running() const
 {
   return _p->_work != std::nullopt;