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 16 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_);
   }
 
   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) {
+    if (!sequence->PopTask()) {
       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_) {}
+  // Constructs a TaskFactory that posts tasks with |execution_mode| to
+  // |thread_pool|.
+  TaskFactory(SchedulerThreadPool* thread_pool, ExecutionMode execution_mode)
+      : cv_(&lock_),
+        task_runner_(thread_pool->CreateTaskRunnerWithTraits(TaskTraits(),
+                                                             execution_mode)),
+        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_);
-    EXPECT_TRUE(task_runner->PostTask(
-        FROM_HERE, Bind(&TaskFactory::RunTaskCallback, Unretained(this),
-                        num_created_tasks_++, task_runner, post_nested_task,
-                        Unretained(event))));
+    task_runner_->PostTask(

[gab, 2016/04/08 16:52:25]

Doesn't hurt to keep the EXPECT_TRUE I guess (although we know the impl never
returns false it's still okay to test it?)

[robliao, 2016/04/08 17:11:11]

If we're operating on the premise that the return value of PostTask doesn't
matter, it would be reasonable to leave this as is and not encourage others to
check PostTask results.

Failures should surface themselves as test failures.

[fdoray, 2016/04/11 14:29:53]

Put back the EXPECT_TRUE. The impl returns false when a
CONTINUE/SKIP_ON_SHUTDOWN task is posted during shutdown.

+        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();
   }
 
+  bool RunsTasksOnCurrentThread() const {
+    return task_runner_->RunsTasksOnCurrentThread();
+  }
+
  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(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_(thread_pool_, execution_mode) {}
 
   const TaskFactory* factory() const { return &factory_; }
 
  private:
   void Run() override {
-    EXPECT_FALSE(task_runner_->RunsTasksOnCurrentThread());
+    EXPECT_FALSE(factory_.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)));
+    threads_posting_tasks.push_back(WrapUnique(
+        new ThreadPostingTasks(thread_pool_.get(), GetParam(), 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.
+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.push_back(WrapUnique(
+        new ThreadPostingTasks(thread_pool_.get(), GetParam(), 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.
+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.push_back(WrapUnique(
+        new ThreadPostingTasks(thread_pool_.get(), GetParam(), 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.

[robliao, 2016/04/08 17:11:12]

Worth mentioning that we do this to avoid sequence checking presumably?

[fdoray, 2016/04/11 14:29:53]

Done.

   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<std::unique_ptr<TaskFactory>> blocked_task_factories;
+  for (size_t i = 0; i < (kNumThreadsInThreadPool - 1); ++i) {
+    blocked_task_factories.push_back(
+        WrapUnique(new TaskFactory(thread_pool_.get(), GetParam())));
+    blocked_task_factories.back()->PostTask(false, &event);
+    blocked_task_factories.back()->WaitForAllTasksToRun();
+  }
 
   // Post |kNumTasksPostedPerThread| tasks that should all run despite the fact
   // that only one thread in |thread_pool_| isn't busy.
+  TaskFactory short_task_factory(thread_pool_.get(), GetParam());
   for (size_t i = 0; i < kNumTasksPostedPerThread; ++i)
-    factory.PostTask(task_runner, false, nullptr);
-  factory.WaitForAllTasksToRun();
+    short_task_factory.PostTask(false, nullptr);
+  short_task_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.
+  // its TaskFactory after it 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

[gab, 2016/04/08 16:52:25]

"tasks running" => "tasks/sequences running" ? to highlight that it's also
testing that independent sequences can also be scheduled when other sequences
are blocked?

[robliao, 2016/04/08 17:11:12]

Hrm... this is an interesting point.
We are running tasks, and every task is associated with a sequence, so that
sequence is also running.

[fdoray, 2016/04/11 14:29:53]

Done.

-  // 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<std::unique_ptr<TaskFactory>> factories;
+  for (size_t i = 0; i < kNumThreadsInThreadPool; ++i) {
+    factories.push_back(
+        WrapUnique(new TaskFactory(thread_pool_.get(), 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
-  // |factory| after it is destroyed.
+  // its TaskFactory after it 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));
+
 }  // namespace
 }  // namespace internal
 }  // namespace base