Handle delayed tasks more natively in the scheduler

content/renderer/scheduler/task_queue_manager.cc

 // Copyright 2014 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 "content/renderer/scheduler/task_queue_manager.h"
 
 #include <queue>
+#include <set>
 
 #include "base/bind.h"
 #include "base/trace_event/trace_event.h"
 #include "base/trace_event/trace_event_argument.h"
 #include "cc/test/test_now_source.h"
 #include "content/renderer/scheduler/nestable_single_thread_task_runner.h"
 #include "content/renderer/scheduler/task_queue_selector.h"
 
 namespace {
 const int64_t kMaxTimeTicks = std::numeric_limits<int64>::max();
 }
 
 namespace content {
 namespace internal {
 
+// Now() is somewhat expensive so it makes sense not to call Now() unless we
+// really need to.
+class LazyNow {
+ public:
+  explicit LazyNow(base::TimeTicks now)
+      : task_queue_manager_(nullptr), now_(now) {
+    DCHECK(!now.is_null());
+  }
+
+  explicit LazyNow(TaskQueueManager* task_queue_manager)
+      : task_queue_manager_(task_queue_manager) {}
+
+  base::TimeTicks Now() {
+    if (now_.is_null())
+      now_ = task_queue_manager_->Now();
+    return now_;
+  }
+
+ private:
+  TaskQueueManager* task_queue_manager_;  // NOT OWNED
+  base::TimeTicks now_;
+};
+
 class TaskQueue : public base::SingleThreadTaskRunner {
  public:
   TaskQueue(TaskQueueManager* task_queue_manager);
 
   // base::SingleThreadTaskRunner implementation.
   bool RunsTasksOnCurrentThread() const override;
   bool PostDelayedTask(const tracked_objects::Location& from_here,
                        const base::Closure& task,
                        base::TimeDelta delay) override {
     return PostDelayedTaskImpl(from_here, task, delay, TaskType::NORMAL);
   }
 
   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
                                   const base::Closure& task,
                                   base::TimeDelta delay) override {
     return PostDelayedTaskImpl(from_here, task, delay, TaskType::NON_NESTABLE);
   }
 
   bool IsQueueEmpty() const;
 
   void SetPumpPolicy(TaskQueueManager::PumpPolicy pump_policy);
   void PumpQueue();
 
-  bool UpdateWorkQueue(base::TimeTicks* next_pending_delayed_task,
+  bool UpdateWorkQueue(LazyNow* lazy_now,
                        const base::PendingTask* previous_task);
   base::PendingTask TakeTaskFromWorkQueue();
 
   void WillDeleteTaskQueueManager();
 
   base::TaskQueue& work_queue() { return work_queue_; }
 
   void set_name(const char* name) { name_ = name; }
 
   void AsValueInto(base::trace_event::TracedValue* state) const;
 
  private:
   enum class TaskType {
     NORMAL,
     NON_NESTABLE,
   };
 
   ~TaskQueue() override;
 
   bool PostDelayedTaskImpl(const tracked_objects::Location& from_here,
                            const base::Closure& task,
                            base::TimeDelta delay,
                            TaskType task_type);
 
-  // Adds a task at the end of the incoming task queue and schedules a call to
-  // TaskQueueManager::DoWork() if the incoming queue was empty and automatic
-  // pumping is enabled. Can be called on an arbitrary thread.
-  void EnqueueTask(const base::PendingTask& pending_task);
+  // Delayed task posted to the underlying run loop, which calls
+  // ScheduleDelayedWorkLocked to enqueue any delayed tasks which should be run
+  // now.
+  void KickDelayedTasks();
+
+  // Posts KickDelayedTasks if there isn't already a task posted on the
+  // underlying runloop for the next task's scheduled run time.
+  void ScheduleDelayedWorkLocked(LazyNow* lazy_now);
+
+  // Enqueues any delayed tasks which should be run now on the incomming_queue_.

[Sami, 2015/03/17 18:40:02]

s/incomming/incoming/

[alex clarke (OOO till 29th), 2015/03/18 10:49:07]

Done.

+  void MoveReadyDelayedTasksToIncomingQueueLocked(LazyNow* lazy_now);
 
   void PumpQueueLocked();
   bool TaskIsOlderThanQueuedTasks(const base::PendingTask* task);
   bool ShouldAutoPumpQueueLocked(const base::PendingTask* previous_task);
   void EnqueueTaskLocked(const base::PendingTask& pending_task);
 
   void TraceQueueSize(bool is_locked) const;
   static const char* PumpPolicyToString(
       TaskQueueManager::PumpPolicy pump_policy);
   static void QueueAsValueInto(const base::TaskQueue& queue,
                                base::trace_event::TracedValue* state);
+  static void QueueAsValueInto(const base::DelayedTaskQueue& queue,
+                               base::trace_event::TracedValue* state);
   static void TaskAsValueInto(const base::PendingTask& task,
                               base::trace_event::TracedValue* state);
 
   // This lock protects all members except the work queue.
   mutable base::Lock lock_;
   base::PlatformThreadId thread_id_;
   TaskQueueManager* task_queue_manager_;
   base::TaskQueue incoming_queue_;
   TaskQueueManager::PumpPolicy pump_policy_;
   const char* name_;
-  std::priority_queue<base::TimeTicks,
-                      std::vector<base::TimeTicks>,
-                      std::greater<base::TimeTicks>> delayed_task_run_times_;
+  base::DelayedTaskQueue delayed_task_queue_;
+  std::set<base::TimeTicks> in_flight_kick_delayed_tasks_;
+  base::ThreadChecker main_thread_checker_;
 
   base::TaskQueue work_queue_;
 
   DISALLOW_COPY_AND_ASSIGN(TaskQueue);
 };
 
 TaskQueue::TaskQueue(TaskQueueManager* task_queue_manager)
     : thread_id_(base::PlatformThread::CurrentId()),
       task_queue_manager_(task_queue_manager),
       pump_policy_(TaskQueueManager::PumpPolicy::AUTO),
       name_(nullptr) {
 }
 
 TaskQueue::~TaskQueue() {
 }
 
 void TaskQueue::WillDeleteTaskQueueManager() {
   base::AutoLock lock(lock_);
   task_queue_manager_ = nullptr;
+  delayed_task_queue_ = base::DelayedTaskQueue();

[Sami, 2015/03/17 18:40:03]

Re: the previous discussion about the MessageLoop, we are purposefully
discarding all the tasks and not running them like MessageLoop does to ensure
anything with a pointer to Blink goes away.

It's probably a better task for another patch, but I think we should be clearing
all the queues here. Mind adding a TODO here?

[alex clarke (OOO till 29th), 2015/03/18 10:49:07]

Done.

 }
 
 bool TaskQueue::RunsTasksOnCurrentThread() const {
   base::AutoLock lock(lock_);
   return base::PlatformThread::CurrentId() == thread_id_;
 }
 
 bool TaskQueue::PostDelayedTaskImpl(const tracked_objects::Location& from_here,
                                     const base::Closure& task,
                                     base::TimeDelta delay,
                                     TaskType task_type) {
   base::AutoLock lock(lock_);
   if (!task_queue_manager_)
     return false;
 
   base::PendingTask pending_task(from_here, task, base::TimeTicks(),
                                  task_type != TaskType::NON_NESTABLE);
   task_queue_manager_->DidQueueTask(&pending_task);
 
   if (delay > base::TimeDelta()) {
-    pending_task.delayed_run_time = task_queue_manager_->Now() + delay;
-    delayed_task_run_times_.push(pending_task.delayed_run_time);
-    return task_queue_manager_->PostDelayedTask(
-        FROM_HERE, Bind(&TaskQueue::EnqueueTask, this, pending_task), delay);
+    base::TimeTicks now = task_queue_manager_->Now();
+    pending_task.delayed_run_time = now + delay;
+    delayed_task_queue_.push(pending_task);
+    // If we changed the topmost task, then it is time to reschedule.
+    if (delayed_task_queue_.top().task.Equals(pending_task.task)) {
+      LazyNow lazy_now(now);
+      ScheduleDelayedWorkLocked(&lazy_now);
+    }
+    return true;
   }
   EnqueueTaskLocked(pending_task);
   return true;
 }
 
+void TaskQueue::MoveReadyDelayedTasksToIncomingQueueLocked(LazyNow* lazy_now) {
+  lock_.AssertAcquired();
+  // Enqueue all delayed tasks that should be running now.
+  while (!delayed_task_queue_.empty() &&
+         delayed_task_queue_.top().delayed_run_time <= lazy_now->Now()) {
+    in_flight_kick_delayed_tasks_.erase(
+        delayed_task_queue_.top().delayed_run_time);

[Sami, 2015/03/17 18:40:02]

Did we lose the code that clears the delayed_run_time here?

[alex clarke (OOO till 29th), 2015/03/18 10:49:07]

It's not needed.  See the implementation of EnqueueTaskLocked.

+    EnqueueTaskLocked(delayed_task_queue_.top());
+    delayed_task_queue_.pop();
+  }
+}
+
+void TaskQueue::ScheduleDelayedWorkLocked(LazyNow* lazy_now) {
+  lock_.AssertAcquired();
+  MoveReadyDelayedTasksToIncomingQueueLocked(lazy_now);

[Sami, 2015/03/17 18:40:03]

Would you mind moving this call to KickDelayedTasks for now? It feels weird to
penalize the posting thread to do this housekeeping work for us, and I'm not
sure it buys us all that much because we have to go through DoWork to run the
tasks anyway -- might as well have the migration happen there.

[alex clarke (OOO till 29th), 2015/03/18 10:49:07]

Done.

+  // Any remaining tasks are in the future, so queue a task to kick them.
+  if (!delayed_task_queue_.empty()) {
+    base::TimeTicks next_run_time = delayed_task_queue_.top().delayed_run_time;
+    DCHECK(next_run_time > lazy_now->Now());
+    // Make sure we don't have more than one KickDelayedTasks posted for a
+    // particular scheduled run time (note it's fine to have multiple ones in
+    // flight for distinct run times).
+    if (in_flight_kick_delayed_tasks_.find(next_run_time) ==
+        in_flight_kick_delayed_tasks_.end()) {
+      in_flight_kick_delayed_tasks_.insert(next_run_time);
+      base::TimeDelta delay = next_run_time - lazy_now->Now();
+      task_queue_manager_->PostDelayedTask(
+          FROM_HERE, Bind(&TaskQueue::KickDelayedTasks, this), delay);
+    }
+  }
+}
+
+void TaskQueue::KickDelayedTasks() {

[rmcilroy, 2015/03/17 19:06:54]

Naming nit - could we just call this ScheduleDelayedWork (or call the one above
KickDelayedWorkLocked) for clarity?

[alex clarke (OOO till 29th), 2015/03/18 10:49:07]

I agree with the sentiment that KickDelayedTasks isn't the best name.  However I
think MoveReadyDelayedTasksToIncomingQueue is a better name.

[rmcilroy, 2015/03/18 11:23:29]

Works for me!

+  DCHECK(main_thread_checker_.CalledOnValidThread());
+  if (!task_queue_manager_)
+    return;
+
+  base::AutoLock lock(lock_);
+  LazyNow lazy_now(task_queue_manager_);
+  ScheduleDelayedWorkLocked(&lazy_now);
+}
+
 bool TaskQueue::IsQueueEmpty() const {
   if (!work_queue_.empty())
     return false;
 
   {
     base::AutoLock lock(lock_);
     return incoming_queue_.empty();
   }
 }
 
@@ skipping 25 matching lines @@
   if (pump_policy_ == TaskQueueManager::PumpPolicy::MANUAL)
     return false;
   if (pump_policy_ == TaskQueueManager::PumpPolicy::AFTER_WAKEUP &&
       TaskIsOlderThanQueuedTasks(previous_task))
     return false;
   if (incoming_queue_.empty())
     return false;
   return true;
 }
 
-bool TaskQueue::UpdateWorkQueue(
-    base::TimeTicks* next_pending_delayed_task,
-    const base::PendingTask* previous_task) {
+bool TaskQueue::UpdateWorkQueue(LazyNow* lazy_now,
+                                const base::PendingTask* previous_task) {
   if (!work_queue_.empty())
     return true;
 
   {
     base::AutoLock lock(lock_);
-    if (!delayed_task_run_times_.empty()) {
-      *next_pending_delayed_task =
-          std::min(*next_pending_delayed_task, delayed_task_run_times_.top());
-    }
+    MoveReadyDelayedTasksToIncomingQueueLocked(lazy_now);

[Sami, 2015/03/17 18:40:03]

Should this happen after the mode check below? It doesn't seem useful to do it
if pumping is disabled.

[alex clarke (OOO till 29th), 2015/03/18 10:49:07]

Done.

     if (!ShouldAutoPumpQueueLocked(previous_task))
       return false;
     work_queue_.Swap(&incoming_queue_);
     TraceQueueSize(true);
     return true;
   }
 }
 
 base::PendingTask TaskQueue::TakeTaskFromWorkQueue() {
   base::PendingTask pending_task = work_queue_.front();
   work_queue_.pop();
   TraceQueueSize(false);
   return pending_task;
 }
 
 void TaskQueue::TraceQueueSize(bool is_locked) const {
   bool is_tracing;
   TRACE_EVENT_CATEGORY_GROUP_ENABLED(
       TRACE_DISABLED_BY_DEFAULT("renderer.scheduler"), &is_tracing);
   if (!is_tracing || !name_)
     return;
   if (!is_locked)
     lock_.Acquire();
   else
     lock_.AssertAcquired();
-  TRACE_COUNTER1(TRACE_DISABLED_BY_DEFAULT("renderer.scheduler"), name_,
-                 incoming_queue_.size() + work_queue_.size());
+  TRACE_COUNTER1(
+      TRACE_DISABLED_BY_DEFAULT("renderer.scheduler"), name_,
+      incoming_queue_.size() + work_queue_.size() + delayed_task_queue_.size());
   if (!is_locked)
     lock_.Release();
 }
 
-void TaskQueue::EnqueueTask(const base::PendingTask& pending_task) {
-  base::AutoLock lock(lock_);
-  EnqueueTaskLocked(pending_task);
-}
-
 void TaskQueue::EnqueueTaskLocked(const base::PendingTask& pending_task) {
   lock_.AssertAcquired();
   if (!task_queue_manager_)
     return;
   if (pump_policy_ == TaskQueueManager::PumpPolicy::AUTO &&
       incoming_queue_.empty())
     task_queue_manager_->MaybePostDoWorkOnMainRunner();
   incoming_queue_.push(pending_task);
 
   if (!pending_task.delayed_run_time.is_null()) {
-    // Update the time of the next pending delayed task.
-    while (!delayed_task_run_times_.empty() &&
-           delayed_task_run_times_.top() <= pending_task.delayed_run_time) {
-      delayed_task_run_times_.pop();
-    }
     // Clear the delayed run time because we've already applied the delay
     // before getting here.
     incoming_queue_.back().delayed_run_time = base::TimeTicks();
   }
   TraceQueueSize(true);
 }
 
 void TaskQueue::SetPumpPolicy(TaskQueueManager::PumpPolicy pump_policy) {
   base::AutoLock lock(lock_);
   if (pump_policy == TaskQueueManager::PumpPolicy::AUTO &&
       pump_policy_ != TaskQueueManager::PumpPolicy::AUTO) {
     PumpQueueLocked();
   }
   pump_policy_ = pump_policy;
 }
 
 void TaskQueue::PumpQueueLocked() {
   lock_.AssertAcquired();
+  if (task_queue_manager_) {
+    LazyNow lazy_now(task_queue_manager_);
+    MoveReadyDelayedTasksToIncomingQueueLocked(&lazy_now);
+  }
   while (!incoming_queue_.empty()) {
     work_queue_.push(incoming_queue_.front());
     incoming_queue_.pop();
   }
   if (!work_queue_.empty())
     task_queue_manager_->MaybePostDoWorkOnMainRunner();
 }
 
 void TaskQueue::PumpQueue() {
   base::AutoLock lock(lock_);
   PumpQueueLocked();
 }
 
 void TaskQueue::AsValueInto(base::trace_event::TracedValue* state) const {
   base::AutoLock lock(lock_);
   state->BeginDictionary();
   if (name_)
     state->SetString("name", name_);
   state->SetString("pump_policy", PumpPolicyToString(pump_policy_));
   state->BeginArray("incoming_queue");
   QueueAsValueInto(incoming_queue_, state);
   state->EndArray();
   state->BeginArray("work_queue");
   QueueAsValueInto(work_queue_, state);
+  state->BeginArray("delayed_task_queue");
+  QueueAsValueInto(delayed_task_queue_, state);
   state->EndArray();
   state->EndDictionary();
 }
 
 // static
 const char* TaskQueue::PumpPolicyToString(
     TaskQueueManager::PumpPolicy pump_policy) {
   switch (pump_policy) {
     case TaskQueueManager::PumpPolicy::AUTO:
       return "auto";
@@ skipping 11 matching lines @@
 void TaskQueue::QueueAsValueInto(const base::TaskQueue& queue,
                                  base::trace_event::TracedValue* state) {
   base::TaskQueue queue_copy(queue);
   while (!queue_copy.empty()) {
     TaskAsValueInto(queue_copy.front(), state);
     queue_copy.pop();
   }
 }
 
 // static
+void TaskQueue::QueueAsValueInto(const base::DelayedTaskQueue& queue,
+                                 base::trace_event::TracedValue* state) {
+  base::DelayedTaskQueue queue_copy(queue);
+  while (!queue_copy.empty()) {
+    TaskAsValueInto(queue_copy.top(), state);
+    queue_copy.pop();
+  }
+}
+
+// static
 void TaskQueue::TaskAsValueInto(const base::PendingTask& task,
                                 base::trace_event::TracedValue* state) {
   state->BeginDictionary();
   state->SetString("posted_from", task.posted_from.ToString());
   state->SetInteger("sequence_num", task.sequence_num);
   state->SetBoolean("nestable", task.nestable);
   state->SetBoolean("is_high_res", task.is_high_res);
   state->SetDouble(
       "delayed_run_time",
       (task.delayed_run_time - base::TimeTicks()).InMicroseconds() / 1000.0L);
@@ skipping 60 matching lines @@
   queue->SetPumpPolicy(pump_policy);
 }
 
 void TaskQueueManager::PumpQueue(size_t queue_index) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   internal::TaskQueue* queue = Queue(queue_index);
   queue->PumpQueue();
 }
 
 bool TaskQueueManager::UpdateWorkQueues(
-    base::TimeTicks* next_pending_delayed_task,
     const base::PendingTask* previous_task) {
   // TODO(skyostil): This is not efficient when the number of queues grows very
   // large due to the number of locks taken. Consider optimizing when we get
   // there.
   DCHECK(main_thread_checker_.CalledOnValidThread());
+  internal::LazyNow lazy_now(this);
   bool has_work = false;
   for (auto& queue : queues_) {
-    has_work |= queue->UpdateWorkQueue(next_pending_delayed_task,
-                                       previous_task);
+    has_work |= queue->UpdateWorkQueue(&lazy_now, previous_task);
     if (!queue->work_queue().empty()) {
       // Currently we should not be getting tasks with delayed run times in any
       // of the work queues.
       DCHECK(queue->work_queue().front().delayed_run_time.is_null());
     }
   }
   return has_work;
 }
 
 void TaskQueueManager::MaybePostDoWorkOnMainRunner() {
@@ skipping 12 matching lines @@
                       on_main_thread));
 }
 
 void TaskQueueManager::DoWork(bool posted_from_main_thread) {
   if (posted_from_main_thread) {
     pending_dowork_count_--;
     DCHECK_GE(pending_dowork_count_, 0);
   }
   DCHECK(main_thread_checker_.CalledOnValidThread());
 
-  base::TimeTicks next_pending_delayed_task(
-      base::TimeTicks::FromInternalValue(kMaxTimeTicks));
-
-  // Pass nullptr to UpdateWorkQueues here to prevent waking up an
+  // Pass nullptr to UpdateWorkQueues here to prevent waking up a
   // pump-after-wakeup queue.
-  if (!UpdateWorkQueues(&next_pending_delayed_task, nullptr))
+  if (!UpdateWorkQueues(nullptr))
     return;
 
   base::PendingTask previous_task((tracked_objects::Location()),
                                   (base::Closure()));
   for (int i = 0; i < work_batch_size_; i++) {
-    // Interrupt the work batch if we should run the next delayed task.
-    if (i > 0 && next_pending_delayed_task.ToInternalValue() != kMaxTimeTicks &&
-        Now() >= next_pending_delayed_task)
-      return;
-
     size_t queue_index;
     if (!SelectWorkQueueToService(&queue_index))
       return;
     // Note that this function won't post another call to DoWork if one is
     // already pending, so it is safe to call it in a loop.
     MaybePostDoWorkOnMainRunner();
     ProcessTaskFromWorkQueue(queue_index, i > 0, &previous_task);
 
-    if (!UpdateWorkQueues(&next_pending_delayed_task, &previous_task))
+    if (!UpdateWorkQueues(&previous_task))
       return;
   }
 }
 
 bool TaskQueueManager::SelectWorkQueueToService(size_t* out_queue_index) {
   bool should_run = selector_->SelectWorkQueueToService(out_queue_index);
   TRACE_EVENT_OBJECT_SNAPSHOT_WITH_ID(
       TRACE_DISABLED_BY_DEFAULT("renderer.scheduler"), "TaskQueueManager", this,
       AsValueWithSelectorResult(should_run, *out_queue_index));
   return should_run;
@@ skipping 93 matching lines @@
   state->EndArray();
   state->BeginDictionary("selector");
   selector_->AsValueInto(state.get());
   state->EndDictionary();
   if (should_run)
     state->SetInteger("selected_queue", selected_queue);
   return state;
 }
 
 }  // namespace content