scheduler: Implement task observers

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 <queue>
 
 #include "base/bind.h"
 #include "base/trace_event/trace_event.h"
@@ skipping 401 matching lines @@
     DCHECK_GE(pending_dowork_count_, 0);
   }
   DCHECK(main_thread_checker_.CalledOnValidThread());
 
   base::TimeTicks next_pending_delayed_task(
       base::TimeTicks::FromInternalValue(kMaxTimeTicks));
 
   if (!UpdateWorkQueues(&next_pending_delayed_task, BEFORE_WAKEUP_EVENT_TYPE))
     return;
 
+  base::PendingTask previous_task((tracked_objects::Location()),
+                                  (base::Closure()));
   for (int i = 0; i < work_batch_size_; i++) {
     // Interrupt the work batch if we should run the next delayed task.
     if (i > 0 && next_pending_delayed_task.ToInternalValue() != kMaxTimeTicks &&
         Now() >= next_pending_delayed_task)
       return;
 
     size_t queue_index;
     if (!SelectWorkQueueToService(&queue_index))
       return;
     // Note that this function won't post another call to DoWork if one is
     // already pending, so it is safe to call it in a loop.
     MaybePostDoWorkOnMainRunner();
-    ProcessTaskFromWorkQueue(queue_index);
+    ProcessTaskFromWorkQueue(queue_index, i > 0, &previous_task);
 
     if (!UpdateWorkQueues(&next_pending_delayed_task, AFTER_WAKEUP_EVENT_TYPE))
       return;
   }
 }
 
 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;
 }
 
 void TaskQueueManager::DidQueueTask(base::PendingTask* pending_task) {
   pending_task->sequence_num = task_sequence_num_.GetNext();
   task_annotator_.DidQueueTask("TaskQueueManager::PostTask", *pending_task);
 }
 
-void TaskQueueManager::ProcessTaskFromWorkQueue(size_t queue_index) {
+void TaskQueueManager::ProcessTaskFromWorkQueue(
+    size_t queue_index,
+    bool has_previous_task,
+    base::PendingTask* previous_task) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   internal::TaskQueue* queue = Queue(queue_index);
   base::PendingTask pending_task = queue->TakeTaskFromWorkQueue();
   if (!pending_task.nestable) {
     // Defer non-nestable work to the main task runner.  NOTE these tasks can be
     // arbitrarily delayed so the additional delay should not be a problem.
     main_task_runner_->PostNonNestableTask(pending_task.posted_from,
                                            pending_task.task);
   } else {
+    // Suppress "will" task observer notifications for the first and "did"
+    // notifications for the last task in the batch to avoid duplicate
+    // notifications.
+    if (has_previous_task) {
+      FOR_EACH_OBSERVER(base::MessageLoop::TaskObserver, task_observers_,
+                        DidProcessTask(*previous_task));
+      FOR_EACH_OBSERVER(base::MessageLoop::TaskObserver, task_observers_,
+                        WillProcessTask(pending_task));
+    }
     task_annotator_.RunTask("TaskQueueManager::PostTask",
                             "TaskQueueManager::RunTask", pending_task);
+    pending_task.task.Reset();
+    *previous_task = pending_task;
   }
 }
 
 bool TaskQueueManager::RunsTasksOnCurrentThread() const {
   return main_task_runner_->RunsTasksOnCurrentThread();
 }
 
 bool TaskQueueManager::PostDelayedTask(
     const tracked_objects::Location& from_here,
     const base::Closure& task,
     base::TimeDelta delay) {
   DCHECK(delay > base::TimeDelta());
   return main_task_runner_->PostDelayedTask(from_here, task, delay);
 }
 
 void TaskQueueManager::SetQueueName(size_t queue_index, const char* name) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   internal::TaskQueue* queue = Queue(queue_index);
   queue->set_name(name);
 }
 
 void TaskQueueManager::SetWorkBatchSize(int work_batch_size) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   DCHECK_GE(work_batch_size, 1);
   work_batch_size_ = work_batch_size;
 }
 
+void TaskQueueManager::AddTaskObserver(
+    base::MessageLoop::TaskObserver* task_observer) {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  base::MessageLoop::current()->AddTaskObserver(task_observer);
+  task_observers_.AddObserver(task_observer);
+}
+
+void TaskQueueManager::RemoveTaskObserver(
+    base::MessageLoop::TaskObserver* task_observer) {
+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  base::MessageLoop::current()->RemoveTaskObserver(task_observer);
+  task_observers_.RemoveObserver(task_observer);
+}
+
 void TaskQueueManager::SetTimeSourceForTesting(
     scoped_refptr<cc::TestNowSource> time_source) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   time_source_ = time_source;
 }
 
 base::TimeTicks TaskQueueManager::Now() const {
   return UNLIKELY(time_source_) ? time_source_->Now() : base::TimeTicks::Now();
 }
 
 scoped_refptr<base::trace_event::ConvertableToTraceFormat>
 TaskQueueManager::AsValueWithSelectorResult(bool should_run,
                                             size_t selected_queue) const {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   scoped_refptr<base::trace_event::TracedValue> state =
       new base::trace_event::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