TaskScheduler [7] SchedulerThreadPool

base/task_scheduler/scheduler_thread_pool.cc

Index: base/task_scheduler/scheduler_thread_pool.cc
diff --git a/base/task_scheduler/scheduler_thread_pool.cc b/base/task_scheduler/scheduler_thread_pool.cc
new file mode 100644
index 0000000000000000000000000000000000000000..12a164d2883aba54323935647d3e88f432673c6b
--- /dev/null
+++ b/base/task_scheduler/scheduler_thread_pool.cc
@@ -0,0 +1,271 @@
+// 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/utils.h"
+#include "base/threading/thread_local.h"
+
+namespace base {
+namespace internal {
+
+namespace {
+
+// SchedulerThreadPool that owns the current thread. Not set for threads that
+// don't belong to a SchedulerThreadPool.
+LazyInstance<ThreadLocalPointer<const SchedulerThreadPool>>::Leaky
+    tls_current_thread_pool = LAZY_INSTANCE_INITIALIZER;
+
+// A task runner that runs tasks with the PARALLEL ExecutionMode.
+class SchedulerParallelTaskRunner : public TaskRunner {
+ public:
+  SchedulerParallelTaskRunner(const TaskTraits& traits,
+                              SchedulerTaskExecutor* executor,
+                              TaskTracker* task_tracker)
+      : traits_(traits), executor_(executor), task_tracker_(task_tracker) {}
+
+  // TaskRunner:
+  bool PostDelayedTask(const tracked_objects::Location& from_here,
+                       const Closure& closure,
+                       TimeDelta delay) override {
+    // Post the task as part of a one-off single-task Sequence.
+    return PostTaskHelper(from_here, closure, traits_, delay,
+                          make_scoped_refptr(new Sequence), executor_,
+                          task_tracker_);
+  }
+
+  bool RunsTasksOnCurrentThread() const override {
+    return tls_current_thread_pool.Get().Get() == executor_;
+  }
+
+ private:
+  ~SchedulerParallelTaskRunner() override = default;
+
+  const TaskTraits traits_;
+  SchedulerTaskExecutor* const executor_;
+  TaskTracker* const task_tracker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerParallelTaskRunner);
+};
+
+}  // namespace
+
+class SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl
+    : public SchedulerWorkerThread::Delegate {
+ public:
+  SchedulerWorkerThreadDelegateImpl(
+      SchedulerThreadPool* outer,
+      const EnqueueSequenceCallback enqueue_sequence_callback);
+  ~SchedulerWorkerThreadDelegateImpl() override;
+
+  // SchedulerWorkerThread::Delegate:
+  void OnMainEntry() override;
+  scoped_refptr<Sequence> GetWork(
+      SchedulerWorkerThread* worker_thread) override;
+  void EnqueueSequence(scoped_refptr<Sequence> sequence) override;
+
+ private:
+  SchedulerThreadPool* outer_;
+  const EnqueueSequenceCallback enqueue_sequence_callback_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerWorkerThreadDelegateImpl);
+};
+
+std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool(
+    ThreadPriority thread_priority,
+    size_t max_threads,
+    const EnqueueSequenceCallback& enqueue_sequence_callback,
+    TaskTracker* task_tracker) {
+  std::unique_ptr<SchedulerThreadPool> thread_pool(
+      new SchedulerThreadPool(enqueue_sequence_callback, task_tracker));
+  if (thread_pool->Initialize(thread_priority, max_threads))
+    return thread_pool;
+  return nullptr;
+}
+
+SchedulerThreadPool::~SchedulerThreadPool() {
+#if DCHECK_IS_ON()
+  // SchedulerThreadPool should never be deleted in production unless its
+  // initialization failed.
+  AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+  DCHECK(join_for_testing_returned_.IsSignaled() || worker_threads_.empty());
+#endif  // DCHECK_IS_ON()
+}
+
+scoped_refptr<TaskRunner> SchedulerThreadPool::CreateTaskRunnerWithTraits(
+    const TaskTraits& traits,
+    ExecutionMode execution_mode) {
+  switch (execution_mode) {
+    case ExecutionMode::PARALLEL:
+      return make_scoped_refptr(
+          new SchedulerParallelTaskRunner(traits, this, task_tracker_));
+
+    case ExecutionMode::SEQUENCED:
+    case ExecutionMode::SINGLE_THREADED:
+      // TODO(fdoray): Support SEQUENCED and SINGLE_THREADED TaskRunners.
+      NOTREACHED();
+      return nullptr;
+  }
+
+  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())
+    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();
+}
+
+SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::
+    SchedulerWorkerThreadDelegateImpl(
+        SchedulerThreadPool* outer,
+        const EnqueueSequenceCallback enqueue_sequence_callback)

