TaskScheduler [7] SchedulerThreadPool

base/task_scheduler/thread_pool_unittest.cc

Index: base/task_scheduler/thread_pool_unittest.cc
diff --git a/base/task_scheduler/thread_pool_unittest.cc b/base/task_scheduler/thread_pool_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..d839b9eaeb76338e836ad2928bb3926808edc670
--- /dev/null
+++ b/base/task_scheduler/thread_pool_unittest.cc
@@ -0,0 +1,239 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "base/task_scheduler/thread_pool.h"
+
+#include <stddef.h>
+
+#include <unordered_set>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/synchronization/condition_variable.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/task_runner.h"
+#include "base/task_scheduler/scheduler_lock.h"
+#include "base/task_scheduler/task_tracker.h"
+#include "base/threading/platform_thread.h"
+#include "base/threading/simple_thread.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace base {
+namespace internal {
+namespace {
+
+const size_t kNumRepetitions = 4;
+const size_t kNumThreadsInThreadPool = 4;
+const size_t kNumTaskRunners = 4;
+const size_t kNumThreadsPostingTasksPerTaskRunner = 2;
+const size_t kNumTasksPostedPerThread = 150;
+
+class TaskSchedulerThreadPoolTest : public testing::Test {
+ protected:
+  TaskSchedulerThreadPoolTest() : cv_(lock_.CreateConditionVariable()) {}
+
+  void SetUp() override {
+    thread_pool_ = ThreadPool::CreateThreadPool(
+        ThreadPriority::NORMAL, kNumThreadsInThreadPool,
+        Bind(&TaskSchedulerThreadPoolTest::RanTaskFromSequenceCallback,
+             Unretained(this)),
+        &task_tracker_);
+    ASSERT_TRUE(thread_pool_);
+  }
+
+  void TearDown() override { thread_pool_->JoinForTesting(); }
+
+  // Waits for all tasks returned by CreateTask() to run.
+  void WaitForAllTasksToRun() {
+    AutoSchedulerLock auto_lock(lock_);
+    while (run_tasks_.size() < num_created_tasks_)
+      cv_->Wait();
+  }
+
+  size_t NumRunTasks() const {
+    AutoSchedulerLock auto_lock(lock_);
+    return run_tasks_.size();
+  }
+
+  scoped_ptr<ThreadPool> thread_pool_;
+
+ public:
+  // |task_runner| is the TaskRunner through which the task will be posted. If
+  // |post_nested_task| is true, the created task will post a new task through
+  // |task_runner| when it runs.
+  Closure CreateTask(scoped_refptr<TaskRunner> task_runner,
+                     bool post_nested_task) {
+    AutoSchedulerLock auto_lock(lock_);
+    return Bind(&TaskSchedulerThreadPoolTest::RunTaskCallback, Unretained(this),
+                num_created_tasks_++, task_runner, post_nested_task);
+  }
+
+ private:
+  void RanTaskFromSequenceCallback(const SchedulerWorkerThread* worker_thread,
+                                   scoped_refptr<Sequence> sequence) {
+    if (!sequence->PopTask()) {
+      const SequenceSortKey sort_key(sequence->GetSortKey());
+      thread_pool_->ReinsertSequence(worker_thread, std::move(sequence),
+                                     sort_key);
+    }
+  }
+
+  // |task_index| is a unique index for this task. |task_runner| is the
+  // TaskRunner through which the task was posted. If |post_nested_task| is
+  // true, the callback will post a new task through |task_runner|.
+  void RunTaskCallback(size_t task_index,
+                       scoped_refptr<TaskRunner> task_runner,
+                       bool post_nested_task) {
+    if (post_nested_task)
+      task_runner->PostTask(FROM_HERE, CreateTask(task_runner, false));
+
+    EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread());
+
+    AutoSchedulerLock auto_lock(lock_);
+
+    if (run_tasks_.find(task_index) != run_tasks_.end())
+      ADD_FAILURE() << "A task ran more than once.";
+    run_tasks_.insert(task_index);
+
+    cv_->Signal();
+  }
+
+  TaskTracker task_tracker_;
+
+  // Synchronizes access to all members below.
+  mutable SchedulerLock lock_;
+
+  // Condition variable signaled when a task completes its execution.
+  scoped_ptr<ConditionVariable> cv_;
+
+  // Number of tasks returned by CreateTask().
+  size_t num_created_tasks_ = 0;
+
+  // Indexes of tasks that ran.
+  std::unordered_set<size_t> run_tasks_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerThreadPoolTest);
+};
+
+class ThreadPostingTasks : public SimpleThread {
+ public:
+  // If |post_nested_task| is true, each task posted by this thread will post
+  // another task when it runs.
+  ThreadPostingTasks(scoped_refptr<TaskRunner> task_runner,
+                     bool post_nested_task,
+                     TaskSchedulerThreadPoolTest* test)
+      : SimpleThread("ThreadPostingTasks"),
+        task_runner_(std::move(task_runner)),
+        post_nested_task_(post_nested_task),
+        test_(test) {}
+
+ private:
+  void Run() override {
+    EXPECT_FALSE(task_runner_->RunsTasksOnCurrentThread());
+
+    for (size_t i = 0; i < kNumTasksPostedPerThread; ++i) {
+      task_runner_->PostTask(
+          FROM_HERE, test_->CreateTask(task_runner_, post_nested_task_));
+    }
+  }
+
+  scoped_refptr<TaskRunner> task_runner_;
+  const bool post_nested_task_;
+  TaskSchedulerThreadPoolTest* const test_;
+
+  DISALLOW_COPY_AND_ASSIGN(ThreadPostingTasks);
+};
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasks) {
+  // Repeat the test |kNumRepetitions| times, waiting for all
+  // SchedulerWorkerThreads to become idle between each repetition. This ensures
+  // that TaskRunners can wake up SchedulerWorkerThreads that have added
+  // themselves to the stack of idle SchedulerWorkerThreads of the ThreadPool.
+  for (size_t i = 0; i < kNumRepetitions; ++i) {
+    // Create |kNumTaskRunners| * |kNumThreadsPostingTasksPerTaskRunner| threads
+    // that will post tasks to |kNumTaskRunners| PARALLEL TaskRunners.
+    std::vector<scoped_ptr<ThreadPostingTasks>> threads_posting_tasks;
+    for (size_t j = 0; j < kNumTaskRunners; ++j) {
+      scoped_refptr<TaskRunner> task_runner =
+          thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                                   ExecutionMode::PARALLEL);
+      for (size_t k = 0; k < kNumThreadsPostingTasksPerTaskRunner; ++k) {
+        threads_posting_tasks.push_back(
+            make_scoped_ptr(new ThreadPostingTasks(task_runner, false, this)));
+        threads_posting_tasks.back()->Start();
+      }
+    }
+
+    for (const auto& thread_posting_tasks : threads_posting_tasks)
+      thread_posting_tasks->Join();
+
+    WaitForAllTasksToRun();
+    EXPECT_EQ((i + 1) * kNumTaskRunners * kNumThreadsPostingTasksPerTaskRunner *
+                  kNumTasksPostedPerThread,
+              NumRunTasks());
+    thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+  }
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWithNestedPostTasks) {
+  // Repeat the test |kNumRepetitions| times, waiting for all
+  // SchedulerWorkerThreads to become idle between each repetition. This ensures
+  // that TaskRunners can wake up SchedulerWorkerThreads that have added
+  // themselves to the stack of idle SchedulerWorkerThreads of the ThreadPool.
+  for (size_t i = 0; i < kNumRepetitions; ++i) {
+    // Create |kNumTaskRunners| * |kNumThreadsPostingTasksPerTaskRunner| threads
+    // that will post tasks to |kNumTaskRunners| PARALLEL TaskRunners. Each task
+    // posted by these threads will post another task when it runs.
+    std::vector<scoped_ptr<ThreadPostingTasks>> threads_posting_tasks;
+    for (size_t j = 0; j < kNumTaskRunners; ++j) {
+      auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+          TaskTraits(), ExecutionMode::PARALLEL);
+      for (size_t k = 0; k < kNumThreadsPostingTasksPerTaskRunner; ++k) {
+        threads_posting_tasks.push_back(
+            make_scoped_ptr(new ThreadPostingTasks(task_runner, true, this)));
+        threads_posting_tasks.back()->Start();
+      }
+    }
+
+    for (const auto& thread_posting_tasks : threads_posting_tasks)
+      thread_posting_tasks->Join();
+
+    WaitForAllTasksToRun();
+    EXPECT_EQ(2 * (i + 1) * kNumTaskRunners *
+                  kNumThreadsPostingTasksPerTaskRunner *
+                  kNumTasksPostedPerThread,
+              NumRunTasks());
+    thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+  }
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWithBlockedTasks) {
+  // Post |kNumThreadsInThreadPool| - 1 tasks that block until |event| is
+  // signaled.
+  WaitableEvent event(
+      true,    // Manual reset, to be able to wake up multiple waiters at once.
+      false);  // Not signaled initially.
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::PARALLEL);
+  for (size_t i = 0; i < (kNumThreadsInThreadPool - 1); ++i) {
+    task_runner->PostTask(FROM_HERE,
+                          Bind(&WaitableEvent::Wait, Unretained(&event)));
+  }
+
+  // Post |kNumTasksPostedPerThread| tasks that should run despite the fact
+  // that all threads in |thread_pool_| are busy except one.
+  for (size_t i = 0; i < kNumTasksPostedPerThread; ++i)
+    task_runner->PostTask(FROM_HERE, CreateTask(task_runner, false));
+
+  WaitForAllTasksToRun();
+  event.Signal();
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}

[robliao, 2016/03/31 22:48:56]

Would it be useful to have a test that saturates the threadpool with a long
running task (e.g. event wait) to verify we're able to wake up all the threads?

[fdoray, 2016/04/01 16:02:52]

Done. TaskSchedulerThreadPoolTest.Saturate

+
+}  // namespace
+}  // namespace internal
+}  // namespace base