Implement ThreadPool.

runtime/vm/thread_pool_test.cc

+// Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+#include "vm/thread.h"
+#include "vm/thread_pool.h"
+#include "vm/unit_test.h"
+
+namespace dart {
+
+DECLARE_FLAG(int, worker_timeout_millis);
+
+UNIT_TEST_CASE(ThreadPool_Create) {
+  ThreadPool thread_pool;
+}
+
+
+class TestTask : public ThreadPool::Task {
+ public:
+  TestTask(Monitor* sync, bool* done)
+      : sync_(sync), done_(done) {
+  }
+
+  void Run() {
+    MonitorLocker ml(sync_);
+    *done_ = true;
+    ml.Notify();
+  }
+
+ private:
+  Monitor* sync_;
+  bool* done_;
+};
+
+
+UNIT_TEST_CASE(ThreadPool_RunOne) {
+  ThreadPool thread_pool;
+  Monitor sync;
+  bool done = false;
+  thread_pool.Run(new TestTask(&sync, &done));
+  {
+    MonitorLocker ml(&sync);
+    while (!done) {
+      ml.Wait();
+    }
+  }
+  EXPECT(done);
+}
+
+
+UNIT_TEST_CASE(ThreadPool_RunMany) {
+  const int kTaskCount = 100;
+  ThreadPool thread_pool;
+  Monitor sync[kTaskCount];
+  bool done[kTaskCount];
+
+  for (int i = 0; i < kTaskCount; i++) {
+    done[i] = false;
+    thread_pool.Run(new TestTask(&sync[i], &done[i]));
+  }
+  for (int i = 0; i < kTaskCount; i++) {
+    MonitorLocker ml(&sync[i]);
+    while (!done[i]) {
+      ml.Wait();
+    }
+    EXPECT(done[i]);
+  }

[siva, 2012/03/14 17:36:09]

Shutdown of the thread pool is done as part of the destructor but the test does
not verify that all the threads did indeed shutdown.

Would it be better if you did an explicit shutdown of the thread pool and then
checked the count_stopped, count_running, count_idle values.

[turnidge, 2012/03/14 21:00:27]

I've added a test that workers eventually shut themselves down.  I added a
private ThreadPool::exit_monitor_ which is only set/used during testing and I'm
using this monitor to notify the test just before workers exit.

+}
+
+
+UNIT_TEST_CASE(ThreadPool_WorkerTimeout) {
+  // Adjust the worker timeout so that we timeout quickly.
+  int saved_timeout = FLAG_worker_timeout_millis;
+  FLAG_worker_timeout_millis = 1;
+
+  ThreadPool thread_pool;
+  EXPECT_EQ(0, thread_pool.workers_started());
+  EXPECT_EQ(0, thread_pool.workers_stopped());
+
+  // Run a worker.
+  Monitor sync;
+  bool done = false;
+  thread_pool.Run(new TestTask(&sync, &done));
+  EXPECT_EQ(1, thread_pool.workers_started());
+  EXPECT_EQ(0, thread_pool.workers_stopped());
+  {
+    MonitorLocker ml(&sync);
+    while (!done) {
+      ml.Wait();
+    }
+  }
+  EXPECT(done);
+
+  // Wait up to 5 seconds to see if a worker times out.
+  const int kMaxWait = 5000;
+  int waited = 0;
+  while (thread_pool.workers_stopped() == 0 && waited < kMaxWait) {
+    OS::Sleep(1);
+    waited += 1;
+  }
+  EXPECT_EQ(1, thread_pool.workers_stopped());
+  FLAG_worker_timeout_millis = saved_timeout;
+}
+
+
+class SpawnTask : public ThreadPool::Task {
+ public:
+  SpawnTask(ThreadPool* pool, Monitor* sync, int todo, int total, int* done)
+      : pool_(pool), sync_(sync), todo_(todo), total_(total), done_(done) {
+  }
+
+  void Run() {
+    todo_--;  // Subtract one for current task.
+    int child_todo = todo_ / 2;
+
+    // Spawn 0-2 children.
+    if (todo_ > 0) {
+      pool_->Run(
+          new SpawnTask(pool_, sync_, todo_ - child_todo, total_, done_));
+    }
+    if (todo_ > 1) {
+      pool_->Run(
+          new SpawnTask(pool_, sync_, child_todo, total_, done_));
+    }
+
+    {
+      MonitorLocker ml(sync_);
+      (*done_)++;
+      if (*done_ >= total_) {
+        ml.Notify();
+      }
+    }
+  }
+
+ private:
+  ThreadPool* pool_;
+  Monitor* sync_;
+  int todo_;
+  int total_;
+  int* done_;
+};
+
+
+UNIT_TEST_CASE(ThreadPool_RecursiveSpawn) {
+  ThreadPool thread_pool;
+  Monitor sync;
+  const int kTotalTasks = 500;
+  int done = 0;
+  thread_pool.Run(
+      new SpawnTask(&thread_pool, &sync, kTotalTasks, kTotalTasks, &done));
+  {
+    MonitorLocker ml(&sync);
+    while (done < kTotalTasks) {
+      ml.Wait();
+    }
+  }
+  EXPECT_EQ(kTotalTasks, done);
+}
+
+
+}  // namespace dart