[scheduler] Implement time-based cpu throttling.

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

Index: third_party/WebKit/Source/platform/scheduler/renderer/task_queue_throttler.cc
diff --git a/third_party/WebKit/Source/platform/scheduler/renderer/throttling_helper.cc b/third_party/WebKit/Source/platform/scheduler/renderer/task_queue_throttler.cc
similarity index 27%
rename from third_party/WebKit/Source/platform/scheduler/renderer/throttling_helper.cc
rename to third_party/WebKit/Source/platform/scheduler/renderer/task_queue_throttler.cc
index 20ff3a1116f6628a01002daf48055ed181c184e0..8f934212a6b811e746416207f69baeb178747056 100644
--- a/third_party/WebKit/Source/platform/scheduler/renderer/throttling_helper.cc
+++ b/third_party/WebKit/Source/platform/scheduler/renderer/task_queue_throttler.cc
@@ -2,9 +2,15 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "platform/scheduler/renderer/throttling_helper.h"
+#include "platform/scheduler/renderer/task_queue_throttler.h"
 
+#include <cstdint>
+
+#include "base/format_macros.h"
 #include "base/logging.h"
+#include "base/memory/ptr_util.h"
+#include "base/optional.h"
+#include "base/strings/stringprintf.h"
 #include "platform/scheduler/base/real_time_domain.h"
 #include "platform/scheduler/child/scheduler_tqm_delegate.h"
 #include "platform/scheduler/renderer/auto_advancing_virtual_time_domain.h"
