[scheduler] Add WakeupBudgetPool.

third_party/WebKit/Source/platform/scheduler/renderer/task_queue_throttler.cc

 // Copyright 2015 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 "platform/scheduler/renderer/task_queue_throttler.h"
 
 #include <cstdint>
 
 #include "base/format_macros.h"
 #include "base/logging.h"
@@ skipping 126 matching lines @@
   if (!task_queue->IsEmpty()) {
     LazyNow lazy_now(tick_clock_);
     OnQueueNextWakeUpChanged(task_queue,
                              NextTaskRunTime(&lazy_now, task_queue).value());
   }
 }
 
 void TaskQueueThrottler::DecreaseThrottleRefCount(TaskQueue* task_queue) {
   TaskQueueMap::iterator iter = queue_details_.find(task_queue);
 
-  if (iter == queue_details_.end() ||
-      --iter->second.throttling_ref_count != 0) {
+  if (iter == queue_details_.end())
     return;
-  }
+  if (iter->second.throttling_ref_count == 0)
+    return;
+  if (--iter->second.throttling_ref_count != 0)
+    return;
 
   TRACE_EVENT1(tracing_category_, "TaskQueueThrottler_TaskQueueUnthrottled",
                "task_queue", task_queue);
 
   task_queue->SetObserver(nullptr);
 
   MaybeDeleteQueueMetadata(iter);
 
   if (!allow_throttling_)
     return;
@@ skipping 10 matching lines @@
   if (find_it == queue_details_.end())
     return false;
   return find_it->second.throttling_ref_count > 0;
 }
 
 void TaskQueueThrottler::UnregisterTaskQueue(TaskQueue* task_queue) {
   auto find_it = queue_details_.find(task_queue);
   if (find_it == queue_details_.end())
     return;
 
-  LazyNow lazy_now(tick_clock_);
   std::unordered_set<BudgetPool*> budget_pools = find_it->second.budget_pools;
   for (BudgetPool* budget_pool : budget_pools) {
-    budget_pool->RemoveQueue(lazy_now.Now(), task_queue);
+    budget_pool->UnregisterQueue(task_queue);
   }
 
   // Iterator may have been deleted by BudgetPool::RemoveQueue, so don't
   // use it here.
   queue_details_.erase(task_queue);
 
   // NOTE: Observer is automatically unregistered when unregistering task queue.
 }
 
 void TaskQueueThrottler::OnQueueNextWakeUpChanged(
     TaskQueue* queue,
     base::TimeTicks next_wake_up) {
   if (!task_runner_->RunsTasksOnCurrentThread()) {
     task_runner_->PostTask(
         FROM_HERE,
         base::Bind(forward_immediate_work_callback_, queue, next_wake_up));
     return;
   }
 
   TRACE_EVENT0(tracing_category_,
                "TaskQueueThrottler::OnQueueNextWakeUpChanged");
 
   // We don't expect this to get called for disabled queues, but we can't DCHECK
   // because of the above thread hop.  Just bail out if the queue is disabled.
   if (!queue->IsQueueEnabled())
     return;
 
   base::TimeTicks now = tick_clock_->NowTicks();
+
+  auto find_it = queue_details_.find(queue);
+  if (find_it == queue_details_.end())
+    return;
+
+  for (BudgetPool* budget_pool : find_it->second.budget_pools) {
+    budget_pool->OnTaskQueueHasWork(queue, now, next_wake_up);

[Sami, 2017/04/26 13:09:08]

Could we also call this OnQueueNextWakeUpChanged because the semantic seems to
be the same as this callback and I can't think of a reason why the name should
be different?

[altimin, 2017/04/26 14:31:29]

Done.

+  }
+
+  base::TimeTicks next_allowed_run_time =
+      GetNextAllowedRunTime(queue, next_wake_up);
+  // TODO(altimin): Remove after moving to budget pools completely.

[Sami, 2017/04/26 13:09:08]

Could you expand this comment? I'm not sure what it means.

[altimin, 2017/04/26 14:31:29]

Done.

   MaybeSchedulePumpThrottledTasks(
-      FROM_HERE, now,
-      std::max(GetNextAllowedRunTime(now, queue), next_wake_up));
+      FROM_HERE, now, std::max(next_wake_up, next_allowed_run_time));
 }
 
 void TaskQueueThrottler::PumpThrottledTasks() {
   TRACE_EVENT0(tracing_category_, "TaskQueueThrottler::PumpThrottledTasks");
   pending_pump_throttled_tasks_runtime_.reset();
 
   LazyNow lazy_now(tick_clock_);
   base::Optional<base::TimeTicks> next_scheduled_delayed_task;
 
+  for (const auto& pair : budget_pools_)
+    pair.first->OnWakeUp(lazy_now.Now());
+
   for (const TaskQueueMap::value_type& map_entry : queue_details_) {
     TaskQueue* task_queue = map_entry.first;
-    if (task_queue->IsEmpty() || !IsThrottled(task_queue))
-      continue;
-
-    // Don't enable queues whose budget pool doesn't allow them to run now.
-    base::TimeTicks next_allowed_run_time =
-        GetNextAllowedRunTime(lazy_now.Now(), task_queue);
-    base::Optional<base::TimeTicks> next_desired_run_time =
-        NextTaskRunTime(&lazy_now, task_queue);
-
-    if (next_desired_run_time &&
-        next_allowed_run_time > next_desired_run_time.value()) {
-      TRACE_EVENT1(
-          tracing_category_,
-          "TaskQueueThrottler::PumpThrottledTasks_ExpensiveTaskThrottled",
-          "throttle_time_in_seconds",
-          (next_allowed_run_time - next_desired_run_time.value()).InSecondsF());
-
-      // Schedule a pump for queue which was disabled because of time budget.
-      next_scheduled_delayed_task =
-          Min(next_scheduled_delayed_task, next_allowed_run_time);
-
-      continue;
-    }
-
-    next_scheduled_delayed_task =
-        Min(next_scheduled_delayed_task, task_queue->GetNextScheduledWakeUp());
-
-    if (next_allowed_run_time > lazy_now.Now())
-      continue;
-
-    // Remove previous fence and install a new one, allowing all tasks posted
-    // on |task_queue| up until this point to run and block all further tasks.
-    task_queue->InsertFence(TaskQueue::InsertFencePosition::NOW);
-  }
-
-  // Maybe schedule a call to TaskQueueThrottler::PumpThrottledTasks if there is
-  // a pending delayed task or a throttled task ready to run.
-  // NOTE: posting a non-delayed task in the future will result in
-  // TaskQueueThrottler::OnTimeDomainHasImmediateWork being called.
-  if (next_scheduled_delayed_task) {
-    MaybeSchedulePumpThrottledTasks(FROM_HERE, lazy_now.Now(),
-                                    *next_scheduled_delayed_task);
+    UpdateQueueThrottlingStateInternal(lazy_now.Now(), task_queue, true);
   }
 }
 
 /* static */
 base::TimeTicks TaskQueueThrottler::AlignedThrottledRunTime(
     base::TimeTicks unthrottled_runtime) {
   const base::TimeDelta one_second = base::TimeDelta::FromSeconds(1);
   return unthrottled_runtime + one_second -
          ((unthrottled_runtime - base::TimeTicks()) % one_second);
 }
