TaskScheduler [11] Support ExecutionMode::SINGLE_THREADED.

base/task_scheduler/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/scheduler_worker_thread.h"
 
 #include <stddef.h>
 
 #include <memory>
 #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/task_scheduler/delayed_task_manager.h"
+#include "base/task_scheduler/priority_queue.h"
 #include "base/task_scheduler/scheduler_lock.h"
 #include "base/task_scheduler/sequence.h"
 #include "base/task_scheduler/task.h"
 #include "base/task_scheduler/task_tracker.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace base {
 namespace internal {
 namespace {
 
-const size_t kNumSequencesPerTest = 150;
-
 // The test parameter is the number of Tasks per Sequence returned by GetWork().
-class TaskSchedulerWorkerThreadTest : public testing::TestWithParam<size_t>,
-                                      public SchedulerWorkerThread::Delegate {
+class TaskSchedulerWorkerThreadDelegateWorkTest
+    : public testing::TestWithParam<size_t>,
+      public SchedulerWorkerThread::Delegate {
  protected:
-  TaskSchedulerWorkerThreadTest()
-      : main_entry_called_(true, false),
+  TaskSchedulerWorkerThreadDelegateWorkTest()
+      : delayed_task_manager_(Bind(&DoNothing)),
+        main_entry_called_(true, false),
         num_get_work_cv_(lock_.CreateConditionVariable()) {}
 
   void SetUp() override {
     worker_thread_ = SchedulerWorkerThread::CreateSchedulerWorkerThread(
-        ThreadPriority::NORMAL, this, &task_tracker_);
+        ThreadPriority::NORMAL, this, &task_tracker_, &delayed_task_manager_,
+        &dummy_priority_queue_);
     ASSERT_TRUE(worker_thread_);
     main_entry_called_.Wait();
   }
 
   void TearDown() override {
     worker_thread_->JoinForTesting();
   }
 
   size_t TasksPerSequence() const { return GetParam(); }
 
@@ skipping 23 matching lines @@
   std::vector<scoped_refptr<Sequence>> CreatedSequences() {
     AutoSchedulerLock auto_lock(lock_);
     return created_sequences_;
   }
 
   std::vector<scoped_refptr<Sequence>> EnqueuedSequences() {
     AutoSchedulerLock auto_lock(lock_);
     return enqueued_sequences_;
   }
 
+  const size_t kNumSequencesPerTest = 150;

[gab, 2016/04/18 18:04:02]

s/const/constexpr/ (see chromium-dev about new C++11 feature for Chromium --
tl;dr; prefer constexpr now for all compile-time constants)

[fdoray, 2016/04/18 19:04:50]

Done.

+
   std::unique_ptr<SchedulerWorkerThread> worker_thread_;
 
  private:
   // SchedulerWorkerThread::Delegate:
   void OnMainEntry() override {
     // Without this |auto_lock|, OnMainEntry() could be called twice without
     // generating an error.
     AutoSchedulerLock auto_lock(lock_);
     EXPECT_FALSE(main_entry_called_.IsSignaled());
     main_entry_called_.Signal();
   }
 
-  scoped_refptr<Sequence> GetWork(
-      SchedulerWorkerThread* worker_thread) override {
+  scoped_refptr<Sequence> GetWork(SchedulerWorkerThread* worker_thread,
+                                  PriorityQueue* single_threaded_priority_queue,
+                                  bool* is_single_threaded_sequence) override {
     EXPECT_EQ(worker_thread_.get(), worker_thread);
 
     {
       AutoSchedulerLock auto_lock(lock_);
 
       // Increment the number of times that this method has been called.
       ++num_get_work_;
       num_get_work_cv_->Signal();
 
       // Verify that this method isn't called more times than expected.
       EXPECT_LE(num_get_work_, max_get_work_);
 
       // Check if a Sequence should be returned.
       if (num_sequences_to_create_ == 0)
         return nullptr;
       --num_sequences_to_create_;
     }
 
     // Create a Sequence with TasksPerSequence() Tasks.
     scoped_refptr<Sequence> sequence(new Sequence);
     for (size_t i = 0; i < TasksPerSequence(); ++i) {
       std::unique_ptr<Task> task(new Task(
-          FROM_HERE, Bind(&TaskSchedulerWorkerThreadTest::RunTaskCallback,
-                          Unretained(this)),
+          FROM_HERE,
+          Bind(&TaskSchedulerWorkerThreadDelegateWorkTest::RunTaskCallback,
+               Unretained(this)),
           TaskTraits()));
       EXPECT_TRUE(task_tracker_.WillPostTask(task.get()));
       sequence->PushTask(std::move(task));
     }
 
     {
       // Add the Sequence to the vector of created Sequences.
       AutoSchedulerLock auto_lock(lock_);
       created_sequences_.push_back(sequence);
     }
@@ skipping 21 matching lines @@
     EXPECT_LE(enqueued_sequences_.size(), created_sequences_.size());
   }
 
   void RunTaskCallback() {
     AutoSchedulerLock auto_lock(lock_);
     ++num_run_tasks_;
     EXPECT_LE(num_run_tasks_, created_sequences_.size());
   }
 
   TaskTracker task_tracker_;
+  DelayedTaskManager delayed_task_manager_;
+  PriorityQueue dummy_priority_queue_;
 
   // Synchronizes access to all members below.
   mutable SchedulerLock lock_;
 
   // Signaled once OnMainEntry() has been called.
   WaitableEvent main_entry_called_;
 
   // Number of Sequences that should be created by GetWork(). When this
   // is 0, GetWork() returns nullptr.
   size_t num_sequences_to_create_ = 0;
 
   // Number of times that GetWork() has been called.
   size_t num_get_work_ = 0;
 
   // Maximum number of times that GetWork() can be called.
   size_t max_get_work_ = 0;
 
   // Condition variable signaled when |num_get_work_| is incremented.
   std::unique_ptr<ConditionVariable> num_get_work_cv_;
 
   // Sequences created by GetWork().
   std::vector<scoped_refptr<Sequence>> created_sequences_;
 
   // Sequences passed to EnqueueSequence().
   std::vector<scoped_refptr<Sequence>> enqueued_sequences_;
 
   // Number of times that RunTaskCallback() has been called.
   size_t num_run_tasks_ = 0;
 
-  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerThreadTest);
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerThreadDelegateWorkTest);
 };
 
 // Verify that when GetWork() continuously returns Sequences, all Tasks in these
 // Sequences run successfully. The test wakes up the SchedulerWorkerThread once.
-TEST_P(TaskSchedulerWorkerThreadTest, ContinuousWork) {
+TEST_P(TaskSchedulerWorkerThreadDelegateWorkTest, ContinuousWork) {
   // Set GetWork() to return |kNumSequencesPerTest| Sequences before starting to
   // return nullptr.
   SetNumSequencesToCreate(kNumSequencesPerTest);
 
   // Expect |kNumSequencesPerTest| calls to GetWork() in which it returns a
   // Sequence and one call in which its returns nullptr.
   const size_t kExpectedNumGetWork = kNumSequencesPerTest + 1;
   SetMaxGetWork(kExpectedNumGetWork);
 
   // Wake up |worker_thread_| and wait until GetWork() has been invoked the
   // expected amount of times.
   worker_thread_->WakeUp();
   WaitForNumGetWork(kExpectedNumGetWork);
 
   // All tasks should have run.
   EXPECT_EQ(kNumSequencesPerTest, NumRunTasks());
 
   // If Sequences returned by GetWork() contain more than one Task, they aren't
   // empty after the worker thread pops Tasks from them and thus should be
   // returned to EnqueueSequence().
   if (TasksPerSequence() > 1)
     EXPECT_EQ(CreatedSequences(), EnqueuedSequences());
   else
     EXPECT_TRUE(EnqueuedSequences().empty());
 }
 
 // Verify that when GetWork() alternates between returning a Sequence and
 // returning nullptr, all Tasks in the returned Sequences run successfully. The
 // test wakes up the SchedulerWorkerThread once for each Sequence.
-TEST_P(TaskSchedulerWorkerThreadTest, IntermittentWork) {
+TEST_P(TaskSchedulerWorkerThreadDelegateWorkTest, IntermittentWork) {
   for (size_t i = 0; i < kNumSequencesPerTest; ++i) {
     // Set GetWork() to return 1 Sequence before starting to return
     // nullptr.
     SetNumSequencesToCreate(1);
 
     // Expect |i + 1| calls to GetWork() in which it returns a Sequence and
     // |i + 1| calls in which it returns nullptr.
     const size_t expected_num_get_work = 2 * (i + 1);
     SetMaxGetWork(expected_num_get_work);
 
     // Wake up |worker_thread_| and wait until GetWork() has been invoked
     // the expected amount of times.
     worker_thread_->WakeUp();
     WaitForNumGetWork(expected_num_get_work);
 
     // The Task should have run
     EXPECT_EQ(i + 1, NumRunTasks());
 
     // If Sequences returned by GetWork() contain more than one Task, they
     // aren't empty after the worker thread pops Tasks from them and thus should
     // be returned to EnqueueSequence().
     if (TasksPerSequence() > 1)
       EXPECT_EQ(CreatedSequences(), EnqueuedSequences());
     else
       EXPECT_TRUE(EnqueuedSequences().empty());
   }
 }
 
 INSTANTIATE_TEST_CASE_P(OneTaskPerSequence,
-                        TaskSchedulerWorkerThreadTest,
+                        TaskSchedulerWorkerThreadDelegateWorkTest,
                         ::testing::Values(1));
 INSTANTIATE_TEST_CASE_P(TwoTasksPerSequence,
-                        TaskSchedulerWorkerThreadTest,
+                        TaskSchedulerWorkerThreadDelegateWorkTest,
                         ::testing::Values(2));
 
+class TaskSchedulerWorkerThreadSingleThreadedWorkTest
+    : public testing::Test,
+      public SchedulerWorkerThread::Delegate {
+ protected:
+  TaskSchedulerWorkerThreadSingleThreadedWorkTest()
+      : delayed_task_manager_(Bind(&DoNothing)),
+        num_run_tasks_cv_(lock_.CreateConditionVariable()) {}
+
+  void SetUp() override {
+    worker_thread_ = SchedulerWorkerThread::CreateSchedulerWorkerThread(
+        ThreadPriority::NORMAL, this, &task_tracker_, &delayed_task_manager_,
+        &dummy_priority_queue_);
+    ASSERT_TRUE(worker_thread_);
+  }
+
+  void TearDown() override { worker_thread_->JoinForTesting(); }
+
+  void PostTestTask(scoped_refptr<TaskRunner> task_runner) {
+    Closure closure;
+    {
+      AutoSchedulerLock auto_lock(lock_);
+      closure = Bind(
+          &TaskSchedulerWorkerThreadSingleThreadedWorkTest::RunTaskCallback,
+          Unretained(this), num_posted_tasks_++, task_runner);

[gab, 2016/04/18 18:04:02]

Only need to lock to get index from |num_posted_tasks_++|, right? Could we
instead build the closure outside the lock to highlight that?

[fdoray, 2016/04/18 19:04:50]

Done.

+    }
+    task_runner->PostTask(FROM_HERE, closure);
+  }
+
+  void WaitForAllTasksToRun() {
+    AutoSchedulerLock auto_lock(lock_);
+    while (num_run_tasks_ < num_posted_tasks_)
+      num_run_tasks_cv_->Wait();
+  }
+
+  const size_t kNumTasksPerTest = 150;
+
+  std::unique_ptr<SchedulerWorkerThread> worker_thread_;
+
+ private:
+  // SchedulerWorkerThread::Delegate:
+  void OnMainEntry() override {}
+
+  scoped_refptr<Sequence> GetWork(SchedulerWorkerThread* worker_thread,
+                                  PriorityQueue* single_threaded_priority_queue,
+                                  bool* is_single_threaded_sequence) override {
+    EXPECT_EQ(worker_thread_.get(), worker_thread);
+
+    // Return a Sequence from |single_threaded_priority_queue| or nullptr if it
+    // is empty.
+    auto transaction = single_threaded_priority_queue->BeginTransaction();
+    auto sequence = transaction->Peek().sequence;
+    if (!sequence)
+      return nullptr;
+    *is_single_threaded_sequence = true;
+    transaction->Pop();
+    return sequence;
+  }
+
+  void EnqueueSequence(scoped_refptr<Sequence> sequence) override {
+    ADD_FAILURE() << "EnqueueSequence shouldn't be called in a test that only "
+                     "posts single-threaded Tasks.";
+  }
+
+  void RunTaskCallback(size_t index, scoped_refptr<TaskRunner> task_runner) {
+    AutoSchedulerLock auto_lock(lock_);
+
+    EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread());
+
+    // Verify that tasks run in posting order.
+    EXPECT_EQ(num_run_tasks_, index);
+
+    // Verify that we don't run more tasks than posted.
+    ++num_run_tasks_;
+    EXPECT_LE(num_run_tasks_, num_posted_tasks_);
+
+    num_run_tasks_cv_->Signal();
+  }
+
+  TaskTracker task_tracker_;
+  DelayedTaskManager delayed_task_manager_;
+  PriorityQueue dummy_priority_queue_;
+
+  // Synchronizes access to all members below.
+  mutable SchedulerLock lock_;
+
+  // Number of tasks posted by PostTestTask().
+  size_t num_posted_tasks_ = 0;
+
+  // Number of times that RunTaskCallback() has been called.
+  size_t num_run_tasks_ = 0;
+
+  // Condition variable signaled when |num_run_tasks_| is incremented.
+  std::unique_ptr<ConditionVariable> num_run_tasks_cv_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerThreadSingleThreadedWorkTest);
+};
+
+TEST_F(TaskSchedulerWorkerThreadSingleThreadedWorkTest, ContinuousWork) {
+  auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+  EXPECT_FALSE(task_runner->RunsTasksOnCurrentThread());
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i)
+    PostTestTask(task_runner);
+  WaitForAllTasksToRun();
+}
+
+TEST_F(TaskSchedulerWorkerThreadSingleThreadedWorkTest, IntermittentWork) {
+  auto task_runner = worker_thread_->CreateTaskRunnerWithTraits(TaskTraits());
+  EXPECT_FALSE(task_runner->RunsTasksOnCurrentThread());
+
+  for (size_t i = 0; i < kNumTasksPerTest; ++i) {
+    PostTestTask(task_runner);
+    WaitForAllTasksToRun();
+  }
+}
+
 }  // namespace
 }  // namespace internal
 }  // namespace base