* Foundation for multi-threaded support.

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

Makefile.am

@@ -6,6 +6,7 @@ EXTRA_DIST= \
 	autogen.sh \
 	rak/address_info.h \
 	rak/algorithm.h \
+	rak/allocators.h \
 	rak/error_number.h \
 	rak/file_stat.h \
 	rak/fs_stat.h \

rak/allocators.h

@@ -0,0 +1,112 @@
+// rak - Rakshasa's toolbox
+// 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
+
+// Some allocators for cacheline aligned chunks of memory, etc.
+
+#ifndef RAK_ALLOCATORS_H
+#define RAK_ALLOCATORS_H
+
+#include <limits>
+#include <stdlib.h>
+
+namespace rak {
+
+template <class T = void*>
+class cacheline_allocator {
+public:
+  typedef std::allocator<T> base_type;
+
+  typedef size_t size_type;
+  typedef ptrdiff_t difference_type;
+  typedef T* pointer;
+  typedef const T* const_pointer;
+  typedef T& reference;
+  typedef const T& const_reference;
+  typedef T value_type;
+
+  cacheline_allocator() throw() { }
+  cacheline_allocator(const cacheline_allocator&) throw() { }
+  template <class U>
+  cacheline_allocator(const cacheline_allocator<U>&) throw() { }
+  ~cacheline_allocator() throw() { }
+
+  template <class U>
+  struct rebind { typedef cacheline_allocator<U> other; };
+
+  // return address of values
+  pointer address (reference value) const { return &value; }
+  const_pointer address (const_reference value) const { return &value; }
+
+  size_type max_size () const throw() { return std::numeric_limits<size_t>::max() / sizeof(T); }
+
+  pointer allocate(size_type num, std::allocator<void>::const_pointer hint = 0) {
+    pointer ptr = NULL;
+    posix_memalign((void**)&ptr, L1_CACHE_BYTES, num*sizeof(T));
+  
+    return ptr;
+  }
+
+  void construct (pointer p, const T& value) { new((void*)p)T(value); }
+  void destroy (pointer p) { p->~T(); }
+  void deallocate (pointer p, size_type num) { ::operator delete((void*)p); }
+};
+
+
+template <class T1, class T2>
+bool operator== (const cacheline_allocator<T1>&, const cacheline_allocator<T2>&) throw() {
+  return true;
+}
+
+template <class T1, class T2>
+bool operator!= (const cacheline_allocator<T1>&, const cacheline_allocator<T2>&) throw() {
+  return false;
+}
+
+}
+
+//
+// Operator new with custom allocators:
+//
+
+template <typename T>
+void* operator new(size_t s, rak::cacheline_allocator<T> a) {
+  typename rak::cacheline_allocator<T>::pointer ptr = NULL;
+  posix_memalign((void**)&ptr, L1_CACHE_BYTES, s);
+  
+  return ptr;
+}
+
+#endif // namespace rak

rak/priority_queue.h

