scheduler: Maintain a constant enqueue order for every task

third_party/WebKit/Source/platform/scheduler/base/task_queue_impl.cc

 // Copyright 2015 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 "platform/scheduler/base/task_queue_impl.h"
 
 #include <utility>
 
 #include "base/format_macros.h"
 #include "base/memory/ptr_util.h"
@@ skipping 92 matching lines @@
   DCHECK(any_thread().task_queue_manager == nullptr)
       << "UnregisterTaskQueue must be called first!";
 #endif
 }
 
 TaskQueueImpl::Task::Task()
     : PendingTask(tracked_objects::Location(),
                   base::Closure(),
                   base::TimeTicks(),
                   true),
-#ifndef NDEBUG
-      enqueue_order_set_(false),
-#endif
-      enqueue_order_(0) {
+      enqueue_order_({base::TimeTicks(), 0}) {
   sequence_num = 0;
 }
 
 TaskQueueImpl::Task::Task(const tracked_objects::Location& posted_from,
                           base::OnceClosure task,
-                          base::TimeTicks desired_run_time,
-                          EnqueueOrder sequence_number,
+                          EnqueueOrder enqueue_order,
                           bool nestable)
-    : PendingTask(posted_from, std::move(task), desired_run_time, nestable),
-#ifndef NDEBUG
-      enqueue_order_set_(false),
-#endif
-      enqueue_order_(0) {
-  sequence_num = sequence_number;
-}
-
-TaskQueueImpl::Task::Task(const tracked_objects::Location& posted_from,
-                          base::OnceClosure task,
-                          base::TimeTicks desired_run_time,
-                          EnqueueOrder sequence_number,
-                          bool nestable,
-                          EnqueueOrder enqueue_order)
-    : PendingTask(posted_from, std::move(task), desired_run_time, nestable),
-#ifndef NDEBUG
-      enqueue_order_set_(true),
-#endif
+    : PendingTask(posted_from,
+                  std::move(task),
+                  enqueue_order.delayed_run_time,
+                  nestable),
       enqueue_order_(enqueue_order) {
-  sequence_num = sequence_number;
+  // Note: these properties are only used for tracing.
+  sequence_num = static_cast<int>(enqueue_order.sequence_num);
+  delayed_run_time = enqueue_order.delayed_run_time;
 }
 
 TaskQueueImpl::AnyThread::AnyThread(TaskQueueManager* task_queue_manager,
                                     TimeDomain* time_domain)
     : task_queue_manager(task_queue_manager),
       time_domain(time_domain),
       observer(nullptr) {}
 
 TaskQueueImpl::AnyThread::~AnyThread() {}
 
@@ skipping 66 matching lines @@
 
 bool TaskQueueImpl::PostImmediateTaskImpl(
     const tracked_objects::Location& from_here,
     base::OnceClosure task,
     TaskType task_type) {
   DCHECK(task);
   base::AutoLock lock(any_thread_lock_);
   if (!any_thread().task_queue_manager)
     return false;
 
-  EnqueueOrder sequence_number =
-      any_thread().task_queue_manager->GetNextSequenceNumber();
+  EnqueueOrder enqueue_order =
+      any_thread().task_queue_manager->GetNextEnqueueOrder(base::TimeTicks());
 
   PushOntoImmediateIncomingQueueLocked(from_here, std::move(task),
-                                       base::TimeTicks(), sequence_number,
+                                       enqueue_order,
                                        task_type != TaskType::NON_NESTABLE);
   return true;
 }
 
 bool TaskQueueImpl::PostDelayedTaskImpl(
     const tracked_objects::Location& from_here,
     base::OnceClosure task,
     base::TimeDelta delay,
     TaskType task_type) {
   DCHECK(task);
   DCHECK_GT(delay, base::TimeDelta());
   if (base::PlatformThread::CurrentId() == thread_id_) {
     // Lock-free fast path for delayed tasks posted from the main thread.
     if (!main_thread_only().task_queue_manager)
       return false;
 
-    EnqueueOrder sequence_number =
-        main_thread_only().task_queue_manager->GetNextSequenceNumber();
-
     base::TimeTicks time_domain_now = main_thread_only().time_domain->Now();
     base::TimeTicks time_domain_delayed_run_time = time_domain_now + delay;
+    EnqueueOrder enqueue_order =
+        main_thread_only().task_queue_manager->GetNextEnqueueOrder(
+            time_domain_delayed_run_time);
     PushOntoDelayedIncomingQueueFromMainThread(
-        Task(from_here, std::move(task), time_domain_delayed_run_time,
-             sequence_number, task_type != TaskType::NON_NESTABLE),
+        Task(from_here, std::move(task), enqueue_order,
+             task_type != TaskType::NON_NESTABLE),
         time_domain_now);
   } else {
     // NOTE posting a delayed task from a different thread is not expected to
     // be common. This pathway is less optimal than perhaps it could be
     // because it causes two main thread tasks to be run.  Should this
     // assumption prove to be false in future, we may need to revisit this.
     base::AutoLock lock(any_thread_lock_);
     if (!any_thread().task_queue_manager)
       return false;
 
-    EnqueueOrder sequence_number =
-        any_thread().task_queue_manager->GetNextSequenceNumber();
-
     base::TimeTicks time_domain_now = any_thread().time_domain->Now();
     base::TimeTicks time_domain_delayed_run_time = time_domain_now + delay;
+    EnqueueOrder enqueue_order =
+        any_thread().task_queue_manager->GetNextEnqueueOrder(
+            time_domain_delayed_run_time);
     PushOntoDelayedIncomingQueueLocked(
-        Task(from_here, std::move(task), time_domain_delayed_run_time,
-             sequence_number, task_type != TaskType::NON_NESTABLE));
+        Task(from_here, std::move(task), enqueue_order,
+             task_type != TaskType::NON_NESTABLE));
   }
   return true;
 }
 
 void TaskQueueImpl::PushOntoDelayedIncomingQueueFromMainThread(
     Task pending_task, base::TimeTicks now) {
   DelayedWakeUp wake_up = pending_task.delayed_wake_up();
   main_thread_only().task_queue_manager->DidQueueTask(pending_task);
   main_thread_only().delayed_incoming_queue.push(std::move(pending_task));
 
   // If |pending_task| is at the head of the queue, then make sure a wake-up
   // is requested if the queue is enabled.  Note we still want to schedule a
   // wake-up even if blocked by a fence, because we'd break throttling logic
   // otherwise.
   DelayedWakeUp new_wake_up =
       main_thread_only().delayed_incoming_queue.top().delayed_wake_up();
   if (wake_up.time == new_wake_up.time &&
       wake_up.sequence_num == new_wake_up.sequence_num) {
     ScheduleDelayedWorkInTimeDomain(now);
   }
 
   TraceQueueSize();
 }
 
 void TaskQueueImpl::PushOntoDelayedIncomingQueueLocked(Task pending_task) {
   any_thread().task_queue_manager->DidQueueTask(pending_task);
 
-  int thread_hop_task_sequence_number =

[alex clarke (OOO till 29th), 2017/04/13 07:46:00]

Oh wow that's a nasty bug fixed here.

[Sami, 2017/04/18 10:47:49]

You mean the overflow? Yeah :)

