Move Task Scheduler Single Thread Task Runners to Dedicated Threads

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 132 matching lines @@
 }  // namespace
 
 // A task runner that runs tasks with the SINGLE_THREADED ExecutionMode.
 class SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner :
     public SingleThreadTaskRunner {
  public:
   // Constructs a SchedulerSingleThreadTaskRunner which can be used to post
   // tasks so long as |worker_pool| and |worker| are alive.
   // TODO(robliao): Find a concrete way to manage the memory of |worker_pool|
   // and |worker|.
-  SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
-                                  SchedulerWorkerPool* worker_pool,
-                                  SchedulerWorker* worker);
+  SchedulerSingleThreadTaskRunner(
+      const TaskTraits& traits,
+      SchedulerWorkerPool* worker_pool,
+      SchedulerWorker* worker,
+      UnregisterSingleThreadWorkerPoolCallback
+          unregister_single_thread_worker_pool_callback);
 
   // SingleThreadTaskRunner:
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const Closure& closure,
                        TimeDelta delay) override {
     std::unique_ptr<Task> task(new Task(from_here, closure, traits_, delay));
     task->single_thread_task_runner_ref = this;
 
     // Post the task to be executed by |worker_| as part of |sequence_|.
     return worker_pool_->PostTaskWithSequence(std::move(task), sequence_,
@@ skipping 111 matching lines @@
   // 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;
+  return CreateInternal(params, task_tracker, delayed_task_manager,
+                        re_enqueue_sequence_callback,
+                        UnregisterSingleThreadWorkerPoolCallback());
+}
+
+// static
+std::unique_ptr<SchedulerWorkerPoolImpl>
+SchedulerWorkerPoolImpl::CreateSingleThreadWorkerPool(
+    const SchedulerWorkerPoolParams& params,
+    UnregisterSingleThreadWorkerPoolCallback
+        unregister_single_thread_worker_pool_callback,
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager) {
+  DCHECK_EQ(params.max_threads(), 1U);
+  return CreateInternal(params, task_tracker, delayed_task_manager,
+                        ReEnqueueSequenceCallback(),
+                        unregister_single_thread_worker_pool_callback);
 }
 
 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));
 }
 
 scoped_refptr<SingleThreadTaskRunner>
 SchedulerWorkerPoolImpl::CreateSingleThreadTaskRunnerWithTraits(
     const TaskTraits& traits) {
+  DCHECK(unregister_single_thread_worker_pool_callback_);
   // TODO(fdoray): Find a way to take load into account when assigning a
   // SchedulerWorker to a SingleThreadTaskRunner.
   size_t worker_index;
   {
     AutoSchedulerLock auto_lock(next_worker_index_lock_);
     worker_index = next_worker_index_;
     next_worker_index_ = (next_worker_index_ + 1) % workers_.size();
   }
   return make_scoped_refptr(new SchedulerSingleThreadTaskRunner(
-      traits, this, workers_[worker_index].get()));
+      traits, this, workers_[worker_index].get(),
+      unregister_single_thread_worker_pool_callback_));
 }
 
 void SchedulerWorkerPoolImpl::ReEnqueueSequence(
     scoped_refptr<Sequence> sequence,
     const SequenceSortKey& sequence_sort_key) {
   shared_priority_queue_.BeginTransaction()->Push(std::move(sequence),
                                                   sequence_sort_key);
 
   // The thread calling this method just ran a Task from |sequence| and will
   // soon try to get another Sequence from which to run a Task. If the thread
@@ skipping 98 matching lines @@
 size_t SchedulerWorkerPoolImpl::NumberOfAliveWorkersForTesting() {
   size_t num_alive_workers = 0;
   for (const auto& worker : workers_) {
     if (worker->ThreadAliveForTesting())
       ++num_alive_workers;
   }
   return num_alive_workers;
 }
 
 SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner::
-    SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
-                                    SchedulerWorkerPool* worker_pool,
-                                    SchedulerWorker* worker)
-    : traits_(traits),
-      worker_pool_(worker_pool),
-      worker_(worker) {
+    SchedulerSingleThreadTaskRunner(
+        const TaskTraits& traits,
+        SchedulerWorkerPool* worker_pool,
+        SchedulerWorker* worker,
+        UnregisterSingleThreadWorkerPoolCallback
+            unregister_single_thread_worker_pool_callback)
+    : traits_(traits), worker_pool_(worker_pool), worker_(worker) {
   DCHECK(worker_pool_);
   DCHECK(worker_);
   static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
       RegisterSingleThreadTaskRunner();
 }
 
 SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner::
     ~SchedulerSingleThreadTaskRunner() {
   static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
       UnregisterSingleThreadTaskRunner();
+  auto worker_pool_impl = static_cast<SchedulerWorkerPoolImpl*>(worker_pool_);
+  worker_pool_impl->unregister_single_thread_worker_pool_callback_.Run(
+      worker_pool_impl);
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::
     SchedulerWorkerDelegateImpl(
         SchedulerWorkerPoolImpl* outer,
         const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
         const PriorityQueue* shared_priority_queue,
         int index)
     : outer_(outer),
       re_enqueue_sequence_callback_(re_enqueue_sequence_callback),
@@ skipping 137 matching lines @@
 
 bool SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl::CanDetach(
     SchedulerWorker* worker) {
   // It's not an issue if |num_single_threaded_runners_| is incremented after
   // this because the newly created SingleThreadTaskRunner (from which no task
   // has run yet) will simply run all its tasks on the next physical thread
   // created by the worker.
   const bool can_detach =
       !idle_start_time_.is_null() &&
       (TimeTicks::Now() - idle_start_time_) > outer_->suggested_reclaim_time_ &&
-      worker != outer_->PeekAtIdleWorkersStack() &&
-      !subtle::NoBarrier_Load(&num_single_threaded_runners_) &&
+      // In a pool which runs sequenced and parallel tasks, always keep
+      // on idle thread alive. In a pool which runs single-threaded tasks,
+      // keep the thread alive only when there are active TaskRunners.
+      (worker != outer_->PeekAtIdleWorkersStack() ||
+       (outer_->unregister_single_thread_worker_pool_callback_ &&
+        !subtle::NoBarrier_Load(&num_single_threaded_runners_))) &&
+      outer_->CanWorkerDetachForTesting();
+  !subtle::NoBarrier_Load(&num_single_threaded_runners_) &&
       outer_->CanWorkerDetachForTesting();
   return can_detach;
 }
 
 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();
 }
 
+// static
+std::unique_ptr<SchedulerWorkerPoolImpl>
+SchedulerWorkerPoolImpl::CreateInternal(
+    const SchedulerWorkerPoolParams& params,
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager,
+    const ReEnqueueSequenceCallback& re_enqueue_sequence_callback,
+    UnregisterSingleThreadWorkerPoolCallback
+        unregister_single_thread_worker_callback) {
+  auto worker_pool = WrapUnique(
+      new SchedulerWorkerPoolImpl(params, task_tracker, delayed_task_manager,
+                                  unregister_single_thread_worker_callback));
+  if (worker_pool->Initialize(params, re_enqueue_sequence_callback))
+    return worker_pool;
+  return nullptr;
+}
+
 SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
     const SchedulerWorkerPoolParams& params,
     TaskTracker* task_tracker,
-    DelayedTaskManager* delayed_task_manager)
+    DelayedTaskManager* delayed_task_manager,
+    UnregisterSingleThreadWorkerPoolCallback
+        unregister_single_thread_worker_pool_callback)
     : 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),
@@ skipping 18 matching lines @@
       // 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) {
+      delayed_task_manager_(delayed_task_manager),
+      unregister_single_thread_worker_pool_callback_(
+          unregister_single_thread_worker_pool_callback) {
   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());
@@ skipping 72 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