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..b9c518e427e14dc922516e4ebc188789632bc7ce
--- /dev/null
+++ b/base/task/task_queue_manager.cc
@@ -0,0 +1,207 @@
+// 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_scheduler.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(size_t index)
+    : index(index), auto_pump(true) {
+}

[picksi1, 2014/10/15 09:07:38]

Should auto_pump be called pump_when_empty?

[Sami, 2014/10/15 12:05:24]

No, because it also involves how DoWork is scheduled. It's not just about work
queue management, which is why we had to think long and hard about its name :)

+
+TaskQueueManager::InternalTaskQueue::~InternalTaskQueue() {
+}
+
+TaskQueueManager::TaskQueueManager(
+    size_t task_queue_count,
+    scoped_refptr<SingleThreadTaskRunner> main_task_runner,
+    TaskQueueScheduler* scheduler)
+    : main_task_runner_(main_task_runner),
+      scheduler_(scheduler),
+      weak_factory_(this) {
+  DCHECK(main_task_runner->RunsTasksOnCurrentThread());
+
+  for (size_t i = 0; i < task_queue_count; i++) {
+    scoped_ptr<InternalTaskQueue> queue(new InternalTaskQueue(i));
+    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);
+  scheduler_->RegisterWorkQueues(work_queues);

[picksi1, 2014/10/15 09:07:38]

Are we exposing our inner workings (i.e. the work_queue) to the scheduler doing
this? Would it be better (ironically!) to pass an InternalTaskQueue to the
scheduler and when a task is posted via the Post function (presumably in this
file?) it would (internally) use the work_queue?

[Sami, 2014/10/15 12:05:24]

I'm not sure I follow the last half of your suggestion, but as to the first part
the selector needs some way to know what work it can select from at the moment
it's called. I'm not sure it needs to know about the work queues specifically,
but just the same information. I guess we could rename lie that they are
"pending" task queues or something, but I'm not sure that abstraction buys us
much. We already need to be a little explicit about the inner workings so that
pumping makes sense.

+}
+
+TaskQueueManager::~TaskQueueManager() {
+}
+
+scoped_refptr<SingleThreadTaskRunner> TaskQueueManager::TaskRunnerForQueue(
+    size_t queue_index) {
+  DCHECK_LT(queue_index, queues_.size());
+  return queues_[queue_index]->task_runner;
+}
+
+bool TaskQueueManager::PollQueue(size_t queue_index) {
+  DCHECK_LT(queue_index, queues_.size());
+  if (!queues_[queue_index]->work_queue.empty())
+    return true;
+  InternalTaskQueue* queue = queues_[queue_index];

[picksi1, 2014/10/15 09:07:38]

Getting a queue from an index is repeated a lot, should we turn it into a
function that can (a) be future proofed in case we change the implementation (b)
allow us to gather the DCHECK_LT's into the same place.

[Sami, 2014/10/15 12:05:24]

Good idea, done.

+  AutoLock lock(queue->incoming_queue_lock);
+  return !queue->incoming_queue.empty();

[petrcermak, 2014/10/15 10:05:40]

Shouldn't this be "return queue->auto_pump && !queue->incoming_queue.empty();"?
How else will the TaskQueueScheduler know that it cannot schedule from the
corresponding queue?

[alexclarke, 2014/10/15 10:09:14]

The TaskQueueScheduler is responsible for setting the state of auto_pump. 
Problem with your suggestion is the TaskQueueScheduler would no longer be able
to know when the idle queue is empty.

[petrcermak, 2014/10/15 10:13:05]

Acknowledged.

+}
+
+bool TaskQueueManager::ReloadWorkQueue(size_t queue_index) {
+  main_thread_checker_.CalledOnValidThread();
+  DCHECK_LT(queue_index, queues_.size());
+  InternalTaskQueue* queue = queues_[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) {
+  DCHECK_LT(queue_index, queues_.size());
+  InternalTaskQueue* queue = queues_[queue_index];
+  AutoLock lock(queue->incoming_queue_lock);
+  if (queue->auto_pump && queue->incoming_queue.empty())
+    ScheduleWork();
+  queue->incoming_queue.push(pending_task);

[picksi1, 2014/10/15 09:07:38]

Is it possible for the incoming queue to be empty but there still to be an
instance of DoWork queued on the main thread? Do we need to have the equvilant
of a 'task runner posted' bool that is set true in scheduleWork and false in
DoWork?

[Sami, 2014/10/15 12:05:24]

Yes, that's possible. This is written to be slightly eager about posting more
DoWork's than are strictly necessary to decrease our changes of getting pushed
to the back of the MessageLoop queue by unrelated things. Because DoWork only
ever runs one task at most I think this should be safe w.r.t. starvation.

A bool like that would work but would need atomic reads and writes, so I'd
prefer to avoid it if we can especially since we're already taking a lock here.

+}
+
+void TaskQueueManager::SetAutoPump(size_t queue_index, bool auto_pump) {
+  DCHECK_LT(queue_index, queues_.size());
+  InternalTaskQueue* queue = queues_[queue_index];
+  {
+    AutoLock lock(queue->incoming_queue_lock);
+    queue->auto_pump = auto_pump;
+  }
+  if (auto_pump)
+    PumpQueue(queue_index);
+}

[picksi1, 2014/10/15 09:07:38]

Would you consider having two different functions here; TurnAutoPumpOn and
TurnAutoPumpOff? It's that last 'if' that makes me squirm a little!

[Sami, 2014/10/15 12:05:24]

Hmm, having two different functions makes this a little more messy to use from
the client's side. However let me see how that would look on the implementation
side. I think it would let use avoid some of the locks.

Edit: Okay I tried this and I still prefer to have all the logic in one place.
Having everything spread out in one-line functions makes it really difficult to
see what is going on. I've still reworked this a bit to make it clearer though.

+
+void TaskQueueManager::PumpQueue(size_t queue_index) {
+  DCHECK_LT(queue_index, queues_.size());
+  InternalTaskQueue* queue = queues_[queue_index];
+  if (queue->work_queue.empty()) {
+    AutoLock lock(queue->incoming_queue_lock);
+    queue->work_queue.Swap(&queue->incoming_queue);
+  }
+  if (!queue->work_queue.empty())
+    ScheduleWork();
+}
+
+void TaskQueueManager::ScheduleWork() {
+  main_task_runner_->PostTask(
+      FROM_HERE, Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr()));
+}

