Stuff.

src/rpc/exec_file.cc

@@ -1,23 +1,22 @@
 #include "config.h"
 
-#include <cassert>
+#include "rpc/exec_file.h"
-#include <cerrno>
+
-#include <cstring>
+// #include <cassert>
-#include <fcntl.h>
+// #include <cerrno>
-#include <spawn.h>
+// #include <cstring>
-#include <string>
+// #include <fcntl.h>
+// #include <spawn.h>
+// #include <string>
 #include <unistd.h>
-#include <sys/types.h>
+// #include <sys/types.h>
-#include <sys/wait.h>
+#include <sys/uio.h>
-#include <torrent/net/fd.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 "rpc/parse.h"
-#include "parse.h"
-
-// Standard POSIX environment pointer
-extern char** environ;
 
 namespace rpc {
 
@@ -25,218 +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.
+  spawn_process.set_log_fd(m_log_fd);
-  [[maybe_unused]] int result;
+  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++) {
+    std::vector<struct iovec> iovecs;
-      if (itr == argv)
+    iovecs.reserve(32);
-        result = write(m_log_fd, "\n---\n", sizeof("\n---\n"));
+    iovecs.push_back({const_cast<char*>("\n---\n"), 5});
-      else
-        result = write(m_log_fd, " ", 1);
 
-      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{};
+  int spawn_status = spawn_process.execute(file, argv);
-
-  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);
 
   if (spawn_status != 0) {
-    if (pipe_0 != -1)
+    if (m_log_fd != -1) {
-      torrent::fd_close(pipe_0);
+      auto prefix    = "\n--- posix_spawn failed: ";
+      auto errno_str = torrent::system::errno_enum_str(spawn_status) + " ---\n";
 
-    if (pipe_1 != -1)
+      struct iovec iovecs[2] = {
-      torrent::fd_close(pipe_1);
+        {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) {
-    m_waitpid_queue.close_pid(child_pid);
+    m_waitpid_queue.close_pid(spawn_process.child_pid());
-
-    if (m_log_fd != -1)
-      result = write(m_log_fd, "\n--- Running in Background ---\n", sizeof("\n--- Running in Background ---\n"));
-
     return 0;
   }
 
-  int status;
-
-  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;
-}
-
-torrent::Object
-ExecFile::execute_object(const torrent::Object& rawArgs, int flags) {
-  char*  argsBuffer[max_args];
-  char** argsCurrent = argsBuffer;
-
-  // Size of value strings are less than 24.
-  char   valueBuffer[buffer_size+1];
-  char*  valueCurrent = valueBuffer;
-
-  if (rawArgs.is_list()) {
-    const torrent::Object::list_type& args = rawArgs.as_list();
-
-    if (args.empty())
-      throw torrent::input_error("Too few arguments.");
-
-    for (torrent::Object::list_const_iterator itr = args.begin(), last = args.end(); itr != last; itr++, argsCurrent++) {
-      if (argsCurrent == argsBuffer + max_args - 1)
-        throw torrent::input_error("Too many arguments.");
-
-      if (itr->is_string() && (!(flags & flag_expand_tilde) || *itr->as_string().c_str() != '~')) {
-        *argsCurrent = const_cast<char*>(itr->as_string().c_str());
-
-      } else {
-        *argsCurrent = valueCurrent;
-        valueCurrent = print_object(valueCurrent, valueBuffer + buffer_size, &*itr, flags) + 1;
-
-        if (valueCurrent >= valueBuffer + buffer_size)
-          throw torrent::input_error("Overflowed execute arg buffer.");
-      }
-    }
-
-  } else {
-    const torrent::Object::string_type& args = rawArgs.as_string();
-
-    if ((flags & flag_expand_tilde) && args.c_str()[0] == '~') {
-      *argsCurrent = valueCurrent;
-      valueCurrent = print_object(valueCurrent, valueBuffer + buffer_size, &rawArgs, flags) + 1;
-    } else {
-      *argsCurrent = const_cast<char*>(args.c_str());
-    }
-
-    argsCurrent++;
-  }
-
-  *argsCurrent = NULL;
-
-  int status = execute(argsBuffer[0], argsBuffer, flags);
-
-  if ((flags & flag_throw) && status != 0)
-    throw torrent::input_error("Bad return code.");
-
   if (flags & flag_capture)
-    return m_capture;
+    m_capture = spawn_process.capture_child_output();
 
-  return torrent::Object((int64_t)status);
+  return spawn_process.wait_for_child();
 }
 
-}
+} // namespace rpc

src/utils/waitpid_queue.cc

@@ -2,7 +2,7 @@
 
 #include "utils/waitpid_queue.h"
 
-#include <unistd.h>
+#include <sys/wait.h>
 #include <torrent/exceptions.h>
 
 namespace utils {
@@ -10,7 +10,7 @@ namespace utils {
 WaitpidQueue::WaitpidQueue() {
   m_worker = std::async(std::launch::async, [this]() {
       bool is_running = true;
-      auto wait_time  = 1s;
+      auto wait_time  = 50ms;
 
       while (is_running) {
         if (!m_queue.empty()) {
@@ -22,14 +22,15 @@ WaitpidQueue::WaitpidQueue() {
             if (elapsed >= wait_time)
               break;
 
-            std::this_thread::sleep_for(100ms);
+            std::this_thread::sleep_for(50ms);
           }
 
         } else {
           m_wakeup_worker.wait(false, std::memory_order_acquire);
+          std::this_thread::sleep_for(50ms);
         }
 
-        std::vector<int> queue;
+        std::set<pid_t> queue;
 
         {
           std::lock_guard<std::mutex> guard(m_mutex);
@@ -49,25 +50,20 @@ WaitpidQueue::WaitpidQueue() {
           m_wakeup_worker.store(false, std::memory_order_release);
         }
 
-        wait_time = std::max(10s, wait_time * 2);
+        wait_time = std::min(10 * 1000ms, wait_time * 2);
 
-        for (int pid : queue) {
+        for (auto 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 (m_queue.erase(pid) != 1)
-
-            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;
+          wait_time = std::max(50ms, wait_time / 2);
 
           m_remaining.fetch_sub(1, std::memory_order_release);
           m_remaining.notify_all();
@@ -98,12 +94,12 @@ WaitpidQueue::close_pid(pid_t pid) {
   {
     std::lock_guard<std::mutex> guard(m_mutex);
 
-    if (!m_queue.empty()) {
+    // if (!m_queue.empty()) {
-      m_queue.push_back(pid);
+    //   m_queue.push_back(pid);
-      return;
+    //   return;
-    }
+    // }
 
-    m_queue.push_back(pid);
+    m_queue.insert(pid);
   }
 
   m_wakeup_worker.store(true, std::memory_order_release);

src/utils/waitpid_queue.h

@@ -1,8 +1,8 @@
 #ifndef RTORRENT_UTILS_WAITPID_QUEUE_H
 #define RTORRENT_UTILS_WAITPID_QUEUE_H
 
-#include <deque>
 #include <future>
+#include <set>
 #include <torrent/system/common.h>
 
 namespace utils {
@@ -27,7 +27,7 @@ private:
   align_cacheline
 
   std::mutex          m_mutex;
-  std::vector<pid_t>  m_queue;
+  std::set<pid_t>     m_queue;
 
   bool                m_should_shutdown{};
 
@@ -39,6 +39,6 @@ private:
 
 inline uint32_t WaitpidQueue::size() const { return m_remaining.load(std::memory_order_acquire); }
 
-} // namespace torrent::utils
+} // namespace utils
 
 #endif