TaskScheduler [8] SEQUENCED TaskRunners in SchedulerThreadPool.

base/task_scheduler/scheduler_thread_pool_unittest.cc

Index: base/task_scheduler/scheduler_thread_pool_unittest.cc
diff --git a/base/task_scheduler/scheduler_thread_pool_unittest.cc b/base/task_scheduler/scheduler_thread_pool_unittest.cc
index 213467b4a9e02ef33c5d04d2891ded88247b42bc..0a1f5ae5c1df0372526c7153797857ee5839a380 100644
--- a/base/task_scheduler/scheduler_thread_pool_unittest.cc
+++ b/base/task_scheduler/scheduler_thread_pool_unittest.cc
@@ -34,7 +34,8 @@ 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;
 
@@ -60,8 +61,7 @@ class TaskSchedulerThreadPoolTest : public testing::Test {
     // 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);
     }
@@ -74,19 +74,22 @@ class TaskSchedulerThreadPoolTest : public testing::Test {
 
 class TaskFactory {
  public:
-  TaskFactory() : cv_(&lock_) {}
-
-  // 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) {
+  // 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. 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
@@ -102,19 +105,27 @@ class TaskFactory {
     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);
@@ -132,6 +143,12 @@ class TaskFactory {
   // 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;
 
@@ -154,26 +171,24 @@ class ThreadPostingTasks : public SimpleThread {
                      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_;
@@ -181,12 +196,12 @@ class ThreadPostingTasks : public SimpleThread {
   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();
   }
 
@@ -203,14 +218,14 @@ TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasks) {
   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();
   }
 
@@ -227,13 +242,13 @@ TEST_F(TaskSchedulerThreadPoolTest, PostParallelTasksWaitAllThreadsIdle) {
   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();
   }
 
@@ -250,51 +265,62 @@ TEST_F(TaskSchedulerThreadPoolTest, NestedPostParallelTasks) {
   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