VM thread shutdown.

runtime/vm/thread_pool.cc

Index: runtime/vm/thread_pool.cc
diff --git a/runtime/vm/thread_pool.cc b/runtime/vm/thread_pool.cc
index 5b3a713d2d09b2c32be7605f622e7671669e36c6..489acef4a8acd7c81a71d3d2d871a29c48b0dbd6 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,9 @@ ThreadPool::ThreadPool()
     count_started_(0),
     count_stopped_(0),
     count_running_(0),
-    count_idle_(0) {
+    count_idle_(0),
+    shutting_down_workers_(NULL),
+    join_list_(NULL) {
 }
 
 
@@ -31,7 +30,7 @@ ThreadPool::~ThreadPool() {
 }
 
 
-void ThreadPool::Run(Task* task) {
+bool ThreadPool::Run(Task* task) {
   Worker* worker = NULL;
   bool new_worker = false;
   {
@@ -39,7 +38,7 @@ void ThreadPool::Run(Task* task) {
     // ThreadPool state.
     MutexLocker ml(&mutex_);
     if (shutting_down_) {
-      return;
+      return false;
     }
     if (idle_workers_ == NULL) {
       worker = new Worker(this);
@@ -51,15 +50,17 @@ void ThreadPool::Run(Task* task) {
       worker->all_next_ = all_workers_;
       all_workers_ = worker;
       worker->owned_ = true;
+      count_running_++;
     } else {
       // Get the first worker from the idle worker list.
       worker = idle_workers_;
       idle_workers_ = worker->idle_next_;
       worker->idle_next_ = NULL;
       count_idle_--;
+      count_running_++;
     }
-    count_running_++;
   }
+
   // Release ThreadPool::mutex_ before calling Worker functions.
   ASSERT(worker != NULL);
   worker->SetTask(task);
@@ -67,6 +68,7 @@ void ThreadPool::Run(Task* task) {
     // Call StartThread after we've assigned the first task.
     worker->StartThread();
   }
+  return true;
 }
 
 
@@ -94,14 +96,30 @@ 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()) {

[Ivan Posva, 2015/09/14 21:52:14]

Cache the current thread it outside the loop?

[zra, 2015/09/14 22:59:01]

Done.

+        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.

[Ivan Posva, 2015/09/14 21:52:13]

Move outside the exit_monitor_ block.

Verify that no workers are added to the join list and synchronize properly on
the required lock, even if not exactly necessary during shutdown as we should be
the only worker running now...

[zra, 2015/09/14 22:59:01]

Done.

+    JoinList::Join(&join_list_);
   }
 }
 
@@ -156,7 +174,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;
   }
 
@@ -174,16 +192,22 @@ bool ThreadPool::RemoveWorkerFromAllList(Worker* worker) {
 }
 
 
-void ThreadPool::SetIdle(Worker* worker) {
-  MutexLocker ml(&mutex_);
-  if (shutting_down_) {
-    return;
+void ThreadPool::SetIdleAndReapExited(Worker* worker) {
+  JoinList* list = NULL;
+  {
+    MutexLocker ml(&mutex_);
+    if (shutting_down_) {
+      return;
+    }
+    ASSERT(worker->owned_ && !IsIdle(worker));
+    worker->idle_next_ = idle_workers_;
+    idle_workers_ = worker;
+    count_idle_++;
+    count_running_--;
+    list = join_list_;
+    join_list_ = NULL;
   }
-  ASSERT(worker->owned_ && !IsIdle(worker));
-  worker->idle_next_ = idle_workers_;
-  idle_workers_ = worker;
-  count_idle_++;
-  count_running_--;
+  JoinList::Join(&list);
 }
 
 
@@ -200,12 +224,62 @@ bool ThreadPool::ReleaseIdleWorker(Worker* worker) {
   bool found = RemoveWorkerFromAllList(worker);
   ASSERT(found);
 
+  // The thread for worker will exit. Add its ThreadId to the join_list_
+  // so that we can join on it at the next opportunity.
+  JoinList::Add(worker->join_id_, &join_list_);
   count_stopped_++;
   count_idle_--;
   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(ThreadJoinId id, JoinList** list) {

[Ivan Posva, 2015/09/14 21:52:13]

AddLocked(). Ideally we want to assert that we hold the ThreadPool mutex.

[zra, 2015/09/14 22:59:01]

Done.

+  *list = new JoinList(id, *list);
+}
+
+
+void ThreadPool::JoinList::Join(JoinList** list) {
+  while (*list) {

[Ivan Posva, 2015/09/14 21:52:14]

*list is not a bool

[zra, 2015/09/14 22:59:01]

Done.

+    JoinList* current = *list;
+    *list = current->next();
+    OSThread::Join(current->id());
+    delete current;
+  }
+}
+
+
 ThreadPool::Task::Task() {
 }
 
@@ -217,9 +291,13 @@ ThreadPool::Task::~Task() {
 ThreadPool::Worker::Worker(ThreadPool* pool)
   : pool_(pool),
     task_(NULL),
+    id_(OSThread::kInvalidThreadId),
+    join_id_(OSThread::kInvalidThreadJoinId),
+    done_(false),
     owned_(false),
     all_next_(NULL),
-    idle_next_(NULL) {
+    idle_next_(NULL),
+    shutdown_next_(NULL) {
 }
 
 
@@ -264,7 +342,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,10 +359,10 @@ void ThreadPool::Worker::Loop() {
 
     ASSERT(task_ == NULL);
     if (IsDone()) {
-      return;
+      return false;
     }
-    ASSERT(pool_ != NULL);
-    pool_->SetIdle(this);
+    ASSERT(!done_);
+    pool_->SetIdleAndReapExited(this);
     idle_start = OS::GetCurrentTimeMillis();
     while (true) {
       Monitor::WaitResult result = ml.Wait(ComputeTimeout(idle_start));
@@ -294,21 +372,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 +395,37 @@ void ThreadPool::Worker::Shutdown() {
 void ThreadPool::Worker::Main(uword args) {
   Thread::EnsureInit();
   Worker* worker = reinterpret_cast<Worker*>(args);
-  worker->Loop();
+
+  {
+    MonitorLocker ml(&(worker->monitor_));
+    ASSERT(worker->task_);
+    worker->id_ = OSThread::GetCurrentThreadId();
+    worker->join_id_ = OSThread::GetCurrentThreadJoinId();
+  }
+
+  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->join_id_, &worker->pool_->join_list_);

[Ivan Posva, 2015/09/14 21:52:14]

JoinList should be consistent on the lock which protects it. Once we added this
thread's id, we can proceed with the exit.

[zra, 2015/09/14 22:59:01]

Done.

+    worker->id_ = OSThread::kInvalidThreadId;
+    worker->join_id_ = OSThread::kInvalidThreadJoinId;
+    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.

[Ivan Posva, 2015/09/14 21:52:13]

Please add comment: If there is a shutdown race, then we will be blocked on this
thread being joined before we entirely unload.

[zra, 2015/09/14 22:59:01]

Done.

+    delete worker;
   }
-  delete worker;
 #if defined(TARGET_OS_WINDOWS)
   Thread::CleanUp();
 #endif