TaskScheduler [6] SchedulerWorkerThread

base/task_scheduler/worker_thread.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/worker_thread.h"
+
+#include <ostream>
+#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) {
+  // Tasks are never nested on WorkerThread.
+  return PostDelayedTask(from_here, task, delay);
+}
+
+SchedulerSingleThreadTaskRunner::~SchedulerSingleThreadTaskRunner() = default;
+
+}  // namespace
+
+scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const SharedSequenceStillHasTasksCallback&
+        shared_sequence_still_has_tasks_callback,
+    const StateChangedCallback& state_changed_callback,
+    TaskTracker* task_tracker) {
+  scoped_ptr<WorkerThread> worker_thread(
+      new WorkerThread(thread_priority, shared_priority_queue,
+                       shared_sequence_still_has_tasks_callback,
+                       state_changed_callback, task_tracker));
+
+  if (worker_thread->thread_handle_.is_null())
+    return scoped_ptr<WorkerThread>();
+  return worker_thread;
+}
+
+WorkerThread::~WorkerThread() {
+  DCHECK(should_exit_for_testing());
+}
+
+void WorkerThread::WakeUp() {
+  wake_up_event_.Signal();
+}
+
+scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits(
+    const TaskTraits& traits) {
+  // A WorkerThread is never destroyed, except in tests in which we don't use
+  // task runners after their WorkerThread has been destroyed. Because of that,
+  // it is correct to keep pointers to WorkerThread members in the constructed
+  // task runner.
+  return scoped_refptr<SingleThreadTaskRunner>(
+      new SchedulerSingleThreadTaskRunner(
+          traits, &single_thread_priority_queue_, task_tracker_));
+}
+
+void WorkerThread::JoinForTesting() {
+  should_exit_for_testing_ = true;
+  base::subtle::MemoryBarrier();
+  WakeUp();
+  PlatformThread::Join(thread_handle_);
+}
+
+WorkerThread::WorkerThread(ThreadPriority thread_priority,
+                           PriorityQueue* shared_priority_queue,
+                           const SharedSequenceStillHasTasksCallback&
+                               shared_sequence_still_has_tasks_callback,
+                           const StateChangedCallback& state_changed_callback,
+                           TaskTracker* task_tracker)
+    : wake_up_event_(false, false),
+      single_thread_priority_queue_(
+          Bind(&WorkerThread::WakeUp, Unretained(this)),
+          shared_priority_queue),
+      shared_priority_queue_(shared_priority_queue),
+      shared_sequence_still_has_tasks_callback_(
+          shared_sequence_still_has_tasks_callback),
+      state_changed_callback_(state_changed_callback),
+      task_tracker_(task_tracker) {
+  DCHECK(shared_priority_queue_);
+  DCHECK(!shared_sequence_still_has_tasks_callback_.is_null());
+  DCHECK(!state_changed_callback_.is_null());
+  DCHECK(task_tracker_);
+
+  const size_t kDefaultStackSize = 0;

[robliao, 2016/03/24 23:03:26]

Might as well make this static const.

[fdoray, 2016/03/29 18:33:34]

Done.

+  PlatformThread::CreateWithPriority(kDefaultStackSize, this, &thread_handle_,
+                                     thread_priority);
+}
+
+scoped_refptr<Sequence> WorkerThread::GetWork(bool* is_single_threaded) {
+  DCHECK(is_single_threaded);

[robliao, 2016/03/24 23:03:26]

I think this is one of the rare cases where a DCHECK might actually not be
necessary given that we're writing to it on the next line. I leave it up to you.

[fdoray, 2016/03/29 18:33:34]

Done.

+  *is_single_threaded = 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();
+  *is_single_threaded = true;
+  return single_thread_sequence.sequence;
+}
+
+void WorkerThread::SetState(State state) {
+  DCHECK_NE(state_, state);
+  state_ = state;
+  state_changed_callback_.Run(this, state);
+}
+
+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;
+    scoped_refptr<Sequence> sequence = GetWork(&sequence_is_single_threaded);
+    if (!sequence) {
+      SetState(State::IDLE);
+
+      // Check one more time if there is work available. Work could have been
+      // added to |shared_priority_queue_| between the first call to GetWork()
+      // and the call to SetState() above in which case this WorkerThread
+      // shouldn't go to sleep (the recipient of |state_changed_callback_|
+      // hadn't be notified that this WorkerThread was idle and might have

[robliao, 2016/03/24 23:03:26]

Nit: hasn't

[fdoray, 2016/03/29 18:33:34]

I don't think that "hasn't" is correct because at this point, the recipient HAS
been notified. I reworded the sentence to express that it hadn't be notified
when the work was posted. Is is clearer?

[robliao, 2016/03/30 00:42:42]

This CL comment is no longer relevant as of patchset 12

+      // assumed that it was gonna run the newly added work). Note that
+      // |wake_up_event_| is always signaled when work is added to
+      // |single_thread_priority_queue_|.
+      sequence = GetWork(&sequence_is_single_threaded);
+
+      if (!sequence) {
+        wake_up_event_.Wait();
+
+        // Always mark this WorkerThread as BUSY after a wake up, regardless of
+        // whether work is actually found in its PriorityQueues. This ensures
+        // that there won't be 2 consecutive IDLE invocations of
+        // |state_changed_callback_| in case of spurious wake up.
+        SetState(State::BUSY);
+
+        continue;
+      }
+
+      SetState(State::BUSY);
+      wake_up_event_.Reset();

[robliao, 2016/03/24 23:03:26]

Why is this reset necessary since our events are currently autoreset?

[fdoray, 2016/03/29 18:33:34]

I moved the reset at the bottom of this method and added a comment to explain
why it's there. I think it's a nice optimization but it's not absolutely
required.

+    }
+
+    DCHECK_EQ(state_, State::BUSY);
+
+    // Peek the next task in |sequence| and run it.
+    task_tracker_->RunTask(sequence->PeekTask());
+
+    // Pop a task from |sequence|. If it is empty after this, reinsert it in

[robliao, 2016/03/24 23:03:26]

I think you mean if the sequence isn't empty.

I would reword the comment as follows...
To maintain the non-empty sequence invariant for the PQ reinsert the sequence
into its original PQ if it's not empty.

[fdoray, 2016/03/29 18:33:34]

We don't necessary reinsert the sequence in its original PQ. A sequence with
BACKGROUND+USER_VISIBLE tasks can change priority queue when its USER_VISIBLE
task is popped.

+    // |single_thread_priority_queue_| or notify
+    // |shared_sequence_still_has_tasks_callback_|.
+    if (!sequence->PopTask()) {
+      if (sequence_is_single_threaded) {
+        const SequenceSortKey sort_key = sequence->GetSortKey();
+        single_thread_priority_queue_.BeginTransaction()->Push(
+            make_scoped_ptr(new PriorityQueue::SequenceAndSortKey(
+                std::move(sequence), sort_key)));
+      } else {
+        shared_sequence_still_has_tasks_callback_.Run(this,
+                                                      std::move(sequence));
+      }
+    }
+  }
+}
+
+std::ostream& operator<<(std::ostream& os, WorkerThread::State state) {
+  switch (state) {
+    case WorkerThread::State::BUSY:
+      os << "BUSY";
+      break;
+    case WorkerThread::State::IDLE:
+      os << "IDLE";
+      break;
+  }
+  return os;
+}
+
+}  // namespace internal
+}  // namespace base