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/lock.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/task_runner.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 kNumThreadsInThreadPool = 4;
+const size_t kNumThreadsPostingTasks = 4;
+const size_t kNumTasksPostedPerThread = 150;
+
+class TaskSchedulerThreadPoolTest : public testing::Test {
+ protected:
+  TaskSchedulerThreadPoolTest() = default;
+
+  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_->WaitForAllWorkerThreadsIdleForTesting();
+    thread_pool_->JoinForTesting();
+  }
+
+  std::unique_ptr<SchedulerThreadPool> thread_pool_;
+
+ private:
+  void RanTaskFromSequenceCallback(scoped_refptr<Sequence> sequence) {
+    // Reinsert |sequence| in |thread_pool_|'s shared PriorityQueue if it isn't
+    // empty after popping one of its Tasks. In production code, this callback
+    // would be implemented by the TaskScheduler which would first determine in
+    // which PriorityQueue the sequence must be reinserted.
+    const bool sequence_became_empty = sequence->PopTask();
+    if (!sequence_became_empty) {

[gab, 2016/04/07 20:32:48]

As discussed offline, make the callback EnqueueSequence() instead and have the
WorkerThread do the pop, only calling this callback if it actually needs to
enqueue, centralizing the pop logic in WorkerThread.

[fdoray, 2016/04/08 14:53:03]

Done.

+      const SequenceSortKey sort_key(sequence->GetSortKey());
+      thread_pool_->InsertSequenceAfterTaskRan(std::move(sequence), sort_key);
+    }
+  }
+
+  TaskTracker task_tracker_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerThreadPoolTest);
+};
+
+class TaskFactory {
+ public:
+  TaskFactory() : cv_(&lock_) {}
+
+  // Posts a task through |task_runner|. 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 it is signaled.
+  void PostTask(scoped_refptr<TaskRunner> task_runner,
+                bool post_nested_task,
+                WaitableEvent* event) {
+    AutoLock auto_lock(lock_);
+    EXPECT_TRUE(task_runner->PostTask(
+        FROM_HERE, Bind(&TaskFactory::RunTaskCallback, Unretained(this),
+                        num_created_tasks_++, task_runner, post_nested_task,
+                        Unretained(event))));
+  }
+
+  // Waits for all tasks posted by PostTask() to start running. It is not
+  // guaranteed that the tasks have completed their execution when this returns.
+  void WaitForAllTasksToRun() const {
+    AutoLock auto_lock(lock_);
+    while (run_tasks_.size() < num_created_tasks_)
+      cv_.Wait();
+  }
+
+  size_t NumRunTasks() const {
+    AutoLock auto_lock(lock_);
+    return run_tasks_.size();
+  }
+
+ private:
+  void RunTaskCallback(size_t task_index,
+                       scoped_refptr<TaskRunner> task_runner,
+                       bool post_nested_task,
+                       WaitableEvent* event) {
+    if (post_nested_task)
+      PostTask(task_runner, false, nullptr);

[gab, 2016/04/07 20:32:48]

I was curious what this top-level PostTask() call was until I realized it was
the factory's, how about PostTestTask() to make it clear that callers are
calling something in this test namespace and not one of our PostTask APIs.

[fdoray, 2016/04/08 14:53:03]

Done.

+
+    EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread());
+
+    {
+      AutoLock 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);

[gab, 2016/04/07 20:32:48]

s/run_tasks_/ran_tasks_/

[fdoray, 2016/04/08 14:53:03]

Done.

+
+      cv_.Signal();
+    }
+
+    if (event)
+      event->Wait();
+  }
+
+  // Synchronizes access to all members below.
+  mutable Lock lock_;
+
+  // Condition variable signaled when a task runs.
+  mutable ConditionVariable cv_;
+
+  // Number of tasks posted by PostTask().
+  size_t num_created_tasks_ = 0;
+
+  // Indexes of tasks that ran.
+  std::unordered_set<size_t> run_tasks_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskFactory);
+};
+
+class ThreadPostingTasks : public SimpleThread {
+ public:
+  // Constructs a thread that posts tasks to |thread_pool| through an
+  // |execution_mode| task runner. If |wait_for_all_threads_idle| is true, the
+  // thread wait until all worker threads in |thread_pool| are idle before
+  // posting a new task. If |post_nested_task| is true, each task posted by this
+  // thread posts another task when it runs.
+  ThreadPostingTasks(SchedulerThreadPool* thread_pool,
+                     ExecutionMode execution_mode,
+                     bool wait_for_all_threads_idle,
+                     bool post_nested_task)
+      : SimpleThread("ThreadPostingTasks"),
+        thread_pool_(thread_pool),
+        task_runner_(thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(),
+                                                              execution_mode)),
+        wait_for_all_threads_idle_(wait_for_all_threads_idle),
+        post_nested_task_(post_nested_task) {}

[gab, 2016/04/07 20:32:48]

DCHECK(thread_pool_)

[fdoray, 2016/04/08 14:53:03]

Done.

+
+  const TaskFactory* factory() const { return &factory_; }
+
+ private:
+  void Run() override {
+    EXPECT_FALSE(task_runner_->RunsTasksOnCurrentThread());
+
+    for (size_t i = 0; i < kNumTasksPostedPerThread; ++i) {
+      if (wait_for_all_threads_idle_)
+        thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+      factory_.PostTask(task_runner_, post_nested_task_, nullptr);
+    }
+  }
+
+  SchedulerThreadPool* const thread_pool_;
+  scoped_refptr<TaskRunner> task_runner_;

[gab, 2016/04/07 20:32:48]

"const scoped_refptr" as well?

[fdoray, 2016/04/08 14:53:03]

Done.

