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 263 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

[robliao, 2017/04/05 23:29:23]

Nit: Linebreak above here.

[fdoray, 2017/04/06 15:07:12]

Done.

+    : 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)));
+
+  // Tasks should not be scheduled before the pool is started.
+  EXPECT_FALSE(task_1_scheduled.IsSignaled());

[robliao, 2017/04/05 23:29:23]

There is an inherent observibility problem here. If the worker pool did create
the threads and was slow to dispatch the the tasks, these events would still not
be signaled.


It might be more reliable to check the number of alive workers as those are done
synchronously. 

[fdoray, 2017/04/06 15:07:12]

Done.

Added NumberOfAliveWorkersForTesting() check. Kept event checks (theoretically,
PostTask() could decide to run tasks synchronously when there are no
workers...).

+  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