@@ skipping 29 matching lines @@
 }
 
 CPUTimeBudgetPool* TaskQueueThrottler::CreateCPUTimeBudgetPool(
     const char* name) {
   CPUTimeBudgetPool* time_budget_pool =
       new CPUTimeBudgetPool(name, this, tick_clock_->NowTicks());
   budget_pools_[time_budget_pool] = base::WrapUnique(time_budget_pool);
   return time_budget_pool;
 }
 
+WakeUpBudgetPool* TaskQueueThrottler::CreateWakeUpBudgetPool(const char* name) {
+  WakeUpBudgetPool* wake_up_budget_pool =
+      new WakeUpBudgetPool(name, this, tick_clock_->NowTicks());
+  budget_pools_[wake_up_budget_pool] = base::WrapUnique(wake_up_budget_pool);
+  return wake_up_budget_pool;
+}
+
 void TaskQueueThrottler::OnTaskRunTimeReported(TaskQueue* task_queue,
                                                base::TimeTicks start_time,
                                                base::TimeTicks end_time) {
   if (!IsThrottled(task_queue))
     return;
 
   auto find_it = queue_details_.find(task_queue);
   if (find_it == queue_details_.end())
     return;
 
   for (BudgetPool* budget_pool : find_it->second.budget_pools) {
-    budget_pool->RecordTaskRunTime(start_time, end_time);
-    if (!budget_pool->HasEnoughBudgetToRun(end_time))
-      budget_pool->BlockThrottledQueues(end_time);
+    budget_pool->RecordTaskRunTime(task_queue, start_time, end_time);
   }
 }
 
