Move Task Scheduler Single Thread Task Runners to Dedicated Threads

base/task_scheduler/scheduler_worker_pool_impl_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_pool_impl.h"
 
 #include <stddef.h>
 
 #include <memory>
 #include <unordered_set>
@@ skipping 49 matching lines @@
 
 using StandbyThreadPolicy = SchedulerWorkerPoolParams::StandbyThreadPolicy;
 
 class TaskSchedulerWorkerPoolImplTest
     : public testing::TestWithParam<test::ExecutionMode> {
  protected:
   TaskSchedulerWorkerPoolImplTest()
       : service_thread_("TaskSchedulerServiceThread") {}
 
   void SetUp() override {
-    InitializeWorkerPool(TimeDelta::Max(), kNumWorkersInWorkerPool);
+    if (GetParam() == test::ExecutionMode::SINGLE_THREADED)
+      InitializeSingleThreadWorkerPool(TimeDelta::Max());
+    else
+      InitializeWorkerPool(TimeDelta::Max(), kNumWorkersInWorkerPool);
   }
 
   void TearDown() override {
     service_thread_.Stop();
     worker_pool_->WaitForAllWorkersIdleForTesting();
     worker_pool_->JoinForTesting();
   }
 
   void InitializeWorkerPool(TimeDelta suggested_reclaim_time,
                             size_t num_workers) {
     ASSERT_FALSE(worker_pool_);
     ASSERT_FALSE(delayed_task_manager_);
     service_thread_.Start();
     delayed_task_manager_ =
         base::MakeUnique<DelayedTaskManager>(service_thread_.task_runner());
     worker_pool_ = SchedulerWorkerPoolImpl::Create(
         SchedulerWorkerPoolParams("TestWorkerPool", ThreadPriority::NORMAL,
                                   StandbyThreadPolicy::LAZY, num_workers,
                                   suggested_reclaim_time),
         Bind(&TaskSchedulerWorkerPoolImplTest::ReEnqueueSequenceCallback,
              Unretained(this)),
         &task_tracker_, delayed_task_manager_.get());
     ASSERT_TRUE(worker_pool_);
   }
 
+  void InitializeSingleThreadWorkerPool(TimeDelta suggested_reclaim_time) {
+    ASSERT_FALSE(worker_pool_);
+    ASSERT_FALSE(delayed_task_manager_);
+    service_thread_.Start();
+    delayed_task_manager_ =
+        base::MakeUnique<DelayedTaskManager>(service_thread_.task_runner());
+    worker_pool_ = CreateSingleThreadWorkerPool(
+        suggested_reclaim_time,
+        Bind(&TaskSchedulerWorkerPoolImplTest::
+                 UnregisterSingleThreadWorkerPoolCallback,
+             Unretained(this)));
+    ASSERT_TRUE(worker_pool_);
+  }
+
+  std::unique_ptr<SchedulerWorkerPoolImpl> CreateSingleThreadWorkerPool(
+      TimeDelta suggested_reclaim_time,
+      SchedulerWorkerPoolImpl::UnregisterSingleThreadWorkerPoolCallback
+          unregister_single_thread_worker_pool_callback) {
+    return SchedulerWorkerPoolImpl::CreateSingleThreadWorkerPool(
+        SchedulerWorkerPoolParams("TestWorkerPool", ThreadPriority::NORMAL,
+                                  StandbyThreadPolicy::LAZY, 1,
+                                  suggested_reclaim_time),
+        unregister_single_thread_worker_pool_callback, &task_tracker_,
+        delayed_task_manager_.get());
+  }
+
   std::unique_ptr<SchedulerWorkerPoolImpl> worker_pool_;
 
   TaskTracker task_tracker_;
   Thread service_thread_;
   std::unique_ptr<DelayedTaskManager> delayed_task_manager_;
 
  private:
   void ReEnqueueSequenceCallback(scoped_refptr<Sequence> sequence) {
     // In production code, this callback would be implemented by the
     // TaskScheduler which would first determine which PriorityQueue the
     // sequence must be re-enqueued.
     const SequenceSortKey sort_key(sequence->GetSortKey());
     worker_pool_->ReEnqueueSequence(std::move(sequence), sort_key);
   }
 
+  void UnregisterSingleThreadWorkerPoolCallback(
+      const SchedulerWorkerPoolImpl* scheduler_worker_pool) {
+    EXPECT_EQ(worker_pool_.get(), scheduler_worker_pool);
+  }
+
   DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolImplTest);
 };
 
 scoped_refptr<TaskRunner> CreateTaskRunnerWithExecutionMode(
     SchedulerWorkerPoolImpl* worker_pool,
     test::ExecutionMode execution_mode) {
   // Allow tasks posted to the returned TaskRunner to wait on a WaitableEvent.
   const TaskTraits traits = TaskTraits().WithBaseSyncPrimitives();
   switch (execution_mode) {
     case test::ExecutionMode::PARALLEL:
@@ skipping 129 matching lines @@
 
   // Wait until all workers are idle to be sure that no task accesses its
   // TestTaskFactory after |thread_posting_tasks| is destroyed.
   worker_pool_->WaitForAllWorkersIdleForTesting();
 }
 
 TEST_P(TaskSchedulerWorkerPoolImplTest, PostTasksWithOneAvailableWorker) {
   // Post blocking tasks to keep all workers busy except one until |event| is
   // signaled. Use different factories so that tasks are added to different
   // sequences and can run simultaneously when the execution mode is SEQUENCED.
+  const size_t num_workers_in_pool =
+      (GetParam() == test::ExecutionMode::SINGLE_THREADED)
+          ? 1
+          : kNumWorkersInWorkerPool;
   WaitableEvent event(WaitableEvent::ResetPolicy::MANUAL,
                       WaitableEvent::InitialState::NOT_SIGNALED);
   std::vector<std::unique_ptr<test::TestTaskFactory>> blocked_task_factories;
-  for (size_t i = 0; i < (kNumWorkersInWorkerPool - 1); ++i) {
+  for (size_t i = 0; i < (num_workers_in_pool - 1); ++i) {
     blocked_task_factories.push_back(MakeUnique<test::TestTaskFactory>(
         CreateTaskRunnerWithExecutionMode(worker_pool_.get(), GetParam()),
         GetParam()));
     EXPECT_TRUE(blocked_task_factories.back()->PostTask(
         PostNestedTask::NO, Bind(&WaitableEvent::Wait, Unretained(&event))));
     blocked_task_factories.back()->WaitForAllTasksToRun();
   }
 
   // Post |kNumTasksPostedPerThread| tasks that should all run despite the fact
   // that only one worker in |worker_pool_| isn't busy.
@@ skipping 10 matching lines @@
   // Wait until all workers are idle to be sure that no task accesses
   // its TestTaskFactory after it is destroyed.
   worker_pool_->WaitForAllWorkersIdleForTesting();
 }
 
 TEST_P(TaskSchedulerWorkerPoolImplTest, Saturate) {
   // Verify that it is possible to have |kNumWorkersInWorkerPool|
   // tasks/sequences running simultaneously. Use different factories so that the
   // blocking tasks are added to different sequences and can run simultaneously
   // when the execution mode is SEQUENCED.
+  const size_t num_workers_in_pool =
+      (GetParam() == test::ExecutionMode::SINGLE_THREADED)
+          ? 1
+          : kNumWorkersInWorkerPool;
   WaitableEvent event(WaitableEvent::ResetPolicy::MANUAL,
                       WaitableEvent::InitialState::NOT_SIGNALED);
   std::vector<std::unique_ptr<test::TestTaskFactory>> factories;
-  for (size_t i = 0; i < kNumWorkersInWorkerPool; ++i) {
+  for (size_t i = 0; i < num_workers_in_pool; ++i) {
     factories.push_back(MakeUnique<test::TestTaskFactory>(
         CreateTaskRunnerWithExecutionMode(worker_pool_.get(), GetParam()),
         GetParam()));
     EXPECT_TRUE(factories.back()->PostTask(
         PostNestedTask::NO, Bind(&WaitableEvent::Wait, Unretained(&event))));
     factories.back()->WaitForAllTasksToRun();
   }
 
   // Release tasks waiting on |event|.
   event.Signal();
@@ skipping 73 matching lines @@
 namespace {
 
 // Same as TaskSchedulerWorkerPoolImplTest but its SchedulerWorkerPoolImpl
 // instance uses |max_threads == 1|.
 class TaskSchedulerWorkerPoolImplSingleWorkerTest
     : public TaskSchedulerWorkerPoolImplTest {
  public:
   TaskSchedulerWorkerPoolImplSingleWorkerTest() = default;
 
  protected:
-  void SetUp() override {
-    InitializeWorkerPool(TimeDelta::Max(), 1);
-  }
+  void SetUp() override { InitializeSingleThreadWorkerPool(TimeDelta::Max()); }
 
  private:
   DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolImplSingleWorkerTest);
 };
 
 }  // namespace
 
 // Verify that the RunsTasksOnCurrentThread() method of a
 // SchedulerSingleThreadTaskRunner returns false when called from a task that
 // isn't part of its sequence even though it's running on that
@@ skipping 41 matching lines @@
   void InitializeThreadChecker() {
     thread_checker_.reset(new ThreadCheckerImpl());
   }
 
   void CheckValidThread() {
     EXPECT_TRUE(thread_checker_->CalledOnValidThread());
   }
 
  protected:
   void SetUp() override {
-    InitializeWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
+    InitializeSingleThreadWorkerPool(kReclaimTimeForDetachTests);
   }
 
   TaskSchedulerWorkerPoolSingleThreadedTest() = default;
 
  private:
   std::unique_ptr<ThreadCheckerImpl> thread_checker_;
 
   DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolSingleThreadedTest);
 };
 
@@ skipping 265 matching lines @@
   *thread_id = PlatformThread::CurrentId();
 }
 
 void VerifyThreadIdIsNot(PlatformThreadId thread_id) {
   EXPECT_NE(thread_id, PlatformThread::CurrentId());
 }
 
 }  // namespace
 
 TEST_F(TaskSchedulerWorkerPoolHistogramTest, NumTasksBeforeDetach) {
-  InitializeWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
+  InitializeSingleThreadWorkerPool(kReclaimTimeForDetachTests);
+  std::unique_ptr<SchedulerWorkerPoolImpl> other_worker_pool =
+      CreateSingleThreadWorkerPool(
+          kReclaimTimeForDetachTests,
+          base::Bind(
+              [](const SchedulerWorkerPoolImpl* scheduler_worker_pool) {}));
 
   // This test assumes that the TaskRunners aren't assigned to the same worker.
   auto task_runner =
       worker_pool_->CreateSingleThreadTaskRunnerWithTraits(TaskTraits());
   auto other_task_runner =

[fdoray, 2017/01/27 16:47:35]

|other_task_runner| was used to change the order of workers on the idle stack
and allow the worker associated with |task_runner| to detach. Now that a worker
in single-threaded pool is always allowed to detach even when it's on top of the
idle stack, you don't need |other_task_runner|. 

[robliao, 2017/01/27 21:25:41]

Removed!

-      worker_pool_->CreateSingleThreadTaskRunnerWithTraits(TaskTraits());
+      other_worker_pool->CreateSingleThreadTaskRunnerWithTraits(TaskTraits());
+  other_worker_pool->DisallowWorkerDetachmentForTesting();
 
   // Post 3 tasks and wait until they run.
   PlatformThreadId thread_id;
   task_runner->PostTask(FROM_HERE,
                         Bind(&CaptureThreadId, Unretained(&thread_id)));
   task_runner->PostTask(FROM_HERE, Bind(&DoNothing));
   task_runner->PostTask(FROM_HERE, Bind(&DoNothing));
   worker_pool_->WaitForAllWorkersIdleForTesting();
 
   // To allow the SchedulerWorker associated with |task_runner| to detach:
   // - Make sure it isn't on top of the idle stack by waking up another
   //   SchedulerWorker and waiting until it goes back to sleep.
   // - Release |task_runner|.
   other_task_runner->PostTask(FROM_HERE, Bind(&VerifyThreadIdIsNot, thread_id));
   worker_pool_->WaitForAllWorkersIdleForTesting();
+  other_worker_pool->WaitForAllWorkersIdleForTesting();
   task_runner = nullptr;
 
   // Allow the SchedulerWorker that was associated with |task_runner| to detach.
   PlatformThread::Sleep(kReclaimTimeForDetachTests + kExtraTimeToWaitForDetach);
   worker_pool_->DisallowWorkerDetachmentForTesting();
 
   // Verify that counts were recorded to the histogram as expected.
   const auto* histogram = worker_pool_->num_tasks_before_detach_histogram();
   EXPECT_EQ(0, histogram->SnapshotSamples()->GetCount(0));
   EXPECT_EQ(1, histogram->SnapshotSamples()->GetCount(3));
   EXPECT_EQ(0, histogram->SnapshotSamples()->GetCount(10));
+  other_worker_pool->JoinForTesting();
 }
 
 TEST(TaskSchedulerWorkerPoolStandbyPolicyTest, InitLazy) {
   TaskTracker task_tracker;
   DelayedTaskManager delayed_task_manager(
       make_scoped_refptr(new TestSimpleTaskRunner));
   auto worker_pool = SchedulerWorkerPoolImpl::Create(
       SchedulerWorkerPoolParams("LazyPolicyWorkerPool", ThreadPriority::NORMAL,
                                 StandbyThreadPolicy::LAZY, 8U,
                                 TimeDelta::Max()),
@@ skipping 13 matching lines @@
                                 StandbyThreadPolicy::ONE, 8U, TimeDelta::Max()),
       Bind(&NotReachedReEnqueueSequenceCallback), &task_tracker,
       &delayed_task_manager);
   ASSERT_TRUE(worker_pool);
   EXPECT_EQ(1U, worker_pool->NumberOfAliveWorkersForTesting());
   worker_pool->JoinForTesting();
 }
 
 }  // namespace internal
 }  // namespace base