Reduce the number of delayed tasks on chromium run loop

Previously if there were N distinct delayed times for dealyed tasks
posted to the scheduler, there would be N delayed tasks posted on the
chromium run loop.  We only really need to post timers for the next
delayed task.  This patch implements that. Caveat: because it's not
possible to cancel tasks posted to the chromium runloop, we will still
have more than one task outstanding if delayed tasks get posted in
reverse order.

BUG=510398, 546953

Review URL: https://codereview.chromium.org/1468443002

Cr-Commit-Position: refs/heads/master@{#361086}

components/scheduler/base/real_time_domain.cc

@@ -10,29 +10,59 @@
 
 namespace scheduler {
 
-RealTimeDomain::RealTimeDomain(
+RealTimeDomain::RealTimeDomain() : weak_factory_(this) {}
+
+RealTimeDomain::~RealTimeDomain() {}
+
+void RealTimeDomain::OnRegisterWithTaskQueueManager(
     TaskQueueManagerDelegate* task_queue_manager_delegate,
-    base::Closure do_work_closure)
-    : task_queue_manager_delegate_(task_queue_manager_delegate),
-      do_work_closure_(do_work_closure) {
+    base::Closure do_work_closure) {
+  task_queue_manager_delegate_ = task_queue_manager_delegate;
+  do_work_closure_ = do_work_closure;
   DCHECK(task_queue_manager_delegate_);
 }
 
-RealTimeDomain::~RealTimeDomain() {}
-
 LazyNow RealTimeDomain::CreateLazyNow() {
+  DCHECK(task_queue_manager_delegate_);
   return LazyNow(task_queue_manager_delegate_);
 }
 
-void RealTimeDomain::RequestWakeup(base::TimeDelta delay) {
-  task_queue_manager_delegate_->PostDelayedTask(FROM_HERE, do_work_closure_,
-                                                delay);
+void RealTimeDomain::RequestWakeup(LazyNow* lazy_now, base::TimeDelta delay) {
+  PostWrappedDoWork(lazy_now->Now(), lazy_now->Now() + delay);
 }
 
 bool RealTimeDomain::MaybeAdvanceTime() {
+  base::TimeTicks next_run_time;
+  if (!NextScheduledRunTime(&next_run_time))
+    return false;
+
+  DCHECK(task_queue_manager_delegate_);
+  base::TimeTicks now = task_queue_manager_delegate_->NowTicks();
+  if (now >= next_run_time)
+    return true;
+
+  PostWrappedDoWork(now, next_run_time);
   return false;
 }
 
+void RealTimeDomain::PostWrappedDoWork(base::TimeTicks now,
+                                       base::TimeTicks run_time) {
+  DCHECK_GE(run_time, now);
+  DCHECK(task_queue_manager_delegate_);
+  if (pending_wakeups_.insert(run_time).second) {
+    task_queue_manager_delegate_->PostDelayedTask(
+        FROM_HERE,
+        base::Bind(&RealTimeDomain::WrappedDoWorkTask,
+                   weak_factory_.GetWeakPtr(), run_time),
+        run_time - now);
+  }
+}
+
+void RealTimeDomain::WrappedDoWorkTask(base::TimeTicks run_time) {
+  pending_wakeups_.erase(run_time);
+  do_work_closure_.Run();
+}
+
 void RealTimeDomain::AsValueIntoInternal(
     base::trace_event::TracedValue* state) const {}
 

components/scheduler/base/real_time_domain.h

@@ -5,7 +5,8 @@
 #ifndef COMPONENTS_SCHEDULER_BASE_REAL_TIME_DOMAIN_H_
 #define COMPONENTS_SCHEDULER_BASE_REAL_TIME_DOMAIN_H_
 
-#include "base/callback.h"
+#include <set>
+
 #include "base/macros.h"
 #include "components/scheduler/base/time_domain.h"
 #include "components/scheduler/scheduler_export.h"
@@ -15,8 +16,7 @@ class TaskQueueManagerDelegate;
 
 class SCHEDULER_EXPORT RealTimeDomain : public TimeDomain {
  public:
-  RealTimeDomain(TaskQueueManagerDelegate* task_queue_manager_delegate,
-                 base::Closure do_work_closure);
+  RealTimeDomain();
 
   // TimeDomain implementation:
   LazyNow CreateLazyNow() override;
@@ -24,13 +24,21 @@ class SCHEDULER_EXPORT RealTimeDomain : public TimeDomain {
   const char* GetName() const override;
 
  protected:
-  void RequestWakeup(base::TimeDelta delay) override;
+  void OnRegisterWithTaskQueueManager(
+      TaskQueueManagerDelegate* task_queue_manager_delegate,
+      base::Closure do_work_closure) override;
+  void RequestWakeup(LazyNow* lazy_now, base::TimeDelta delay) override;
   void AsValueIntoInternal(
       base::trace_event::TracedValue* state) const override;
 
  private:
+  void PostWrappedDoWork(base::TimeTicks now, base::TimeTicks run_time);
+  void WrappedDoWorkTask(base::TimeTicks run_time);
+
   TaskQueueManagerDelegate* task_queue_manager_delegate_;  // NOT OWNED
+  std::set<base::TimeTicks> pending_wakeups_;
   base::Closure do_work_closure_;
+  base::WeakPtrFactory<RealTimeDomain> weak_factory_;
 
   ~RealTimeDomain() override;
 

components/scheduler/base/task_queue_impl.cc

@@ -325,10 +325,6 @@ void TaskQueueImpl::EnqueueDelayedTaskLocked(const Task& pending_task) {
     return;
   if (any_thread().incoming_queue.empty())
     any_thread().time_domain->RegisterAsUpdatableTaskQueue(this);
-  if (any_thread().pump_policy == PumpPolicy::AUTO &&
-      any_thread().incoming_queue.empty()) {
-    any_thread().task_queue_manager->MaybePostDoWorkOnMainRunner();
-  }
   // TODO(alexclarke): consider std::move() when allowed.
   any_thread().incoming_queue.push(pending_task);
   any_thread().incoming_queue.back().set_enqueue_order(

components/scheduler/base/task_queue_manager.cc

@@ -43,8 +43,8 @@ TaskQueueManager::TaskQueueManager(
   do_work_closure_ =
       base::Bind(&TaskQueueManager::DoWork, weak_factory_.GetWeakPtr(), false);
 
-  real_time_domain_ =
-      make_scoped_refptr(new RealTimeDomain(delegate.get(), do_work_closure_));
+  // TODO(alexclarke): Change this to be a parameter that's passed in.
+  real_time_domain_ = make_scoped_refptr(new RealTimeDomain());
   RegisterTimeDomain(real_time_domain_);
 }
 
@@ -61,6 +61,8 @@ TaskQueueManager::~TaskQueueManager() {
 void TaskQueueManager::RegisterTimeDomain(
     const scoped_refptr<TimeDomain>& time_domain) {
   time_domains_.insert(time_domain);
+  time_domain->OnRegisterWithTaskQueueManager(delegate_.get(),
+                                              do_work_closure_);
 }
 
 void TaskQueueManager::UnregisterTimeDomain(

components/scheduler/base/task_queue_manager_unittest.cc

@@ -1458,4 +1458,28 @@ TEST_F(TaskQueueManagerTest, TimeDomainMigration) {
   manager_->UnregisterTimeDomain(domain_b);
 }
 
+namespace {
+void ChromiumRunloopInspectionTask(
+    scoped_refptr<cc::OrderedSimpleTaskRunner> test_task_runner) {
+  EXPECT_EQ(1u, test_task_runner->NumPendingTasks());
+}
+}  // namespace
+
+TEST_F(TaskQueueManagerTest, NumberOfPendingTasksOnChromiumRunLoop) {
+  Initialize(1u);
+
+  // NOTE because tasks posted to the chromiumrun loop are not cancellable, we
+  // will end up with a lot more tasks posted if the delayed tasks were posted
+  // in the reverse order.
+  // TODO(alexclarke): Consider talking to the message pump directly.
+  test_task_runner_->SetAutoAdvanceNowToPendingTasks(true);
+  for (int i = 1; i < 100; i++) {
+    runners_[0]->PostDelayedTask(
+        FROM_HERE,
+        base::Bind(&ChromiumRunloopInspectionTask, test_task_runner_),
+        base::TimeDelta::FromMilliseconds(i));
+  }
+  test_task_runner_->RunUntilIdle();
+}
+
 }  // namespace scheduler

components/scheduler/base/time_domain.cc

@@ -12,7 +12,7 @@
 
 namespace scheduler {
 
-TimeDomain::TimeDomain() : weak_factory_(this) {}
+TimeDomain::TimeDomain() {}
 
 TimeDomain::~TimeDomain() {}
 
@@ -31,7 +31,7 @@ void TimeDomain::UnregisterQueue(internal::TaskQueueImpl* queue) {
     }
   }
 
-  // |newly_updatable_| might contain |task_queue|, we use
+  // |newly_updatable_| might contain |queue|, we use
   // MoveNewlyUpdatableQueuesIntoUpdatableQueueSet to flush it out.
   MoveNewlyUpdatableQueuesIntoUpdatableQueueSet();
   updatable_queue_set_.erase(queue);
@@ -66,12 +66,12 @@ void TimeDomain::ScheduleDelayedWork(internal::TaskQueueImpl* queue,
                                      base::TimeTicks delayed_run_time,
                                      LazyNow* lazy_now) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
-  // Dedupe wakeups.
-  if (delayed_wakeup_multimap_.find(delayed_run_time) ==
-      delayed_wakeup_multimap_.end()) {
+
+  if (delayed_wakeup_multimap_.empty() ||
+      delayed_run_time < delayed_wakeup_multimap_.begin()->first) {
     base::TimeDelta delay =
         std::max(base::TimeDelta(), delayed_run_time - lazy_now->Now());
-    RequestWakeup(delay);
+    RequestWakeup(lazy_now, delay);
   }
   delayed_wakeup_multimap_.insert(std::make_pair(delayed_run_time, queue));
 }

components/scheduler/base/time_domain.h

@@ -7,6 +7,7 @@
 
 #include <map>
 
+#include "base/callback.h"
 #include "base/macros.h"
 #include "base/memory/ref_counted.h"
 #include "base/memory/weak_ptr.h"
@@ -20,6 +21,7 @@ namespace internal {
 class TaskQueueImpl;
 }  // internal
 class TaskQueueManager;
+class TaskQueueManagerDelegate;
 
 class SCHEDULER_EXPORT TimeDomain : public base::RefCounted<TimeDomain> {
  public:
@@ -79,9 +81,14 @@ class SCHEDULER_EXPORT TimeDomain : public base::RefCounted<TimeDomain> {
   void UpdateWorkQueues(bool should_trigger_wakeup,
                         const internal::TaskQueueImpl::Task* previous_task);
 
+  // Called by the TaskQueueManager when the TimeDomain is registered.
+  virtual void OnRegisterWithTaskQueueManager(
+      TaskQueueManagerDelegate* task_queue_manager_delegate,
+      base::Closure do_work_closure) = 0;
+
   // The implementaion will secedule task processing to run with |delay| with
   // respect to the TimeDomain's time source.
-  virtual void RequestWakeup(base::TimeDelta delay) = 0;
+  virtual void RequestWakeup(LazyNow* lazy_now, base::TimeDelta delay) = 0;
 
   // For implementation specific tracing.
   virtual void AsValueIntoInternal(
@@ -109,7 +116,6 @@ class SCHEDULER_EXPORT TimeDomain : public base::RefCounted<TimeDomain> {
   std::set<internal::TaskQueueImpl*> updatable_queue_set_;
 
   base::ThreadChecker main_thread_checker_;
-  base::WeakPtrFactory<TimeDomain> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(TimeDomain);
 };

components/scheduler/base/time_domain_unittest.cc

@@ -13,6 +13,7 @@
 #include "testing/gmock/include/gmock/gmock.h"
 
 using testing::_;
+using testing::Mock;
 
 namespace scheduler {
 
@@ -34,10 +35,12 @@ class MockTimeDomain : public TimeDomain {
       base::trace_event::TracedValue* state) const override {}
 
   bool MaybeAdvanceTime() override { return false; }
-
   const char* GetName() const override { return "Test"; }
+  void OnRegisterWithTaskQueueManager(
+      TaskQueueManagerDelegate* task_queue_manager_delegate,
+      base::Closure do_work_closure) override {}
 
-  MOCK_METHOD1(RequestWakeup, void(base::TimeDelta delay));
+  MOCK_METHOD2(RequestWakeup, void(LazyNow* lazy_now, base::TimeDelta delay));
 
   void SetNow(base::TimeTicks now) { now_ = now; }
 
@@ -67,7 +70,7 @@ class TimeDomainTest : public testing::Test {
 TEST_F(TimeDomainTest, ScheduleDelayedWork) {
   base::TimeDelta delay = base::TimeDelta::FromMilliseconds(10);
   base::TimeTicks delayed_runtime = time_domain_->Now() + delay;
-  EXPECT_CALL(*time_domain_.get(), RequestWakeup(delay));
+  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_, delay));
   LazyNow lazy_now = time_domain_->CreateLazyNow();
   time_domain_->ScheduleDelayedWork(task_queue_.get(),
                                     time_domain_->Now() + delay, &lazy_now);
@@ -81,13 +84,41 @@ TEST_F(TimeDomainTest, ScheduleDelayedWork) {
   EXPECT_EQ(task_queue_.get(), next_task_queue);
 }
 
+TEST_F(TimeDomainTest, RequestWakeup_OnlyCalledForEarlierTasks) {
+  base::TimeDelta delay1 = base::TimeDelta::FromMilliseconds(10);
+  base::TimeDelta delay2 = base::TimeDelta::FromMilliseconds(20);
+  base::TimeDelta delay3 = base::TimeDelta::FromMilliseconds(30);
+  base::TimeDelta delay4 = base::TimeDelta::FromMilliseconds(1);
+
+  // RequestWakeup should always be called if there are no other wakeups.
+  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_, delay1));
+  LazyNow lazy_now = time_domain_->CreateLazyNow();
+  time_domain_->ScheduleDelayedWork(task_queue_.get(),
+                                    time_domain_->Now() + delay1, &lazy_now);
+
+  Mock::VerifyAndClearExpectations(time_domain_.get());
+
+  // RequestWakeup should not be called when scheduling later tasks.
+  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_, _)).Times(0);
+  time_domain_->ScheduleDelayedWork(task_queue_.get(),
+                                    time_domain_->Now() + delay2, &lazy_now);
+  time_domain_->ScheduleDelayedWork(task_queue_.get(),
+                                    time_domain_->Now() + delay3, &lazy_now);
+
+  // RequestWakeup should be called when scheduling earlier tasks.
+  Mock::VerifyAndClearExpectations(time_domain_.get());
+  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_, delay4));
+  time_domain_->ScheduleDelayedWork(task_queue_.get(),
+                                    time_domain_->Now() + delay4, &lazy_now);
+}
+
 TEST_F(TimeDomainTest, UnregisterQueue) {
   scoped_refptr<internal::TaskQueueImpl> task_queue2_ =
       make_scoped_refptr(new internal::TaskQueueImpl(
           nullptr, time_domain_, TaskQueue::Spec("test_queue2"),
           "test.category", "test.category"));
 
-  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_)).Times(2);
+  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_, _)).Times(1);
   LazyNow lazy_now = time_domain_->CreateLazyNow();
   time_domain_->ScheduleDelayedWork(
       task_queue_.get(),
@@ -132,7 +163,7 @@ TEST_F(TimeDomainTest, UpdateWorkQueues) {
   // MoveReadyDelayedTasksToIncomingQueue as expected.
   base::TimeDelta delay = base::TimeDelta::FromMilliseconds(50);
   base::TimeTicks delayed_runtime = time_domain_->Now() + delay;
-  EXPECT_CALL(*time_domain_.get(), RequestWakeup(delay));
+  EXPECT_CALL(*time_domain_.get(), RequestWakeup(_, delay));
   LazyNow lazy_now = time_domain_->CreateLazyNow();
   time_domain_->ScheduleDelayedWork(dummy_queue.get(), delayed_runtime,
                                     &lazy_now);

components/scheduler/base/virtual_time_domain.cc

@@ -15,13 +15,22 @@ VirtualTimeDomain::VirtualTimeDomain(base::TimeTicks initial_time)
 
 VirtualTimeDomain::~VirtualTimeDomain() {}
 
+void VirtualTimeDomain::OnRegisterWithTaskQueueManager(
+    TaskQueueManagerDelegate* task_queue_manager_delegate,
+    base::Closure do_work_closure) {
+  task_queue_manager_delegate_ = task_queue_manager_delegate;
+  do_work_closure_ = do_work_closure;
+  DCHECK(task_queue_manager_delegate_);
+}
+
 LazyNow VirtualTimeDomain::CreateLazyNow() {
   base::AutoLock lock(lock_);
   return LazyNow(now_);
 }
 
-void VirtualTimeDomain::RequestWakeup(base::TimeDelta delay) {
-  // We don't need to do anything here because AdvanceTo triggers delayed tasks.
+void VirtualTimeDomain::RequestWakeup(LazyNow* lazy_now,
+                                      base::TimeDelta delay) {
+  // We don't need to do anything here because AdvanceTo posts a DoWork.
 }
 
 bool VirtualTimeDomain::MaybeAdvanceTime() {
@@ -35,8 +44,7 @@ void VirtualTimeDomain::AdvanceTo(base::TimeTicks now) {
   base::AutoLock lock(lock_);
   DCHECK_GE(now, now_);
   now_ = now;
-  LazyNow lazy_now(now_);
-  WakeupReadyDelayedQueues(&lazy_now);
+  task_queue_manager_delegate_->PostTask(FROM_HERE, do_work_closure_);
 }
 
 const char* VirtualTimeDomain::GetName() const {

components/scheduler/base/virtual_time_domain.h

@@ -27,7 +27,10 @@ class SCHEDULER_EXPORT VirtualTimeDomain : public TimeDomain {
   void AdvanceTo(base::TimeTicks now);
 
  protected:
-  void RequestWakeup(base::TimeDelta delay) override;
+  void OnRegisterWithTaskQueueManager(
+      TaskQueueManagerDelegate* task_queue_manager_delegate,
+      base::Closure do_work_closure) override;
+  void RequestWakeup(LazyNow* lazy_now, base::TimeDelta delay) override;
   void AsValueIntoInternal(
       base::trace_event::TracedValue* state) const override;
 
@@ -35,6 +38,9 @@ class SCHEDULER_EXPORT VirtualTimeDomain : public TimeDomain {
   mutable base::Lock lock_;  // Protects |now_|.
   base::TimeTicks now_;
 
+  TaskQueueManagerDelegate* task_queue_manager_delegate_;  // NOT OWNED
+  base::Closure do_work_closure_;
+
   ~VirtualTimeDomain() override;
 
   DISALLOW_COPY_AND_ASSIGN(VirtualTimeDomain);