content: Add task queue manager

content/renderer/scheduler/task_queue_manager.cc

+// Copyright 2014 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 "content/renderer/scheduler/task_queue_manager.h"
+
+#include "base/bind.h"
+#include "base/debug/trace_event.h"
+#include "content/renderer/scheduler/task_queue_selector.h"
+
+namespace content {
+
+TaskQueueManager::TaskRunner::TaskRunner(
+    base::WeakPtr<TaskQueueManager> task_queue_manager,
+    size_t queue_index)
+    : task_queue_manager_(task_queue_manager), queue_index_(queue_index) {
+}
+
+TaskQueueManager::TaskRunner::~TaskRunner() {
+}
+
+bool TaskQueueManager::TaskRunner::RunsTasksOnCurrentThread() const {
+  if (!task_queue_manager_)
+    return false;
+  return task_queue_manager_->RunsTasksOnCurrentThread();
+}
+
+bool TaskQueueManager::TaskRunner::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta delay) {
+  if (!task_queue_manager_)
+    return false;
+  return task_queue_manager_->PostDelayedTask(
+      queue_index_, from_here, task, delay);
+}
+
+bool TaskQueueManager::TaskRunner::PostNonNestableDelayedTask(
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta delay) {
+  if (!task_queue_manager_)
+    return false;
+  return task_queue_manager_->PostNonNestableDelayedTask(
+      queue_index_, from_here, task, delay);
+}
+
+TaskQueueManager::InternalTaskQueue::InternalTaskQueue() : auto_pump(true) {
+}
+
+TaskQueueManager::InternalTaskQueue::~InternalTaskQueue() {
+}
+
+TaskQueueManager::TaskQueueManager(
+    size_t task_queue_count,
+    scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
+    TaskQueueSelector* selector)
+    : main_task_runner_(main_task_runner),
+      selector_(selector),
+      weak_factory_(this) {
+  DCHECK(main_task_runner->RunsTasksOnCurrentThread());
+
+  for (size_t i = 0; i < task_queue_count; i++) {
+    scoped_ptr<InternalTaskQueue> queue(new InternalTaskQueue());
+    queue->task_runner =
+        make_scoped_refptr(new TaskRunner(weak_factory_.GetWeakPtr(), i));
+    queues_.push_back(queue.release());
+  }
+
+  std::vector<const base::TaskQueue*> work_queues;
+  for (size_t i = 0; i < queues_.size(); i++)
+    work_queues.push_back(&queues_[i]->work_queue);
+  selector_->RegisterWorkQueues(work_queues);
+}
+
+TaskQueueManager::~TaskQueueManager() {
+}
+
+TaskQueueManager::InternalTaskQueue* TaskQueueManager::Queue(
+    size_t queue_index) const {
+  DCHECK_LT(queue_index, queues_.size());
+  return queues_[queue_index];
+}
+
+scoped_refptr<base::SingleThreadTaskRunner>
+TaskQueueManager::TaskRunnerForQueue(size_t queue_index) {
+  return Queue(queue_index)->task_runner;
+}
+
+bool TaskQueueManager::PollQueue(size_t queue_index) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  if (!queue->work_queue.empty())
+    return true;
+  base::AutoLock lock(queue->incoming_queue_lock);
+  return !queue->incoming_queue.empty();
+}
+
+bool TaskQueueManager::ReloadWorkQueue(size_t queue_index) {
+  main_thread_checker_.CalledOnValidThread();
+  InternalTaskQueue* queue = Queue(queue_index);
+  DCHECK(queue->work_queue.empty());
+  base::AutoLock lock(queue->incoming_queue_lock);
+  if (!queue->auto_pump)
+    return false;
+  queue->work_queue.Swap(&queue->incoming_queue);
+  return !queue->work_queue.empty();
+}
+
+void TaskQueueManager::EnqueueTask(size_t queue_index,
+                                   const base::PendingTask& pending_task) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  base::AutoLock lock(queue->incoming_queue_lock);
+  if (queue->auto_pump && queue->incoming_queue.empty())
+    PostDoWorkOnMainRunner();
+  queue->incoming_queue.push(pending_task);
+}
+
+void TaskQueueManager::SetAutoPump(size_t queue_index, bool auto_pump) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  base::AutoLock lock(queue->incoming_queue_lock);
+  if (auto_pump) {
+    queue->auto_pump = true;
+    PumpQueueLocked(queue);
+  } else {
+    queue->auto_pump = false;
+  }
+}
+
+void TaskQueueManager::PumpQueueLocked(InternalTaskQueue* queue) {
+  queue->incoming_queue_lock.AssertAcquired();
+  while (!queue->incoming_queue.empty()) {
+    queue->work_queue.push(queue->incoming_queue.front());
+    queue->incoming_queue.pop();
+  }
+  if (!queue->work_queue.empty())
+    PostDoWorkOnMainRunner();
+}
+
+void TaskQueueManager::PumpQueue(size_t queue_index) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  base::AutoLock lock(queue->incoming_queue_lock);
+  PumpQueueLocked(queue);
+}
+
+bool TaskQueueManager::UpdateWorkQueues() {
+  // TODO(skyostil): This is not efficient when the number of queues grows very
+  // large due to the number of locks taken. Consider optimizing when we get
+  // there.
+  bool has_work = false;
+  for (size_t i = 0; i < queues_.size(); i++) {
+    if (!queues_[i]->work_queue.empty())
+      has_work = true;
+    else if (ReloadWorkQueue(i))
+      has_work = true;
+  }
+  return has_work;
+}
+
+void TaskQueueManager::PostDoWorkOnMainRunner() {
+  main_task_runner_->PostTask(
+      FROM_HERE, Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr()));
+}
+
+void TaskQueueManager::DoWork() {
+  main_thread_checker_.CalledOnValidThread();
+  if (!UpdateWorkQueues())
+    return;
+
+  size_t queue_index;
+  if (!selector_->SelectWorkQueueToService(&queue_index))
+    return;
+  PostDoWorkOnMainRunner();
+  RunTaskFromWorkQueue(queue_index);
+}
+
+void TaskQueueManager::RunTaskFromWorkQueue(size_t queue_index) {
+  main_thread_checker_.CalledOnValidThread();
+  InternalTaskQueue* queue = Queue(queue_index);
+  DCHECK(!queue->work_queue.empty());
+  base::PendingTask pending_task = queue->work_queue.front();
+  queue->work_queue.pop();
+  task_annotator_.RunTask(
+      "TaskQueueManager::PostTask", "TaskQueueManager::RunTask", pending_task);
+}
+
+bool TaskQueueManager::RunsTasksOnCurrentThread() const {
+  return main_task_runner_->RunsTasksOnCurrentThread();
+}
+
+bool TaskQueueManager::PostDelayedTask(
+    size_t queue_index,
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta delay) {
+  int sequence_num = task_sequence_num_.GetNext();
+
+  base::PendingTask pending_task(from_here, task);
+  pending_task.sequence_num = sequence_num;
+
+  task_annotator_.DidQueueTask("TaskQueueManager::PostTask", pending_task);
+  if (delay > base::TimeDelta()) {
+    return main_task_runner_->PostDelayedTask(
+        from_here,
+        Bind(&TaskQueueManager::EnqueueTask,
+             weak_factory_.GetWeakPtr(),
+             queue_index,
+             pending_task),
+        delay);
+  }
+  EnqueueTask(queue_index, pending_task);
+  return true;
+}
+
+bool TaskQueueManager::PostNonNestableDelayedTask(
+    size_t queue_index,
+    const tracked_objects::Location& from_here,
+    const base::Closure& task,
+    base::TimeDelta delay) {
+  // Defer non-nestable work to the main task runner.
+  return main_task_runner_->PostNonNestableDelayedTask(from_here, task, delay);
+}
+
+}  // namespace content