TaskScheduler [12] Support SINGLE_THREADED in SchedulerThreadPool DO NOT SUBMIT

base/task_scheduler/scheduler_thread_pool.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_thread_pool.h"
 
 #include <utility>
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
@@ skipping 156 matching lines @@
     ExecutionMode execution_mode) {
   switch (execution_mode) {
     case ExecutionMode::PARALLEL:
       return make_scoped_refptr(new SchedulerParallelTaskRunner(
           traits, this, task_tracker_, delayed_task_manager_));
 
     case ExecutionMode::SEQUENCED:
       return make_scoped_refptr(new SchedulerSequencedTaskRunner(
           traits, this, task_tracker_, delayed_task_manager_));
 
-    case ExecutionMode::SINGLE_THREADED:
-      // TODO(fdoray): Support SINGLE_THREADED TaskRunners.
-      NOTREACHED();
-      return nullptr;
+    case ExecutionMode::SINGLE_THREADED: {
+      // Acquire a lock to ensure that the number of single-thread TaskRunners
+      // for a given SchedulerWorkerThread doesn't increase between the calls to
+      // GetNumSingleThreadTaskRunners() and CreateTaskRunnerWithTraits() below.
+      // This number can still decrease.
+      AutoSchedulerLock auto_lock(single_thread_task_runner_creation_lock_);
+
+      SchedulerWorkerThread* worker_thread = nullptr;
+      size_t min_num_single_thread_task_runners = -1;
+      for (const auto& current_worker_thread : worker_threads_) {
+        const size_t current_num_single_thread_task_runners =
+            current_worker_thread->GetNumSingleThreadTaskRunners();
+        if (current_num_single_thread_task_runners <
+            min_num_single_thread_task_runners) {
+          worker_thread = current_worker_thread.get();
+          min_num_single_thread_task_runners =
+              current_num_single_thread_task_runners;
+          if (min_num_single_thread_task_runners == 0)
+            break;
+        }
+      }
+
+      // |worker_thread| is the SchedulerWorkerThread with the fewer single-
+      // thread TaskRunners.
+      DCHECK(worker_thread);
+      return worker_thread->CreateTaskRunnerWithTraits(traits);
+    }
   }
 
   NOTREACHED();
   return nullptr;
 }
 
 void SchedulerThreadPool::EnqueueSequence(
     scoped_refptr<Sequence> sequence,
     const SequenceSortKey& sequence_sort_key) {
   shared_priority_queue_.BeginTransaction()->Push(
       WrapUnique(new PriorityQueue::SequenceAndSortKey(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
   // belongs to this pool, it will get that Sequence from
   // |shared_priority_queue_|. When that's the case, there is no need to wake up
   // another thread after |sequence| is inserted in |shared_priority_queue_|. If
   // we did wake up another thread, we would waste resources by having more
   // threads trying to get a Sequence from |shared_priority_queue_| than the
   // number of Sequences in it.
   if (tls_current_thread_pool.Get().Get() != this)
     WakeUpOneThread();
 }
 
 void SchedulerThreadPool::WaitForAllWorkerThreadsIdleForTesting() {
   AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
-  while (idle_worker_threads_stack_.size() < worker_threads_.size())
+  while (idle_worker_threads_stack_.Size() < worker_threads_.size())
     idle_worker_threads_stack_cv_for_testing_->Wait();
 }
 
 void SchedulerThreadPool::JoinForTesting() {
   for (const auto& worker_thread : worker_threads_)
     worker_thread->JoinForTesting();
 
   DCHECK(!join_for_testing_returned_.IsSignaled());
   join_for_testing_returned_.Signal();
 }
@@ skipping 25 matching lines @@
 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::OnMainEntry() {
   DCHECK(!tls_current_thread_pool.Get().Get());
   tls_current_thread_pool.Get().Set(outer_);
 }
 
 scoped_refptr<Sequence>
 SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::GetWork(
     SchedulerWorkerThread* worker_thread,
     PriorityQueue* alternate_priority_queue,
     bool* alternate_priority_queue_used) {
-  // TODO(fdoray): Return a Sequence from |alternate_priority_queue| when
-  // appropriate.
+  DCHECK(worker_thread);
+  DCHECK(alternate_priority_queue);
+  DCHECK(alternate_priority_queue_used);
 
-  std::unique_ptr<PriorityQueue::Transaction> transaction(
+  *is_single_threaded_sequence = false;
+
+  std::unique_ptr<PriorityQueue::Transaction> shared_transaction(
       outer_->shared_priority_queue_.BeginTransaction());
-  const auto sequence_and_sort_key = transaction->Peek();
+  const auto shared_sequence_and_sort_key = shared_transaction->Peek();
 
-  if (sequence_and_sort_key.is_null()) {
-    // |transaction| is kept alive while |worker_thread| is added to
+  std::unique_ptr<PriorityQueue::Transaction> alternate_transaction(
+      alternate_priority_queue->BeginTransaction());
+  const auto alternate_sequence_and_sort_key =
+      alternate_transaction->Peek();
+
+  if (shared_sequence_and_sort_key.is_null() &&
+      alternate_sequence_and_sort_key.is_null()) {
+    alternate_transaction.reset();
+
+    // |shared_transaction| is kept alive while |worker_thread| is added to
     // |idle_worker_threads_stack_| to avoid this race:
     // 1. This thread creates a Transaction, finds |shared_priority_queue_|
     //    empty and ends the Transaction.
     // 2. Other thread creates a Transaction, inserts a Sequence into
     //    |shared_priority_queue_| and ends the Transaction. This can't happen
     //    if the Transaction of step 1 is still active because because there can
     //    only be one active Transaction per PriorityQueue at a time.
     // 3. Other thread calls WakeUpOneThread(). No thread is woken up because
     //    |idle_worker_threads_stack_| is empty.
     // 4. This thread adds itself to |idle_worker_threads_stack_| and goes to
     //    sleep. No thread runs the Sequence inserted in step 2.
     outer_->AddToIdleWorkerThreadsStack(worker_thread);
     return nullptr;
   }
 
-  transaction->Pop();
-  return sequence_and_sort_key.sequence;
+  scoped_refptr<Sequence> sequence;
+
+  if (alternate_sequence_and_sort_key.is_null() ||
+      (!shared_sequence_and_sort_key.is_null() &&
+       alternate_sequence_and_sort_key.sort_key <
+           shared_sequence_and_sort_key.sort_key)) {
+    shared_transaction->Pop();
+    sequence = std::move(shared_sequence_and_sort_key.sequence);
+  } else {
+    DCHECK(!alternate_sequence_and_sort_key.is_null());
+    alternate_transaction->Pop();
+    sequence = std::move(alternate_sequence_and_sort_key.sequence);
+    *alternate_priority_queue_used = true;
+  }
+
+  shared_transaction.reset();
+  alternate_transaction.reset();
+  outer_->RemoveFromIdleWorkerThreadsStack(worker_thread);
+
+  return sequence;
 }
 
 void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::EnqueueSequence(
     scoped_refptr<Sequence> sequence) {
   enqueue_sequence_callback_.Run(std::move(sequence));
 }
 
 SchedulerThreadPool::SchedulerThreadPool(
     const EnqueueSequenceCallback& enqueue_sequence_callback,
     TaskTracker* task_tracker,
@@ skipping 17 matching lines @@
 
   DCHECK(worker_threads_.empty());
 
   for (size_t i = 0; i < max_threads; ++i) {
     std::unique_ptr<SchedulerWorkerThread> worker_thread =
         SchedulerWorkerThread::CreateSchedulerWorkerThread(
             thread_priority, worker_thread_delegate_.get(), task_tracker_,
             delayed_task_manager_, &shared_priority_queue_);
     if (!worker_thread)
       break;
-    idle_worker_threads_stack_.push(worker_thread.get());
+    idle_worker_threads_stack_.Push(worker_thread.get());
     worker_threads_.push_back(std::move(worker_thread));
   }
 
   return !worker_threads_.empty();
 }
 
 void SchedulerThreadPool::WakeUpOneThread() {
-  SchedulerWorkerThread* worker_thread = PopOneIdleWorkerThread();
+  SchedulerWorkerThread* worker_thread = nullptr;
+  {
+    AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+    if (!idle_worker_threads_stack_.Empty())
+      worker_thread = idle_worker_threads_stack_.Pop();
+  }
+
   if (worker_thread)
     worker_thread->WakeUp();
 }
 
 void SchedulerThreadPool::AddToIdleWorkerThreadsStack(
     SchedulerWorkerThread* worker_thread) {
   AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
-  idle_worker_threads_stack_.push(worker_thread);
-  DCHECK_LE(idle_worker_threads_stack_.size(), worker_threads_.size());
+  idle_worker_threads_stack_.Push(worker_thread);
+  DCHECK_LE(idle_worker_threads_stack_.Size(), worker_threads_.size());
 
-  if (idle_worker_threads_stack_.size() == worker_threads_.size())
+  if (idle_worker_threads_stack_.Size() == worker_threads_.size())
     idle_worker_threads_stack_cv_for_testing_->Broadcast();
 }
 
-SchedulerWorkerThread* SchedulerThreadPool::PopOneIdleWorkerThread() {
+void SchedulerThreadPool::RemoveFromIdleWorkerThreadsStack(
+    SchedulerWorkerThread* worker_thread) {
   AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
-
-  if (idle_worker_threads_stack_.empty())
-    return nullptr;
-
-  auto worker_thread = idle_worker_threads_stack_.top();
-  idle_worker_threads_stack_.pop();
-  return worker_thread;
+  idle_worker_threads_stack_.Remove(worker_thread);
 }
 
 }  // namespace internal
 }  // namespace base