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 {
+namespace internal {
+
+class TaskRunner : public base::SingleThreadTaskRunner {
+ public:
+  TaskRunner(base::WeakPtr<TaskQueueManager> task_queue_manager,
+             size_t queue_index);
+
+  // base::SingleThreadTaskRunner implementation.
+  virtual bool RunsTasksOnCurrentThread() const override;
+  virtual bool PostDelayedTask(const tracked_objects::Location& from_here,
+                               const base::Closure& task,
+                               base::TimeDelta delay) override;
+  virtual bool PostNonNestableDelayedTask(
+      const tracked_objects::Location& from_here,
+      const base::Closure& task,
+      base::TimeDelta delay) override;
+
+ private:
+  virtual ~TaskRunner();
+
+  base::WeakPtr<TaskQueueManager> task_queue_manager_;

[cpu_(ooo_6.6-7.5), 2014/10/20 20:00:10]

why is this weak?

[Sami, 2014/10/21 10:32:54]

It's needed to avoid posting new tasks after the scheduler has shut down.
SingleThreadTaskRunner is refcounted, so we can't guarantee that the queue
manager outlives it.

[cpu_(ooo_6.6-7.5), 2014/10/21 19:46:07]

Should the QM take a reference instead?

[Sami, 2014/10/22 12:09:08]

I'm not sure I got that -- QM already holds a reference to the task runner
(although it stricly doesn't need one). The basic problem is that TaskRunner is
refcounted (through inheritance) and QM isn't. This means QM may get deleted
before all the references to the task runners have been cleared, which is what
the weak pointer guards against.

We don't think QM should become refcounted because it is supposed to have a
single owner and a very clear life cycle. Having the weak pointer here is the
simplest way we could think of of achieving this. (The other ideas we had were
just weak pointers in disguise.) Does that make sense?

+  const size_t queue_index_;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskRunner);
+};
+
+TaskRunner::TaskRunner(base::WeakPtr<TaskQueueManager> task_queue_manager,
+                       size_t queue_index)
+    : task_queue_manager_(task_queue_manager), queue_index_(queue_index) {
+}
+
+TaskRunner::~TaskRunner() {
+}
+
+bool TaskRunner::RunsTasksOnCurrentThread() const {
+  if (!task_queue_manager_)
+    return false;
+  return task_queue_manager_->RunsTasksOnCurrentThread();
+}
+
+bool 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 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);
+}

[cpu_(ooo_6.6-7.5), 2014/10/20 20:00:10]

It would seem at first blush that the only value that TaskRunner adds to the
thing is the queue_index ? that is, it feels that we could not have the
taskrunner at all and just have the TQM expose PostNonNestableDelayedTask et al
forcing the client to remember the queue number?

I am not totally set against it, but seems a lot of work for just a number,
specially if the clients are going to be modified:

touch_task_runner->PostTask(...)   vs  task_runner->PostTask(TOUCH_TASK, ...)

[Sami, 2014/10/21 10:32:54]

The reason we went for the TaskRunner was that all the clients that need to
submit tasks to the manager (e.g., cc, the input event filter) are already
parameterized using a SingleThreadTaskRunner. This way we can just hook them up
to the appropriate queue and the knowledge about individual queues doesn't need
to be baked in everywhere.

[cpu_(ooo_6.6-7.5), 2014/10/21 19:46:07]

Acknowledged.

+
+class TaskQueue {
+ public:

[cpu_(ooo_6.6-7.5), 2014/10/21 19:46:07]

this should be a struct.

[Sami, 2014/10/22 12:09:07]

I turned it into a class on Jochen's suggestion, but a struct seems fine too.
Done.

+  TaskQueue() : auto_pump(true) {}
+  ~TaskQueue() {}
+
+  scoped_refptr<TaskRunner> task_runner;
+
+  base::Lock incoming_queue_lock;
+  base::TaskQueue incoming_queue;
+
+  bool auto_pump;
+  base::TaskQueue work_queue;
+
+  DISALLOW_COPY_AND_ASSIGN(TaskQueue);
+};
+
+}  // namespace
+
+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<internal::TaskQueue> queue(new internal::TaskQueue());
+    queue->task_runner = make_scoped_refptr(
+        new internal::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++)

[cpu_(ooo_6.6-7.5), 2014/10/21 19:46:07]

use range for, that has been approved, right?

[Sami, 2014/10/22 12:09:08]

Ah, that's true. Thanks, done.

