inspector: enable Inspector JS API in workers

PR-URL: nodejs#22769
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>

lib/inspector.js

@@ -14,7 +14,7 @@ const util = require('util');
 const { Connection, open, url } = process.binding('inspector');
 const { originalConsole } = require('internal/process/per_thread');
 
-if (!Connection || !require('internal/worker').isMainThread)
+if (!Connection)
   throw new ERR_INSPECTOR_NOT_AVAILABLE();
 
 const connectionSymbol = Symbol('connectionProperty');

lib/internal/util/inspector.js

@@ -1,8 +1,6 @@
 'use strict';
 
-// TODO(addaleax): Figure out how to integrate the inspector with workers.
-const hasInspector = process.config.variables.v8_enable_inspector === 1 &&
-  require('internal/worker').isMainThread;
+const hasInspector = process.config.variables.v8_enable_inspector === 1;
 const inspector = hasInspector ? require('inspector') : undefined;
 
 let session;

lib/internal/worker.js

@@ -17,6 +17,7 @@ const { MessagePort, MessageChannel } = internalBinding('messaging');
 const { handle_onclose } = internalBinding('symbols');
 const { clearAsyncIdStack } = require('internal/async_hooks');
 const { serializeError, deserializeError } = require('internal/error-serdes');
+const { pathToFileURL } = require('url');
 
 util.inherits(MessagePort, EventEmitter);
 
@@ -257,8 +258,9 @@ class Worker extends EventEmitter {
       }
     }
 
+    const url = options.eval ? null : pathToFileURL(filename);
     // Set up the C++ handle for the worker, as well as some internal wiring.
-    this[kHandle] = new WorkerImpl();
+    this[kHandle] = new WorkerImpl(url);
     this[kHandle].onexit = (code) => this[kOnExit](code);
     this[kPort] = this[kHandle].messagePort;
     this[kPort].on('message', (data) => this[kOnMessage](data));

src/inspector/main_thread_interface.cc

@@ -354,13 +354,6 @@ void MainThreadHandle::Reset() {
   main_thread_ = nullptr;
 }
 
