Run task queue manager work in batches

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 "base/debug/trace_event_argument.h"
 #include "content/renderer/scheduler/task_queue_selector.h"
@@ skipping 227 matching lines @@
 
 }  // namespace internal
 
 TaskQueueManager::TaskQueueManager(
     size_t task_queue_count,
     scoped_refptr<base::SingleThreadTaskRunner> main_task_runner,
     TaskQueueSelector* selector)
     : main_task_runner_(main_task_runner),
       selector_(selector),
       pending_dowork_count_(0),
+      work_batch_size_(1),
       weak_factory_(this) {
   DCHECK(main_task_runner->RunsTasksOnCurrentThread());
   TRACE_EVENT_OBJECT_CREATED_WITH_ID(
       TRACE_DISABLED_BY_DEFAULT("renderer.scheduler"), "TaskQueueManager",
       this);
 
   task_queue_manager_weak_ptr_ = weak_factory_.GetWeakPtr();
   for (size_t i = 0; i < task_queue_count; i++) {
     scoped_refptr<internal::TaskQueue> queue(
         make_scoped_refptr(new internal::TaskQueue(this)));
@@ skipping 67 matching lines @@
       FROM_HERE, Bind(&TaskQueueManager::DoWork, task_queue_manager_weak_ptr_,
                       on_main_thread));
 }
 
 void TaskQueueManager::DoWork(bool posted_from_main_thread) {
   if (posted_from_main_thread) {
     pending_dowork_count_--;
     DCHECK_GE(pending_dowork_count_, 0);
   }
   main_thread_checker_.CalledOnValidThread();
-  if (!UpdateWorkQueues())
-    return;
 
-  size_t queue_index;
-  if (!SelectWorkQueueToService(&queue_index))
-    return;
-  MaybePostDoWorkOnMainRunner();
-  ProcessTaskFromWorkQueue(queue_index);
+  for (int i = 0; i < work_batch_size_; i++) {
+    if (!UpdateWorkQueues())
+      return;
+
+    size_t queue_index;
+    if (!SelectWorkQueueToService(&queue_index))
+      return;
+    if (i == 0)
+      MaybePostDoWorkOnMainRunner();
+    ProcessTaskFromWorkQueue(queue_index);
+  }

[rmcilroy, 2015/01/14 17:56:37]

Would the following work:

if (!UpdateWorkQueues())
  return;
MaybePostDoWorkOnMainRunner();

for (int i = 0; i < work_batch_size_; i++) {
  if (!SelectWorkQueueToService(&queue_index))
     return;
  ProcessTaskFromWorkQueue(queue_index);
}

It's a bit simpler, and ensures that we only take the locks in UpdateWorkQueues
once per batch, but does mean that that a given batch won't include any work
posted from any DoWorks for that batch. Would this prevent any batching being
possible on the indexdb perf test?

[Sami, 2015/01/14 18:11:49]

I had the same idea and chose to do polling inside the batch to minimize the
chances that we yield and then immediately come back since there was more work.
The idea was simply that it's more likely there's more work waiting for us
inside the task queue manager than outside of it. I can definitely see the point
about taking fewer locks if there really isn't anything to do. 

Let me check how this fares with the indexeddb benchmark.

On 2015/01/14 17:56:37, rmcilroy wrote:
I don't think we can do this here because the selector should be able to prevent
any DoWorks from getting posted if it wants.
Here we need to check if any of the queues became empty and bail out.

[rmcilroy, 2015/01/14 19:34:48]

Good point
Ack
Ahh that's sad, but having thought about it a bit more it probably wouldn't be a
great plan anyway because without updating the work queues every cycle the
selector wouldn't necessarily make the correct decision.

 }
 
 bool TaskQueueManager::SelectWorkQueueToService(size_t* out_queue_index) {
   bool should_run = selector_->SelectWorkQueueToService(out_queue_index);
   TRACE_EVENT_OBJECT_SNAPSHOT_WITH_ID(
       TRACE_DISABLED_BY_DEFAULT("renderer.scheduler"), "TaskQueueManager", this,
       AsValueWithSelectorResult(should_run, *out_queue_index));
   return should_run;
 }
 
@@ skipping 28 matching lines @@
   DCHECK(delay > base::TimeDelta());
   return main_task_runner_->PostDelayedTask(from_here, task, delay);
 }
 
 void TaskQueueManager::SetQueueName(size_t queue_index, const char* name) {
   main_thread_checker_.CalledOnValidThread();
   internal::TaskQueue* queue = Queue(queue_index);
   queue->set_name(name);
 }
 
+void TaskQueueManager::SetWorkBatchSize(int work_batch_size) {
+  main_thread_checker_.CalledOnValidThread();
+  DCHECK_GE(work_batch_size, 1);
+  work_batch_size_ = work_batch_size;
+}
+
 scoped_refptr<base::debug::ConvertableToTraceFormat>
 TaskQueueManager::AsValueWithSelectorResult(bool should_run,
                                             size_t selected_queue) const {
   main_thread_checker_.CalledOnValidThread();
   scoped_refptr<base::debug::TracedValue> state =
       new base::debug::TracedValue();
   state->BeginArray("queues");
   for (auto& queue : queues_)
     queue->AsValueInto(state.get());
   state->EndArray();
   state->BeginDictionary("selector");
   selector_->AsValueInto(state.get());
   state->EndDictionary();
   if (should_run)
     state->SetInteger("selected_queue", selected_queue);
   return state;
 }
 
 }  // namespace content