From 25b101705d04bb1a6a867123dc7fc687adcd3f11 Mon Sep 17 00:00:00 2001 From: Shelley Vohr Date: Mon, 17 Aug 2026 14:46:21 +0000 Subject: [PATCH] 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 --- src/node_platform.cc | 72 +++++++++++++++--------------------- src/node_platform.h | 41 +++++++++++--------- test/cctest/test_platform.cc | 50 +++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 60 deletions(-) diff --git a/src/node_platform.cc b/src/node_platform.cc index 57a43eeb6459..71a4e3d5ee83 100644 --- a/src/node_platform.cc +++ b/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 = - std::move(const_cast&>(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 : 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 delayed = - std::move(const_cast&>( - delayed_tasks_to_schedule.top())); - delayed_tasks_to_schedule.pop(); - + for (std::unique_ptr& 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::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 entry = - std::move(const_cast&>(tasks.top())); - tasks.pop(); + for (std::unique_ptr& entry : + foreground_tasks_.Lock().PopAll()) { did_work = true; RunForegroundTask(std::move(entry->task)); } @@ -788,12 +764,21 @@ template TaskQueue::Locked::Locked(TaskQueue* queue) : queue_(queue), lock_(queue->lock_) {} +template +std::unique_ptr TaskQueue::PopTask() { + // std::priority_queue::top() only hands out a const reference. + Item& top = const_cast(task_queue_.top()); + std::unique_ptr task = std::move(top.task); + task_queue_.pop(); + return task; +} + template void TaskQueue::Locked::Push(std::unique_ptr 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 TaskQueue::Locked::Pop() { if (queue_->task_queue_.empty()) { return std::unique_ptr(nullptr); } - std::unique_ptr result = std::move( - std::move(const_cast&>(queue_->task_queue_.top()))); - queue_->task_queue_.pop(); - return result; + return queue_->PopTask(); } template @@ -816,10 +798,7 @@ std::unique_ptr TaskQueue::Locked::BlockingPop() { if (queue_->stopped_) { return std::unique_ptr(nullptr); } - std::unique_ptr result = std::move( - std::move(const_cast&>(queue_->task_queue_.top()))); - queue_->task_queue_.pop(); - return result; + return queue_->PopTask(); } template @@ -843,12 +822,19 @@ void TaskQueue::Locked::Stop() { } template -TaskQueue::PriorityQueue TaskQueue::Locked::PopAll() { - TaskQueue::PriorityQueue result; - result.swap(queue_->task_queue_); +std::vector> TaskQueue::Locked::PopAll() { + std::vector> result; + result.reserve(queue_->task_queue_.size()); + while (!queue_->task_queue_.empty()) { + result.push_back(queue_->PopTask()); + } return result; } +template class TaskQueue; +template class TaskQueue; +template class TaskQueue; + 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 diff --git a/src/node_platform.h b/src/node_platform.h index e98ecf322802..bd6fe024d5e8 100644 --- a/src/node_platform.h +++ b/src/node_platform.h @@ -27,22 +27,6 @@ concept has_priority = requires(T t) { t.priority; }; template 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& a, - const std::unique_ptr& b) const { - if constexpr (has_priority) { - return a->priority < b->priority; - } else { - return false; - } - } - }; - - using PriorityQueue = std::priority_queue, - std::vector>, - EntryCompare>; class Locked { public: void Push(std::unique_ptr 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> PopAll(); private: friend class TaskQueue; @@ -67,11 +52,33 @@ class TaskQueue { Locked Lock() { return Locked(this); } private: + struct Item { + std::unique_ptr 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) { + if (a.task->priority != b.task->priority) { + return a.task->priority < b.task->priority; + } + } + return a.sequence > b.sequence; + } + }; + using PriorityQueue = + std::priority_queue, ItemCompare>; + + std::unique_ptr PopTask(); + Mutex lock_; ConditionVariable tasks_available_; ConditionVariable outstanding_tasks_drained_; int outstanding_tasks_; bool stopped_; + uint64_t next_sequence_ = 0; PriorityQueue task_queue_; }; diff --git a/test/cctest/test_platform.cc b/test/cctest/test_platform.cc index f1c1d52d92c7..170c43ba7441 100644 --- a/test/cctest/test_platform.cc +++ b/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* log, int id) : log_(log), id_(id) {} + void Run() override { log_->push_back(id_); } + + private: + std::vector* log_; + int id_; +}; + +TEST(TaskQueueTest, HigherPriorityFirstThenPostingOrder) { + std::vector log; + { + node::TaskQueue queue; + for (int i = 0; i < 64; i++) { + queue.Lock().Push(std::make_unique(&log, i)); + } + for (std::unique_ptr& task : queue.Lock().PopAll()) task->Run(); + for (int i = 64; i < 96; i++) { + queue.Lock().Push(std::make_unique(&log, i)); + } + while (std::unique_ptr 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 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( + std::make_unique(&log, id++), priority)); + } + for (std::unique_ptr& entry : queue.Lock().PopAll()) { + entry->task->Run(); + } + } + const std::vector expected = {2, 4, 0, 3, 6, 1, 5}; + EXPECT_EQ(log, expected); +}