From 667c4c35625804a883bbf0c1b9e7277cce5ef3e1 Mon Sep 17 00:00:00 2001 From: rakshasa Date: Thu, 12 Nov 2009 09:28:35 +0000 Subject: [PATCH] * 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 | 3 +- scripts/common.m4 | 22 ++++- src/command_network.cc | 3 +- src/control.cc | 8 +- src/control.h | 1 - src/core/curl_socket.cc | 14 +-- src/core/manager.cc | 42 +-------- src/core/manager.h | 3 - src/core/poll_manager.cc | 45 +++++++++ src/core/poll_manager.h | 3 + src/core/poll_manager_epoll.cc | 12 +++ src/core/poll_manager_epoll.h | 1 + src/core/poll_manager_kqueue.cc | 12 +++ src/core/poll_manager_kqueue.h | 1 + src/core/poll_manager_select.cc | 24 +++++ src/core/poll_manager_select.h | 1 + src/globals.cc | 2 + src/globals.h | 9 ++ src/main.cc | 8 +- src/rpc/scgi.cc | 23 +++-- src/rpc/scgi.h | 5 +- src/rpc/scgi_task.cc | 18 ++-- src/utils/Makefile.am | 4 +- src/utils/thread_base.cc | 159 ++++++++++++++++++++++++++++++++ src/utils/thread_base.h | 87 +++++++++++++++++ 25 files changed, 428 insertions(+), 82 deletions(-) create mode 100644 src/utils/thread_base.cc create mode 100644 src/utils/thread_base.h diff --git a/rak/timer.h b/rak/timer.h index 701a752d..e1b6af38 100644 --- a/rak/timer.h +++ b/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; } diff --git a/scripts/common.m4 b/scripts/common.m4 index fe7ef3c3..81cfe920 100644 --- a/scripts/common.m4 +++ b/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 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() + ]) +]) diff --git a/src/command_network.cc b/src/command_network.cc index 9eb7c795..34f75ab1 100644 --- a/src/command_network.cc +++ b/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", ""); } diff --git a/src/control.cc b/src/control.cc index 62e76091..66c2cdfb 100644 --- a/src/control.cc +++ b/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(); -} diff --git a/src/control.h b/src/control.h index 92ea9b63..06b7e871 100644 --- a/src/control.h +++ b/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; } diff --git a/src/core/curl_socket.cc b/src/core/curl_socket.cc index c91668d9..87d310d0 100644 --- a/src/core/curl_socket.cc +++ b/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; } diff --git a/src/core/manager.cc b/src/core/manager.cc index 3886c4fa..d7a8c765 100644 --- a/src/core/manager.cc +++ b/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 diff --git a/src/core/manager.h b/src/core/manager.h index 595f4a4b..3b23da3e 100644 --- a/src/core/manager.h +++ b/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; }; diff --git a/src/core/poll_manager.cc b/src/core/poll_manager.cc index 7e9600f3..90df840e 100644 --- a/src/core/poll_manager.cc +++ b/src/core/poll_manager.cc @@ -39,7 +39,13 @@ #include #include +#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) diff --git a/src/core/poll_manager.h b/src/core/poll_manager.h index 1da2f432..1649edf5 100644 --- a/src/core/poll_manager.h +++ b/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&); diff --git a/src/core/poll_manager_epoll.cc b/src/core/poll_manager_epoll.cc index c09c85f2..e8cd4f12 100644 --- a/src/core/poll_manager_epoll.cc +++ b/src/core/poll_manager_epoll.cc @@ -74,4 +74,16 @@ PollManagerEPoll::poll(rak::timer timeout) { static_cast(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(m_poll)->poll((timeout.usec() + 999) / 1000) == -1) + return check_error(); + + static_cast(m_poll)->perform(); +} + } diff --git a/src/core/poll_manager_epoll.h b/src/core/poll_manager_epoll.h index 4fa148e7..3e4f2ac5 100644 --- a/src/core/poll_manager_epoll.h +++ b/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) {} diff --git a/src/core/poll_manager_kqueue.cc b/src/core/poll_manager_kqueue.cc index 641b4d51..1c6e5197 100644 --- a/src/core/poll_manager_kqueue.cc +++ b/src/core/poll_manager_kqueue.cc @@ -75,4 +75,16 @@ PollManagerKQueue::poll(rak::timer timeout) { static_cast(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(m_poll)->poll((timeout.usec() + 999) / 1000) == -1) + return check_error(); + + static_cast(m_poll)->perform(); +} + } diff --git a/src/core/poll_manager_kqueue.h b/src/core/poll_manager_kqueue.h index 26f0fbce..05165042 100644 --- a/src/core/poll_manager_kqueue.h +++ b/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) {} diff --git a/src/core/poll_manager_select.cc b/src/core/poll_manager_select.cc index aca87724..929c547a 100644 --- a/src/core/poll_manager_select.cc +++ b/src/core/poll_manager_select.cc @@ -120,4 +120,28 @@ PollManagerSelect::poll(rak::timer timeout) { static_cast(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(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(m_poll)->perform(m_readSet, m_writeSet, m_errorSet); +} + } diff --git a/src/core/poll_manager_select.h b/src/core/poll_manager_select.h index 3b9124f7..c869509c 100644 --- a/src/core/poll_manager_select.h +++ b/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); diff --git a/src/globals.cc b/src/globals.cc index dc9bf371..2d04ddfe 100644 --- a/src/globals.cc +++ b/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; diff --git a/src/globals.h b/src/globals.h index 3ac81c20..165d433b 100644 --- a/src/globals.h +++ b/src/globals.h @@ -40,11 +40,20 @@ #include #include +#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 diff --git a/src/main.cc b/src/main.cc index 02951848..18e904bb 100644 --- a/src/main.cc +++ b/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; } diff --git a/src/rpc/scgi.cc b/src/rpc/scgi.cc index f1d22fa6..61627342 100644 --- a/src/rpc/scgi.cc +++ b/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; diff --git a/src/rpc/scgi.h b/src/rpc/scgi.h index 4e75f842..f0fa9e36 100644 --- a/src/rpc/scgi.h +++ b/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); } diff --git a/src/rpc/scgi_task.cc b/src/rpc/scgi_task.cc index efd8683c..4f783354 100644 --- a/src/rpc/scgi_task.cc +++ b/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... diff --git a/src/utils/Makefile.am b/src/utils/Makefile.am index 55ea9ff1..f2daa2bb 100644 --- a/src/utils/Makefile.am +++ b/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) diff --git a/src/utils/thread_base.cc b/src/utils/thread_base.cc new file mode 100644 index 00000000..bef62b7e --- /dev/null +++ b/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 +// +// Skomakerveien 33 +// 3185 Skoppum, NORWAY + +#include "config.h" + +#include "thread_base.h" + +#include +#include + +#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); +// } + +} diff --git a/src/utils/thread_base.h b/src/utils/thread_base.h new file mode 100644 index 00000000..1442710d --- /dev/null +++ b/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 +// +// Skomakerveien 33 +// 3185 Skoppum, NORWAY + +#ifndef RTORRENT_UTILS_THREAD_BASE_H +#define RTORRENT_UTILS_THREAD_BASE_H + +#include +#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