Separate the create and start phases in SchedulerWorkerPoolImpl.

base/task_scheduler/scheduler_worker_pool_impl.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 <algorithm>
 #include <utility>
@@ skipping 174 matching lines @@
 
   // Number of tasks executed since the last time the
   // TaskScheduler.NumTasksBeforeDetach histogram was recorded.
   size_t num_tasks_since_last_detach_ = 0;
 
   const int index_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerDelegateImpl);
 };
 
+SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
+    const std::string& name,
+    ThreadPriority priority_hint,
+    ReEnqueueSequenceCallback re_enqueue_sequence_callback,
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager)
+    : name_(name),
+      priority_hint_(priority_hint),
+      re_enqueue_sequence_callback_(std::move(re_enqueue_sequence_callback)),
+      idle_workers_stack_lock_(shared_priority_queue_.container_lock()),
+      idle_workers_stack_cv_for_testing_(
+          idle_workers_stack_lock_.CreateConditionVariable()),
+      join_for_testing_returned_(WaitableEvent::ResetPolicy::MANUAL,
+                                 WaitableEvent::InitialState::NOT_SIGNALED),
+#if DCHECK_IS_ON()
+      workers_created_(WaitableEvent::ResetPolicy::MANUAL,
+                       WaitableEvent::InitialState::NOT_SIGNALED),
+#endif
+      // Mimics the UMA_HISTOGRAM_LONG_TIMES macro.
+      detach_duration_histogram_(Histogram::FactoryTimeGet(
+          kDetachDurationHistogramPrefix + name_ + kPoolNameSuffix,
+          TimeDelta::FromMilliseconds(1),
+          TimeDelta::FromHours(1),
+          50,
+          HistogramBase::kUmaTargetedHistogramFlag)),
+      // Mimics the UMA_HISTOGRAM_COUNTS_1000 macro. When a worker runs more
+      // than 1000 tasks before detaching, there is no need to know the exact
+      // number of tasks that ran.
+      num_tasks_before_detach_histogram_(Histogram::FactoryGet(
+          kNumTasksBeforeDetachHistogramPrefix + name_ + kPoolNameSuffix,
+          1,
+          1000,
+          50,
+          HistogramBase::kUmaTargetedHistogramFlag)),
+      // Mimics the UMA_HISTOGRAM_COUNTS_100 macro. A SchedulerWorker is
+      // expected to run between zero and a few tens of tasks between waits.
+      // When it runs more than 100 tasks, there is no need to know the exact
+      // number of tasks that ran.
+      num_tasks_between_waits_histogram_(Histogram::FactoryGet(
+          kNumTasksBetweenWaitsHistogramPrefix + name_ + kPoolNameSuffix,
+          1,
+          100,
+          50,
+          HistogramBase::kUmaTargetedHistogramFlag)),
+      task_tracker_(task_tracker),
+      delayed_task_manager_(delayed_task_manager) {
+  DCHECK(task_tracker_);
+  DCHECK(delayed_task_manager_);
+}
+
+void SchedulerWorkerPoolImpl::Start(const SchedulerWorkerPoolParams& params) {
+  suggested_reclaim_time_ = params.suggested_reclaim_time();
+
+  std::vector<SchedulerWorker*> workers_to_wake_up;
+
+  {
+    // Get the number of enqueued Sequences to know how many workers to wake up
+    // at the end of this method. Keep the transaction alive to prevent new
+    // Sequences from being enqueued in the meantime.
+    auto priority_queue_transaction = shared_priority_queue_.BeginTransaction();
+    const size_t num_workers_to_wake_up = priority_queue_transaction->Size();
+
+    // The number of workers created alive is |num_workers_to_wake_up|, plus one
+    // if the standby thread policy is ONE.
+    const size_t num_alive_workers =

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

Can do this computation at the end to avoid holding the transaction during the
entire thread creation phase?

Our failure guarantee is at least one thread was created successfully. The
WakeUp at the end will create any additional ones if we can.

It would be nice to use the counts of calls to WakeUpOneWorker to get the same
effect here, but I can't convince myself that this is clearly better than
checking the size.

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

PQ lock is a predecessor of |idle_workers_stack_lock_|. If I wanted to get the
number of workers to wake up from the PQ after creating threads, I would first
need to release |idle_workers_stack_lock_|. But as soon as I do that, tasks can
be posted and wake up workers (i.e. Start() should not wake up workers for these
tasks).

// PQ contains 2 sequences.

{
  AutoSchedulerLock auto_lock(idle_workers_stack_lock_);

  // Create workers
  // ...
}

// Another thread posts a tasks (adds a sequence to the
// PQ and wakes up a worker).

size_t num_workers_to_wake_up =
  shared_priority_queue_.BeginTransaction()->Size();

// |num_workers_to_wake_up| is 3 but should be 2.

A good solution to avoid keeping a PQ transaction while workers are created
would be to have a |num_wake_up_before_start_| member.

+        num_workers_to_wake_up +
+        (params.standby_thread_policy() ==
+                 SchedulerWorkerPoolParams::StandbyThreadPolicy::ONE
+             ? 1
+             : 0);
+
+    AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
+
+#if DCHECK_IS_ON()
+    DCHECK(!workers_created_.IsSignaled());
+#endif
+
+    DCHECK(workers_.empty());
+    workers_.resize(params.max_threads());
+
+    // Create workers in reverse order of index so that the worker with the
+    // highest index is at the bottom of the idle stack.
+    for (int index = params.max_threads() - 1; index >= 0; --index) {
+      const SchedulerWorker::InitialState initial_state =
+          static_cast<size_t>(index) < num_alive_workers
+              ? SchedulerWorker::InitialState::ALIVE
+              : SchedulerWorker::InitialState::DETACHED;
+      scoped_refptr<SchedulerWorker> worker = SchedulerWorker::Create(
+          params.priority_hint(),
+          MakeUnique<SchedulerWorkerDelegateImpl>(
+              this, re_enqueue_sequence_callback_, index),
+          task_tracker_, initial_state, params.backward_compatibility());
+      if (!worker)
+        break;
+
+      if (static_cast<size_t>(index) < num_workers_to_wake_up)
+        workers_to_wake_up.push_back(worker.get());
+      else
+        idle_workers_stack_.Push(worker.get());
+
+      workers_[index] = std::move(worker);
+    }
+
+#if DCHECK_IS_ON()
+    workers_created_.Signal();
+#endif
+
+    CHECK(!workers_.empty());
+  }
+
+  for (SchedulerWorker* worker : workers_to_wake_up)
+    worker->WakeUp();
+}
+
 SchedulerWorkerPoolImpl::~SchedulerWorkerPoolImpl() {
   // SchedulerWorkerPool should never be deleted in production unless its
   // initialization failed.
   DCHECK(join_for_testing_returned_.IsSignaled() || workers_.empty());
 }
 
