Merge branch 'master' into feature/ipc-worker

SocketManager throws internal_error for category_generic, aborting rtorrent over RPC.

.github/workflows/commit-lint.yml

@@ -0,0 +1,65 @@
+name: "Lint Commit Message Size"
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened]
+
+jobs:
+  check-commit-bounds:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Validate Line Count, Width, and Spacing
+        run: |
+          # Fetch commit hashes unique to this PR branch
+          COMMITS=$(git log --no-merges --pretty=format:"%H" origin/${{ github.base_ref }}..HEAD)
+
+          MAX_LINES=3
+          MAX_CHARS=90
+          FAILED=0
+
+          for commit in $COMMITS; do
+            SUBJECT=$(git log --format="%s" -n 1 $commit)
+
+            # Extract clean commit message, trimming trailing blank lines
+            RAW_MSG=$(git log --format="%B" -n 1 $commit)
+            CLEAN_MSG=$(echo "$RAW_MSG" | awk '{msg[NR]=$0} END {while(NR>0 && msg[NR]=="") NR--; for(i=1;i<=NR;i++) print msg[i]}')
+
+            # 1. Check total line count
+            LINE_COUNT=$(echo "$CLEAN_MSG" | wc -l)
+            if [ "$LINE_COUNT" -gt "$MAX_LINES" ]; then
+              echo "❌ Error: Commit message has too many lines ($LINE_COUNT/$MAX_LINES)."
+              echo "   Commit: '$SUBJECT'"
+              FAILED=1
+            fi
+
+            # 2. Check for multiple consecutive blank lines
+            # This regex looks for 2 or more empty lines anywhere in the message
+            if echo "$CLEAN_MSG" | grep -pz '(\r?\n){3,}'; then
+              echo "❌ Error: Commit message contains multiple consecutive line breaks."
+              echo "   Commit: '$SUBJECT'"
+              FAILED=1
+            fi
+
+            # 3. Check maximum width of any individual line
+            while IFS= read -r line; do
+              LINE_LENGTH=${#line}
+              if [ "$LINE_LENGTH" -gt "$MAX_CHARS" ]; then
+                echo "❌ Error: Line length exceeds limit ($LINE_LENGTH / $MAX_CHARS chars)."
+                echo "   Offending line: '$line'"
+                FAILED=1
+              fi
+            done <<< "$CLEAN_MSG"
+
+          done
+
+          # Fail the job if any check failed
+          if [ "$FAILED" -ne 0 ]; then
+            exit 1
+          fi
+
+          echo "✅ All commit messages passed style rules!"

configure.ac

@@ -1,6 +1,6 @@
 m4_pattern_allow([PKG_CHECK_EXISTS])
 
-AC_INIT([rtorrent],[0.16.18],[sundell.software@gmail.com])
+AC_INIT([rtorrent],[0.16.19],[sundell.software@gmail.com])
 
 AC_CONFIG_HEADERS([config.h])
 AC_CONFIG_MACRO_DIRS([scripts])
@@ -14,7 +14,7 @@ AX_CXX_COMPILE_STDCXX(20, noext, mandatory)
 
 PKG_PROG_PKG_CONFIG
 
-AC_DEFINE([API_VERSION], [23], [api version])
+AC_DEFINE([API_VERSION], [24], [api version])
 
 RAK_CHECK_CFLAGS
 RAK_CHECK_CXXFLAGS
@@ -47,7 +47,7 @@ fi
 
 PKG_CHECK_MODULES([CPPUNIT], [cppunit],, [no_cppunit="yes"])
 PKG_CHECK_MODULES([ZLIB], [zlib])
-PKG_CHECK_MODULES([DEPENDENCIES], [libtorrent >= 0.16.18])
+PKG_CHECK_MODULES([DEPENDENCIES], [libtorrent >= 0.16.19])
 
 AC_LANG_PUSH(C++)
 TORRENT_WITH_XMLRPC_C

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/command_local.cc

@@ -117,7 +117,7 @@ group_insert(const torrent::Object::list_type& args) {
   rpc::commands.call("method.insert", rpc::create_object_list("group." + name + ".ratio.enable", "simple",
                                                               "schedule=group." + name + ".ratio,5,60,on_ratio=" + name));
   rpc::commands.call("method.insert", rpc::create_object_list("group." + name + ".ratio.disable", "simple",
-                                                              "schedule_remove=group." + name + ".ratio"));
+                                                              "schedule.remove=group." + name + ".ratio"));
   rpc::commands.call("method.insert", rpc::create_object_list("group."  + name + ".ratio.command", "simple",
                                                               "d.try_close= ;d.ignore_commands.set=1"));
   rpc::commands.call("method.insert", rpc::create_object_list("group." + name + ".view", "string", view));
