Files
node/src/node_platform.cc
theanarkh 83c955d8c2 src: add name for more threads
PR-URL: https://github.com/nodejs/node/pull/59601
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Chengzhong Wu <legendecas@gmail.com>
Reviewed-By: Darshan Sen <raisinten@gmail.com>
2025-08-27 17:20:32 +00:00

863 lines
29 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("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();
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();
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);
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.
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_;
};
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 suggets. As a compromise, we currently only block on
// user-blocking tasks to reduce the chance of deadlocks while making sure
// that criticl user-blocking tasks are not lost.
worker_thread_task_runner_->BlockingDrain();
} while (per_isolate->FlushForegroundTasksInternal());
}
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();
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();
});
});
}
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();
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>
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_->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);
}
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;
}
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);
}
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;
}
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>
TaskQueue<T>::PriorityQueue TaskQueue<T>::Locked::PopAll() {
TaskQueue<T>::PriorityQueue result;
result.swap(queue_->task_queue_);
return result;
}
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