[scheduler] Create TaskQueueThrottler for worker thread

Add behind-a-feature-flag functionality to throttle tasks on the worker
thread to 1Hz.

R=alexclarke@chromium.org
BUG=776416

Change-Id: I1698aea8788f8688027e01af4d5e72405e2c1a23
Reviewed-on: https://chromium-review.googlesource.com/1070983Reviewed-by: Sami Kyöstilä <skyostil@chromium.org>
Reviewed-by: Alex Clarke <alexclarke@chromium.org>
Commit-Queue: Alexander Timin <altimin@chromium.org>
Cr-Commit-Position: refs/heads/master@{#561578}

third_party/blink/renderer/platform/scheduler/child/features.h

@@ -13,6 +13,9 @@ namespace scheduler {
 const base::Feature kHighPriorityInput{"BlinkSchedulerHighPriorityInput",
                                        base::FEATURE_DISABLED_BY_DEFAULT};
 
+const base::Feature kDedicatedWorkerThrottling{
+    "kBlinkSchedulerWorkerThrottling", base::FEATURE_DISABLED_BY_DEFAULT};
+
 }  // namespace scheduler
 }  // namespace blink
 

third_party/blink/renderer/platform/scheduler/child/worker_scheduler.cc

@@ -5,6 +5,8 @@
 #include "third_party/blink/renderer/platform/scheduler/child/worker_scheduler.h"
 
 #include "third_party/blink/renderer/platform/scheduler/child/task_queue_with_task_type.h"
