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"
+#include "base/threading/thread_local.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
+    tls_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());
+
+    // Create a new Sequence to allow parallel execution of Tasks posted through

[gab, 2016/04/07 20:32:47]

s/new Sequence/single-task Sequence/

("create" implies "new", but "single-task" is what's important to highlight
here)

[fdoray, 2016/04/08 14:53:02]

Done.

+    // this TaskRunner.
+    scoped_refptr<Sequence> sequence(new Sequence);

[gab, 2016/04/07 20:32:47]

I'd also be happy to inline this, something like:

     // Post the task as part of a one-off single-task Sequence.
     PostTaskHelper(WrapUnique(new Task(from_here, closure, traits_)),
                    make_scoped_refptr(new Sequence), priority_queue_,
task_tracker_);

[fdoray, 2016/04/08 14:53:02]

Done.

+
+    PostTaskHelper(WrapUnique(new Task(from_here, closure, traits_)),
+                   std::move(sequence), priority_queue_, task_tracker_);
+
+    return true;
+  }
+
+  bool RunsTasksOnCurrentThread() const override {
+    return tls_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);
+};
+
+}  // namespace
+
+std::unique_ptr<SchedulerThreadPool> SchedulerThreadPool::CreateThreadPool(
+    ThreadPriority thread_priority,
+    size_t max_threads,
+    const RanTaskFromSequenceCallback& ran_task_from_sequence_callback,
+    TaskTracker* task_tracker) {
+  std::unique_ptr<SchedulerThreadPool> thread_pool(
+      new SchedulerThreadPool(ran_task_from_sequence_callback, task_tracker));
+  thread_pool->Initialize(thread_priority, max_threads);
+  if (thread_pool->worker_threads_.empty())
+    return nullptr;
+  return thread_pool;
+}
+
+SchedulerThreadPool::~SchedulerThreadPool() {
+  AutoSchedulerLock auto_lock(join_for_testing_returned_lock_);
+  DCHECK(join_for_testing_returned_ || worker_threads_.empty());

[gab, 2016/04/07 20:32:47]

Wrap these two lines in #if DCHECK_IS_ON() to avoid locking in Release builds?
(or even compiling it in given it's never supposed to run)

Also add a comment verbally explaining this check like:

// SchedulerThreadPool should never be deleted in production unless its
initialization failed.

[fdoray, 2016/04/08 14:53:02]

Done.

+}
+
+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:
+      // TODO(fdoray): Support SEQUENCED and SINGLE_THREADED TaskRunners.
+      NOTREACHED();
+      return nullptr;
+  }
+
+  NOTREACHED();
+  return nullptr;
+}
+
+void SchedulerThreadPool::InsertSequenceAfterTaskRan(
+    scoped_refptr<Sequence> sequence,
+    const SequenceSortKey& sequence_sort_key) {
+  auto sequence_and_sort_key = WrapUnique(new PriorityQueue::SequenceAndSortKey(
+      std::move(sequence), sequence_sort_key));
+  auto transaction = shared_priority_queue_.BeginTransaction();
+
+  // 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_shared_priority_queue.Get().Get() == &shared_priority_queue_)
+    transaction->PushNoWakeUp(std::move(sequence_and_sort_key));
+  else
+    transaction->Push(std::move(sequence_and_sort_key));
+}
+
+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_);

[gab, 2016/04/07 20:32:47]

We could get away with a MemoryBarrier instead of a lock?

Or shall we use a WaitableEvent? Essentially this is a boolean whose state needs
to be synced across threads. WaitableEvent sounds fine to me (I'd like to reduce
the number of members which are test only).

[fdoray, 2016/04/08 14:53:02]

I wanted to use a MemoryBarrier but
https://codereview.chromium.org/1704113002/diff/280001/base/task_scheduler/wo...
Let's use a WaitableEvent.

+  DCHECK(!join_for_testing_returned_);
+  join_for_testing_returned_ = true;
+}
+
+SchedulerThreadPool::SchedulerThreadPool(
+    const RanTaskFromSequenceCallback& ran_task_from_sequence_callback,
+    TaskTracker* task_tracker)
+    : shared_priority_queue_(
+          Bind(&SchedulerThreadPool::WakeUpOneThread, Unretained(this))),
+      idle_worker_threads_stack_lock_(shared_priority_queue_.container_lock()),
+      idle_worker_threads_stack_cv_(
+          idle_worker_threads_stack_lock_.CreateConditionVariable()),
+      ran_task_from_sequence_callback_(ran_task_from_sequence_callback),
+      task_tracker_(task_tracker) {
+  DCHECK(task_tracker_);
+}
+
+void SchedulerThreadPool::Initialize(ThreadPriority thread_priority,
+                                     size_t max_threads) {
+  DCHECK(worker_threads_.empty());
+
+  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,
+                                                           this, 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();

[gab, 2016/04/07 20:32:47]

Could technically release the lock before invoking |WakeUp()| if we think this
may help performance.

[fdoray, 2016/04/08 14:53:02]

Done.

+}
+
+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_->Broadcast();

