Compare commits

...

5 Commits

Author SHA1 Message Date
rakshasa e56bffff4d Fixed socket category unit test. 2026-08-05 11:18:31 +02:00
xirvik d20c3234da Add system.sockets allocation budget commands.
They report the budget the socket manager enforces, so a client can show the limit.
2026-08-05 11:18:31 +02:00
Maks Verver dda5e4fa04 Fix deprecated command redirects (rtorrent -D) 2026-08-05 10:14:35 +02:00
Jari Sundell 9ab73352e1 Minor fixes to WaitpidQueue. 2026-08-05 00:59:33 +09:00
Jari Sundell f90a96d57a Added WaitpidQueue to handle background process reaping. 2026-08-04 19:03:26 +09:00
19 changed files with 249 additions and 186 deletions
+2
View File
@@ -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 \
\
+10 -2
View File
@@ -243,6 +243,8 @@ initialize_command_local() {
CMD_ANY ("system.sockets.max_size", [](auto, auto) { return torrent::runtime::socket_manager()->max_size(); });
CMD_ANY_VALUE_V ("system.sockets.max_size.set", [](auto, auto& value) { return torrent::runtime::socket_manager()->set_max_size_and_adjust(value); });
CMD_ANY_V ("system.sockets.adjust_alloc", [](auto, auto) { torrent::runtime::socket_manager()->adjust_allocation(); });
CMD_ANY ("system.sockets.reserved_alloc", [](auto, auto) { return torrent::runtime::socket_manager()->reserved_allocation(); });
CMD_ANY ("system.sockets.available_alloc", [](auto, auto) { return torrent::runtime::socket_manager()->available_allocation(); });
for (uint32_t i = 0; i < torrent::runtime::SocketManager::category_count; ++i) {
auto category = static_cast<torrent::runtime::socket_manager_category_t>(i);
@@ -251,8 +253,10 @@ 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); });
if (i == 0)
if (i == 0) {
CMD_ANY (category_name + ".min_alloc", [](auto, auto) { return torrent::runtime::socket_manager()->generic_min_allocation(); });
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); });
@@ -348,6 +352,8 @@ initialize_command_local() {
rpc::rpc.mark_safe("system.sockets.size");
rpc::rpc.mark_safe("system.sockets.max_size");
rpc::rpc.mark_safe("system.sockets.reserved_alloc");
rpc::rpc.mark_safe("system.sockets.available_alloc");
for (uint32_t i = 0; i < torrent::runtime::SocketManager::category_count; ++i) {
auto category_name = "system.sockets." + torrent::option_to_str_or_throw(torrent::OPTION_SOCKET_CATEGORY, i);
@@ -355,8 +361,10 @@ initialize_command_local() {
rpc::rpc.mark_safe(category_name + ".size");
rpc::rpc.mark_safe(category_name + ".max_size");
if (i == 0)
if (i == 0) {
rpc::rpc.mark_safe(category_name + ".min_alloc");
continue;
}
rpc::rpc.mark_safe(category_name + ".min_alloc");
rpc::rpc.mark_safe(category_name + ".max_alloc");
+3 -3
View File
@@ -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{};
+3 -3
View File
@@ -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{};
+4 -4
View File
@@ -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);
+3 -3
View File
@@ -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
+4 -4
View File
@@ -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;
};
}
+2 -2
View File
@@ -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
+1 -1
View File
@@ -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;
};
}
+6 -4
View File
@@ -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{};
};
}
+2 -3
View File
@@ -392,13 +392,12 @@ main(int argc, char** argv) {
CMD_REDIRECT("bind", "network.bind_address.set");
CMD_REDIRECT("ip", "network.local_address.set");
CMD_REDIRECT("port_range", "network.port_range.set");
// TODO: Check if dht is on by default.
CMD_REDIRECT("dht", "dht.mode.set");
CMD_REDIRECT("port_random", "network.port_random.set");
CMD_REDIRECT("proxy_address", "network.proxy_address.set");
CMD_REDIRECT("port_random", "network.listen.port.random.set");
CMD_REDIRECT("proxy_address", "network.proxy.http.set");
CMD_REDIRECT("key_layout", "keys.layout.set");
+2 -2
View File
@@ -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.
};
+44 -148
View File
@@ -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
+4
View File
@@ -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;
};
}
+107
View File
@@ -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
+44
View File
@@ -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
+7 -5
View File
@@ -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;
};
}
-1
View File
@@ -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() {
+1 -1
View File
@@ -45,8 +45,8 @@ TestCommandLocal::test_socket_category_commands() {
CPPUNIT_ASSERT(rpc::commands.has(name + ".size"));
CPPUNIT_ASSERT(rpc::commands.has(name + ".max_size"));
CPPUNIT_ASSERT(rpc::commands.has(name + ".min_alloc"));
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"));