From 6a94909647a8113a9883545eb2cc44bc1ea7b6f3 Mon Sep 17 00:00:00 2001 From: rakshasa Date: Mon, 14 Dec 2009 21:18:51 +0000 Subject: [PATCH] * 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 | 1 + src/core/poll_manager_select.cc | 7 +++++- src/thread_base.cc | 23 +++++++++++++++--- src/thread_base.h | 41 +++++++++++++++++++++++++++++++++ src/thread_main.cc | 3 +++ src/thread_worker.cc | 27 +++++++++++++++++++--- src/thread_worker.h | 7 +++++- 7 files changed, 101 insertions(+), 8 deletions(-) diff --git a/src/core/poll_manager.cc b/src/core/poll_manager.cc index 90df840e..458fef5d 100644 --- a/src/core/poll_manager.cc +++ b/src/core/poll_manager.cc @@ -37,6 +37,7 @@ #include "config.h" #include +#include #include #include "globals.h" diff --git a/src/core/poll_manager_select.cc b/src/core/poll_manager_select.cc index f62dd212..6decb9b7 100644 --- a/src/core/poll_manager_select.cc +++ b/src/core/poll_manager_select.cc @@ -45,6 +45,7 @@ #include #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(); diff --git a/src/thread_base.cc b/src/thread_base.cc index eec69491..3235cd40 100644 --- a/src/thread_base.cc +++ b/src/thread_base.cc @@ -38,11 +38,18 @@ #include "thread_base.h" +#include #include +#include #include #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; diff --git a/src/thread_base.h b/src/thread_base.h index 977209f3..d8249215 100644 --- a/src/thread_base.h +++ b/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 diff --git a/src/thread_main.cc b/src/thread_main.cc index ff2ee126..ca3599cf 100644 --- a/src/thread_main.cc +++ b/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; diff --git a/src/thread_worker.cc b/src/thread_worker.cc index feab0f03..eeceb8d8 100644 --- a/src/thread_worker.cc +++ b/src/thread_worker.cc @@ -38,10 +38,17 @@ #include "thread_worker.h" #include "globals.h" +#include "control.h" #include #include +#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(); } diff --git a/src/thread_worker.h b/src/thread_worker.h index 3b068865..33d4de07 100644 --- a/src/thread_worker.h +++ b/src/thread_worker.h @@ -39,12 +39,14 @@ #include "thread_base.h" +#include + // 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