+    work_queues.push_back(&queues_[i]->work_queue);
+  selector_->RegisterWorkQueues(work_queues);
+}
+
+TaskQueueManager::~TaskQueueManager() {
+}
+
+internal::TaskQueue* 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) {

[cpu_(ooo_6.6-7.5), 2014/10/21 19:46:07]

what is the use case for poll?  wouldn't we want to reload anyhow?

[Sami, 2014/10/22 12:09:08]

The use case is checking whether new high priority tasks are pending either in
the work or incoming queues so that the foreground task can choose to yield back
to the message loop. We do this, for example, with the HTML parser so it knows
when to give room for other work.

You're right that we'll end up reloading anyway on the next iteration of the
message loop. We thought about this and still prefer to expose the operation as
a poll (now called IsQueueEmpty), because that's what the client actually wants
to do instead of reloading anything.

+  internal::TaskQueue* 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();
+  internal::TaskQueue* 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) {
+  internal::TaskQueue* queue = Queue(queue_index);
+  base::AutoLock lock(queue->incoming_queue_lock);
+  if (queue->auto_pump && queue->incoming_queue.empty())
+    PostDoWorkOnMainRunner();

[jar (doing other things), 2014/10/28 22:33:41]

In general, it is a bad idea to do much work while a lock is held, and more
often than not, an error to call out to an arbitrary function while holding a
lock.

Is there a reason why you need the lock held when calling this function.

[Sami, 2014/10/29 11:36:19]

You're right, PostDoWorkOnMainRunner() doesn't need any protection from the
lock. I'll write a CL to move it outside the lock scope.

[Sami, 2014/10/29 12:49:08]

On further reflection, we do need to hold the lock because it prevents the
TaskQueueManager from getting deleted from under us as we're posting the task.
This code has been refactored since this patch to make the locking dependencies
more explicit.

[jar (doing other things), 2014/10/29 21:14:07]

Can you point to how the lock acquisition precludes destruction?

If that were the only thing blocking destruction... then couldn't destruction
happen between line 145 and 146??  ...and wouldn't that cause problems on line
146 when trying to lock?

[Sami, 2014/10/30 10:43:00]

The locking wasn't quite right in this version of the patch. See here:
https://code.google.com/p/chromium/codesearch#chromium/src/content/renderer/s...

Each queue is notified about TQM going away in its destructor. This notification
is protected by the same lock that is held while PostDoWorkOnMainRunner() is
called.

+  queue->incoming_queue.push(pending_task);
+}
+
+void TaskQueueManager::SetAutoPump(size_t queue_index, bool auto_pump) {
+  internal::TaskQueue* 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(internal::TaskQueue* queue) {
+  queue->incoming_queue_lock.AssertAcquired();
+  while (!queue->incoming_queue.empty()) {
+    queue->work_queue.push(queue->incoming_queue.front());

[jar (doing other things), 2014/10/28 22:33:41]

I'm not sure if I'm understanding this full context... but...

The act of pumping one-task-at-a-time gives up some super valuable results of
the current message loop / message pump system.

The current system:
a) Gets faster and faster at servicing a queue the further the queue falls
behind (the larger the incoming queue).
b) Provides asymptotically zero-locking-cost to the thread running the tasks as
the queue gets further and further behind (as the number of unserviced tasks
grows).

The above system level features were critical for precluding a large class of
denial of service attacks (I was really proud of resolving them).

Can you comment on why the above is not an issue for you?

...or...

Could you consider other mechanisms for processing the external queue??  For
instance, if you're not willing to do the swap (loading the whole external queue
into a thread-specific queue each time the thread specific queue empties), would
would consider (for example) loading half of the external queue (rounded up)
each time you try to pump?

Is there a reason why the above is bad?

[Sami, 2014/10/29 11:36:19]

First, thanks for shedding some light on the reasoning behind the message loop's
swapping logic. It's easy to see how the mechanism works by reading the code,
but the *why* is always more interesting.

In our implementation we actually do full queue swaps most of the time (see
ReloadWorkQueue). We call this mode automatic pumping, and it is the default for
all but one of the queues. The exception is the idle task queue, which we want
to pump exactly once a frame. This lets us synchronize idle task execution with
graphics frame rendering. In other words, PumpQueueLocked will get called once
per frame instead of once per posted task, which makes the locking cost
constant.

That said, there is a higher level issue here too: because we need to coexist
with the base message loop, we're only letting ourselves run exactly one task
every time our scheduling function (DoWork) runs. This isn't optimal because it
increases the number of queue swaps done in the message loop. I think we could
experiment with running N tasks (where N is a small constant) per iteration to
reduce this overhead, but for now N = 1 seems like a safer way to get started.

[jar (doing other things), 2014/10/29 21:14:07]

Acknowledged.

+    queue->incoming_queue.pop();
+  }
+  if (!queue->work_queue.empty())
+    PostDoWorkOnMainRunner();
+}
+
+void TaskQueueManager::PumpQueue(size_t queue_index) {
+  internal::TaskQueue* 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();
+  internal::TaskQueue* 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()) {

[cpu_(ooo_6.6-7.5), 2014/10/21 19:46:07]

why is the logic here different from the logic in PostNonNestableDelayedTask ?

[Sami, 2014/10/22 12:09:07]

We want to defer any non-nestable work directly to the main message loop because
it is rare enough (as far as I know) that how it is scheduled doesn't make a
huge difference, and doing anything else would complicate the task queue manager
significantly because we'd need to add a separate deferred work queue for it.

+    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