@@ -16,8 +22,167 @@
 namespace blink {
 namespace scheduler {
 
-ThrottlingHelper::ThrottlingHelper(RendererSchedulerImpl* renderer_scheduler,
-                                   const char* tracing_category)
+namespace {
+const int kMaxBudgetLevelInSeconds = 1;
+
+base::Optional<base::TimeTicks> NextTaskRunTime(LazyNow* lazy_now,
+                                                TaskQueue* queue) {
+  if (queue->HasPendingImmediateWork())
+    return lazy_now->Now();
+  return queue->GetNextScheduledWakeUp();
+}
+}
+
+TaskQueueThrottler::TimeBudgetPool::TimeBudgetPool(
+    const char* name,
+    TaskQueueThrottler* task_queue_throttler,
+    base::TimeTicks now)
+    : name_(name),
+      task_queue_throttler_(task_queue_throttler),
+      max_budget_level_(base::TimeDelta::FromSeconds(kMaxBudgetLevelInSeconds)),
+      last_checkpoint_(now),
+      cpu_percentage_(1),
+      is_enabled_(true) {}
+
+TaskQueueThrottler::TimeBudgetPool::~TimeBudgetPool() {}
+
+void TaskQueueThrottler::TimeBudgetPool::SetTimeBudget(base::TimeTicks now,
+                                                       double cpu_percentage) {
+  Advance(now);
+  cpu_percentage_ = cpu_percentage;
+}
+
+void TaskQueueThrottler::TimeBudgetPool::AddQueue(base::TimeTicks now,
+                                                  TaskQueue* queue) {
+  DCHECK(task_queue_throttler_->time_budget_pool_for_queue_.find(queue) ==
+         task_queue_throttler_->time_budget_pool_for_queue_.end());
+  task_queue_throttler_->time_budget_pool_for_queue_[queue] = this;
+
+  associated_task_queues_.insert(queue);
+
+  if (!task_queue_throttler_->IsThrottled(queue))

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

Note: if |time_budget_pool_for_queue_| gets rolled into Metadata we can do one
map lookup instead of two.

[altimin, 2016/09/15 15:52:11]

It's not feasible to do this, because queue can be unthrottled and can have a
time budget pool.

[alex clarke (OOO till 29th), 2016/09/15 16:18:40]

That's not true.  Just because |throttled_queues_| currently only contains
throttled queues doesn't mean it has to stay that way.

Obviously we'd need to change the name, perhaps |queue_details_| is a better
name.

[altimin, 2016/09/16 13:38:48]

Done.

[alex clarke (OOO till 29th), 2016/09/16 14:36:18]

I'm not seeing this change, did you forget up load the patchset?

[altimin, 2016/09/16 14:46:59]

Sorry, fixed now.

+    return;
+
+  queue->SetQueueEnabled(false);
+
+  task_queue_throttler_->MaybeSchedulePumpQueue(FROM_HERE, now, queue,
+                                                GetNextAllowedRunTime());
+}
+
+void TaskQueueThrottler::TimeBudgetPool::RemoveQueue(base::TimeTicks now,
+                                                     TaskQueue* queue) {
+  DCHECK_EQ(task_queue_throttler_->time_budget_pool_for_queue_[queue], this);
+  task_queue_throttler_->time_budget_pool_for_queue_.erase(queue);
+
+  associated_task_queues_.erase(queue);
+
+  if (!task_queue_throttler_->IsThrottled(queue))
+    return;
+
+  task_queue_throttler_->MaybeSchedulePumpQueue(FROM_HERE, now, queue);
+}
+
+void TaskQueueThrottler::TimeBudgetPool::EnableThrottling(LazyNow* lazy_now) {
+  if (is_enabled_)
+    return;
+  is_enabled_ = true;
+
+  for (TaskQueue* queue : associated_task_queues_) {
+    if (!task_queue_throttler_->IsThrottled(queue))
+      continue;
+
+    queue->SetQueueEnabled(false);
+
+    task_queue_throttler_->MaybeSchedulePumpQueue(
+        FROM_HERE, lazy_now->Now(), queue, GetNextAllowedRunTime());
+  }
+}
+
+void TaskQueueThrottler::TimeBudgetPool::DisableThrottling(LazyNow* lazy_now) {
+  if (!is_enabled_)
+    return;
+  is_enabled_ = false;
+
+  for (TaskQueue* queue : associated_task_queues_) {
+    if (!task_queue_throttler_->IsThrottled(queue))
+      continue;
+
+    task_queue_throttler_->MaybeSchedulePumpQueue(FROM_HERE, lazy_now->Now(),
+                                                  queue);
+  }
+}
+
+bool TaskQueueThrottler::TimeBudgetPool::IsThrottlingEnabled() const {
+  return is_enabled_;
+}
+
+void TaskQueueThrottler::TimeBudgetPool::Close() {
+  DCHECK_EQ(0u, associated_task_queues_.size());
+
+  task_queue_throttler_->time_budget_pools_.erase(this);
+}
+
+bool TaskQueueThrottler::TimeBudgetPool::IsAllowedToRun(base::TimeTicks now) {

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

I wonder if HasEnoughBudgetToRun is a better name, Sami WDYT?

[altimin, 2016/09/15 15:52:11]

Done.

+  Advance(now);
+  return !is_enabled_ || current_budget_level_.InMicroseconds() >= 0;
+}
+
+base::TimeTicks TaskQueueThrottler::TimeBudgetPool::GetNextAllowedRunTime() {
+  if (!is_enabled_ || current_budget_level_.InMicroseconds() >= 0) {
+    return last_checkpoint_;
+  } else {
+    // Subtract because current_budget is negative.
+    return last_checkpoint_ - current_budget_level_ / cpu_percentage_;
+  }
+}
+
+void TaskQueueThrottler::TimeBudgetPool::RecordTaskRunTime(
+    base::TimeDelta task_run_time) {
+  if (is_enabled_) {
+    current_budget_level_ -= task_run_time;
+  }
+}
+
+const char* TaskQueueThrottler::TimeBudgetPool::Name() const {
+  return name_;
+}
+
+void TaskQueueThrottler::TimeBudgetPool::AsValueInto(
+    base::trace_event::TracedValue* state,
+    base::TimeTicks now) const {
+  state->BeginDictionary();
+
+  state->SetString("name", name_);
+  state->SetDouble("time_budget", cpu_percentage_);
+  state->SetDouble("time_budget_level_in_seconds",
+                   current_budget_level_.InSecondsF());
+  state->SetDouble("last_checkpoint_seconds_ago",
+                   (now - last_checkpoint_).InSecondsF());
+
+  state->BeginArray("task_queues");
+  for (TaskQueue* queue : associated_task_queues_) {
+    state->AppendString(base::StringPrintf(
+        "%" PRIx64, static_cast<uint64_t>(reinterpret_cast<uintptr_t>(queue))));
+  }
+  state->EndArray();
+
+  state->EndDictionary();
+}
+
+void TaskQueueThrottler::TimeBudgetPool::Advance(base::TimeTicks now) {
+  if (now > last_checkpoint_) {
+    if (is_enabled_) {
+      current_budget_level_ = std::min(
+          current_budget_level_ + cpu_percentage_ * (now - last_checkpoint_),
+          max_budget_level_);
+    }
+    last_checkpoint_ = now;
+  }
+}
+
+TaskQueueThrottler::TaskQueueThrottler(
+    RendererSchedulerImpl* renderer_scheduler,
+    const char* tracing_category)
     : task_runner_(renderer_scheduler->ControlTaskRunner()),
       renderer_scheduler_(renderer_scheduler),
       tick_clock_(renderer_scheduler->tick_clock()),
@@ -26,15 +191,15 @@ ThrottlingHelper::ThrottlingHelper(RendererSchedulerImpl* renderer_scheduler,
       virtual_time_(false),
       weak_factory_(this) {
   pump_throttled_tasks_closure_.Reset(base::Bind(
-      &ThrottlingHelper::PumpThrottledTasks, weak_factory_.GetWeakPtr()));
-  forward_immediate_work_closure_ =
-      base::Bind(&ThrottlingHelper::OnTimeDomainHasImmediateWork,
+      &TaskQueueThrottler::PumpThrottledTasks, weak_factory_.GetWeakPtr()));
+  forward_immediate_work_callback_ =
+      base::Bind(&TaskQueueThrottler::OnTimeDomainHasImmediateWork,
                  weak_factory_.GetWeakPtr());
 
   renderer_scheduler_->RegisterTimeDomain(time_domain_.get());
 }
 
-ThrottlingHelper::~ThrottlingHelper() {
+TaskQueueThrottler::~TaskQueueThrottler() {
   // It's possible for queues to be still throttled, so we need to tidy up
   // before unregistering the time domain.
   for (const TaskQueueMap::value_type& map_entry : throttled_queues_) {
@@ -46,7 +211,7 @@ ThrottlingHelper::~ThrottlingHelper() {
   renderer_scheduler_->UnregisterTimeDomain(time_domain_.get());
 }
 
-void ThrottlingHelper::SetQueueEnabled(TaskQueue* task_queue, bool enabled) {
+void TaskQueueThrottler::SetQueueEnabled(TaskQueue* task_queue, bool enabled) {
   TaskQueueMap::iterator find_it = throttled_queues_.find(task_queue);
 
   if (find_it == throttled_queues_.end()) {
@@ -63,7 +228,7 @@ void ThrottlingHelper::SetQueueEnabled(TaskQueue* task_queue, bool enabled) {
     task_queue->SetQueueEnabled(false);
 }
 
-void ThrottlingHelper::IncreaseThrottleRefCount(TaskQueue* task_queue) {
+void TaskQueueThrottler::IncreaseThrottleRefCount(TaskQueue* task_queue) {
   DCHECK_NE(task_queue, task_runner_.get());
 
   if (virtual_time_)
@@ -74,25 +239,28 @@ void ThrottlingHelper::IncreaseThrottleRefCount(TaskQueue* task_queue) {
           task_queue, Metadata(1, task_queue->IsQueueEnabled())));
 
   if (insert_result.second) {
-    // The insert was succesful so we need to throttle the queue.
+    // The insert was successful so we need to throttle the queue.
     task_queue->SetTimeDomain(time_domain_.get());
     task_queue->RemoveFence();
     task_queue->SetQueueEnabled(false);
 
     if (!task_queue->IsEmpty()) {
       if (task_queue->HasPendingImmediateWork()) {
-        OnTimeDomainHasImmediateWork();
+        OnTimeDomainHasImmediateWork(task_queue);
       } else {
-        OnTimeDomainHasDelayedWork();
+        OnTimeDomainHasDelayedWork(task_queue);
       }
     }
+
+    TRACE_EVENT1(tracing_category_, "TaskQueueThrottler_TaskQueueThrottled",
+                 "task_queue", task_queue);
   } else {
     // An entry already existed in the map so we need to increment the refcount.
     insert_result.first->second.throttling_ref_count++;
   }
 }
 
-void ThrottlingHelper::DecreaseThrottleRefCount(TaskQueue* task_queue) {
+void TaskQueueThrottler::DecreaseThrottleRefCount(TaskQueue* task_queue) {
   if (virtual_time_)
     return;
 
@@ -107,104 +275,170 @@ void ThrottlingHelper::DecreaseThrottleRefCount(TaskQueue* task_queue) {
     task_queue->SetTimeDomain(renderer_scheduler_->real_time_domain());
     task_queue->RemoveFence();
     task_queue->SetQueueEnabled(enabled);
+
+    TRACE_EVENT1(tracing_category_, "TaskQueueThrottler_TaskQueueUntrottled",
+                 "task_queue", task_queue);
   }
 }
 
-bool ThrottlingHelper::IsThrottled(TaskQueue* task_queue) const {
+bool TaskQueueThrottler::IsThrottled(TaskQueue* task_queue) const {
   return throttled_queues_.find(task_queue) != throttled_queues_.end();
 }
 
-void ThrottlingHelper::UnregisterTaskQueue(TaskQueue* task_queue) {
+void TaskQueueThrottler::UnregisterTaskQueue(TaskQueue* task_queue) {
   throttled_queues_.erase(task_queue);
 }
 
-void ThrottlingHelper::OnTimeDomainHasImmediateWork() {
-  // Forward to the main thread if called from another thread.
+void TaskQueueThrottler::OnTimeDomainHasImmediateWork(TaskQueue* queue) {
+  // Forward to the main thread if called from another thread
   if (!task_runner_->RunsTasksOnCurrentThread()) {
-    task_runner_->PostTask(FROM_HERE, forward_immediate_work_closure_);
+    task_runner_->PostTask(FROM_HERE,
+                           base::Bind(forward_immediate_work_callback_, queue));
     return;
   }
   TRACE_EVENT0(tracing_category_,
-               "ThrottlingHelper::OnTimeDomainHasImmediateWork");
+               "TaskQueueThrottler::OnTimeDomainHasImmediateWork");
+
   base::TimeTicks now = tick_clock_->NowTicks();
-  MaybeSchedulePumpThrottledTasksLocked(FROM_HERE, now, now);
+  base::TimeTicks next_allowed_run_time = GetNextAllowedRunTime(now, queue);
+  MaybeSchedulePumpThrottledTasks(FROM_HERE, now, next_allowed_run_time);
 }
 
-void ThrottlingHelper::OnTimeDomainHasDelayedWork() {
+void TaskQueueThrottler::OnTimeDomainHasDelayedWork(TaskQueue* queue) {
   TRACE_EVENT0(tracing_category_,
-               "ThrottlingHelper::OnTimeDomainHasDelayedWork");
-  base::TimeTicks next_scheduled_delayed_task;
-  bool has_delayed_task =
-      time_domain_->NextScheduledRunTime(&next_scheduled_delayed_task);
-  DCHECK(has_delayed_task);
+               "TaskQueueThrottler::OnTimeDomainHasDelayedWork");
   base::TimeTicks now = tick_clock_->NowTicks();
-  MaybeSchedulePumpThrottledTasksLocked(FROM_HERE, now,
-                                        next_scheduled_delayed_task);
+  LazyNow lazy_now(now);
+
+  base::Optional<base::TimeTicks> next_scheduled_delayed_task =
+      NextTaskRunTime(&lazy_now, queue);
+  DCHECK(next_scheduled_delayed_task);
+  MaybeSchedulePumpThrottledTasks(FROM_HERE, now,
+                                  next_scheduled_delayed_task.value());
+}
+
+namespace {
+
+template <class T>
+T Min(const base::Optional<T>& optional, const T& value) {
+  if (!optional) {
+    return value;
+  }
+  return std::min(optional.value(), value);
+}
+
+template <class T>
+base::Optional<T> Min(const base::Optional<T>& a, const base::Optional<T>& b) {
+  if (!b)
+    return a;
+  if (!a)
+    return b;
+  return std::min(a.value(), b.value());
+}
+
+template <class T>
+T Max(const base::Optional<T>& optional, const T& value) {
+  if (!optional)
+    return value;
+  return std::max(optional.value(), value);
+}
+
+template <class T>
+base::Optional<T> Max(const base::Optional<T>& a, const base::Optional<T>& b) {
+  if (!b)
+    return a;
+  if (!a)
+    return b;
+  return std::max(a.value(), b.value());
 }
 
-void ThrottlingHelper::PumpThrottledTasks() {
-  TRACE_EVENT0(tracing_category_, "ThrottlingHelper::PumpThrottledTasks");
-  pending_pump_throttled_tasks_runtime_ = base::TimeTicks();
+}  // namespace
+
+void TaskQueueThrottler::PumpThrottledTasks() {
+  TRACE_EVENT0("renderer.scheduler", "TaskQueueThrottler::PumpThrottledTasks");

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

Why change tracing_category_?

[altimin, 2016/09/15 15:52:10]

Done.

+  pending_pump_throttled_tasks_runtime_.reset();
+
+  LazyNow lazy_now(tick_clock_);
+  base::Optional<base::TimeTicks> next_scheduled_delayed_task;
 
-  LazyNow lazy_low(tick_clock_);
   for (const TaskQueueMap::value_type& map_entry : throttled_queues_) {
     TaskQueue* task_queue = map_entry.first;
     if (!map_entry.second.enabled || task_queue->IsEmpty())
       continue;
 
-    task_queue->SetQueueEnabled(true);
-    task_queue->InsertFence();
+    base::TimeTicks next_allowed_run_time =

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

Please add a comment here saying: Don't pump queues whose budget pool doesn't
allow them to run now.

[altimin, 2016/09/15 15:52:10]

Done.

+        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(
+          "renderer.scheduler",

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

tracing_category_?

[altimin, 2016/09/15 15:52:11]

Done.

+          "TaskQueueThrottler::PumpThrottledTasks_ExpensiveTaskThrottled",
+          "throttle_time_in_seconds",
+          (next_allowed_run_time - next_desired_run_time.value()).InSecondsF());
+
+      next_scheduled_delayed_task =
+          Min(next_scheduled_delayed_task, next_allowed_run_time);
+    }

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

can we put continue;  in there for clarity?

[altimin, 2016/09/15 15:52:10]

Done.

+
+    next_scheduled_delayed_task =
+        Min(next_scheduled_delayed_task, task_queue->GetNextScheduledWakeUp());
+
+    if (next_allowed_run_time == lazy_now.Now()) {

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

For readability I would be tempted instead to write;

if (next_allowed_run_time > lazy_now.Now())
  continue;

task_queue->SetQueueEnabled(true);
task_queue->InsertFence();

[altimin, 2016/09/15 15:52:11]

Done.

+      task_queue->SetQueueEnabled(true);
+      task_queue->InsertFence();
+    }
   }
-  // Make sure NextScheduledRunTime gives us an up-to date result.
-  time_domain_->ClearExpiredWakeups();
 
-  base::TimeTicks next_scheduled_delayed_task;
-  // Maybe schedule a call to ThrottlingHelper::PumpThrottledTasks if there is
-  // a pending delayed task. NOTE posting a non-delayed task in the future will
-  // result in ThrottlingHelper::OnTimeDomainHasImmediateWork being called.
-  if (time_domain_->NextScheduledRunTime(&next_scheduled_delayed_task)) {
-    MaybeSchedulePumpThrottledTasksLocked(FROM_HERE, lazy_low.Now(),
-                                          next_scheduled_delayed_task);
+  // 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);
   }
 }
 
 /* static */
-base::TimeTicks ThrottlingHelper::ThrottledRunTime(
+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);
 }
 
-void ThrottlingHelper::MaybeSchedulePumpThrottledTasksLocked(
+void TaskQueueThrottler::MaybeSchedulePumpThrottledTasks(
     const tracked_objects::Location& from_here,
     base::TimeTicks now,
-    base::TimeTicks unthrottled_runtime) {
+    base::TimeTicks runtime) {

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

unaligned_runtiume

[altimin, 2016/09/15 15:52:11]

Done.

   if (virtual_time_)
     return;
 
-  base::TimeTicks throttled_runtime =
-      ThrottledRunTime(std::max(now, unthrottled_runtime));
+  runtime = std::max(now, AlignedThrottledRunTime(runtime));
+
   // If there is a pending call to PumpThrottledTasks and it's sooner than
-  // |unthrottled_runtime| then return.
-  if (!pending_pump_throttled_tasks_runtime_.is_null() &&
-      throttled_runtime >= pending_pump_throttled_tasks_runtime_) {
+  // |runtime| then return.
+  if (pending_pump_throttled_tasks_runtime_ &&
+      runtime >= pending_pump_throttled_tasks_runtime_.value()) {
     return;
   }
 
-  pending_pump_throttled_tasks_runtime_ = throttled_runtime;
+  pending_pump_throttled_tasks_runtime_ = runtime;
 
   pump_throttled_tasks_closure_.Cancel();
 
-  base::TimeDelta delay = pending_pump_throttled_tasks_runtime_ - now;
+  base::TimeDelta delay = pending_pump_throttled_tasks_runtime_.value() - now;
   TRACE_EVENT1(tracing_category_,
-               "ThrottlingHelper::MaybeSchedulePumpThrottledTasksLocked",
+               "TaskQueueThrottler::MaybeSchedulePumpThrottledTasks",
                "delay_till_next_pump_ms", delay.InMilliseconds());
   task_runner_->PostDelayedTask(
       from_here, pump_throttled_tasks_closure_.callback(), delay);
 }
 
-void ThrottlingHelper::EnableVirtualTime() {
+void TaskQueueThrottler::EnableVirtualTime() {
   virtual_time_ = true;
 
   pump_throttled_tasks_closure_.Cancel();
@@ -221,5 +455,90 @@ void ThrottlingHelper::EnableVirtualTime() {
   }
 }
 
+TaskQueueThrottler::TimeBudgetPool* TaskQueueThrottler::CreateTimeBudgetPool(
+    const char* name) {
+  TimeBudgetPool* time_budget_pool =
+      new TimeBudgetPool(name, this, tick_clock_->NowTicks());
+  time_budget_pools_[time_budget_pool] = base::WrapUnique(time_budget_pool);
+  return time_budget_pool;
+}
+
+void TaskQueueThrottler::OnTaskRunTimeReported(TaskQueue* task_queue,
+                                               base::TimeTicks start_time,
+                                               base::TimeTicks end_time) {
+  if (!IsThrottled(task_queue))
+    return;
+
+  TimeBudgetPool* time_budget_pool = GetTimeBudgetPoolForQueue(task_queue);
+  if (time_budget_pool) {

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

For readability I think we should prefer:

if (!time_budget_pool)
  return;

[altimin, 2016/09/15 15:52:10]

Done.

+    time_budget_pool->RecordTaskRunTime(end_time - start_time);
+    if (!time_budget_pool->IsAllowedToRun(end_time)) {
+      // This task was too expensive and all following tasks are throttled
+      // until explicitly allowed.
+      task_queue->SetQueueEnabled(false);
+
+      if (task_queue->HasPendingImmediateWork()) {
+        MaybeSchedulePumpThrottledTasks(
+            FROM_HERE, end_time,
+            std::max(end_time, time_budget_pool->GetNextAllowedRunTime()));
+      }
+    }
+  }
+}
+
+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());
+  }
+
+  state->BeginDictionary("time_budget_pools");
+
+  for (const auto& map_entry : time_budget_pools_) {
+    TaskQueueThrottler::TimeBudgetPool* pool = map_entry.first;
+    pool->AsValueInto(state, now);
+  }
+
+  state->EndDictionary();
+}
+
+TaskQueueThrottler::TimeBudgetPool*
+TaskQueueThrottler::GetTimeBudgetPoolForQueue(TaskQueue* queue) {
+  auto find_it = time_budget_pool_for_queue_.find(queue);
+  if (find_it == time_budget_pool_for_queue_.end()) {
+    return nullptr;
+  } else {
+    TimeBudgetPool* result = find_it->second;
+    DCHECK(result);
+    return result;
+  }
+}
+
+void TaskQueueThrottler::MaybeSchedulePumpQueue(
+    const tracked_objects::Location& from_here,
+    base::TimeTicks now,
+    TaskQueue* queue,
+    base::Optional<base::TimeTicks> next_possible_run_time) {
+  LazyNow lazy_now(now);
+  base::Optional<base::TimeTicks> next_run_time =
+      Max(NextTaskRunTime(&lazy_now, queue), next_possible_run_time);
+
+  if (next_run_time) {

[alex clarke (OOO till 29th), 2016/09/15 12:19:35]

Can this ever actually be false?  Should we DCHECK?

[altimin, 2016/09/15 15:52:10]

Yes. We're calling MaybeSchedulePumpQueue when we're adding/removing queues, and
they can be empty, and |next_run_time| can be null.

+    MaybeSchedulePumpThrottledTasks(from_here, now, next_run_time.value());
+  }
+}
+
+base::TimeTicks TaskQueueThrottler::GetNextAllowedRunTime(base::TimeTicks now,
+                                                          TaskQueue* queue) {
+  TimeBudgetPool* time_budget_pool = GetTimeBudgetPoolForQueue(queue);
+  if (!time_budget_pool) {
+    return now;
+  } else {
+    return std::max(now, time_budget_pool->GetNextAllowedRunTime());
+  }
+}
+
 }  // namespace scheduler
 }  // namespace blink