From 588f9e80c79e5cd762f6078f0cd7f05496d14dce Mon Sep 17 00:00:00 2001 From: rakshasa Date: Mon, 25 Jan 2010 13:14:09 +0000 Subject: [PATCH] * Threaded XMLRPC support. Sponsored by Xirvik. git-svn-id: svn://rakshasa.no/libtorrent/trunk/rtorrent@1123 e378c898-3ddf-0310-93e7-cc216c733640 --- src/command_local.cc | 10 +++++----- src/command_network.cc | 13 ++++++------- src/control.cc | 13 +++++-------- src/control.h | 17 +++++------------ src/core/poll_manager_select.cc | 4 ---- src/globals.cc | 4 ++-- src/globals.h | 6 ++++-- src/rpc/scgi.cc | 20 +++++++++++++------- src/rpc/scgi.h | 10 ++++++---- src/rpc/scgi_task.cc | 23 ++++++++++++----------- src/thread_worker.cc | 28 ++++++++++++++++++++++++++++ src/thread_worker.h | 18 +++++++++++++++++- 12 files changed, 103 insertions(+), 63 deletions(-) diff --git a/src/command_local.cc b/src/command_local.cc index 3de02a76..4d14870a 100644 --- a/src/command_local.cc +++ b/src/command_local.cc @@ -67,10 +67,10 @@ apply_log(int logType, const torrent::Object& rawArgs) { switch (logType) { case 0: ::close(rpc::execFile.log_fd()); rpc::execFile.set_log_fd(-1); break; case 1: - if (control->scgi()) { - ::close(control->scgi()->log_fd()); - control->scgi()->set_log_fd(-1); - } +// if (control->scgi()) { +// ::close(control->scgi()->log_fd()); +// control->scgi()->set_log_fd(-1); +// } break; default: break; } @@ -84,7 +84,7 @@ apply_log(int logType, const torrent::Object& rawArgs) { switch (logType) { case 0: rpc::execFile.set_log_fd(logFd); break; - case 1: if (control->scgi()) control->scgi()->set_log_fd(logFd); break; +// case 1: if (control->scgi()) control->scgi()->set_log_fd(logFd); break; default: break; } diff --git a/src/command_network.cc b/src/command_network.cc index 34f75ab1..1173bff2 100644 --- a/src/command_network.cc +++ b/src/command_network.cc @@ -324,14 +324,13 @@ initialize_xmlrpc() { void apply_scgi(const std::string& arg, int type) { - if (control->scgi() != NULL) + if (worker_thread->scgi() != NULL) throw torrent::input_error("SCGI already enabled."); if (!rpc::xmlrpc.is_valid()) initialize_xmlrpc(); - // Fix this... - control->set_scgi(new rpc::SCgi); + rpc::SCgi* scgi = new rpc::SCgi; rak::address_info* ai = NULL; rak::socket_address sa; @@ -366,13 +365,13 @@ apply_scgi(const std::string& arg, int type) { throw torrent::input_error("Invalid port number."); saPtr->set_port(port); - control->scgi()->open_port(saPtr, saPtr->length(), rpc::call_command_value("get_scgi_dont_route")); + scgi->open_port(saPtr, saPtr->length(), rpc::call_command_value("get_scgi_dont_route")); break; case 2: default: - control->scgi()->open_named(rak::path_expand(arg)); + scgi->open_named(rak::path_expand(arg)); break; } @@ -381,11 +380,11 @@ apply_scgi(const std::string& arg, int type) { } catch (torrent::local_error& e) { if (ai != NULL) rak::address_info::free_address_info(ai); + delete scgi; throw torrent::input_error(e.what()); } - control->scgi()->set_slot_process(rak::mem_fn(&rpc::xmlrpc, &rpc::XmlRpc::process)); - control->scgi()->activate(); + worker_thread->set_scgi(scgi); } void diff --git a/src/control.cc b/src/control.cc index 652a25dd..dd06aecd 100644 --- a/src/control.cc +++ b/src/control.cc @@ -58,9 +58,6 @@ #include "control.h" Control::Control() : - m_shutdownReceived(false), - m_shutdownQuick(false), - m_ui(new ui::Root()), m_display(new display::Manager()), m_input(new input::Manager()), @@ -68,9 +65,9 @@ Control::Control() : m_commandScheduler(new rpc::CommandScheduler()), - m_scgi(NULL), - - m_tick(0) { + m_tick(0), + m_shutdownReceived(false), + m_shutdownQuick(false) { m_core = new core::Manager(); m_viewManager = new core::ViewManager(); @@ -119,7 +116,7 @@ Control::initialize() { void Control::cleanup() { - delete m_scgi; m_scgi = NULL; + // delete m_scgi; m_scgi = NULL; rpc::xmlrpc.cleanup(); priority_queue_erase(&taskScheduler, &m_taskShutdown); @@ -139,7 +136,7 @@ Control::cleanup() { void Control::cleanup_exception() { - delete m_scgi; m_scgi = NULL; + // delete m_scgi; m_scgi = NULL; display::Canvas::cleanup(); } diff --git a/src/control.h b/src/control.h index 4b08c28a..47406912 100644 --- a/src/control.h +++ b/src/control.h @@ -64,8 +64,6 @@ namespace input { namespace rpc { class CommandScheduler; - class FastCgi; - class SCgi; class XmlRpc; } @@ -84,8 +82,8 @@ public: void handle_shutdown(); - void receive_normal_shutdown() { m_shutdownReceived = true; } - void receive_quick_shutdown() { m_shutdownReceived = true; m_shutdownQuick = true; } + void receive_normal_shutdown() { m_shutdownReceived = true; __sync_synchronize(); } + void receive_quick_shutdown() { m_shutdownReceived = true; m_shutdownQuick = true; __sync_synchronize(); } core::Manager* core() { return m_core; } core::ViewManager* view_manager() { return m_viewManager; } @@ -99,9 +97,6 @@ public: rpc::CommandScheduler* command_scheduler() { return m_commandScheduler; } - rpc::SCgi* scgi() { return m_scgi; } - void set_scgi(rpc::SCgi* f) { m_scgi = f; } - uint64_t tick() const { return m_tick; } void inc_tick() { m_tick++; } @@ -112,9 +107,6 @@ private: Control(const Control&); void operator = (const Control&); - bool m_shutdownReceived; - bool m_shutdownQuick; - core::Manager* m_core; core::ViewManager* m_viewManager; core::DhtManager* m_dhtManager; @@ -126,14 +118,15 @@ private: rpc::CommandScheduler* m_commandScheduler; - rpc::SCgi* m_scgi; - uint64_t m_tick; mode_t m_umask; std::string m_workingDirectory; rak::priority_item m_taskShutdown; + + bool m_shutdownReceived lt_cacheline_aligned; + bool m_shutdownQuick lt_cacheline_aligned; }; #endif diff --git a/src/core/poll_manager_select.cc b/src/core/poll_manager_select.cc index cc658e64..297c45fe 100644 --- a/src/core/poll_manager_select.cc +++ b/src/core/poll_manager_select.cc @@ -111,10 +111,6 @@ PollManagerSelect::poll_simple(rak::timer timeout) { std::memset(m_readSet, 0, 3 * m_setSize); unsigned int maxFd = currentPoll->fdset(m_readSet, m_writeSet, m_errorSet); - // unsigned int maxFd = 0; - - if ((unsigned int)std::count((char*)m_readSet, (char*)m_readSet + 3 * m_setSize, 0) != 3 * m_setSize) - throw torrent::internal_error("Got stray bits set."); timeval t = timeout.tval(); diff --git a/src/globals.cc b/src/globals.cc index f2247c68..0d7ff709 100644 --- a/src/globals.cc +++ b/src/globals.cc @@ -43,5 +43,5 @@ rak::timer cachedTime; Control* control = NULL; //__thread ThreadBase* main_thread = NULL; -ThreadBase* main_thread = NULL; -ThreadBase* worker_thread = NULL; +ThreadMain* main_thread = NULL; +ThreadWorker* worker_thread = NULL; diff --git a/src/globals.h b/src/globals.h index d6940d6d..240af6d8 100644 --- a/src/globals.h +++ b/src/globals.h @@ -41,6 +41,8 @@ #include #include "thread_base.h" +#include "thread_main.h" +#include "thread_worker.h" class Control; @@ -54,7 +56,7 @@ extern rak::timer cachedTime; extern Control* control; // extern __thread ThreadBase* main_thread; // Only use for worker threads for now. -extern ThreadBase* main_thread; -extern ThreadBase* worker_thread; +extern ThreadMain* main_thread; +extern ThreadWorker* worker_thread; #endif diff --git a/src/rpc/scgi.cc b/src/rpc/scgi.cc index 63fc7648..19bc2dee 100644 --- a/src/rpc/scgi.cc +++ b/src/rpc/scgi.cc @@ -49,6 +49,7 @@ #include "control.h" #include "globals.h" #include "scgi.h" +#include "parse_commands.h" namespace rpc { @@ -118,16 +119,16 @@ SCgi::open(void* sa, unsigned int length) { void SCgi::activate() { - main_thread->poll()->open(this); - main_thread->poll()->insert_read(this); - main_thread->poll()->insert_error(this); + worker_thread->poll()->open(this); + worker_thread->poll()->insert_read(this); + worker_thread->poll()->insert_error(this); } void SCgi::deactivate() { - main_thread->poll()->remove_read(this); - main_thread->poll()->remove_error(this); - main_thread->poll()->close(this); + worker_thread->poll()->remove_read(this); + worker_thread->poll()->remove_error(this); + worker_thread->poll()->close(this); } void @@ -163,7 +164,12 @@ SCgi::receive_call(SCgiTask* task, const char* buffer, uint32_t length) { slot_write slotWrite; slotWrite.set(rak::mem_fn(task, &SCgiTask::receive_write)); - return m_slotProcess(buffer, length, slotWrite); + ThreadBase::acquire_global_lock(); + // bool result = m_slotProcess(buffer, length, slotWrite); + bool result = xmlrpc.process(buffer, length, slotWrite); + ThreadBase::release_global_lock(); + + return result; } } diff --git a/src/rpc/scgi.h b/src/rpc/scgi.h index f0fa9e36..9f39f220 100644 --- a/src/rpc/scgi.h +++ b/src/rpc/scgi.h @@ -49,13 +49,14 @@ namespace utils { namespace rpc { -class SCgi : public torrent::Event { +class lt_cacheline_aligned SCgi : public torrent::Event { public: typedef rak::function2 slot_write; - typedef rak::function3 slot_process; +// typedef rak::function3 slot_process; static const int max_tasks = 10; + // Global lock: SCgi() : m_logFd(-1) {} virtual ~SCgi(); @@ -67,11 +68,12 @@ public: const std::string& path() const { return m_path; } - void set_slot_process(slot_process::base_type* s) { m_slotProcess.set(s); } +// void set_slot_process(slot_process::base_type* s) { m_slotProcess.set(s); } int log_fd() const { return m_logFd; } void set_log_fd(int fd) { m_logFd = fd; } + // Thread local: virtual void event_read(); virtual void event_write(); virtual void event_error(); @@ -85,7 +87,7 @@ private: std::string m_path; int m_logFd; - slot_process m_slotProcess; +// slot_process m_slotProcess; SCgiTask m_task[max_tasks]; }; diff --git a/src/rpc/scgi_task.cc b/src/rpc/scgi_task.cc index ecff6d9a..f7d664a6 100644 --- a/src/rpc/scgi_task.cc +++ b/src/rpc/scgi_task.cc @@ -36,6 +36,7 @@ #include "config.h" +#include #include #include #include @@ -60,7 +61,7 @@ namespace rpc { // If bufferSize is zero then memcpy won't do anything. inline void SCgiTask::realloc_buffer(uint32_t size, const char* buffer, uint32_t bufferSize) { - char* tmp = new char[size]; + char* tmp = rak::cacheline_allocator::alloc_size(size); std::memcpy(tmp, buffer, bufferSize); delete [] m_buffer; @@ -71,13 +72,13 @@ void SCgiTask::open(SCgi* parent, int fd) { m_parent = parent; m_fileDesc = fd; - m_buffer = new char[(m_bufferSize = default_buffer_size) + 1]; + m_buffer = rak::cacheline_allocator::alloc_size((m_bufferSize = default_buffer_size) + 1); m_position = m_buffer; m_body = NULL; - main_thread->poll()->open(this); - main_thread->poll()->insert_read(this); - main_thread->poll()->insert_error(this); + worker_thread->poll()->open(this); + worker_thread->poll()->insert_read(this); + worker_thread->poll()->insert_error(this); // scgiTimer = rak::timer::current(); } @@ -87,10 +88,10 @@ SCgiTask::close() { if (!get_fd().is_valid()) return; - main_thread->poll()->remove_read(this); - main_thread->poll()->remove_write(this); - main_thread->poll()->remove_error(this); - main_thread->poll()->close(this); + worker_thread->poll()->remove_read(this); + worker_thread->poll()->remove_write(this); + worker_thread->poll()->remove_error(this); + worker_thread->poll()->close(this); get_fd().close(); get_fd().clear(); @@ -171,8 +172,8 @@ SCgiTask::event_read() { if ((unsigned int)std::distance(m_buffer, m_position) != m_bufferSize) return; - main_thread->poll()->remove_read(this); - main_thread->poll()->insert_write(this); + worker_thread->poll()->remove_read(this); + worker_thread->poll()->insert_write(this); if (m_parent->log_fd() >= 0) { // Clean up logging, this is just plain ugly... diff --git a/src/thread_worker.cc b/src/thread_worker.cc index 073ab9ee..62b8d8d0 100644 --- a/src/thread_worker.cc +++ b/src/thread_worker.cc @@ -44,6 +44,9 @@ #include #include "core/manager.h" +#include "rpc/scgi.h" +#include "rpc/xmlrpc.h" +#include "rpc/parse_commands.h" ThreadWorker::ThreadWorker() { m_taskTouchLog.set_slot(rak::mem_fn(this, &ThreadWorker::task_touch_log)); @@ -59,6 +62,31 @@ ThreadWorker::init_thread() { m_state = STATE_INITIALIZED; } +bool +ThreadWorker::set_scgi(rpc::SCgi* scgi) { + if (!__sync_bool_compare_and_swap(&m_safe.scgi, NULL, scgi)) + return false; + + // The xmlrpc process call requires a global lock. +// m_safe.scgi->set_slot_process(rak::mem_fn(&rpc::xmlrpc, &rpc::XmlRpc::process)); + + // Synchronize in order to ensure the worker thread sees the updated + // SCgi object. + __sync_synchronize(); + queue_item((thread_base_func)&start_scgi); + return true; +} + +void +ThreadWorker::start_scgi(ThreadBase* baseThread) { + ThreadWorker* thread = (ThreadWorker*)baseThread; + + if (thread->m_safe.scgi == NULL) + throw torrent::internal_error("Tried to start SCGI but object was not present."); + + thread->m_safe.scgi->activate(); +} + void ThreadWorker::start_log_counter(ThreadBase* baseThread) { ThreadWorker* thread = (ThreadWorker*)baseThread; diff --git a/src/thread_worker.h b/src/thread_worker.h index c04d5157..6d7e0412 100644 --- a/src/thread_worker.h +++ b/src/thread_worker.h @@ -41,6 +41,10 @@ #include +namespace rpc { +class SCgi; +} + // Check if cacheline aligned with inheritance ends up taking two // cachelines. @@ -51,12 +55,24 @@ public: virtual void init_thread(); - static void start_log_counter(ThreadBase* thread); + rpc::SCgi* scgi() { return m_safe.scgi; } + bool set_scgi(rpc::SCgi* scgi); + + static void start_scgi(ThreadBase* thread); + static void start_log_counter(ThreadBase* thread); private: void task_touch_log(); rak::priority_item m_taskTouchLog; + + struct lt_cacheline_aligned safe_type { + safe_type() : scgi(NULL) {} + + rpc::SCgi* scgi; + }; + + safe_type m_safe; }; #endif