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 <utility>
+
+#include "base/atomicops.h"
+#include "base/bind.h"
+#include "base/debug/task_annotator.h"
+#include "base/logging.h"
+#include "base/task_scheduler/delayed_task_manager.h"
+#include "base/task_scheduler/priority_queue.h"
+#include "base/task_scheduler/shutdown_manager.h"
+#include "base/task_scheduler/utils.h"
+#include "base/time/time.h"
+#include "build/build_config.h"
+
+namespace base {
+namespace internal {
+
+namespace {
+
+// A task runner that runs tasks on a single WorkerThread.
+class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {

[robliao, 2016/03/07 19:28:08]

Working through the ChildTaskRunner, do these TaskRunners have to extend the
specific type (SingleThreadTaskRunner, SequencedTaskRunner)? We only return
TaskRunners, and things appear to work fine if they all extend TaskRunner
instead.

[gab, 2016/03/07 19:31:37]

They do, we will eventually need the underlying types to set
ThreadTaskRunnerHandles (ref. a_3_thread_handles in private repo).

+ public:
+  // Tasks posted through this task runner have |traits| and are inserted in
+  // |single_thread_priority_queue|. |delayed_task_manager| is used to post
+  // delayed tasks. |shutdown_manager| is notified when a task is posted.
+  SchedulerSingleThreadTaskRunner(const TaskTraits& traits,
+                                  PriorityQueue* single_thread_priority_queue,
+                                  DelayedTaskManager* delayed_task_manager,
+                                  ShutdownManager* shutdown_manager);
+
+  // 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;
+
+  TaskTraits traits_;
+  scoped_refptr<Sequence> sequence_;
+  PriorityQueue* priority_queue_;
+  DelayedTaskManager* delayed_task_manager_;
+  ShutdownManager* shutdown_manager_;
+
+  DISALLOW_COPY_AND_ASSIGN(SchedulerSingleThreadTaskRunner);
+};
+
+SchedulerSingleThreadTaskRunner::SchedulerSingleThreadTaskRunner(
+    const TaskTraits& traits,
+    PriorityQueue* priority_queue,
+    DelayedTaskManager* delayed_task_manager,
+    ShutdownManager* shutdown_manager)
+    : traits_(traits),
+      sequence_(new Sequence),
+      priority_queue_(priority_queue),
+      delayed_task_manager_(delayed_task_manager),
+      shutdown_manager_(shutdown_manager) {}
+
+bool SchedulerSingleThreadTaskRunner::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& closure,
+    TimeDelta delay) {
+  Task task(from_here, closure, traits_, TimeTicks::Now());
+  if (!delay.is_zero())
+    task.delayed_run_time = task.post_time + delay;
+  PostTaskHelper(task, sequence_, priority_queue_, shutdown_manager_,
+                 delayed_task_manager_);
+  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;
+
+}  // namespace
+
+scoped_ptr<WorkerThread> WorkerThread::CreateWorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const MainEntryCallback& main_entry_callback,
+    const ReinsertSequenceCallback& reinsert_sequence_callback,
+    const BecomesIdleCallback& becomes_idle_callback,
+    DelayedTaskManager* delayed_task_manager,
+    ShutdownManager* shutdown_manager) {
+  scoped_ptr<WorkerThread> worker_thread(new WorkerThread(
+      thread_priority, shared_priority_queue, main_entry_callback,
+      reinsert_sequence_callback, becomes_idle_callback, delayed_task_manager,
+      shutdown_manager));
+  return worker_thread->IsValid() ? std::move(worker_thread)
+                                  : scoped_ptr<WorkerThread>();
+}
+
+WorkerThread::~WorkerThread() {
+  DCHECK(thread_handle_.is_null());
+}
+
+void WorkerThread::WakeUp() {
+  wakeup_event_.Signal();
+}
+
+scoped_refptr<SingleThreadTaskRunner> WorkerThread::CreateTaskRunnerWithTraits(
+    const TaskTraits& traits,
+    ExecutionMode execution_mode) {
+#if defined(OS_WIN)
+  DCHECK(execution_mode == ExecutionMode::SINGLE_THREADED);
+#else
+  DCHECK(execution_mode == ExecutionMode::SINGLE_THREADED);
+#endif  // defined(OS_WIN)
+
+  return scoped_refptr<SingleThreadTaskRunner>(
+      new SchedulerSingleThreadTaskRunner(
+          traits, &single_thread_priority_queue_, delayed_task_manager_,
+          shutdown_manager_));
+}
+
+bool WorkerThread::HasSingleThreadedTasks() const {
+  subtle::MemoryBarrier();
+  return !single_thread_priority_queue_.UnsynchronizedEmpty() ||
+         is_running_single_threaded_task_;
+}
+
+void WorkerThread::ShutdownAndJoinForTesting() {
+  DCHECK(!thread_handle_.is_null());
+  shutdown_manager_->Shutdown();
+  WakeUp();
+  PlatformThread::Join(thread_handle_);
+  thread_handle_ = PlatformThreadHandle();
+}
+
+WorkerThread::WorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    const MainEntryCallback& main_entry_callback,
+    const ReinsertSequenceCallback& reinsert_sequence_callback,
+    const BecomesIdleCallback& becomes_idle_callback,
+    DelayedTaskManager* delayed_task_manager,
+    ShutdownManager* shutdown_manager)
+    : wakeup_event_(false, false),
+      is_running_single_threaded_task_(false),
+      single_thread_priority_queue_(
+          Bind(&WorkerThread::WakeUp, Unretained(this)),
+          shared_priority_queue),
+      shared_priority_queue_(shared_priority_queue),
+      main_entry_callback_(main_entry_callback),
+      reinsert_sequence_callback_(reinsert_sequence_callback),
+      becomes_idle_callback_(becomes_idle_callback),
+      delayed_task_manager_(delayed_task_manager),
+      shutdown_manager_(shutdown_manager) {
+  DCHECK(shared_priority_queue_);
+  DCHECK(!reinsert_sequence_callback.is_null());
+  DCHECK(!becomes_idle_callback.is_null());
+  DCHECK(delayed_task_manager_);
+  DCHECK(shutdown_manager_);
+
+#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);
+}
+
+bool WorkerThread::IsValid() const {
+  return !thread_handle_.is_null();
+}
+
+scoped_refptr<Sequence> WorkerThread::GetWork() {
+  scoped_ptr<PriorityQueue::Transaction> shared_transaction(
+      shared_priority_queue_->BeginTransaction());
+  SequenceSortKey shared_sort_key;
+  scoped_refptr<Sequence> shared_sequence =
+      shared_transaction->PeekSequence(&shared_sort_key);
+
+  scoped_ptr<PriorityQueue::Transaction> single_thread_transaction(
+      single_thread_priority_queue_.BeginTransaction());
+  SequenceSortKey single_thread_sort_key;
+  scoped_refptr<Sequence> single_thread_sequence =
+      single_thread_transaction->PeekSequence(&single_thread_sort_key);
+
+  if (single_thread_sequence.get() == nullptr &&
+      shared_sequence.get() == nullptr) {
+    return scoped_refptr<Sequence>();
+  }
+
+  if (single_thread_sequence.get() == nullptr ||
+      (shared_sequence.get() != nullptr &&
+       single_thread_sort_key < shared_sort_key)) {
+    shared_transaction->PopSequence();
+    return shared_sequence;
+  }
+
+  DCHECK(single_thread_sequence.get());
+
+  is_running_single_threaded_task_ = true;
+  single_thread_transaction->PopSequence();
+  return single_thread_sequence;
+}
+
+void WorkerThread::ReinsertSequenceInSingleThreadPriorityQueue(
+    scoped_refptr<Sequence> sequence) {
+  // Get the sort key of |sequence| before creating a priority queue
+  // transaction, to avoid holding 2 locks at the same time.
+  SequenceSortKey sort_key = sequence->GetSortKey();
+
+  // Insert the sequence in the single-thread priority queue.
+  single_thread_priority_queue_.BeginTransaction()->PushSequence(sequence,
+                                                                 sort_key);
+}
+
+void WorkerThread::WaitUntilWorkIsAvailable() {
+  const TimeTicks next_delayed_task_ready_time =
+      delayed_task_manager_->GetNextDelayedRunTime();
+
+  if (next_delayed_task_ready_time.is_null()) {
+    // There is no delayed tasks. Wait until |wakeup_event_| is signaled.
+    wakeup_event_.Wait();
+  } else {
+    // There is delayed tasks. Wait until either a delayed task becomes ready
+    // for execution or |wakeup_event_| is signaled. Note: Multiple threads
+    // sharing the same DelayedTaskManager may wake up at the same time when a
+    // delayed task becomes ready for execution. This isn't optimal. However,
+    // since most delayed tasks should be posted to BACKGROUND thread pools
+    // (which have a single thread), this behavior shouldn't occur frequently.
+    const TimeDelta wait_time = next_delayed_task_ready_time - TimeTicks::Now();
+    if (wait_time.InMilliseconds() > 0)
+      wakeup_event_.TimedWait(wait_time);
+  }
+}
+
+void WorkerThread::ThreadMain() {
+  main_entry_callback_.Run();
+  while (!shutdown_manager_->shutdown_completed()) {
+    // Get the sequence containing the next task to execute.
+    scoped_refptr<Sequence> sequence = GetWork();
+    if (sequence.get() == nullptr) {
+      // Add the thread to the stack of idle threads of the parent thread pool.
+      becomes_idle_callback_.Run(this);
+
+      // Check one more time whether there is pending work. Without this, it
+      // could be that work has been added to |shared_priority_queue_| after the
+      // first call to GetWork() and before this thread was added to the stack
+      // of idle threads. In such a case, |wake_up_event_| hasn't been signaled
+      // because this thread wasn't in the stack of idle threads. However, this
+      // thread is needed to execute the newly added work.
+      sequence = GetWork();
+
+      if (sequence.get() == nullptr) {
+        WaitUntilWorkIsAvailable();
+        sequence = GetWork();
+      }
+    }
+
+    if (sequence.get() != nullptr) {
+      // Peek the next task in the sequence.
+      const Task* task = sequence->PeekTask();
+      const TaskShutdownBehavior shutdown_behavior =
+          task->traits.shutdown_behavior();
+
+      // Run the task.
+      if (shutdown_manager_->ShouldScheduleTask(shutdown_behavior)) {
+        debug::TaskAnnotator task_annotator;
+        task_annotator.RunTask(kQueueFunctionName, *task);
+        shutdown_manager_->DidExecuteTask(shutdown_behavior);
+      }
+
+      // Pop the task from the sequence.
+      size_t new_num_tasks_in_sequence;
+      sequence->PopTask(&new_num_tasks_in_sequence);
+
+      // Put the sequence back in the appropriate priority queue.
+      if (new_num_tasks_in_sequence > 0) {
+        if (is_running_single_threaded_task_)
+          ReinsertSequenceInSingleThreadPriorityQueue(sequence);
+        else
+          reinsert_sequence_callback_.Run(sequence, this);
+      }
+
+      // Note that the thread is no longer running a single-threaded task.
+      is_running_single_threaded_task_ = false;
+    }
+
+    // Post delayed tasks that are ready for execution.
+    delayed_task_manager_->PostReadyTasks();
+  }
+}
+
+}  // namespace internal
+}  // namespace base