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 "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/callback_forward.h"
+#include "base/logging.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/synchronization/condition_variable.h"
+#include "base/synchronization/waitable_event.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 "testing/gtest/include/gtest/gtest.h"
+
+namespace base {
+namespace internal {
+
+class TaskSchedulerWorkerThreadTest : public testing::Test {
+ protected:
+  TaskSchedulerWorkerThreadTest()
+      : shared_priority_queue_(Bind(&DoNothing)),
+        worker_thread_(WorkerThread::CreateWorkerThread(

[robliao, 2016/03/09 15:41:12]

The worker thread may start and call the idle callback before the idle callback
CV has been initialized due to the order.

Fixed in
https://luckyluke-private.googlesource.com/src/+/0041eafd8efbdac3aa5ebbdf1e01...

[fdoray, 2016/03/14 20:46:35]

Thanks!

+            ThreadPriority::NORMAL,
+            &shared_priority_queue_,
+            Bind(&TaskSchedulerWorkerThreadTest::ReinsertSequenceCallback,
+                 Unretained(this)),
+            Bind(&TaskSchedulerWorkerThreadTest::BecomesIdleCallback,
+                 Unretained(this)),
+            &task_tracker_)),
+        run_task_cv_(lock_.CreateConditionVariable()),
+        becomes_idle_callback_cv_(lock_.CreateConditionVariable()),
+        num_becomes_idle_callback_invocations_(0),
+        last_posted_task_index_(0),
+        last_run_task_index_(0),
+        ran_task_that_should_not_run_(false),
+        ran_tasks_in_wrong_order_(true) {}
+
+  Closure GetTaskThatShouldRunClosure() {
+    ++last_posted_task_index_;
+    return Bind(&TaskSchedulerWorkerThreadTest::RunTaskThatShouldRun,
+                Unretained(this), last_posted_task_index_);
+  }
+
+  Closure GetTaskThatShouldNotRunClosure() {
+    return Bind(&TaskSchedulerWorkerThreadTest::RunTaskThatShouldNotRun,
+                Unretained(this));
+  }
+
+  void WaitUntilLastPostedTaskHasRun() {
+    AutoSchedulerLock auto_lock(lock_);
+    while (last_posted_task_index_ != last_run_task_index_)
+      run_task_cv_->Wait();
+  }
+
+  void ExpectLastPostedTaskHasRun() {
+    AutoSchedulerLock auto_lock(lock_);
+    EXPECT_EQ(last_posted_task_index_, last_run_task_index_);
+  }
+
+  void WaitUntilNumBecomesIdleCallbackInvocations(
+      size_t expected_num_becomes_idle_callback_invocations) {
+    AutoSchedulerLock auto_lock(lock_);
+    while (num_becomes_idle_callback_invocations_ !=
+           expected_num_becomes_idle_callback_invocations) {
+      EXPECT_LT(num_becomes_idle_callback_invocations_,
+                expected_num_becomes_idle_callback_invocations);
+      becomes_idle_callback_cv_->Wait();
+    }
+  }
+
+  bool ran_task_that_should_not_run() const {
+    return ran_task_that_should_not_run_;
+  }
+
+ protected:
+  PriorityQueue shared_priority_queue_;
+  TaskTracker task_tracker_;
+  scoped_ptr<WorkerThread> worker_thread_;
+
+ private:
+  void ReinsertSequenceCallback(scoped_refptr<Sequence> sequence,
+                                const WorkerThread* worker_thread) {
+    const SequenceSortKey sort_key = sequence->GetSortKey();
+    shared_priority_queue_.BeginTransaction()->Push(make_scoped_ptr(
+        new PriorityQueue::SequenceAndSortKey(std::move(sequence), sort_key)));
+  }
+
+  void BecomesIdleCallback(WorkerThread* worker_thread) {
+    AutoSchedulerLock auto_lock(lock_);
+    ++num_becomes_idle_callback_invocations_;
+    becomes_idle_callback_cv_->Signal();
+  }
+
+  void RunTaskThatShouldRun(size_t index) {
+    AutoSchedulerLock auto_lock(lock_);
+
+    if (index != last_run_task_index_ + 1)
+      ran_tasks_in_wrong_order_ = true;
+
+    last_run_task_index_ = index;
+    run_task_cv_->Signal();
+  }
+
+  void RunTaskThatShouldNotRun() { ran_task_that_should_not_run_ = true; }
+
+  // Lock protecting |run_task_cv_|.
+  SchedulerLock lock_;
+
+  // Condition variable signaled each time a task completes its execution.
+  scoped_ptr<ConditionVariable> run_task_cv_;
+
+  // Condition variable signaled when BecomesIdleCallback() is invoked.
+  scoped_ptr<ConditionVariable> becomes_idle_callback_cv_;
+
+  // Number of times that BecomesIdleCallback() has been called.
+  size_t num_becomes_idle_callback_invocations_;
+
+  // 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_;
+
+  // True if tasks were run in the wrong order.
+  bool ran_tasks_in_wrong_order_;
+
+};
+
+TEST_F(TaskSchedulerWorkerThreadTest, PostOneSingleThreadedTask) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  worker_thread_->CreateTaskRunnerWithTraits(TaskTraits())
+      ->PostTask(FROM_HERE, GetTaskThatShouldRunClosure());
+
+  WaitUntilNumBecomesIdleCallbackInvocations(2);
+  ExpectLastPostedTaskHasRun();
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleSingleThreadedTasksNoWaitBetweenPosts) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+  for (size_t i = 0; i < 100; ++i)
+    task_runner->PostTask(FROM_HERE, GetTaskThatShouldRunClosure());
+
+  WaitUntilLastPostedTaskHasRun();
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleSingleThreadedTasksWaitBetweenPosts) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+  for (size_t i = 0; i < 100; ++i) {
+    task_runner->PostTask(FROM_HERE, GetTaskThatShouldRunClosure());
+    WaitUntilLastPostedTaskHasRun();
+    WaitUntilNumBecomesIdleCallbackInvocations(2 + i);
+  }
+
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest,
+       PostMultipleTasksTwoSingleThreadedTaskRunners) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  auto task_runner_a = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+  auto task_runner_b = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+
+  for (size_t i = 0; i < 100; ++i) {
+    task_runner_a->PostTask(FROM_HERE, GetTaskThatShouldRunClosure());
+    task_runner_b->PostTask(FROM_HERE, GetTaskThatShouldRunClosure());
+  }
+
+  WaitUntilLastPostedTaskHasRun();
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest, PostOneTaskInSharedPriorityQueue) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  PostTaskHelper(make_scoped_ptr(new Task(
+                     FROM_HERE, GetTaskThatShouldRunClosure(), TaskTraits())),
+                 new Sequence, &shared_priority_queue_, &task_tracker_);
+  EXPECT_TRUE(worker_thread_->WakeUp());
+
+  WaitUntilNumBecomesIdleCallbackInvocations(2);
+  ExpectLastPostedTaskHasRun();
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest, PostSharedAndSingleThreadedTasks) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  // Post a task in the shared priority queue.
+  PostTaskHelper(make_scoped_ptr(new Task(
+                     FROM_HERE, GetTaskThatShouldRunClosure(), TaskTraits())),
+                 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, GetTaskThatShouldRunClosure());
+
+  WaitUntilNumBecomesIdleCallbackInvocations(2);
+  ExpectLastPostedTaskHasRun();
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest, WakeUpIdleThread) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  EXPECT_TRUE(worker_thread_->WakeUp());
+  WaitUntilNumBecomesIdleCallbackInvocations(2);
+
+  worker_thread_->JoinForTesting();
+}
+
+TEST_F(TaskSchedulerWorkerThreadTest, WakeUpBusyThread) {
+  ASSERT_NE(nullptr, worker_thread_.get());
+  WaitUntilNumBecomesIdleCallbackInvocations(1);
+
+  // Post a task.
+  WaitableEvent event(false, false);
+  worker_thread_->CreateTaskRunnerWithTraits(TaskTraits())
+      ->PostTask(FROM_HERE, Bind(&WaitableEvent::Wait, Unretained(&event)));
+
+  // Because the thread is busy, WakeUp() should return false.
+  EXPECT_FALSE(worker_thread_->WakeUp());
+
+  event.Signal();
+  WaitUntilNumBecomesIdleCallbackInvocations(2);
+  worker_thread_->JoinForTesting();
+}
+
+}  // namespace internal
+}  // namespace base