src: run same-priority platform tasks in posting order

src/node_platform.cc

@@ -157,16 +157,9 @@ class WorkerThreadsTaskRunner::DelayedTaskScheduler {
     DelayedTaskScheduler* scheduler =
         ContainerOf(&DelayedTaskScheduler::loop_, flush_tasks->loop);
 
-    auto tasks_to_run = scheduler->tasks_.Lock().PopAll();
-    while (!tasks_to_run.empty()) {
-      // We have to use const_cast because std::priority_queue::top() does not
-      // return a movable item.
-      std::unique_ptr<Task> task =
-          std::move(const_cast<std::unique_ptr<Task>&>(tasks_to_run.top()));
-      tasks_to_run.pop();
-      // This runs either the ScheduleTasks that scheduels the timers to
-      // pop the tasks back into the worker task runner queue, or the
-      // or the StopTasks to stop the timers and drop all the pending tasks.
+    // ScheduleTasks (start a timer that pops the task into the worker queue)
+    // in posting order, then, once Stop() was called, the StopTask.
+    for (std::unique_ptr<Task>& task : scheduler->tasks_.Lock().PopAll()) {
       task->Run();
     }
   }
@@ -611,15 +604,8 @@ void NodePlatform::DrainTasks(Isolate* isolate) {
 bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
   bool did_work = false;
 
-  auto delayed_tasks_to_schedule = foreground_delayed_tasks_.Lock().PopAll();
-  while (!delayed_tasks_to_schedule.empty()) {
-    // We have to use const_cast because std::priority_queue::top() does not
-    // return a movable item.
-    std::unique_ptr<DelayedTask> delayed =
-        std::move(const_cast<std::unique_ptr<DelayedTask>&>(
-            delayed_tasks_to_schedule.top()));
-    delayed_tasks_to_schedule.pop();
-
+  for (std::unique_ptr<DelayedTask>& delayed :
+       foreground_delayed_tasks_.Lock().PopAll()) {
     did_work = true;
     uint64_t delay_millis = llround(delayed->timeout * 1000);
 
@@ -642,18 +628,8 @@ bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
         });
   }
 
-  TaskQueue<TaskQueueEntry>::PriorityQueue tasks;
-  {
-    auto locked = foreground_tasks_.Lock();
-    tasks = locked.PopAll();
-  }
-
-  while (!tasks.empty()) {
-    // We have to use const_cast because std::priority_queue::top() does not
-    // return a movable item.
-    std::unique_ptr<TaskQueueEntry> entry =
-        std::move(const_cast<std::unique_ptr<TaskQueueEntry>&>(tasks.top()));
-    tasks.pop();
+  for (std::unique_ptr<TaskQueueEntry>& entry :
+       foreground_tasks_.Lock().PopAll()) {
     did_work = true;
     RunForegroundTask(std::move(entry->task));
   }
@@ -788,12 +764,21 @@ template <class T>
 TaskQueue<T>::Locked::Locked(TaskQueue* queue)
     : queue_(queue), lock_(queue->lock_) {}
 
+template <class T>
+std::unique_ptr<T> TaskQueue<T>::PopTask() {
+  // std::priority_queue::top() only hands out a const reference.
+  Item& top = const_cast<Item&>(task_queue_.top());
+  std::unique_ptr<T> task = std::move(top.task);
+  task_queue_.pop();
+  return task;
+}
+
 template <class T>
 void TaskQueue<T>::Locked::Push(std::unique_ptr<T> task, bool outstanding) {
   if (outstanding) {
     queue_->outstanding_tasks_++;
   }
-  queue_->task_queue_.push(std::move(task));
+  queue_->task_queue_.push({std::move(task), queue_->next_sequence_++});
   queue_->tasks_available_.Signal(lock_);
 }
 
@@ -802,10 +787,7 @@ std::unique_ptr<T> TaskQueue<T>::Locked::Pop() {
   if (queue_->task_queue_.empty()) {
     return std::unique_ptr<T>(nullptr);
   }
-  std::unique_ptr<T> result = std::move(
-      std::move(const_cast<std::unique_ptr<T>&>(queue_->task_queue_.top())));
-  queue_->task_queue_.pop();
-  return result;
+  return queue_->PopTask();
 }
 
 template <class T>
@@ -816,10 +798,7 @@ std::unique_ptr<T> TaskQueue<T>::Locked::BlockingPop() {
   if (queue_->stopped_) {
     return std::unique_ptr<T>(nullptr);
   }
-  std::unique_ptr<T> result = std::move(
-      std::move(const_cast<std::unique_ptr<T>&>(queue_->task_queue_.top())));
-  queue_->task_queue_.pop();
-  return result;
+  return queue_->PopTask();
 }
 
 template <class T>
@@ -843,12 +822,19 @@ void TaskQueue<T>::Locked::Stop() {
 }
 
 template <class T>
-TaskQueue<T>::PriorityQueue TaskQueue<T>::Locked::PopAll() {
-  TaskQueue<T>::PriorityQueue result;
-  result.swap(queue_->task_queue_);
+std::vector<std::unique_ptr<T>> TaskQueue<T>::Locked::PopAll() {
+  std::vector<std::unique_ptr<T>> result;
+  result.reserve(queue_->task_queue_.size());
+  while (!queue_->task_queue_.empty()) {
+    result.push_back(queue_->PopTask());
+  }
   return result;
 }
 
+template class TaskQueue<Task>;
+template class TaskQueue<TaskQueueEntry>;
+template class TaskQueue<DelayedTask>;
+
 void MultiIsolatePlatform::DisposeIsolate(Isolate* isolate) {
   // The order of these calls is important. When the Isolate is disposed,
   // it may still post tasks to the platform, so it must still be registered

src/node_platform.h

@@ -27,22 +27,6 @@ concept has_priority = requires(T t) { t.priority; };
 template <class T>
 class TaskQueue {
  public:
-  // If the entry type has a priority member, order the priority queue by
-  // that - higher priority first. Otherwise, maintain insertion order.
-  struct EntryCompare {
-    bool operator()(const std::unique_ptr<T>& a,
-                    const std::unique_ptr<T>& b) const {
-      if constexpr (has_priority<T>) {
-        return a->priority < b->priority;
-      } else {
-        return false;
-      }
-    }
-  };
-
-  using PriorityQueue = std::priority_queue<std::unique_ptr<T>,
-                                            std::vector<std::unique_ptr<T>>,
-                                            EntryCompare>;
   class Locked {
    public:
     void Push(std::unique_ptr<T> task, bool outstanding = false);
@@ -51,7 +35,8 @@ class TaskQueue {
     void NotifyOfOutstandingCompletion();
     void BlockingDrain();
     void Stop();
-    PriorityQueue PopAll();
+    // All queued tasks, in the order Pop() would have returned them.
+    std::vector<std::unique_ptr<T>> PopAll();
 
    private:
     friend class TaskQueue;
@@ -67,11 +52,33 @@ class TaskQueue {
   Locked Lock() { return Locked(this); }
 
  private:
+  struct Item {
+    std::unique_ptr<T> task;
+    uint64_t sequence;
+  };
+  // Higher priority first if the entry type has one; posting order otherwise
+  // and among equal priorities (a sequence number breaks the tie).
+  struct ItemCompare {
+    bool operator()(const Item& a, const Item& b) const {
+      if constexpr (has_priority<T>) {
+        if (a.task->priority != b.task->priority) {
+          return a.task->priority < b.task->priority;
+        }
+      }
+      return a.sequence > b.sequence;
+    }
+  };
+  using PriorityQueue =
+      std::priority_queue<Item, std::vector<Item>, ItemCompare>;
+
+  std::unique_ptr<T> PopTask();
+
   Mutex lock_;
   ConditionVariable tasks_available_;
   ConditionVariable outstanding_tasks_drained_;
   int outstanding_tasks_;
   bool stopped_;
+  uint64_t next_sequence_ = 0;
   PriorityQueue task_queue_;
 };
 

test/cctest/test_platform.cc

@@ -128,3 +128,53 @@ TEST_F(PlatformTest, TracingControllerNullptr) {
   node::SetTracingController(orig_controller);
   EXPECT_EQ(node::GetTracingController(), orig_controller);
 }
+
+class RecordingTask : public v8::Task {
+ public:
+  RecordingTask(std::vector<int>* log, int id) : log_(log), id_(id) {}
+  void Run() override { log_->push_back(id_); }
+
+ private:
+  std::vector<int>* log_;
+  int id_;
+};
+
+TEST(TaskQueueTest, HigherPriorityFirstThenPostingOrder) {
+  std::vector<int> log;
+  {
+    node::TaskQueue<v8::Task> queue;
+    for (int i = 0; i < 64; i++) {
+      queue.Lock().Push(std::make_unique<RecordingTask>(&log, i));
+    }
+    for (std::unique_ptr<v8::Task>& task : queue.Lock().PopAll()) task->Run();
+    for (int i = 64; i < 96; i++) {
+      queue.Lock().Push(std::make_unique<RecordingTask>(&log, i));
+    }
+    while (std::unique_ptr<v8::Task> task = queue.Lock().Pop()) task->Run();
+  }
+  ASSERT_EQ(log.size(), 96u);
+  for (int i = 0; i < 96; i++) EXPECT_EQ(log[i], i);
+
+  log.clear();
+  {
+    using v8::TaskPriority;
+    node::TaskQueue<node::TaskQueueEntry> queue;
+    const TaskPriority priorities[] = {TaskPriority::kUserVisible,
+                                       TaskPriority::kBestEffort,
+                                       TaskPriority::kUserBlocking,
+                                       TaskPriority::kUserVisible,
+                                       TaskPriority::kUserBlocking,
+                                       TaskPriority::kBestEffort,
+                                       TaskPriority::kUserVisible};
+    int id = 0;
+    for (TaskPriority priority : priorities) {
+      queue.Lock().Push(std::make_unique<node::TaskQueueEntry>(
+          std::make_unique<RecordingTask>(&log, id++), priority));
+    }
+    for (std::unique_ptr<node::TaskQueueEntry>& entry : queue.Lock().PopAll()) {
+      entry->task->Run();
+    }
+  }
+  const std::vector<int> expected = {2, 4, 0, 3, 6, 1, 5};
+  EXPECT_EQ(log, expected);
+}