-void TaskQueueThrottler::BlockQueue(base::TimeTicks now, TaskQueue* queue) {
-  if (!IsThrottled(queue))
+void TaskQueueThrottler::UpdateQueueThrottlingState(base::TimeTicks now,
+                                                    TaskQueue* queue) {
+  UpdateQueueThrottlingStateInternal(now, queue, false);
+}
+
+void TaskQueueThrottler::UpdateQueueThrottlingStateInternal(base::TimeTicks now,
+                                                            TaskQueue* queue,
+                                                            bool is_wake_up) {
+  if (!queue->IsQueueEnabled() || !IsThrottled(queue)) {
     return;
+  }
 
-  queue->InsertFence(TaskQueue::InsertFencePosition::BEGINNING_OF_TIME);
-  SchedulePumpQueue(FROM_HERE, now, queue);
+  LazyNow lazy_now(now);
+
+  base::Optional<base::TimeTicks> next_desired_run_time =
+      NextTaskRunTime(&lazy_now, queue);
+
+  if (!next_desired_run_time) {
+    // This queue is empty. Given that new task can arrive at any moment,
+    // block the queue completely and update the state upon the notification
+    // about a new task.
+    queue->InsertFence(TaskQueue::InsertFencePosition::NOW);
+    return;
+  }
+
+  if (CanRunTasksUntil(queue, now, next_desired_run_time.value())) {

[Sami, 2017/04/27 12:12:30]

Reading this more, I guess I'm still a bit confused about the three different
cases here. Maybe it would be more natural to set a fence for the point (in
time) at which we want to stop running new tasks? Isn't that effectively what
we're doing here?

[altimin, 2017/04/27 12:23:50]

Yes, but it will be at least as complex as fence is enqueue-order based, not
timestamp-based. We will have the same problem with delayed tasks and ensuring
that fence blocks them but does not block immediate tasks (or delayed tasks
coming before the fence).

+    // We can run up until the next task uninterrupted. Remove the fence
+    // to allow new tasks to run immediately.

[Sami, 2017/04/26 13:09:08]

This is a little confusing: the next task doesn't want to run until
|next_desired_run_time|, so why are we allowing new tasks to run immediately?
Are we actually just allowing the next wake up to happen when removing the
fence?

[altimin, 2017/04/26 14:31:29]

Yes. Also there can be new tasks coming from different task queues, which should
be run if they are inside the wake-up window.

[Sami, 2017/04/27 12:12:30]

I see. I wonder if a fence with a timestamp would be a cleaner way of solving
this problem? Maybe we don't need to do that right now though.

[altimin, 2017/04/27 12:23:51]

See comment above.

+    queue->RemoveFence();
+
+    // TaskQueueThrottler does not schedule wake-ups implicitly, we need
+    // to be explicit.
+    if (next_desired_run_time.value() != now) {
+      time_domain_->SetNextTask(next_desired_run_time.value());

[Sami, 2017/04/26 13:09:08]

SetNextTaskRunTime? (also, why doesn't this just schedule a normal wake up in
the time domain?)

[altimin, 2017/04/26 14:31:29]

ThrottledTimeDomain does not schedule wake-ups (do prevent unnecessary ones).

+    }
+    return;
+  }
+
+  if (CanRunTasksAt(queue, now, is_wake_up)) {
+    // We can run task now, but we can't run until the next scheduled task.
+    // Insert a fresh fence to unblock queue and schedule a pump for the
+    // next wake-up.
+    queue->InsertFence(TaskQueue::InsertFencePosition::NOW);
+
+    base::Optional<base::TimeTicks> next_wake_up =

[Sami, 2017/04/26 13:09:08]

What's the difference between next_desired_run_time and next_wake_up?

[altimin, 2017/04/26 14:31:29]

next_desired_run_time takes into account tasks that can run now, next_wake_up
tells us when we should wake up to enqueue a new delayed task. 

+        queue->GetNextScheduledWakeUp();
+    if (next_wake_up) {
+      MaybeSchedulePumpThrottledTasks(
+          FROM_HERE, now, GetNextAllowedRunTime(queue, next_wake_up.value()));
+    }
+    return;
+  }
+
+  // Ensure that correct type of a fence is blocking queue which can't run.
+  base::Optional<QueueBlockType> block_type = GetQueueBlockType(now, queue);
+  if (block_type == QueueBlockType::kAllTasks) {
+    queue->InsertFence(TaskQueue::InsertFencePosition::BEGINNING_OF_TIME);
+  } else if (block_type == QueueBlockType::kNewTasksOnly &&
+             !queue->HasFence()) {
+    // Insert a new non-fully blocking fence only when there is no fence already
+    // in order avoid undesired unblocking of old tasks.
+    queue->InsertFence(TaskQueue::InsertFencePosition::NOW);
+  }
+  DCHECK(block_type);

[Sami, 2017/04/26 13:09:08]

nit: DCHECK before actually using |block_type|

[Sami, 2017/04/27 12:12:30]

Missed this?

[altimin, 2017/04/27 12:23:51]

Done, forgot to reply.

+
+  // Schedule a pump.
+  base::TimeTicks next_run_time =
+      GetNextAllowedRunTime(queue, next_desired_run_time.value());
+  MaybeSchedulePumpThrottledTasks(FROM_HERE, now, next_run_time);
+}
+
+base::Optional<QueueBlockType> TaskQueueThrottler::GetQueueBlockType(
+    base::TimeTicks now,
+    TaskQueue* queue) {
+  auto find_it = queue_details_.find(queue);
+  if (find_it == queue_details_.end())
+    return base::nullopt;
+
+  bool has_new_tasks_only_block = false;
+
+  for (BudgetPool* budget_pool : find_it->second.budget_pools) {
+    if (!budget_pool->CanRunTasksAt(now, false)) {
+      if (budget_pool->GetBlockType() == QueueBlockType::kAllTasks)
+        return QueueBlockType::kAllTasks;
+      // GetQueueBlockType() == QueueBlockType::kNewTasksOnly

[Sami, 2017/04/26 13:09:08]

DCHECK?

[altimin, 2017/04/26 14:31:29]

Done.

+      has_new_tasks_only_block = true;

[Sami, 2017/04/26 13:09:08]

break?

[altimin, 2017/04/26 14:31:29]

No, we can still encounter QueueBlockType::kAllTasks and this takes precedence. 

[Sami, 2017/04/27 12:12:30]

Ah got it.

+    }
+  }
+
+  if (has_new_tasks_only_block)
+    return QueueBlockType::kNewTasksOnly;
+  return base::nullopt;
 }
 
 void TaskQueueThrottler::AsValueInto(base::trace_event::TracedValue* state,
                                      base::TimeTicks now) const {
   if (pending_pump_throttled_tasks_runtime_) {
     state->SetDouble(
         "next_throttled_tasks_pump_in_seconds",
         (pending_pump_throttled_tasks_runtime_.value() - now).InSecondsF());
   }
 
@@ skipping 40 matching lines @@
 
   find_it->second.budget_pools.erase(budget_pool);
 
   MaybeDeleteQueueMetadata(find_it);
 }
 
 void TaskQueueThrottler::UnregisterBudgetPool(BudgetPool* budget_pool) {
   budget_pools_.erase(budget_pool);
 }
 
-void TaskQueueThrottler::UnblockQueue(base::TimeTicks now, TaskQueue* queue) {
-  SchedulePumpQueue(FROM_HERE, now, queue);
-}
-
-void TaskQueueThrottler::SchedulePumpQueue(
-    const tracked_objects::Location& from_here,
-    base::TimeTicks now,
-    TaskQueue* queue) {
-  if (!IsThrottled(queue))
-    return;
-
-  LazyNow lazy_now(now);
-  base::Optional<base::TimeTicks> next_desired_run_time =
-      NextTaskRunTime(&lazy_now, queue);
-  if (!next_desired_run_time)
-    return;
-
-  base::Optional<base::TimeTicks> next_run_time =
-      Max(next_desired_run_time, GetNextAllowedRunTime(now, queue));
-
-  MaybeSchedulePumpThrottledTasks(from_here, now, next_run_time.value());
-}
-
-base::TimeTicks TaskQueueThrottler::GetNextAllowedRunTime(base::TimeTicks now,
-                                                          TaskQueue* queue) {
-  base::TimeTicks next_run_time = now;
+base::TimeTicks TaskQueueThrottler::GetNextAllowedRunTime(
+    TaskQueue* queue,
+    base::TimeTicks desired_run_time) {
+  base::TimeTicks next_run_time = desired_run_time;
 
   auto find_it = queue_details_.find(queue);
   if (find_it == queue_details_.end())
     return next_run_time;
 
   for (BudgetPool* budget_pool : find_it->second.budget_pools) {
-    next_run_time =
-        std::max(next_run_time, budget_pool->GetNextAllowedRunTime());
+    next_run_time = std::max(
+        next_run_time, budget_pool->GetNextAllowedRunTime(desired_run_time));
   }
 
   return next_run_time;
 }
 
+bool TaskQueueThrottler::CanRunTasksAt(TaskQueue* queue,
+                                       base::TimeTicks moment,
+                                       bool is_wake_up) {
+  auto find_it = queue_details_.find(queue);
+  if (find_it == queue_details_.end())
+    return true;
+
+  for (BudgetPool* budget_pool : find_it->second.budget_pools) {
+    if (!budget_pool->CanRunTasksAt(moment, is_wake_up))
+      return false;
+  }
+
+  return true;
+}
+
+bool TaskQueueThrottler::CanRunTasksUntil(TaskQueue* queue,
+                                          base::TimeTicks now,
+                                          base::TimeTicks moment) {
+  auto find_it = queue_details_.find(queue);
+  if (find_it == queue_details_.end())
+    return true;
+
+  for (BudgetPool* budget_pool : find_it->second.budget_pools) {
+    if (!budget_pool->CanRunTasksUntil(now, moment))
+      return false;
+  }
+
+  return true;
+}
+
 void TaskQueueThrottler::MaybeDeleteQueueMetadata(TaskQueueMap::iterator it) {
   if (it->second.throttling_ref_count == 0 && it->second.budget_pools.empty())
     queue_details_.erase(it);
 }
 
 void TaskQueueThrottler::DisableThrottling() {
   if (!allow_throttling_)
     return;
 
   allow_throttling_ = false;
@@ skipping 26 matching lines @@
   for (const auto& map_entry : queue_details_) {
     if (map_entry.second.throttling_ref_count == 0)
       continue;
 
     TaskQueue* queue = map_entry.first;
 
     // Throttling is enabled and task queue should be blocked immediately
     // to enforce task alignment.
     queue->InsertFence(TaskQueue::InsertFencePosition::BEGINNING_OF_TIME);
     queue->SetTimeDomain(time_domain_.get());
-    SchedulePumpQueue(FROM_HERE, lazy_now.Now(), queue);
+    UpdateQueueThrottlingState(lazy_now.Now(), queue);
   }
 
   TRACE_EVENT0(tracing_category_, "TaskQueueThrottler_EnableThrottling");
 }
 
 }  // namespace scheduler
 }  // namespace blink