[picksi1, 2014/10/15 09:07:38]

Is ScheduleWork the right name? Should it be more like
PostTaskQueueManagerExecution ... or something?

[Sami, 2014/10/15 12:05:24]

ScheduleWork/DoWork seems like a reasonable pairing to me, no? Posting is just
an implementation detail.

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

I somewhat agree with Simon that it's difficult to work out from the name
whether this schedules work here or posts it for scheduling.  I also think
"schedule" shouldn't be used since this function doesn't actually do any
scheduling decisions.  I would prefer PostDoWorkOnMainRunner or similar.

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

Ah, I was hoping it was now more clear since we have a selector instead of a
scheduler. Happy to concede defeat and go with PostDoWorkOnMainRunner :) Done.

+
+bool TaskQueueManager::ReloadWorkQueues() {
+  bool has_work = false;
+  for (auto& queue : queues_) {
+    if (!queue->work_queue.empty())
+      has_work = true;
+    else if (ReloadWorkQueue(queue->index))
+      has_work = true;
+  }
+  return has_work;
+}
+
+void TaskQueueManager::DoWork() {
+  main_thread_checker_.CalledOnValidThread();
+  if (!ReloadWorkQueues())
+    return;
+
+  size_t queue_index;
+  if (!scheduler_->ChooseWorkQueueToService(&queue_index))
+    return;
+  ScheduleWork();
+  RunTaskFromWorkQueue(queue_index);
+}
+
+void TaskQueueManager::RunTaskFromWorkQueue(size_t queue_index) {
+  main_thread_checker_.CalledOnValidThread();
+  DCHECK_LT(queue_index, queues_.size());
+  InternalTaskQueue* queue = queues_[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