* Added shutdown for worker thread.

git-svn-id: svn://rakshasa.no/libtorrent/trunk/rtorrent@1117 e378c898-3ddf-0310-93e7-cc216c733640
This commit is contained in:
rakshasa
2009-12-22 03:29:47 +00:00
parent 5a1d0be323
commit 2c7563825c
5 changed files with 54 additions and 12 deletions
+13
View File
@@ -144,9 +144,18 @@ Control::cleanup_exception() {
display::Canvas::cleanup();
}
bool
Control::is_shutdown_completed() {
return m_shutdownQuick && !worker_thread->is_active() && torrent::is_inactive();
}
void
Control::handle_shutdown() {
if (!m_shutdownQuick) {
// Temporary hack:
if (worker_thread->is_active())
worker_thread->queue_item(&ThreadBase::stop_thread);
torrent::connection_manager()->listen_close();
m_core->shutdown(false);
@@ -154,6 +163,10 @@ Control::handle_shutdown() {
priority_queue_insert(&taskScheduler, &m_taskShutdown, cachedTime + rak::timer::from_seconds(5));
} else {
// Temporary hack:
if (worker_thread->is_active())
worker_thread->queue_item(&ThreadBase::stop_thread);
m_core->shutdown(true);
}
+1 -1
View File
@@ -74,7 +74,7 @@ public:
Control();
~Control();
bool is_shutdown_completed() { return m_shutdownQuick && torrent::is_inactive(); }
bool is_shutdown_completed();
bool is_shutdown_received() { return m_shutdownReceived; }
bool is_shutdown_started() { return m_shutdownQuick; }
+1
View File
@@ -340,6 +340,7 @@ main(int argc, char** argv) {
}
delete control;
delete worker_thread;
delete main_thread;
return 0;
+32 -9
View File
@@ -44,6 +44,8 @@
#include <torrent/exceptions.h>
#include "globals.h"
#include "control.h"
#include "core/manager.h"
// Temporarly injected into config.h.
/* temp hack */
@@ -99,6 +101,8 @@ public:
value_type m_queue[max_size + 1];
};
void throw_shutdown_exception() { throw torrent::shutdown_exception(); }
ThreadBase::ThreadBase() :
m_state(STATE_UNKNOWN),
m_pollManager(NULL) {
@@ -106,6 +110,8 @@ ThreadBase::ThreadBase() :
// thread itself. Need to be careful with what external stuff
// create_poll_manager calls in that case.
m_taskShutdown.set_slot(rak::ptr_fn(&throw_shutdown_exception));
m_threadQueue = new thread_queue_hack;
}
@@ -120,13 +126,15 @@ ThreadBase::start_thread() {
}
void
ThreadBase::stop_thread() {
ThreadBase::stop_thread(ThreadBase* thread) {
if (!thread->m_taskShutdown.is_queued())
priority_queue_insert(&thread->m_taskScheduler, &thread->m_taskShutdown, cachedTime);
}
inline rak::timer
ThreadBase::client_next_timeout() {
if (m_taskScheduler.empty())
return rak::timer::from_seconds(10);
return rak::timer::from_seconds(2);
else if (m_taskScheduler.top()->time() <= cachedTime)
return 0;
else
@@ -140,18 +148,30 @@ ThreadBase::event_loop(ThreadBase* threadBase) {
// Set local poll and priority queue.
while (true) {
// Check for new queued items set by other threads.
if (!threadBase->m_threadQueue->empty())
threadBase->call_queued_items();
try {
// // Remember to add global lock thing to the main poll loop ++.
while (true) {
// Check for new queued items set by other threads.
if (!threadBase->m_threadQueue->empty())
threadBase->call_queued_items();
rak::priority_queue_perform(&threadBase->m_taskScheduler, cachedTime);
// // Remember to add global lock thing to the main poll loop ++.
threadBase->m_pollManager->poll_simple(threadBase->client_next_timeout());
rak::priority_queue_perform(&threadBase->m_taskScheduler, cachedTime);
threadBase->m_pollManager->poll_simple(threadBase->client_next_timeout());
}
} catch (torrent::shutdown_exception& e) {
acquire_global_lock();
control->core()->push_log("Shutting down thread.");
release_global_lock();
// sleep(20);
}
threadBase->m_state = STATE_INACTIVE;
__sync_synchronize();
return NULL;
}
@@ -168,4 +188,7 @@ ThreadBase::call_queued_items() {
void
ThreadBase::queue_item(thread_base_func newFunc) {
m_threadQueue->push_back(newFunc);
// TODO: Need to send a signal that interrupts polling so as to
// ensure we react immediately.
}
+7 -2
View File
@@ -59,12 +59,15 @@ public:
enum state_type {
STATE_UNKNOWN,
STATE_INITIALIZED,
STATE_ACTIVE
STATE_ACTIVE,
STATE_INACTIVE
};
ThreadBase();
virtual ~ThreadBase();
bool is_active() const { return m_state == STATE_ACTIVE; }
torrent::Poll* poll() { return m_pollManager->get_torrent_poll(); }
core::PollManager* poll_manager() { return m_pollManager; }
priority_queue& task_scheduler() { return m_taskScheduler; }
@@ -72,7 +75,7 @@ public:
virtual void init_thread() = 0;
void start_thread();
void stop_thread();
static void stop_thread(ThreadBase* thread);
// ATM, only interaction with a thread's allowed by other threads is
// through the queue_item call.
@@ -94,6 +97,8 @@ protected:
core::PollManager* m_pollManager;
rak::priority_queue_default m_taskScheduler;
rak::priority_item m_taskShutdown;
// 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;