-      any_thread().task_queue_manager->GetNextSequenceNumber();
+  EnqueueOrder thread_hop_task_enqueue_order =
+      any_thread().task_queue_manager->GetNextEnqueueOrder(base::TimeTicks());
   PushOntoImmediateIncomingQueueLocked(
-     FROM_HERE,
-     base::Bind(&TaskQueueImpl::ScheduleDelayedWorkTask, this,
-                base::Passed(&pending_task)),
-     base::TimeTicks(),
-     thread_hop_task_sequence_number,
-     false);
+      FROM_HERE,
+      base::Bind(&TaskQueueImpl::ScheduleDelayedWorkTask, this,
+                 base::Passed(&pending_task)),
+      thread_hop_task_enqueue_order, false);
 }
 
 void TaskQueueImpl::ScheduleDelayedWorkTask(Task pending_task) {
   DCHECK(main_thread_checker_.CalledOnValidThread());
   base::TimeTicks delayed_run_time = pending_task.delayed_run_time;
   base::TimeTicks time_domain_now = main_thread_only().time_domain->Now();
   if (delayed_run_time <= time_domain_now) {
     // If |delayed_run_time| is in the past then push it onto the work queue
     // immediately. To ensure the right task ordering we need to temporarily
     // push it onto the |delayed_incoming_queue|.
     delayed_run_time = time_domain_now;
     pending_task.delayed_run_time = time_domain_now;
     main_thread_only().delayed_incoming_queue.push(std::move(pending_task));
     LazyNow lazy_now(time_domain_now);
     WakeUpForDelayedWork(&lazy_now);
   } else {
     // If |delayed_run_time| is in the future we can queue it as normal.
     PushOntoDelayedIncomingQueueFromMainThread(std::move(pending_task),
                                                time_domain_now);
   }
   TraceQueueSize();
 }
 
 void TaskQueueImpl::PushOntoImmediateIncomingQueueLocked(
     const tracked_objects::Location& posted_from,
     base::OnceClosure task,
-    base::TimeTicks desired_run_time,
-    EnqueueOrder sequence_number,
+    EnqueueOrder enqueue_order,
     bool nestable) {
   // If the |immediate_incoming_queue| is empty we need a DoWork posted to make
   // it run.
   bool was_immediate_incoming_queue_empty;
 
   {
     base::AutoLock lock(immediate_incoming_queue_lock_);
     was_immediate_incoming_queue_empty = immediate_incoming_queue().empty();
     immediate_incoming_queue().emplace_back(posted_from, std::move(task),
-                                            desired_run_time, sequence_number,
-                                            nestable, sequence_number);
+                                            enqueue_order, nestable);
     any_thread().task_queue_manager->DidQueueTask(
         immediate_incoming_queue().back());
   }
 
   if (was_immediate_incoming_queue_empty) {
     // However there's no point posting a DoWork for a blocked queue. NB we can
     // only tell if it's disabled from the main thread.
     bool queue_is_blocked =
         RunsTasksOnCurrentThread() &&
         (!IsQueueEnabled() || main_thread_only().current_fence);
     any_thread().task_queue_manager->OnQueueHasIncomingImmediateWork(
-        this, sequence_number, queue_is_blocked);
+        this, enqueue_order, queue_is_blocked);
     if (any_thread().observer)
