From 5dbb0020dcff5cda7699bb23790fa5bba6168979 Mon Sep 17 00:00:00 2001 From: Jari Sundell Date: Wed, 17 Dec 2025 23:43:43 +0100 Subject: [PATCH] Replace ThreadWorker with scgi::ThreadScgi. --- src/Makefile.am | 7 +- src/command_local.cc | 1 - src/command_logging.cc | 14 +- src/command_network.cc | 5 +- src/control.cc | 17 ++- src/globals.cc | 1 - src/globals.h | 23 +++- src/main.cc | 12 +- src/rpc/scgi.cc | 5 +- src/scgi/thread_scgi.cc | 143 ++++++++++++++++++++ src/{thread_worker.h => scgi/thread_scgi.h} | 41 ++++-- src/session/thread_session.cc | 2 +- src/thread_worker.cc | 97 ------------- 13 files changed, 218 insertions(+), 150 deletions(-) create mode 100644 src/scgi/thread_scgi.cc rename src/{thread_worker.h => scgi/thread_scgi.h} (52%) delete mode 100644 src/thread_worker.cc diff --git a/src/Makefile.am b/src/Makefile.am index 7552ac9d..690631eb 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -125,6 +125,9 @@ libsub_root_a_SOURCES = \ rpc/tinyxml2/tinyxml2.cc \ rpc/nlohmann/json.h \ \ + scgi/thread_scgi.cc \ + scgi/thread_scgi.h \ + \ session/session_manager.cc \ session/session_manager.h \ session/thread_session.cc \ @@ -195,9 +198,7 @@ libsub_root_a_SOURCES = \ option_parser.cc \ option_parser.h \ signal_handler.cc \ - signal_handler.h \ - thread_worker.cc \ - thread_worker.h + signal_handler.h AM_CPPFLAGS = -I$(srcdir) -I$(top_srcdir) -DPACKAGE_DATADIR=\"$(pkgdatadir)\" diff --git a/src/command_local.cc b/src/command_local.cc index 3bc24634..736b969f 100644 --- a/src/command_local.cc +++ b/src/command_local.cc @@ -185,7 +185,6 @@ cmd_file_append(const torrent::Object::list_type& args) { void initialize_command_local() { core::DownloadList* dList = control->core()->download_list(); - core::DownloadStore* dStore = control->core()->download_store(); torrent::ChunkManager* chunkManager = torrent::chunk_manager(); torrent::FileManager* fileManager = torrent::file_manager(); diff --git a/src/command_logging.cc b/src/command_logging.cc index 0e22e30d..ecadd1fe 100644 --- a/src/command_logging.cc +++ b/src/command_logging.cc @@ -11,7 +11,6 @@ #include "globals.h" #include "control.h" #include "command_helpers.h" -#include "thread_worker.h" #include "core/download.h" #include "core/download_list.h" #include "core/manager.h" @@ -135,12 +134,13 @@ initialize_command_logging() { CMD2_ANY_LIST ("log.append_file", std::bind(&apply_log_open, log_flag_append_file, std::placeholders::_2)); CMD2_ANY_LIST ("log.append_gz_file", std::bind(&apply_log_open, log_flag_append_file, std::placeholders::_2)); - CMD2_ANY_STRING_V("log.close", std::bind(&torrent::log_close_output_str, std::placeholders::_2)); + CMD2_ANY_STRING_V("log.close", std::bind(&torrent::log_close_output_str, std::placeholders::_2)); - CMD2_ANY_LIST ("log.add_output", std::bind(&apply_log_add_output, std::placeholders::_2)); + CMD2_ANY_LIST ("log.add_output", std::bind(&apply_log_add_output, std::placeholders::_2)); - CMD2_ANY_STRING ("log.execute", std::bind(&apply_log, std::placeholders::_2, 0)); - CMD2_ANY_STRING ("log.vmmap.dump", std::bind(&log_vmmap_dump, std::placeholders::_2)); - CMD2_ANY_STRING_V("log.rpc", std::bind(&ThreadWorker::set_rpc_log, worker_thread, std::placeholders::_2)); - CMD2_REDIRECT ("log.xmlrpc", "log.rpc"); // For backwards compatibility + CMD2_ANY_STRING ("log.execute", std::bind(&apply_log, std::placeholders::_2, 0)); + CMD2_ANY_STRING ("log.vmmap.dump", std::bind(&log_vmmap_dump, std::placeholders::_2)); + CMD2_ANY_STRING_V("log.rpc", [](const auto&, const auto& str) { scgi_thread::set_rpc_log(str); }); + + CMD2_REDIRECT ("log.xmlrpc", "log.rpc"); // For backwards compatibility } diff --git a/src/command_network.cc b/src/command_network.cc index b7ae6126..9d68a256 100644 --- a/src/command_network.cc +++ b/src/command_network.cc @@ -20,7 +20,6 @@ #include "globals.h" #include "control.h" #include "command_helpers.h" -#include "thread_worker.h" #include "core/download.h" #include "core/manager.h" #include "rpc/scgi.h" @@ -79,7 +78,7 @@ initialize_rpc_handlers() { torrent::Object apply_scgi(const std::string& arg, int type) { - if (worker_thread->scgi() != NULL) + if (scgi_thread::scgi() != nullptr) throw torrent::input_error("SCGI already enabled."); initialize_rpc_handlers(); @@ -142,7 +141,7 @@ apply_scgi(const std::string& arg, int type) { throw torrent::input_error(e.what()); } - worker_thread->set_scgi(scgi); + scgi_thread::set_scgi(scgi); return torrent::Object(); } diff --git a/src/control.cc b/src/control.cc index d33b1f28..1542018c 100644 --- a/src/control.cc +++ b/src/control.cc @@ -8,7 +8,6 @@ #include #include -#include "thread_worker.h" #include "core/dht_manager.h" #include "core/download_store.h" #include "core/http_queue.h" @@ -63,7 +62,7 @@ Control::initialize() { session_thread::manager()->start(); session_thread::thread()->start_thread(); - worker_thread->start_thread(); + scgi_thread::thread()->start_thread(); display::Canvas::initialize(); display::Window::slot_schedule([this](display::Window* w, std::chrono::microseconds t) { m_display->schedule(w, t); }); @@ -90,6 +89,9 @@ Control::cleanup() { if(!display::Canvas::daemon()) m_inputStdin->remove(torrent::this_thread::poll()); + if (scgi_thread::thread()->is_active()) + scgi_thread::thread()->stop_thread_wait(); + // Wait for all session files to be written. session_thread::thread()->stop_thread_wait(); @@ -109,7 +111,7 @@ Control::cleanup_exception() { bool Control::is_shutdown_completed() { - if (!m_shutdownQuick || worker_thread->is_active()) + if (!m_shutdownQuick) return false; // Tracker requests can be disowned, so wait for these to @@ -128,10 +130,10 @@ void Control::handle_shutdown() { rpc::commands.call_catch("event.system.shutdown", rpc::make_target(), "shutdown", "System shutdown event action failed: "); - if (!m_shutdownQuick) { - if (worker_thread->is_active()) - worker_thread->stop_thread_wait(); + if (scgi_thread::thread()->is_active()) + scgi_thread::thread()->stop_thread_wait(); + if (!m_shutdownQuick) { torrent::runtime::network_manager()->listen_close(); m_directory_events->close(); @@ -141,9 +143,6 @@ Control::handle_shutdown() { torrent::this_thread::scheduler()->wait_for_ceil_seconds(&m_task_shutdown, 5s); } else { - if (worker_thread->is_active()) - worker_thread->stop_thread_wait(); - m_core->shutdown(true); } diff --git a/src/globals.cc b/src/globals.cc index 41704f68..bbbd8d8d 100644 --- a/src/globals.cc +++ b/src/globals.cc @@ -5,4 +5,3 @@ rpc::ip_table_list ip_tables; Control* control{}; -ThreadWorker* worker_thread{}; diff --git a/src/globals.h b/src/globals.h index eda79244..e9c02d06 100644 --- a/src/globals.h +++ b/src/globals.h @@ -6,20 +6,33 @@ #include "rpc/ip_table_list.h" class Control; -class ThreadWorker; extern rpc::ip_table_list ip_tables; extern Control* control; -// TODO: Update to new thread model. -extern ThreadWorker* worker_thread; +namespace rpc { +class SCgi; +} namespace session { - class SessionManager; +} -} // namespace session +namespace scgi_thread { + +torrent::utils::Thread* thread(); +std::thread::id thread_id(); + +void callback(void* target, std::function&& fn); +void cancel_callback(void* target); +void cancel_callback_and_wait(void* target); + +rpc::SCgi* scgi(); +void set_scgi(rpc::SCgi* scgi); +void set_rpc_log(const std::string& filename); + +} // namespace torrent::scgi_thread namespace session_thread { diff --git a/src/main.cc b/src/main.cc index 8a98b22f..4c9893c0 100644 --- a/src/main.cc +++ b/src/main.cc @@ -31,6 +31,7 @@ #include "rpc/command_scheduler.h" #include "rpc/command_scheduler_item.h" #include "rpc/parse_commands.h" +#include "scgi/thread_scgi.h" #include "session/thread_session.h" #include "ui/root.h" #include "utils/directory.h" @@ -41,8 +42,6 @@ #include "signal_handler.h" #include "option_parser.h" -#include "thread_worker.h" - #define LT_LOG(log_fmt, ...) \ lt_log_print(torrent::LOG_SYSTEM, "system: " log_fmt, __VA_ARGS__); @@ -215,12 +214,9 @@ main(int argc, char** argv) { torrent::initialize(); torrent::set_main_thread_slots(std::bind(&client_perform)); + scgi::ThreadScgi::create_thread(); session::ThreadSession::create_thread(); - // TODO: Move to controller. - worker_thread = new ThreadWorker(); - worker_thread->init_thread(); - // Initialize option handlers after libtorrent to ensure // torrent::ConnectionManager* are valid etc. initialize_commands(); @@ -535,11 +531,9 @@ main(int argc, char** argv) { delete control; control = nullptr; + scgi::ThreadScgi::destroy_thread(); session::ThreadSession::destroy_thread(); - delete worker_thread; - worker_thread = nullptr; - torrent::log_cleanup(); return 0; diff --git a/src/rpc/scgi.cc b/src/rpc/scgi.cc index 271b18c3..0c81586b 100644 --- a/src/rpc/scgi.cc +++ b/src/rpc/scgi.cc @@ -11,7 +11,6 @@ #include "control.h" #include "globals.h" -#include "thread_worker.h" #include "rpc/scgi_task.h" #include "utils/socket_fd.h" @@ -93,7 +92,7 @@ SCgi::open(void* sa, unsigned int length) { void SCgi::activate() { - assert(std::this_thread::get_id() == worker_thread->thread_id() && "SCgi::activate() must be called from the worker thread."); + assert(torrent::this_thread::thread() == scgi_thread::thread()); torrent::this_thread::poll()->open(this); torrent::this_thread::poll()->insert_read(this); @@ -102,7 +101,7 @@ SCgi::activate() { void SCgi::deactivate() { - assert(std::this_thread::get_id() == worker_thread->thread_id() && "SCgi::deactivate() must be called from the worker thread."); + assert(torrent::this_thread::thread() == scgi_thread::thread()); torrent::this_thread::poll()->remove_and_close(this); } diff --git a/src/scgi/thread_scgi.cc b/src/scgi/thread_scgi.cc new file mode 100644 index 00000000..cf01e35a --- /dev/null +++ b/src/scgi/thread_scgi.cc @@ -0,0 +1,143 @@ +#include "config.h" + +#include "scgi/thread_scgi.h" + +#include +#include +#include +#include +#include + +#include "rpc/scgi.h" + +namespace scgi { + +class ThreadScgiInternal { +public: + static ThreadScgi* thread_scgi() { return ThreadScgi::internal_thread_scgi(); } +}; + +ThreadScgi* ThreadScgi::m_thread_scgi{}; + +void +ThreadScgi::create_thread() { + auto thread = new ThreadScgi; + + m_thread_scgi = thread; + m_thread_scgi->m_state = STATE_INITIALIZED; +} + +void +ThreadScgi::destroy_thread() { + delete m_thread_scgi; + m_thread_scgi = nullptr; +} + +ThreadScgi* +ThreadScgi::thread_scgi() { + return m_thread_scgi; +} + +// TODO: Remove '= 0'. +void +ThreadScgi::init_thread() { +} + +void +ThreadScgi::cleanup_thread() { + if (m_scgi != nullptr) + m_scgi.load()->deactivate(); +} + +rpc::SCgi* +ThreadScgi::scgi() { + return m_scgi; +} + +bool +ThreadScgi::set_scgi(rpc::SCgi* scgi) { + rpc::SCgi* expected = nullptr; + + if (!m_scgi.compare_exchange_strong(expected, scgi)) + return false; + + change_rpc_log(); + + callback(nullptr, [this]() { + if (m_scgi == NULL) + throw torrent::internal_error("Tried to start SCGI but object was not present."); + + m_scgi.load()->activate(); + }); + + return true; +} + +void +ThreadScgi::set_rpc_log(const std::string& filename) { + callback(nullptr, [this, filename]() { + m_rpc_log_filename = filename; + change_rpc_log(); + }); +} + +void +ThreadScgi::change_rpc_log() { + if (scgi() == NULL) + return; + + if (scgi()->log_fd() != -1) { + ::close(scgi()->log_fd()); + scgi()->set_log_fd(-1); + lt_log_print(torrent::LOG_NOTICE, "Closed RPC log.", 0); + } + + if (m_rpc_log_filename.empty()) + return; + + scgi()->set_log_fd(open(rak::path_expand(m_rpc_log_filename).c_str(), O_WRONLY | O_APPEND | O_CREAT, 0644)); + + if (scgi()->log_fd() == -1) { + lt_log_print(torrent::LOG_NOTICE, "Could not open RPC log file '%s'.", m_rpc_log_filename.c_str()); + return; + } + + lt_log_print(torrent::LOG_NOTICE, "Logging RPC events to '%s'.", m_rpc_log_filename.c_str()); +} + +void +ThreadScgi::call_events() { + if ((m_flags & flag_do_shutdown)) { + if ((m_flags & flag_did_shutdown)) + throw torrent::internal_error("Already trigged shutdown."); + + m_flags |= flag_did_shutdown; + throw torrent::shutdown_exception(); + } + + process_callbacks(); +} + +std::chrono::microseconds +ThreadScgi::next_timeout() { + // TODO: This leads to kqueue crash? + // return std::chrono::microseconds(1h); + return std::chrono::microseconds(10min); +} + +} // namespace scgi + +namespace scgi_thread { + +torrent::utils::Thread* thread() { return scgi::ThreadScgiInternal::thread_scgi(); } +std::thread::id thread_id() { return scgi::ThreadScgiInternal::thread_scgi()->thread_id(); } + +void callback(void* target, std::function&& fn) { scgi::ThreadScgiInternal::thread_scgi()->callback(target, std::move(fn)); } +void cancel_callback(void* target) { scgi::ThreadScgiInternal::thread_scgi()->cancel_callback(target); } +void cancel_callback_and_wait(void* target) { scgi::ThreadScgiInternal::thread_scgi()->cancel_callback_and_wait(target); } + +rpc::SCgi* scgi() { return scgi::ThreadScgiInternal::thread_scgi()->scgi(); } +void set_scgi(rpc::SCgi* scgi) { scgi::ThreadScgiInternal::thread_scgi()->set_scgi(scgi); } +void set_rpc_log(const std::string& filename) { scgi::ThreadScgiInternal::thread_scgi()->set_rpc_log(filename); } + +} // namespace scgi_thread diff --git a/src/thread_worker.h b/src/scgi/thread_scgi.h similarity index 52% rename from src/thread_worker.h rename to src/scgi/thread_scgi.h index 736fc060..66f7a083 100644 --- a/src/thread_worker.h +++ b/src/scgi/thread_scgi.h @@ -1,5 +1,5 @@ -#ifndef RTORRENT_THREAD_WORKER_H -#define RTORRENT_THREAD_WORKER_H +#ifndef RTORRENT_SCGI_THREAD_SCGI_H +#define RTORRENT_SCGI_THREAD_SCGI_H #include #include @@ -9,30 +9,49 @@ namespace rpc { class SCgi; } -class ThreadWorker : public torrent::utils::Thread { -public: - ThreadWorker() = default; - ~ThreadWorker(); +namespace scgi { - const char* name() const override { return "rtorrent scgi"; } +class ThreadScgiInternal; + +class ThreadScgi : public torrent::utils::Thread { +public: + + static void create_thread(); + static void destroy_thread(); + static ThreadScgi* thread_scgi(); + + const char* name() const override { return "rtorrent-scgi"; } void init_thread() override; void cleanup_thread() override; - rpc::SCgi* scgi() { return m_scgi; } + rpc::SCgi* scgi(); bool set_scgi(rpc::SCgi* scgi); void set_rpc_log(const std::string& filename); +protected: + friend class ThreadScgiInternal; + + ThreadScgi() = default; + + static auto internal_thread_scgi() { return m_thread_scgi; } + + void call_events() override; + std::chrono::microseconds next_timeout() override; + private: void task_touch_log(); void change_rpc_log(); - void call_events() override; - std::chrono::microseconds next_timeout() override; + static ThreadScgi* m_thread_scgi; std::atomic m_scgi{nullptr}; std::string m_rpc_log_filename; + + // std::unique_ptr m_manager; }; -#endif // RTORRENT_THREAD_WORKER_H +} // namespace scgi + +#endif // RTORRENT_SCGI_THREAD_SCGI_H diff --git a/src/session/thread_session.cc b/src/session/thread_session.cc index a7dee96d..5215e4e2 100644 --- a/src/session/thread_session.cc +++ b/src/session/thread_session.cc @@ -76,7 +76,7 @@ std::chrono::microseconds ThreadSession::next_timeout() { // TODO: This leads to kqueue crash? // return std::chrono::microseconds(1h); - return std::chrono::microseconds(10s); + return std::chrono::microseconds(10min); } } // namespace session diff --git a/src/thread_worker.cc b/src/thread_worker.cc deleted file mode 100644 index 23ad3e03..00000000 --- a/src/thread_worker.cc +++ /dev/null @@ -1,97 +0,0 @@ -#include "config.h" - -#include "thread_worker.h" - -#include -#include -#include -#include -#include - -#include "globals.h" -#include "control.h" -#include "core/manager.h" -#include "rpc/scgi.h" -#include "rpc/parse_commands.h" - -ThreadWorker::~ThreadWorker() = default; - -void -ThreadWorker::init_thread() { - m_state = STATE_INITIALIZED; -} - -void -ThreadWorker::cleanup_thread() { - if (m_scgi != nullptr) - m_scgi.load()->deactivate(); -} - -bool -ThreadWorker::set_scgi(rpc::SCgi* scgi) { - rpc::SCgi* expected = nullptr; - - if (!m_scgi.compare_exchange_strong(expected, scgi)) - return false; - - change_rpc_log(); - - callback(nullptr, [this]() { - if (m_scgi == NULL) - throw torrent::internal_error("Tried to start SCGI but object was not present."); - - m_scgi.load()->activate(); - }); - - return true; -} - -void -ThreadWorker::set_rpc_log(const std::string& filename) { - callback(nullptr, [this, filename]() { - m_rpc_log_filename = filename; - change_rpc_log(); - }); -} - -void -ThreadWorker::change_rpc_log() { - if (scgi() == NULL) - return; - - if (scgi()->log_fd() != -1) { - ::close(scgi()->log_fd()); - scgi()->set_log_fd(-1); - control->core()->push_log("Closed RPC log."); - } - - if (m_rpc_log_filename.empty()) - return; - - scgi()->set_log_fd(open(rak::path_expand(m_rpc_log_filename).c_str(), O_WRONLY | O_APPEND | O_CREAT, 0644)); - - if (scgi()->log_fd() == -1) { - control->core()->push_log_std("Could not open RPC log file '" + m_rpc_log_filename + "'."); - return; - } - - control->core()->push_log_std("Logging RPC events to '" + m_rpc_log_filename + "'."); -} - -void -ThreadWorker::call_events() { - if ((m_flags & flag_do_shutdown)) { - if ((m_flags & flag_did_shutdown)) - throw torrent::internal_error("Already trigged shutdown."); - - m_flags |= flag_did_shutdown; - throw torrent::shutdown_exception(); - } - - process_callbacks(); -} - -std::chrono::microseconds -ThreadWorker::next_timeout() { - return std::chrono::microseconds(10min); -}