worker: add cpu profile APIs for worker

Author: theanarkh

doc/api/errors.md

@@ -3291,6 +3291,24 @@ added: v18.1.0
 The `Response` that has been passed to `WebAssembly.compileStreaming` or to
 `WebAssembly.instantiateStreaming` is not a valid WebAssembly response.
 
+<a id="ERR_WORKER_CPU_PROFILE_ALREADY_STARTED"></a>
+
+### `ERR_WORKER_CPU_PROFILE_ALREADY_STARTED`
+
+The `Worker` CPU profile with the given name is already started.
+
+<a id="ERR_WORKER_CPU_PROFILE_NOT_STARTED"></a>
+
+### `ERR_WORKER_CPU_PROFILE_NOT_STARTED`
+
+The `Worker` CPU profile with the given name is not started.
+
+<a id="ERR_WORKER_CPU_PROFILE_TOO_MANY"></a>
+
+### `ERR_WORKER_CPU_PROFILE_TOO_MANY`
+
+There are too many CPU profiles being collected in `Worker`.
+
 <a id="ERR_WORKER_INIT_FAILED"></a>
 
 ### `ERR_WORKER_INIT_FAILED`

doc/api/worker_threads.md

@@ -1957,6 +1957,18 @@ this matches its values.
 
 If the worker has stopped, the return value is an empty object.
 
+### `worker.startCpuProfile(name)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* name: {string}
+* Returns: {Promise}
+
+Starting a CPU profile with the given `name` for the worker thread. The profile can be stopped
+with [`worker.stopCpuProfile(name)`][].
+
 ### `worker.stderr`
 
 <!-- YAML
@@ -1995,6 +2007,18 @@ inside the worker thread. If `stdout: true` was not passed to the
 [`Worker`][] constructor, then data is piped to the parent thread's
 [`process.stdout`][] stream.
 
+### `worker.stopCpuProfile(name)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* name: {string}
+* Returns: {Promise}
+
+Stopping a CPU profile with the given `name` which is passed to [`worker.startCpuProfile(name)`][]
+for the worker thread. Returns a Promise that fulfills with the profile data or throws an error.
+
 ### `worker.terminate()`
 
 <!-- YAML
@@ -2176,6 +2200,8 @@ thread spawned will spawn another until the application crashes.
 [`worker.SHARE_ENV`]: #workershare_env
 [`worker.on('message')`]: #event-message_1
 [`worker.postMessage()`]: #workerpostmessagevalue-transferlist
+[`worker.startCpuProfile(name)`]: #workerstartcpuprofilename
+[`worker.stopCpuProfile(name)`]: #workerstopcpuprofilename
 [`worker.terminate()`]: #workerterminate
 [`worker.threadId`]: #workerthreadid_1
 [`worker.threadName`]: #workerthreadname_1

lib/internal/errors.js