-      any_thread().observer->OnQueueNextWakeUpChanged(this, desired_run_time);
+      any_thread().observer->OnQueueNextWakeUpChanged(
+          this, enqueue_order.delayed_run_time);
   }
 
   TraceQueueSize();
 }
 
 void TaskQueueImpl::ReloadImmediateWorkQueueIfEmpty() {
   if (!main_thread_only().immediate_work_queue->Empty())
     return;
 
   main_thread_only().immediate_work_queue->ReloadEmptyImmediateQueue();
@@ skipping 62 matching lines @@
   // have been canceled.
   while (!main_thread_only().delayed_incoming_queue.empty()) {
     Task& task =
         const_cast<Task&>(main_thread_only().delayed_incoming_queue.top());
     if (task.task.IsCancelled()) {
       main_thread_only().delayed_incoming_queue.pop();
       continue;
     }
     if (task.delayed_run_time > lazy_now->Now())
       break;
-    task.set_enqueue_order(
-        main_thread_only().task_queue_manager->GetNextSequenceNumber());
     main_thread_only().delayed_work_queue->Push(std::move(task));

[alex clarke (OOO till 29th), 2017/04/13 07:46:00]

Something for a future patch, but I wonder if we actually need the
delayed_work_queue anymore.

[Sami, 2017/04/18 10:47:49]

Interesting point. We'd still need some way to round-robin immediate and delayed
work. Let's revisit this.

     main_thread_only().delayed_incoming_queue.pop();
   }
 
   // Make sure the next wake up is scheduled.
   if (!main_thread_only().delayed_incoming_queue.empty()) {
     return main_thread_only().delayed_incoming_queue.top().delayed_wake_up();
   }
 
   return base::nullopt;
 }