@@ -250,12 +250,12 @@ initialize_command_local() {
 
     CMD_ANY        (category_name + ".size",      [category](auto, auto) { return torrent::runtime::socket_manager()->category_managed_size(category); });
     CMD_ANY        (category_name + ".max_size",  [category](auto, auto) { return torrent::runtime::socket_manager()->category_max_size(category); });
-    CMD_ANY        (category_name + ".min_alloc", [category](auto, auto) { return torrent::runtime::socket_manager()->category_min_allocation(category); });
-    CMD_ANY        (category_name + ".max_alloc", [category](auto, auto) { return torrent::runtime::socket_manager()->category_max_allocation(category); });
 
     if (i == 0)
       continue;
 
+    CMD_ANY        (category_name + ".min_alloc",     [category](auto, auto) { return torrent::runtime::socket_manager()->category_min_allocation(category); });
+    CMD_ANY        (category_name + ".max_alloc",     [category](auto, auto) { return torrent::runtime::socket_manager()->category_max_allocation(category); });
     CMD_ANY_VALUE_V(category_name + ".min_alloc.set", [category](auto, auto& value) { torrent::runtime::socket_manager()->set_category_min_allocation(category, value); });
     CMD_ANY_VALUE_V(category_name + ".max_alloc.set", [category](auto, auto& value) { torrent::runtime::socket_manager()->set_category_max_allocation(category, value); });
   }
@@ -354,6 +354,10 @@ initialize_command_local() {
 
     rpc::rpc.mark_safe(category_name + ".size");
     rpc::rpc.mark_safe(category_name + ".max_size");
+
+    if (i == 0)
+      continue;
+
     rpc::rpc.mark_safe(category_name + ".min_alloc");
     rpc::rpc.mark_safe(category_name + ".max_alloc");
   }

src/control.h

@@ -6,7 +6,7 @@
 #include <memory>
 #include <sys/types.h>
 #include <torrent/torrent.h>
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 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{};
 

src/core/dht_manager.h

@@ -2,7 +2,7 @@
 #define RTORRENT_CORE_DHT_MANAGER_H
 
 #include <torrent/object.h>
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 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{};

src/core/download_factory.h

@@ -9,7 +9,7 @@
 #include <iosfwd>
 
 #include <torrent/object.h>
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 #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);

src/core/view.h

@@ -18,7 +18,7 @@
 #include <string>
 #include <vector>
 #include <torrent/object.h>
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 #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

src/display/manager.h

@@ -1,7 +1,7 @@
 #ifndef RTORRENT_DISPLAY_MANAGER_H
 #define RTORRENT_DISPLAY_MANAGER_H
 
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 #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;
 };
 
 }

src/display/window.h

@@ -2,7 +2,7 @@
 #define RTORRENT_WINDOW_BASE_H
 
 #include <functional>
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 #include <torrent/system/thread.h>
 
 #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

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;
 };
 
 }

src/display/window_log.h

@@ -1,8 +1,8 @@
 #ifndef RTORRENT_DISPLAY_WINDOW_LOG_H
 #define RTORRENT_DISPLAY_WINDOW_LOG_H
 
+#include <torrent/system/scheduler.h>
 #include <torrent/utils/log_buffer.h>
-#include <torrent/utils/scheduler.h>
 
 #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<bool> m_log_updating{};
+  align_cacheline
+
+  std::atomic<bool>   m_log_updating{};
 };
 
 }

src/input/input_event.cc

