Separate the create and start phases in SchedulerWorkerPoolImpl.

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);
+    CreateAndStartWorkerPool(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) {
+  void CreateWorkerPool() {
     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),
+    worker_pool_ = MakeUnique<SchedulerWorkerPoolImpl>(
+        "TestWorkerPool", ThreadPriority::NORMAL,
         Bind(&TaskSchedulerWorkerPoolImplTest::ReEnqueueSequenceCallback,
              Unretained(this)),
         &task_tracker_, delayed_task_manager_.get());
     ASSERT_TRUE(worker_pool_);
   }
 
+  void StartWorkerPool(TimeDelta suggested_reclaim_time, size_t num_workers) {
+    ASSERT_TRUE(worker_pool_);
+    worker_pool_->Start(SchedulerWorkerPoolParams(
+        "TestWorkerPool", ThreadPriority::NORMAL, StandbyThreadPolicy::LAZY,
+        num_workers, suggested_reclaim_time));
+  }
+
+  void CreateAndStartWorkerPool(TimeDelta suggested_reclaim_time,
+                                size_t num_workers) {
+    CreateWorkerPool();
+    StartWorkerPool(suggested_reclaim_time, num_workers);
+  }
+
   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
@@ skipping 264 matching lines @@
 
 INSTANTIATE_TEST_CASE_P(Parallel,
                         TaskSchedulerWorkerPoolImplTest,
                         ::testing::Values(test::ExecutionMode::PARALLEL));
 INSTANTIATE_TEST_CASE_P(Sequenced,
                         TaskSchedulerWorkerPoolImplTest,
                         ::testing::Values(test::ExecutionMode::SEQUENCED));
 
 namespace {
 
+class TaskSchedulerWorkerPoolImplPostTaskBeforeStartTest

[gab, 2017/04/10 16:16:25]

Why even bother with the fixture? I'd favor inlining CreateWorkerPool() in test
below.

+    : public TaskSchedulerWorkerPoolImplTest {
+ public:
+  void SetUp() override {
+    CreateWorkerPool();
+    // Let the test start the worker pool.
+  }
+};
+
+void TaskPostedBeforeStart(PlatformThreadRef* platform_thread_ref,
+                           WaitableEvent* task_scheduled,
+                           WaitableEvent* barrier) {
+  *platform_thread_ref = PlatformThread::CurrentRef();
+  task_scheduled->Signal();
+  barrier->Wait();
+}
+
+}  // namespace
+
+// Verify that 2 tasks posted before Start() to a SchedulerWorkerPoolImpl with
+// more than 2 workers are scheduled on different workers when Start() is
+// called.
+TEST_F(TaskSchedulerWorkerPoolImplPostTaskBeforeStartTest,
+       PostTasksBeforeStart) {
+  PlatformThreadRef task_1_thread_ref;
+  PlatformThreadRef task_2_thread_ref;
+  WaitableEvent task_1_scheduled(WaitableEvent::ResetPolicy::MANUAL,
+                                 WaitableEvent::InitialState::NOT_SIGNALED);
+  WaitableEvent task_2_scheduled(WaitableEvent::ResetPolicy::MANUAL,
+                                 WaitableEvent::InitialState::NOT_SIGNALED);
+
+  // This event is used to prevent a task from completing before the other task
+  // is scheduled. If that happened, both tasks could run on the same worker and
+  // this test couldn't verify that the correct number of workers were woken up.
+  WaitableEvent barrier(WaitableEvent::ResetPolicy::MANUAL,
+                        WaitableEvent::InitialState::NOT_SIGNALED);
+
+  worker_pool_
+      ->CreateTaskRunnerWithTraits(TaskTraits().WithBaseSyncPrimitives())
+      ->PostTask(FROM_HERE,
+                 Bind(&TaskPostedBeforeStart, Unretained(&task_1_thread_ref),
+                      Unretained(&task_1_scheduled), Unretained(&barrier)));
+  worker_pool_
+      ->CreateTaskRunnerWithTraits(TaskTraits().WithBaseSyncPrimitives())
+      ->PostTask(FROM_HERE,
+                 Bind(&TaskPostedBeforeStart, Unretained(&task_2_thread_ref),
+                      Unretained(&task_2_scheduled), Unretained(&barrier)));
+
+  // Workers should not be created and tasks should not run before the pool is
+  // started.

[gab, 2017/04/10 19:09:07]

PlatformThread::Sleep(TestTimeouts::tiny_timeout());

(I thought this initially and then figured, meh... but now that you did it in
https://codereview.chromium.org/2806413002/, might as well be consistent :))

+  EXPECT_EQ(0U, worker_pool_->NumberOfAliveWorkersForTesting());
+  EXPECT_FALSE(task_1_scheduled.IsSignaled());
+  EXPECT_FALSE(task_2_scheduled.IsSignaled());
+
+  StartWorkerPool(TimeDelta::Max(), kNumWorkersInWorkerPool);
+
+  // Tasks should be scheduled shortly after the pool is started.
+  task_1_scheduled.Wait();
+  task_2_scheduled.Wait();
+
+  // Tasks should be scheduled on different threads.
+  EXPECT_NE(task_1_thread_ref, task_2_thread_ref);
+
+  barrier.Signal();
+  task_tracker_.Flush();
+}
+
+namespace {
+
 constexpr size_t kMagicTlsValue = 42;
 
 class TaskSchedulerWorkerPoolCheckTlsReuse
     : public TaskSchedulerWorkerPoolImplTest {
  public:
   void SetTlsValueAndWait() {
     slot_.Set(reinterpret_cast<void*>(kMagicTlsValue));
     waiter_.Wait();
   }
 
   void CountZeroTlsValuesAndWait(WaitableEvent* count_waiter) {
     if (!slot_.Get())
       subtle::NoBarrier_AtomicIncrement(&zero_tls_values_, 1);
 
     count_waiter->Signal();
     waiter_.Wait();
   }
 
  protected:
   TaskSchedulerWorkerPoolCheckTlsReuse() :
       waiter_(WaitableEvent::ResetPolicy::MANUAL,
               WaitableEvent::InitialState::NOT_SIGNALED) {}
 
   void SetUp() override {
-    InitializeWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
+    CreateAndStartWorkerPool(kReclaimTimeForDetachTests,
+                             kNumWorkersInWorkerPool);
   }
 
   subtle::Atomic32 zero_tls_values_ = 0;
 
   WaitableEvent waiter_;
 
  private:
   ThreadLocalStorage::Slot slot_;
 
   DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolCheckTlsReuse);
@@ skipping 71 matching lines @@
       StatisticsRecorder::CreateTemporaryForTesting();
 
   DISALLOW_COPY_AND_ASSIGN(TaskSchedulerWorkerPoolHistogramTest);
 };
 
 }  // namespace
 
 TEST_F(TaskSchedulerWorkerPoolHistogramTest, NumTasksBetweenWaits) {
   WaitableEvent event(WaitableEvent::ResetPolicy::MANUAL,
                       WaitableEvent::InitialState::NOT_SIGNALED);
-  InitializeWorkerPool(TimeDelta::Max(), kNumWorkersInWorkerPool);
+  CreateAndStartWorkerPool(TimeDelta::Max(), kNumWorkersInWorkerPool);
   auto task_runner = worker_pool_->CreateSequencedTaskRunnerWithTraits(
       TaskTraits().WithBaseSyncPrimitives());
 
   // Post a task.
   task_runner->PostTask(FROM_HERE,
                         Bind(&WaitableEvent::Wait, Unretained(&event)));
 
   // Post 2 more tasks while the first task hasn't completed its execution. It
   // is guaranteed that these tasks will run immediately after the first task,
   // without allowing the worker to sleep.
@@ skipping 23 matching lines @@
                         WaitableEvent* wait_event) {
   signal_event->Signal();
   wait_event->Wait();
 }
 
 }  // namespace
 
 TEST_F(TaskSchedulerWorkerPoolHistogramTest, NumTasksBetweenWaitsWithDetach) {
   WaitableEvent tasks_can_exit_event(WaitableEvent::ResetPolicy::MANUAL,
                                      WaitableEvent::InitialState::NOT_SIGNALED);
-  InitializeWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
+  CreateAndStartWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
   auto task_runner = worker_pool_->CreateTaskRunnerWithTraits(
       TaskTraits().WithBaseSyncPrimitives());
 
   // Post tasks to saturate the pool.
   std::vector<std::unique_ptr<WaitableEvent>> task_started_events;
   for (size_t i = 0; i < kNumWorkersInWorkerPool; ++i) {
     task_started_events.push_back(
         MakeUnique<WaitableEvent>(WaitableEvent::ResetPolicy::MANUAL,
                                   WaitableEvent::InitialState::NOT_SIGNALED));
     task_runner->PostTask(
@@ skipping 41 matching lines @@
   EXPECT_EQ(static_cast<int>(kNumWorkersInWorkerPool),
             histogram->SnapshotSamples()->GetCount(1));
   EXPECT_EQ(0, histogram->SnapshotSamples()->GetCount(10));
 
   tasks_can_exit_event.Signal();
   worker_pool_->WaitForAllWorkersIdleForTesting();
   worker_pool_->DisallowWorkerDetachmentForTesting();
 }
 
 TEST_F(TaskSchedulerWorkerPoolHistogramTest, NumTasksBeforeDetach) {
-  InitializeWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
+  CreateAndStartWorkerPool(kReclaimTimeForDetachTests, kNumWorkersInWorkerPool);
 
   auto histogrammed_thread_task_runner =
       worker_pool_->CreateSequencedTaskRunnerWithTraits(
           TaskTraits().WithBaseSyncPrimitives());
 
   // Post 3 tasks and hold the thread for idle thread stack ordering.
   // This test assumes |histogrammed_thread_task_runner| gets assigned the same
   // thread for each of its tasks.
   PlatformThreadRef thread_ref;
   histogrammed_thread_task_runner->PostTask(
@@ skipping 94 matching lines @@
   ADD_FAILURE()
       << "Unexpected invocation of NotReachedReEnqueueSequenceCallback.";
 }
 
 }  // namespace
 
 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()),
+  auto worker_pool = MakeUnique<SchedulerWorkerPoolImpl>(
+      "LazyPolicyWorkerPool", ThreadPriority::NORMAL,
       Bind(&NotReachedReEnqueueSequenceCallback), &task_tracker,
       &delayed_task_manager);
+  worker_pool->Start(SchedulerWorkerPoolParams(StandbyThreadPolicy::LAZY, 8U,
+                                               TimeDelta::Max()));
   ASSERT_TRUE(worker_pool);
   EXPECT_EQ(0U, worker_pool->NumberOfAliveWorkersForTesting());
   worker_pool->JoinForTesting();
 }
 
 TEST(TaskSchedulerWorkerPoolStandbyPolicyTest, InitOne) {
   TaskTracker task_tracker;
   DelayedTaskManager delayed_task_manager(
       make_scoped_refptr(new TestSimpleTaskRunner));
-  auto worker_pool = SchedulerWorkerPoolImpl::Create(
-      SchedulerWorkerPoolParams("LazyPolicyWorkerPool", ThreadPriority::NORMAL,
-                                StandbyThreadPolicy::ONE, 8U, TimeDelta::Max()),
+  auto worker_pool = MakeUnique<SchedulerWorkerPoolImpl>(
+      "OnePolicyWorkerPool", ThreadPriority::NORMAL,
       Bind(&NotReachedReEnqueueSequenceCallback), &task_tracker,
       &delayed_task_manager);
+  worker_pool->Start(SchedulerWorkerPoolParams(StandbyThreadPolicy::ONE, 8U,
+                                               TimeDelta::Max()));
   ASSERT_TRUE(worker_pool);
   EXPECT_EQ(1U, worker_pool->NumberOfAliveWorkersForTesting());
   worker_pool->JoinForTesting();
 }
 
 }  // namespace internal
 }  // namespace base