TaskScheduler [8] SEQUENCED TaskRunners in 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>
@@ skipping 13 matching lines @@
 #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 {
+class TaskSchedulerThreadPoolTest
+    : public testing::TestWithParam<ExecutionMode> {
  protected:
   TaskSchedulerThreadPoolTest() = default;
 
   void SetUp() override {
     thread_pool_ = SchedulerThreadPool::CreateThreadPool(
         ThreadPriority::NORMAL, kNumThreadsInThreadPool,
         Bind(&TaskSchedulerThreadPoolTest::RanTaskFromSequenceCallback,
              Unretained(this)),
         &task_tracker_);
     ASSERT_TRUE(thread_pool_);
@@ skipping 19 matching lines @@
     }
   }
 
   TaskTracker task_tracker_;
 
   DISALLOW_COPY_AND_ASSIGN(TaskSchedulerThreadPoolTest);
 };
 
 class TaskFactory {
  public:
-  TaskFactory() : cv_(&lock_) {}
+  // Constructs a TaskFactory that posts tasks through |task_runner|.
+  // |execution_mode| is the execution mode of |task_runner|.
+  TaskFactory(scoped_refptr<TaskRunner> task_runner,

[robliao, 2016/04/04 18:45:50]

I wonder if there's a way we can extract this out of a task runner. I currently
don't see an easy way to do this at the moment.

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

I changed the code to avoid relying on the caller to provide matching task
runner / execution mode.

+              ExecutionMode execution_mode)
+      : cv_(&lock_),
+        task_runner_(std::move(task_runner)),
+        execution_mode_(execution_mode) {}
 
-  // 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) {
+  // Posts a task. If |post_nested_task| is true, the task will post a new task
+  // when it runs. If |event| is set, the task will block until it is signaled.
+  void PostTask(bool post_nested_task, WaitableEvent* event) {
     AutoLock auto_lock(lock_);
-    task_runner->PostTask(
-        FROM_HERE, Bind(&TaskFactory::RunTaskCallback, Unretained(this),
-                        num_created_tasks_++, task_runner, post_nested_task,
-                        Unretained(event)));
+    task_runner_->PostTask(
+        FROM_HERE,
+        Bind(&TaskFactory::RunTaskCallback, Unretained(this),
+             num_created_tasks_++, 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);
+      PostTask(false, nullptr);
 
-    EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread());
+    EXPECT_TRUE(task_runner_->RunsTasksOnCurrentThread());
 
     {
       AutoLock auto_lock(lock_);
 
+      if (execution_mode_ == ExecutionMode::SEQUENCED &&
+          task_index != run_tasks_.size()) {
+        ADD_FAILURE() << "A SEQUENCED task didn't run in the expected order.";
+      }
+
       if (run_tasks_.find(task_index) != run_tasks_.end())
         ADD_FAILURE() << "A task ran more than once.";
       run_tasks_.insert(task_index);
 
       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_;
 
+  // Task runner through which this factory posts tasks.
+  const scoped_refptr<TaskRunner> task_runner_;
+
+  // Execution mode of |task_runner_|.
+  const ExecutionMode execution_mode_;
+
   // 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) {}
+        post_nested_task_(post_nested_task),
+        factory_(task_runner_, execution_mode) {}
 
   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);
+      factory_.PostTask(post_nested_task_, nullptr);
     }
   }
 
   SchedulerThreadPool* const thread_pool_;
   scoped_refptr<TaskRunner> task_runner_;
   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.
+TEST_P(TaskSchedulerThreadPoolTest, PostTasks) {
+  // Create threads to post tasks.
   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)));

[robliao, 2016/04/04 18:45:50]

Should PARALLEL be parameterized? All of the tests below do not change this
parameter.

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

Done.

     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());
   }
 }
 
-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.
+TEST_P(TaskSchedulerThreadPoolTest, PostTasksWaitAllThreadsIdle) {
+  // Create threads to post tasks. 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());
   }
 }
 
-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.
+TEST_P(TaskSchedulerThreadPoolTest, NestedPostTasks) {
+  // Create threads to post tasks. 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;
-
+TEST_P(TaskSchedulerThreadPoolTest, PostTasksWithOneAvailableThread) {
   // Post tasks to keep all threads busy except one until |event| is signaled.
+  // Use different factories so that tasks are added to different sequences when
+  // the execution mode is SEQUENCED.
   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();
+  std::vector<scoped_ptr<TaskFactory>> factories;
+  for (size_t i = 0; i < (kNumThreadsInThreadPool - 1); ++i) {
+    factories.push_back(make_scoped_ptr(new TaskFactory(
+        thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(), GetParam()),
+        GetParam())));
+    factories.back()->PostTask(false, &event);
+    factories.back()->WaitForAllTasksToRun();
+  }
 
   // Post |kNumTasksPostedPerThread| tasks that should all run despite the fact
   // that only one thread in |thread_pool_| isn't busy.
+  factories.push_back(make_scoped_ptr(new TaskFactory(

[robliao, 2016/04/04 18:45:50]

Should this really be a part of |factories| or does it make more sense for this
to be outside of |factories|?

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

Done.

+      thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(), GetParam()),
+      GetParam())));
   for (size_t i = 0; i < kNumTasksPostedPerThread; ++i)
-    factory.PostTask(task_runner, false, nullptr);
-  factory.WaitForAllTasksToRun();
+    factories.back()->PostTask(false, nullptr);
+  factories.back()->WaitForAllTasksToRun();
 
   // Release tasks waiting on |event|.
   event.Signal();
 
   // Wait until all worker threads are idle to be sure that no task accesses
-  // its TaskFactory after |factory| is destroyed.
+  // its TaskFactory after |factories| is destroyed.
   thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
 }
 
-TEST_F(TaskSchedulerThreadPoolTest, Saturate) {
-  TaskFactory factory;
-
+TEST_P(TaskSchedulerThreadPoolTest, Saturate) {
   // Verify that it is possible to have |kNumThreadsInThreadPool| tasks running
-  // simultaneously.
+  // simultaneously. Use different factories so that tasks are added to
+  // different sequences when the execution mode is SEQUENCED.
   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();
+  std::vector<scoped_ptr<TaskFactory>> factories;
+  for (size_t i = 0; i < kNumThreadsInThreadPool; ++i) {
+    factories.push_back(make_scoped_ptr(new TaskFactory(
+        thread_pool_->CreateTaskRunnerWithTraits(TaskTraits(), GetParam()),
+        GetParam())));
+    factories.back()->PostTask(false, &event);
+    factories.back()->WaitForAllTasksToRun();
+  }
 
   // Release tasks waiting on |event|.
   event.Signal();
 
   // Wait until all worker threads are idle to be sure that no task accesses
-  // its TaskFactory after |factory| is destroyed.
+  // its TaskFactory after |factories| is destroyed.
   thread_pool_->WaitForAllWorkerThreadsIdleForTesting();
 }
 
+INSTANTIATE_TEST_CASE_P(Parallel,
+                        TaskSchedulerThreadPoolTest,
+                        ::testing::Values(ExecutionMode::PARALLEL));
+INSTANTIATE_TEST_CASE_P(Sequenced,
+                        TaskSchedulerThreadPoolTest,
+                        ::testing::Values(ExecutionMode::SEQUENCED));
+
 // 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) {
   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;
 
@@ skipping 34 matching lines @@
   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