* Created ThreadBase holding poll manager, etc, and added a this_thread global object.

* Cleaned up main initialization, SCGI and polling code.

* Fixed a bug that would cause reading of a piece to hang if the incoming data contains only data up to the file boundary, but not the next file's data. The bug did not trigger if the file boundary and piece boundaries were the same.


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

rak/timer.h

@@ -46,8 +46,7 @@ namespace rak {
 // Don't convert negative Timer to timeval and then back to Timer, that will bork.
 class timer {
  public:
-  timer() : m_time(0) {}
-  timer(int64_t usec) : m_time(usec) {}
+  timer(int64_t usec = 0) : m_time(usec) {}
   timer(timeval tv) : m_time((int64_t)(uint32_t)tv.tv_sec * 1000000 + (int64_t)(uint32_t)tv.tv_usec % 1000000) {}
 
   int32_t             seconds() const                    { return m_time / 1000000; }

scripts/common.m4

@@ -86,8 +86,9 @@ AC_DEFUN([TORRENT_ENABLE_ARCH], [
         AC_MSG_RESULT($enableval)
 
         for i in `IFS=,; echo $enableval`; do
-          CXXFLAGS="$CXXFLAGS -arch $i"
-          LDFLAGS="$LDFLAGS -arch $i"
+          CFLAGS="$CFLAGS -march=$i"
+          CXXFLAGS="$CXXFLAGS -march=$i"
+          LDFLAGS="$LDFLAGS -march=$i"
         done
       fi
     ])
@@ -183,7 +184,7 @@ AC_DEFUN([TORRENT_CHECK_MADVISE], [
 AC_DEFUN([TORRENT_CHECK_EXECINFO], [
   AC_MSG_CHECKING(for execinfo.h)
 
-  AC_LINK_IFELSE(
+  AC_RUN_IFELSE(
     [[#include <execinfo.h>
       int main() { backtrace((void**)0, 0); backtrace_symbols((char**)0, 0); return 0;}
     ]],
@@ -239,3 +240,18 @@ AC_DEFUN([TORRENT_DISABLE_IPV6], [
         fi
     ])
 ])
+
+AC_DEFUN([TORRENT_ENABLE_TR1], [
+  AC_ARG_ENABLE(std_tr1,
+    [  --disable-std_tr1       disable check for support for TR1 [[default=enable]]],
+    [
+      if test "$enableval" = "yes"; then
+        TORRENT_CHECK_TR1()
+      else
+        AC_MSG_CHECKING(for TR1 support)
+        AC_MSG_RESULT(disabled)
+      fi
+    ],[
+        TORRENT_CHECK_TR1()
+    ])
+])

src/command_network.cc

@@ -385,6 +385,7 @@ apply_scgi(const std::string& arg, int type) {
   }
 
   control->scgi()->set_slot_process(rak::mem_fn(&rpc::xmlrpc, &rpc::XmlRpc::process));
+  control->scgi()->activate();
 }
 
 void
@@ -492,6 +493,6 @@ initialize_command_network() {
   ADD_VARIABLE_BOOL("peer_exchange", true);
 
   // Not really network stuff:
-  ADD_VARIABLE_BOOL  ("log.handshake", false);
+  ADD_VARIABLE_BOOL  ("handshake_log", false);
   ADD_VARIABLE_STRING("log.tracker", "");
 }

src/control.cc

@@ -114,7 +114,7 @@ Control::initialize() {
 
   m_ui->init(this);
 
-  m_inputStdin->insert(m_core->get_poll_manager()->get_torrent_poll());
+  m_inputStdin->insert(this_thread->poll());
 }
 
 void
@@ -124,7 +124,7 @@ Control::cleanup() {
 
   priority_queue_erase(&taskScheduler, &m_taskShutdown);
 
-  m_inputStdin->remove(m_core->get_poll_manager()->get_torrent_poll());
+  m_inputStdin->remove(this_thread->poll());
 
   m_core->download_store()->disable();
 
@@ -161,7 +161,3 @@ Control::handle_shutdown() {
   m_shutdownReceived = false;
 }
 
-torrent::Poll*
-Control::poll() {
-  return m_core->get_poll_manager()->get_torrent_poll();
-}

src/control.h

@@ -91,7 +91,6 @@ public:
   core::ViewManager*  view_manager()                { return m_viewManager; }
   core::DhtManager*   dht_manager()                 { return m_dhtManager; }
 
-  torrent::Poll*      poll();
 
   ui::Root*           ui()                          { return m_ui; }
   display::Manager*   display()                     { return m_display; }

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);
-    control->poll()->open(socket);
+    this_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.
-    control->poll()->insert_error(socket);
+    this_thread->poll()->insert_error(socket);
   } 
 
   if (what == CURL_POLL_NONE || what == CURL_POLL_OUT)
-    control->poll()->remove_read(socket);
+    this_thread->poll()->remove_read(socket);
   else
-    control->poll()->insert_read(socket);
+    this_thread->poll()->insert_read(socket);
 
   if (what == CURL_POLL_NONE || what == CURL_POLL_IN)
-    control->poll()->remove_write(socket);
+    this_thread->poll()->remove_write(socket);
   else
-    control->poll()->insert_write(socket);
+    this_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.");
 
-  control->poll()->closed(this);
+  this_thread->poll()->closed(this);
   m_fileDesc = -1;
 }
 

src/core/manager.cc

@@ -158,10 +158,9 @@ Manager::push_log(const char* msg) {
 }
 
 Manager::Manager() :
-  m_hashingView(NULL),
-
-  m_pollManager(NULL) {
-
+  m_hashingView(NULL)
+//   m_pollManager(NULL) {
+{
   m_downloadStore   = new DownloadStore();
   m_downloadList    = new DownloadList();
   m_fileStatusCache = new FileStatusCache();
@@ -216,40 +215,6 @@ Manager::get_address_throttle(const sockaddr* addr) {
   return m_addressThrottles.get(rak::socket_address::cast_from(addr)->sa_inet()->address_h(), torrent::ThrottlePair(NULL, NULL));
 }
 
-void
-Manager::initialize_first() {
-  const char* poll = getenv("RTORRENT_POLL");
-  if (poll != NULL) {
-    if (!strcmp(poll, "epoll"))
-      m_pollManager = PollManagerEPoll::create(sysconf(_SC_OPEN_MAX));
-    else if (!strcmp(poll, "kqueue"))
-      m_pollManager = PollManagerKQueue::create(sysconf(_SC_OPEN_MAX));
-    else if (!strcmp(poll, "select"))
-      m_pollManager = PollManagerSelect::create(sysconf(_SC_OPEN_MAX));
-
-    if (m_pollManager == NULL)
-      m_logImportant.push_front(std::string("Cannot enable '") + poll + "' based polling.");
-  }
-
-  if (m_pollManager != NULL)
-      m_logImportant.push_front(std::string("Using '") + poll + "' based polling.");
-
-  else if ((m_pollManager = PollManagerEPoll::create(sysconf(_SC_OPEN_MAX))) != NULL)
-    m_logImportant.push_front("Using 'epoll' based polling.");
-
-  else if ((m_pollManager = PollManagerKQueue::create(sysconf(_SC_OPEN_MAX))) != NULL)
-    m_logImportant.push_front("Using 'kqueue' based polling.");
-
-  else if ((m_pollManager = PollManagerSelect::create(sysconf(_SC_OPEN_MAX))) != NULL)
-    m_logImportant.push_front("Using 'select' based polling.");
-
-  else
-    throw std::runtime_error("Could not create any PollManager.");
-
-  // Need to initialize this before parseing options.
-  torrent::initialize(m_pollManager->get_torrent_poll());
-}
-
 // Most of this should be possible to move out.
 void
 Manager::initialize_second() {
@@ -276,7 +241,6 @@ Manager::cleanup() {
   delete m_httpStack;
   CurlStack::global_cleanup();
 
-  delete m_pollManager;
 }
 
 void

src/core/manager.h

@@ -85,7 +85,6 @@ public:
   View*               hashing_view()                      { return m_hashingView; }
   void                set_hashing_view(View* v);
 
-  PollManager*        get_poll_manager()                  { return m_pollManager; }
   Log&                get_log_important()                 { return m_logImportant; }
   Log&                get_log_complete()                  { return m_logComplete; }
 
@@ -97,7 +96,6 @@ public:
   torrent::ThrottlePair get_address_throttle(const sockaddr* addr);
 
   // Really should find a more descriptive name.
-  void                initialize_first();
   void                initialize_second();
   void                cleanup();
 
@@ -153,7 +151,6 @@ private:
   ThrottleMap         m_throttles;
   AddressThrottleMap  m_addressThrottles;
 
-  PollManager*        m_pollManager;
   Log                 m_logImportant;
   Log                 m_logComplete;
 };

src/core/poll_manager.cc

@@ -39,7 +39,13 @@
 #include <stdexcept>
 #include <rak/error_number.h>
 
+#include "globals.h"
+#include "control.h"
+#include "manager.h"
 #include "poll_manager.h"
+#include "poll_manager_epoll.h"
+#include "poll_manager_kqueue.h"
+#include "poll_manager_select.h"
 
 namespace core {
 
@@ -54,6 +60,45 @@ PollManager::~PollManager() {
   delete m_poll;
 }
 
+PollManager*
+PollManager::create_poll_manager() {
+  PollManager* pollManager = NULL;
+  Log* log = &control->core()->get_log_important();
+
+  const char* poll = getenv("RTORRENT_POLL");
+
+  int maxOpen = sysconf(_SC_OPEN_MAX);
+
+  if (poll != NULL) {
+    if (!strcmp(poll, "epoll"))
+      pollManager = PollManagerEPoll::create(maxOpen);
+    else if (!strcmp(poll, "kqueue"))
+      pollManager = PollManagerKQueue::create(maxOpen);
+    else if (!strcmp(poll, "select"))
+      pollManager = PollManagerSelect::create(maxOpen);
+
+    if (pollManager == NULL)
+      log->push_front(std::string("Cannot enable '") + poll + "' based polling.");
+  }
+
+  if (pollManager != NULL)
+      log->push_front(std::string("Using '") + poll + "' based polling.");
+
+  else if ((pollManager = PollManagerEPoll::create(maxOpen)) != NULL)
+    log->push_front("Using 'epoll' based polling.");
+
+  else if ((pollManager = PollManagerKQueue::create(maxOpen)) != NULL)
+    log->push_front("Using 'kqueue' based polling.");
+
+  else if ((pollManager = PollManagerSelect::create(maxOpen)) != NULL)
+    log->push_front("Using 'select' based polling.");
+
+  else
+    throw std::runtime_error("Could not create any PollManager.");
+
+  return pollManager;
+}
+
 void
 PollManager::check_error() {
   if (rak::error_number::current().value() != rak::error_number::e_intr)

src/core/poll_manager.h

@@ -60,6 +60,9 @@ public:
   torrent::Poll*      get_torrent_poll()           { return m_poll; }
 
   virtual void        poll(rak::timer timeout) = 0;
+  virtual void        poll_simple(rak::timer timeout) = 0;
+
+  static PollManager* create_poll_manager();
 
 protected:
   PollManager(const PollManager&);

src/core/poll_manager_epoll.cc

@@ -74,4 +74,16 @@ PollManagerEPoll::poll(rak::timer timeout) {
   static_cast<torrent::PollEPoll*>(m_poll)->perform();
 }
 
+void
+PollManagerEPoll::poll_simple(rak::timer timeout) {
+  // Add 1ms to ensure we don't idle loop due to the lack of
+  // resolution.
+  timeout = timeout + 1000;
+
+  if (static_cast<torrent::PollEPoll*>(m_poll)->poll((timeout.usec() + 999) / 1000) == -1)
+    return check_error();
+
+  static_cast<torrent::PollEPoll*>(m_poll)->perform();
+}
+
 }

src/core/poll_manager_epoll.h

@@ -53,6 +53,7 @@ public:
   torrent::Poll*      get_torrent_poll();
 
   void                poll(rak::timer timeout);
+  void                poll_simple(rak::timer timeout);
 
 private:
   PollManagerEPoll(torrent::Poll* p) : PollManager(p) {}

src/core/poll_manager_kqueue.cc

@@ -75,4 +75,16 @@ PollManagerKQueue::poll(rak::timer timeout) {
   static_cast<torrent::PollKQueue*>(m_poll)->perform();
 }
 
+void
+PollManagerKQueue::poll_simple(rak::timer timeout) {
+  // Add 1ms to ensure we don't idle loop due to the lack of
+  // resolution.
+  timeout = std::min(timeout, rak::timer(torrent::next_timeout())) + 1000;
+
+  if (static_cast<torrent::PollKQueue*>(m_poll)->poll((timeout.usec() + 999) / 1000) == -1)
+    return check_error();
+
+  static_cast<torrent::PollKQueue*>(m_poll)->perform();
+}
+
 }

src/core/poll_manager_kqueue.h

@@ -53,6 +53,7 @@ public:
   torrent::Poll*      get_torrent_poll();
 
   void                poll(rak::timer timeout);
+  void                poll_simple(rak::timer timeout);
 
 private:
   PollManagerKQueue(torrent::Poll* p) : PollManager(p) {}

src/core/poll_manager_select.cc

@@ -120,4 +120,28 @@ PollManagerSelect::poll(rak::timer timeout) {
   static_cast<torrent::PollSelect*>(m_poll)->perform(m_readSet, m_writeSet, m_errorSet);
 }
 
+void
+PollManagerSelect::poll_simple(rak::timer timeout) {
+  timeout = timeout + 1000;
+
+#if defined USE_VARIABLE_FDSET
+  std::memset(m_readSet, 0, m_setSize);
+  std::memset(m_writeSet, 0, m_setSize);
+  std::memset(m_errorSet, 0, m_setSize);
+#else
+  FD_ZERO(m_readSet);
+  FD_ZERO(m_writeSet);
+  FD_ZERO(m_errorSet);
+#endif    
+
+  unsigned int maxFd = static_cast<torrent::PollSelect*>(m_poll)->fdset(m_readSet, m_writeSet, m_errorSet);
+
+  timeval t = timeout.tval();
+
+  if (select(maxFd + 1, m_readSet, m_writeSet, m_errorSet, &t) == -1)
+    return check_error();
+
+  static_cast<torrent::PollSelect*>(m_poll)->perform(m_readSet, m_writeSet, m_errorSet);
+}
+
 }

src/core/poll_manager_select.h

@@ -51,6 +51,7 @@ public:
   ~PollManagerSelect();
 
   void                poll(rak::timer timeout);
+  void                poll_simple(rak::timer timeout);
 
 private:
   PollManagerSelect(torrent::Poll* p);

src/globals.cc

@@ -42,3 +42,5 @@ rak::priority_queue_default taskScheduler;
 rak::timer                  cachedTime;
 
 Control*                    control = NULL;
+//__thread utils::ThreadBase* this_thread = NULL;
+utils::ThreadBase* this_thread = NULL;

src/globals.h

@@ -40,11 +40,20 @@
 #include <rak/timer.h>
 #include <rak/priority_queue_default.h>
 
+#include "utils/thread_base.h"
+
 class Control;
 
+// The cachedTime timer should only be updated by the main thread to
+// avoid potential problems in timing calculations. Code really should
+// be reviewed and fixed in order to avoid any potential problems, and
+// then made updates properly sync'ed with memory barriers.
+
 extern rak::priority_queue_default taskScheduler;
 extern rak::timer                  cachedTime;
 
 extern Control*                    control;
+// extern __thread utils::ThreadBase* this_thread; // Only use for worker threads for now.
+extern utils::ThreadBase* this_thread; // Only use for main threads for now.
 
 #endif

src/main.cc

@@ -154,6 +154,9 @@ main(int argc, char** argv) {
 
     control = new Control;
     
+    this_thread = new utils::ThreadBase();
+    this_thread->init_thread();
+
     srandom(cachedTime.usec());
     srand48(cachedTime.usec());
 
@@ -165,7 +168,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));
 
-    control->core()->initialize_first();
+    torrent::initialize(this_thread->poll());
 
     // Initialize option handlers after libtorrent to ensure
     // torrent::ConnectionManager* are valid etc.
@@ -315,7 +318,7 @@ main(int argc, char** argv) {
       rak::priority_queue_perform(&taskScheduler, cachedTime);
 
       // Do shutdown check before poll, not after.
-      control->core()->get_poll_manager()->poll(client_next_timeout(control));
+      this_thread->poll_manager()->poll(client_next_timeout(control));
     }
 
     control->core()->download_list()->session_save();
@@ -329,6 +332,7 @@ main(int argc, char** argv) {
   }
 
   delete control;
+  delete this_thread;
 
   return 0;
 }

src/rpc/scgi.cc

@@ -60,9 +60,8 @@ SCgi::~SCgi() {
     if (itr->is_open())
       itr->close();
 
-  control->poll()->remove_read(this);
-  control->poll()->remove_error(this);
-  control->poll()->close(this);
+  deactivate();
+  torrent::connection_manager()->dec_socket_count();
 
   get_fd().close();
   get_fd().clear();
@@ -109,10 +108,6 @@ SCgi::open(void* sa, unsigned int length) {
 
     torrent::connection_manager()->inc_socket_count();
 
-    control->poll()->open(this);
-    control->poll()->insert_read(this);
-    control->poll()->insert_error(this);
-
   } catch (torrent::resource_error& e) {
     get_fd().close();
     get_fd().clear();
@@ -121,6 +116,20 @@ 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);
+}
+
+void
+SCgi::deactivate() {
+  this_thread->poll()->remove_read(this);
+  this_thread->poll()->remove_error(this);
+  this_thread->poll()->close(this);
+}
+
 void
 SCgi::event_read() {
   rak::socket_address sa;

src/rpc/scgi.h

@@ -62,7 +62,10 @@ public:
   void                open_port(void* sa, unsigned int length, bool dontRoute);
   void                open_named(const std::string& filename);
 
-  const std::string   path() const { return m_path; }
+  void                activate();
+  void                deactivate();
+
+  const std::string&  path() const { return m_path; }
 
   void                set_slot_process(slot_process::base_type* s) { m_slotProcess.set(s); }
 

src/rpc/scgi_task.cc

@@ -75,9 +75,9 @@ SCgiTask::open(SCgi* parent, int fd) {
   m_position = m_buffer;
   m_body     = NULL;
 
-  control->poll()->open(this);
-  control->poll()->insert_read(this);
-  control->poll()->insert_error(this);
+  this_thread->poll()->open(this);
+  this_thread->poll()->insert_read(this);
+  this_thread->poll()->insert_error(this);
 
 //   scgiTimer = rak::timer::current();
 }
@@ -87,10 +87,10 @@ SCgiTask::close() {
   if (!get_fd().is_valid())
     return;
 
-  control->poll()->remove_read(this);
-  control->poll()->remove_write(this);
-  control->poll()->remove_error(this);
-  control->poll()->close(this);
+  this_thread->poll()->remove_read(this);
+  this_thread->poll()->remove_write(this);
+  this_thread->poll()->remove_error(this);
+  this_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;
 
-  control->poll()->remove_read(this);
-  control->poll()->insert_write(this);
+  this_thread->poll()->remove_read(this);
+  this_thread->poll()->insert_write(this);
 
   if (m_parent->log_fd() >= 0) {
     // Clean up logging, this is just plain ugly...

src/utils/Makefile.am

@@ -9,6 +9,8 @@ libsub_utils_a_SOURCES = \
 	lockfile.cc \
 	lockfile.h \
 	socket_fd.cc \
-	socket_fd.h
+	socket_fd.h \
+	thread_base.cc \
+	thread_base.h
 
 INCLUDES = -I$(srcdir) -I$(srcdir)/.. -I$(top_srcdir)

src/utils/thread_base.cc

@@ -0,0 +1,159 @@
+// libTorrent - 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_base.h"
+
+#include <cstdlib>
+#include <torrent/exceptions.h>
+
+#include "globals.h"
+
+namespace utils {
+
+// Temporarly injected into config.h.
+/* temp hack */
+//#define __cacheline_aligned __attribute__((__aligned__(128)))
+
+// class __cacheline_aligned thread_queue_hack {
+// public:
+//   typedef ThreadBase::thread_base_func value_type;
+//   typedef ThreadBase::thread_base_func* iterator;
+
+//   static const unsigned int max_size = 32;
+
+//   thread_queue_hack() { std::memset(m_queue, 0, sizeof(thread_queue_hack)); }
+
+//   void     lock()   { while (!__sync_bool_compare_and_swap(&m_lock, 0, 1)) usleep(0); }
+//   void     unlock() { __sync_bool_compare_and_swap(&m_lock, 1, 0); }
+
+//   iterator begin() { return m_queue; }
+//   iterator max_capacity() { return m_queue + max_size; }
+
+//   iterator end_and_lock() { lock(); return std::find(begin(), max_capacity(), (value_type)NULL); }
+
+//   bool     empty() const { return m_queue[0] == NULL; }
+
+//   void push_back(value_type v) {
+//     iterator itr = end_and_lock();
+  
+//     if (itr == max_capacity())
+//       throw torrent::internal_error("Overflowed thread_queue.");
+
+//     __sync_bool_compare_and_swap(itr, NULL, v);
+//     __sync_bool_compare_and_swap(&m_lock, 1, 0);
+//   }
+
+//   value_type* copy_and_clear(value_type* dest) {
+//     iterator itr = begin();
+//     lock();
+
+//     while (*itr != NULL) *++dest = *++itr;
+
+//     clear_and_unlock();
+//     return dest;
+//   }
+
+//   void clear_and_unlock() {
+//     std::memset(m_queue, 0, sizeof(value_type) * (max_size + 1));
+//     m_lock = 0;
+//     __sync_synchronize();
+//   }
+
+//  private:
+//   int        m_lock;
+//   value_type m_queue[max_size + 1];
+// };
+
+ThreadBase::ThreadBase()
+  : m_pollManager(NULL) {
+  // Init the poll manager in a special init function called by the
+  // thread itself. Need to be careful with what external stuff
+  // create_poll_manager calls in that case.
+
+//   m_threadQueue = new thread_queue_hack;
+}
+
+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::event_loop() {
+//   // Setup stuff...
+
+//   // Set local poll and priority queue.
+  
+//   while (true) {
+//     // Check for new queued items set by other threads.
+//     if (!m_threadQueue->empty())
+//       call_queued_items();
+
+//     // Remember to add global lock thing to the main poll loop ++.
+
+//     rak::priority_queue_perform(&taskScheduler, cachedTime);
+
+//     m_pollManager->poll_simple(rak::timer::from_seconds(10));
+//   }
+
+//   return NULL;
+// }
+
+// void
+// ThreadBase::call_queued_items() {
+//   thread_base_func result[thread_queue_hack::max_size];
+//   thread_base_func* first = result;
+//   thread_base_func* last = m_threadQueue->copy_and_clear((thread_base_func*)result);
+
+//   while (first != last)
+//     (*first)(this);
+// }
+
+// void
+// ThreadBase::queue_item(thread_base_func newFunc) {
+//   m_threadQueue->push_back(newFunc);
+// }
+
+}

src/utils/thread_base.h

@@ -0,0 +1,87 @@
+// libTorrent - 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_UTILS_THREAD_BASE_H
+#define RTORRENT_UTILS_THREAD_BASE_H
+
+#include <sys/types.h>
+#include "rak/priority_queue_default.h"
+#include "core/poll_manager.h"
+
+namespace utils {
+
+struct thread_queue_hack;
+
+// Move this class to libtorrent.
+
+// class __cacheline_aligned ThreadBase {
+class ThreadBase {
+public:
+  typedef rak::priority_queue_default priority_queue;
+  typedef void (*thread_base_func)(ThreadBase*);
+
+  ThreadBase();
+  ~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();
+
+  // ATM, only interaction with a thread's allowed by other threads is
+  // through the queue_item call.
+
+//   void                queue_item(thread_base_func newFunc);
+
+//   void*               event_loop();
+
+private:
+//   void                call_queued_items();
+
+  // The timer needs to be sync'ed when updated...
+
+  core::PollManager*          m_pollManager;
+  rak::priority_queue_default m_taskScheduler;
+
+  // Temporary hack to pass messages to a thread. This really needs to
+  // be cleaned up and/or integrated into the priority queue itself.
+//   thread_queue_hack*  m_threadQueue;
+};
+
+}
+
+#endif