From b635f8c8521f138b94dc6897fc066f9bd2202c99 Mon Sep 17 00:00:00 2001 From: rakshasa Date: Sat, 12 Dec 2009 21:30:52 +0000 Subject: [PATCH] * More work on threading. git-svn-id: svn://rakshasa.no/libtorrent/trunk/rtorrent@1112 e378c898-3ddf-0310-93e7-cc216c733640 --- src/Makefile.am | 2 ++ src/command_events.cc | 11 ++++++++ src/control.cc | 4 +-- src/core/curl_socket.cc | 14 +++++----- src/globals.cc | 4 +-- src/globals.h | 4 +-- src/main.cc | 11 ++++---- src/rpc/scgi.cc | 12 ++++----- src/rpc/scgi_task.cc | 18 ++++++------- src/thread_base.cc | 16 +++-------- src/thread_base.h | 6 ++--- src/thread_main.cc | 52 +++++++++++++++++++++++++++++++++++ src/thread_main.h | 55 +++++++++++++++++++++++++++++++++++++ src/thread_worker.cc | 60 +++++++++++++++++++++++++++++++++++++++++ src/thread_worker.h | 57 +++++++++++++++++++++++++++++++++++++++ 15 files changed, 278 insertions(+), 48 deletions(-) create mode 100644 src/thread_main.cc create mode 100644 src/thread_main.h create mode 100644 src/thread_worker.cc create mode 100644 src/thread_worker.h diff --git a/src/Makefile.am b/src/Makefile.am index 90a4c13b..9ffe9517 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -41,6 +41,8 @@ rtorrent_SOURCES = \ signal_handler.h \ thread_base.cc \ thread_base.h \ + thread_main.cc \ + thread_main.h \ thread_worker.cc \ thread_worker.h diff --git a/src/command_events.cc b/src/command_events.cc index b25dfbc2..c5bcb84e 100644 --- a/src/command_events.cc +++ b/src/command_events.cc @@ -59,6 +59,8 @@ #include "control.h" #include "command_helpers.h" +#include "thread_worker.h" + torrent::Object apply_on_state_change(const char* name, const torrent::Object& rawArgs) { const torrent::Object::list_type& args = rawArgs.as_list(); @@ -350,8 +352,17 @@ d_multicall(const torrent::Object& rawArgs) { return resultRaw; } +torrent::Object +test_thread_locking(const torrent::Object& rawArgs) { + worker_thread->queue_item(&ThreadWorker::start_log_counter); + + return torrent::Object(); +} + void initialize_command_events() { + ADD_COMMAND_NONE("test.thread_locking", rak::ptr_fn(&test_thread_locking)); + ADD_VARIABLE_BOOL("check_hash", true); ADD_VARIABLE_BOOL("session_lock", true); diff --git a/src/control.cc b/src/control.cc index 66c2cdfb..01825de7 100644 --- a/src/control.cc +++ b/src/control.cc @@ -114,7 +114,7 @@ Control::initialize() { m_ui->init(this); - m_inputStdin->insert(this_thread->poll()); + m_inputStdin->insert(main_thread->poll()); } void @@ -124,7 +124,7 @@ Control::cleanup() { priority_queue_erase(&taskScheduler, &m_taskShutdown); - m_inputStdin->remove(this_thread->poll()); + m_inputStdin->remove(main_thread->poll()); m_core->download_store()->disable(); diff --git a/src/core/curl_socket.cc b/src/core/curl_socket.cc index 87d310d0..b8587edc 100644 --- a/src/core/curl_socket.cc +++ b/src/core/curl_socket.cc @@ -69,22 +69,22 @@ CurlSocket::receive_socket(void* easy_handle, curl_socket_t fd, int what, void* if (socket == NULL) { socket = stack->new_socket(fd); - this_thread->poll()->open(socket); + main_thread->poll()->open(socket); // No interface for libcurl to signal when it's interested in error events. // Assume that hence it must always be interested in them. - this_thread->poll()->insert_error(socket); + main_thread->poll()->insert_error(socket); } if (what == CURL_POLL_NONE || what == CURL_POLL_OUT) - this_thread->poll()->remove_read(socket); + main_thread->poll()->remove_read(socket); else - this_thread->poll()->insert_read(socket); + main_thread->poll()->insert_read(socket); if (what == CURL_POLL_NONE || what == CURL_POLL_IN) - this_thread->poll()->remove_write(socket); + main_thread->poll()->remove_write(socket); else - this_thread->poll()->insert_write(socket); + main_thread->poll()->insert_write(socket); return 0; } @@ -99,7 +99,7 @@ CurlSocket::close() { if (m_fileDesc == -1) throw torrent::internal_error("CurlSocket::close() m_fileDesc == -1."); - this_thread->poll()->closed(this); + main_thread->poll()->closed(this); m_fileDesc = -1; } diff --git a/src/globals.cc b/src/globals.cc index cd3c5fbb..f2247c68 100644 --- a/src/globals.cc +++ b/src/globals.cc @@ -42,6 +42,6 @@ rak::priority_queue_default taskScheduler; rak::timer cachedTime; Control* control = NULL; -//__thread ThreadBase* this_thread = NULL; -ThreadBase* this_thread = NULL; +//__thread ThreadBase* main_thread = NULL; +ThreadBase* main_thread = NULL; ThreadBase* worker_thread = NULL; diff --git a/src/globals.h b/src/globals.h index a1dd393f..d6940d6d 100644 --- a/src/globals.h +++ b/src/globals.h @@ -53,8 +53,8 @@ extern rak::priority_queue_default taskScheduler; extern rak::timer cachedTime; extern Control* control; -// extern __thread ThreadBase* this_thread; // Only use for worker threads for now. -extern ThreadBase* this_thread; +// extern __thread ThreadBase* main_thread; // Only use for worker threads for now. +extern ThreadBase* main_thread; extern ThreadBase* worker_thread; #endif diff --git a/src/main.cc b/src/main.cc index a59e4f31..40c03c40 100644 --- a/src/main.cc +++ b/src/main.cc @@ -69,6 +69,7 @@ #include "signal_handler.h" #include "option_parser.h" +#include "thread_main.h" #include "thread_worker.h" void do_panic(int signum); @@ -156,8 +157,8 @@ main(int argc, char** argv) { control = new Control; - this_thread = new ThreadBase(); - this_thread->init_thread(); + main_thread = new ThreadMain(); + main_thread->init_thread(); worker_thread = new ThreadWorker(); worker_thread->init_thread(); @@ -173,7 +174,7 @@ main(int argc, char** argv) { SignalHandler::set_handler(SIGBUS, sigc::bind(sigc::ptr_fun(&do_panic), SIGBUS)); SignalHandler::set_handler(SIGFPE, sigc::bind(sigc::ptr_fun(&do_panic), SIGFPE)); - torrent::initialize(this_thread->poll()); + torrent::initialize(main_thread->poll()); // Initialize option handlers after libtorrent to ensure // torrent::ConnectionManager* are valid etc. @@ -325,7 +326,7 @@ main(int argc, char** argv) { rak::priority_queue_perform(&taskScheduler, cachedTime); // Do shutdown check before poll, not after. - this_thread->poll_manager()->poll(client_next_timeout(control)); + main_thread->poll_manager()->poll(client_next_timeout(control)); } control->core()->download_list()->session_save(); @@ -339,7 +340,7 @@ main(int argc, char** argv) { } delete control; - delete this_thread; + delete main_thread; return 0; } diff --git a/src/rpc/scgi.cc b/src/rpc/scgi.cc index 61627342..63fc7648 100644 --- a/src/rpc/scgi.cc +++ b/src/rpc/scgi.cc @@ -118,16 +118,16 @@ SCgi::open(void* sa, unsigned int length) { void SCgi::activate() { - this_thread->poll()->open(this); - this_thread->poll()->insert_read(this); - this_thread->poll()->insert_error(this); + main_thread->poll()->open(this); + main_thread->poll()->insert_read(this); + main_thread->poll()->insert_error(this); } void SCgi::deactivate() { - this_thread->poll()->remove_read(this); - this_thread->poll()->remove_error(this); - this_thread->poll()->close(this); + main_thread->poll()->remove_read(this); + main_thread->poll()->remove_error(this); + main_thread->poll()->close(this); } void diff --git a/src/rpc/scgi_task.cc b/src/rpc/scgi_task.cc index 4f783354..ecff6d9a 100644 --- a/src/rpc/scgi_task.cc +++ b/src/rpc/scgi_task.cc @@ -75,9 +75,9 @@ SCgiTask::open(SCgi* parent, int fd) { m_position = m_buffer; m_body = NULL; - this_thread->poll()->open(this); - this_thread->poll()->insert_read(this); - this_thread->poll()->insert_error(this); + main_thread->poll()->open(this); + main_thread->poll()->insert_read(this); + main_thread->poll()->insert_error(this); // scgiTimer = rak::timer::current(); } @@ -87,10 +87,10 @@ SCgiTask::close() { if (!get_fd().is_valid()) return; - this_thread->poll()->remove_read(this); - this_thread->poll()->remove_write(this); - this_thread->poll()->remove_error(this); - this_thread->poll()->close(this); + main_thread->poll()->remove_read(this); + main_thread->poll()->remove_write(this); + main_thread->poll()->remove_error(this); + main_thread->poll()->close(this); get_fd().close(); get_fd().clear(); @@ -171,8 +171,8 @@ SCgiTask::event_read() { if ((unsigned int)std::distance(m_buffer, m_position) != m_bufferSize) return; - this_thread->poll()->remove_read(this); - this_thread->poll()->insert_write(this); + main_thread->poll()->remove_read(this); + main_thread->poll()->insert_write(this); if (m_parent->log_fd() >= 0) { // Clean up logging, this is just plain ugly... diff --git a/src/thread_base.cc b/src/thread_base.cc index 19049a05..eec69491 100644 --- a/src/thread_base.cc +++ b/src/thread_base.cc @@ -80,7 +80,7 @@ public: iterator itr = begin(); lock(); - while (*itr != NULL) *++dest = *++itr; + while (*itr != NULL) *dest++ = *itr++; clear_and_unlock(); return dest; @@ -111,14 +111,6 @@ ThreadBase::~ThreadBase() { // Cleanup... } -// Main thread init... Always replace this. -void -ThreadBase::init_thread() { - this_thread = this; - - m_pollManager = core::PollManager::create_poll_manager(); -} - void ThreadBase::start_thread() { if (m_state != STATE_INITIALIZED || @@ -144,7 +136,7 @@ ThreadBase::event_loop(ThreadBase* threadBase) { // // Remember to add global lock thing to the main poll loop ++. -// rak::priority_queue_perform(&threadBase->m_taskScheduler, cachedTime); + rak::priority_queue_perform(&threadBase->m_taskScheduler, cachedTime); threadBase->m_pollManager->poll_simple(rak::timer::from_seconds(10)); } @@ -158,8 +150,8 @@ ThreadBase::call_queued_items() { thread_base_func* first = result; thread_base_func* last = m_threadQueue->copy_and_clear((thread_base_func*)result); - while (first != last) - (*first)(this); + while (first != last && *first) + (*first++)(this); } void diff --git a/src/thread_base.h b/src/thread_base.h index 18903356..977209f3 100644 --- a/src/thread_base.h +++ b/src/thread_base.h @@ -49,7 +49,7 @@ struct thread_queue_hack; struct thread_queue_hack; -class __cacheline_aligned ThreadBase { +class ThreadBase { public: typedef rak::priority_queue_default priority_queue; typedef void (*thread_base_func)(ThreadBase*); @@ -62,13 +62,13 @@ public: }; ThreadBase(); - ~ThreadBase(); + virtual ~ThreadBase(); torrent::Poll* poll() { return m_pollManager->get_torrent_poll(); } core::PollManager* poll_manager() { return m_pollManager; } priority_queue& task_scheduler() { return m_taskScheduler; } - virtual void init_thread(); + virtual void init_thread() = 0; void start_thread(); void stop_thread(); diff --git a/src/thread_main.cc b/src/thread_main.cc new file mode 100644 index 00000000..ff2ee126 --- /dev/null +++ b/src/thread_main.cc @@ -0,0 +1,52 @@ +// rTorrent - BitTorrent library +// Copyright (C) 2005-2007, 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 "thread_main.h" +#include "globals.h" + +#include + +ThreadMain::~ThreadMain() { +} + +void +ThreadMain::init_thread() { + m_pollManager = core::PollManager::create_poll_manager(); + + m_state = STATE_INITIALIZED; +} diff --git a/src/thread_main.h b/src/thread_main.h new file mode 100644 index 00000000..00312a29 --- /dev/null +++ b/src/thread_main.h @@ -0,0 +1,55 @@ +// rTorrent - BitTorrent library +// Copyright (C) 2005-2007, 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 RTORRENT_THREAD_MAIN_H +#define RTORRENT_THREAD_MAIN_H + +#include "thread_base.h" + +// Check if cacheline aligned with inheritance ends up taking two +// cachelines. + +class __cacheline_aligned ThreadMain : public ThreadBase { +public: + ThreadMain() {} + ~ThreadMain(); + + virtual void init_thread(); + +private: +}; + +#endif diff --git a/src/thread_worker.cc b/src/thread_worker.cc new file mode 100644 index 00000000..feab0f03 --- /dev/null +++ b/src/thread_worker.cc @@ -0,0 +1,60 @@ +// rTorrent - BitTorrent library +// Copyright (C) 2005-2007, 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 "thread_worker.h" +#include "globals.h" + +#include +#include + +ThreadWorker::~ThreadWorker() { +} + +void +ThreadWorker::init_thread() { + m_pollManager = core::PollManager::create_poll_manager(); + + m_state = STATE_INITIALIZED; +} + +void +ThreadWorker::start_log_counter(ThreadBase* thread) { + assert(false); + + throw torrent::internal_error("PRRREREREFFFERERE"); +} diff --git a/src/thread_worker.h b/src/thread_worker.h new file mode 100644 index 00000000..3b068865 --- /dev/null +++ b/src/thread_worker.h @@ -0,0 +1,57 @@ +// rTorrent - BitTorrent library +// Copyright (C) 2005-2007, 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 RTORRENT_THREAD_WORKER_H +#define RTORRENT_THREAD_WORKER_H + +#include "thread_base.h" + +// Check if cacheline aligned with inheritance ends up taking two +// cachelines. + +class __cacheline_aligned ThreadWorker : public ThreadBase { +public: + ThreadWorker() {} + ~ThreadWorker(); + + virtual void init_thread(); + + static void start_log_counter(ThreadBase* thread); + +private: +}; + +#endif