@@ -46,10 +46,10 @@
 
 namespace rak {
 
-template <typename Value, typename Compare, typename Equal>
-class priority_queue : public std::vector<Value> {
+template <typename Value, typename Compare, typename Equal, typename Alloc = std::allocator<Value> >
+class priority_queue : public std::vector<Value, Alloc> {
 public:
-  typedef std::vector<Value>                  base_type;
+  typedef std::vector<Value, Alloc>           base_type;
   typedef typename base_type::reference       reference;
   typedef typename base_type::const_reference const_reference;
   typedef typename base_type::iterator        iterator;

rak/priority_queue_default.h

@@ -38,6 +38,7 @@
 #define RAK_PRIORITY_QUEUE_DEFAULT_H
 
 #include <stdexcept>
+#include <rak/allocators.h>
 #include <rak/functional.h>
 #include <rak/functional_fun.h>
 #include <rak/priority_queue.h>
@@ -83,7 +84,8 @@ struct priority_compare {
 };
 
 typedef std::equal_to<priority_item*> priority_equal;
-typedef priority_queue<priority_item*, priority_compare, priority_equal> priority_queue_default;
+typedef priority_queue<priority_item*, priority_compare, priority_equal,
+                       cacheline_allocator<priority_item*> > priority_queue_default;
 
 inline void
 priority_queue_perform(priority_queue_default* queue, timer t) {

src/Makefile.am

@@ -40,6 +40,8 @@ rtorrent_SOURCES = \
 	signal_handler.cc \
 	signal_handler.h \
 	thread_base.cc \
-	thread_base.h
+	thread_base.h \
+	thread_worker.cc \
+	thread_worker.h
 
 INCLUDES = -I$(srcdir) -I$(top_srcdir)

src/core/poll_manager_select.cc

@@ -40,6 +40,7 @@
 #include <stdexcept>
 #include <unistd.h>
 #include <sys/time.h>
+#include <torrent/exceptions.h>
 #include <torrent/poll_select.h>
 #include <torrent/torrent.h>
 
@@ -49,26 +50,18 @@ namespace core {
 
 PollManagerSelect::PollManagerSelect(torrent::Poll* p) : PollManager(p) {
 #if defined USE_VARIABLE_FDSET
-  m_setSize = m_poll->open_max() / 8;
-  m_readSet = (fd_set*)new char[m_setSize];
-  m_writeSet = (fd_set*)new char[m_setSize];
-  m_errorSet = (fd_set*)new char[m_setSize];
+  m_setSize = (m_poll->open_max() + 7) / 8;
 
-  std::memset(m_readSet, 0, m_setSize);
-  std::memset(m_writeSet, 0, m_setSize);
-  std::memset(m_errorSet, 0, m_setSize);
+  char* buffer;
+  posix_memalign((void**)&buffer, L1_CACHE_BYTES, 3 * m_setSize);
+
+  std::memset(buffer, 0, 3 * m_setSize);
+
+  m_readSet = (fd_set*)buffer;
+  m_writeSet = (fd_set*)(buffer += m_setSize);
+  m_errorSet = (fd_set*)(buffer += m_setSize);
 #else
-  if (m_poll->open_max() > FD_SETSIZE)
-    throw std::logic_error("PollManagerSelect::create(...) received a max open sockets >= FD_SETSIZE, but USE_VARIABLE_FDSET was not defined");
-
-  m_setSize = FD_SETSIZE / 8;
-  m_readSet = new fd_set;
-  m_writeSet = new fd_set;
-  m_errorSet = new fd_set;
-
-  FD_ZERO(m_readSet);
-  FD_ZERO(m_writeSet);
-  FD_ZERO(m_errorSet);
+#error Only variable fdset supported atm.
 #endif
 }
 
@@ -78,20 +71,12 @@ PollManagerSelect::create(int maxOpenSockets) {
 
   if (p == NULL)
     return NULL;
-  else
-    return new PollManagerSelect(p);
+
+  return new PollManagerSelect(p);
 }
 
 PollManagerSelect::~PollManagerSelect() {
-#if defined USE_VARIABLE_FDSET
-  delete [] m_readSet;
-  delete [] m_writeSet;
-  delete [] m_errorSet;
-#else
-  delete m_readSet;
-  delete m_writeSet;
-  delete m_errorSet;
-#endif
+  free(m_readSet);
 }
 
 void
@@ -99,15 +84,9 @@ PollManagerSelect::poll(rak::timer timeout) {
   torrent::perform();
   timeout = std::min(timeout, rak::timer(torrent::next_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);
 
@@ -122,26 +101,23 @@ PollManagerSelect::poll(rak::timer timeout) {
 
 void
 PollManagerSelect::poll_simple(rak::timer timeout) {
+  torrent::PollSelect* currentPoll = static_cast<torrent::PollSelect*>(m_poll);
+
   timeout = timeout + 1000;
+  std::memset(m_readSet, 0, 3 * m_setSize);
 
-#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 = currentPoll->fdset(m_readSet, m_writeSet, m_errorSet);
+  //  unsigned int maxFd = 0;
 
-  unsigned int maxFd = static_cast<torrent::PollSelect*>(m_poll)->fdset(m_readSet, m_writeSet, m_errorSet);
+  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();
 
   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);
+  currentPoll->perform(m_readSet, m_writeSet, m_errorSet);
 }
 
 }

src/core/poll_manager_select.h

@@ -45,7 +45,7 @@ namespace torrent {
 
 namespace core {
 
-class PollManagerSelect : public PollManager {
+class __cacheline_aligned PollManagerSelect : public PollManager {
 public:
   static PollManagerSelect* create(int maxOpenSockets);
   ~PollManagerSelect();

src/globals.cc

@@ -44,3 +44,4 @@ rak::timer                  cachedTime;
 Control*                    control = NULL;
 //__thread ThreadBase* this_thread = NULL;
 ThreadBase* this_thread = NULL;
+ThreadBase* worker_thread = NULL;

src/globals.h

@@ -54,6 +54,7 @@ extern rak::timer                  cachedTime;
 
 extern Control*                    control;
 // extern __thread ThreadBase* this_thread; // Only use for worker threads for now.
-extern ThreadBase* this_thread; // Only use for main threads for now.
+extern ThreadBase* this_thread;
+extern ThreadBase* worker_thread;
 
 #endif

src/main.cc

@@ -69,6 +69,8 @@
 #include "signal_handler.h"
 #include "option_parser.h"
 
+#include "thread_worker.h"
+
 void do_panic(int signum);
 void print_help();
 void initialize_commands();
@@ -157,6 +159,9 @@ main(int argc, char** argv) {
     this_thread = new ThreadBase();
     this_thread->init_thread();
 
+    worker_thread = new ThreadWorker();
+    worker_thread->init_thread();
+
     srandom(cachedTime.usec());
     srand48(cachedTime.usec());
 
@@ -308,6 +313,8 @@ main(int argc, char** argv) {
     control->display()->adjust_layout();
     control->display()->receive_update();
 
+    worker_thread->start_thread();
+
     while (!control->is_shutdown_completed()) {
       if (control->is_shutdown_received())
         control->handle_shutdown();

src/thread_base.cc

@@ -47,63 +47,64 @@
 /* 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;
+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;
+  static const unsigned int max_size = 32;
 
-//   thread_queue_hack() { std::memset(m_queue, 0, sizeof(thread_queue_hack)); }
+  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); }
+  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 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); }
+  iterator end_and_lock() { lock(); return std::find(begin(), max_capacity(), (value_type)NULL); }
 
-//   bool     empty() const { return m_queue[0] == NULL; }
+  bool     empty() const { return m_queue[0] == NULL; }
 
-//   void push_back(value_type v) {
-//     iterator itr = end_and_lock();
+  void push_back(value_type v) {
+    iterator itr = end_and_lock();
   
-//     if (itr == max_capacity())
-//       throw torrent::internal_error("Overflowed thread_queue.");
+    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);
-//   }
+    __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();
+  value_type* copy_and_clear(value_type* dest) {
+    iterator itr = begin();
+    lock();
 
-//     while (*itr != NULL) *++dest = *++itr;
+    while (*itr != NULL) *++dest = *++itr;
 
-//     clear_and_unlock();
-//     return dest;
-//   }
+    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();
-//   }
+  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];
-// };
+ private:
+  int        m_lock;
+  value_type m_queue[max_size + 1];
+};
 
-ThreadBase::ThreadBase()
-  : m_pollManager(NULL) {
+ThreadBase::ThreadBase() :
+  m_state(STATE_UNKNOWN),
+  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;
+  m_threadQueue = new thread_queue_hack;
 }
 
 ThreadBase::~ThreadBase() {
@@ -118,38 +119,50 @@ ThreadBase::init_thread() {
   m_pollManager = core::PollManager::create_poll_manager();
 }
 
-// void*
-// ThreadBase::event_loop() {
-//   // Setup stuff...
+void
+ThreadBase::start_thread() {
+  if (m_state != STATE_INITIALIZED ||
+      pthread_create(&m_thread, NULL, (pthread_func)&ThreadBase::event_loop, this))
+    throw torrent::internal_error("Failed to create thread.");
+}
 
-//   // Set local poll and priority queue.
+void
+ThreadBase::stop_thread() {
+}
+
+void*
+ThreadBase::event_loop(ThreadBase* threadBase) {
+  // Setup stuff...
+  threadBase->m_state = STATE_ACTIVE;
+
+  // Set local poll and priority queue.
   
-//   while (true) {
-//     // Check for new queued items set by other threads.
-//     if (!m_threadQueue->empty())
-//       call_queued_items();
+  while (true) {
+    // Check for new queued items set by other threads.
+    if (!threadBase->m_threadQueue->empty())
+      threadBase->call_queued_items();
 
 //     // Remember to add global lock thing to the main poll loop ++.
 
-//     rak::priority_queue_perform(&taskScheduler, cachedTime);
+//    rak::priority_queue_perform(&threadBase->m_taskScheduler, cachedTime);
 
-//     m_pollManager->poll_simple(rak::timer::from_seconds(10));
-//   }
+    threadBase->m_pollManager->poll_simple(rak::timer::from_seconds(10));
+  }
 
-//   return NULL;
-// }
+  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);
+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);
-// }
+  while (first != last)
+    (*first)(this);
+}
 
-// void
-// ThreadBase::queue_item(thread_base_func newFunc) {
-//   m_threadQueue->push_back(newFunc);
-// }
+void
+ThreadBase::queue_item(thread_base_func newFunc) {
+  m_threadQueue->push_back(newFunc);
+}

src/thread_base.h

@@ -37,7 +37,9 @@
 #ifndef RTORRENT_UTILS_THREAD_BASE_H
 #define RTORRENT_UTILS_THREAD_BASE_H
 
+#include <pthread.h>
 #include <sys/types.h>
+
 #include "rak/priority_queue_default.h"
 #include "core/poll_manager.h"
 
@@ -45,11 +47,19 @@ struct thread_queue_hack;
 
 // Move this class to libtorrent.
 
-// class __cacheline_aligned ThreadBase {
-class ThreadBase {
+struct thread_queue_hack;
+
+class __cacheline_aligned ThreadBase {
 public:
   typedef rak::priority_queue_default priority_queue;
   typedef void (*thread_base_func)(ThreadBase*);
+  typedef void* (*pthread_func)(void*);
+
+  enum state_type {
+    STATE_UNKNOWN,
+    STATE_INITIALIZED,
+    STATE_ACTIVE
+  };
 
   ThreadBase();
   ~ThreadBase();
@@ -60,15 +70,21 @@ public:
 
   virtual void        init_thread();
 
+  void                start_thread();
+  void                stop_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                queue_item(thread_base_func newFunc);
 
-//   void*               event_loop();
+  static void*        event_loop(ThreadBase* threadBase);
 
-private:
-//   void                call_queued_items();
+protected:
+  void                call_queued_items();
+
+  pthread_t           m_thread;
+  state_type          m_state;
 
   // The timer needs to be sync'ed when updated...
 
@@ -77,7 +93,7 @@ private:
 
   // 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;
+  thread_queue_hack*  m_threadQueue;
 };
 
 #endif