@@ -11,8 +11,7 @@ namespace input {
 
 void
 InputEvent::insert() {
-  torrent::this_thread::poll()->open(this);
-  torrent::this_thread::poll()->insert_read(this);
+  torrent::this_thread::poll()->open_and_insert_read(this);
 }
 
 void

src/input/input_event.h

@@ -2,12 +2,11 @@
 #define RTORRENT_INPUT_INPUT_EVENT_H
 
 #include <functional>
-
-#include <torrent/event.h>
+#include <torrent/system/event.h>
 
 namespace input {
 
-class InputEvent : public torrent::Event {
+class InputEvent : public torrent::system::Event {
 public:
   typedef std::function<void (int)> slot_int;
 

src/main.cc

@@ -112,8 +112,7 @@ main(int argc, char** argv) {
 
     torrent::log_initialize();
 
-    // TODO: Create a fake thread object for initializing other processes and enabling logging.
-    torrent::initialize_main_thread();
+    torrent::runtime::initialize_worker_process_and_main_thread();
 
     // Block SIGCHLD until all threads are created, then unblock on main-thread, to avoid SIGCHLD
     // interrupting other threads.

src/rpc/command_scheduler_item.h

@@ -5,7 +5,7 @@
 
 #include <functional>
 #include <torrent/object.h>
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 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.
 };

src/rpc/exec_file.cc

@@ -1,23 +1,22 @@
 #include "config.h"
 
-#include <cassert>
-#include <cerrno>
-#include <cstring>
-#include <fcntl.h>
-#include <spawn.h>
-#include <string>
+#include "rpc/exec_file.h"
+
+// #include <cassert>
+// #include <cerrno>
+// #include <cstring>
+// #include <fcntl.h>
+// #include <spawn.h>
+// #include <string>
 #include <unistd.h>
-#include <sys/types.h>
-#include <sys/wait.h>
-#include <torrent/net/fd.h>
+// #include <sys/types.h>
+#include <sys/uio.h>
+// #include <torrent/net/fd.h>
 #include <torrent/system/thread.h>
+#include <torrent/system/spawn_process.h>
 #include <torrent/system/types.h>
 
-#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<struct iovec> iovecs;
+    iovecs.reserve(32);
+    iovecs.push_back({const_cast<char*>("\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<char*>(" "), 1});
+
+      iovecs.push_back({*itr, std::strlen(*itr)});
     }
 
-    result = write(m_log_fd, "\n---\n", sizeof("\n---\n"));
+    iovecs.push_back({const_cast<char*>("\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<char*>(prefix), std::strlen(prefix)},
+        {const_cast<char*>(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

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/rpc/scgi.cc

@@ -109,8 +109,7 @@ void
 SCgi::activate() {
   assert(torrent::this_thread::thread() == scgi_thread::thread());
 
-  torrent::this_thread::poll()->open(this);
-  torrent::this_thread::poll()->insert_read(this);
+  torrent::this_thread::poll()->open_and_insert_read(this);
 }
 
 // TODO: This should close the fd to avoid reuse.

src/rpc/scgi.h

@@ -3,13 +3,13 @@
 
 #include <array>
 #include <memory>
-#include <torrent/event.h>
+#include <torrent/system/event.h>
 
 #include "rpc/scgi_task.h"
 
 namespace rpc {
 
-class SCgi : public torrent::Event {
+class SCgi : public torrent::system::Event {
 public:
   static const int max_tasks = 100;
 

src/rpc/scgi_task.cc

@@ -47,8 +47,7 @@ SCgiTask::open(SCgi* parent, int fd) {
                                  // m_trusted=false into the next reuse, given that the
                                  // UNTRUSTED_CONNECTION=0 parse branch is a no-op.
 
-  torrent::this_thread::poll()->open(this);
-  torrent::this_thread::poll()->insert_read(this);
+  torrent::this_thread::poll()->open_and_insert_read(this);
 
   auto lock = std::lock_guard<std::mutex>(m_result_mutex);
 

src/rpc/scgi_task.h

@@ -4,13 +4,13 @@
 #include <memory>
 #include <mutex>
 #include <vector>
-#include <torrent/event.h>
+#include <torrent/system/event.h>
 
 namespace rpc {
 
 class SCgi;
 
-class SCgiTask : public torrent::Event {
+class SCgiTask : public torrent::system::Event {
 public:
   static constexpr int default_buffer_size = 8191;
   static constexpr int max_header_size     = 2000;

src/utils/waitpid_queue.cc

@@ -0,0 +1,107 @@
+#include "config.h"
+
+#include "utils/waitpid_queue.h"
+
+#include <sys/wait.h>
+#include <torrent/exceptions.h>
+
+namespace utils {
+
+WaitpidQueue::WaitpidQueue() {
+  m_worker = std::async(std::launch::async, [this]() {
+      auto wait_time  = 50ms;
+
+      while (true) {
+        if (!m_queue.empty()) {
+          auto start_time = std::chrono::steady_clock::now();
+
+          while (std::chrono::steady_clock::now() - start_time < wait_time) {
+            if (m_should_shutdown.load(std::memory_order_acquire))
+              return;
+
+            std::this_thread::sleep_for(50ms);
+
+            if (m_wakeup_worker.load(std::memory_order_acquire))
+              break;
+          }
+
+        } else {
+          m_wakeup_worker.wait(false, std::memory_order_acquire);
+        }
+
+        // Adds a small delay to allow new processes to finish if they're quickly spawned and
+        // terminated.
+        std::this_thread::sleep_for(50ms);
+
+        std::set<pid_t> queue;
+
+        {
+          std::lock_guard<std::mutex> guard(m_mutex);
+
+          if (m_should_shutdown)
+            return;
+
+          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<std::mutex> 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();
+        }
+      }
+    });
+}
+
+// We don't wait for the worker thread to finish as waitpid isn't needed to be called on shutdown.
+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();
+}
+
+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);
+    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

src/utils/waitpid_queue.h

@@ -0,0 +1,44 @@
+#ifndef RTORRENT_UTILS_WAITPID_QUEUE_H
+#define RTORRENT_UTILS_WAITPID_QUEUE_H
+
+#include <future>
+#include <set>
+#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::set<pid_t>     m_queue;
+
+  align_cacheline
+
+  std::atomic<bool>     m_wakeup_worker{};
+  std::atomic<bool>     m_should_shutdown{};
+
+  std::atomic<uint32_t> m_remaining{};
+};
+
+inline uint32_t WaitpidQueue::size() const { return m_remaining.load(std::memory_order_acquire); }
+
+} // namespace utils
+
+#endif

src/utils/watch_ready_queue.h

@@ -9,7 +9,7 @@
 #include <utility>
 #include <vector>
 
-#include <torrent/utils/scheduler.h>
+#include <torrent/system/scheduler.h>
 
 namespace utils {
 
@@ -49,10 +49,12 @@ private:
   void               update_status(Entry* entry);
   void               schedule();
 
-  std::map<std::string, Entry>   m_entries;
-  std::vector<Entry*>            m_entry_queue;
-  torrent::utils::SchedulerEntry m_task_process;
-  bool                           m_active{true};
+  std::map<std::string, Entry> m_entries;
+  std::vector<Entry*>          m_entry_queue;
+
+  bool                         m_active{true};
+
+  torrent::system::SchedulerEntry m_task_process;
 };
 
 }

test/Makefile.am

@@ -49,6 +49,8 @@ rtorrent_Test_Rpc_SOURCES = $(rtorrent_Test_Common) \
 rtorrent_Test_Src_SOURCES = $(rtorrent_Test_Common) \
 	src/test_command_dynamic.cc \
 	src/test_command_dynamic.h \
+	src/test_command_local.cc \
+	src/test_command_local.h \
 	src/test_watch_ready_queue.cc \
 	src/test_watch_ready_queue.h
 

test/helpers/mock_compare.h

@@ -2,9 +2,10 @@
 #define LIBTORRENT_HELPERS_MOCK_COMPARE_H
 
 #include <algorithm>
+#include <map>
 #include <type_traits>
-#include <torrent/event.h>
 #include <torrent/net/socket_address.h>
+#include <torrent/system/event.h>
 
 // Compare arguments to mock functions with what is expected. The lhs
 // are the expected arguments, rhs are the ones called with.
@@ -13,13 +14,13 @@ template <typename Arg>
 inline bool mock_compare_arg(Arg lhs, Arg rhs) { return lhs == rhs; }
 
 template <int I, typename A, typename... Args>
-typename std::enable_if<I == 1, int>::type
+std::enable_if_t<I == 1, int>
 mock_compare_tuple(const std::tuple<A, Args...>& lhs, const std::tuple<Args...>& rhs) {
   return mock_compare_arg(std::get<I>(lhs), std::get<I - 1>(rhs)) ? 0 : 1;
 }
 
 template <int I, typename A, typename... Args>
-typename std::enable_if<1 < I, int>::type
+std::enable_if_t<1 < I, int>
 mock_compare_tuple(const std::tuple<A, Args...>& lhs, const std::tuple<Args...>& rhs) {
   auto res = mock_compare_tuple<I - 1>(lhs, rhs);
 
@@ -70,24 +71,37 @@ void mock_compare_add(T* v) {
 // Specialize:
 //
 
+constexpr int mock_compare_gt_two_int = -0xFD30;
+
+template <>
+inline bool mock_compare_arg<int>(int lhs, int rhs) {
+  if (lhs == mock_compare_gt_two_int)
+    return rhs > 2;
+  if (rhs == mock_compare_gt_two_int)
+    return lhs > 2;
+
+  return lhs == rhs;
+}
+
 template <>
 inline bool mock_compare_arg<sockaddr*>(sockaddr* lhs, sockaddr* rhs) {
   return lhs != nullptr && rhs != nullptr && torrent::sa_equal(lhs, rhs);
 }
+
 template <>
 inline bool mock_compare_arg<const sockaddr*>(const sockaddr* lhs, const sockaddr* rhs) {
   return lhs != nullptr && rhs != nullptr && torrent::sa_equal(lhs, rhs);
 }
 
 template <>
-inline bool mock_compare_arg<torrent::Event*>(torrent::Event* lhs, torrent::Event* rhs) {
-  if (mock_compare_map<torrent::Event>::is_key(lhs)) {
-    if (!mock_compare_map<torrent::Event>::has_value(rhs)) {
-      mock_compare_map<torrent::Event>::values[lhs] = rhs;
+inline bool mock_compare_arg<torrent::system::Event*>(torrent::system::Event* lhs, torrent::system::Event* rhs) {
+  if (mock_compare_map<torrent::system::Event>::is_key(lhs)) {
+    if (!mock_compare_map<torrent::system::Event>::has_value(rhs)) {
+      mock_compare_map<torrent::system::Event>::values[lhs] = rhs;
       return true;
     }
 
-    return mock_compare_map<torrent::Event>::has_key(lhs) && mock_compare_map<torrent::Event>::get(lhs) == rhs;
+    return mock_compare_map<torrent::system::Event>::has_key(lhs) && mock_compare_map<torrent::system::Event>::get(lhs) == rhs;
   }
 
   return lhs == rhs;

test/helpers/mock_function.cc

@@ -6,9 +6,9 @@
 #include <iostream>
 #include <unistd.h>
 
-#include "torrent/event.h"
 #include "torrent/net/socket_address.h"
 #include "torrent/net/fd.h"
+#include "torrent/system/event.h"
 #include "torrent/utils/log.h"
 #include "torrent/utils/random.h"
 
@@ -32,7 +32,7 @@ mock_clear(bool ignore_assert) {
   MOCK_CLEANUP_MAP(torrent::random_uniform_uint16);
   MOCK_CLEANUP_MAP(torrent::random_uniform_uint32);
 
-  mock_compare_map<torrent::Event>::values.clear();
+  mock_compare_map<torrent::system::Event>::values.clear();
 }
 } // namespace
 
@@ -55,7 +55,9 @@ mock_redirect_defaults([[maybe_unused]] mock_redirect_flags flags) {
 
   mock_redirect(torrent::fd__close, std::function<int(int fildes)>([](int fildes) { return ::close(fildes); }));
   mock_redirect(torrent::fd__fcntl_int, std::function<int(int fildes, int cmd, int arg)>([](int fildes, int cmd, int arg) { return ::fcntl(fildes, cmd, arg); }));
-  mock_redirect(torrent::fd__setsockopt_int, std::function<int(int socket, int level, int option_name, int option_value)>([](int socket, int level, int option_name, int option_value) { return ::setsockopt(socket, level, option_name, &option_value, sizeof(int)); }));
+  mock_redirect(torrent::fd__setsockopt_int, std::function<int(int socket, int level, int option_name, int option_value)>([](int socket, int level, int option_name, int option_value) {
+      return ::setsockopt(socket, level, option_name, &option_value, sizeof(int));
+    }));
   mock_redirect(torrent::fd__socket, std::function<int(int domain, int type, int protocol)>([](int domain, int type, int protocol) { return ::socket(domain, type, protocol); }));
 }
 
@@ -112,69 +114,6 @@ int fd__socket(int domain, int type, int protocol) {
   return mock_call<int>(__func__, &torrent::fd__socket, domain, type, protocol);
 }
 
-//
-// Mock functions for 'torrent/common.h':
-//
-
-namespace this_thread {
-
-void event_open(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_open, event);
-}
-
-void event_open_and_count(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_open_and_count, event);
-}
-
-void event_close_and_count(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_close_and_count, event);
-}
-
-void event_closed_and_count(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_closed_and_count, event);
-}
-
-void event_insert_read(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_insert_read, event);
-}
-
-void event_insert_write(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_insert_write, event);
-}
-
-void event_insert_error(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_insert_error, event);
-}
-
-void event_remove_read(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_remove_read, event);
-}
-
-void event_remove_write(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_remove_write, event);
-}
-
-void event_remove_error(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_remove_error, event);
-}
-
-void event_remove_and_close(Event* event) {
-  MOCK_LOG("fd:%i type_name:%s", event->file_descriptor(), event->type_name());
-  return mock_call<void>(__func__, &torrent::this_thread::event_remove_and_close, event);
-}
-
-}
-
 //
 // Mock functions for 'torrent/utils/random.h':
 //

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>
 TestMainThread::create() {

test/src/test_command_local.cc

@@ -0,0 +1,54 @@
+#include "config.h"
+
+#include "test/src/test_command_local.h"
+
+#include <torrent/torrent.h>
+#include <torrent/runtime/socket_manager.h>
+#include <torrent/utils/option_strings.h>
+
+#include "control.h"
+#include "globals.h"
+#include "rpc/parse_commands.h"
+
+CPPUNIT_TEST_SUITE_REGISTRATION(TestCommandLocal);
+
+void initialize_command_local();
+
+void
+TestCommandLocal::setUp() {
+  torrent::initialize_main_thread();
+  torrent::initialize();
+
+  if (control == nullptr)
+    control = new Control;
+
+  if (!rpc::commands.has("system.sockets.size"))
+    initialize_command_local();
+}
+
+void
+TestCommandLocal::tearDown() {
+  torrent::cleanup();
+}
+
+void
+TestCommandLocal::test_socket_category_commands() {
+  for (uint32_t i = 0; i < torrent::runtime::SocketManager::category_count; ++i) {
+    auto category = static_cast<torrent::runtime::socket_manager_category_t>(i);
+    auto name     = "system.sockets." + torrent::option_to_str_or_throw(torrent::OPTION_SOCKET_CATEGORY, i);
+
+    for (const auto suffix : {".size", ".max_size", ".min_alloc", ".max_alloc"})
+      if (rpc::commands.has(name + suffix))
+        CPPUNIT_ASSERT_NO_THROW(rpc::commands.call(name + suffix));
+
+    const bool has_allocation = category != torrent::runtime::category_generic;
+
+    CPPUNIT_ASSERT(rpc::commands.has(name + ".size"));
+    CPPUNIT_ASSERT(rpc::commands.has(name + ".max_size"));
+
+    CPPUNIT_ASSERT_EQUAL(has_allocation, rpc::commands.has(name + ".min_alloc"));
+    CPPUNIT_ASSERT_EQUAL(has_allocation, rpc::commands.has(name + ".max_alloc"));
+    CPPUNIT_ASSERT_EQUAL(has_allocation, rpc::commands.has(name + ".min_alloc.set"));
+    CPPUNIT_ASSERT_EQUAL(has_allocation, rpc::commands.has(name + ".max_alloc.set"));
+  }
+}

test/src/test_command_local.h

@@ -0,0 +1,15 @@
+#include "test/helpers/test_fixture.h"
+
+class TestCommandLocal : public test_fixture {
+  CPPUNIT_TEST_SUITE(TestCommandLocal);
+
+  CPPUNIT_TEST(test_socket_category_commands);
+
+  CPPUNIT_TEST_SUITE_END();
+
+public:
+  void setUp();
+  void tearDown();
+
+  void test_socket_category_commands();
+};