From 97415c656ba483cd902730b586afffcf7d762f2c Mon Sep 17 00:00:00 2001 From: Jari Sundell Date: Thu, 1 May 2025 09:38:49 +0200 Subject: [PATCH] Various scheduler and therad cleanups. --- src/Makefile.am | 2 -- src/command_ui.cc | 3 +- src/control.cc | 12 ++------ src/globals.h | 37 ------------------------- src/rpc/exec_file.cc | 46 ++++-------------------------- src/rpc/jsonrpc.cc | 1 - src/thread_base.cc | 56 ------------------------------------- src/thread_base.h | 51 ---------------------------------- src/thread_worker.cc | 66 +++++++++++++++++++++++--------------------- src/thread_worker.h | 28 +++++++------------ 10 files changed, 55 insertions(+), 247 deletions(-) delete mode 100644 src/thread_base.cc delete mode 100644 src/thread_base.h diff --git a/src/Makefile.am b/src/Makefile.am index d9e93d60..f6861c9f 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -200,8 +200,6 @@ libsub_root_a_SOURCES = \ option_parser.h \ signal_handler.cc \ signal_handler.h \ - thread_base.cc \ - thread_base.h \ thread_worker.cc \ thread_worker.h diff --git a/src/command_ui.cc b/src/command_ui.cc index 86f99b29..d92e9119 100644 --- a/src/command_ui.cc +++ b/src/command_ui.cc @@ -95,7 +95,8 @@ apply_view_set(const torrent::Object::list_type& args) { // if (args.front().as_string() == "main") // control->ui()->download_list()->set_view(*itr); // else - throw torrent::input_error("No such target."); + + throw torrent::input_error("No such target."); } torrent::Object diff --git a/src/control.cc b/src/control.cc index d3f3e79a..3cc20803 100644 --- a/src/control.cc +++ b/src/control.cc @@ -96,7 +96,6 @@ Control::initialize() { void Control::cleanup() { - // delete m_scgi; m_scgi = NULL; rpc::rpc.cleanup(); priority_queue_erase(&taskScheduler, &m_taskShutdown); @@ -118,8 +117,6 @@ Control::cleanup() { void Control::cleanup_exception() { - // delete m_scgi; m_scgi = NULL; - display::Canvas::cleanup(); } @@ -143,11 +140,10 @@ Control::handle_shutdown() { if (!m_shutdownQuick) { if (worker_thread->is_active()) - worker_thread->callback(nullptr, []() { - worker_thread->queue_stop_thread(); - }); + worker_thread->stop_thread(); torrent::connection_manager()->listen_close(); + m_directory_events->close(); m_core->shutdown(false); @@ -156,9 +152,7 @@ Control::handle_shutdown() { } else { if (worker_thread->is_active()) - worker_thread->callback(nullptr, []() { - worker_thread->queue_stop_thread(); - }); + worker_thread->stop_thread_wait(); m_core->shutdown(true); } diff --git a/src/globals.h b/src/globals.h index 69d01f3f..ed4adef1 100644 --- a/src/globals.h +++ b/src/globals.h @@ -1,46 +1,9 @@ -// rTorrent - BitTorrent library -// Copyright (C) 2005-2011, Jari Sundell -// -// This program is free software; you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation; either version 2 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. -// -// You should have received a copy of the GNU General Public License -// along with this program; if not, write to the Free Software -// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -// -// In addition, as a special exception, the copyright holders give -// permission to link the code of portions of this program with the -// OpenSSL library under certain conditions as described in each -// individual source file, and distribute linked combinations -// including the two. -// -// You must obey the GNU General Public License in all respects for -// all of the code used other than OpenSSL. If you modify file(s) -// with this exception, you may extend this exception to your version -// of the file(s), but you are not obligated to do so. If you do not -// wish to do so, delete this exception statement from your version. -// If you delete this exception statement from all source files in the -// program, then also delete it here. -// -// Contact: Jari Sundell -// -// Skomakerveien 33 -// 3185 Skoppum, NORWAY - #ifndef TORRENT_GLOBALS_H #define TORRENT_GLOBALS_H #include #include -#include "thread_base.h" #include "thread_worker.h" #include "rpc/ip_table_list.h" diff --git a/src/rpc/exec_file.cc b/src/rpc/exec_file.cc index 18991439..d97f582c 100644 --- a/src/rpc/exec_file.cc +++ b/src/rpc/exec_file.cc @@ -1,39 +1,3 @@ -// rTorrent - BitTorrent client -// Copyright (C) 2005-2011, Jari Sundell -// -// This program is free software; you can redistribute it and/or modify -// it under the terms of the GNU General Public License as published by -// the Free Software Foundation; either version 2 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU General Public License for more details. -// -// You should have received a copy of the GNU General Public License -// along with this program; if not, write to the Free Software -// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA -// -// In addition, as a special exception, the copyright holders give -// permission to link the code of portions of this program with the -// OpenSSL library under certain conditions as described in each -// individual source file, and distribute linked combinations -// including the two. -// -// You must obey the GNU General Public License in all respects for -// all of the code used other than OpenSSL. If you modify file(s) -// with this exception, you may extend this exception to your version -// of the file(s), but you are not obligated to do so. If you do not -// wish to do so, delete this exception statement from your version. -// If you delete this exception statement from all source files in the -// program, then also delete it here. -// -// Contact: Jari Sundell -// -// Skomakerveien 33 -// 3185 Skoppum, NORWAY - #include "config.h" #include @@ -43,16 +7,16 @@ #include #include #include +#include #include "exec_file.h" #include "parse.h" -#include "thread_base.h" namespace rpc { const unsigned int ExecFile::max_args; const unsigned int ExecFile::buffer_size; - + const int ExecFile::flag_expand_tilde; const int ExecFile::flag_throw; const int ExecFile::flag_capture; @@ -98,7 +62,7 @@ ExecFile::execute(const char* file, char* const* argv, int flags) { if (detached_pid != 0) { if (m_logFd != -1) result = write(m_logFd, "\n--- Background task ---\n", sizeof("\n--- Background task ---\n")); - + _exit(0); } @@ -139,7 +103,7 @@ ExecFile::execute(const char* file, char* const* argv, int flags) { // We yield the global lock when waiting for the executed command to // finish so that XMLRPC and other threads can continue working. - ThreadBase::release_global_lock(); + torrent::utils::Thread::release_global_lock(); if (flags & flag_capture) { m_capture = std::string(); @@ -170,7 +134,7 @@ ExecFile::execute(const char* file, char* const* argv, int flags) { wpid = waitpid(childPid, &status, 0); } while (wpid == -1 && rak::error_number::current().value() == rak::error_number::e_intr); - ThreadBase::acquire_global_lock(); + torrent::utils::Thread::acquire_global_lock(); if (wpid != childPid) throw torrent::internal_error("ExecFile::execute(...) waitpid failed."); diff --git a/src/rpc/jsonrpc.cc b/src/rpc/jsonrpc.cc index 223742c3..a9fad266 100644 --- a/src/rpc/jsonrpc.cc +++ b/src/rpc/jsonrpc.cc @@ -7,7 +7,6 @@ #include #include -#include "thread_base.h" #include "rpc/rpc_manager.h" #include "rpc/command.h" #include "rpc/command_map.h" diff --git a/src/thread_base.cc b/src/thread_base.cc deleted file mode 100644 index 9d56d24e..00000000 --- a/src/thread_base.cc +++ /dev/null @@ -1,56 +0,0 @@ -#include "config.h" - -#include "thread_base.h" - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include "globals.h" - -void -throw_shutdown_exception() { throw torrent::shutdown_exception(); } - -ThreadBase::ThreadBase() { - m_taskShutdown.slot() = std::bind(&throw_shutdown_exception); -} - -// Defined in here and not the header so that the default destructor -// can properly deduce how to destruct the unique_ptr -ThreadBase::~ThreadBase() = default; - -// Move to libtorrent... -void -ThreadBase::queue_stop_thread() { - if (!m_taskShutdown.is_queued()) - priority_queue_insert(&m_taskScheduler, &m_taskShutdown, cachedTime); -} - -std::chrono::microseconds -ThreadBase::next_timeout() { - if (m_taskScheduler.empty()) - return std::chrono::microseconds(10min); - - cachedTime = rak::timer::current(); - - if (m_taskScheduler.top()->time() <= cachedTime) - return std::chrono::microseconds(0); - - return std::chrono::microseconds((m_taskScheduler.top()->time() - cachedTime).usec()); -} - -void -ThreadBase::call_events() { - cachedTime = rak::timer::current(); - - rak::priority_queue_perform(&m_taskScheduler, cachedTime); - - process_callbacks(); -} diff --git a/src/thread_base.h b/src/thread_base.h deleted file mode 100644 index b092d36e..00000000 --- a/src/thread_base.h +++ /dev/null @@ -1,51 +0,0 @@ -#ifndef RTORRENT_UTILS_THREAD_BASE_H -#define RTORRENT_UTILS_THREAD_BASE_H - -#include -#include -#include -#include - -#include "rak/priority_queue_default.h" -#include "core/poll_manager.h" - -// Move this class to libtorrent. - -class thread_queue_hack; - -class ThreadBase : public torrent::utils::Thread { -public: - typedef rak::priority_queue_default priority_queue; - typedef void (*thread_base_func)(ThreadBase*); - - ThreadBase(); - virtual ~ThreadBase(); - - priority_queue& task_scheduler() { return m_taskScheduler; } - - // Throw torrent::shutdown_exception to stop the thread. - void queue_stop_thread(); - - // ATM, only interaction with a thread's allowed by other threads is - // through the queue_item call. - - // void queue_item(thread_base_func newFunc); - -protected: - void call_events() override; - std::chrono::microseconds next_timeout() override; - - // TODO: Add thread name. - - // The timer needs to be sync'ed when updated... - - rak::priority_queue_default m_taskScheduler; - - rak::priority_item m_taskShutdown; - - // Temporary hack to pass messages to a thread. This really needs to - // be cleaned up and/or integrated into the priority queue itself. - // std::unique_ptr m_threadQueue; -}; - -#endif diff --git a/src/thread_worker.cc b/src/thread_worker.cc index f1e9d37d..7842adfa 100644 --- a/src/thread_worker.cc +++ b/src/thread_worker.cc @@ -1,8 +1,6 @@ #include "config.h" #include "thread_worker.h" -#include "globals.h" -#include "control.h" #include #include @@ -11,12 +9,14 @@ #include #include +#include "globals.h" +#include "control.h" #include "core/manager.h" #include "rpc/scgi.h" #include "rpc/parse_commands.h" ThreadWorker::~ThreadWorker() { - if (m_scgi) + if (m_scgi != nullptr) m_scgi.load()->deactivate(); } @@ -29,13 +29,17 @@ ThreadWorker::init_thread() { 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]() { - start_scgi(this); + if (m_scgi == NULL) + throw torrent::internal_error("Tried to start SCGI but object was not present."); + + m_scgi.load()->activate(); }); return true; @@ -43,32 +47,12 @@ ThreadWorker::set_scgi(rpc::SCgi* scgi) { void ThreadWorker::set_rpc_log(const std::string& filename) { - m_rpcLog = filename; - - callback(nullptr, [this]() { - msg_change_rpc_log(this); + callback(nullptr, [this, filename]() { + m_rpc_log_filename = filename; + change_rpc_log(); }); } -void -ThreadWorker::start_scgi(ThreadBase* baseThread) { - ThreadWorker* thread = (ThreadWorker*)baseThread; - - if (thread->scgi() == NULL) - throw torrent::internal_error("Tried to start SCGI but object was not present."); - - thread->scgi()->activate(); -} - -void -ThreadWorker::msg_change_rpc_log(ThreadBase* baseThread) { - ThreadWorker* thread = (ThreadWorker*)baseThread; - - acquire_global_lock(); - thread->change_rpc_log(); - release_global_lock(); -} - void ThreadWorker::change_rpc_log() { if (scgi() == NULL) @@ -80,15 +64,35 @@ ThreadWorker::change_rpc_log() { control->core()->push_log("Closed RPC log."); } - if (m_rpcLog.empty()) + if (m_rpc_log_filename.empty()) return; - scgi()->set_log_fd(open(rak::path_expand(m_rpcLog).c_str(), O_WRONLY | O_APPEND | O_CREAT, 0644)); + 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_rpcLog + "'."); + 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_rpcLog + "'."); + 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(); + } + + cachedTime = rak::timer::current(); + + process_callbacks(); +} + +std::chrono::microseconds +ThreadWorker::next_timeout() { + return std::chrono::microseconds(10min); } diff --git a/src/thread_worker.h b/src/thread_worker.h index dfb5820b..ff41a244 100644 --- a/src/thread_worker.h +++ b/src/thread_worker.h @@ -2,44 +2,36 @@ #define RTORRENT_THREAD_WORKER_H #include - -#include "thread_base.h" -#include "rak/priority_queue_default.h" +#include +#include namespace rpc { class SCgi; } -// Check if cacheline aligned with inheritance ends up taking two -// cachelines. - -class ThreadWorker : public ThreadBase { +class ThreadWorker : public torrent::utils::Thread { public: ThreadWorker() = default; ~ThreadWorker(); - const char* name() const { return "rtorrent scgi"; } + const char* name() const override { return "rtorrent scgi"; } - virtual void init_thread(); + void init_thread() override; - rpc::SCgi* scgi() { return m_scgi; } + rpc::SCgi* scgi() { return m_scgi; } bool set_scgi(rpc::SCgi* scgi); void set_rpc_log(const std::string& filename); - static void start_scgi(ThreadBase* thread); - static void msg_change_rpc_log(ThreadBase* thread); - private: void task_touch_log(); - void change_rpc_log(); - std::atomic m_scgi{ nullptr }; + void call_events() override; + std::chrono::microseconds next_timeout() override; - // The following types shall only be modified while holding the - // global lock. - std::string m_rpcLog; + std::atomic m_scgi{nullptr}; + std::string m_rpc_log_filename; }; #endif