-Agent* MainThreadHandle::GetInspectorAgent() {
-  Mutex::ScopedLock scoped_lock(block_lock_);
-  if (main_thread_ == nullptr)
-    return nullptr;
-  return main_thread_->inspector_agent();
-}
-
 std::unique_ptr<InspectorSessionDelegate>
 MainThreadHandle::MakeDelegateThreadSafe(
     std::unique_ptr<InspectorSessionDelegate> delegate) {

src/inspector/main_thread_interface.h

@@ -54,7 +54,6 @@ class MainThreadHandle : public std::enable_shared_from_this<MainThreadHandle> {
     return ++next_object_id_;
   }
   bool Post(std::unique_ptr<Request> request);
-  Agent* GetInspectorAgent();
   std::unique_ptr<InspectorSessionDelegate> MakeDelegateThreadSafe(
       std::unique_ptr<InspectorSessionDelegate> delegate);
   bool Expired();

src/inspector/tracing_agent.cc

@@ -74,11 +74,14 @@ DispatchResponse TracingAgent::start(
   if (categories_set.empty())
     return DispatchResponse::Error("At least one category should be enabled");
 
-  trace_writer_ = env_->tracing_agent_writer()->agent()->AddClient(
-      categories_set,
-      std::unique_ptr<InspectorTraceWriter>(
-          new InspectorTraceWriter(frontend_.get())),
-      tracing::Agent::kIgnoreDefaultCategories);
+  auto* writer = env_->tracing_agent_writer();
+  if (writer != nullptr) {
+    trace_writer_ = env_->tracing_agent_writer()->agent()->AddClient(
+        categories_set,
+        std::unique_ptr<InspectorTraceWriter>(
+            new InspectorTraceWriter(frontend_.get())),
+        tracing::Agent::kIgnoreDefaultCategories);
+  }
   return DispatchResponse::OK();
 }
 

src/inspector_agent.cc

@@ -190,6 +190,12 @@ static int StartDebugSignalHandler() {
 
 const int CONTEXT_GROUP_ID = 1;
 
+std::string GetWorkerLabel(node::Environment* env) {
+  std::ostringstream result;
+  result << "Worker[" << env->thread_id() << "]";
+  return result.str();
+}
+
 class ChannelImpl final : public v8_inspector::V8Inspector::Channel,
                           public protocol::FrontendChannel {
  public:
@@ -393,10 +399,13 @@ bool IsFilePath(const std::string& path) {
 
 class NodeInspectorClient : public V8InspectorClient {
  public:
-  explicit NodeInspectorClient(node::Environment* env) : env_(env) {
+  explicit NodeInspectorClient(node::Environment* env, bool is_main)
+      : env_(env), is_main_(is_main) {
     client_ = V8Inspector::create(env->isolate(), this);
     // TODO(bnoordhuis) Make name configurable from src/node.cc.
-    ContextInfo info(GetHumanReadableProcessName());
+    std::string name =
+        is_main_ ? GetHumanReadableProcessName() : GetWorkerLabel(env);
+    ContextInfo info(name);
     info.is_default = true;
     contextCreated(env->context(), info);
   }
@@ -625,6 +634,7 @@ class NodeInspectorClient : public V8InspectorClient {
   }
 
   node::Environment* env_;
+  bool is_main_;
   bool running_nested_loop_ = false;
   std::unique_ptr<V8Inspector> client_;
   std::unordered_map<int, std::unique_ptr<ChannelImpl>> channels_;
@@ -642,13 +652,23 @@ Agent::Agent(Environment* env)
   : parent_env_(env),
     debug_options_(env->options()->debug_options) {}
 
-Agent::~Agent() = default;
+Agent::~Agent() {
+  if (start_io_thread_async.data == this) {
+    start_io_thread_async.data = nullptr;
+    // This is global, will never get freed
+    uv_close(reinterpret_cast<uv_handle_t*>(&start_io_thread_async), nullptr);
+  }
+}
 
 bool Agent::Start(const std::string& path,
-                  std::shared_ptr<DebugOptions> options) {
+                  std::shared_ptr<DebugOptions> options,
+                  bool is_main) {
+  if (options == nullptr) {
+    options = std::make_shared<DebugOptions>();
+  }
   path_ = path;
   debug_options_ = options;
-  client_ = std::make_shared<NodeInspectorClient>(parent_env_);
+  client_ = std::make_shared<NodeInspectorClient>(parent_env_, is_main);
   if (parent_env_->is_main_thread()) {
     CHECK_EQ(0, uv_async_init(parent_env_->event_loop(),
                               &start_io_thread_async,

src/inspector_agent.h

@@ -47,7 +47,9 @@ class Agent {
   ~Agent();
 
   // Create client_, may create io_ if option enabled
-  bool Start(const std::string& path, std::shared_ptr<DebugOptions> options);
+  bool Start(const std::string& path,
+             std::shared_ptr<DebugOptions> options,
+             bool is_worker);
   // Stop and destroy io_
   void Stop();
 

src/node.cc

@@ -327,7 +327,7 @@ static struct {
     // right away on the websocket port and fails to bind/etc, this will return
     // false.
     return env->inspector_agent()->Start(
-        script_path == nullptr ? "" : script_path, options);
+        script_path == nullptr ? "" : script_path, options, true);
   }
 
   bool InspectorStarted(Environment* env) {

src/node_worker.cc

@@ -35,10 +35,26 @@ namespace {
 uint64_t next_thread_id = 1;
 Mutex next_thread_id_mutex;
 
+#if NODE_USE_V8_PLATFORM && HAVE_INSPECTOR
+void StartWorkerInspector(Environment* child, const std::string& url) {
+  child->inspector_agent()->Start(url, nullptr, false);
+}
+
+void WaitForWorkerInspectorToStop(Environment* child) {
+  child->inspector_agent()->WaitForDisconnect();
+  child->inspector_agent()->Stop();
+}
+
+#else
+// No-ops
+void StartWorkerInspector(Environment* child, const std::string& url) {}
+void WaitForWorkerInspectorToStop(Environment* child) {}
+#endif
+
 }  // anonymous namespace
 
-Worker::Worker(Environment* env, Local<Object> wrap)
-    : AsyncWrap(env, wrap, AsyncWrap::PROVIDER_WORKER) {
+Worker::Worker(Environment* env, Local<Object> wrap, const std::string& url)
+    : AsyncWrap(env, wrap, AsyncWrap::PROVIDER_WORKER), url_(url) {
   // Generate a new thread id.
   {
     Mutex::ScopedLock next_thread_id_lock(next_thread_id_mutex);
@@ -155,6 +171,7 @@ void Worker::Run() {
         env_->async_hooks()->pop_async_id(1);
 
         Debug(this, "Loaded environment for worker %llu", thread_id_);
+        StartWorkerInspector(env_.get(), url_);
       }
 
       {
@@ -215,6 +232,7 @@ void Worker::Run() {
       env_->stop_sub_worker_contexts();
       env_->RunCleanup();
       RunAtExit(env_.get());
+      WaitForWorkerInspectorToStop(env_.get());
 
       {
         Mutex::ScopedLock stopped_lock(stopped_mutex_);
@@ -351,7 +369,15 @@ void Worker::New(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
-  new Worker(env, args.This());
+  std::string url;
+  // Argument might be a string or URL
+  if (args.Length() == 1 && !args[0]->IsNullOrUndefined()) {
+    Utf8Value value(
+        args.GetIsolate(),
+        args[0]->ToString(env->context()).FromMaybe(v8::Local<v8::String>()));
+    url.append(value.out(), value.length());
+  }
+  new Worker(env, args.This(), url);
 }
 
 void Worker::StartThread(const FunctionCallbackInfo<Value>& args) {

src/node_worker.h

@@ -12,7 +12,7 @@ namespace worker {
 // A worker thread, as represented in its parent thread.
 class Worker : public AsyncWrap {
  public:
-  Worker(Environment* env, v8::Local<v8::Object> wrap);
+  Worker(Environment* env, v8::Local<v8::Object> wrap, const std::string& url);
   ~Worker();
 
   // Run the worker. This is only called from the worker thread.
@@ -52,6 +52,7 @@ class Worker : public AsyncWrap {
   uv_loop_t loop_;
   DeleteFnPtr<IsolateData, FreeIsolateData> isolate_data_;
   DeleteFnPtr<Environment, FreeEnvironment> env_;
+  const std::string url_;
   v8::Isolate* isolate_ = nullptr;
   DeleteFnPtr<ArrayBufferAllocator, FreeArrayBufferAllocator>
       array_buffer_allocator_;

test/common/README.md

@@ -361,6 +361,11 @@ was disabled at compile time.
 Skip the rest of the tests in the current file when the Node.js executable
 was compiled with a pointer size smaller than 64 bits.
 
+### skipIfWorker()
+
+Skip the rest of the tests in the current file when not running on a main
+thread.
+
 ## ArrayStream Module
 
 The `ArrayStream` module provides a simple `Stream` that pushes elements from

test/common/index.js

@@ -614,10 +614,6 @@ function skipIfInspectorDisabled() {
   if (process.config.variables.v8_enable_inspector === 0) {
     skip('V8 inspector is disabled');
   }
-  if (!isMainThread) {
-    // TODO(addaleax): Fix me.
-    skip('V8 inspector is not available in Workers');
-  }
 }
 
 function skipIf32Bits() {
@@ -626,6 +622,12 @@ function skipIf32Bits() {
   }
 }
 
+function skipIfWorker() {
+  if (!isMainThread) {
+    skip('This test only works on a main thread');
+  }
+}
+
 function getArrayBufferViews(buf) {
   const { buffer, byteOffset, byteLength } = buf;
 
@@ -744,6 +746,7 @@ module.exports = {
   skipIf32Bits,
   skipIfEslintMissing,
   skipIfInspectorDisabled,
+  skipIfWorker,
 
   get localhostIPv6() { return '::1'; },
 

test/parallel/test-bootstrap-modules.js

@@ -11,4 +11,4 @@ const list = process.moduleLoadList.slice();
 
 const assert = require('assert');
 
-assert(list.length <= 74, list);
+assert(list.length <= 75, list);

test/parallel/test-inspector-port-zero-cluster.js

@@ -3,6 +3,7 @@
 const common = require('../common');
 
 common.skipIfInspectorDisabled();
+common.skipIfWorker();
 
 // Assert that even when started with `--inspect=0` workers are assigned
 // consecutive (i.e. deterministically predictable) debug ports

test/parallel/test-inspector-tracing-domain.js

@@ -3,6 +3,7 @@
 const common = require('../common');
 
 common.skipIfInspectorDisabled();
+common.skipIfWorker(); // https://github.com/nodejs/node/issues/22767
 
 const assert = require('assert');
 const { Session } = require('inspector');

test/parallel/test-trace-events-dynamic-enable.js

@@ -3,6 +3,7 @@
 const common = require('../common');
 
 common.skipIfInspectorDisabled();
+common.skipIfWorker(); // https://github.com/nodejs/node/issues/22767
 
 const assert = require('assert');
 const { performance } = require('perf_hooks');

test/parallel/test-warn-sigprof.js

@@ -10,6 +10,8 @@ common.skipIfInspectorDisabled();
 if (common.isWindows)
   common.skip('test does not apply to Windows');
 
+common.skipIfWorker(); // Worker inspector never has a server running
+
 common.expectWarning('Warning',
                      'process.on(SIGPROF) is reserved while debugging',
                      common.noWarnCode);

test/sequential/test-inspector-async-hook-setup-at-signal.js

@@ -6,6 +6,8 @@ common.skipIf32Bits();
 const { NodeInstance } = require('../common/inspector-helper.js');
 const assert = require('assert');
 
+common.skipIfWorker(); // Signal starts a server for a main thread inspector
+
 const script = `
 process._rawDebug('Waiting until a signal enables the inspector...');
 let waiting = setInterval(waitUntilDebugged, 50);

test/sequential/test-inspector-contexts.js

@@ -33,7 +33,11 @@ async function testContextCreatedAndDestroyed() {
       // the PID.
       strictEqual(name.includes(`[${process.pid}]`), true);
     } else {
-      strictEqual(`${process.argv0}[${process.pid}]`, name);
+      let expects = `${process.argv0}[${process.pid}]`;
+      if (!common.isMainThread) {
+        expects = `Worker[${require('worker_threads').threadId}]`;
+      }
+      strictEqual(expects, name);
     }
     strictEqual(origin, '',
                 JSON.stringify(contextCreated));

test/sequential/test-inspector-port-cluster.js

@@ -3,6 +3,7 @@
 const common = require('../common');
 
 common.skipIfInspectorDisabled();
+common.skipIfWorker();
 
 const assert = require('assert');
 const cluster = require('cluster');