TaskScheduler [7] SchedulerThreadPool

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"
+#include "base/lazy_instance.h"
+#include "base/logging.h"
+#include "base/memory/ptr_util.h"
+#include "base/task_scheduler/task_tracker.h"
+
+namespace base {
+namespace internal {
+
+namespace {
+
+// Shared PriorityQueue of a thread's SchedulerThreadPool. Not set for threads
+// that don't belong to a SchedulerThreadPool.
+LazyInstance<ThreadLocalPointer<const PriorityQueue>>::Leaky
+    g_current_shared_priority_queue = LAZY_INSTANCE_INITIALIZER;
+
+// A task runner that runs tasks with the PARALLEL ExecutionMode.
+class SchedulerParallelTaskRunner : public TaskRunner {
+ public:
+  SchedulerParallelTaskRunner(const TaskTraits& traits,
+                              PriorityQueue* priority_queue,
+                              TaskTracker* task_tracker)
+      : traits_(traits),
+        priority_queue_(priority_queue),
+        task_tracker_(task_tracker) {}
+
+  // TaskRunner:
+  bool PostDelayedTask(const tracked_objects::Location& from_here,
+                       const Closure& closure,
+                       TimeDelta delay) override {
+    // TODO(fdoray): Support delayed tasks.
+    DCHECK(delay.is_zero());
+    PostTaskHelper(WrapUnique(new Task(from_here, closure, traits_)),
+                   make_scoped_refptr(new Sequence), priority_queue_,
+                   task_tracker_);
+    return true;
+  }
+
+  bool RunsTasksOnCurrentThread() const override {
+    return g_current_shared_priority_queue.Get().Get() == priority_queue_;
+  }
+
+ private:
+  ~SchedulerParallelTaskRunner() override = default;
+
+  const TaskTraits traits_;
+  PriorityQueue* const priority_queue_;
+  TaskTracker* const task_tracker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner);
+};
+
+void PostTaskCallback(scoped_refptr<Sequence> sequence,
+                      PriorityQueue* priority_queue,
+                      std::unique_ptr<Task> task) {
+  DCHECK(sequence);
+  DCHECK(priority_queue);
+  DCHECK(task);
+
+  if (sequence->PushTask(std::move(task))) {
+    // |sequence| must be inserted into |priority_queue| because it was empty
+    // before |task| was inserted into it.
+    const SequenceSortKey sequence_sort_key = sequence->GetSortKey();
+    priority_queue->BeginTransaction()->Push(
+        WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence),
+                                                         sequence_sort_key)));
+  }
+}
+
+}  // namespace
+
+SchedulerThreadPool::~SchedulerThreadPool() {
+  AutoSchedulerLock auto_lock(join_for_testing_returned_lock_);
+  DCHECK(join_for_testing_returned_);
+}
+
+std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool(
+    ThreadPriority thread_priority,
+    size_t max_threads,
+    const SchedulerWorkerThread::RanTaskFromSequenceCallback&
+        ran_task_from_sequence_callback,
+    TaskTracker* task_tracker) {
+  std::unique_ptr<SchedulerThreadPool> thread_pool(
+      new SchedulerThreadPool(thread_priority, max_threads,
+                              ran_task_from_sequence_callback, task_tracker));
+
+  if (thread_pool->worker_threads_.empty())
+    return nullptr;
+  return thread_pool;
+}
+
+scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits(
+    const TaskTraits& traits,
+    ExecutionMode execution_mode) {
+  switch (execution_mode) {
+    case ExecutionMode::PARALLEL:
+      return make_scoped_refptr(new SchedulerParallelTaskRunner(
+          traits, &shared_priority_queue_, task_tracker_));
+
+    case ExecutionMode::SEQUENCED:
+    case ExecutionMode::SINGLE_THREADED:
+      NOTIMPLEMENTED();
+      return nullptr;
+  }
+
+  NOTREACHED();
+  return nullptr;
+}
+
+void SchedulerThreadPool::ReinsertSequence(
+    scoped_refptr<Sequence> sequence,
+    const SequenceSortKey& sequence_sort_key) {
+  DCHECK(!g_current_shared_priority_queue.Get().Get());
+
+  // If |worker_thread| belongs to this thread pool, set a flag to avoid waking

[robliao, 2016/04/01 21:05:51]

Update this comment.

[fdoray, 2016/04/01 21:45:30]

Done.

+  // up a SchedulerWorkerThread when |sequence| is reinserted in
+  // |shared_priority_queue_|. In such cases, |worker_thread| will soon get a
+  // Sequence from |shared_priority_queue_| via GetWorkCallback() and there is
+  // no need to wake up another SchedulerWorkerThread to do so.
+  if (g_current_shared_priority_queue.Get().Get() == &shared_priority_queue_)
+    no_wake_up_on_sequence_insertion_.Set(true);
+
+  shared_priority_queue_.BeginTransaction()->Push(
+      WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence),
+                                                       sequence_sort_key)));
+  no_wake_up_on_sequence_insertion_.Set(false);
+}
+
+void SchedulerThreadPool::WaitForAllWorkerThreadsIdleForTesting() {
+  AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+  while (idle_worker_threads_stack_.size() < worker_threads_.size())
+    idle_worker_threads_stack_cv_->Wait();
+}
+
+void SchedulerThreadPool::JoinForTesting() {
+  for (const auto& worker_thread : worker_threads_)
+    worker_thread->JoinForTesting();
+
+  AutoSchedulerLock auto_lock(join_for_testing_returned_lock_);
+  DCHECK(!join_for_testing_returned_);
+  join_for_testing_returned_ = true;
+}
+
+SchedulerThreadPool::SchedulerThreadPool(
+    ThreadPriority thread_priority,
+    size_t max_threads,
+    const SchedulerWorkerThread::RanTaskFromSequenceCallback&
+        ran_task_from_sequence_callback,
+    TaskTracker* task_tracker)
+    : shared_priority_queue_(Bind(
+          &SchedulerThreadPool::SequenceInsertedInSharedPriorityQueueCallback,
+          Unretained(this))),
+      idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()),
+      idle_worker_threads_stack_cv_(
+          idle_worker_threads_stack_lock_.CreateConditionVariable()),
+      task_tracker_(task_tracker) {
+  DCHECK_GT(max_threads, 0U);
+  DCHECK(!ran_task_from_sequence_callback.is_null());
+  DCHECK(task_tracker_);
+
+  // |this| always outlives the worker threads to which these callbacks are
+  // passed.
+  const Closure main_entry_callback(
+      Bind(&SchedulerThreadPool::MainEntryCallback, Unretained(this)));
+  const SchedulerWorkerThread::GetWorkCallback get_work_callback(
+      Bind(&SchedulerThreadPool::GetWorkCallback, Unretained(this)));
+
+  AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+
+  for (size_t i = 0; i < max_threads; ++i) {
+    std::unique_ptr<SchedulerWorkerThread> worker_thread =
+        SchedulerWorkerThread::CreateSchedulerWorkerThread(
+            thread_priority, main_entry_callback, get_work_callback,
+            ran_task_from_sequence_callback, task_tracker);
+    if (!worker_thread)
+      break;
+    idle_worker_threads_stack_.push(worker_thread.get());
+    worker_threads_.push_back(std::move(worker_thread));
+  }
+}
+
+void SchedulerThreadPool::WakeUpOneThread() {
+  AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+
+  if (idle_worker_threads_stack_.empty())
+    return;
+
+  SchedulerWorkerThread* worker_thread = idle_worker_threads_stack_.top();
+  idle_worker_threads_stack_.pop();
+  worker_thread->WakeUp();
+}
+
+void SchedulerThreadPool::AddToIdleSchedulerWorkerThreadsStack(
+    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());
+
+  if (idle_worker_threads_stack_.size() == worker_threads_.size())
+    idle_worker_threads_stack_cv_->Signal();
+}
+
+void SchedulerThreadPool::SequenceInsertedInSharedPriorityQueueCallback() {
+  if (!no_wake_up_on_sequence_insertion_.Get())
+    WakeUpOneThread();
+}
+
+void SchedulerThreadPool::MainEntryCallback() const {
+  DCHECK(!g_current_shared_priority_queue.Get().Get());
+  g_current_shared_priority_queue.Get().Set(&shared_priority_queue_);
+}
+
+scoped_refptr<Sequence> SchedulerThreadPool::GetWorkCallback(
+    SchedulerWorkerThread* worker_thread) {
+  std::unique_ptr<PriorityQueue::Transaction> transaction(
+      shared_priority_queue_.BeginTransaction());
+  const PriorityQueue::SequenceAndSortKey sequence_and_sort_key(
+      transaction->Peek());
+
+  if (sequence_and_sort_key.is_null()) {
+    // |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.
+    // 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.
+    AddToIdleSchedulerWorkerThreadsStack(worker_thread);
+    return nullptr;
+  }
+
+  transaction->Pop();
+  return sequence_and_sort_key.sequence;
+}
+
+void PostTaskHelper(std::unique_ptr<Task> task,
+                    scoped_refptr<Sequence> sequence,
+                    PriorityQueue* priority_queue,
+                    TaskTracker* task_tracker) {
+  DCHECK(task);
+  DCHECK(sequence);
+  DCHECK(priority_queue);
+  DCHECK(task_tracker);
+
+  task_tracker->PostTask(
+      Bind(&PostTaskCallback, std::move(sequence), priority_queue),
+      std::move(task));
+}
+
+}  // namespace internal
+}  // namespace base