TaskScheduler [6] SchedulerWorkerThread

base/task_scheduler/worker_thread.cc

Index: base/task_scheduler/worker_thread.cc
diff --git a/base/task_scheduler/worker_thread.cc b/base/task_scheduler/worker_thread.cc
new file mode 100644
index 0000000000000000000000000000000000000000..27f919168cb2205ea74cf21cde33f71403bb3f44
--- /dev/null
+++ b/base/task_scheduler/worker_thread.cc
@@ -0,0 +1,251 @@
+// 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/worker_thread.h"
+
+#include <utility>
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "base/task_scheduler/task_tracker.h"
+#include "base/task_scheduler/utils.h"
+#include "base/time/time.h"
+
+namespace base {
+namespace internal {
+
+namespace {
+
+class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
+ public:
+  SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
+                                  PriorityQueue* single_thread_priority_queue,
+                                  TaskTracker* task_tracker);
+
+  // SingleThreadTaskRunner:
+  bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
+                                  const Closure& task,
+                                  TimeDelta delay) override;
+  bool PostDelayedTask(const tracked_objects::Location& from_here,
+                       const Closure& closure,
+                       TimeDelta delay) override;
+  bool RunsTasksOnCurrentThread() const override;
+
+ private:
+  ~SchedulerSingleThreadTaskRunner() override;
+
+  const TaskTraits traits_;
+  const scoped_refptr<Sequence> sequence_;
+  PriorityQueue* const priority_queue_;
+  TaskTracker* const task_tracker_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
+};
+
+SchedulerSingleThreadTaskRunner::SchedulerSingleThreadTaskRunner(
+    const TaskTraits& traits,
+    PriorityQueue* single_thread_priority_queue,
+    TaskTracker* task_tracker)
+    : traits_(traits),
+      sequence_(new Sequence),
+      priority_queue_(single_thread_priority_queue),
+      task_tracker_(task_tracker) {}
+
+bool SchedulerSingleThreadTaskRunner::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& closure,
+    TimeDelta delay) {
+  // TODO(fdoray): Support delayed tasks.
+  DCHECK(delay.is_zero());
+  PostTaskHelper(make_scoped_ptr(new Task(from_here, closure, traits_)),
+                 sequence_, priority_queue_, task_tracker_);
+  return true;
+}
+
+bool SchedulerSingleThreadTaskRunner::RunsTasksOnCurrentThread() const {
+  // TODO(fdoray): Return true only if tasks posted may actually run on the
+  // current thread. It is valid, but not ideal, to always return true.
+  return true;
+}
+
+bool SchedulerSingleThreadTaskRunner::PostNonNestableDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& task,
+    TimeDelta delay) {
+  return PostDelayedTask(from_here, task, delay);
+}
+
+SchedulerSingleThreadTaskRunner::~SchedulerSingleThreadTaskRunner() = default;
+
+void PushSequenceInPriorityQueue(scoped_refptr<Sequence> sequence,
+                                 PriorityQueue* priority_queue) {
+  const SequenceSortKey sort_key = sequence->GetSortKey();
+  priority_queue->BeginTransaction()->Push(make_scoped_ptr(
+      new PriorityQueue::SequenceAndSortKey(std::move(sequence), sort_key)));
+}
+
+}  // namespace
+
+scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const ReinsertSequenceCallback& reinsert_sequence_callback,
+    const BecomesIdleCallback& becomes_idle_callback,
+    TaskTracker* task_tracker) {
+  scoped_ptr<WorkerThread> worker_thread(new WorkerThread(
+      thread_priority, shared_priority_queue, reinsert_sequence_callback,
+      becomes_idle_callback, task_tracker));
+
+  if (worker_thread->thread_handle_.is_null())
+    return scoped_ptr<WorkerThread>();
+  return worker_thread;
+}
+
+WorkerThread::~WorkerThread() {
+  AutoSchedulerLock auto_lock(lock_);
+  DCHECK(should_exit_for_testing_);
+}
+
+bool WorkerThread::WakeUp() {
+  AutoSchedulerLock auto_lock(lock_);
+  if (is_awake_)
+    return false;
+  is_awake_ = true;
+  wake_up_cv_->Signal();
+  return true;
+}
+
+scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits(
+    const TaskTraits& traits) {
+  return scoped_refptr<SingleThreadTaskRunner>(
+      new SchedulerSingleThreadTaskRunner(
+          traits, &single_thread_priority_queue_, task_tracker_));
+}
+
+void WorkerThread::JoinForTesting() {
+  {
+    AutoSchedulerLock auto_lock(lock_);
+    DCHECK(!should_exit_for_testing_);
+    should_exit_for_testing_ = true;
+  }
+  WakeUp();
+  PlatformThread::Join(thread_handle_);
+}
+
+WorkerThread::WorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const ReinsertSequenceCallback& reinsert_sequence_callback,
+    const BecomesIdleCallback& becomes_idle_callback,
+    TaskTracker* task_tracker)
+    : wake_up_cv_(lock_.CreateConditionVariable()),
+      is_awake_(true),
+      should_exit_for_testing_(false),
+      single_thread_priority_queue_(
+          Bind(IgnoreResult(&WorkerThread::WakeUp), Unretained(this)),
+          shared_priority_queue),
+      shared_priority_queue_(shared_priority_queue),
+      reinsert_sequence_callback_(reinsert_sequence_callback),
+      becomes_idle_callback_(becomes_idle_callback),
+      task_tracker_(task_tracker) {
+  DCHECK(shared_priority_queue_);
+  DCHECK(!reinsert_sequence_callback_.is_null());
+  DCHECK(!becomes_idle_callback_.is_null());
+  DCHECK(task_tracker_);
+
+#if defined(OS_MACOSX)
+  // Mac only supports 2 priorities. crbug.com/554651
+  if (thread_priority != ThreadPriority::NORMAL &&
+      thread_priority != ThreadPriority::REALTIME_AUDIO) {
+    thread_priority = ThreadPriority::NORMAL;
+  }
+#endif  // defined(OS_MACOSX)
+
+  const size_t kDefaultStackSize = 0;
+  PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_,
+                                     thread_priority);
+}
+
+scoped_refptr<Sequence> WorkerThread::GetWork(bool* single_thread) {
+  DCHECK(single_thread);
+  *single_thread = false;
+
+  scoped_ptr<PriorityQueue::Transaction> shared_transaction(
+      shared_priority_queue_->BeginTransaction());
+  const PriorityQueue::SequenceAndSortKey shared_sequence =
+      shared_transaction->Peek();
+
+  scoped_ptr<PriorityQueue::Transaction> single_thread_transaction(
+      single_thread_priority_queue_.BeginTransaction());
+  const PriorityQueue::SequenceAndSortKey single_thread_sequence =
+      single_thread_transaction->Peek();
+
+  if (single_thread_sequence.is_null() && shared_sequence.is_null()) {
+    return scoped_refptr<Sequence>();
+  }
+
+  if (single_thread_sequence.is_null() ||
+      (!shared_sequence.is_null() &&
+       single_thread_sequence.sort_key < shared_sequence.sort_key)) {
+    shared_transaction->Pop();
+    return shared_sequence.sequence;
+  }
+
+  DCHECK(!single_thread_sequence.is_null());
+  single_thread_transaction->Pop();
+  *single_thread = true;
+  return single_thread_sequence.sequence;
+}
+
+void WorkerThread::WaitUntilWakeUp() {
+  AutoSchedulerLock auto_lock(lock_);
+  while (!is_awake_)
+    wake_up_cv_->Wait();
+}
+
+void WorkerThread::ThreadMain() {
+  while (!task_tracker_->shutdown_completed() && !should_exit_for_testing_) {
+    // Get the sequence containing the next task to execute.
+    bool sequence_is_single_threaded = false;

[robliao, 2016/03/09 15:41:12]

For unit tests:
If we're exiting the test, call WakeUp, and the thread has not marked itself
as is_awake_ = false, WakeUp will have no effect.

if is_awake_ is false and we call WakeUp, we will signal the condition
variable. If no thread is waiting, the signal will have no effect.

Fixed by rechecking should_exit_for_testing in:
https://luckyluke-private.googlesource.com/src/+/c8f5bbaff4f03637366ba67b9ab4...

We'll probably need a closer audit of the state in ThreadMain.

[fdoray, 2016/03/14 20:46:35]

You're right. Thanks for the fix. I moved it inside ThreadMain, where it was
easier to explain it.

+    scoped_refptr<Sequence> sequence = GetWork(&sequence_is_single_threaded);
+    if (sequence.get() == nullptr) {
+      // Mark the WorkerThread as idle.
+      {
+        AutoSchedulerLock auto_lock(lock_);
+        is_awake_ = false;
+      }
+      becomes_idle_callback_.Run(this);
+
+      // Check one more time if there is work available. If work has been added
+      // to |shared_priority_queue_| after the first call to GetWork() but
+      // before the |becomes_idle_callback_| invocation, this WorkerThread
+      // should run this work.
+      sequence = GetWork(&sequence_is_single_threaded);
+
+      if (sequence.get() == nullptr) {
+        WaitUntilWakeUp();
+        sequence = GetWork(&sequence_is_single_threaded);
+      }
+    }
+
+    if (sequence.get() != nullptr) {
+      // Peek the next task in the sequence and run it.
+      task_tracker_->RunTask(sequence->PeekTask());
+
+      // Pop the task from its sequence. If the sequence isn't empty, reinsert
+      // it in the appropriate PriorityQueue.
+      if (!sequence->PopTask()) {
+        if (sequence_is_single_threaded) {
+          PushSequenceInPriorityQueue(std::move(sequence),
+                                      &single_thread_priority_queue_);
+        } else {
+          reinsert_sequence_callback_.Run(std::move(sequence), this);
+        }
+      }
+    }
+  }
+}
+
+}  // namespace internal
+}  // namespace base