+#include "third_party/blink/renderer/platform/scheduler/common/throttling/task_queue_throttler.h"
+#include "third_party/blink/renderer/platform/scheduler/common/throttling/wake_up_budget_pool.h"
 #include "third_party/blink/renderer/platform/scheduler/public/non_main_thread_scheduler.h"
 
 namespace blink {
@@ -16,6 +18,11 @@ WorkerScheduler::WorkerScheduler(
       throttleable_task_queue_(non_main_thread_scheduler->CreateTaskRunner()),
       thread_scheduler_(non_main_thread_scheduler) {
   thread_scheduler_->RegisterWorkerScheduler(this);
+  if (WakeUpBudgetPool* wake_up_budget_pool =
+          thread_scheduler_->wake_up_budget_pool()) {
+    wake_up_budget_pool->AddQueue(thread_scheduler_->GetTickClock()->NowTicks(),
+                                  throttleable_task_queue_.get());
+  }
 }
 
 WorkerScheduler::~WorkerScheduler() {
@@ -30,9 +37,16 @@ WorkerScheduler::OnActiveConnectionCreated() {
 }
 
 void WorkerScheduler::Dispose() {
+  if (TaskQueueThrottler* throttler =
+          thread_scheduler_->task_queue_throttler()) {
+    throttler->ShutdownTaskQueue(throttleable_task_queue_.get());
+  }
+
   thread_scheduler_->UnregisterWorkerScheduler(this);
+
   default_task_queue_->ShutdownTaskQueue();
   throttleable_task_queue_->ShutdownTaskQueue();
+
 #if DCHECK_IS_ON()
   is_disposed_ = true;
 #endif
@@ -100,5 +114,31 @@ scoped_refptr<base::SingleThreadTaskRunner> WorkerScheduler::GetTaskRunner(
   return nullptr;
 }
 
+void WorkerScheduler::OnThrottlingStateChanged(
+    FrameScheduler::ThrottlingState throttling_state) {
+  if (throttling_state_ == throttling_state)
+    return;
+  throttling_state_ = throttling_state;
+
+  if (TaskQueueThrottler* throttler =
+          thread_scheduler_->task_queue_throttler()) {
+    if (throttling_state_ == FrameScheduler::ThrottlingState::kThrottled) {
+      throttler->IncreaseThrottleRefCount(throttleable_task_queue_.get());
+    } else {
+      throttler->DecreaseThrottleRefCount(throttleable_task_queue_.get());
+    }
+  }
+}
+
+scoped_refptr<base::sequence_manager::TaskQueue>
+WorkerScheduler::DefaultTaskQueue() {
+  return default_task_queue_.get();
+}
+
+scoped_refptr<base::sequence_manager::TaskQueue>
+WorkerScheduler::ThrottleableTaskQueue() {
+  return throttleable_task_queue_.get();
+}
+
 }  // namespace scheduler
 }  // namespace blink

third_party/blink/renderer/platform/scheduler/child/worker_scheduler.h

@@ -9,6 +9,7 @@
 #include "third_party/blink/public/platform/task_type.h"
 #include "third_party/blink/renderer/platform/scheduler/base/task_queue.h"
 #include "third_party/blink/renderer/platform/scheduler/public/frame_or_worker_scheduler.h"
+#include "third_party/blink/renderer/platform/scheduler/public/frame_scheduler.h"
 
 namespace blink {
 
@@ -36,13 +37,20 @@ class PLATFORM_EXPORT WorkerScheduler : public FrameOrWorkerScheduler {
   // This must be called only from WorkerThread::GetTaskRunner().
   scoped_refptr<base::SingleThreadTaskRunner> GetTaskRunner(TaskType) const;
 
-  // TODO(nhiroki): Add mechanism to throttle/suspend tasks in response to the
-  // state of the parent document (https://crbug.com/670534).
+  void OnThrottlingStateChanged(
+      FrameScheduler::ThrottlingState throtting_state);
+
+ protected:
+  scoped_refptr<base::sequence_manager::TaskQueue> DefaultTaskQueue();
+  scoped_refptr<base::sequence_manager::TaskQueue> ThrottleableTaskQueue();
 
  private:
   scoped_refptr<base::sequence_manager::TaskQueue> default_task_queue_;
   scoped_refptr<base::sequence_manager::TaskQueue> throttleable_task_queue_;
 
+  FrameScheduler::ThrottlingState throttling_state_ =
+      FrameScheduler::ThrottlingState::kNotThrottled;
+
   NonMainThreadScheduler* thread_scheduler_;  // NOT OWNED
 
 #if DCHECK_IS_ON()

third_party/blink/renderer/platform/scheduler/child/worker_scheduler_unittest.cc

@@ -7,14 +7,16 @@
 #include <memory>
 #include "base/macros.h"
 #include "base/test/simple_test_tick_clock.h"
-#include "base/test/test_simple_task_runner.h"
+#include "base/test/test_mock_time_task_runner.h"
 #include "testing/gmock/include/gmock/gmock.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "third_party/blink/renderer/platform/scheduler/base/test/task_queue_manager_for_test.h"
+#include "third_party/blink/renderer/platform/scheduler/common/throttling/task_queue_throttler.h"
 #include "third_party/blink/renderer/platform/scheduler/worker/worker_thread_scheduler.h"
 #include "third_party/blink/renderer/platform/wtf/functional.h"
 
 using testing::ElementsAreArray;
+using testing::ElementsAre;
 
 namespace blink {
 namespace scheduler {
@@ -26,6 +28,24 @@ void AppendToVectorTestTask(std::vector<std::string>* vector,
   vector->push_back(value);
 }
 
+void RunChainedTask(scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+                    int count,
+                    const base::TickClock* clock,
+                    std::vector<base::TimeTicks>* tasks) {
+  tasks->push_back(clock->NowTicks());
+
+  if (count == 1)
+    return;
+
+  // Add a delay of 50ms to ensure that wake-up based throttling does not affect
+  // us.
+  task_runner->PostDelayedTask(
+      FROM_HERE,
+      base::BindOnce(&RunChainedTask, task_runner, count - 1,
+                     base::Unretained(clock), base::Unretained(tasks)),
+      base::TimeDelta::FromMilliseconds(50));
+}
+
 class WorkerThreadSchedulerForTest : public WorkerThreadScheduler {
  public:
   WorkerThreadSchedulerForTest(
@@ -37,30 +57,55 @@ class WorkerThreadSchedulerForTest : public WorkerThreadScheduler {
   const std::unordered_set<WorkerScheduler*>& worker_schedulers() {
     return worker_schedulers_;
   }
+
+  using WorkerThreadScheduler::CreateTaskQueueThrottler;
+};
+
+class WorkerSchedulerForTest : public WorkerScheduler {
+ public:
+  explicit WorkerSchedulerForTest(
+      WorkerThreadSchedulerForTest* thread_scheduler)
+      : WorkerScheduler(thread_scheduler) {}
+
+  using WorkerScheduler::DefaultTaskQueue;
+  using WorkerScheduler::ThrottleableTaskQueue;
 };
 
 class WorkerSchedulerTest : public testing::Test {
  public:
   WorkerSchedulerTest()
-      : mock_task_runner_(new base::TestSimpleTaskRunner()),
+      : mock_task_runner_(new base::TestMockTimeTaskRunner()),
         scheduler_(new WorkerThreadSchedulerForTest(
             WebThreadType::kTestThread,
             base::sequence_manager::TaskQueueManagerForTest::Create(
                 nullptr,
                 mock_task_runner_,
-                &clock_),
+                mock_task_runner_->GetMockTickClock()),
             nullptr /* proxy */)) {
-    clock_.Advance(base::TimeDelta::FromMicroseconds(5000));
+    mock_task_runner_->AdvanceMockTickClock(
+        base::TimeDelta::FromMicroseconds(5000));
   }
 
   ~WorkerSchedulerTest() override = default;
 
   void SetUp() override {
     scheduler_->Init();
-    worker_scheduler_ = std::make_unique<WorkerScheduler>(scheduler_.get());
+    worker_scheduler_ =
+        std::make_unique<WorkerSchedulerForTest>(scheduler_.get());
+  }
+
+  void TearDown() override {
+    if (worker_scheduler_) {
+      worker_scheduler_->Dispose();
+      worker_scheduler_.reset();
+    }
   }
 
-  void RunUntilIdle() { mock_task_runner_->RunUntilIdle(); }
+  const base::TickClock* GetClock() {
+    return mock_task_runner_->GetMockTickClock();
+  }
+
+  void RunUntilIdle() { mock_task_runner_->FastForwardUntilNoTasksRemain(); }
 
   // Helper for posting a task.
   void PostTestTask(std::vector<std::string>* run_order,
@@ -72,11 +117,10 @@ class WorkerSchedulerTest : public testing::Test {
   }
 
  protected:
-  base::SimpleTestTickClock clock_;
-  scoped_refptr<base::TestSimpleTaskRunner> mock_task_runner_;
+  scoped_refptr<base::TestMockTimeTaskRunner> mock_task_runner_;
 
   std::unique_ptr<WorkerThreadSchedulerForTest> scheduler_;
-  std::unique_ptr<WorkerScheduler> worker_scheduler_;
+  std::unique_ptr<WorkerSchedulerForTest> worker_scheduler_;
 
   DISALLOW_COPY_AND_ASSIGN(WorkerSchedulerTest);
 };
@@ -97,20 +141,23 @@ TEST_F(WorkerSchedulerTest, TestPostTasks) {
   PostTestTask(&run_order, "T5");
   RunUntilIdle();
   EXPECT_TRUE(run_order.empty());
+
+  worker_scheduler_.reset();
 }
 
 TEST_F(WorkerSchedulerTest, RegisterWorkerSchedulers) {
   EXPECT_THAT(scheduler_->worker_schedulers(),
               testing::ElementsAre(worker_scheduler_.get()));
 
-  std::unique_ptr<WorkerScheduler> worker_scheduler2 =
-      std::make_unique<WorkerScheduler>(scheduler_.get());
+  std::unique_ptr<WorkerSchedulerForTest> worker_scheduler2 =
+      std::make_unique<WorkerSchedulerForTest>(scheduler_.get());
 
   EXPECT_THAT(scheduler_->worker_schedulers(),
               testing::UnorderedElementsAre(worker_scheduler_.get(),
                                             worker_scheduler2.get()));
 
   worker_scheduler_->Dispose();
+  worker_scheduler_.reset();
 
   EXPECT_THAT(scheduler_->worker_schedulers(),
               testing::ElementsAre(worker_scheduler2.get()));
@@ -120,6 +167,75 @@ TEST_F(WorkerSchedulerTest, RegisterWorkerSchedulers) {
   EXPECT_THAT(scheduler_->worker_schedulers(), testing::ElementsAre());
 }
 
+TEST_F(WorkerSchedulerTest, ThrottleWorkerScheduler) {
+  scheduler_->CreateTaskQueueThrottler();
+
+  EXPECT_FALSE(scheduler_->task_queue_throttler()->IsThrottled(
+      worker_scheduler_->ThrottleableTaskQueue().get()));
+
+  scheduler_->OnThrottlingStateChanged(
+      FrameScheduler::ThrottlingState::kThrottled);
+  EXPECT_TRUE(scheduler_->task_queue_throttler()->IsThrottled(
+      worker_scheduler_->ThrottleableTaskQueue().get()));
+
+  scheduler_->OnThrottlingStateChanged(
+      FrameScheduler::ThrottlingState::kThrottled);
+  EXPECT_TRUE(scheduler_->task_queue_throttler()->IsThrottled(
+      worker_scheduler_->ThrottleableTaskQueue().get()));
+
+  // Ensure that two calls with kThrottled do not mess with throttling
+  // refcount.
+  scheduler_->OnThrottlingStateChanged(
+      FrameScheduler::ThrottlingState::kNotThrottled);
+  EXPECT_FALSE(scheduler_->task_queue_throttler()->IsThrottled(
+      worker_scheduler_->ThrottleableTaskQueue().get()));
+}
+
+TEST_F(WorkerSchedulerTest, ThrottleWorkerScheduler_CreateThrottled) {
+  scheduler_->CreateTaskQueueThrottler();
+
+  scheduler_->OnThrottlingStateChanged(
+      FrameScheduler::ThrottlingState::kThrottled);
+
+  std::unique_ptr<WorkerSchedulerForTest> worker_scheduler2 =
+      std::make_unique<WorkerSchedulerForTest>(scheduler_.get());
+
+  // Ensure that newly created scheduler is throttled.
+  EXPECT_TRUE(scheduler_->task_queue_throttler()->IsThrottled(
+      worker_scheduler2->ThrottleableTaskQueue().get()));
+
+  worker_scheduler2->Dispose();
+}
+
+TEST_F(WorkerSchedulerTest, ThrottleWorkerScheduler_RunThrottledTasks) {
+  scheduler_->CreateTaskQueueThrottler();
+
+  // Create a new |worker_scheduler| to ensure that it's properly initialised.
+  worker_scheduler_->Dispose();
+  worker_scheduler_ =
+      std::make_unique<WorkerSchedulerForTest>(scheduler_.get());
+
+  scheduler_->OnThrottlingStateChanged(
+      FrameScheduler::ThrottlingState::kThrottled);
+
+  std::vector<base::TimeTicks> tasks;
+
+  worker_scheduler_->ThrottleableTaskQueue()->PostTask(
+      FROM_HERE,
+      base::BindOnce(&RunChainedTask,
+                     worker_scheduler_->ThrottleableTaskQueue(), 5,
+                     base::Unretained(GetClock()), base::Unretained(&tasks)));
+
+  RunUntilIdle();
+
+  EXPECT_THAT(tasks,
+              ElementsAre(base::TimeTicks() + base::TimeDelta::FromSeconds(1),
+                          base::TimeTicks() + base::TimeDelta::FromSeconds(2),
+                          base::TimeTicks() + base::TimeDelta::FromSeconds(3),
+                          base::TimeTicks() + base::TimeDelta::FromSeconds(4),
+                          base::TimeTicks() + base::TimeDelta::FromSeconds(5)));
+}
+
 }  // namespace worker_scheduler_unittest
 }  // namespace scheduler
 }  // namespace blink

third_party/blink/renderer/platform/scheduler/common/throttling/task_queue_throttler.cc

@@ -336,7 +336,7 @@ void TaskQueueThrottler::UpdateQueueThrottlingStateInternal(base::TimeTicks now,
     base::Optional<base::TimeTicks> unblock_until =
         GetTimeTasksCanRunUntil(queue, now, is_wake_up);
     DCHECK(unblock_until);
-    if (!unblock_until || unblock_until.value() > now) {
+    if (unblock_until.value() > now) {
       queue->InsertFenceAt(unblock_until.value());
     } else if (unblock_until.value() == now) {
       queue->InsertFence(TaskQueue::InsertFencePosition::kNow);

third_party/blink/renderer/platform/scheduler/public/frame_scheduler.h

@@ -27,6 +27,7 @@ class FrameScheduler : public FrameOrWorkerScheduler {
   enum class ObserverType { kLoader, kWorkerScheduler };
 
   // Represents throttling state.
+  // TODO(altimin): Move it into standalone LifecycleState.
   enum class ThrottlingState {
     // Frame is active and should not be throttled.
     kNotThrottled,

third_party/blink/renderer/platform/scheduler/public/non_main_thread_scheduler.h

@@ -16,6 +16,7 @@
 #include "third_party/blink/renderer/platform/scheduler/child/worker_task_queue.h"
 #include "third_party/blink/renderer/platform/scheduler/common/thread_scheduler_impl.h"
 #include "third_party/blink/renderer/platform/scheduler/public/thread_scheduler.h"
+#include "third_party/blink/renderer/platform/scheduler/util/tracing_helper.h"
 #include "third_party/blink/renderer/platform/scheduler/worker/non_main_thread_scheduler_helper.h"
 
 namespace blink {
@@ -23,6 +24,8 @@ namespace scheduler {
 class TaskQueueWithTaskType;
 class WorkerSchedulerProxy;
 class WorkerScheduler;
+class TaskQueueThrottler;
+class WakeUpBudgetPool;
 
 // TODO(yutak): Rename this class to NonMainThreadSchedulerImpl and consider
 // changing all non-impl scheduler classes to have only static methods.
@@ -93,26 +96,41 @@ class PLATFORM_EXPORT NonMainThreadScheduler : public ThreadSchedulerImpl {
 
   scoped_refptr<WorkerTaskQueue> CreateTaskRunner();
 
+  // TaskQueueThrottler might be null if throttling is not enabled or
+  // not supported.
+  TaskQueueThrottler* task_queue_throttler() const {
+    return task_queue_throttler_.get();
+  }
+  WakeUpBudgetPool* wake_up_budget_pool() const { return wake_up_budget_pool_; }
+
  protected:
   explicit NonMainThreadScheduler(
       std::unique_ptr<NonMainThreadSchedulerHelper> helper);
 
+  friend class WorkerScheduler;
+
+  // Each WorkerScheduler should notify NonMainThreadScheduler when it is
+  // created or destroyed.
+  virtual void RegisterWorkerScheduler(WorkerScheduler* worker_scheduler);
+  virtual void UnregisterWorkerScheduler(WorkerScheduler* worker_scheduler);
+
   // Called during Init() for delayed initialization for subclasses.
   virtual void InitImpl() = 0;
 
+  // This controller should be initialized before any TraceableVariables
+  // because they require one to initialize themselves.
+  TraceableVariableController traceable_variable_controller_;
+
   std::unique_ptr<NonMainThreadSchedulerHelper> helper_;
 
   // Worker schedulers associated with this thread.
   std::unordered_set<WorkerScheduler*> worker_schedulers_;
 
- private:
-  friend class WorkerScheduler;
-
-  // Each WorkerScheduler should notify NonMainThreadScheduler when it is
-  // created or destroyed.
-  void RegisterWorkerScheduler(WorkerScheduler* worker_scheduler);
-  void UnregisterWorkerScheduler(WorkerScheduler* worker_scheduler);
+  std::unique_ptr<TaskQueueThrottler> task_queue_throttler_;
+  // Owned by |task_queue_throttler_|.
+  WakeUpBudgetPool* wake_up_budget_pool_ = nullptr;
 
+ private:
   static void RunIdleTask(WebThread::IdleTask task, base::TimeTicks deadline);
   scoped_refptr<TaskQueueWithTaskType> v8_task_runner_;
 

third_party/blink/renderer/platform/scheduler/worker/non_main_thread_scheduler.cc

@@ -8,6 +8,8 @@
 
 #include "third_party/blink/public/platform/platform.h"
 #include "third_party/blink/renderer/platform/scheduler/child/task_queue_with_task_type.h"
+#include "third_party/blink/renderer/platform/scheduler/common/throttling/task_queue_throttler.h"
+#include "third_party/blink/renderer/platform/scheduler/common/throttling/wake_up_budget_pool.h"
 #include "third_party/blink/renderer/platform/scheduler/worker/worker_thread_scheduler.h"
 
 namespace blink {

third_party/blink/renderer/platform/scheduler/worker/worker_thread_scheduler.cc

@@ -15,8 +15,11 @@
 #include "third_party/blink/renderer/platform/scheduler/base/task_queue.h"
 #include "third_party/blink/renderer/platform/scheduler/base/task_queue_manager.h"
 #include "third_party/blink/renderer/platform/scheduler/child/default_params.h"
+#include "third_party/blink/renderer/platform/scheduler/child/features.h"
 #include "third_party/blink/renderer/platform/scheduler/child/task_queue_with_task_type.h"
+#include "third_party/blink/renderer/platform/scheduler/child/worker_scheduler.h"
 #include "third_party/blink/renderer/platform/scheduler/child/worker_scheduler_proxy.h"
+#include "third_party/blink/renderer/platform/scheduler/common/throttling/task_queue_throttler.h"
 #include "third_party/blink/renderer/platform/scheduler/worker/non_main_thread_scheduler_helper.h"
 
 namespace blink {
@@ -83,6 +86,11 @@ WorkerThreadScheduler::WorkerThreadScheduler(
     proxy->OnWorkerSchedulerCreated(GetWeakPtr());
   }
 
+  if (thread_type == WebThreadType::kDedicatedWorkerThread &&
+      base::FeatureList::IsEnabled(kDedicatedWorkerThrottling)) {
+    CreateTaskQueueThrottler();
+  }
+
   TRACE_EVENT_OBJECT_CREATED_WITH_ID(
       TRACE_DISABLED_BY_DEFAULT("worker.scheduler"), "WorkerScheduler", this);
 }
@@ -144,6 +152,7 @@ void WorkerThreadScheduler::Shutdown() {
   UMA_HISTOGRAM_CUSTOM_TIMES(
       "WorkerThread.Runtime", delta, base::TimeDelta::FromSeconds(1),
       base::TimeDelta::FromDays(1), 50 /* bucket count */);
+  task_queue_throttler_.reset();
   helper_->Shutdown();
 }
 
@@ -193,7 +202,18 @@ void WorkerThreadScheduler::DidProcessTask(double start_time, double end_time) {
 
 void WorkerThreadScheduler::OnThrottlingStateChanged(
     FrameScheduler::ThrottlingState throttling_state) {
+  if (throttling_state_ == throttling_state)
+    return;
   throttling_state_ = throttling_state;
+
+  for (WorkerScheduler* worker_scheduler : worker_schedulers_)
+    worker_scheduler->OnThrottlingStateChanged(throttling_state);
+}
+
+void WorkerThreadScheduler::RegisterWorkerScheduler(
+    WorkerScheduler* worker_scheduler) {
+  NonMainThreadScheduler::RegisterWorkerScheduler(worker_scheduler);
+  worker_scheduler->OnThrottlingStateChanged(throttling_state_);
 }
 
 scoped_refptr<WorkerTaskQueue> WorkerThreadScheduler::ControlTaskQueue() {
@@ -204,5 +224,14 @@ base::WeakPtr<WorkerThreadScheduler> WorkerThreadScheduler::GetWeakPtr() {
   return weak_factory_.GetWeakPtr();
 }
 
+void WorkerThreadScheduler::CreateTaskQueueThrottler() {
+  if (task_queue_throttler_)
+    return;
+  task_queue_throttler_ = std::make_unique<TaskQueueThrottler>(
+      this, &traceable_variable_controller_);
+  wake_up_budget_pool_ =
+      task_queue_throttler_->CreateWakeUpBudgetPool("worker_wake_up_pool");
+}
+
 }  // namespace scheduler
 }  // namespace blink

third_party/blink/renderer/platform/scheduler/worker/worker_thread_scheduler.h

@@ -94,6 +94,10 @@ class PLATFORM_EXPORT WorkerThreadScheduler
     return throttling_state_;
   }
 
+  void RegisterWorkerScheduler(WorkerScheduler* worker_scheduler) override;
+
+  void CreateTaskQueueThrottler();
+
  private:
   void MaybeStartLongIdlePeriod();