Refactored thread pool

runtime/bin/thread_pool.cc

Index: runtime/bin/thread_pool.cc
diff --git a/runtime/bin/thread_pool.cc b/runtime/bin/thread_pool.cc
index cf1ec3b9d4cc941322d0857bc81f98f987613489..581945f11e1f9cb5a60e452e09d455b0636262b7 100644
--- a/runtime/bin/thread_pool.cc
+++ b/runtime/bin/thread_pool.cc
@@ -6,9 +6,11 @@
 
 #include "bin/thread.h"
 
-void TaskQueue::Insert(TaskQueueEntry* entry) {
+
+bool ThreadPool::InsertTask(Task task) {
+  TaskQueueEntry* entry = new TaskQueueEntry(task);
   MonitorLocker monitor(&monitor_);

[Mads Ager (google), 2012/01/24 12:29:39]

MonitorLocker locker(&monitor_); ?

Similarly for the other ones?

[Søren Gjesse, 2012/01/24 13:04:35]

Done.

-  monitor_.Enter();
+  if (terminate_) return false;
   if (head_ == NULL) {
     head_ = entry;
     tail_ = entry;
@@ -17,20 +19,19 @@ void TaskQueue::Insert(TaskQueueEntry* entry) {
     tail_->set_next(entry);
     tail_ = entry;
   }
+  return true;
 }
 
 
-TaskQueueEntry* TaskQueue::Remove() {
+ThreadPool::TaskQueueEntry* ThreadPool::WaitForTask() {
   MonitorLocker monitor(&monitor_);
+  if (terminate_ && !drain_) return NULL;
   TaskQueueEntry* result = head_;
   while (result == NULL) {
-    if (terminate_) {
+    if (terminate_ && (!drain_ || result == NULL)) {

[Mads Ager (google), 2012/01/24 12:29:39]

I don't understand this condition. Shouldn't this be

if (terminate_ && !drain_)

If you have drain_ set to true and you get a spurious wakeup from the monitor
you can get here with: terminate_ == true, drain_ == true and result == NULL. In
that case you will terminate without draining the queue?

[Søren Gjesse, 2012/01/24 13:04:35]

Good catch! result would always be NULL here.

Moved the condition to be just after Wait an changed it to:

if (terminate_ && (drain_flag_ == kDoNotDrain || head_ == NULL))

This should mean "if terminating return NULL if either not draining or queue is
empty".

       return NULL;
     }
     monitor.Wait();
-    if (terminate_) {
-      return NULL;
-    }
     result = head_;
   }
   head_ = result->next();
@@ -39,46 +40,26 @@ TaskQueueEntry* TaskQueue::Remove() {
 }
 
 
-void TaskQueue::Shutdown() {
-  MonitorLocker monitor(&monitor_);
-  terminate_ = true;
-  monitor.NotifyAll();
-}
-
-
-void ThreadPool::InsertTask(Task task) {
-  TaskQueueEntry* entry = new TaskQueueEntry(task);
-  queue_.Insert(entry);
-}
-
-
-Task ThreadPool::WaitForTask() {
-  TaskQueueEntry* entry = queue_.Remove();
-  if (entry == NULL) {
-    return NULL;
-  }
-  Task task = entry->task();
-  delete entry;
-  return task;
-}
-
-
 void ThreadPool::Start() {
   MonitorLocker monitor(&monitor_);
-  for (int i = 0; i < size_; i++) {
+  terminate_ = false;
+  drain_ = false;
+  for (int i = 0; i < initial_size_; i++) {
     int result = dart::Thread::Start(&ThreadPool::Main,
                                      reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start thread pool thread %d", result);
     }
+    size_++;
   }
 }
 
 
-void ThreadPool::Shutdown() {
-  terminate_ = true;
-  queue_.Shutdown();
+void ThreadPool::Shutdown(bool drain) {
   MonitorLocker monitor(&monitor_);
+  terminate_ = true;
+  drain_ = drain;
+  monitor.NotifyAll();
   while (size_ > 0) {
     monitor.Wait();
   }
@@ -97,13 +78,13 @@ void ThreadPool::Main(uword args) {
     printf("Thread pool thread started\n");
   }
   ThreadPool* pool = reinterpret_cast<ThreadPool*>(args);
-  while (!pool->terminate_) {
+  while (true) {
     if (Dart_IsVMFlagSet("trace_thread_pool")) {
       printf("Waiting for task\n");
     }
-    Task task = pool->WaitForTask();
-    if (pool->terminate_) break;
-    (*(pool->task_handler_))(task);
+    TaskQueueEntry* task = pool->WaitForTask();
+    if (task == NULL) break;
+    (*(pool->task_handler_))(task->task());
   }
   pool->ThreadTerminated();
 };