Stuff.

src/Makefile.am

@@ -174,6 +174,8 @@ libsub_root_a_SOURCES = \
 	utils/list_focus.h \
 	utils/lockfile.cc \
 	utils/lockfile.h \
+	utils/waitpid_queue.cc \
+	utils/waitpid_queue.h \
 	utils/watch_ready_queue.cc \
 	utils/watch_ready_queue.h \
 	\

src/rpc/exec_file.cc

@@ -148,6 +148,8 @@ ExecFile::execute(const char* file, char* const* argv, int flags) {
   }
 
   if (flags & flag_background) {
+    m_waitpid_queue.close_pid(child_pid);
+
     if (m_log_fd != -1)
       result = write(m_log_fd, "\n--- Running in Background ---\n", sizeof("\n--- Running in Background ---\n"));
 

src/rpc/exec_file.h

@@ -3,6 +3,8 @@
 
 #include <torrent/object.h>
 
+#include "utils/waitpid_queue.h"
+
 namespace rpc {
 
 class ExecFile {
@@ -24,6 +26,8 @@ public:
 private:
   int                 m_log_fd{-1};
   std::string         m_capture;
+
+  utils::WaitpidQueue m_waitpid_queue;
 };
 
 }

src/utils/waitpid_queue.cc

@@ -0,0 +1,119 @@
+#include "config.h"
+
+#include "utils/waitpid_queue.h"
+
+#include <unistd.h>
+#include <torrent/exceptions.h>
+
+namespace utils {
+
+WaitpidQueue::WaitpidQueue() {
+  m_worker = std::async(std::launch::async, [this]() {
+      bool is_running = true;
+      auto wait_time  = 1s;
+
+      while (is_running) {
+        if (!m_queue.empty()) {
+          auto start_time = std::chrono::steady_clock::now();
+
+          while (!m_wakeup_worker.load(std::memory_order_acquire) && !m_should_shutdown) {
+            auto elapsed = std::chrono::steady_clock::now() - start_time;
+
+            if (elapsed >= wait_time)
+              break;
+
+            std::this_thread::sleep_for(100ms);
+          }
+
+        } else {
+          m_wakeup_worker.wait(false, std::memory_order_acquire);
+        }
+
+        std::vector<int> queue;
+
+        {
+          std::lock_guard<std::mutex> guard(m_mutex);
+
+          if (m_should_shutdown) {
+            if (m_queue.empty())
+              return;
+
+            is_running = false;
+          }
+
+          if (m_queue.empty())
+            throw torrent::internal_error("WaitpidQueue worker thread woke up but queue is empty.");
+
+          queue = m_queue;
+
+          m_wakeup_worker.store(false, std::memory_order_release);
+        }
+
+        wait_time = std::max(10s, wait_time * 2);
+
+        for (int pid : queue) {
+          if (::waitpid(pid, nullptr, WNOHANG) == 0)
+            continue;
+
+          {
+            std::lock_guard<std::mutex> guard(m_mutex);
+
+            auto itr = std::find(m_queue.begin(), m_queue.end(), pid);
+
+            if (itr == m_queue.end())
+              throw torrent::internal_error("WaitpidQueue worker thread could not find pid in queue.");
+
+            *itr = m_queue.back();
+            m_queue.pop_back();
+          }
+
+          wait_time = 1s;
+
+          m_remaining.fetch_sub(1, std::memory_order_release);
+          m_remaining.notify_all();
+        }
+      }
+    });
+}
+
+WaitpidQueue::~WaitpidQueue() {
+  {
+    std::lock_guard<std::mutex> guard(m_mutex);
+    m_should_shutdown = true;
+  }
+
+  m_wakeup_worker.store(true, std::memory_order_release);
+  m_wakeup_worker.notify_all();
+
+  // m_worker.wait();
+}
+
+void
+WaitpidQueue::close_pid(pid_t pid) {
+  if (pid < 0)
+    throw torrent::internal_error("WaitpidQueue::close_pid() called with invalid pid.");
+
+  m_remaining.fetch_add(1, std::memory_order_acquire);
+
+  {
+    std::lock_guard<std::mutex> guard(m_mutex);
+
+    if (!m_queue.empty()) {
+      m_queue.push_back(pid);
+      return;
+    }
+
+    m_queue.push_back(pid);
+  }
+
+  m_wakeup_worker.store(true, std::memory_order_release);
+  m_wakeup_worker.notify_all();
+}
+
+void
+WaitpidQueue::wait_for(uint32_t max_remaining) {
+  while (m_remaining.load(std::memory_order_acquire) > max_remaining)
+    m_remaining.wait(max_remaining, std::memory_order_acquire);
+}
+
+} // namespace torrent::utils

src/utils/waitpid_queue.h

@@ -0,0 +1,44 @@
+#ifndef RTORRENT_UTILS_WAITPID_QUEUE_H
+#define RTORRENT_UTILS_WAITPID_QUEUE_H
+
+#include <deque>
+#include <future>
+#include <torrent/system/common.h>
+
+namespace utils {
+
+class WaitpidQueue {
+public:
+  WaitpidQueue();
+  ~WaitpidQueue();
+
+  uint32_t            size() const;
+
+  void                close_pid(int pid);
+
+  void                wait_for(uint32_t max_remaining);
+
+private:
+  WaitpidQueue(const WaitpidQueue&) = delete;
+  WaitpidQueue& operator=(const WaitpidQueue&) = delete;
+
+  std::future<void>   m_worker;
+
+  align_cacheline
+
+  std::mutex          m_mutex;
+  std::vector<pid_t>  m_queue;
+
+  bool                m_should_shutdown{};
+
+  align_cacheline
+
+  std::atomic<bool>     m_wakeup_worker{};
+  std::atomic<uint32_t> m_remaining{};
+};
+
+inline uint32_t WaitpidQueue::size() const { return m_remaining.load(std::memory_order_acquire); }
+
+} // namespace torrent::utils
+
+#endif