[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->OnQueueNextWakeUpChanged(queue, now, next_wake_up);
+  }
+
+  // TODO(altimin): This probably can be removed —- budget pools should
+  // schedule this.
+  base::TimeTicks next_allowed_run_time =
+      GetNextAllowedRunTime(queue, next_wake_up);
   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())) {
+    // We can run up until the next task uninterrupted. Remove the fence
+    // to allow new tasks to run immediately.
+    queue->RemoveFence();

[alex clarke (OOO till 29th), 2017/05/02 10:51:54]

Can we DCHECK(queue->HasFence()) or is that going to fail for newly added
queues?

[altimin, 2017/05/02 18:16:58]

It is going to fail when the fence is removed and new work is coming from a
different thread during this period.

+
+    // TaskQueueThrottler does not schedule wake-ups implicitly, we need
+    // to be explicit.
+    if (next_desired_run_time.value() != now) {
+      time_domain_->SetNextTaskRunTime(next_desired_run_time.value());
+    }
+    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 =
+        queue->GetNextScheduledWakeUp();
+    if (next_wake_up) {
+      MaybeSchedulePumpThrottledTasks(
+          FROM_HERE, now, GetNextAllowedRunTime(queue, next_wake_up.value()));
+    }
+    return;
+  }
+
+  base::TimeTicks next_run_time =
+      GetNextAllowedRunTime(queue, next_desired_run_time.value());
+
+  // Ensure that correct type of a fence is blocking queue which can't run.

[alex clarke (OOO till 29th), 2017/05/02 10:51:54]

Could you please re-word this comment?  It doesn't make much sense to me.

[altimin, 2017/05/02 18:16:58]

Done.

+  base::Optional<QueueBlockType> block_type = GetQueueBlockType(now, queue);
+  DCHECK(block_type);
+
+  if (block_type == QueueBlockType::kAllTasks) {

[alex clarke (OOO till 29th), 2017/05/02 10:51:54]

Are you planning on adding more QueueBlockTypes?  If so then this should be a
switch because you'll get a compile error if you forget to handle the new type
here. 

[altimin, 2017/05/02 18:16:58]

Agreed.

+    queue->InsertFence(TaskQueue::InsertFencePosition::BEGINNING_OF_TIME);
+
+    TRACE_EVENT1(
+        "renderer.scheduler",
+        "TaskQueueThrottler::PumpThrottledTasks_ExpensiveTaskThrottled",
+        "throttle_time_in_seconds",
+        (next_run_time - next_desired_run_time.value()).InSecondsF());
+  } 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);
+  }

[alex clarke (OOO till 29th), 2017/05/02 10:51:54]

So just to make sure I understad, if block_type == QueueBlockType::kNewTasksOnly
and there already is a fence we don't need to do anything.  Does it matter of
the pre-existing fence is InsertFencePosition::BEGINNING_OF_TIME?

+
+  // Schedule a pump.
+  MaybeSchedulePumpThrottledTasks(FROM_HERE, now, next_run_time);
+}
+
+base::Optional<QueueBlockType> TaskQueueThrottler::GetQueueBlockType(

[alex clarke (OOO till 29th), 2017/05/02 10:51:54]

Do we need this function? Could we instead expose QueueBlockType via
TaskQueueThrottler::CanRunTasksAt.  They seem to do a lot of the same work.

+    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;
+      DCHECK_EQ(budget_pool->GetBlockType(), QueueBlockType::kNewTasksOnly);
+      has_new_tasks_only_block = true;
+    }
+  }
+
+  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