mirror of
https://github.com/nodejs/node.git
synced 2026-11-09 15:11:11 -05:00
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> PR-URL: https://github.com/nodejs/node/pull/65353 Refs: https://github.com/nodejs/node/pull/58047 Refs: https://github.com/nodejs/node/pull/61999 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
853 lines
28 KiB
C++
853 lines
28 KiB
C++
#include "node_platform.h"
|
|
#include "node_internals.h"
|
|
|
|
#include "env-inl.h"
|
|
#include "debug_utils-inl.h"
|
|
#include <algorithm> // find_if(), find(), move()
|
|
#include <cmath> // llround()
|
|
#include <memory> // unique_ptr(), shared_ptr(), make_shared()
|
|
|
|
namespace node {
|
|
|
|
using v8::Isolate;
|
|
using v8::Object;
|
|
using v8::Platform;
|
|
using v8::Task;
|
|
using v8::TaskPriority;
|
|
|
|
namespace {
|
|
|
|
struct PlatformWorkerData {
|
|
TaskQueue<TaskQueueEntry>* task_queue;
|
|
Mutex* platform_workers_mutex;
|
|
ConditionVariable* platform_workers_ready;
|
|
int* pending_platform_workers;
|
|
int id;
|
|
PlatformDebugLogLevel debug_log_level;
|
|
};
|
|
|
|
const char* GetTaskPriorityName(TaskPriority priority) {
|
|
switch (priority) {
|
|
case TaskPriority::kUserBlocking:
|
|
return "UserBlocking";
|
|
case TaskPriority::kUserVisible:
|
|
return "UserVisible";
|
|
case TaskPriority::kBestEffort:
|
|
return "BestEffort";
|
|
default:
|
|
return "Unknown";
|
|
}
|
|
}
|
|
|
|
static void PrintSourceLocation(const v8::SourceLocation& location) {
|
|
auto loc = location.ToString();
|
|
if (!loc.empty()) {
|
|
fprintf(stderr, " %s\n", loc.c_str());
|
|
} else {
|
|
fprintf(stderr, " <no location>\n");
|
|
}
|
|
}
|
|
|
|
static void PlatformWorkerThread(void* data) {
|
|
uv_thread_setname("node-V8Worker");
|
|
std::unique_ptr<PlatformWorkerData>
|
|
worker_data(static_cast<PlatformWorkerData*>(data));
|
|
|
|
TaskQueue<TaskQueueEntry>* pending_worker_tasks = worker_data->task_queue;
|
|
TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
|
|
"PlatformWorkerThread");
|
|
|
|
// Notify the main thread that the platform worker is ready.
|
|
{
|
|
Mutex::ScopedLock lock(*worker_data->platform_workers_mutex);
|
|
(*worker_data->pending_platform_workers)--;
|
|
worker_data->platform_workers_ready->Signal(lock);
|
|
}
|
|
|
|
bool debug_log_enabled =
|
|
worker_data->debug_log_level != PlatformDebugLogLevel::kNone;
|
|
int id = worker_data->id;
|
|
while (std::unique_ptr<TaskQueueEntry> entry =
|
|
pending_worker_tasks->Lock().BlockingPop()) {
|
|
if (debug_log_enabled) {
|
|
fprintf(stderr,
|
|
"\nPlatformWorkerThread %d running task %p %s\n",
|
|
id,
|
|
entry->task.get(),
|
|
GetTaskPriorityName(entry->priority));
|
|
fflush(stderr);
|
|
}
|
|
entry->task->Run();
|
|
// See NodePlatform::DrainTasks().
|
|
if (entry->is_outstanding()) {
|
|
pending_worker_tasks->Lock().NotifyOfOutstandingCompletion();
|
|
}
|
|
}
|
|
}
|
|
|
|
static int GetActualThreadPoolSize(int thread_pool_size) {
|
|
if (thread_pool_size < 1) {
|
|
thread_pool_size = uv_available_parallelism() - 1;
|
|
}
|
|
return std::max(thread_pool_size, 1);
|
|
}
|
|
|
|
} // namespace
|
|
|
|
class WorkerThreadsTaskRunner::DelayedTaskScheduler {
|
|
public:
|
|
explicit DelayedTaskScheduler(TaskQueue<TaskQueueEntry>* tasks)
|
|
: pending_worker_tasks_(tasks) {}
|
|
|
|
std::unique_ptr<uv_thread_t> Start() {
|
|
auto start_thread = [](void* data) {
|
|
uv_thread_setname("DelayedTaskSchedulerWorker");
|
|
static_cast<DelayedTaskScheduler*>(data)->Run();
|
|
};
|
|
std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
|
|
uv_sem_init(&ready_, 0);
|
|
CHECK_EQ(0, uv_thread_create(t.get(), start_thread, this));
|
|
uv_sem_wait(&ready_);
|
|
uv_sem_destroy(&ready_);
|
|
return t;
|
|
}
|
|
|
|
void PostDelayedTask(v8::TaskPriority priority,
|
|
std::unique_ptr<Task> task,
|
|
double delay_in_seconds) {
|
|
auto locked = tasks_.Lock();
|
|
|
|
if (has_shut_down_) return;
|
|
|
|
auto entry = std::make_unique<TaskQueueEntry>(std::move(task), priority);
|
|
auto delayed = std::make_unique<ScheduleTask>(
|
|
this, std::move(entry), delay_in_seconds);
|
|
|
|
// The delayed task scheuler is on is own thread with its own loop that
|
|
// runs the timers for the scheduled tasks to pop the original task back
|
|
// into the the worker task queue. This first pushes the tasks that
|
|
// schedules the timers into the local task queue that will be flushed
|
|
// by the local event loop.
|
|
locked.Push(std::move(delayed));
|
|
uv_async_send(&flush_tasks_);
|
|
}
|
|
|
|
void Stop() {
|
|
auto locked = tasks_.Lock();
|
|
has_shut_down_ = true;
|
|
locked.Push(std::make_unique<StopTask>(this));
|
|
uv_async_send(&flush_tasks_);
|
|
}
|
|
|
|
private:
|
|
void Run() {
|
|
TRACE_EVENT_METADATA1("__metadata", "thread_name", "name",
|
|
"WorkerThreadsTaskRunner::DelayedTaskScheduler");
|
|
loop_.data = this;
|
|
CHECK_EQ(0, uv_loop_init(&loop_));
|
|
flush_tasks_.data = this;
|
|
CHECK_EQ(0, uv_async_init(&loop_, &flush_tasks_, FlushTasks));
|
|
uv_sem_post(&ready_);
|
|
|
|
uv_run(&loop_, UV_RUN_DEFAULT);
|
|
CheckedUvLoopClose(&loop_);
|
|
}
|
|
|
|
static void FlushTasks(uv_async_t* flush_tasks) {
|
|
DelayedTaskScheduler* scheduler =
|
|
ContainerOf(&DelayedTaskScheduler::loop_, flush_tasks->loop);
|
|
|
|
// 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();
|
|
}
|
|
}
|
|
|
|
class StopTask : public Task {
|
|
public:
|
|
explicit StopTask(DelayedTaskScheduler* scheduler): scheduler_(scheduler) {}
|
|
|
|
void Run() override {
|
|
std::vector<uv_timer_t*> timers;
|
|
for (uv_timer_t* timer : scheduler_->timers_)
|
|
timers.push_back(timer);
|
|
for (uv_timer_t* timer : timers)
|
|
scheduler_->TakeTimerTask(timer);
|
|
uv_close(reinterpret_cast<uv_handle_t*>(&scheduler_->flush_tasks_),
|
|
[](uv_handle_t* handle) {});
|
|
}
|
|
|
|
private:
|
|
DelayedTaskScheduler* scheduler_;
|
|
};
|
|
|
|
class ScheduleTask : public Task {
|
|
public:
|
|
ScheduleTask(DelayedTaskScheduler* scheduler,
|
|
std::unique_ptr<TaskQueueEntry> task,
|
|
double delay_in_seconds)
|
|
: scheduler_(scheduler),
|
|
task_(std::move(task)),
|
|
delay_in_seconds_(delay_in_seconds) {}
|
|
|
|
void Run() override {
|
|
uint64_t delay_millis = llround(delay_in_seconds_ * 1000);
|
|
std::unique_ptr<uv_timer_t> timer(new uv_timer_t());
|
|
CHECK_EQ(0, uv_timer_init(&scheduler_->loop_, timer.get()));
|
|
timer->data = task_.release();
|
|
CHECK_EQ(0, uv_timer_start(timer.get(), RunTask, delay_millis, 0));
|
|
scheduler_->timers_.insert(timer.release());
|
|
}
|
|
|
|
private:
|
|
DelayedTaskScheduler* scheduler_;
|
|
std::unique_ptr<TaskQueueEntry> task_;
|
|
double delay_in_seconds_;
|
|
};
|
|
|
|
static void RunTask(uv_timer_t* timer) {
|
|
DelayedTaskScheduler* scheduler =
|
|
ContainerOf(&DelayedTaskScheduler::loop_, timer->loop);
|
|
auto entry = scheduler->TakeTimerTask(timer);
|
|
bool is_outstanding = entry->is_outstanding();
|
|
scheduler->pending_worker_tasks_->Lock().Push(std::move(entry),
|
|
is_outstanding);
|
|
}
|
|
|
|
std::unique_ptr<TaskQueueEntry> TakeTimerTask(uv_timer_t* timer) {
|
|
std::unique_ptr<TaskQueueEntry> task_entry(
|
|
static_cast<TaskQueueEntry*>(timer->data));
|
|
uv_timer_stop(timer);
|
|
uv_close(reinterpret_cast<uv_handle_t*>(timer), [](uv_handle_t* handle) {
|
|
delete reinterpret_cast<uv_timer_t*>(handle);
|
|
});
|
|
timers_.erase(timer);
|
|
return task_entry;
|
|
}
|
|
|
|
uv_sem_t ready_;
|
|
// Task queue in the worker thread task runner, we push the delayed task back
|
|
// to it when the timer expires.
|
|
TaskQueue<TaskQueueEntry>* pending_worker_tasks_;
|
|
|
|
// Locally scheduled tasks to be poped into the worker task runner queue.
|
|
// It is flushed whenever the next closest timer expires.
|
|
TaskQueue<Task> tasks_;
|
|
uv_loop_t loop_;
|
|
uv_async_t flush_tasks_;
|
|
std::unordered_set<uv_timer_t*> timers_;
|
|
bool has_shut_down_ = false;
|
|
};
|
|
|
|
WorkerThreadsTaskRunner::WorkerThreadsTaskRunner(
|
|
int thread_pool_size, PlatformDebugLogLevel debug_log_level)
|
|
: debug_log_level_(debug_log_level) {
|
|
Mutex platform_workers_mutex;
|
|
ConditionVariable platform_workers_ready;
|
|
|
|
Mutex::ScopedLock lock(platform_workers_mutex);
|
|
int pending_platform_workers = thread_pool_size;
|
|
|
|
delayed_task_scheduler_ = std::make_unique<DelayedTaskScheduler>(
|
|
&pending_worker_tasks_);
|
|
threads_.push_back(delayed_task_scheduler_->Start());
|
|
|
|
for (int i = 0; i < thread_pool_size; i++) {
|
|
PlatformWorkerData* worker_data =
|
|
new PlatformWorkerData{&pending_worker_tasks_,
|
|
&platform_workers_mutex,
|
|
&platform_workers_ready,
|
|
&pending_platform_workers,
|
|
i,
|
|
debug_log_level_};
|
|
std::unique_ptr<uv_thread_t> t { new uv_thread_t() };
|
|
if (uv_thread_create(t.get(), PlatformWorkerThread,
|
|
worker_data) != 0) {
|
|
break;
|
|
}
|
|
threads_.push_back(std::move(t));
|
|
}
|
|
|
|
// Wait for platform workers to initialize before continuing with the
|
|
// bootstrap.
|
|
while (pending_platform_workers > 0) {
|
|
platform_workers_ready.Wait(lock);
|
|
}
|
|
}
|
|
|
|
void WorkerThreadsTaskRunner::PostTask(v8::TaskPriority priority,
|
|
std::unique_ptr<v8::Task> task,
|
|
const v8::SourceLocation& location) {
|
|
auto entry = std::make_unique<TaskQueueEntry>(std::move(task), priority);
|
|
bool is_outstanding = entry->is_outstanding();
|
|
pending_worker_tasks_.Lock().Push(std::move(entry), is_outstanding);
|
|
}
|
|
|
|
void WorkerThreadsTaskRunner::PostDelayedTask(
|
|
v8::TaskPriority priority,
|
|
std::unique_ptr<v8::Task> task,
|
|
const v8::SourceLocation& location,
|
|
double delay_in_seconds) {
|
|
delayed_task_scheduler_->PostDelayedTask(
|
|
priority, std::move(task), delay_in_seconds);
|
|
}
|
|
|
|
void WorkerThreadsTaskRunner::BlockingDrain() {
|
|
pending_worker_tasks_.Lock().BlockingDrain();
|
|
}
|
|
|
|
void WorkerThreadsTaskRunner::Shutdown() {
|
|
pending_worker_tasks_.Lock().Stop();
|
|
delayed_task_scheduler_->Stop();
|
|
for (size_t i = 0; i < threads_.size(); i++) {
|
|
CHECK_EQ(0, uv_thread_join(threads_[i].get()));
|
|
}
|
|
}
|
|
|
|
int WorkerThreadsTaskRunner::NumberOfWorkerThreads() const {
|
|
return threads_.size();
|
|
}
|
|
|
|
PerIsolatePlatformData::PerIsolatePlatformData(
|
|
Isolate* isolate, uv_loop_t* loop, PlatformDebugLogLevel debug_log_level)
|
|
: isolate_(isolate), loop_(loop), debug_log_level_(debug_log_level) {
|
|
flush_tasks_ = new uv_async_t();
|
|
CHECK_EQ(0, uv_async_init(loop, flush_tasks_, FlushTasks));
|
|
flush_tasks_->data = static_cast<void*>(this);
|
|
uv_unref(reinterpret_cast<uv_handle_t*>(flush_tasks_));
|
|
}
|
|
|
|
std::shared_ptr<v8::TaskRunner>
|
|
PerIsolatePlatformData::GetForegroundTaskRunner() {
|
|
return shared_from_this();
|
|
}
|
|
|
|
void PerIsolatePlatformData::FlushTasks(uv_async_t* handle) {
|
|
auto platform_data = static_cast<PerIsolatePlatformData*>(handle->data);
|
|
platform_data->FlushForegroundTasksInternal();
|
|
}
|
|
|
|
void PerIsolatePlatformData::PostIdleTaskImpl(
|
|
std::unique_ptr<v8::IdleTask> task, const v8::SourceLocation& location) {
|
|
UNREACHABLE();
|
|
}
|
|
|
|
void PerIsolatePlatformData::PostTaskImpl(std::unique_ptr<Task> task,
|
|
const v8::SourceLocation& location) {
|
|
// The task can be posted from any V8 background worker thread, even when
|
|
// the foreground task runner is being cleaned up by Shutdown(). In that
|
|
// case, make sure we wait until the shutdown is completed (which leads
|
|
// to flush_tasks_ == nullptr, and the task will be discarded).
|
|
if (debug_log_level_ != PlatformDebugLogLevel::kNone) {
|
|
fprintf(stderr, "\nPerIsolatePlatformData::PostTaskImpl %p", task.get());
|
|
PrintSourceLocation(location);
|
|
if (debug_log_level_ == PlatformDebugLogLevel::kVerbose) {
|
|
DumpNativeBacktrace(stderr);
|
|
}
|
|
fflush(stderr);
|
|
}
|
|
|
|
auto locked = foreground_tasks_.Lock();
|
|
if (flush_tasks_ == nullptr) return;
|
|
// All foreground tasks are treated as user blocking tasks.
|
|
locked.Push(std::make_unique<TaskQueueEntry>(
|
|
std::move(task), v8::TaskPriority::kUserBlocking));
|
|
uv_async_send(flush_tasks_);
|
|
}
|
|
|
|
void PerIsolatePlatformData::PostDelayedTaskImpl(
|
|
std::unique_ptr<Task> task,
|
|
double delay_in_seconds,
|
|
const v8::SourceLocation& location) {
|
|
if (debug_log_level_ != PlatformDebugLogLevel::kNone) {
|
|
fprintf(stderr,
|
|
"\nPerIsolatePlatformData::PostDelayedTaskImpl %p %f",
|
|
task.get(),
|
|
delay_in_seconds);
|
|
PrintSourceLocation(location);
|
|
if (debug_log_level_ == PlatformDebugLogLevel::kVerbose) {
|
|
DumpNativeBacktrace(stderr);
|
|
}
|
|
fflush(stderr);
|
|
}
|
|
|
|
auto locked = foreground_delayed_tasks_.Lock();
|
|
if (flush_tasks_ == nullptr) return;
|
|
std::unique_ptr<DelayedTask> delayed(new DelayedTask());
|
|
delayed->task = std::move(task);
|
|
delayed->platform_data = shared_from_this();
|
|
delayed->timeout = delay_in_seconds;
|
|
// All foreground tasks are treated as user blocking tasks.
|
|
delayed->priority = v8::TaskPriority::kUserBlocking;
|
|
locked.Push(std::move(delayed));
|
|
uv_async_send(flush_tasks_);
|
|
}
|
|
|
|
void PerIsolatePlatformData::PostNonNestableTaskImpl(
|
|
std::unique_ptr<Task> task, const v8::SourceLocation& location) {
|
|
PostTaskImpl(std::move(task), location);
|
|
}
|
|
|
|
void PerIsolatePlatformData::PostNonNestableDelayedTaskImpl(
|
|
std::unique_ptr<Task> task,
|
|
double delay_in_seconds,
|
|
const v8::SourceLocation& location) {
|
|
PostDelayedTaskImpl(std::move(task), delay_in_seconds, location);
|
|
}
|
|
|
|
PerIsolatePlatformData::~PerIsolatePlatformData() {
|
|
CHECK(!flush_tasks_);
|
|
}
|
|
|
|
void PerIsolatePlatformData::AddShutdownCallback(void (*callback)(void*),
|
|
void* data) {
|
|
shutdown_callbacks_.emplace_back(ShutdownCallback { callback, data });
|
|
}
|
|
|
|
void PerIsolatePlatformData::Shutdown() {
|
|
auto foreground_tasks_locked = foreground_tasks_.Lock();
|
|
auto foreground_delayed_tasks_locked = foreground_delayed_tasks_.Lock();
|
|
|
|
foreground_delayed_tasks_locked.PopAll();
|
|
foreground_tasks_locked.PopAll();
|
|
scheduled_delayed_tasks_.clear();
|
|
|
|
if (flush_tasks_ != nullptr) {
|
|
// Both destroying the scheduled_delayed_tasks_ lists and closing
|
|
// flush_tasks_ handle add tasks to the event loop. We keep a count of all
|
|
// non-closed handles, and when that reaches zero, we inform any shutdown
|
|
// callbacks that the platform is done as far as this Isolate is concerned.
|
|
self_reference_ = shared_from_this();
|
|
uv_close(reinterpret_cast<uv_handle_t*>(flush_tasks_),
|
|
[](uv_handle_t* handle) {
|
|
std::unique_ptr<uv_async_t> flush_tasks{
|
|
reinterpret_cast<uv_async_t*>(handle)};
|
|
PerIsolatePlatformData* platform_data =
|
|
static_cast<PerIsolatePlatformData*>(flush_tasks->data);
|
|
platform_data->DecreaseHandleCount();
|
|
platform_data->self_reference_.reset();
|
|
});
|
|
flush_tasks_ = nullptr;
|
|
}
|
|
}
|
|
|
|
void PerIsolatePlatformData::DecreaseHandleCount() {
|
|
CHECK_GE(uv_handle_count_, 1);
|
|
if (--uv_handle_count_ == 0) {
|
|
for (const auto& callback : shutdown_callbacks_)
|
|
callback.cb(callback.data);
|
|
}
|
|
}
|
|
|
|
NodePlatform::NodePlatform(int thread_pool_size,
|
|
v8::TracingController* tracing_controller,
|
|
v8::PageAllocator* page_allocator) {
|
|
if (per_process::enabled_debug_list.enabled(
|
|
DebugCategory::PLATFORM_VERBOSE)) {
|
|
debug_log_level_ = PlatformDebugLogLevel::kVerbose;
|
|
} else if (per_process::enabled_debug_list.enabled(
|
|
DebugCategory::PLATFORM_MINIMAL)) {
|
|
debug_log_level_ = PlatformDebugLogLevel::kMinimal;
|
|
} else {
|
|
debug_log_level_ = PlatformDebugLogLevel::kNone;
|
|
}
|
|
|
|
if (tracing_controller != nullptr) {
|
|
tracing_controller_ = tracing_controller;
|
|
} else {
|
|
tracing_controller_ = new v8::TracingController();
|
|
}
|
|
|
|
// V8 will default to its built in allocator if none is provided.
|
|
page_allocator_ = page_allocator;
|
|
|
|
// TODO(addaleax): It's a bit icky that we use global state here, but we can't
|
|
// really do anything about it unless V8 starts exposing a way to access the
|
|
// current v8::Platform instance.
|
|
SetTracingController(tracing_controller_);
|
|
DCHECK_EQ(GetTracingController(), tracing_controller_);
|
|
|
|
thread_pool_size = GetActualThreadPoolSize(thread_pool_size);
|
|
worker_thread_task_runner_ = std::make_shared<WorkerThreadsTaskRunner>(
|
|
thread_pool_size, debug_log_level_);
|
|
}
|
|
|
|
NodePlatform::~NodePlatform() {
|
|
Shutdown();
|
|
}
|
|
|
|
void NodePlatform::RegisterIsolate(Isolate* isolate, uv_loop_t* loop) {
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
auto delegate =
|
|
std::make_shared<PerIsolatePlatformData>(isolate, loop, debug_log_level_);
|
|
IsolatePlatformDelegate* ptr = delegate.get();
|
|
auto insertion = per_isolate_.emplace(
|
|
isolate,
|
|
std::make_pair(ptr, std::move(delegate)));
|
|
CHECK(insertion.second);
|
|
}
|
|
|
|
void NodePlatform::RegisterIsolate(Isolate* isolate,
|
|
IsolatePlatformDelegate* delegate) {
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
auto insertion = per_isolate_.emplace(
|
|
isolate,
|
|
std::make_pair(delegate, std::shared_ptr<PerIsolatePlatformData>{}));
|
|
CHECK(insertion.second);
|
|
}
|
|
|
|
void NodePlatform::UnregisterIsolate(Isolate* isolate) {
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
auto existing_it = per_isolate_.find(isolate);
|
|
CHECK_NE(existing_it, per_isolate_.end());
|
|
auto& existing = existing_it->second;
|
|
if (existing.second) {
|
|
existing.second->Shutdown();
|
|
}
|
|
per_isolate_.erase(existing_it);
|
|
}
|
|
|
|
void NodePlatform::AddIsolateFinishedCallback(Isolate* isolate,
|
|
void (*cb)(void*), void* data) {
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
auto it = per_isolate_.find(isolate);
|
|
if (it == per_isolate_.end()) {
|
|
cb(data);
|
|
return;
|
|
}
|
|
CHECK(it->second.second);
|
|
it->second.second->AddShutdownCallback(cb, data);
|
|
}
|
|
|
|
void NodePlatform::Shutdown() {
|
|
if (has_shut_down_) return;
|
|
has_shut_down_ = true;
|
|
worker_thread_task_runner_->Shutdown();
|
|
|
|
{
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
per_isolate_.clear();
|
|
}
|
|
}
|
|
|
|
int NodePlatform::NumberOfWorkerThreads() {
|
|
return worker_thread_task_runner_->NumberOfWorkerThreads();
|
|
}
|
|
|
|
void PerIsolatePlatformData::RunForegroundTask(std::unique_ptr<Task> task) {
|
|
if (isolate_->IsExecutionTerminating()) return;
|
|
DebugSealHandleScope scope(isolate_);
|
|
Environment* env = Environment::GetCurrent(isolate_);
|
|
if (env != nullptr) {
|
|
v8::HandleScope scope(isolate_);
|
|
InternalCallbackScope cb_scope(env, Object::New(isolate_), { 0, 0 },
|
|
InternalCallbackScope::kNoFlags);
|
|
task->Run();
|
|
} else {
|
|
// When the Environment was freed, the tasks of the Isolate should also be
|
|
// canceled by `NodePlatform::UnregisterIsolate`. However, if the embedder
|
|
// request to run the foreground task after the Environment was freed, run
|
|
// the task without InternalCallbackScope.
|
|
|
|
// The task is moved out of InternalCallbackScope if env is not available.
|
|
// This is a required else block, and should not be removed.
|
|
// See comment: https://github.com/nodejs/node/pull/34688#pullrequestreview-463867489
|
|
task->Run();
|
|
}
|
|
}
|
|
|
|
void PerIsolatePlatformData::DeleteFromScheduledTasks(DelayedTask* task) {
|
|
auto it =
|
|
std::ranges::find_if(scheduled_delayed_tasks_,
|
|
[task](const DelayedTaskPointer& delayed) -> bool {
|
|
return delayed.get() == task;
|
|
});
|
|
CHECK_NE(it, scheduled_delayed_tasks_.end());
|
|
scheduled_delayed_tasks_.erase(it);
|
|
}
|
|
|
|
void PerIsolatePlatformData::RunForegroundTask(uv_timer_t* handle) {
|
|
DelayedTask* delayed = ContainerOf(&DelayedTask::timer, handle);
|
|
delayed->platform_data->RunForegroundTask(std::move(delayed->task));
|
|
delayed->platform_data->DeleteFromScheduledTasks(delayed);
|
|
}
|
|
|
|
void NodePlatform::DrainTasks(Isolate* isolate) {
|
|
std::shared_ptr<PerIsolatePlatformData> per_isolate = ForNodeIsolate(isolate);
|
|
if (!per_isolate) return;
|
|
|
|
do {
|
|
// FIXME(54918): we should not be blocking on the worker tasks on the
|
|
// main thread in one go. Doing so leads to two problems:
|
|
// 1. If any of the worker tasks post another foreground task and wait
|
|
// for it to complete, and that foreground task is posted right after
|
|
// we flush the foreground task queue and before the foreground thread
|
|
// goes into sleep, we'll never be able to wake up to execute that
|
|
// foreground task and in turn the worker task will never complete, and
|
|
// we have a deadlock.
|
|
// 2. Worker tasks can be posted from any thread, not necessarily associated
|
|
// with the current isolate, and we can be blocking on a worker task that
|
|
// is associated with a completely unrelated isolate in the event loop.
|
|
// This is suboptimal.
|
|
//
|
|
// However, not blocking on the worker tasks at all can lead to loss of some
|
|
// critical user-blocking worker tasks e.g. wasm async compilation tasks,
|
|
// which should block the main thread until they are completed, as the
|
|
// documentation suggests. As a compromise, we currently only block on
|
|
// user-blocking tasks to reduce the chance of deadlocks while making sure
|
|
// that critical user-blocking tasks are not lost.
|
|
worker_thread_task_runner_->BlockingDrain();
|
|
} while (per_isolate->FlushForegroundTasksInternal());
|
|
}
|
|
|
|
bool PerIsolatePlatformData::FlushForegroundTasksInternal() {
|
|
bool did_work = false;
|
|
|
|
for (std::unique_ptr<DelayedTask>& delayed :
|
|
foreground_delayed_tasks_.Lock().PopAll()) {
|
|
did_work = true;
|
|
uint64_t delay_millis = llround(delayed->timeout * 1000);
|
|
|
|
delayed->timer.data = static_cast<void*>(delayed.get());
|
|
uv_timer_init(loop_, &delayed->timer);
|
|
// Timers may not guarantee queue ordering of events with the same delay
|
|
// if the delay is non-zero. This should not be a problem in practice.
|
|
uv_timer_start(&delayed->timer, RunForegroundTask, delay_millis, 0);
|
|
uv_unref(reinterpret_cast<uv_handle_t*>(&delayed->timer));
|
|
uv_handle_count_++;
|
|
|
|
scheduled_delayed_tasks_.emplace_back(
|
|
delayed.release(), [](DelayedTask* delayed) {
|
|
uv_close(reinterpret_cast<uv_handle_t*>(&delayed->timer),
|
|
[](uv_handle_t* handle) {
|
|
std::unique_ptr<DelayedTask> task{
|
|
static_cast<DelayedTask*>(handle->data)};
|
|
task->platform_data->DecreaseHandleCount();
|
|
});
|
|
});
|
|
}
|
|
|
|
for (std::unique_ptr<TaskQueueEntry>& entry :
|
|
foreground_tasks_.Lock().PopAll()) {
|
|
did_work = true;
|
|
RunForegroundTask(std::move(entry->task));
|
|
}
|
|
|
|
return did_work;
|
|
}
|
|
|
|
void NodePlatform::PostTaskOnWorkerThreadImpl(
|
|
v8::TaskPriority priority,
|
|
std::unique_ptr<v8::Task> task,
|
|
const v8::SourceLocation& location) {
|
|
if (debug_log_level_ != PlatformDebugLogLevel::kNone) {
|
|
fprintf(stderr,
|
|
"\nNodePlatform::PostTaskOnWorkerThreadImpl %s %p",
|
|
GetTaskPriorityName(priority),
|
|
task.get());
|
|
PrintSourceLocation(location);
|
|
if (debug_log_level_ == PlatformDebugLogLevel::kVerbose) {
|
|
DumpNativeBacktrace(stderr);
|
|
}
|
|
fflush(stderr);
|
|
}
|
|
worker_thread_task_runner_->PostTask(priority, std::move(task), location);
|
|
}
|
|
|
|
void NodePlatform::PostDelayedTaskOnWorkerThreadImpl(
|
|
v8::TaskPriority priority,
|
|
std::unique_ptr<v8::Task> task,
|
|
double delay_in_seconds,
|
|
const v8::SourceLocation& location) {
|
|
if (debug_log_level_ != PlatformDebugLogLevel::kNone) {
|
|
fprintf(stderr,
|
|
"\nNodePlatform::PostDelayedTaskOnWorkerThreadImpl %s %p %f",
|
|
GetTaskPriorityName(priority),
|
|
task.get(),
|
|
delay_in_seconds);
|
|
PrintSourceLocation(location);
|
|
if (debug_log_level_ == PlatformDebugLogLevel::kVerbose) {
|
|
DumpNativeBacktrace(stderr);
|
|
}
|
|
fflush(stderr);
|
|
}
|
|
worker_thread_task_runner_->PostDelayedTask(
|
|
priority, std::move(task), location, delay_in_seconds);
|
|
}
|
|
|
|
IsolatePlatformDelegate* NodePlatform::ForIsolate(Isolate* isolate) {
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
auto data = per_isolate_[isolate];
|
|
CHECK_NOT_NULL(data.first);
|
|
return data.first;
|
|
}
|
|
|
|
std::shared_ptr<PerIsolatePlatformData>
|
|
NodePlatform::ForNodeIsolate(Isolate* isolate) {
|
|
Mutex::ScopedLock lock(per_isolate_mutex_);
|
|
auto data = per_isolate_[isolate];
|
|
CHECK_NOT_NULL(data.first);
|
|
return data.second;
|
|
}
|
|
|
|
bool NodePlatform::FlushForegroundTasks(Isolate* isolate) {
|
|
std::shared_ptr<PerIsolatePlatformData> per_isolate = ForNodeIsolate(isolate);
|
|
if (!per_isolate) return false;
|
|
return per_isolate->FlushForegroundTasksInternal();
|
|
}
|
|
|
|
std::unique_ptr<v8::JobHandle> NodePlatform::CreateJobImpl(
|
|
v8::TaskPriority priority,
|
|
std::unique_ptr<v8::JobTask> job_task,
|
|
const v8::SourceLocation& location) {
|
|
if (debug_log_level_ != PlatformDebugLogLevel::kNone) {
|
|
fprintf(stderr,
|
|
"\nNodePlatform::CreateJobImpl %s %p",
|
|
GetTaskPriorityName(priority),
|
|
job_task.get());
|
|
PrintSourceLocation(location);
|
|
if (debug_log_level_ == PlatformDebugLogLevel::kVerbose) {
|
|
DumpNativeBacktrace(stderr);
|
|
}
|
|
fflush(stderr);
|
|
}
|
|
return v8::platform::NewDefaultJobHandle(
|
|
this, priority, std::move(job_task), NumberOfWorkerThreads());
|
|
}
|
|
|
|
bool NodePlatform::IdleTasksEnabled(Isolate* isolate) {
|
|
return ForIsolate(isolate)->IdleTasksEnabled();
|
|
}
|
|
|
|
std::shared_ptr<v8::TaskRunner> NodePlatform::GetForegroundTaskRunner(
|
|
Isolate* isolate, v8::TaskPriority priority) {
|
|
return ForIsolate(isolate)->GetForegroundTaskRunner();
|
|
}
|
|
|
|
double NodePlatform::MonotonicallyIncreasingTime() {
|
|
// Convert nanos to seconds.
|
|
return uv_hrtime() / 1e9;
|
|
}
|
|
|
|
double NodePlatform::CurrentClockTimeMillis() {
|
|
return SystemClockTimeMillis();
|
|
}
|
|
|
|
v8::TracingController* NodePlatform::GetTracingController() {
|
|
CHECK_NOT_NULL(tracing_controller_);
|
|
return tracing_controller_;
|
|
}
|
|
|
|
Platform::StackTracePrinter NodePlatform::GetStackTracePrinter() {
|
|
return []() {
|
|
fprintf(stderr, "\n");
|
|
DumpNativeBacktrace(stderr);
|
|
fflush(stderr);
|
|
};
|
|
}
|
|
|
|
v8::PageAllocator* NodePlatform::GetPageAllocator() {
|
|
return page_allocator_;
|
|
}
|
|
|
|
template <class T>
|
|
TaskQueue<T>::TaskQueue()
|
|
: lock_(),
|
|
tasks_available_(),
|
|
outstanding_tasks_drained_(),
|
|
outstanding_tasks_(0),
|
|
stopped_(false),
|
|
task_queue_() {}
|
|
|
|
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_->next_sequence_++});
|
|
queue_->tasks_available_.Signal(lock_);
|
|
}
|
|
|
|
template <class T>
|
|
std::unique_ptr<T> TaskQueue<T>::Locked::Pop() {
|
|
if (queue_->task_queue_.empty()) {
|
|
return std::unique_ptr<T>(nullptr);
|
|
}
|
|
return queue_->PopTask();
|
|
}
|
|
|
|
template <class T>
|
|
std::unique_ptr<T> TaskQueue<T>::Locked::BlockingPop() {
|
|
while (queue_->task_queue_.empty() && !queue_->stopped_) {
|
|
queue_->tasks_available_.Wait(lock_);
|
|
}
|
|
if (queue_->stopped_) {
|
|
return std::unique_ptr<T>(nullptr);
|
|
}
|
|
return queue_->PopTask();
|
|
}
|
|
|
|
template <class T>
|
|
void TaskQueue<T>::Locked::NotifyOfOutstandingCompletion() {
|
|
if (--queue_->outstanding_tasks_ == 0) {
|
|
queue_->outstanding_tasks_drained_.Broadcast(lock_);
|
|
}
|
|
}
|
|
|
|
template <class T>
|
|
void TaskQueue<T>::Locked::BlockingDrain() {
|
|
while (queue_->outstanding_tasks_ > 0) {
|
|
queue_->outstanding_tasks_drained_.Wait(lock_);
|
|
}
|
|
}
|
|
|
|
template <class T>
|
|
void TaskQueue<T>::Locked::Stop() {
|
|
queue_->stopped_ = true;
|
|
queue_->tasks_available_.Broadcast(lock_);
|
|
}
|
|
|
|
template <class T>
|
|
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
|
|
// for the task runner to be found from the map. After the isolate is torn
|
|
// down, we need to remove it from the map before we can free the address,
|
|
// so that when another Isolate::Allocate() is called, that would not be
|
|
// allocated to the same address and be registered on an existing map
|
|
// entry.
|
|
// Refs: https://github.com/nodejs/node/issues/30846
|
|
isolate->Deinitialize();
|
|
this->UnregisterIsolate(isolate);
|
|
Isolate::Free(isolate);
|
|
}
|
|
|
|
} // namespace node
|