TaskScheduler [7] SchedulerThreadPool

base/task_scheduler/scheduler_thread_pool_unittest.cc

+// 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/scheduler_thread_pool.h"
+
+#include <stddef.h>
+
+#include <memory>
+#include <unordered_set>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/macros.h"
+#include "base/memory/ptr_util.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 kNumThreadsPostingTasks = 4;
+const size_t kNumTasksPostedPerThread = 150;
+
+class TaskSchedulerThreadPoolTest : public testing::Test {
+ protected:
+  TaskSchedulerThreadPoolTest() : cv_(lock_.CreateConditionVariable()) {}
+
+  void SetUp() override {
+    thread_pool_ = SchedulerThreadPool::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 start running. It is not
+  // guaranteed that the tasks have completed their execution when this returns.
+  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();
+  }
+
+  std::unique_ptr<SchedulerThreadPool> thread_pool_;
+
+ public:
+  // Creates a task. |task_runner| is the TaskRunner through which the task will
+  // be posted. If |post_nested_task| is true, the task will post a new task
+  // through |task_runner| when it runs. If |event| is set, the task will block
+  // until |event| is signaled.
+  Closure CreateTask(scoped_refptr<TaskRunner> task_runner,
+                     bool post_nested_task,
+                     WaitableEvent* event) {
+    AutoSchedulerLock auto_lock(lock_);
+    return Bind(&TaskSchedulerThreadPoolTest::RunTaskCallback, Unretained(this),
+                num_created_tasks_++, task_runner, post_nested_task,
+                Unretained(event));
+  }
+
+ private:
+  void RanTaskFromSequenceCallback(const SchedulerWorkerThread* worker_thread,
+                                   scoped_refptr<Sequence> sequence) {
+    if (!sequence->PopTask()) {
+      const SequenceSortKey sort_key(sequence->GetSortKey());
+      thread_pool_->ReinsertSequence(std::move(sequence), sort_key);
+    }
+  }
+
+  void RunTaskCallback(size_t task_index,
+                       scoped_refptr<TaskRunner> task_runner,
+                       bool post_nested_task,
+                       WaitableEvent* event) {
+    if (post_nested_task)
+      task_runner->PostTask(FROM_HERE, CreateTask(task_runner, false, nullptr));
+
+    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();
+    }
+
+    if (event)
+      event->Wait();
+  }
+
+  TaskTracker task_tracker_;
+
+  // Synchronizes access to all members below.
+  mutable SchedulerLock lock_;
+
+  // Condition variable signaled when a task completes its execution.
+  std::unique_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_, nullptr));
+    }
+  }
+
+  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 worker threads that have added themselves to
+  // the stack of idle threads of their thread pool.
+  for (size_t i = 0; i < kNumRepetitions; ++i) {
+    // Create |kNumTaskRunners| * |kNumThreadsPostingTasksPerTaskRunner| threads
+    // that will post tasks to |kNumTaskRunners| PARALLEL TaskRunners.
+    std::vector<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
+    for (size_t j = 0; j < kNumThreadsPostingTasks; ++j) {
+      scoped_refptr<TaskRunner> task_runner =
+          thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                                   ExecutionMode::PARALLEL);
+      threads_posting_tasks.push_back(
+          WrapUnique(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) * kNumThreadsPostingTasks * 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 worker threads that have added themselves to
+  // the stack of idle threads of their thread pool.
+  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<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
+    for (size_t j = 0; j < kNumThreadsPostingTasks; ++j) {
+      auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+          TaskTraits(), ExecutionMode::PARALLEL);
+      threads_posting_tasks.push_back(
+          WrapUnique(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) * kNumThreadsPostingTasks * kNumTasksPostedPerThread,
+              NumRunTasks());
+    thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+  }
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWithOneAvailableThread) {
+  // Post tasks to keep all threads busy except one until |event| is signaled.
+  WaitableEvent event(
+      true,    // Manual reset, to wake up multiple waiters at once.

[robliao, 2016/04/01 21:05:51]

Nit: I think we can omit these comments. The premise being that readers are
considered to be "experts" in this code.

It's unfortunate, but this is why I prefer enums to booleans for these cases :-)
.

[fdoray, 2016/04/01 21:45:30]

Done.

+      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, CreateTask(task_runner, false, &event));
+  WaitForAllTasksToRun();
+
+  // Post |kNumTasksPostedPerThread| tasks that should run despite the fact
+  // that only one thread in |thread_pool_| isn't busy.
+  for (size_t i = 0; i < kNumTasksPostedPerThread; ++i)
+    task_runner->PostTask(FROM_HERE, CreateTask(task_runner, false, nullptr));
+  WaitForAllTasksToRun();
+
+  // Release the tasks waiting on |event|.
+  event.Signal();
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, Saturate) {
+  // Verify that it is possible to have |kNumThreadsInThreadPool| tasks running
+  // simultaneously.
+  WaitableEvent event(
+      true,    // Manual reset, 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; ++i)
+    task_runner->PostTask(FROM_HERE, CreateTask(task_runner, false, &event));
+  WaitForAllTasksToRun();
+  event.Signal();
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}
+
+// Checks that when PostTaskHelper is called with an empty sequence, the task
+// is added to the sequence and the sequence is added to the priority queue.
+TEST(TaskSchedulerPostTaskHelperTest, PostTaskInEmptySequence) {
+  std::unique_ptr<Task> task(
+      new Task(FROM_HERE, Bind(&DoNothing), TaskTraits()));
+  const Task* task_raw = task.get();
+  scoped_refptr<Sequence> sequence(new Sequence);
+  PriorityQueue priority_queue(Bind(&DoNothing));
+  TaskTracker task_tracker;
+
+  // Post |task|.
+  PostTaskHelper(std::move(task), sequence, &priority_queue, &task_tracker);
+
+  // Expect to find the sequence in the priority queue.
+  EXPECT_EQ(sequence, priority_queue.BeginTransaction()->Peek().sequence);
+
+  // Expect to find |task| alone in |sequence|.
+  EXPECT_EQ(task_raw, sequence->PeekTask());
+  sequence->PopTask();
+  EXPECT_EQ(nullptr, sequence->PeekTask());
+}
+
+// Checks that when PostTaskHelper is called with a sequence that already
+// contains a task, the task is added to the sequence but the sequence is not
+// added to the priority queue.
+TEST(TaskSchedulerPostTaskHelperTest, PostTaskInNonEmptySequence) {
+  std::unique_ptr<Task> task(
+      new Task(FROM_HERE, Bind(&DoNothing), TaskTraits()));
+  const Task* task_raw = task.get();
+  scoped_refptr<Sequence> sequence(new Sequence);
+  PriorityQueue priority_queue(Bind(&DoNothing));
+  TaskTracker task_tracker;
+
+  // Add an initial task in |sequence|.
+  sequence->PushTask(
+      WrapUnique(new Task(FROM_HERE, Bind(&DoNothing), TaskTraits())));
+
+  // Post |task|.
+  PostTaskHelper(std::move(task), sequence, &priority_queue, &task_tracker);
+
+  // Expect to find the priority queue empty.
+  EXPECT_TRUE(priority_queue.BeginTransaction()->Peek().is_null());
+
+  // Expect to find |task| in |sequence| behind the initial task.
+  EXPECT_NE(task_raw, sequence->PeekTask());
+  sequence->PopTask();
+  EXPECT_EQ(task_raw, sequence->PeekTask());
+  sequence->PopTask();
+  EXPECT_EQ(nullptr, sequence->PeekTask());
+}
+
+}  // namespace
+}  // namespace internal
+}  // namespace base