[scheduler] Add TaskQueue::Observer

third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc

Index: third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc
diff --git a/third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc b/third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc
index f0758bae6e5793818902f518ec407e5c2d5a4072..68369b2d71f4c82b6b866ba7bdd2c1ac75a7a7f8 100644
--- a/third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc
+++ b/third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc
@@ -103,6 +103,8 @@ TaskQueueImpl::~TaskQueueImpl() {
   DCHECK(any_thread().task_queue_manager == nullptr)
       << "UnregisterTaskQueue must be called first!";
 
+  DCHECK(any_thread().observer == nullptr)
+      << "TaskQueue::Observer must be unregistered manually!";
 #endif
 }
 
@@ -147,7 +149,9 @@ TaskQueueImpl::Task::Task(const tracked_objects::Location& posted_from,
 
 TaskQueueImpl::AnyThread::AnyThread(TaskQueueManager* task_queue_manager,
                                     TimeDomain* time_domain)
-    : task_queue_manager(task_queue_manager), time_domain(time_domain) {}
+    : task_queue_manager(task_queue_manager),
+      time_domain(time_domain),
+      observer(nullptr) {}
 
 TaskQueueImpl::AnyThread::~AnyThread() {}
 
@@ -157,6 +161,7 @@ TaskQueueImpl::MainThreadOnly::MainThreadOnly(
     TimeDomain* time_domain)
     : task_queue_manager(task_queue_manager),
       time_domain(time_domain),
+      observer(nullptr),
       delayed_work_queue(
           new WorkQueue(task_queue, "delayed", WorkQueue::QueueType::DELAYED)),
       immediate_work_queue(new WorkQueue(task_queue,
@@ -183,6 +188,8 @@ void TaskQueueImpl::UnregisterTaskQueue() {
 
   any_thread().task_queue_manager = nullptr;
   main_thread_only().task_queue_manager = nullptr;
+  any_thread().observer = nullptr;
+  main_thread_only().observer = nullptr;
   main_thread_only().delayed_incoming_queue = std::priority_queue<Task>();
   immediate_incoming_queue().clear();
   main_thread_only().immediate_work_queue.reset();
@@ -277,19 +284,15 @@ bool TaskQueueImpl::PostDelayedTaskImpl(
 
 void TaskQueueImpl::PushOntoDelayedIncomingQueueFromMainThread(
     Task pending_task, base::TimeTicks now) {
-  base::TimeTicks delayed_run_time = pending_task.delayed_run_time;
+  DelayedWakeUp wakeup = pending_task.delayed_wakeup();
   main_thread_only().task_queue_manager->DidQueueTask(pending_task);
   main_thread_only().delayed_incoming_queue.push(std::move(pending_task));
 
-  // If |pending_task| is at the head of the queue, then make sure a wakeup
-  // is requested if the queue is enabled.  Note we still want to schedule a
-  // wakeup even if blocked by a fence, because we'd break throttling logic
-  // otherwise.
-  base::TimeTicks next_delayed_task =
-      main_thread_only().delayed_incoming_queue.top().delayed_run_time;
-  if (next_delayed_task == delayed_run_time && IsQueueEnabled()) {
-    main_thread_only().time_domain->ScheduleDelayedWork(
-        this, {delayed_run_time, pending_task.sequence_num}, now);
+  DelayedWakeUp new_wakeup =
+      main_thread_only().delayed_incoming_queue.top().delayed_wakeup();
+  if (wakeup.time == new_wakeup.time &&
+      wakeup.sequence_num == new_wakeup.sequence_num) {
+    TimeDomainScheduleDelayedWork(now);
   }
 
   TraceQueueSize();
@@ -358,7 +361,8 @@ void TaskQueueImpl::PushOntoImmediateIncomingQueueLocked(
         (!IsQueueEnabled() || main_thread_only().current_fence);
     any_thread().task_queue_manager->OnQueueHasIncomingImmediateWork(
         this, sequence_number, queue_is_blocked);
-    any_thread().time_domain->OnQueueHasImmediateWork(this);
+    if (any_thread().observer)
+      any_thread().observer->OnQueueNextWakeupChanged(this, desired_run_time);
   }
 
   TraceQueueSize();
@@ -449,9 +453,7 @@ TaskQueueImpl::WakeUpForDelayedWork(LazyNow* lazy_now) {
 
   // Make sure the next wake up is scheduled.
   if (!main_thread_only().delayed_incoming_queue.empty()) {
-    return DelayedWakeUp{
-        main_thread_only().delayed_incoming_queue.top().delayed_run_time,
-        main_thread_only().delayed_incoming_queue.top().sequence_num};
+    return main_thread_only().delayed_incoming_queue.top().delayed_wakeup();
   }
 
   return base::nullopt;
@@ -598,13 +600,7 @@ void TaskQueueImpl::SetTimeDomain(TimeDomain* time_domain) {
   main_thread_only().time_domain = time_domain;
   time_domain->RegisterQueue(this);
 
-  if (IsQueueEnabled() && !main_thread_only().delayed_incoming_queue.empty()) {
-    time_domain->ScheduleDelayedWork(
-        this,
-        {main_thread_only().delayed_incoming_queue.top().delayed_run_time,
-         main_thread_only().delayed_incoming_queue.top().sequence_num},
-        time_domain->Now());
-  }
+  TimeDomainScheduleDelayedWork(time_domain->Now());
 }
 
 TimeDomain* TaskQueueImpl::GetTimeDomain() const {
@@ -816,24 +812,11 @@ void TaskQueueImpl::EnableOrDisableWithSelector(bool enable) {
     return;
 
   if (enable) {
-    // Check if there's any immediate work on either queue.
-    bool immediate_queues_empty =
-        main_thread_only().immediate_work_queue->Empty();
-    if (immediate_queues_empty) {
-      base::AutoLock lock(immediate_incoming_queue_lock_);
-      immediate_queues_empty = immediate_incoming_queue().empty();
-    }
-    // Avoid holding the lock while we fire the notification.
-    if (!immediate_queues_empty)
-      main_thread_only().time_domain->OnQueueHasImmediateWork(this);
-
-    if (!main_thread_only().delayed_incoming_queue.empty()) {
-      main_thread_only().time_domain->ScheduleDelayedWork(
-          this,
-          {main_thread_only().delayed_incoming_queue.top().delayed_run_time,
-           main_thread_only().delayed_incoming_queue.top().sequence_num},
-          main_thread_only().time_domain->Now());
-    }
+    if (HasPendingImmediateWork())
+      NotifyWakeupChangedOnMainThread(base::TimeTicks());
+
+    TimeDomainScheduleDelayedWork(main_thread_only().time_domain->Now());
+
     // Note the selector calls TaskQueueManager::OnTaskQueueEnabled which posts
     // a DoWork if needed.
     main_thread_only().task_queue_manager->selector_.EnableQueue(this);
@@ -875,13 +858,7 @@ void TaskQueueImpl::SweepCanceledDelayedTasks(base::TimeTicks now) {
     main_thread_only().time_domain->CancelDelayedWork(this);
   } else if (first_task_runtime !=
              main_thread_only().delayed_incoming_queue.top().delayed_run_time) {
-    if (IsQueueEnabled()) {
-      main_thread_only().time_domain->ScheduleDelayedWork(
-          this,
-          {main_thread_only().delayed_incoming_queue.top().delayed_run_time,
-           main_thread_only().delayed_incoming_queue.top().sequence_num},
-          main_thread_only().time_domain->Now());
-    }
+    TimeDomainScheduleDelayedWork(main_thread_only().time_domain->Now());
   }
 }
 
@@ -891,6 +868,39 @@ void TaskQueueImpl::PushImmediateIncomingTaskForTest(
   immediate_incoming_queue().push_back(std::move(task));
 }
 
+void TaskQueueImpl::AddObserver(Observer* observer) {

[Sami, 2017/04/04 19:07:29]

These should probably be just SetObserver since you can't add multiple ones.

[altimin, 2017/04/05 11:35:58]

Done.

+  base::AutoLock lock(any_thread_lock_);
+  any_thread().observer = observer;
+  main_thread_only().observer = observer;
+}
+
+void TaskQueueImpl::RemoveObserver(Observer* observer) {
+  base::AutoLock lock(any_thread_lock_);
+  DCHECK(any_thread().observer == observer);
+  any_thread().observer = nullptr;
+  main_thread_only().observer = nullptr;
+}
+
+void TaskQueueImpl::TimeDomainScheduleDelayedWork(base::TimeTicks now) {

[Sami, 2017/04/04 19:07:30]

ScheduleDelayedWorkInTimeDomain to make this read a bit better?

[altimin, 2017/04/05 11:35:58]

Done.

+  if (!IsQueueEnabled())
+    return;
+  if (main_thread_only().delayed_incoming_queue.empty())
+    return;
+
+  main_thread_only().time_domain->ScheduleDelayedWork(
+      this, main_thread_only().delayed_incoming_queue.top().delayed_wakeup(),
+      now);
+
+  if (!HasPendingImmediateWork())

[Sami, 2017/04/04 19:07:29]

nit: braces please

[altimin, 2017/04/05 11:35:58]

Done.

+    NotifyWakeupChangedOnMainThread(
+        main_thread_only().delayed_incoming_queue.top().delayed_run_time);
+}
+
+void TaskQueueImpl::NotifyWakeupChangedOnMainThread(base::TimeTicks wakeup) {
+  if (main_thread_only().observer)
+    main_thread_only().observer->OnQueueNextWakeupChanged(this, wakeup);
+}
+
 }  // namespace internal
 }  // namespace scheduler
 }  // namespace blink