fujin/rtorrent
1
0
Fork 0
mirror of https://github.com/rakshasa/rtorrent.git synced 2026-08-12 21:22:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Various scheduler and therad cleanups.

Browse Source
This commit is contained in:
Jari Sundell
2025-05-01 09:38:49 +02:00
committed by GitHub
parent 3d9c083032
commit 97415c656b
10 changed files with 55 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Makefile.am
-2
View File
@@ -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
src/command_ui.cc
+2 -1
View File
@@ -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
src/control.cc
+3 -9
View File
@@ -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);
}
src/globals.h
-37
View File
@@ -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 <jaris@ifi.uio.no>
//
// Skomakerveien 33
// 3185 Skoppum, NORWAY
#ifndef TORRENT_GLOBALS_H
#define TORRENT_GLOBALS_H
#include <rak/timer.h>
#include <rak/priority_queue_default.h>
#include "thread_base.h"
#include "thread_worker.h"
#include "rpc/ip_table_list.h"
src/rpc/exec_file.cc
+5 -41
View File
@@ -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 <jaris@ifi.uio.no>
//
// Skomakerveien 33
// 3185 Skoppum, NORWAY
#include "config.h"
#include <fcntl.h>
@@ -43,16 +7,16 @@
#include <rak/path.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <torrent/utils/thread.h>
#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.");
src/rpc/jsonrpc.cc
-1
View File
@@ -7,7 +7,6 @@
#include <torrent/common.h>
#include <torrent/torrent.h>
#include "thread_base.h"
#include "rpc/rpc_manager.h"
#include "rpc/command.h"
#include "rpc/command_map.h"
src/thread_base.cc
-56
View File
@@ -1,56 +0,0 @@
#include "config.h"
#include "thread_base.h"
#include <cassert>
#include <cstdlib>
#include <cstring>
#include <mutex>
#include <rak/error_number.h>
#include <torrent/exceptions.h>
#include <torrent/torrent.h>
#include <torrent/utils/chrono.h>
#include <torrent/utils/log.h>
#include <unistd.h>
#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<thread_queue_hack>
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();
}
src/thread_base.h
-51
View File
@@ -1,51 +0,0 @@
#ifndef RTORRENT_UTILS_THREAD_BASE_H
#define RTORRENT_UTILS_THREAD_BASE_H
#include <memory>
#include <pthread.h>
#include <sys/types.h>
#include <torrent/utils/thread.h>
#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<thread_queue_hack> m_threadQueue;
};
#endif
src/thread_worker.cc
+35 -31
View File
@@ -1,8 +1,6 @@
#include "config.h"
#include "thread_worker.h"
#include "globals.h"
#include "control.h"
#include <fcntl.h>
#include <unistd.h>
@@ -11,12 +9,14 @@
#include <torrent/exceptions.h>
#include <torrent/poll.h>
#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);
}
src/thread_worker.h
+10 -18
View File
@@ -2,44 +2,36 @@
#define RTORRENT_THREAD_WORKER_H
#include <atomic>
#include "thread_base.h"
#include "rak/priority_queue_default.h"
#include <string>
#include <torrent/utils/thread.h>
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<rpc::SCgi*> 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<rpc::SCgi*> m_scgi{nullptr};
std::string m_rpc_log_filename;
};
#endif