* 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

@@ -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()) {
+//       if (control->scgi()) {
-        ::close(control->scgi()->log_fd());
+//         ::close(control->scgi()->log_fd());
-        control->scgi()->set_log_fd(-1);
+//         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;
     }
 

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...
+  rpc::SCgi* scgi = new rpc::SCgi;
-  control->set_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));
+  worker_thread->set_scgi(scgi);
-  control->scgi()->activate();
 }
 
 void

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_shutdownReceived(false),
-  m_tick(0) {
+  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();
 }

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_normal_shutdown()     { m_shutdownReceived = true; __sync_synchronize(); }
-  void                receive_quick_shutdown()      { m_shutdownReceived = true; m_shutdownQuick = true; }
+  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

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();
 

src/globals.cc

@@ -43,5 +43,5 @@ rak::timer                  cachedTime;
 
 Control*                    control = NULL;
 //__thread ThreadBase* main_thread = NULL;
-ThreadBase* main_thread = NULL;
+ThreadMain* main_thread = NULL;
-ThreadBase* worker_thread = NULL;
+ThreadWorker* worker_thread = NULL;

src/globals.h

@@ -41,6 +41,8 @@
 #include <rak/priority_queue_default.h>
 
 #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 ThreadMain*   main_thread;
-extern ThreadBase* worker_thread;
+extern ThreadWorker* worker_thread;
 
 #endif

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);
+  worker_thread->poll()->open(this);
-  main_thread->poll()->insert_read(this);
+  worker_thread->poll()->insert_read(this);
-  main_thread->poll()->insert_error(this);
+  worker_thread->poll()->insert_error(this);
 }
 
 void
 SCgi::deactivate() {
-  main_thread->poll()->remove_read(this);
+  worker_thread->poll()->remove_read(this);
-  main_thread->poll()->remove_error(this);
+  worker_thread->poll()->remove_error(this);
-  main_thread->poll()->close(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;
 }
 
 }

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<bool, const char*, uint32_t>             slot_write;
-  typedef rak::function3<bool, const char*, uint32_t, slot_write> slot_process;
+//   typedef rak::function3<bool, const char*, uint32_t, slot_write> 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];
 };
 

src/rpc/scgi_task.cc

@@ -36,6 +36,7 @@
 
 #include "config.h"
 
+#include <rak/allocators.h>
 #include <rak/error_number.h>
 #include <cstdio>
 #include <sys/types.h>
@@ -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<char>::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<char>::alloc_size((m_bufferSize = default_buffer_size) + 1);
   m_position = m_buffer;
   m_body     = NULL;
 
-  main_thread->poll()->open(this);
+  worker_thread->poll()->open(this);
-  main_thread->poll()->insert_read(this);
+  worker_thread->poll()->insert_read(this);
-  main_thread->poll()->insert_error(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);
+  worker_thread->poll()->remove_read(this);
-  main_thread->poll()->remove_write(this);
+  worker_thread->poll()->remove_write(this);
-  main_thread->poll()->remove_error(this);
+  worker_thread->poll()->remove_error(this);
-  main_thread->poll()->close(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);
+  worker_thread->poll()->remove_read(this);
-  main_thread->poll()->insert_write(this);
+  worker_thread->poll()->insert_write(this);
 
   if (m_parent->log_fd() >= 0) {
     // Clean up logging, this is just plain ugly...

src/thread_worker.cc

@@ -44,6 +44,9 @@
 #include <torrent/exceptions.h>
 
 #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;

src/thread_worker.h

@@ -41,6 +41,10 @@
 
 #include <rak/priority_queue_default.h>
 
+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