[gab, 2016/04/07 20:32:47]

Would a manual-reset WaitableEvent be better here? It feels weird to make a
broadcast of something that might become false any second (I understand it
probably doesn't in the controlled way the test is set up but the reader of this
code doesn't know that).

[fdoray, 2016/04/08 14:53:02]

The WaitableEvent would be reset in WakeUpOneThread, when a thread is removed
from the stack? I don't like this solution because it means that we'll have to
think about resetting the event whenever we remove a thread from the stack
(eventually, we'll also remove a thread from the stack when single-threaded work
is posted). Currently, WaitForAllWorkerThreadsIdleForTesting only returns when
all threads are idle. This state can change immediately after, but it would also
be the case with a WaitableEvent.

[gab, 2016/04/08 17:56:00]

But the Broadcast's call can racily be lost, no? i.e. if the broadcast occurs
before the WaitForAllWorkerThreadsIdleForTesting call, the wait will block
forever?

[fdoray, 2016/04/08 19:00:05]

WaitForAllWorkerThreadsIdleForTesting will return immediately if all threads are
idle, even if the broadcast is done before:

while (idle_worker_threads_stack_.size() < worker_threads_.size())  <- this
condition will be false
  idle_worker_threads_stack_cv_->Wait();

That's how ConditionVariable are supposed to be used
https://code.google.com/p/chromium/codesearch#chromium/src/base/synchronizati...

+}
+
+void SchedulerThreadPool::OnMainEntry() {
+  DCHECK(!tls_current_shared_priority_queue.Get().Get());
+  tls_current_shared_priority_queue.Get().Set(&shared_priority_queue_);
+}
+
+void SchedulerThreadPool::OnMainExit() {
+  DCHECK(tls_current_shared_priority_queue.Get().Get());
+  tls_current_shared_priority_queue.Get().Set(nullptr);

[gab, 2016/04/07 20:32:47]

Actually looking into this deeper turns out this is not required.
This line merely sets the mapping to null instead of an invalid pointer it
doesn't actually remove the slot allocated for this thread.

But this automatically does
https://code.google.com/p/chromium/codesearch#chromium/src/base/threading/thr....

In SchedulerLock we actually had to use ThreadLocalStorage::Slot directly
instead of ThreadLocalPointer per the pointer being put in there being owned by
the TLS state and us thus needing to register custom destructor handling. In
this case it's not owned and letting the map auto clear itself on thread exit is
sufficient.

(ThreadLocalPointer is implemented in terms of ThreadLocalStorage::StaticSlot on
Windows anyways -- Rob can rant some more here as to why we still have the two
very similar impls, mostly an unfinished cleanup FWIR)

[fdoray, 2016/04/08 14:53:02]

Done. Interesting.

+}
+
+scoped_refptr<Sequence> SchedulerThreadPool::GetWork(
+    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 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.
+    AddToIdleWorkerThreadsStack(worker_thread);
+    return nullptr;
+  }
+
+  transaction->Pop();

[gab, 2016/04/07 20:32:47]

Actually since this is all done under a single transaction could we just have
Pop() return the value and remove Peek()?

[fdoray, 2016/04/08 14:53:02]

Eventually, we'll have to:
- peek from a shared and a single-threaded PQ
- determine which sequence has the highest priority
- pop it from its PQ
We need separate methods for peek/pop to do that.

+  return sequence_and_sort_key.sequence;
+}
+
+void SchedulerThreadPool::RanTaskFromSequence(
+    scoped_refptr<Sequence> sequence) {
+  ran_task_from_sequence_callback_.Run(std::move(sequence));
+}
+
+bool 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);
+
+  if (!task_tracker->WillPostTask(task.get()))
+    return false;
+
+  const bool sequence_was_empty = sequence->PushTask(std::move(task));
+  if (sequence_was_empty) {
+    // Insert |sequence| in |priority_queue| if it was empty before |task| was
+    // inserted into it. When that's not the case, one of these must be true:
+    // - |sequence| is already in a PriorityQueue, or,
+    // - A worker thread is running a Task from |sequence|. It will insert
+    //   |sequence| in a PriorityQueue once it's done running the Task.
+    const SequenceSortKey sequence_sort_key = sequence->GetSortKey();

[gab, 2016/04/07 20:32:47]

inline this?

[fdoray, 2016/04/08 14:53:02]

I can't. I have to make sure that sequence->GetSortKey() runs before
std::move(sequence).

+    priority_queue->BeginTransaction()->Push(
+        WrapUnique(new PriorityQueue::SequenceAndSortKey(std::move(sequence),
+                                                         sequence_sort_key)));
+  }
+
+  return true;
+}
+
+}  // namespace internal
+}  // namespace base