Index: runtime/vm/thread_pool.cc |
diff --git a/runtime/vm/thread_pool.cc b/runtime/vm/thread_pool.cc |
index 5b3a713d2d09b2c32be7605f622e7671669e36c6..d3afe3a4a6429dda7153468caeb8a5942bb4d550 100644 |
--- a/runtime/vm/thread_pool.cc |
+++ b/runtime/vm/thread_pool.cc |
@@ -12,9 +12,6 @@ namespace dart { |
DEFINE_FLAG(int, worker_timeout_millis, 5000, |
"Free workers when they have been idle for this amount of time."); |
-Monitor* ThreadPool::exit_monitor_ = NULL; |
-int* ThreadPool::exit_count_ = NULL; |
- |
ThreadPool::ThreadPool() |
: shutting_down_(false), |
all_workers_(NULL), |
@@ -22,7 +19,10 @@ ThreadPool::ThreadPool() |
count_started_(0), |
count_stopped_(0), |
count_running_(0), |
- count_idle_(0) { |
+ count_idle_(0), |
+ idle_join_list_(NULL), |
+ shutting_down_workers_(NULL), |
+ join_list_(NULL) { |
} |
@@ -94,15 +94,34 @@ void ThreadPool::Shutdown() { |
} |
// Release ThreadPool::mutex_ before calling Worker functions. |
- Worker* current = saved; |
- while (current != NULL) { |
- // We may access all_next_ without holding ThreadPool::mutex_ here |
- // because the worker is no longer owned by the ThreadPool. |
- Worker* next = current->all_next_; |
- current->all_next_ = NULL; |
- current->Shutdown(); |
- current = next; |
+ { |
+ MonitorLocker eml(&exit_monitor_); |
+ |
+ // First tell all the workers to shut down. |
+ Worker* current = saved; |
+ while (current != NULL) { |
+ Worker* next = current->all_next_; |
+ if (current->id_ != OSThread::GetCurrentThreadId()) { |
+ AddWorkerToShutdownList(current); |
+ } |
+ current->Shutdown(); |
+ current = next; |
+ } |
+ saved = NULL; |
+ |
+ // Wait until all workers will exit. |
+ while (shutting_down_workers_ != NULL) { |
+ // Here, we are waiting for workers to exit. When a worker exits we will |
+ // be notified. |
+ eml.Wait(); |
+ } |
+ |
+ // Join non-idle threads. |
+ JoinList::Join(&join_list_); |
} |
+ |
+ // Join any remaining idle threads. |
+ JoinList::Join(&idle_join_list_); |
} |
@@ -156,7 +175,7 @@ bool ThreadPool::RemoveWorkerFromAllList(Worker* worker) { |
all_workers_ = worker->all_next_; |
worker->all_next_ = NULL; |
worker->owned_ = false; |
- worker->pool_ = NULL; |
+ worker->done_ = true; |
return true; |
} |
@@ -187,25 +206,118 @@ void ThreadPool::SetIdle(Worker* worker) { |
} |
-bool ThreadPool::ReleaseIdleWorker(Worker* worker) { |
- MutexLocker ml(&mutex_); |
- if (shutting_down_) { |
- return false; |
+void ThreadPool::ReapExitedIdleThreads() { |
+ JoinList* list = NULL; |
+ { |
+ MutexLocker ml(&mutex_); |
+ if (shutting_down_) { |
+ // If we're shutting down, the threads will be reaped in Shutdown. |
+ return; |
+ } |
+ list = idle_join_list_; |
+ idle_join_list_ = NULL; |
} |
- // Remove from idle list. |
- if (!RemoveWorkerFromIdleList(worker)) { |
- return false; |
+ JoinList::Join(&list); |
+} |
+ |
+ |
+class ReaperTask : public ThreadPool::Task { |
+ public: |
+ explicit ReaperTask(ThreadPool* pool) : pool_(pool) {} |
+ virtual void Run() { |
+ pool_->ReapExitedIdleThreads(); |
} |
- // Remove from all list. |
- bool found = RemoveWorkerFromAllList(worker); |
- ASSERT(found); |
- count_stopped_++; |
- count_idle_--; |
+ private: |
+ ThreadPool* pool_; |
+}; |
+ |
+ |
+bool ThreadPool::ReleaseIdleWorker(Worker* worker) { |
+ Worker* idle_worker = NULL; |
+ { |
+ MutexLocker ml(&mutex_); |
+ if (shutting_down_) { |
+ return false; |
+ } |
+ // Remove from idle list. |
+ if (!RemoveWorkerFromIdleList(worker)) { |
+ return false; |
+ } |
+ // Remove from all list. |
+ bool found = RemoveWorkerFromAllList(worker); |
+ ASSERT(found); |
+ |
+ // The thread for worker will exit. Add its ThreadId to the idle_join_list_ |
+ // so that we can join on it at the next opportunity. |
+ JoinList::Add(worker->id_, &idle_join_list_); |
+ |
+ // If there's an idle worker hanging around, set it up to reap the exiting |
+ // idle thread. |
+ if (idle_workers_ != NULL) { |
+ Worker* idle_worker = idle_workers_; |
+ idle_workers_ = idle_workers_->idle_next_; |
+ idle_worker->idle_next_ = NULL; |
+ count_idle_--; |
+ } |
+ |
+ count_stopped_++; |
+ count_idle_--; |
+ } |
+ if (idle_worker != NULL) { |
+ idle_worker->SetTask(new ReaperTask(this)); |
+ } |
return true; |
} |
+// Only call while holding the exit_monitor_ |
+void ThreadPool::AddWorkerToShutdownList(Worker* worker) { |
+ worker->shutdown_next_ = shutting_down_workers_; |
+ shutting_down_workers_ = worker; |
+} |
+ |
+ |
+// Only call while holding the exit_monitor_ |
+bool ThreadPool::RemoveWorkerFromShutdownList(Worker* worker) { |
+ ASSERT(worker != NULL); |
+ ASSERT(shutting_down_workers_ != NULL); |
+ |
+ // Special case head of list. |
+ if (shutting_down_workers_ == worker) { |
+ shutting_down_workers_ = worker->shutdown_next_; |
+ worker->shutdown_next_ = NULL; |
+ return true; |
+ } |
+ |
+ for (Worker* current = shutting_down_workers_; |
+ current->shutdown_next_ != NULL; |
+ current = current->shutdown_next_) { |
+ if (current->shutdown_next_ == worker) { |
+ current->shutdown_next_ = worker->shutdown_next_; |
+ worker->shutdown_next_ = NULL; |
+ return true; |
+ } |
+ } |
+ return false; |
+} |
+ |
+ |
+void ThreadPool::JoinList::Add(ThreadId id, JoinList** list) { |
+ *list = new JoinList(id, *list); |
+} |
+ |
+ |
+void ThreadPool::JoinList::Join(JoinList** list) { |
+ while (*list) { |
+ JoinList* current = *list; |
+ *list = current->next(); |
+ OSThread::Join(current->id()); |
+ delete current; |
+ } |
+} |
+ |
+ |
ThreadPool::Task::Task() { |
} |
@@ -216,10 +328,14 @@ ThreadPool::Task::~Task() { |
ThreadPool::Worker::Worker(ThreadPool* pool) |
: pool_(pool), |
+ done_(false), |
task_(NULL), |
+ id_(OSThread::kInvalidThreadId), |
+ started_(false), |
owned_(false), |
all_next_(NULL), |
- idle_next_(NULL) { |
+ idle_next_(NULL), |
+ shutdown_next_(NULL) { |
} |
@@ -264,7 +380,7 @@ static int64_t ComputeTimeout(int64_t idle_start) { |
} |
-void ThreadPool::Worker::Loop() { |
+bool ThreadPool::Worker::Loop() { |
MonitorLocker ml(&monitor_); |
int64_t idle_start; |
while (true) { |
@@ -281,9 +397,10 @@ void ThreadPool::Worker::Loop() { |
ASSERT(task_ == NULL); |
if (IsDone()) { |
- return; |
+ return false; |
} |
- ASSERT(pool_ != NULL); |
+ ASSERT(!done_); |
+ pool_->ReapExitedIdleThreads(); |
Ivan Posva
2015/08/17 13:35:52
There is no need to hold the monitor_ when calling
zra
2015/08/18 06:23:14
Done.
|
pool_->SetIdle(this); |
idle_start = OS::GetCurrentTimeMillis(); |
while (true) { |
@@ -294,21 +411,21 @@ void ThreadPool::Worker::Loop() { |
break; |
} |
if (IsDone()) { |
- return; |
+ return false; |
} |
- if (result == Monitor::kTimedOut && |
- pool_->ReleaseIdleWorker(this)) { |
- return; |
+ if ((result == Monitor::kTimedOut) && pool_->ReleaseIdleWorker(this)) { |
+ return true; |
} |
} |
} |
UNREACHABLE(); |
+ return false; |
} |
void ThreadPool::Worker::Shutdown() { |
MonitorLocker ml(&monitor_); |
- pool_ = NULL; // Fail fast if someone tries to access pool_. |
+ done_ = true; |
ml.Notify(); |
} |
@@ -317,20 +434,58 @@ void ThreadPool::Worker::Shutdown() { |
void ThreadPool::Worker::Main(uword args) { |
Thread::EnsureInit(); |
Worker* worker = reinterpret_cast<Worker*>(args); |
- worker->Loop(); |
+ bool delete_self = false; |
+ |
+ { |
+ MonitorLocker ml(&(worker->monitor_)); |
+ if (worker->IsDone()) { |
+ // id_ hasn't been set yet, but the ThreadPool is being shutdown. |
+ // Delete the task, and return. |
+ ASSERT(worker->task_); |
+ delete worker->task_; |
+ worker->task_ = NULL; |
+ delete_self = true; |
+ } else { |
+ worker->id_ = OSThread::GetCurrentThreadId(); |
+ worker->started_ = true; |
+ } |
+ } |
+ |
+ // We aren't able to delete the worker while holding the worker's monitor. |
+ // Now that we have released it, and we know that ThreadPool::Shutdown |
+ // won't touch it again, we can delete it and return. |
+ if (delete_self) { |
+ MonitorLocker eml(&worker->pool_->exit_monitor_); |
+ ThreadPool::JoinList::Add( |
+ OSThread::GetCurrentThreadId(), &worker->pool_->join_list_); |
+ worker->pool_->RemoveWorkerFromShutdownList(worker); |
+ delete worker; |
+ eml.Notify(); |
+ return; |
+ } |
+ |
+ bool released = worker->Loop(); |
// It should be okay to access these unlocked here in this assert. |
- ASSERT(!worker->owned_ && |
- worker->all_next_ == NULL && |
- worker->idle_next_ == NULL); |
- |
- // The exit monitor is only used during testing. |
- if (ThreadPool::exit_monitor_) { |
- MonitorLocker ml(ThreadPool::exit_monitor_); |
- (*ThreadPool::exit_count_)++; |
- ml.Notify(); |
+ // worker->all_next_ is retained by the pool for shutdown monitoring. |
+ ASSERT(!worker->owned_ && (worker->idle_next_ == NULL)); |
+ |
+ if (!released) { |
+ // This worker is exiting because the thread pool is being shut down. |
+ // Inform the thread pool that we are exiting. We remove this worker from |
+ // shutting_down_workers_ list because there will be no need for the |
+ // ThreadPool to take action for this worker. |
+ MonitorLocker eml(&worker->pool_->exit_monitor_); |
+ JoinList::Add(worker->id_, &worker->pool_->join_list_); |
+ worker->id_ = OSThread::kInvalidThreadId; |
+ worker->pool_->RemoveWorkerFromShutdownList(worker); |
+ delete worker; |
+ eml.Notify(); |
+ } else { |
+ // This worker is going down because it was idle for too long. This case |
+ // is not due to a ThreadPool Shutdown. Thus, we simply delete the worker. |
+ delete worker; |
} |
- delete worker; |
#if defined(TARGET_OS_WINDOWS) |
Thread::CleanUp(); |
#endif |