@@ -1898,6 +1898,9 @@ E('ERR_VM_MODULE_NOT_MODULE',
 E('ERR_VM_MODULE_STATUS', 'Module status %s', Error);
 E('ERR_WASI_ALREADY_STARTED', 'WASI instance has already started', Error);
 E('ERR_WEBASSEMBLY_RESPONSE', 'WebAssembly response %s', TypeError);
+E('ERR_WORKER_CPU_PROFILE_ALREADY_STARTED', 'Worker CPU profile already started with name(%s)', Error);
+E('ERR_WORKER_CPU_PROFILE_NOT_STARTED', 'Worker CPU profile not started with name(%s)', Error);
+E('ERR_WORKER_CPU_PROFILE_TOO_MANY', 'Worker has too many CPU profiles', Error);
 E('ERR_WORKER_INIT_FAILED', 'Worker initialization failure: %s', Error);
 E('ERR_WORKER_INVALID_EXEC_ARGV', (errors, msg = 'invalid execArgv flags') =>
  `Initiated Worker with ${msg}: ${ArrayPrototypeJoin(errors, ', ')}`,

lib/internal/worker.js

@@ -36,6 +36,9 @@ const {
 
 const errorCodes = require('internal/errors').codes;
 const {
+ ERR_WORKER_CPU_PROFILE_ALREADY_STARTED,
+ ERR_WORKER_CPU_PROFILE_NOT_STARTED,
+ ERR_WORKER_CPU_PROFILE_TOO_MANY,
  ERR_WORKER_NOT_RUNNING,
  ERR_WORKER_PATH,
  ERR_WORKER_UNSERIALIZABLE_ERROR,
@@ -506,6 +509,32 @@ class Worker extends EventEmitter {
  };
  });
  }
+
+ // TODO(theanarkh): add options
+ startCpuProfile(name) {
+ validateString(name, 'name');
+ const startTaker = this[kHandle]?.startCpuProfile(name);
+ return new Promise((resolve, reject) => {
+ if (!startTaker) return reject(new ERR_WORKER_NOT_RUNNING());
+ startTaker.ondone = (status) => {
+ if (status === 1) return reject(new ERR_WORKER_CPU_PROFILE_ALREADY_STARTED(name));
+ if (status === 2) return reject(new ERR_WORKER_CPU_PROFILE_TOO_MANY());
+ resolve();
+ };
+ });
+ }
+
+ stopCpuProfile(name) {
+ validateString(name, 'name');
+ const stopTaker = this[kHandle]?.stopCpuProfile(name);
+ return new Promise((resolve, reject) => {
+ if (!stopTaker) return reject(new ERR_WORKER_NOT_RUNNING());
+ stopTaker.ondone = (status, profile) => {
+ if (status === 1) return reject(new ERR_WORKER_CPU_PROFILE_NOT_STARTED(name));
+ resolve(profile);
+ };
+ });
+ }
 }
 
 /**

src/async_wrap.h

@@ -79,6 +79,7 @@ namespace node {
  V(UDPWRAP) \
  V(SIGINTWATCHDOG) \
  V(WORKER) \
+ V(WORKERCPUPROFILE) \
  V(WORKERCPUUSAGE) \
  V(WORKERHEAPSNAPSHOT) \
  V(WORKERHEAPSTATISTICS) \

src/env_properties.h

@@ -483,6 +483,7 @@
  V(tcp_constructor_template, v8::FunctionTemplate) \
  V(tty_constructor_template, v8::FunctionTemplate) \
  V(write_wrap_template, v8::ObjectTemplate) \
+ V(worker_cpu_profile_taker_template, v8::ObjectTemplate) \
  V(worker_cpu_usage_taker_template, v8::ObjectTemplate) \
  V(worker_heap_snapshot_taker_template, v8::ObjectTemplate) \
  V(worker_heap_statistics_taker_template, v8::ObjectTemplate) \

src/node_worker.cc

@@ -23,6 +23,9 @@ using v8::Array;
 using v8::ArrayBuffer;
 using v8::Boolean;
 using v8::Context;
+using v8::CpuProfile;
+using v8::CpuProfiler;
+using v8::CpuProfilingStatus;
 using v8::Float64Array;
 using v8::FunctionCallbackInfo;
 using v8::FunctionTemplate;
@@ -495,7 +498,10 @@ Worker::~Worker() {
  CHECK(stopped_);
  CHECK_NULL(env_);
  CHECK(!tid_.has_value());
-
+ if (!cpu_profiler_) {
+ cpu_profiler_->Dispose();
+ cpu_profiler_ = nullptr;
+ }
  Debug(this, "Worker %llu destroyed", thread_id_.id);
 }
 
@@ -897,6 +903,164 @@ void Worker::CpuUsage(const FunctionCallbackInfo<Value>& args) {
  }
 }
 
+class WorkerCpuProfileTaker : public AsyncWrap {
+ public:
+ WorkerCpuProfileTaker(Environment* env, Local<Object> obj)
+ : AsyncWrap(env, obj, AsyncWrap::PROVIDER_WORKERCPUPROFILE) {}
+
+ SET_NO_MEMORY_INFO()
+ SET_MEMORY_INFO_NAME(WorkerCpuProfileTaker)
+ SET_SELF_SIZE(WorkerCpuProfileTaker)
+};
+
+void Worker::StartCpuProfile(const FunctionCallbackInfo<Value>& args) {
+ Worker* w;
+ ASSIGN_OR_RETURN_UNWRAP(&w, args.This());
+ Environment* env = w->env();
+
+ CHECK(args[0]->IsString());
+ node::Utf8Value name(env->isolate(), args[0]);
+
+ AsyncHooks::DefaultTriggerAsyncIdScope trigger_id_scope(w);
+ Local<Object> wrap;
+ if (!env->worker_cpu_profile_taker_template()
+ ->NewInstance(env->context())
+ .ToLocal(&wrap)) {
+ return;
+ }
+
+ BaseObjectPtr<WorkerCpuProfileTaker> taker =
+ MakeDetachedBaseObject<WorkerCpuProfileTaker>(env, wrap);
+
+ bool scheduled = w->RequestInterrupt([taker = std::move(taker),
+ name = name.ToString(),
+ env,
+ w](Environment* worker_env) mutable {
+ Isolate* isolate = worker_env->isolate();
+ if (!w->cpu_profiler_) {
+ w->cpu_profiler_ = CpuProfiler::New(isolate);
+ }
+ Local<String> title = String::NewFromUtf8(isolate,
+ name.data(),
+ NewStringType::kNormal,
+ name.size()).ToLocalChecked();
+ CpuProfilingStatus status = w->cpu_profiler_->StartProfiling(title, true);
+ env->SetImmediateThreadsafe(
+ [taker = std::move(taker),
+ status
+ ](Environment* env) mutable {
+ Isolate* isolate = env->isolate();
+ HandleScope handle_scope(isolate);
+ Context::Scope context_scope(env->context());
+ AsyncHooks::DefaultTriggerAsyncIdScope trigger_id_scope(taker.get());
+ Local<Value> argv[] = {
+ Number::New(isolate, static_cast<double>(status)) // status
+ };
+ taker->MakeCallback(env->ondone_string(), arraysize(argv), argv);
+ },
+ CallbackFlags::kUnrefed);
+ });
+
+ if (scheduled) {
+ args.GetReturnValue().Set(wrap);
+ }
+}
+
+class JSONOutputStream : public v8::OutputStream {
+ public:
+ JSONOutputStream() {}
+
+ int GetChunkSize() override {
+ return 65536;
+ }
+
+ void EndOfStream() override {}
+
+ WriteResult WriteAsciiChunk(char* data, const int size) override {
+ out_stream_.write(data, size);
+ return kContinue;
+ }
+
+ std::ostringstream& out_stream() { return out_stream_; }
+
+ private:
+ std::ostringstream out_stream_;
+};
+
+void Worker::StopCpuProfile(const FunctionCallbackInfo<Value>& args) {
+ Worker* w;
+ ASSIGN_OR_RETURN_UNWRAP(&w, args.This());
+
+ Environment* env = w->env();
+ CHECK(args[0]->IsString());
+ node::Utf8Value name(env->isolate(), args[0]);
+
+ AsyncHooks::DefaultTriggerAsyncIdScope trigger_id_scope(w);
+ Local<Object> wrap;
+ if (!env->worker_cpu_profile_taker_template()
+ ->NewInstance(env->context())
+ .ToLocal(&wrap)) {
+ return;
+ }
+
+ BaseObjectPtr<WorkerCpuProfileTaker> taker =
+ MakeDetachedBaseObject<WorkerCpuProfileTaker>(env, wrap);
+
+ bool scheduled = w->RequestInterrupt([taker = std::move(taker),
+ name = name.ToString(),
+ env,
+ w](Environment* worker_env) mutable {
+ Local<String> title = String::NewFromUtf8(worker_env->isolate(),
+ name.data(),
+ NewStringType::kNormal,
+ name.size()).ToLocalChecked();
+ bool found = false;
+ auto json_out_stream = std::make_unique<JSONOutputStream>();
+ if (w->cpu_profiler_) {
+ CpuProfile* profile = w->cpu_profiler_->StopProfiling(title);
+ if (profile) {
+ profile->Serialize(json_out_stream.get(),
+ CpuProfile::SerializationFormat::kJSON);
+ profile->Delete();
+ found = true;
+ }
+ }
+ env->SetImmediateThreadsafe(
+ [taker = std::move(taker),
+ json_out_stream = std::move(json_out_stream),
+ found
+ ](Environment* env) mutable {
+ Isolate* isolate = env->isolate();
+ HandleScope handle_scope(isolate);
+ Context::Scope context_scope(env->context());
+ AsyncHooks::DefaultTriggerAsyncIdScope trigger_id_scope(taker.get());
+ Local<Value> argv[] = {
+ Undefined(isolate), // status
+ Undefined(isolate), // profile
+ };
+ if (found) {
+ argv[0] = Number::New(isolate, 0);
+ Local<Value> result;
+ if (ToV8Value(env->context(),
+ json_out_stream->out_stream().str(),
+ isolate).ToLocal(&result)) {
+ argv[1] = result;
+ } else {
+ argv[1] = FIXED_ONE_BYTE_STRING(isolate, "{}");
+ }
+ } else {
+ argv[0] = Number::New(isolate, 1);
+ }
+ taker->MakeCallback(env->ondone_string(), arraysize(argv), argv);
+ },
+ CallbackFlags::kUnrefed);
+ });
+
+ if (scheduled) {
+ args.GetReturnValue().Set(wrap);
+ }
+}
+
 class WorkerHeapStatisticsTaker : public AsyncWrap {
  public:
  WorkerHeapStatisticsTaker(Environment* env, Local<Object> obj)
@@ -1189,6 +1353,8 @@ void CreateWorkerPerIsolateProperties(IsolateData* isolate_data,
  SetProtoMethod(isolate, w, "loopStartTime", Worker::LoopStartTime);
  SetProtoMethod(isolate, w, "getHeapStatistics", Worker::GetHeapStatistics);
  SetProtoMethod(isolate, w, "cpuUsage", Worker::CpuUsage);
+ SetProtoMethod(isolate, w, "startCpuProfile", Worker::StartCpuProfile);
+ SetProtoMethod(isolate, w, "stopCpuProfile", Worker::StopCpuProfile);
 
  SetConstructorFunction(isolate, target, "Worker", w);
  }
@@ -1234,6 +1400,20 @@ void CreateWorkerPerIsolateProperties(IsolateData* isolate_data,
  isolate_data->set_worker_cpu_usage_taker_template(wst->InstanceTemplate());
  }
 
+ {
+ Local<FunctionTemplate> wst = NewFunctionTemplate(isolate, nullptr);
+
+ wst->InstanceTemplate()->SetInternalFieldCount(
+ WorkerCpuProfileTaker::kInternalFieldCount);
+ wst->Inherit(AsyncWrap::GetConstructorTemplate(isolate_data));
+
+ Local<String> wst_string =
+ FIXED_ONE_BYTE_STRING(isolate, "WorkerCpuProfileTaker");
+ wst->SetClassName(wst_string);
+ isolate_data->set_worker_cpu_profile_taker_template(
+ wst->InstanceTemplate());
+ }
+
  SetMethod(isolate, target, "getEnvMessagePort", GetEnvMessagePort);
 }
 
@@ -1311,6 +1491,8 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
  registry->Register(Worker::LoopStartTime);
  registry->Register(Worker::GetHeapStatistics);
  registry->Register(Worker::CpuUsage);
+ registry->Register(Worker::StartCpuProfile);
+ registry->Register(Worker::StopCpuProfile);
 }
 
 } // anonymous namespace

src/node_worker.h

@@ -8,6 +8,8 @@
 #include "node_exit_code.h"
 #include "node_messaging.h"
 #include "uv.h"
+#include "v8-profiler.h"
+#include "json_utils.h"
 
 namespace node {
 
@@ -81,6 +83,8 @@ class Worker : public AsyncWrap {
  static void GetHeapStatistics(
  const v8::FunctionCallbackInfo<v8::Value>& args);
  static void CpuUsage(const v8::FunctionCallbackInfo<v8::Value>& args);
+ static void StartCpuProfile(const v8::FunctionCallbackInfo<v8::Value>& args);
+ static void StopCpuProfile(const v8::FunctionCallbackInfo<v8::Value>& args);
 
  private:
  bool CreateEnvMessagePort(Environment* env);
@@ -107,6 +111,7 @@ class Worker : public AsyncWrap {
  uintptr_t stack_base_ = 0;
  // Optional name used for debugging in inspector and trace events.
  std::string name_;
+ v8::CpuProfiler* cpu_profiler_ = nullptr;
 
  // Custom resource constraints:
  double resource_limits_[kTotalResourceLimitCount];