Replace the secondary queue logic for delayed task scheduling by a boolean flag.

Simplify logic and code by getting rid of the secondary priority queue that was tracking the times at which ProcessRipeTasks() was scheduled, and replacing it by a simple boolean flag for each delayed task.

See https://chromium-review.googlesource.com/c/chromium/src/+/1196844/13/base/task/task_scheduler/delayed_task_manager.h#89

Change-Id: I1ea6312183eaf8fc4ce45c5d668e4e5a13506e11
Reviewed-on: https://chromium-review.googlesource.com/1230694
Commit-Queue: Alexandre Frechette <frechette@chromium.org>
Reviewed-by: François Doray <fdoray@chromium.org>
Cr-Commit-Position: refs/heads/master@{#592803}

base/task/task_scheduler/delayed_task_manager.cc

@@ -15,6 +15,31 @@
 namespace base {
 namespace internal {
 
+DelayedTaskManager::DelayedTask::DelayedTask(Task task,
+                                             PostTaskNowCallback callback)
+    : task(std::move(task)), callback(std::move(callback)) {}
+
+DelayedTaskManager::DelayedTask::DelayedTask(
+    DelayedTaskManager::DelayedTask&& other) = default;
+
+DelayedTaskManager::DelayedTask::~DelayedTask() = default;
+
+DelayedTaskManager::DelayedTask& DelayedTaskManager::DelayedTask::operator=(
+    DelayedTaskManager::DelayedTask&& other) = default;
+
+bool DelayedTaskManager::DelayedTask::operator>(
+    const DelayedTask& other) const {
+  return task.delayed_run_time > other.task.delayed_run_time;
+}
+
+bool DelayedTaskManager::DelayedTask::IsScheduled() const {
+  return scheduled_;
+}
+void DelayedTaskManager::DelayedTask::SetScheduled() {
+  DCHECK(!scheduled_);
+  scheduled_ = true;
+}
+
 DelayedTaskManager::DelayedTaskManager(
     std::unique_ptr<const TickClock> tick_clock)
     : process_ripe_tasks_closure_(
@@ -30,62 +55,47 @@ void DelayedTaskManager::Start(
     scoped_refptr<TaskRunner> service_thread_task_runner) {
   DCHECK(service_thread_task_runner);
 
-  TimeTicks next_delayed_task_run_time;
+  TimeTicks process_ripe_tasks_time;
   {
     AutoSchedulerLock auto_lock(queue_lock_);
     DCHECK(!service_thread_task_runner_);
     service_thread_task_runner_ = std::move(service_thread_task_runner);
-    if (delayed_task_queue_.empty()) {
-      return;
-    }
-    next_delayed_task_run_time = GetNextDelayedTaskRunTimeLockRequired();
-    process_ripe_tasks_time_queue_.push(next_delayed_task_run_time);
+    process_ripe_tasks_time = GetTimeToScheduleProcessRipeTasksLockRequired();
   }
-  ScheduleProcessRipeTasksOnServiceThread(next_delayed_task_run_time);
+  ScheduleProcessRipeTasksOnServiceThread(process_ripe_tasks_time);
 }
 
 void DelayedTaskManager::AddDelayedTask(
     Task task,
     PostTaskNowCallback post_task_now_callback) {
   DCHECK(task.task);
+  DCHECK(!task.delayed_run_time.is_null());
 
   // Use CHECK instead of DCHECK to crash earlier. See http://crbug.com/711167
   // for details.
   CHECK(task.task);
-
-  TimeTicks next_delayed_task_run_time;
+  TimeTicks process_ripe_tasks_time;
   {
     AutoSchedulerLock auto_lock(queue_lock_);
-    TimeTicks previous_next_delayed_task_run_time =
-        GetNextDelayedTaskRunTimeLockRequired();
-    delayed_task_queue_.push(
-        {std::move(task), std::move(post_task_now_callback)});
-    next_delayed_task_run_time = GetNextDelayedTaskRunTimeLockRequired();
+    delayed_task_queue_.emplace(std::move(task),
+                                std::move(post_task_now_callback));
     // Not started yet.
-    if (service_thread_task_runner_ == nullptr) {
-      return;
-    }
-    // ProcessRipeTasks() callback already scheduled before or at |task|'s
-    // scheduled run time.
-    if (GetNextProcessRipeTaskTimeLockRequired() <=
-        next_delayed_task_run_time) {
+    if (service_thread_task_runner_ == nullptr)
       return;
-    }
-    DCHECK_LT(next_delayed_task_run_time, previous_next_delayed_task_run_time);
-    process_ripe_tasks_time_queue_.push(next_delayed_task_run_time);
+    process_ripe_tasks_time = GetTimeToScheduleProcessRipeTasksLockRequired();
   }
-  ScheduleProcessRipeTasksOnServiceThread(next_delayed_task_run_time);
+  ScheduleProcessRipeTasksOnServiceThread(process_ripe_tasks_time);
 }
 
 void DelayedTaskManager::ProcessRipeTasks() {
   std::vector<DelayedTask> ripe_delayed_tasks;
-  TimeTicks next_delayed_task_run_time;
-  bool schedule_process_ripe_tasks = false;
+  TimeTicks process_ripe_tasks_time;
 
   {
     AutoSchedulerLock auto_lock(queue_lock_);
     const TimeTicks now = tick_clock_->NowTicks();
-    while (GetNextDelayedTaskRunTimeLockRequired() <= now) {
+    while (!delayed_task_queue_.empty() &&
+           delayed_task_queue_.top().task.delayed_run_time <= now) {
       // The const_cast on top is okay since the DelayedTask is
       // transactionnaly being popped from |delayed_task_queue_| right after
       // and the move doesn't alter the sort order (a requirement for the
@@ -95,47 +105,35 @@ void DelayedTaskManager::ProcessRipeTasks() {
           std::move(const_cast<DelayedTask&>(delayed_task_queue_.top())));
       delayed_task_queue_.pop();
     }
-    while (GetNextProcessRipeTaskTimeLockRequired() <= now) {
-      process_ripe_tasks_time_queue_.pop();
-    }
-    if (!delayed_task_queue_.empty()) {
-      next_delayed_task_run_time = GetNextDelayedTaskRunTimeLockRequired();
-      // ProcessRipeTasks() callback already scheduled before or at the next
-      // task's scheduled run time.
-      if (next_delayed_task_run_time <
-          GetNextProcessRipeTaskTimeLockRequired()) {
-        schedule_process_ripe_tasks = true;
-        process_ripe_tasks_time_queue_.push(next_delayed_task_run_time);
-      }
-    }
-  }
-
-  if (schedule_process_ripe_tasks) {
-    ScheduleProcessRipeTasksOnServiceThread(next_delayed_task_run_time);
+    process_ripe_tasks_time = GetTimeToScheduleProcessRipeTasksLockRequired();
   }
+  ScheduleProcessRipeTasksOnServiceThread(process_ripe_tasks_time);
 
   for (auto& delayed_task : ripe_delayed_tasks) {
-    std::move(delayed_task.second).Run(std::move(delayed_task.first));
+    std::move(delayed_task.callback).Run(std::move(delayed_task.task));
   }
 }
 
-const TimeTicks DelayedTaskManager::GetNextDelayedTaskRunTimeLockRequired() {
-  queue_lock_.AssertAcquired();
-  return delayed_task_queue_.empty()
-             ? TimeTicks::Max()
-             : delayed_task_queue_.top().first.delayed_run_time;
-}
-
-const TimeTicks DelayedTaskManager::GetNextProcessRipeTaskTimeLockRequired() {
+TimeTicks DelayedTaskManager::GetTimeToScheduleProcessRipeTasksLockRequired() {
   queue_lock_.AssertAcquired();
-  return process_ripe_tasks_time_queue_.empty()
-             ? TimeTicks::Max()
-             : process_ripe_tasks_time_queue_.top();
+  if (delayed_task_queue_.empty())
+    return TimeTicks::Max();
+  // The const_cast on top is okay since |IsScheduled()| and |SetScheduled()|
+  // don't alter the sort order (a requirement for the Windows STL's consistency
+  // debug-checks for std::priority_queue::top())
+  DelayedTask& ripest_delayed_task =
+      const_cast<DelayedTask&>(delayed_task_queue_.top());
+  if (ripest_delayed_task.IsScheduled())
+    return TimeTicks::Max();
+  ripest_delayed_task.SetScheduled();
+  return ripest_delayed_task.task.delayed_run_time;
 }
 
 void DelayedTaskManager::ScheduleProcessRipeTasksOnServiceThread(
     TimeTicks next_delayed_task_run_time) {
   DCHECK(!next_delayed_task_run_time.is_null());
+  if (next_delayed_task_run_time.is_max())
+    return;
   const TimeTicks now = tick_clock_->NowTicks();
   TimeDelta delay = std::max(TimeDelta(), next_delayed_task_run_time - now);
   service_thread_task_runner_->PostDelayedTask(

base/task/task_scheduler/delayed_task_manager.h

@@ -27,8 +27,6 @@ class TaskRunner;
 
 namespace internal {
 
-struct Task;
-
 // The DelayedTaskManager forwards tasks to post task callbacks when they become
 // ripe for execution. Tasks are not forwarded before Start() is called. This
 // class is thread-safe.
@@ -53,22 +51,42 @@ class BASE_EXPORT DelayedTaskManager {
   void AddDelayedTask(Task task, PostTaskNowCallback post_task_now_callback);
 
  private:
-  using DelayedTask = std::pair<Task, PostTaskNowCallback>;
+  struct DelayedTask {
+    DelayedTask(Task task, PostTaskNowCallback callback);
+    DelayedTask(DelayedTask&& other);
+    ~DelayedTask();
+
+    // Required by priority_queue::pop().
+    DelayedTask& operator=(DelayedTask&& other);
+
+    // Required by priority_queue.
+    bool operator>(const DelayedTask& other) const;
+
+    Task task;
+    PostTaskNowCallback callback;
+
+    // True iff the delayed task has been marked as scheduled.
+    bool IsScheduled() const;
+
+    // Mark the delayed task as scheduled. Since the sort key is
+    // |task.delayed_run_time|, it does not alter sort order when it is called.
+    void SetScheduled();
+
+   private:
+    bool scheduled_ = false;
+  };
 
   // Pop and post all the ripe tasks in the delayed task queue.
   void ProcessRipeTasks();
 
-  // Return the run time of the delayed task that needs to be processed the
-  // soonest.
-  const TimeTicks GetNextDelayedTaskRunTimeLockRequired();
-
-  // Return the run time of the soonest scheduled `ProcessRipeTasks` call.
-  const TimeTicks GetNextProcessRipeTaskTimeLockRequired();
+  // Get the time at which to schedule the next |ProcessRipeTasks()| execution.
+  TimeTicks GetTimeToScheduleProcessRipeTasksLockRequired();
 
-  // Schedule the ProcessRipeTasks method on the service thread to be executed
-  // in the given |next_delayed_task_run_time|.
+  // Schedule |ProcessRipeTasks()| on the service thread to be executed at the
+  // given |process_ripe_tasks_time|, provided the given time is not
+  // TimeTicks::Max().
   void ScheduleProcessRipeTasksOnServiceThread(
-      TimeTicks next_delayed_task_run_time);
+      TimeTicks process_ripe_tasks_time);
 
   const RepeatingClosure process_ripe_tasks_closure_;
 
@@ -76,24 +94,12 @@ class BASE_EXPORT DelayedTaskManager {
 
   scoped_refptr<TaskRunner> service_thread_task_runner_;
 
-  struct TaskDelayedRuntimeComparator {
-    inline bool operator()(const DelayedTask& lhs,
-                           const DelayedTask& rhs) const {
-      return lhs.first.delayed_run_time > rhs.first.delayed_run_time;
-    }
-  };
-
   std::priority_queue<DelayedTask,
                       std::vector<DelayedTask>,
-                      TaskDelayedRuntimeComparator>
+                      std::greater<DelayedTask>>
       delayed_task_queue_;
 
-  std::
-      priority_queue<TimeTicks, std::vector<TimeTicks>, std::greater<TimeTicks>>
-          process_ripe_tasks_time_queue_;
-
-  // Synchronizes access to |delayed_task_queue_|,
-  // |process_ripe_task_time_queue_| and the setting of
+  // Synchronizes access to |delayed_task_queue_| and the setting of
   // |service_thread_task_runner|. Once |service_thread_task_runner_| is set,
   // it is never modified. It is therefore safe to access
   // |service_thread_task_runner_| without synchronization once it is observed