Task Scheduler.

base/task_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/thread_pool.h"
+
+#include "base/bind.h"
+#include "base/callback_forward.h"
+#include "base/synchronization/condition_variable.h"
+#include "base/task_scheduler/scheduler_lock.h"
+#include "base/task_scheduler/shutdown_manager.h"
+#include "base/task_scheduler/worker_thread.h"
+#include "base/threading/platform_thread.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace base {
+namespace task_scheduler {
+
+namespace {
+class SingleThreadChecker {
+ public:
+  SingleThreadChecker() : ran_all_tasks_on_same_thread_(true) {}
+
+  void RunTask() {
+    AutoSchedulerLock auto_lock(lock_);
+
+    if (!last_thread_handle_.is_null() &&
+        !PlatformThread::CurrentHandle().is_equal(last_thread_handle_)) {
+      ran_all_tasks_on_same_thread_ = false;
+    }
+    last_thread_handle_ = PlatformThread::CurrentHandle();
+  }
+
+  bool ran_all_tasks_on_same_thread() const {
+    return ran_all_tasks_on_same_thread_;
+  }
+
+ private:
+  PlatformThreadHandle last_thread_handle_;
+  bool ran_all_tasks_on_same_thread_;
+  SchedulerLock lock_;
+};
+}  // namespace
+
+class TaskSchedulerThreadPoolTest : public testing::Test {
+ protected:
+  TaskSchedulerThreadPoolTest()
+      : thread_pool_(ThreadPool::CreateThreadPool(
+            ThreadPriority::NORMAL,
+            4,
+            Bind(&TaskSchedulerThreadPoolTest::ReinsertSequenceCallback,
+                 Unretained(this)),
+            &shutdown_manager_)),
+        cv_(lock_.RawLockForConditionVariable()),
+        last_posted_task_index_(0),
+        last_run_task_index_(0),
+        ran_task_that_should_not_run_(false) {}
+
+  Closure GetTaskThatShouldRunClosure(
+      SingleThreadChecker* single_thread_checker) {
+    ++last_posted_task_index_;
+    return Bind(&TaskSchedulerThreadPoolTest::RunTaskThatShouldRun,
+                Unretained(this), last_posted_task_index_,
+                single_thread_checker);
+  }
+
+  Closure GetTaskThatShouldNotRunClosure() {
+    return Bind(&TaskSchedulerThreadPoolTest::RunTaskThatShouldNotRun,
+                Unretained(this));
+  }
+
+  void WaitUntilLastPostedTaskHasRun() {
+    AutoSchedulerLock auto_lock(lock_);
+    while (last_posted_task_index_ != last_run_task_index_)
+      cv_.Wait();
+  }
+
+  void Shutdown() {
+    shutdown_manager_.Shutdown();
+    thread_pool_->JoinAllThreadsForTesting();
+  }
+
+  bool ran_task_that_should_not_run() const {
+    return ran_task_that_should_not_run_;
+  }
+
+  scoped_ptr<ThreadPool> thread_pool_;
+  ShutdownManager shutdown_manager_;
+
+ private:
+  void ReinsertSequenceCallback(scoped_refptr<Sequence> sequence,
+                                const WorkerThread* worker_thread) {
+    thread_pool_->ReinsertSequence(sequence, sequence->GetSortKey(),
+                                   worker_thread);
+  }
+
+  void RunTaskThatShouldRun(size_t index,
+                            SingleThreadChecker* single_thread_checker) {
+    AutoSchedulerLock auto_lock(lock_);
+
+    if (single_thread_checker)
+      single_thread_checker->RunTask();
+
+    // Wait until the task with index (|index| - 1) has run. If tasks are
+    // executed in the wrong order, this can block forever and make the test
+    // fail.
+    //
+    // Note: It isn't correct to assume that this condition will always be
+    // immediatly true if tasks are popped from the priority queue in the right
+    // order. Indeed, a thread A could start running task #2 before a thread B
+    // starts running task #1 despite the fact that thread B has popped task #1
+    // before thread A has popped task #2.
+    while (last_run_task_index_ != index - 1)
+      cv_.Wait();
+
+    ++last_run_task_index_;
+    cv_.Broadcast();
+  }
+
+  void RunTaskThatShouldNotRun() { ran_task_that_should_not_run_ = true; }
+
+  // Lock protecting |cv_|.
+  SchedulerLock lock_;
+
+  // Condition variable signaled each time a task completes its execution.
+  ConditionVariable cv_;
+
+  // Index of the last posted task.
+  size_t last_posted_task_index_;
+
+  // Index of the last run task.
+  size_t last_run_task_index_;
+
+  // True if a task that shouldn't run has run.
+  bool ran_task_that_should_not_run_;
+};
+
+TEST_F(TaskSchedulerThreadPoolTest, PostSingleParallelTask) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                           ExecutionMode::PARALLEL)
+      ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostSingleSequencedTask) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                           ExecutionMode::SEQUENCED)
+      ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostSingleSingleThreadedTask) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                           ExecutionMode::SINGLE_THREADED)
+      ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostMultipleTasksNoWaitBetweenPosts) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  for (size_t i = 0; i < 100; ++i) {
+    thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                             ExecutionMode::PARALLEL)
+        ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+  }
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostMultipleTasksWaitBetweenPosts) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  for (size_t i = 0; i < 100; ++i) {
+    thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                             ExecutionMode::PARALLEL)
+        ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+    WaitUntilLastPostedTaskHasRun();
+  }
+
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostMultipleTasksInSameSequence) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::SEQUENCED);
+
+  for (size_t i = 0; i < 100; ++i)
+    task_runner->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostMultipleTasksInTwoSequences) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  auto task_runner_a = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::SEQUENCED);
+  auto task_runner_b = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::SEQUENCED);
+
+  for (size_t i = 0; i < 100; ++i) {
+    task_runner_a->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+    task_runner_b->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+  }
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostMultipleSingleThreadedTasks) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::SINGLE_THREADED);
+  SingleThreadChecker single_thread_checker;
+
+  for (size_t i = 0; i < 100; ++i) {
+    task_runner->PostTask(FROM_HERE,
+                          GetTaskThatShouldRunClosure(&single_thread_checker));
+  }
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+  EXPECT_TRUE(single_thread_checker.ran_all_tasks_on_same_thread());
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelDelayedTasks) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::PARALLEL);
+
+  for (size_t i = 0; i < 10; ++i) {
+    task_runner->PostDelayedTask(FROM_HERE,
+                                 GetTaskThatShouldRunClosure(nullptr),
+                                 TimeDelta::FromMilliseconds(i * 50));
+  }
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostSequencedDelayedTasks) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::SEQUENCED);
+
+  for (size_t i = 0; i < 10; ++i) {
+    task_runner->PostDelayedTask(FROM_HERE,
+                                 GetTaskThatShouldRunClosure(nullptr),
+                                 TimeDelta::FromMilliseconds(i * 50));
+  }
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, ShutdownBehavior) {
+  ASSERT_TRUE(thread_pool_.get());
+
+  shutdown_manager_.SetIsShuttingDownForTesting();
+
+  // Post tasks with different shutdown behaviors.
+  thread_pool_->CreateTaskRunnerWithTraits(
+                  TaskTraits().WithShutdownBehavior(
+                      TaskShutdownBehavior::CONTINUE_ON_SHUTDOWN),
+                  ExecutionMode::PARALLEL)
+      ->PostTask(FROM_HERE, GetTaskThatShouldNotRunClosure());
+  thread_pool_->CreateTaskRunnerWithTraits(
+                  TaskTraits().WithShutdownBehavior(
+                      TaskShutdownBehavior::SKIP_ON_SHUTDOWN),
+                  ExecutionMode::PARALLEL)
+      ->PostTask(FROM_HERE, GetTaskThatShouldNotRunClosure());
+  thread_pool_->CreateTaskRunnerWithTraits(
+                  TaskTraits().WithShutdownBehavior(
+                      TaskShutdownBehavior::BLOCK_SHUTDOWN),
+                  ExecutionMode::PARALLEL)
+      ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure(nullptr));
+
+  WaitUntilLastPostedTaskHasRun();
+  Shutdown();
+  EXPECT_FALSE(ran_task_that_should_not_run());
+}
+
+}  // namespace task_scheduler
+}  // namespace base