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 16 matching lines @@
 
  private:
   ~SchedulerSingleThreadTaskRunner() override;
 
   // Sequence for all Tasks posted through this TaskRunner.
   const scoped_refptr<Sequence> sequence_ = new Sequence;
 
   const TaskTraits traits_;
   SchedulerWorkerPool* const worker_pool_;
   SchedulerWorker* const worker_;
+  const UnregisterSingleThreadWorkerPoolCallback

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

Not needed. SchedulerSingleThreadTaskRunner can access
worker_pool_->unregister_single_thread_worker_pool_callback_ directly.

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

Indeed! Done.

+      unregister_single_thread_worker_pool_callback_;
 
   DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
 };
 
 class SchedulerWorkerPoolImpl::SchedulerWorkerDelegateImpl
     : public SchedulerWorker::Delegate {
  public:
   // |outer| owns the worker for which this delegate is constructed.
   // |re_enqueue_sequence_callback| is invoked when ReEnqueueSequence() is
   // called with a non-single-threaded Sequence. |shared_priority_queue| is a
@@ skipping 82 matching lines @@
     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;
 }
 
+// static
+std::unique_ptr<SchedulerWorkerPoolImpl>
+SchedulerWorkerPoolImpl::CreateSingleThreadWorkerPool(

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

Forward Create() and CreateSingleThreadWorkerPool() to CreateInternal()?

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

I wasn't convinced that it would make things clearer since CreateInternal would
have to take the union of all arguments, but I've done it here as an experiment.
It does make one path for construction, which is nice, but at the same time,
construction is already simple.

+    const SchedulerWorkerPoolParams& params,
+    UnregisterSingleThreadWorkerPoolCallback
+        unregister_single_thread_worker_pool_callback,
+    TaskTracker* task_tracker,
+    DelayedTaskManager* delayed_task_manager) {
+  DCHECK_EQ(params.max_threads(), 1U);
+  auto worker_pool = WrapUnique(new SchedulerWorkerPoolImpl(
+      params, task_tracker, delayed_task_manager,
+      unregister_single_thread_worker_pool_callback));
+  if (worker_pool->Initialize(params, ReEnqueueSequenceCallback())) {

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

no braces

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

Done.

+    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));
 }
 
 scoped_refptr<SingleThreadTaskRunner>
 SchedulerWorkerPoolImpl::CreateSingleThreadTaskRunnerWithTraits(
     const TaskTraits& traits) {
+  DCHECK(!unregister_single_thread_worker_pool_callback_.is_null());

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

Tip: You can write
DCHECK(unregister_single_thread_worker_pool_callback_);
https://cs.chromium.org/chromium/src/base/callback_internal.h?l=87

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

Nice! Done.

   // 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)
+    SchedulerSingleThreadTaskRunner(
+        const TaskTraits& traits,
+        SchedulerWorkerPool* worker_pool,
+        SchedulerWorker* worker,
+        UnregisterSingleThreadWorkerPoolCallback
+            unregister_single_thread_worker_pool_callback)
     : traits_(traits),
       worker_pool_(worker_pool),
-      worker_(worker) {
+      worker_(worker),
+      unregister_single_thread_worker_pool_callback_(
+          unregister_single_thread_worker_pool_callback) {
   DCHECK(worker_pool_);
   DCHECK(worker_);
+  DCHECK(!unregister_single_thread_worker_pool_callback.is_null());
   static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
       RegisterSingleThreadTaskRunner();
 }
 
 SchedulerWorkerPoolImpl::SchedulerSingleThreadTaskRunner::
     ~SchedulerSingleThreadTaskRunner() {
   static_cast<SchedulerWorkerDelegateImpl*>(worker_->delegate())->
       UnregisterSingleThreadTaskRunner();
+  unregister_single_thread_worker_pool_callback_.Run(
+      static_cast<SchedulerWorkerPoolImpl*>(worker_pool_));
 }
 
 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() &&
+      (worker != outer_->PeekAtIdleWorkersStack() ||
+       outer_->workers_.size() == 1) &&  // Single Thread Worker Pools

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

Should be

...
// 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() ||
 (unregister_single_thread_worker_pool_callback_ &&
  !subtle::NoBarrier_Load(&num_single_threaded_runners_))) &&
outer_->CanWorkerDetachForTesting();

Note: (outer_->workers_.size() == 1) is true in pools with a single thread that
runs parallel/sequenced tasks
https://cs.chromium.org/chromium/src/content/renderer/render_process_impl.cc?...
I think we want to keep the thread active in these pools.

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

Good point. Done.

       !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();
 }
 
 SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
     const SchedulerWorkerPoolParams& params,
     TaskTracker* task_tracker,
     DelayedTaskManager* delayed_task_manager)
+    : SchedulerWorkerPoolImpl(params,
+                              task_tracker,
+                              delayed_task_manager,
+                              UnregisterSingleThreadWorkerPoolCallback()) {}
+
+SchedulerWorkerPoolImpl::SchedulerWorkerPoolImpl(
+    const SchedulerWorkerPoolParams& params,
+    TaskTracker* task_tracker,
+    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