From 06a391f9d82b15e8aed47c5f79e1c5fa6c268eef Mon Sep 17 00:00:00 2001 From: rakshasa Date: Mon, 3 Aug 2026 09:11:23 +0200 Subject: [PATCH] Stuff. --- src/Makefile.am | 2 + src/rpc/exec_file.cc | 2 + src/rpc/exec_file.h | 4 ++ src/utils/waitpid_queue.cc | 119 +++++++++++++++++++++++++++++++++++++ src/utils/waitpid_queue.h | 44 ++++++++++++++ 5 files changed, 171 insertions(+) create mode 100644 src/utils/waitpid_queue.cc create mode 100644 src/utils/waitpid_queue.h diff --git a/src/Makefile.am b/src/Makefile.am index 147a1997..d322db72 100644 --- a/src/Makefile.am +++ b/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 \ \ diff --git a/src/rpc/exec_file.cc b/src/rpc/exec_file.cc index d1467526..c355e90b 100644 --- a/src/rpc/exec_file.cc +++ b/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")); diff --git a/src/rpc/exec_file.h b/src/rpc/exec_file.h index dc0aa160..2a60578d 100644 --- a/src/rpc/exec_file.h +++ b/src/rpc/exec_file.h @@ -3,6 +3,8 @@ #include +#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; }; } diff --git a/src/utils/waitpid_queue.cc b/src/utils/waitpid_queue.cc new file mode 100644 index 00000000..b3730b1a --- /dev/null +++ b/src/utils/waitpid_queue.cc @@ -0,0 +1,119 @@ +#include "config.h" + +#include "utils/waitpid_queue.h" + +#include +#include + +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 queue; + + { + std::lock_guard 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 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 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 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 diff --git a/src/utils/waitpid_queue.h b/src/utils/waitpid_queue.h new file mode 100644 index 00000000..b034af9c --- /dev/null +++ b/src/utils/waitpid_queue.h @@ -0,0 +1,44 @@ +#ifndef RTORRENT_UTILS_WAITPID_QUEUE_H +#define RTORRENT_UTILS_WAITPID_QUEUE_H + +#include +#include +#include + +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 m_worker; + + align_cacheline + + std::mutex m_mutex; + std::vector m_queue; + + bool m_should_shutdown{}; + + align_cacheline + + std::atomic m_wakeup_worker{}; + std::atomic m_remaining{}; +}; + +inline uint32_t WaitpidQueue::size() const { return m_remaining.load(std::memory_order_acquire); } + +} // namespace torrent::utils + +#endif