content: Add task queue manager

base/task/task_queue_manager.cc

Index: base/task/task_queue_manager.cc
diff --git a/base/task/task_queue_manager.cc b/base/task/task_queue_manager.cc
new file mode 100644
index 0000000000000000000000000000000000000000..30415976bdab53101c8bb61277ef754a20127056
--- /dev/null
+++ b/base/task/task_queue_manager.cc
@@ -0,0 +1,212 @@
+// 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 "base/task/task_queue_manager.h"
+
+#include "base/bind.h"
+#include "base/debug/trace_event.h"
+#include "base/task/task_queue_selector.h"
+
+namespace base {
+
+TaskQueueManager::TaskRunner::TaskRunner(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 {
+  return task_queue_manager_->RunsTasksOnCurrentThread();
+}
+
+bool TaskQueueManager::TaskRunner::PostDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& task,
+    TimeDelta delay) {
+  return task_queue_manager_->PostDelayedTask(
+      queue_index_, from_here, task, delay);
+}
+
+bool TaskQueueManager::TaskRunner::PostNonNestableDelayedTask(
+    const tracked_objects::Location& from_here,
+    const Closure& task,
+    TimeDelta delay) {
+  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<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(this, i));
+    queues_.push_back(queue.release());
+  }
+
+  std::vector<const 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<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;
+  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());
+  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 PendingTask& pending_task) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  AutoLock lock(queue->incoming_queue_lock);
+  if (queue->auto_pump && queue->incoming_queue.empty())
+    ScheduleWork();
+  queue->incoming_queue.push(pending_task);
+}
+
+void TaskQueueManager::SetAutoPump(size_t queue_index, bool auto_pump) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  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())
+    ScheduleWork();
+}
+
+void TaskQueueManager::PumpQueue(size_t queue_index) {
+  InternalTaskQueue* queue = Queue(queue_index);
+  AutoLock lock(queue->incoming_queue_lock);
+  PumpQueueLocked(queue);
+}
+
+void TaskQueueManager::ScheduleWork() {
+  main_task_runner_->PostTask(
+      FROM_HERE, Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr()));
+}
+
+bool TaskQueueManager::ReloadWorkQueues() {

[rmcilroy, 2014/10/15 13:02:45]

I was going to suggest renaming this function to ReloadAutoPumpedWorkQueues,
since this function only does reloads for auto pumped queues, but it also
returns true if there is work in any of the queues (including manual pumped
queues), which makes it a bit confusing.  How about UpdateWorkQueues (which is a
bit less specific about whether it reloads the queues or not). 

[Sami, 2014/10/15 14:34:30]

UpdateWorkQueues sgtm. I've now documented what the function actually does.

+  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::DoWork() {
+  main_thread_checker_.CalledOnValidThread();
+  if (!ReloadWorkQueues())
+    return;
+
+  size_t queue_index;
+  if (!selector_->SelectWorkQueueToService(&queue_index))
+    return;
+  ScheduleWork();
+  RunTaskFromWorkQueue(queue_index);
+}
+
+void TaskQueueManager::RunTaskFromWorkQueue(size_t queue_index) {
+  main_thread_checker_.CalledOnValidThread();
+  InternalTaskQueue* queue = Queue(queue_index);
+  DCHECK(!queue->work_queue.empty());
+  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 Closure& task,
+    TimeDelta delay) {
+  int sequence_num = task_sequence_num_.GetNext();
+
+  PendingTask pending_task(from_here, task);
+  pending_task.sequence_num = sequence_num;
+
+  task_annotator_.DidQueueTask("TaskQueueManager::PostTask", pending_task);
+  if (delay > 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 Closure& task,
+    TimeDelta delay) {
+  // Defer non-nestable work to the main task runner.
+  return main_task_runner_->PostNonNestableDelayedTask(from_here, task, delay);
+}
+
+}  // namespace content