-// static
-std::unique_ptr<SchedulerWorkerPoolImpl> SchedulerWorkerPoolImpl::Create(
-    const SchedulerWorkerPoolParams& params,
-    const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
-    TaskTracker* task_tracker,
-    DelayedTaskManager* delayed_task_manager) {
-  auto worker_pool = WrapUnique(
-      new SchedulerWorkerPoolImpl(params, task_tracker, delayed_task_manager));
-  if (worker_pool->Initialize(params, re_enqueue_sequence_callback))
-    return worker_pool;
-  return nullptr;
-}
-
 scoped_refptr<TaskRunner> SchedulerWorkerPoolImpl::CreateTaskRunnerWithTraits(
     const TaskTraits& traits) {
   return make_scoped_refptr(new SchedulerParallelTaskRunner(traits, this));
 }
 
 scoped_refptr<SequencedTaskRunner>
 SchedulerWorkerPoolImpl::CreateSequencedTaskRunnerWithTraits(
     const TaskTraits& traits) {
   return make_scoped_refptr(new SchedulerSequencedTaskRunner(traits, this));
 }
@@ skipping 69 matching lines @@
   }
 }
 
 void SchedulerWorkerPoolImpl::GetHistograms(
     std::vector<const HistogramBase*>* histograms) const {
   histograms->push_back(detach_duration_histogram_);
   histograms->push_back(num_tasks_between_waits_histogram_);
 }
 
 int SchedulerWorkerPoolImpl::GetMaxConcurrentTasksDeprecated() const {
+#if DCHECK_IS_ON()
+  DCHECK(workers_created_.IsSignaled());
+#endif
   return workers_.size();
 }
 
 void SchedulerWorkerPoolImpl::WaitForAllWorkersIdleForTesting() {
+#if DCHECK_IS_ON()
+  DCHECK(workers_created_.IsSignaled());
+#endif
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
   while (idle_workers_stack_.Size() < workers_.size())
     idle_workers_stack_cv_for_testing_->Wait();
 }
 
 void SchedulerWorkerPoolImpl::JoinForTesting() {
+#if DCHECK_IS_ON()
+  DCHECK(workers_created_.IsSignaled());
+#endif
   DCHECK(!CanWorkerDetachForTesting() || suggested_reclaim_time_.is_max())
       << "Workers can detach during join.";
   for (const auto& worker : workers_)
     worker->JoinForTesting();
 
   DCHECK(!join_for_testing_returned_.IsSignaled());
   join_for_testing_returned_.Signal();
 }
 
 void SchedulerWorkerPoolImpl::DisallowWorkerDetachmentForTesting() {
   worker_detachment_disallowed_.Set();
 }
 
 size_t SchedulerWorkerPoolImpl::NumberOfAliveWorkersForTesting() {
+#if DCHECK_IS_ON()
+  DCHECK(workers_created_.IsSignaled());
+#endif
   size_t num_alive_workers = 0;
   for (const auto& worker : workers_) {
     if (worker->ThreadAliveForTesting())
       ++num_alive_workers;
   }
   return num_alive_workers;
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     SchedulerWorkerDelegateImpl(
@@ skipping 121 matching lines @@
 }
 
 void SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::OnDetach() {
   DCHECK(!did_detach_since_last_get_work_);
   outer_->num_tasks_before_detach_histogram_->Add(num_tasks_since_last_detach_);
   num_tasks_since_last_detach_ = 0;
   did_detach_since_last_get_work_ = true;
   last_detach_time_ = TimeTicks::Now();
 }
 
-SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
-    const SchedulerWorkerPoolParams& params,
-    TaskTracker* task_tracker,
-    DelayedTaskManager* delayed_task_manager)
-    : name_(params.name()),
-      suggested_reclaim_time_(params.suggested_reclaim_time()),
-      idle_workers_stack_lock_(shared_priority_queue_.container_lock()),
-      idle_workers_stack_cv_for_testing_(
-          idle_workers_stack_lock_.CreateConditionVariable()),
-      join_for_testing_returned_(WaitableEvent::ResetPolicy::MANUAL,
-                                 WaitableEvent::InitialState::NOT_SIGNALED),
-#if DCHECK_IS_ON()
-      workers_created_(WaitableEvent::ResetPolicy::MANUAL,
-                       WaitableEvent::InitialState::NOT_SIGNALED),
-#endif
-      // Mimics the UMA_HISTOGRAM_LONG_TIMES macro.
-      detach_duration_histogram_(Histogram::FactoryTimeGet(
-          kDetachDurationHistogramPrefix + name_ + kPoolNameSuffix,
-          TimeDelta::FromMilliseconds(1),
-          TimeDelta::FromHours(1),
-          50,
-          HistogramBase::kUmaTargetedHistogramFlag)),
-      // Mimics the UMA_HISTOGRAM_COUNTS_1000 macro. When a worker runs more
-      // than 1000 tasks before detaching, there is no need to know the exact
-      // number of tasks that ran.
-      num_tasks_before_detach_histogram_(Histogram::FactoryGet(
-          kNumTasksBeforeDetachHistogramPrefix + name_ + kPoolNameSuffix,
-          1,
-          1000,
-          50,
-          HistogramBase::kUmaTargetedHistogramFlag)),
-      // Mimics the UMA_HISTOGRAM_COUNTS_100 macro. A SchedulerWorker is
-      // expected to run between zero and a few tens of tasks between waits.
-      // When it runs more than 100 tasks, there is no need to know the exact
-      // number of tasks that ran.
-      num_tasks_between_waits_histogram_(Histogram::FactoryGet(
-          kNumTasksBetweenWaitsHistogramPrefix + name_ + kPoolNameSuffix,
-          1,
-          100,
-          50,
-          HistogramBase::kUmaTargetedHistogramFlag)),
-      task_tracker_(task_tracker),
-      delayed_task_manager_(delayed_task_manager) {
-  DCHECK(task_tracker_);
-  DCHECK(delayed_task_manager_);
-}
-
-bool SchedulerWorkerPoolImpl::Initialize(
-    const SchedulerWorkerPoolParams& params,
-    const ReEnqueueSequenceCallback& re_enqueue_sequence_callback) {
-  AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
-
-  DCHECK(workers_.empty());
-  workers_.resize(params.max_threads());
-
-  // Create workers and push them to the idle stack in reverse order of index.
-  // This ensures that they are woken up in order of index and that the ALIVE
-  // worker is on top of the stack.
-  for (int index = params.max_threads() - 1; index >= 0; --index) {
-    const bool is_standby_lazy =
-        params.standby_thread_policy() ==
-        SchedulerWorkerPoolParams::StandbyThreadPolicy::LAZY;
-    const SchedulerWorker::InitialState initial_state =
-        (index == 0 && !is_standby_lazy)
-            ? SchedulerWorker::InitialState::ALIVE
-            : SchedulerWorker::InitialState::DETACHED;
-    scoped_refptr<SchedulerWorker> worker = SchedulerWorker::Create(
-        params.priority_hint(),
-        MakeUnique<SchedulerWorkerDelegateImpl>(
-            this, re_enqueue_sequence_callback, index),
-        task_tracker_, initial_state, params.backward_compatibility());
-    if (!worker)
-      break;
-    idle_workers_stack_.Push(worker.get());
-    workers_[index] = std::move(worker);
-  }
-
-#if DCHECK_IS_ON()
-  workers_created_.Signal();
-#endif
-
-  return !workers_.empty();
-}
-
 void SchedulerWorkerPoolImpl::WakeUpOneWorker() {
   SchedulerWorker* worker;
   {
     AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
     worker = idle_workers_stack_.Pop();
   }
   if (worker)
     worker->WakeUp();
   // TODO(robliao): Honor StandbyThreadPolicy::ONE here and consider adding
   // hysteresis to the CanDetach check. See https://crbug.com/666041.
@@ skipping 25 matching lines @@
   AutoSchedulerLock auto_lock(idle_workers_stack_lock_);
   idle_workers_stack_.Remove(worker);
 }
 
 bool SchedulerWorkerPoolImpl::CanWorkerDetachForTesting() {
   return !worker_detachment_disallowed_.IsSet();
 }
 
 }  // namespace internal
 }  // namespace base