* More work on the threading stuff.

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

src/core/poll_manager.cc

@@ -37,6 +37,7 @@
 #include "config.h"
 
 #include <stdexcept>
+#include <unistd.h>
 #include <rak/error_number.h>
 
 #include "globals.h"

src/core/poll_manager_select.cc

@@ -45,6 +45,7 @@
 #include <torrent/torrent.h>
 
 #include "poll_manager_select.h"
+#include "thread_base.h"
 
 namespace core {
 
@@ -92,7 +93,11 @@ PollManagerSelect::poll(rak::timer timeout) {
 
   timeval t = timeout.tval();
 
-  if (select(maxFd + 1, m_readSet, m_writeSet, m_errorSet, &t) == -1)
+  ThreadBase::release_global_lock();
+  int status = select(maxFd + 1, m_readSet, m_writeSet, m_errorSet, &t);
+  ThreadBase::release_global_lock();
+
+  if (status == -1)
     return check_error();
 
   torrent::perform();

src/thread_base.cc

@@ -38,11 +38,18 @@
 
 #include "thread_base.h"
 
+#include <cassert>
 #include <cstdlib>
+#include <iostream>
 #include <torrent/exceptions.h>
 
 #include "globals.h"
 
+// A preliminary implementation of a global lock, to be moved
+// somewhere more appropriate when it's put to use.
+
+ThreadBase::global_lock_type ThreadBase::m_global = { 0, PTHREAD_MUTEX_INITIALIZER };
+
 // Temporarly injected into config.h.
 /* temp hack */
 //#define __cacheline_aligned __attribute__((__aligned__(128)))
@@ -108,7 +115,7 @@ ThreadBase::ThreadBase() :
 }
 
 ThreadBase::~ThreadBase() {
-  // Cleanup...
+  pthread_mutex_destroy(&ThreadBase::m_global.lock);
 }
 
 void
@@ -122,6 +129,16 @@ void
 ThreadBase::stop_thread() {
 }
 
+inline rak::timer
+ThreadBase::client_next_timeout() {
+  if (m_taskScheduler.empty())
+    return rak::timer::from_seconds(10);
+  else if (m_taskScheduler.top()->time() <= cachedTime)
+    return 0;
+  else
+    return m_taskScheduler.top()->time() - cachedTime;
+}
+
 void*
 ThreadBase::event_loop(ThreadBase* threadBase) {
   // Setup stuff...
@@ -134,11 +151,11 @@ ThreadBase::event_loop(ThreadBase* threadBase) {
     if (!threadBase->m_threadQueue->empty())
       threadBase->call_queued_items();
 
-//     // Remember to add global lock thing to the main poll loop ++.
+    //     // Remember to add global lock thing to the main poll loop ++.
 
     rak::priority_queue_perform(&threadBase->m_taskScheduler, cachedTime);
 
-    threadBase->m_pollManager->poll_simple(rak::timer::from_seconds(10));
+    threadBase->m_pollManager->poll_simple(threadBase->client_next_timeout());
   }
 
   return NULL;

src/thread_base.h

@@ -80,7 +80,15 @@ public:
 
   static void*        event_loop(ThreadBase* threadBase);
 
+  static inline int   global_queue_size() { return m_global.waiting; }
+
+  static inline void  acquire_global_lock();
+  static inline void  release_global_lock();
+  static inline void  waive_global_lock();
+
 protected:
+  inline rak::timer   client_next_timeout();
+
   void                call_queued_items();
 
   pthread_t           m_thread;
@@ -94,6 +102,39 @@ protected:
   // 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;
+
+  struct __cacheline_aligned global_lock_type {
+    int             waiting;
+    pthread_mutex_t lock;
+  };
+
+  static global_lock_type m_global;
 };
 
+inline void
+ThreadBase::acquire_global_lock() {
+  __sync_add_and_fetch(&ThreadBase::m_global.waiting, 1);
+
+  pthread_mutex_lock(&ThreadBase::m_global.lock);
+
+//   if (pthread_mutex_lock(&ThreadBase::m_global.lock))
+//     throw internal_error("Mutex failed.");
+  
+  __sync_fetch_and_sub(&ThreadBase::m_global.waiting, 1);
+}
+
+inline void
+ThreadBase::release_global_lock() {
+  pthread_mutex_unlock(&ThreadBase::m_global.lock);
+}
+
+inline void
+ThreadBase::waive_global_lock() {
+  __sync_synchronize();
+  pthread_mutex_unlock(&ThreadBase::m_global.lock);
+
+  // Do we need to sleep here? Make a CppUnit test for this.
+  acquire_global_lock();
+}
+
 #endif

src/thread_main.cc

@@ -46,6 +46,9 @@ ThreadMain::~ThreadMain() {
 
 void
 ThreadMain::init_thread() {
+  // The main thread always holds the lock while running.
+  acquire_global_lock();
+
   m_pollManager = core::PollManager::create_poll_manager();
 
   m_state = STATE_INITIALIZED;

src/thread_worker.cc

@@ -38,10 +38,17 @@
 
 #include "thread_worker.h"
 #include "globals.h"
+#include "control.h"
 
 #include <cassert>
 #include <torrent/exceptions.h>
 
+#include "core/manager.h"
+
+ThreadWorker::ThreadWorker() {
+  m_taskTouchLog.set_slot(rak::mem_fn(this, &ThreadWorker::task_touch_log));
+}
+
 ThreadWorker::~ThreadWorker() {
 }
 
@@ -53,8 +60,22 @@ ThreadWorker::init_thread() {
 }
 
 void
-ThreadWorker::start_log_counter(ThreadBase* thread) {
-  assert(false);
+ThreadWorker::start_log_counter(ThreadBase* baseThread) {
+  ThreadWorker* thread = (ThreadWorker*)baseThread;
 
-  throw torrent::internal_error("PRRREREREFFFERERE");
+  if (!thread->m_taskTouchLog.is_queued())
+    priority_queue_insert(&thread->m_taskScheduler, &thread->m_taskTouchLog, cachedTime);
+}
+
+void
+ThreadWorker::task_touch_log() {
+  priority_queue_insert(&m_taskScheduler, &m_taskTouchLog, cachedTime + rak::timer::from_seconds(1));
+
+  acquire_global_lock();
+  __sync_synchronize();
+
+  control->core()->push_log("Tick Tock.");
+
+  __sync_synchronize();
+  release_global_lock();
 }

src/thread_worker.h

@@ -39,12 +39,14 @@
 
 #include "thread_base.h"
 
+#include <rak/priority_queue_default.h>
+
 // Check if cacheline aligned with inheritance ends up taking two
 // cachelines.
 
 class __cacheline_aligned ThreadWorker : public ThreadBase {
 public:
-  ThreadWorker() {}
+  ThreadWorker();
   ~ThreadWorker();
 
   virtual void        init_thread();
@@ -52,6 +54,9 @@ public:
   static void start_log_counter(ThreadBase* thread);
 
 private:
+  void                task_touch_log();
+
+  rak::priority_item  m_taskTouchLog;
 };
 
 #endif