[danakj, 2016/04/14 19:48:58]

const&?

[fdoray, 2016/04/14 23:46:58]

Done.

+    : outer_(outer), enqueue_sequence_callback_(enqueue_sequence_callback) {}
+
+SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::
+    ~SchedulerWorkerThreadDelegateImpl() = default;
+
+void SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::OnMainEntry() {

[danakj, 2016/04/14 19:48:58]

Just in case, if these are in a DelegateImpl and not just on SchedulerThreadPool
because of previous talking about moving the Delegate to not be a nested class
(where I meant the delegate definition itself), you could put them back, not
sure.

[fdoray, 2016/04/14 23:46:58]

These methods are in DelegateImpl instead of SchedulerThreadPool because we
didn't want 2 methods called EnqueueSequence on SchedulerThreadPool.

+  DCHECK(!tls_current_thread_pool.Get().Get());
+  tls_current_thread_pool.Get().Set(outer_);
+}
+
+scoped_refptr<Sequence>
+SchedulerThreadPool::SchedulerWorkerThreadDelegateImpl::GetWork(
+    SchedulerWorkerThread* worker_thread) {
+  std::unique_ptr<PriorityQueue::Transaction> transaction(
+      outer_->shared_priority_queue_.BeginTransaction());
+  const auto 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 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;
+}
+
+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)
+    : idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()),
+      idle_worker_threads_stack_cv_for_testing_(
+          idle_worker_threads_stack_lock_.CreateConditionVariable()),
+      join_for_testing_returned_(true, false),
+      worker_thread_delegate_(
+          new SchedulerWorkerThreadDelegateImpl(this,
+                                                enqueue_sequence_callback)),
+      task_tracker_(task_tracker) {
+  DCHECK(task_tracker_);
+}
+
+bool SchedulerThreadPool::Initialize(ThreadPriority thread_priority,
+                                     size_t max_threads) {
+  AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+
+  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_);
+    if (!worker_thread)
+      break;
+    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;
+  {

[robliao, 2016/04/13 20:50:54]

It might be cleaner to take the below and encapsulate it in PopOneIdleThread.

[fdoray, 2016/04/13 23:13:12]

Done.

[gab, 2016/04/14 16:11:34]

I disagree, now this is one more hop for the reader, effectively this scope was
an inlined anonymous method with an unambiguous purpose (i.e. didn't require an
explicit name to be understandable). I prefer this Patch Set's way to PS32's.

Not feeling very strongly, but less methods makes for an easier read IMO and
this extra one doesn't feel necessary.

[robliao, 2016/04/14 16:43:38]

The reason why I like the separation is that it keeps methods either fully
synchronized or not, making it easy to check concurrency.

[fdoray, 2016/04/14 18:41:11]

Keeping the code as-is because I plan to create a custom stack that supports
atomic Peek+Pop anyway (the custom stack will support removal of arbitrary
elements, which will be useful to remove threads from the stack when they are
woken up for single-threaded work).

+    AutoSchedulerLock auto_lock(idle_worker_threads_stack_lock_);
+
+    if (idle_worker_threads_stack_.empty())
+      return;
+
+    worker_thread = idle_worker_threads_stack_.top();
+    idle_worker_threads_stack_.pop();
+  }
+  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());
+
+  if (idle_worker_threads_stack_.size() == worker_threads_.size())
+    idle_worker_threads_stack_cv_for_testing_->Broadcast();
+}
+
+void SchedulerThreadPool::PostTaskNow(std::unique_ptr<Task> task,
+                                      scoped_refptr<Sequence> sequence) {
+  const bool sequence_was_empty = PostTaskNowHelper(
+      std::move(task), std::move(sequence), &shared_priority_queue_);
+
+  // No thread has already been woken up to run Tasks from |sequence| if it was
+  // empty before |task| was inserted into it.
+  if (sequence_was_empty)
+    WakeUpOneThread();
+}
+
+}  // namespace internal
+}  // namespace base