Task Scheduler.

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..d6efa05e578650cf4bc7b46c08931f33e0c850e0
--- /dev/null
+++ b/base/task_scheduler/worker_thread.cc
@@ -0,0 +1,308 @@
+// 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/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 task_scheduler {
+
+namespace {
+
+// A task runner that runs tasks on a single WorkerThread.
+class SchedulerSingleThreadTaskRunner : public SingleThreadTaskRunner {
+ 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 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, reinsert_sequence_callback,
+      becomes_idle_callback, delayed_task_manager, shutdown_manager));
+  return worker_thread->IsValid() ? std::move(worker_thread)
+                                  : scoped_ptr<WorkerThread>();
+}
+
+WorkerThread::~WorkerThread() = default;
+
+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 ||
+         execution_mode == ExecutionMode::SINGLE_THREADED_COM_STA);
+#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 {
+  return !single_thread_priority_queue_.UnsynchronizedEmpty() ||
+         is_running_single_threaded_task_;
+}
+
+void WorkerThread::JoinForTesting() {
+  DCHECK(shutdown_manager_->shutdown_completed());
+  WakeUp();
+  PlatformThread::Join(thread_handle_);
+}
+
+WorkerThread::WorkerThread(
+    ThreadPriority thread_priority,
+    PriorityQueue* shared_priority_queue,
+    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),
+      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_->GetNextDelayedTaskReadyTime();
+
+  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() {
+  for (;;) {

[robliao, 2016/02/11 21:56:27]

Maybe this should be while(!shutdown_manager->shutdown_completed())

[fdoray, 2016/02/12 04:16:20]

Done.

+    // 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();

[robliao, 2016/02/11 22:30:07]

Can we remove this somehow?

[fdoray, 2016/02/12 04:16:20]

We could if we built something that atomically checks if there is work to do and
calls WaitUntilWorkIsAvailable if there isn't. I'll work on this tomorrow.

+
+      if (sequence.get() == nullptr) {
+        WaitUntilWorkIsAvailable();
+        sequence = GetWork();
+      }
+    }
+
+    if (sequence.get() != nullptr) {
+      // Peek the next task in the sequence.
+      const Task* task = sequence->PeekTask();
+      DCHECK(task);

[robliao, 2016/02/11 22:30:07]

Remove DCHECK.

[fdoray, 2016/02/12 04:16:20]

Done.

+      const TaskShutdownBehavior shutdown_behavior =
+          task->traits.shutdown_behavior();
+
+      // Run the task.
+      if (shutdown_manager_->ShouldScheduleTask(shutdown_behavior)) {
+        debug::TaskAnnotator task_annotator;
+        task_annotator.RunTask("task_scheduler::PostTask", *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;
+    }
+
+    // Exit the thread if shutdown is complete.
+    if (shutdown_manager_->shutdown_completed())
+      break;
+
+    // Post delayed tasks that are ready for execution.
+    delayed_task_manager_->PostReadyTasks();
+  }
+}
+
+}  // namespace task_scheduler
+}  // namespace base