+  const bool wait_for_all_threads_idle_;
+  const bool post_nested_task_;
+  TaskFactory factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(ThreadPostingTasks);
+};
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasks) {
+  // Create threads to post tasks to PARALLEL TaskRunners.
+  std::vector<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
+  for (size_t j = 0; j < kNumThreadsPostingTasks; ++j) {
+    threads_posting_tasks.push_back(WrapUnique(new ThreadPostingTasks(
+        thread_pool_.get(), ExecutionMode::PARALLEL, false, false)));
+    threads_posting_tasks.back()->Start();
+  }
+
+  // Wait for all tasks to run.
+  for (const auto& thread_posting_tasks : threads_posting_tasks) {
+    thread_posting_tasks->Join();
+    thread_posting_tasks->factory()->WaitForAllTasksToRun();
+    EXPECT_EQ(kNumTasksPostedPerThread,
+              thread_posting_tasks->factory()->NumRunTasks());
+  }
+
+  // Wait until all worker threads are idle to be sure that no task accesses
+  // its TaskFactory after |thread_posting_tasks| is destroyed.
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWaitAllThreadsIdle) {
+  // Create threads to post tasks to PARALLEL TaskRunners. To verify that
+  // worker threads can sleep and be woken up when new tasks are posted, wait
+  // for all threads to become idle before posting a new task.
+  std::vector<std::unique_ptr<ThreadPostingTasks>> threads_posting_tasks;
+  for (size_t j = 0; j < kNumThreadsPostingTasks; ++j) {
+    threads_posting_tasks.push_back(WrapUnique(new ThreadPostingTasks(
+        thread_pool_.get(), ExecutionMode::PARALLEL, true, false)));
+    threads_posting_tasks.back()->Start();
+  }
+
+  // Wait for all tasks to run.
+  for (const auto& thread_posting_tasks : threads_posting_tasks) {
+    thread_posting_tasks->Join();
+    thread_posting_tasks->factory()->WaitForAllTasksToRun();
+    EXPECT_EQ(kNumTasksPostedPerThread,
+              thread_posting_tasks->factory()->NumRunTasks());
+  }
+
+  // Wait until all worker threads are idle to be sure that no task accesses
+  // its TaskFactory after |thread_posting_tasks| is destroyed.
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, NestedPostParallelTasks) {
+  // Create threads to post tasks to 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) {
+    threads_posting_tasks.push_back(WrapUnique(new ThreadPostingTasks(
+        thread_pool_.get(), ExecutionMode::PARALLEL, false, true)));
+    threads_posting_tasks.back()->Start();
+  }
+
+  // Wait for all tasks to run.
+  for (const auto& thread_posting_tasks : threads_posting_tasks) {
+    thread_posting_tasks->Join();
+    thread_posting_tasks->factory()->WaitForAllTasksToRun();
+    EXPECT_EQ(2 * kNumTasksPostedPerThread,
+              thread_posting_tasks->factory()->NumRunTasks());
+  }
+
+  // Wait until all worker threads are idle to be sure that no task accesses
+  // its TaskFactory after |thread_posting_tasks| is destroyed.
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWithOneAvailableThread) {
+  TaskFactory factory;
+
+  // Post tasks to keep all threads busy except one until |event| is signaled.
+  WaitableEvent event(true, false);
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::PARALLEL);
+  for (size_t i = 0; i < (kNumThreadsInThreadPool - 1); ++i)
+    factory.PostTask(task_runner, false, &event);
+  factory.WaitForAllTasksToRun();
+
+  // Post |kNumTasksPostedPerThread| tasks that should all run despite the fact
+  // that only one thread in |thread_pool_| isn't busy.
+  for (size_t i = 0; i < kNumTasksPostedPerThread; ++i)
+    factory.PostTask(task_runner, false, nullptr);
+  factory.WaitForAllTasksToRun();
+
+  // Release tasks waiting on |event|.
+  event.Signal();
+
+  // Wait until all worker threads are idle to be sure that no task accesses
+  // |factory| after it is destroyed.
+  thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
+}
+
+TEST_F(TaskSchedulerThreadPoolTest, Saturate) {
+  TaskFactory factory;
+
+  // Verify that it is possible to have |kNumThreadsInThreadPool| tasks running
+  // simultaneously.
+  WaitableEvent event(true, false);
+  auto task_runner = thread_pool_->CreateTaskRunnerWithTraits(
+      TaskTraits(), ExecutionMode::PARALLEL);
+  for (size_t i = 0; i < kNumThreadsInThreadPool; ++i)
+    factory.PostTask(task_runner, false, &event);
+  factory.WaitForAllTasksToRun();
+
+  // Release tasks waiting on |event|.
+  event.Signal();
+
+  // Wait until all worker threads are idle to be sure that no task accesses
+  // |factory| after it is destroyed.
+  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) {

[gab, 2016/04/07 20:32:48]

Don't use the fixture name (TaskSchedulerPostTaskHelperTest) for tests that
aren't TEST_F

[fdoray, 2016/04/08 14:53:03]

Moved these tests to utils_unittest.cc. Not sure what your comment is supposed
to mean.

[gab, 2016/04/08 17:56:00]

Oops, actually no I had somehow read "TaskSchedulerPostTaskHelperTest" as being
"TaskSchedulerThreadPoolTest"... "TaskSchedulerPostTaskHelperTest" is another
name and was correct, please do keep them here.

[fdoray, 2016/04/08 19:00:05]

I moved PostTaskHelper + tests to utils.cc/.h/_unitttest.cc because I plan to
use it from single-threaded TaskRunners defined in scheduler_worker_thread.cc,
not just because of your comment.

+  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|.
+  EXPECT_TRUE(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|.
+  EXPECT_TRUE(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