TaskScheduler [6] SchedulerWorkerThread

base/task_scheduler/worker_thread_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/worker_thread.h"
+
+#include <utility>
+#include <vector>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/callback_forward.h"
+#include "base/logging.h"
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/synchronization/condition_variable.h"
+#include "base/task_scheduler/priority_queue.h"
+#include "base/task_scheduler/scheduler_lock.h"
+#include "base/task_scheduler/task_tracker.h"
+#include "base/task_scheduler/utils.h"
+#include "base/threading/simple_thread.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace base {
+namespace internal {
+
+namespace {
+
+const size_t kNumTasksPerTest = 500;

[robliao, 2016/03/24 23:03:27]

Why 500?

[fdoray, 2016/03/29 18:33:35]

A high number of tasks increases the chances to catch bugs (I discovered+fixed
bugs that occurred only after 100 iterations when I wrote these tests). No
particular reason for choosing 500.

+
+class TaskClosureFactory {

[robliao, 2016/03/24 23:03:26]

Optional: Maybe SequencedTaskClosureFactory?

[fdoray, 2016/03/29 18:33:34]

Now I use the factory to create non-sequenced tasks...

+ public:
+  TaskClosureFactory() : run_task_cv_(lock_.CreateConditionVariable()) {}
+
+  ~TaskClosureFactory() {
+    AutoSchedulerLock auto_lock(lock_);
+    EXPECT_EQ(num_factored_tasks_, num_run_tasks_);
+  }
+
+  Closure CreateTaskClosure() {
+    AutoSchedulerLock auto_lock(lock_);
+    ++num_factored_tasks_;
+    return Bind(&TaskClosureFactory::RunTask, Unretained(this),
+                num_factored_tasks_);
+  }
+
+  void WaitUntilLastFactoredTaskRan() {

[robliao, 2016/03/24 23:03:26]

WaitForAllTasksToRun

[fdoray, 2016/03/29 18:33:35]

Done.

+    AutoSchedulerLock auto_lock(lock_);
+    while (num_factored_tasks_ != num_run_tasks_)

[robliao, 2016/03/24 23:03:26]

Nit:  num_run_tasks_ < num_created_tasks_
Makes it easier to see the expected relationship between the two values.

[fdoray, 2016/03/29 18:33:35]

Done.

+      run_task_cv_->Wait();
+  }
+
+ private:
+  void RunTask(size_t task_index) {
+    AutoSchedulerLock auto_lock(lock_);
+
+    if (task_index != num_run_tasks_ + 1)

[robliao, 2016/03/24 23:03:27]

If you allow task_index to be zero, you can avoid the + 1 here.

Make the bind use num_created_tasks_++

[fdoray, 2016/03/29 18:33:34]

Done.

+      ADD_FAILURE() << "Unexpected task execution order.";
+
+    ++num_run_tasks_;
+    run_task_cv_->Signal();
+  }
+
+  // Synchronizes access to all members.
+  SchedulerLock lock_;
+
+  // Signaled when a task runs.
+  scoped_ptr<ConditionVariable> run_task_cv_;
+
+  // Number of times that CreateTaskClosure() has been called.
+  size_t num_factored_tasks_ = 0;

[robliao, 2016/03/24 23:03:26]

num_created_tasks_

[fdoray, 2016/03/29 18:33:34]

Done.

+
+  // Number of times that RunTask() has been called.
+  size_t num_run_tasks_ = 0;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskClosureFactory);
+};
+
+class ThreadPostingTasks : public SimpleThread {
+ public:
+  ThreadPostingTasks(WorkerThread* worker_thread)
+      : SimpleThread("ThreadPostingTasks"), worker_thread_(worker_thread) {}
+
+  void WaitUntilLastFactoredTaskRan() {
+    factory_.WaitUntilLastFactoredTaskRan();
+  }
+
+ private:
+  void Run() override {
+    auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+    for (size_t i = 0; i < kNumTasksPerTest; ++i)
+      task_runner->PostTask(FROM_HERE, factory_.CreateTaskClosure());
+  }
+
+  WorkerThread* const worker_thread_;
+  TaskClosureFactory factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(ThreadPostingTasks);
+};
+
+}  // namespace
+
+class TaskSchedulerWorkerThreadTest : public testing::Test {
+ protected:
+  TaskSchedulerWorkerThreadTest()
+      : shared_priority_queue_(Bind(&DoNothing)),
+        state_changed_callback_cv_(lock_.CreateConditionVariable()) {}
+
+  virtual void SetUp() override {

[robliao, 2016/03/24 23:03:26]

Nit: virtual not needed with override.
Here and below.

[fdoray, 2016/03/29 18:33:35]

Done.

+    worker_thread_ = WorkerThread::CreateWorkerThread(
+        ThreadPriority::NORMAL, &shared_priority_queue_,
+        Bind(&TaskSchedulerWorkerThreadTest::
+                 PoppedTaskFromSharedSequenceCallback,
+             Unretained(this)),
+        Bind(&TaskSchedulerWorkerThreadTest::StateChangedCallback,
+             Unretained(this)),
+        &task_tracker_);
+    ASSERT_TRUE(worker_thread_);
+    WaitUntilIdle();
+
+    AutoSchedulerLock auto_lock(lock_);
+    num_state_changes_ = 0;
+  }
+
+  virtual void TearDown() override { worker_thread_->JoinForTesting(); }
+
+  void WaitUntilIdle() {
+    AutoSchedulerLock auto_lock(lock_);
+    while (last_state_ != WorkerThread::State::IDLE)
+      state_changed_callback_cv_->Wait();
+  }
+
+  void WaitUntilNumStateChanges(size_t expected_num_state_changes) {
+    AutoSchedulerLock auto_lock(lock_);
+    while (num_state_changes_ != expected_num_state_changes)

[robliao, 2016/03/24 23:03:27]

Nit: num_state_changes_ < expected_num_state_changes

[fdoray, 2016/03/29 18:33:35]

Done.

+      state_changed_callback_cv_->Wait();
+  }
+
+  // Returns the number of state changes that occurred after SetUp() completed
+  // its execution.
+  size_t num_state_changes() const {
+    AutoSchedulerLock auto_lock(lock_);
+    return num_state_changes_;
+  }
+
+  size_t num_popped_task_from_shared_sequence() const {
+    AutoSchedulerLock auto_lock(lock_);
+    return num_popped_task_from_shared_sequence_;
+  }
+
+  PriorityQueue shared_priority_queue_;
+  TaskTracker task_tracker_;
+  scoped_ptr<WorkerThread> worker_thread_;
+
+ private:
+  void PoppedTaskFromSharedSequenceCallback(const WorkerThread* worker_thread,
+                                            scoped_refptr<Sequence> sequence) {
+    {
+      AutoSchedulerLock auto_lock(lock_);
+      ++num_popped_task_from_shared_sequence_;
+    }
+
+    // Reinsert sequence in |shared_priority_queue_|.
+    const SequenceSortKey sort_key = sequence->GetSortKey();
+    shared_priority_queue_.BeginTransaction()->Push(make_scoped_ptr(
+        new PriorityQueue::SequenceAndSortKey(std::move(sequence), sort_key)));
+  }
+
+  void StateChangedCallback(WorkerThread* worker_thread,
+                            WorkerThread::State state) {
+    AutoSchedulerLock auto_lock(lock_);
+    EXPECT_EQ(worker_thread_.get(), worker_thread);
+    EXPECT_NE(last_state_, state);
+    last_state_ = state;
+    ++num_state_changes_;
+    state_changed_callback_cv_->Signal();
+  }
+
+  // Synchronizes access to all members below.
+  mutable SchedulerLock lock_;
+
+  // Condition variable signaled when StateChangedCallback() is invoked.
+  scoped_ptr<ConditionVariable> state_changed_callback_cv_;
+
+  // Last state reported to StateChangedCallback().
+  WorkerThread::State last_state_ = WorkerThread::State::BUSY;
+
+  // Number of times that StateChangedCallback() has been invoked.
+  size_t num_state_changes_ = 0;
+
+  // Number of times that PoppedTaskFromSharedSequenceCallback() has been
+  // invoked.
+  size_t num_popped_task_from_shared_sequence_ = 0;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerThreadTest);
+};
+
+// Verify each call to WakeUp() on an idle WorkerThread causes 2 state changes.

