diff --git a/src/control.cc b/src/control.cc index 01825de7..652a25dd 100644 --- a/src/control.cc +++ b/src/control.cc @@ -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); } diff --git a/src/control.h b/src/control.h index 06b7e871..4b08c28a 100644 --- a/src/control.h +++ b/src/control.h @@ -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; } diff --git a/src/main.cc b/src/main.cc index 40c03c40..6e2ab348 100644 --- a/src/main.cc +++ b/src/main.cc @@ -340,6 +340,7 @@ main(int argc, char** argv) { } delete control; + delete worker_thread; delete main_thread; return 0; diff --git a/src/thread_base.cc b/src/thread_base.cc index 520dedb2..8365b168 100644 --- a/src/thread_base.cc +++ b/src/thread_base.cc @@ -44,6 +44,8 @@ #include #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. } diff --git a/src/thread_base.h b/src/thread_base.h index 51667519..f21aa0c9 100644 --- a/src/thread_base.h +++ b/src/thread_base.h @@ -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;