src: run same-priority platform tasks in posting order

TaskQueue became a std::priority_queue when worker tasks started to
honor v8::TaskPriority. Its comparator returns false for entry types
without a priority member, and for entries of equal priority, on the
assumption that the heap then keeps insertion order. It does not: three
tasks pushed A, B, C pop as A, C, B, and larger batches come out in
heap order. That affects the per-isolate foreground task queue (tasks
of one priority no longer run in the order they were posted), the
foreground delayed task queue, and the delayed task scheduler of the
worker thread task runner, whose local queue is drained in one batch:
when v8 posts a delayed worker task shortly before the platform shuts
down, the StopTask pushed by Stop() can run before a ScheduleTask that
was pushed earlier, that ScheduleTask then starts a timer on the
scheduler's loop after all timers were supposed to be stopped, and
Shutdown() blocks in uv_thread_join() until the delay (e.g. the 8 s of
the memory reducer) expires.

Give every queued item a sequence number and use it as the tie
breaker, so that tasks of equal priority, and tasks without one, come
out in FIFO order again; higher priorities still come first. PopAll()
now returns the tasks in that order instead of handing out the heap.

Signed-off-by: Shelley Vohr <shelley.vohr@gmail.com>

Author: codebytere

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.
+  std::unique_ptr<T> task =
+      std::move(const_cast<Item&>(task_queue_.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);
+}