diff --git a/Makefile.am b/Makefile.am index 65fc633e..2f20a875 100644 --- a/Makefile.am +++ b/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 \ diff --git a/rak/allocators.h b/rak/allocators.h new file mode 100644 index 00000000..bc1c48c7 --- /dev/null +++ b/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 +// +// Skomakerveien 33 +// 3185 Skoppum, NORWAY + +// Some allocators for cacheline aligned chunks of memory, etc. + +#ifndef RAK_ALLOCATORS_H +#define RAK_ALLOCATORS_H + +#include +#include + +namespace rak { + +template +class cacheline_allocator { +public: + typedef std::allocator 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 + cacheline_allocator(const cacheline_allocator&) throw() { } + ~cacheline_allocator() throw() { } + + template + struct rebind { typedef cacheline_allocator 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::max() / sizeof(T); } + + pointer allocate(size_type num, std::allocator::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 +bool operator== (const cacheline_allocator&, const cacheline_allocator&) throw() { + return true; +} + +template +bool operator!= (const cacheline_allocator&, const cacheline_allocator&) throw() { + return false; +} + +} + +// +// Operator new with custom allocators: +// + +template +void* operator new(size_t s, rak::cacheline_allocator a) { + typename rak::cacheline_allocator::pointer ptr = NULL; + posix_memalign((void**)&ptr, L1_CACHE_BYTES, s); + + return ptr; +} + +#endif // namespace rak diff --git a/rak/priority_queue.h b/rak/priority_queue.h index a8bff52d..69c89811 100644 --- a/rak/priority_queue.h +++ b/rak/priority_queue.h @@ -46,10 +46,10 @@ namespace rak { -template -class priority_queue : public std::vector { +template > +class priority_queue : public std::vector { public: - typedef std::vector base_type; + typedef std::vector base_type; typedef typename base_type::reference reference; typedef typename base_type::const_reference const_reference; typedef typename base_type::iterator iterator; diff --git a/rak/priority_queue_default.h b/rak/priority_queue_default.h index 83fb8466..0b4da804 100644 --- a/rak/priority_queue_default.h +++ b/rak/priority_queue_default.h @@ -38,6 +38,7 @@ #define RAK_PRIORITY_QUEUE_DEFAULT_H #include +#include #include #include #include @@ -83,7 +84,8 @@ struct priority_compare { }; typedef std::equal_to priority_equal; -typedef priority_queue priority_queue_default; +typedef priority_queue > priority_queue_default; inline void priority_queue_perform(priority_queue_default* queue, timer t) { diff --git a/src/Makefile.am b/src/Makefile.am index 7493bdf6..90a4c13b 100644 --- a/src/Makefile.am +++ b/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) diff --git a/src/core/poll_manager_select.cc b/src/core/poll_manager_select.cc index 929c547a..f62dd212 100644 --- a/src/core/poll_manager_select.cc +++ b/src/core/poll_manager_select.cc @@ -40,6 +40,7 @@ #include #include #include +#include #include #include @@ -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(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(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(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(m_poll)->perform(m_readSet, m_writeSet, m_errorSet); + currentPoll->perform(m_readSet, m_writeSet, m_errorSet); } } diff --git a/src/core/poll_manager_select.h b/src/core/poll_manager_select.h index c869509c..be75c51e 100644 --- a/src/core/poll_manager_select.h +++ b/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(); diff --git a/src/globals.cc b/src/globals.cc index 42bec88a..cd3c5fbb 100644 --- a/src/globals.cc +++ b/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; diff --git a/src/globals.h b/src/globals.h index ea25a3e5..a1dd393f 100644 --- a/src/globals.h +++ b/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 diff --git a/src/main.cc b/src/main.cc index b8427178..a59e4f31 100644 --- a/src/main.cc +++ b/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(); diff --git a/src/thread_base.cc b/src/thread_base.cc index dec63825..19049a05 100644 --- a/src/thread_base.cc +++ b/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); +} diff --git a/src/thread_base.h b/src/thread_base.h index 5684924b..18903356 100644 --- a/src/thread_base.h +++ b/src/thread_base.h @@ -37,7 +37,9 @@ #ifndef RTORRENT_UTILS_THREAD_BASE_H #define RTORRENT_UTILS_THREAD_BASE_H +#include #include + #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