[robliao, 2016/03/24 23:03:26]

This seems like a brittle test, especially if for some reason we need more
tests. It may be more helpful to verify that the states go through an expected
state ordering instead (e.g. from IDLE to BUSY and then back to IDLE)

[fdoray, 2016/03/29 18:33:34]

StateChangedCallback() verifies that the states go through an expected state
ordering. I added a comment to highlight that fact.

+TEST_F(TaskSchedulerWorkerThreadTest, WakeUp) {
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    worker_thread_->WakeUp();
+    WaitUntilNumStateChanges(2 * (i + 1));

[robliao, 2016/03/24 23:03:26]

Should this instead WaitUntilIdle? If so, you can remove
WaitUntilNumStateChanges

[fdoray, 2016/03/29 18:33:34]

Replacing WaitUntilNumStateChanges() with WaitUntilIdle() isn't correct because
it doesn't verify that the thread has gone through the BUSY state since it was
woken up (the test could pass if the state is always IDLE).

+    EXPECT_EQ(2 * (i + 1), num_state_changes());
+  }
+
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that a task runs successfully when it is posted through a
+// task runner returned by WorkerThread::CreateTaskRunnerWithTraits().
+TEST_F(TaskSchedulerWorkerThreadTest, PostOneSingleThreadedTask) {
+  TaskClosureFactory factory;
+  worker_thread_->CreateTaskRunnerWithTraits(TaskTraits())
+      ->PostTask(FROM_HERE, factory.CreateTaskClosure());
+
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_EQ(2, num_state_changes());
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 500 tasks run successfully when they are posted through a task

[robliao, 2016/03/24 23:03:27]

Maybe |kNumTasksPerTest| instead of 500.

[fdoray, 2016/03/29 18:33:34]

Done.

+// runner returned by WorkerThread::CreateTaskRunnerWithTraits(). Don't wait
+// between posts.
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleSingleThreadedTasksNoWaitBetweenPosts) {
+  TaskClosureFactory factory;
+  auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i)
+    task_runner->PostTask(FROM_HERE, factory.CreateTaskClosure());
+
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_GE(num_state_changes(), 2);
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 500 tasks run successfully when they are posted through a task
+// runner returned by WorkerThread::CreateTaskRunnerWithTraits(). Wait until the
+// previous task has completed its execution before posting a new task.
+TEST_F(TaskSchedulerWorkerThreadTest,

[robliao, 2016/03/24 23:03:27]

Isn't this similar to PostOneSingleThreadedTask, just called kNumTasksPerTest?

[fdoray, 2016/03/29 18:33:34]

Yes, PostOneSingleThreadedTask is equivalent to this test with 1 loop iteration.
I had both tests because PostOneSingleThreadedTask is easier to debug and
PostMultipleSingleThreadedTasksWaitBetweenPosts is more likely to catch
synchronization errors. In the latest patch set, I removed
PostOneSingleThreadedTask to reduce the amount of code to review.

+       PostMultipleSingleThreadedTasksWaitBetweenPosts) {
+  TaskClosureFactory factory;
+  auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    task_runner->PostTask(FROM_HERE, factory.CreateTaskClosure());
+    factory.WaitUntilLastFactoredTaskRan();
+    WaitUntilIdle();
+    EXPECT_EQ(2 * (i + 1), num_state_changes());
+  }
+
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 1000 tasks run successfully when they are posted through 2 task
+// runners returned by WorkerThread::CreateTaskRunnerWithTraits(). Don't wait
+// between posts.
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleTasksTwoSingleThreadedTaskRunners) {
+  TaskClosureFactory factory;
+  auto task_runner_a = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+  auto task_runner_b = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    task_runner_a->PostTask(FROM_HERE, factory.CreateTaskClosure());
+    task_runner_b->PostTask(FROM_HERE, factory.CreateTaskClosure());

[robliao, 2016/03/24 23:03:26]

Do we guarantee task execution ordering across task runners?

[fdoray, 2016/03/29 18:33:35]

The current implementation runs tasks posted to the same WorkerThread / same
priority in posting order. However, we might want to change that in the future
(e.g. if we realize that there is less contention on the pq lock if we run tasks
from the same sequence in batch). For that reason, I now have a different
factory for each task runner.

+  }
+
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_GE(num_state_changes(), 2);
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that a task runs successfully when it is added to the shared priority
+// queue of a WorkerThread. The test wakes up the WorkerThread after adding the
+// task to the priority queue.

