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); +}