@@ skipping 152 matching lines @@
 
 void TaskQueueImpl::SetBlameContext(
     base::trace_event::BlameContext* blame_context) {
   main_thread_only().blame_context = blame_context;
 }
 
 void TaskQueueImpl::InsertFence(TaskQueue::InsertFencePosition position) {
   if (!main_thread_only().task_queue_manager)
     return;
 
-  EnqueueOrder previous_fence = main_thread_only().current_fence;
+  uint64_t previous_fence = main_thread_only().current_fence;
   main_thread_only().current_fence =
       position == TaskQueue::InsertFencePosition::NOW
-          ? main_thread_only().task_queue_manager->GetNextSequenceNumber()
-          : static_cast<EnqueueOrder>(EnqueueOrderValues::BLOCKING_FENCE);
+          ? main_thread_only()
+                .task_queue_manager->GetNextEnqueueOrder(base::TimeTicks())
+                .sequence_num
+          : static_cast<uint64_t>(
+                EnqueueOrderSequenceNumberValues::BLOCKING_FENCE);
 
   // Tasks posted after this point will have a strictly higher enqueue order
   // and will be blocked from running.
   bool task_unblocked = main_thread_only().immediate_work_queue->InsertFence(
       main_thread_only().current_fence);
   task_unblocked |= main_thread_only().delayed_work_queue->InsertFence(
       main_thread_only().current_fence);
 
   if (!task_unblocked && previous_fence &&
       previous_fence < main_thread_only().current_fence) {
     base::AutoLock lock(immediate_incoming_queue_lock_);
     if (!immediate_incoming_queue().empty() &&
-        immediate_incoming_queue().front().enqueue_order() > previous_fence &&
-        immediate_incoming_queue().front().enqueue_order() <
+        immediate_incoming_queue().front().enqueue_order().sequence_num >
+            previous_fence &&
+        immediate_incoming_queue().front().enqueue_order().sequence_num <
             main_thread_only().current_fence) {
       task_unblocked = true;
     }
   }
 
   if (IsQueueEnabled() && task_unblocked) {
     main_thread_only().task_queue_manager->MaybeScheduleImmediateWork(
         FROM_HERE);
   }
 }
 
 void TaskQueueImpl::RemoveFence() {
   if (!main_thread_only().task_queue_manager)
     return;
 
-  EnqueueOrder previous_fence = main_thread_only().current_fence;
+  uint64_t previous_fence = main_thread_only().current_fence;
   main_thread_only().current_fence = 0;
 
   bool task_unblocked = main_thread_only().immediate_work_queue->RemoveFence();
   task_unblocked |= main_thread_only().delayed_work_queue->RemoveFence();
 
   if (!task_unblocked && previous_fence) {
     base::AutoLock lock(immediate_incoming_queue_lock_);
     if (!immediate_incoming_queue().empty() &&
-        immediate_incoming_queue().front().enqueue_order() > previous_fence) {
+        immediate_incoming_queue().front().enqueue_order().sequence_num >
+            previous_fence) {
       task_unblocked = true;
     }
   }
 
   if (IsQueueEnabled() && task_unblocked) {
     main_thread_only().task_queue_manager->MaybeScheduleImmediateWork(
         FROM_HERE);
   }
 }
 
 bool TaskQueueImpl::BlockedByFence() const {
   if (!main_thread_only().current_fence)
     return false;
 
   if (!main_thread_only().immediate_work_queue->BlockedByFence() ||
       !main_thread_only().delayed_work_queue->BlockedByFence()) {
     return false;
   }
 
   base::AutoLock lock(immediate_incoming_queue_lock_);
   if (immediate_incoming_queue().empty())
     return true;
 
-  return immediate_incoming_queue().front().enqueue_order() >
+  return immediate_incoming_queue().front().enqueue_order().sequence_num >
          main_thread_only().current_fence;
 }
 
 bool TaskQueueImpl::CouldTaskRun(EnqueueOrder enqueue_order) const {
   if (!IsQueueEnabled())
     return false;
 
   if (!main_thread_only().current_fence)
     return true;
 
-  return enqueue_order < main_thread_only().current_fence;
+  return enqueue_order.sequence_num < main_thread_only().current_fence;
 }
 
-EnqueueOrder TaskQueueImpl::GetFenceForTest() const {
+uint64_t TaskQueueImpl::GetFenceForTest() const {
   return main_thread_only().current_fence;
 }
 
 // static
 void TaskQueueImpl::QueueAsValueInto(const WTF::Deque<Task>& queue,
                                      base::trace_event::TracedValue* state) {
   for (const Task& task : queue) {
     TaskAsValueInto(task, state);
   }
 }
@@ skipping 12 matching lines @@
     mutable_queue->pop();
   }
   *mutable_queue = std::move(visited);
 }
 
 // static
 void TaskQueueImpl::TaskAsValueInto(const Task& task,
                                     base::trace_event::TracedValue* state) {
   state->BeginDictionary();
   state->SetString("posted_from", task.posted_from.ToString());
-#ifndef NDEBUG
-  if (task.enqueue_order_set())
-    state->SetInteger("enqueue_order", task.enqueue_order());
-#else
-  state->SetInteger("enqueue_order", task.enqueue_order());
-#endif
+  state->SetInteger("enqueue_order.delayed_run_time",
+                    task.enqueue_order().delayed_run_time.ToInternalValue());
+  state->SetInteger("enqueue_order.sequence_num",
+                    task.enqueue_order().sequence_num);
   state->SetInteger("sequence_num", task.sequence_num);
   state->SetBoolean("nestable", task.nestable);
   state->SetBoolean("is_high_res", task.is_high_res);
   state->SetBoolean("is_cancelled", task.task.IsCancelled());
   state->SetDouble(
       "delayed_run_time",
       (task.delayed_run_time - base::TimeTicks()).InMicroseconds() / 1000.0L);
   state->EndDictionary();
 }
 
@@ skipping 145 matching lines @@
 }
 
 void TaskQueueImpl::NotifyWakeUpChangedOnMainThread(base::TimeTicks wake_up) {
   if (main_thread_only().observer)
     main_thread_only().observer->OnQueueNextWakeUpChanged(this, wake_up);
 }
 
 }  // namespace internal
 }  // namespace scheduler
 }  // namespace blink