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/control.h b/src/control.h index 397c996e..dfc287d4 100644 --- a/src/control.h +++ b/src/control.h @@ -6,7 +6,7 @@ #include #include #include -#include +#include namespace ui { class Root; @@ -107,8 +107,8 @@ private: mode_t m_umask; std::string m_workingDirectory; - torrent::utils::SchedulerEntry m_task_shutdown; - torrent::utils::SchedulerEntry m_task_shutdown_clear_requests; + torrent::system::SchedulerEntry m_task_shutdown; + torrent::system::SchedulerEntry m_task_shutdown_clear_requests; int m_clear_requests_count{}; diff --git a/src/core/dht_manager.h b/src/core/dht_manager.h index eb6d6474..613a2d15 100644 --- a/src/core/dht_manager.h +++ b/src/core/dht_manager.h @@ -2,7 +2,7 @@ #define RTORRENT_CORE_DHT_MANAGER_H #include -#include +#include namespace core { @@ -42,8 +42,8 @@ private: uint64_t m_dhtPrevBytesUp; uint64_t m_dhtPrevBytesDown; - torrent::utils::SchedulerEntry m_update_timeout; - torrent::utils::SchedulerEntry m_stop_timeout; + torrent::system::SchedulerEntry m_update_timeout; + torrent::system::SchedulerEntry m_stop_timeout; bool m_warned{}; bool m_set_by_user{}; diff --git a/src/core/download_factory.h b/src/core/download_factory.h index 71513807..435b09d8 100644 --- a/src/core/download_factory.h +++ b/src/core/download_factory.h @@ -9,7 +9,7 @@ #include #include -#include +#include #include "http_queue.h" @@ -79,9 +79,9 @@ private: command_list_type m_commands; torrent::Object::map_type m_variables; - slot_void m_slot_finished; - torrent::utils::SchedulerEntry m_task_load; - torrent::utils::SchedulerEntry m_task_commit; + slot_void m_slot_finished; + torrent::system::SchedulerEntry m_task_load; + torrent::system::SchedulerEntry m_task_commit; }; bool is_network_uri(const std::string& uri); diff --git a/src/core/view.h b/src/core/view.h index 73b3dc0b..abbf0bef 100644 --- a/src/core/view.h +++ b/src/core/view.h @@ -18,7 +18,7 @@ #include #include #include -#include +#include #include "globals.h" @@ -153,8 +153,8 @@ private: std::chrono::microseconds m_last_changed{}; - signal_void m_signal_changed; - torrent::utils::SchedulerEntry m_delay_changed; + signal_void m_signal_changed; + torrent::system::SchedulerEntry m_delay_changed; }; } // namespace core diff --git a/src/display/manager.h b/src/display/manager.h index 32274a3c..57696112 100644 --- a/src/display/manager.h +++ b/src/display/manager.h @@ -1,7 +1,7 @@ #ifndef RTORRENT_DISPLAY_MANAGER_H #define RTORRENT_DISPLAY_MANAGER_H -#include +#include #include "display/frame.h" @@ -31,9 +31,9 @@ private: bool m_force_redraw{false}; Frame m_root_frame; - std::chrono::microseconds m_time_last_update{}; - torrent::utils::ExternalScheduler m_scheduler; - torrent::utils::SchedulerEntry m_task_update; + std::chrono::microseconds m_time_last_update{}; + torrent::system::ExternalScheduler m_scheduler; + torrent::system::SchedulerEntry m_task_update; }; } diff --git a/src/display/window.h b/src/display/window.h index 12e3dfbc..f09bb1e3 100644 --- a/src/display/window.h +++ b/src/display/window.h @@ -2,7 +2,7 @@ #define RTORRENT_WINDOW_BASE_H #include -#include +#include #include #include "canvas.h" @@ -99,7 +99,7 @@ protected: extent_type m_max_width; extent_type m_max_height; - torrent::utils::SchedulerEntry m_task_update; + torrent::system::SchedulerEntry m_task_update; }; // Return a range with a distance of no more than __distance and diff --git a/src/display/window_http_queue.h b/src/display/window_http_queue.h index 1fd46043..e7e6ffbc 100644 --- a/src/display/window_http_queue.h +++ b/src/display/window_http_queue.h @@ -47,7 +47,7 @@ private: signal_curl_get::iterator m_conn_insert; signal_curl_get::iterator m_conn_erase; - torrent::utils::SchedulerEntry m_task_deactivate; + torrent::system::SchedulerEntry m_task_deactivate; }; } diff --git a/src/display/window_log.h b/src/display/window_log.h index bb00c9c7..f3b30ecc 100644 --- a/src/display/window_log.h +++ b/src/display/window_log.h @@ -1,8 +1,8 @@ #ifndef RTORRENT_DISPLAY_WINDOW_LOG_H #define RTORRENT_DISPLAY_WINDOW_LOG_H +#include #include -#include #include "window.h" @@ -22,10 +22,12 @@ public: private: inline iterator find_older(); - torrent::log_buffer* m_log; - torrent::utils::SchedulerEntry m_task_update; + torrent::log_buffer* m_log; + torrent::system::SchedulerEntry m_task_update; - align_cacheline std::atomic m_log_updating{}; + align_cacheline + + std::atomic m_log_updating{}; }; } diff --git a/src/rpc/command_scheduler_item.h b/src/rpc/command_scheduler_item.h index e70e55bf..abe1d78e 100644 --- a/src/rpc/command_scheduler_item.h +++ b/src/rpc/command_scheduler_item.h @@ -5,7 +5,7 @@ #include #include -#include +#include namespace rpc { @@ -43,7 +43,7 @@ private: uint32_t m_interval{}; std::chrono::microseconds m_time_scheduled; - torrent::utils::SchedulerEntry m_task; + torrent::system::SchedulerEntry m_task; // Flags for various things. }; diff --git a/src/rpc/exec_file.cc b/src/rpc/exec_file.cc index d1467526..114922de 100644 --- a/src/rpc/exec_file.cc +++ b/src/rpc/exec_file.cc @@ -1,23 +1,22 @@ #include "config.h" -#include -#include -#include -#include -#include -#include +#include "rpc/exec_file.h" + +// #include +// #include +// #include +// #include +// #include +// #include #include -#include -#include -#include +// #include +#include +// #include #include +#include #include -#include "exec_file.h" -#include "parse.h" - -// Standard POSIX environment pointer -extern char** environ; +#include "rpc/parse.h" namespace rpc { @@ -25,159 +24,56 @@ namespace rpc { int ExecFile::execute(const char* file, char* const* argv, int flags) { - assert(!((flags & flag_capture) && (flags & flag_background))); + torrent::system::SpawnProcess spawn_process; - // Write the executed command and its parameters to the log fd. - [[maybe_unused]] int result; + spawn_process.set_log_fd(m_log_fd); + spawn_process.set_background(flags & flag_background); + spawn_process.set_capture_output(flags & flag_capture); if (m_log_fd != -1) { - for (char* const* itr = argv; *itr != NULL; itr++) { - if (itr == argv) - result = write(m_log_fd, "\n---\n", sizeof("\n---\n")); - else - result = write(m_log_fd, " ", 1); + std::vector iovecs; + iovecs.reserve(32); + iovecs.push_back({const_cast("\n---\n"), 5}); - result = write(m_log_fd, *itr, std::strlen(*itr)); + for (auto* itr = argv; *itr != nullptr; itr++) { + if (itr != argv) + iovecs.push_back({const_cast(" "), 1}); + + iovecs.push_back({*itr, std::strlen(*itr)}); } - result = write(m_log_fd, "\n---\n", sizeof("\n---\n")); + iovecs.push_back({const_cast("\n---\n"), 5}); + + [[maybe_unused]] int result = ::writev(m_log_fd, iovecs.data(), iovecs.size()); } - posix_spawn_file_actions_t actions{}; - - if (posix_spawn_file_actions_init(&actions) != 0) - throw torrent::internal_error("ExecFile::execute(...) posix_spawn_file_actions_init failed."); - - posix_spawnattr_t attr; - posix_spawnattr_init(&attr); - - // Try to avoid leaking open fds to the spawned process. Prefer POSIX_SPAWN_CLOEXEC_DEFAULT - // (macOS-only) or posix_spawn_file_actions_addclosefrom_np (glibc >= 2.34, FreeBSD >= 13.1). - // - // Other platforms like musl libc, OpenBSD and NetBSD must rely on explicit O_CLOEXEC. - - // Handle standard input redirection (/dev/null), posix_spawn_file_actions_addopen handles opening - // and dup2 natively - if (posix_spawn_file_actions_addopen(&actions, 0, "/dev/null", O_RDWR, 0) != 0) { - // Fallback if open fails inside action setup - posix_spawn_file_actions_addclose(&actions, 0); - } - - int pipe_0 = -1; - int pipe_1 = -1; - - // Handle standard output redirection - if (flags & flag_capture) { - torrent::fd_open_pipe(pipe_0, pipe_1); - - posix_spawn_file_actions_adddup2(&actions, pipe_1, 1); - - // Ensure the write end of the pipe is closed in the child after duplicating. - posix_spawn_file_actions_addclose(&actions, pipe_0); - posix_spawn_file_actions_addclose(&actions, pipe_1); - - } else if (m_log_fd != -1) { - posix_spawn_file_actions_adddup2(&actions, m_log_fd, 1); - - } else { - posix_spawn_file_actions_addopen(&actions, 1, "/dev/null", O_RDWR, 0); - } - - if (m_log_fd != -1) { - posix_spawn_file_actions_adddup2(&actions, m_log_fd, 2); - } else { - posix_spawn_file_actions_addopen(&actions, 2, "/dev/null", O_RDWR, 0); - } - - short spawn_flags = 0; - -#if defined(POSIX_SPAWN_CLOEXEC_DEFAULT) - spawn_flags |= POSIX_SPAWN_CLOEXEC_DEFAULT; -#elif defined(HAVE_POSIX_SPAWN_FILE_ACTIONS_ADDCLOSEFROM_NP) - posix_spawn_file_actions_addclosefrom_np(&actions, 3); -#endif - - if (flags & flag_background) { -#ifdef POSIX_SPAWN_SETSID - spawn_flags |= POSIX_SPAWN_SETSID; -#else - spawn_flags |= POSIX_SPAWN_SETPGROUP; - posix_spawnattr_setpgroup(&attr, 0); -#endif - } - - posix_spawnattr_setflags(&attr, spawn_flags); - - pid_t child_pid{}; - int spawn_status = posix_spawnp(&child_pid, file, &actions, &attr, argv, environ); - - posix_spawn_file_actions_destroy(&actions); - posix_spawnattr_destroy(&attr); + int spawn_status = spawn_process.execute(file, argv); if (spawn_status != 0) { - if (pipe_0 != -1) - torrent::fd_close(pipe_0); + if (m_log_fd != -1) { + auto prefix = "\n--- posix_spawn failed: "; + auto errno_str = torrent::system::errno_enum_str(spawn_status) + " ---\n"; - if (pipe_1 != -1) - torrent::fd_close(pipe_1); + struct iovec iovecs[2] = { + {const_cast(prefix), std::strlen(prefix)}, + {const_cast(errno_str.c_str()), errno_str.size()}, + }; + + [[maybe_unused]] int result = ::writev(m_log_fd, iovecs, 2); + } throw torrent::input_error("ExecFile::execute() posix_spawn failed: " + torrent::system::errno_enum_str(spawn_status)); } - if (flags & flag_capture) { - m_capture = std::string(); - torrent::fd_close(pipe_1); - - char buffer[4096]; - ssize_t length; - - do { - length = read(pipe_0, buffer, sizeof(buffer)); - - if (length > 0) - m_capture += std::string(buffer, length); - - } while (length > 0); - - torrent::fd_close(pipe_0); - - if (m_log_fd != -1) { - result = write(m_log_fd, "Captured output:\n", sizeof("Captured output:\n")); - result = write(m_log_fd, m_capture.data(), m_capture.length()); - } - } - if (flags & flag_background) { - if (m_log_fd != -1) - result = write(m_log_fd, "\n--- Running in Background ---\n", sizeof("\n--- Running in Background ---\n")); - + m_waitpid_queue.close_pid(spawn_process.child_pid()); return 0; } - int status; + if (flags & flag_capture) + m_capture = spawn_process.capture_child_output(); - while (::waitpid(child_pid, &status, 0) == -1) { - switch (errno) { - case EINTR: - continue; - case ECHILD: - throw torrent::internal_error("ExecFile::execute(...) waitpid failed with ECHILD, child process not found."); - case EINVAL: - throw torrent::internal_error("ExecFile::execute(...) waitpid failed with EINVAL."); - default: - throw torrent::internal_error("ExecFile::execute(...) waitpid failed with unexpected error: " + std::string(std::strerror(errno))); - } - }; - - // Check return value? - if (m_log_fd != -1) { - if (WIFEXITED(status) && WEXITSTATUS(status) == 0) - result = write(m_log_fd, "\n--- Success ---\n", sizeof("\n--- Success ---\n")); - else - result = write(m_log_fd, "\n--- Error ---\n", sizeof("\n--- Error ---\n")); - } - - return status; + return spawn_process.wait_for_child(); } torrent::Object @@ -237,4 +133,4 @@ ExecFile::execute_object(const torrent::Object& rawArgs, int flags) { return torrent::Object((int64_t)status); } -} +} // namespace rpc 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..ffe3f7fa --- /dev/null +++ b/src/utils/waitpid_queue.cc @@ -0,0 +1,115 @@ +#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 = 50ms; + + 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(50ms); + } + + } else { + m_wakeup_worker.wait(false, std::memory_order_acquire); + std::this_thread::sleep_for(50ms); + } + + std::set 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::min(10 * 1000ms, wait_time * 2); + + for (auto pid : queue) { + if (::waitpid(pid, nullptr, WNOHANG) == 0) + continue; + + { + std::lock_guard guard(m_mutex); + + if (m_queue.erase(pid) != 1) + throw torrent::internal_error("WaitpidQueue worker thread could not find pid in queue."); + } + + wait_time = std::max(50ms, wait_time / 2); + + 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.insert(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..4fef82cd --- /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::set 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 utils + +#endif diff --git a/src/utils/watch_ready_queue.h b/src/utils/watch_ready_queue.h index 1f0515ec..a18160bf 100644 --- a/src/utils/watch_ready_queue.h +++ b/src/utils/watch_ready_queue.h @@ -9,7 +9,7 @@ #include #include -#include +#include namespace utils { @@ -49,10 +49,12 @@ private: void update_status(Entry* entry); void schedule(); - std::map m_entries; - std::vector m_entry_queue; - torrent::utils::SchedulerEntry m_task_process; - bool m_active{true}; + std::map m_entries; + std::vector m_entry_queue; + + bool m_active{true}; + + torrent::system::SchedulerEntry m_task_process; }; } diff --git a/test/helpers/test_main_thread.cc b/test/helpers/test_main_thread.cc index 7a48a07b..f9b3d1aa 100644 --- a/test/helpers/test_main_thread.cc +++ b/test/helpers/test_main_thread.cc @@ -8,7 +8,6 @@ #include "torrent/exceptions.h" #include "torrent/net/resolver.h" #include "torrent/utils/log.h" -#include "torrent/utils/scheduler.h" std::unique_ptr TestMainThread::create() {