[robliao, 2016/03/24 23:03:27]

Mention that the wakeup is necessary since shared tasks do not automatically
wake up the thread.

[fdoray, 2016/03/29 18:33:35]

Done.

+TEST_F(TaskSchedulerWorkerThreadTest, PostOneSharedTask) {
+  TaskClosureFactory factory;
+  PostTaskHelper(make_scoped_ptr(new Task(
+                     FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                 make_scoped_refptr(new Sequence), &shared_priority_queue_,
+                 &task_tracker_);
+
+  worker_thread_->WakeUp();
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_EQ(2, num_state_changes());
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 500 tasks run successfully when they are added to the shared
+// priority queue of a WorkerThread in different Sequences. The test wakes up
+// the WorkerThread each time it adds a task to the priority queue.

[robliao, 2016/03/24 23:03:27]

The worker thread looks like it's only woken up once.

[fdoray, 2016/03/29 18:33:34]

Done.

+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleSharedTasksNoWaitBetweenPosts) {
+  TaskClosureFactory factory;
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    PostTaskHelper(make_scoped_ptr(new Task(
+                       FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                   make_scoped_refptr(new Sequence), &shared_priority_queue_,
+                   &task_tracker_);
+  }
+
+  worker_thread_->WakeUp();
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_EQ(2, num_state_changes());
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 500 tasks run successfully when they are added to the shared
+// priority queue of a WorkerThread in different Sequences. The test wakes up
+// the WorkerThread and waits until the task completes its execution each time
+// it adds a task to the priority queue.
+TEST_F(TaskSchedulerWorkerThreadTest, PostMultipleSharedTasksWaitBetweenPosts) {
+  TaskClosureFactory factory;
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    PostTaskHelper(make_scoped_ptr(new Task(
+                       FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                   make_scoped_refptr(new Sequence), &shared_priority_queue_,
+                   &task_tracker_);
+    worker_thread_->WakeUp();
+
+    factory.WaitUntilLastFactoredTaskRan();
+    WaitUntilIdle();
+    EXPECT_EQ(2 * (i + 1), num_state_changes());
+  }
+
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 500 tasks run successfully when they are added to the shared
+// priority queue of a WorkerThread (all tasks in the same Sequence). The test
+// wakes up the WorkerThread once it has added the 500 tasks to the priority
+// queue.
+TEST_F(TaskSchedulerWorkerThreadTest, PostMultipleSharedTasksInSameSequence) {
+  TaskClosureFactory factory;
+  scoped_refptr<Sequence> sequence(new Sequence);
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    PostTaskHelper(make_scoped_ptr(new Task(
+                       FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                   sequence, &shared_priority_queue_, &task_tracker_);
+  }
+
+  worker_thread_->WakeUp();
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_EQ(2, num_state_changes());
+  EXPECT_EQ(kNumTasksPerTest - 1, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 1000 tasks run successfully when they are added to the shared
+// priority queue of a WorkerThread. Tasks are split into 2 sequences. The test
+// wakes up the WorkerThread once it has added the 1000 tasks to the priority
+// queue.
+TEST_F(TaskSchedulerWorkerThreadTest, PostMultipleSharedTasksInTwoSequences) {
+  TaskClosureFactory factory;
+  scoped_refptr<Sequence> sequence_a(new Sequence);
+  scoped_refptr<Sequence> sequence_b(new Sequence);
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    PostTaskHelper(make_scoped_ptr(new Task(
+                       FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                   sequence_a, &shared_priority_queue_, &task_tracker_);

[robliao, 2016/03/24 23:03:27]

We definitely don't guarantee interleaving across sequences. This should
probably relax the run check.

[fdoray, 2016/03/29 18:33:34]

Done.

+    PostTaskHelper(make_scoped_ptr(new Task(
+                       FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                   sequence_b, &shared_priority_queue_, &task_tracker_);
+  }
+
+  worker_thread_->WakeUp();
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_EQ(2, num_state_changes());
+  EXPECT_EQ(2 * (kNumTasksPerTest - 1), num_popped_task_from_shared_sequence());
+}
+
+// Verify that a shared task and a single-threaded task run successfully when
+// they are posted to a WorkerThread. The WorkerThread is not woken up by the
+// test after the shared task is posted. The wake-up is done by the task runner
+// through which the single-thread task is posted.
+TEST_F(TaskSchedulerWorkerThreadTest, PostSharedAndSingleThreadedTasks) {
+  TaskClosureFactory factory;
+
+  // Post a task in the shared priority queue. Don't wake up the WorkerThread.
+  PostTaskHelper(make_scoped_ptr(new Task(
+                     FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                 make_scoped_refptr(new Sequence), &shared_priority_queue_,
+                 &task_tracker_);
+
+  // Post a task in the single-threaded priority queue. The TaskRunner will wake
+  // up the WorkerThread.
+  worker_thread_->CreateTaskRunnerWithTraits(TaskTraits())
+      ->PostTask(FROM_HERE, factory.CreateTaskClosure());
+
+  factory.WaitUntilLastFactoredTaskRan();
+  WaitUntilIdle();
+  EXPECT_EQ(2, num_state_changes());
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 500 shared tasks and 500 single-threaded tasks run successfully
+// when they are posted to a WorkerThread. The WorkerThread is not woken up by
+// the test after a shared task is posted. The wake-ups are done by the task
+// runner through which the single-thread tasks are posted.
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleSharedAndSingleThreadedTasks) {
+  TaskClosureFactory factory;
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    // Post a task in the shared priority queue. Don't wake up the WorkerThread.
+    PostTaskHelper(make_scoped_ptr(new Task(
+                       FROM_HERE, factory.CreateTaskClosure(), TaskTraits())),
+                   make_scoped_refptr(new Sequence), &shared_priority_queue_,
+                   &task_tracker_);
+
+    // Post a task in the single-threaded priority queue. The TaskRunner will
+    // wake up the WorkerThread.
+    worker_thread_->CreateTaskRunnerWithTraits(TaskTraits())
+        ->PostTask(FROM_HERE, factory.CreateTaskClosure());
+
+    factory.WaitUntilLastFactoredTaskRan();
+    WaitUntilIdle();
+    EXPECT_EQ(2 * (i + 1), num_state_changes());
+  }
+
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+// Verify that 10000 single-threaded tasks posted to a single WorkerThread from
+// 10 threads run successfully.
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostSingleThreadedTasksFromMultipleThreads) {
+  const size_t kNumThreads = 20;
+  std::vector<scoped_ptr<ThreadPostingTasks>> threads;
+  for (size_t i = 0; i < kNumThreads; ++i) {
+    threads.push_back(
+        make_scoped_ptr(new ThreadPostingTasks(worker_thread_.get())));
+    threads.back()->Start();
+  }
+
+  for (const auto& thread : threads) {
+    thread->Join();
+    thread->WaitUntilLastFactoredTaskRan();
+  }
+
+  WaitUntilIdle();
+  EXPECT_GE(num_state_changes(), 2);
+  EXPECT_EQ(0, num_popped_task_from_shared_sequence());
+}
+
+}  // namespace internal
+}  // namespace base