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 {
 
+class LazyNow {

[Sami, 2015/03/17 15:54:45]

Neat helper!

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Acknowledged.

+ 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 chromium run loop, which calls

[Sami, 2015/03/17 15:54:44]

nit: Replace "chromium" with "base" or "underlying" since this is all in
chromium anyway. Also s/timers/tasks/ (here and below).

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

+  // ScheduleDelayedWorkLocked to enqueue any delayed timers which should be run
+  // now.
+  void KickDelayedTasks();
+
+  // Enqueues any delayed timers which should be run now, and (maybe) posts
+  // KickDelayedTasks if there isn't already a timer posted on the chromium
+  // runloop for the next task's scheduled run time.
+  bool ScheduleDelayedWorkLocked(LazyNow* lazy_now);
+
+  // Enqueues any delayed timers which should be run now.
+  bool EnqueueReadyDelayedTasksLocked(LazyNow* lazy_now);

[Sami, 2015/03/17 15:54:45]

We should come up with a better name for this because it's too easy to confuse
with EnqueueTaskLocked which puts stuff on the incoming queue as opposed to the
work queue.

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

How about MoveReadyDelayedTasksToWorkQueueLocked?

 
   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 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::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;
+
+  // Clear all delayed tasks because we need them to be deleted before the blink

[Sami, 2015/03/17 15:54:44]

TQM shouldn't really know anything about blink so I'd remove this comment.

By the way, should we be clearing the incoming and work task queues here too?

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Probably no since the existing implementations of base::MessageLoop try to run
any remaining tasks (up to a limit of 100 tasks iirc).

+  // heap goes away.
+  while (!delayed_task_queue_.empty())

[Sami, 2015/03/17 15:54:45]

nit: would delayed_task_queue_ = base::DelayedTaskQueue() work? Probably no
reason to do this in order.

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

+    delayed_task_queue_.pop();
 }
 
 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);
+      return ScheduleDelayedWorkLocked(&lazy_now);
+    }
+    return true;
   }
   EnqueueTaskLocked(pending_task);
   return true;
 }
 
+bool TaskQueue::EnqueueReadyDelayedTasksLocked(LazyNow* lazy_now) {
+  lock_.AssertAcquired();

[Sami, 2015/03/17 15:54:45]

Could you DCHECK that we're on the main thread here?

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

+  bool task_enqueued = false;
+  // 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);
+    base::PendingTask task = delayed_task_queue_.top();
+    delayed_task_queue_.pop();
+    task.delayed_run_time = base::TimeTicks();

[Sami, 2015/03/17 15:54:45]

Could you add a TODO here about not clearing this so that the selector can
estimate delayed task starvation?

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

+    work_queue_.push(task);
+    task_enqueued = true;
+  }
+  return task_enqueued;
+}
+
+bool TaskQueue::ScheduleDelayedWorkLocked(LazyNow* lazy_now) {
+  lock_.AssertAcquired();
+  bool task_enqueued = false;
+  if (pump_policy_ == TaskQueueManager::PumpPolicy::AUTO)
+    task_enqueued = EnqueueReadyDelayedTasksLocked(lazy_now);

[Sami, 2015/03/17 15:54:45]

We can't call this here because the work queue is owned by the main thread.

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

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;
+    // 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);
+    }
+  }
+  return task_enqueued;
+}
+
+void TaskQueue::KickDelayedTasks() {
+  bool task_enqueued;

[Sami, 2015/03/17 15:54:45]

DCHECK main thread.

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

+  {
+    base::AutoLock lock(lock_);
+    LazyNow lazy_now(task_queue_manager_);

[Sami, 2015/03/17 15:54:45]

task_queue_manager_ might be deleted by now.

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

+    task_enqueued = ScheduleDelayedWorkLocked(&lazy_now);
+  }
+  if (task_enqueued)
+    task_queue_manager_->DoWork(false);
+}
+
 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());
-    }
+    EnqueueReadyDelayedTasksLocked(lazy_now);
     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_,

[Sami, 2015/03/17 15:54:45]

Could you add the delayed task queue here too?

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

                  incoming_queue_.size() + work_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_);
+    EnqueueReadyDelayedTasksLocked(&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");

[Sami, 2015/03/17 15:54:45]

Could you dump the delayed task queue here too? Also might be good to record how
many delayed task runners we have posted.

[alex clarke (OOO till 29th), 2015/03/17 16:53:06]

Done.

   QueueAsValueInto(work_queue_, state);
   state->EndArray();
   state->EndDictionary();
 }
 
 // static
 const char* TaskQueue::PumpPolicyToString(
     TaskQueueManager::PumpPolicy pump_policy) {
   switch (pump_policy) {
     case TaskQueueManager::PumpPolicy::AUTO:
@@ skipping 92 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