* More work on threading.

git-svn-id: svn://rakshasa.no/libtorrent/trunk/rtorrent@1112 e378c898-3ddf-0310-93e7-cc216c733640

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
 

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

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

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;
 }
 

src/globals.cc

@@ -42,6 +42,6 @@ rak::priority_queue_default taskScheduler;
 rak::timer                  cachedTime;
 
 Control*                    control = NULL;
-//__thread ThreadBase* this_thread = NULL;
+//__thread ThreadBase* main_thread = NULL;
-ThreadBase* this_thread = NULL;
+ThreadBase* main_thread = NULL;
 ThreadBase* worker_thread = NULL;

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 __thread ThreadBase* main_thread; // Only use for worker threads for now.
-extern ThreadBase* this_thread;
+extern ThreadBase* main_thread;
 extern ThreadBase* worker_thread;
 
 #endif

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();
+    main_thread = new ThreadMain();
-    this_thread->init_thread();
+    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;
 }

src/rpc/scgi.cc

@@ -118,16 +118,16 @@ SCgi::open(void* sa, unsigned int length) {
 
 void
 SCgi::activate() {
-  this_thread->poll()->open(this);
+  main_thread->poll()->open(this);
-  this_thread->poll()->insert_read(this);
+  main_thread->poll()->insert_read(this);
-  this_thread->poll()->insert_error(this);
+  main_thread->poll()->insert_error(this);
 }
 
 void
 SCgi::deactivate() {
-  this_thread->poll()->remove_read(this);
+  main_thread->poll()->remove_read(this);
-  this_thread->poll()->remove_error(this);
+  main_thread->poll()->remove_error(this);
-  this_thread->poll()->close(this);
+  main_thread->poll()->close(this);
 }
 
 void

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);
+  main_thread->poll()->open(this);
-  this_thread->poll()->insert_read(this);
+  main_thread->poll()->insert_read(this);
-  this_thread->poll()->insert_error(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);
+  main_thread->poll()->remove_read(this);
-  this_thread->poll()->remove_write(this);
+  main_thread->poll()->remove_write(this);
-  this_thread->poll()->remove_error(this);
+  main_thread->poll()->remove_error(this);
-  this_thread->poll()->close(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);
+  main_thread->poll()->remove_read(this);
-  this_thread->poll()->insert_write(this);
+  main_thread->poll()->insert_write(this);
 
   if (m_parent->log_fd() >= 0) {
     // Clean up logging, this is just plain ugly...

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)
+  while (first != last && *first)
-    (*first)(this);
+    (*first++)(this);
 }
 
 void

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

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 <jaris@ifi.uio.no>
+//
+//           Skomakerveien 33
+//           3185 Skoppum, NORWAY
+
+#include "config.h"
+
+#include "thread_main.h"
+#include "globals.h"
+
+#include <torrent/exceptions.h>
+
+ThreadMain::~ThreadMain() {
+}
+
+void
+ThreadMain::init_thread() {
+  m_pollManager = core::PollManager::create_poll_manager();
+
+  m_state = STATE_INITIALIZED;
+}

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 <jaris@ifi.uio.no>
+//
+//           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

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 <jaris@ifi.uio.no>
+//
+//           Skomakerveien 33
+//           3185 Skoppum, NORWAY
+
+#include "config.h"
+
+#include "thread_worker.h"
+#include "globals.h"
+
+#include <cassert>
+#include <torrent/exceptions.h>
+
+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");
+}

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 <jaris@ifi.uio.no>
+//
+//           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