From a26865849631b370866db04ac808373606164191 Mon Sep 17 00:00:00 2001
From: Eugene Ostroukhov <eostroukhov@gmail.com>
Date: Fri, 26 Jul 2019 10:40:54 -0700
Subject: [PATCH] inspector: new API - Session.connectToMainThread

This API is designed to enable worker threads use Inspector protocol
on main thread (and other workers through NodeWorker domain).

Note that worker can cause dead lock by suspending itself. I will
work on a new API that will allow workers to be hidden from the
inspector.

Fixes: https://github.com/nodejs/node/issues/28828
PR-URL: https://github.com/nodejs/node/pull/28870
Reviewed-By: Aleksei Koziatinskii <ak239spb@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
---
 doc/api/errors.md                             |   6 +
 doc/api/inspector.md                          |  12 +-
 lib/inspector.js                              |  14 ++
 lib/internal/errors.js                        |   1 +
 src/inspector/worker_inspector.cc             |   6 +
 src/inspector/worker_inspector.h              |   5 +
 src/inspector_agent.cc                        |  11 ++
 src/inspector_agent.h                         |   9 +-
 src/inspector_js_api.cc                       |  56 ++++--
 .../test-inspector-bindings.js                |   3 +-
 .../test-inspector-connect-main-thread.js     | 179 ++++++++++++++++++
 11 files changed, 282 insertions(+), 20 deletions(-)
 rename test/{sequential => parallel}/test-inspector-bindings.js (98%)
 create mode 100644 test/parallel/test-inspector-connect-main-thread.js

diff --git a/doc/api/errors.md b/doc/api/errors.md
index d2946435ca3b98..87a080b081f01b 100644
--- a/doc/api/errors.md
+++ b/doc/api/errors.md
@@ -1213,6 +1213,12 @@ The `inspector` module is not available for use.
 While using the `inspector` module, an attempt was made to use the inspector
 before it was connected.
 
+<a id="ERR_INSPECTOR_NOT_WORKER"></a>
+### ERR_INSPECTOR_NOT_WORKER
+
+An API was called on the main thread that can only be used from
+the worker thread.
+
 <a id="ERR_INVALID_ADDRESS_FAMILY"></a>
 ### ERR_INVALID_ADDRESS_FAMILY
 
diff --git a/doc/api/inspector.md b/doc/api/inspector.md
index f82c53f5bc3f51..96c3ad03c06354 100644
--- a/doc/api/inspector.md
+++ b/doc/api/inspector.md
@@ -121,9 +121,15 @@ session.on('Debugger.paused', ({ params }) => {
 added: v8.0.0
 -->
 
-Connects a session to the inspector back-end. An exception will be thrown
-if there is already a connected session established either through the API or by
-a front-end connected to the Inspector WebSocket port.
+Connects a session to the inspector back-end.
+
+### session.connectToMainThread()
+<!-- YAML
+added: REPLACEME
+-->
+
+Connects a session to the main thread inspector back-end. An exception will
+be thrown if this API was not called on a Worker thread.
 
 ### session.disconnect()
 <!-- YAML
diff --git a/lib/inspector.js b/lib/inspector.js
index 198d87ae44ccb0..08cf938999183a 100644
--- a/lib/inspector.js
+++ b/lib/inspector.js
@@ -9,6 +9,7 @@ const {
   ERR_INSPECTOR_NOT_AVAILABLE,
   ERR_INSPECTOR_NOT_CONNECTED,
   ERR_INSPECTOR_NOT_ACTIVE,
+  ERR_INSPECTOR_NOT_WORKER,
   ERR_INVALID_ARG_TYPE,
   ERR_INVALID_CALLBACK
 } = require('internal/errors').codes;
@@ -20,8 +21,11 @@ if (!hasInspector)
 const EventEmitter = require('events');
 const { validateString } = require('internal/validators');
 const util = require('util');
+const { isMainThread } = require('worker_threads');
+
 const {
   Connection,
+  MainThreadConnection,
   open,
   url,
   waitForDebugger
@@ -47,6 +51,16 @@ class Session extends EventEmitter {
       new Connection((message) => this[onMessageSymbol](message));
   }
 
+  connectToMainThread() {
+    if (isMainThread)
+      throw new ERR_INSPECTOR_NOT_WORKER();
+    if (this[connectionSymbol])
+      throw new ERR_INSPECTOR_ALREADY_CONNECTED('The inspector session');
+    this[connectionSymbol] =
+      new MainThreadConnection(
+        (message) => queueMicrotask(() => this[onMessageSymbol](message)));
+  }
+
   [onMessageSymbol](message) {
     const parsed = JSON.parse(message);
     try {
diff --git a/lib/internal/errors.js b/lib/internal/errors.js
index a5d66965171ebd..eff688e6f5fc07 100644
--- a/lib/internal/errors.js
+++ b/lib/internal/errors.js
@@ -888,6 +888,7 @@ E('ERR_INSPECTOR_COMMAND', 'Inspector error %d: %s', Error);
 E('ERR_INSPECTOR_NOT_ACTIVE', 'Inspector is not active', Error);
 E('ERR_INSPECTOR_NOT_AVAILABLE', 'Inspector is not available', Error);
 E('ERR_INSPECTOR_NOT_CONNECTED', 'Session is not connected', Error);
+E('ERR_INSPECTOR_NOT_WORKER', 'Current thread is not a worker', Error);
 E('ERR_INTERNAL_ASSERTION', (message) => {
   const suffix = 'This is caused by either a bug in Node.js ' +
     'or incorrect usage of Node.js internals.\n' +
diff --git a/src/inspector/worker_inspector.cc b/src/inspector/worker_inspector.cc
index 48679d2f6986b3..deeddcc0836dd1 100644
--- a/src/inspector/worker_inspector.cc
+++ b/src/inspector/worker_inspector.cc
@@ -72,6 +72,12 @@ void ParentInspectorHandle::WorkerStarted(
   parent_thread_->Post(std::move(request));
 }
 
+std::unique_ptr<inspector::InspectorSession> ParentInspectorHandle::Connect(
+    std::unique_ptr<inspector::InspectorSessionDelegate> delegate,
+    bool prevent_shutdown) {
+  return parent_thread_->Connect(std::move(delegate), prevent_shutdown);
+}
+
 void WorkerManager::WorkerFinished(int session_id) {
   children_.erase(session_id);
 }
diff --git a/src/inspector/worker_inspector.h b/src/inspector/worker_inspector.h
index c0961496cec1bd..b01063e01fc890 100644
--- a/src/inspector/worker_inspector.h
+++ b/src/inspector/worker_inspector.h
@@ -12,6 +12,8 @@
 
 namespace node {
 namespace inspector {
+class InspectorSession;
+class InspectorSessionDelegate;
 class MainThreadHandle;
 class WorkerManager;
 
@@ -68,6 +70,9 @@ class ParentInspectorHandle {
     return wait_;
   }
   const std::string& url() const { return url_; }
+  std::unique_ptr<inspector::InspectorSession> Connect(
+      std::unique_ptr<inspector::InspectorSessionDelegate> delegate,
+      bool prevent_shutdown);
 
  private:
   int id_;
diff --git a/src/inspector_agent.cc b/src/inspector_agent.cc
index 323e6646d6f1bb..cf0110144a535d 100644
--- a/src/inspector_agent.cc
+++ b/src/inspector_agent.cc
@@ -828,6 +828,17 @@ std::unique_ptr<InspectorSession> Agent::Connect(
       new SameThreadInspectorSession(session_id, client_));
 }
 
+std::unique_ptr<InspectorSession> Agent::ConnectToMainThread(
+    std::unique_ptr<InspectorSessionDelegate> delegate,
+    bool prevent_shutdown) {
+  CHECK_NOT_NULL(parent_handle_);
+  CHECK_NOT_NULL(client_);
+  auto thread_safe_delegate =
+      client_->getThreadHandle()->MakeDelegateThreadSafe(std::move(delegate));
+  return parent_handle_->Connect(std::move(thread_safe_delegate),
+                                 prevent_shutdown);
+}
+
 void Agent::WaitForDisconnect() {
   CHECK_NOT_NULL(client_);
   bool is_worker = parent_handle_ != nullptr;
diff --git a/src/inspector_agent.h b/src/inspector_agent.h
index 4fb544f85bd145..d5088a1b546904 100644
--- a/src/inspector_agent.h
+++ b/src/inspector_agent.h
@@ -86,12 +86,19 @@ class Agent {
   std::unique_ptr<ParentInspectorHandle> GetParentHandle(
       int thread_id, const std::string& url);
 
-  // Called to create inspector sessions that can be used from the main thread.
+  // Called to create inspector sessions that can be used from the same thread.
   // The inspector responds by using the delegate to send messages back.
   std::unique_ptr<InspectorSession> Connect(
       std::unique_ptr<InspectorSessionDelegate> delegate,
       bool prevent_shutdown);
 
+  // Called from the worker to create inspector sessions that is connected
+  // to the main thread.
+  // The inspector responds by using the delegate to send messages back.
+  std::unique_ptr<InspectorSession> ConnectToMainThread(
+      std::unique_ptr<InspectorSessionDelegate> delegate,
+      bool prevent_shutdown);
+
   void PauseOnNextJavascriptStatement(const std::string& reason);
 
   std::string GetWsUrl() const;
diff --git a/src/inspector_js_api.cc b/src/inspector_js_api.cc
index 8b891a2d611ece..45b1c83bea7f34 100644
--- a/src/inspector_js_api.cc
+++ b/src/inspector_js_api.cc
@@ -37,6 +37,29 @@ std::unique_ptr<StringBuffer> ToProtocolString(Isolate* isolate,
   return StringBuffer::create(StringView(*buffer, buffer.length()));
 }
 
+struct LocalConnection {
+  static std::unique_ptr<InspectorSession> Connect(
+      Agent* inspector, std::unique_ptr<InspectorSessionDelegate> delegate) {
+    return inspector->Connect(std::move(delegate), false);
+  }
+
+  static Local<String> GetClassName(Environment* env) {
+    return FIXED_ONE_BYTE_STRING(env->isolate(), "Connection");
+  }
+};
+
+struct MainThreadConnection {
+  static std::unique_ptr<InspectorSession> Connect(
+      Agent* inspector, std::unique_ptr<InspectorSessionDelegate> delegate) {
+    return inspector->ConnectToMainThread(std::move(delegate), true);
+  }
+
+  static Local<String> GetClassName(Environment* env) {
+    return FIXED_ONE_BYTE_STRING(env->isolate(), "MainThreadConnection");
+  }
+};
+
+template <typename ConnectionType>
 class JSBindingsConnection : public AsyncWrap {
  public:
   class JSBindingsSessionDelegate : public InspectorSessionDelegate {
@@ -70,14 +93,29 @@ class JSBindingsConnection : public AsyncWrap {
                        : AsyncWrap(env, wrap, PROVIDER_INSPECTORJSBINDING),
                          callback_(env->isolate(), callback) {
     Agent* inspector = env->inspector_agent();
-    session_ = inspector->Connect(std::make_unique<JSBindingsSessionDelegate>(
-        env, this), false);
+    session_ = ConnectionType::Connect(
+        inspector, std::make_unique<JSBindingsSessionDelegate>(env, this));
   }
 
   void OnMessage(Local<Value> value) {
     MakeCallback(callback_.Get(env()->isolate()), 1, &value);
   }
 
+  static void Bind(Environment* env, Local<Object> target) {
+    Local<String> class_name = ConnectionType::GetClassName(env);
+    Local<FunctionTemplate> tmpl =
+        env->NewFunctionTemplate(JSBindingsConnection::New);
+    tmpl->InstanceTemplate()->SetInternalFieldCount(1);
+    tmpl->SetClassName(class_name);
+    tmpl->Inherit(AsyncWrap::GetConstructorTemplate(env));
+    env->SetProtoMethod(tmpl, "dispatch", JSBindingsConnection::Dispatch);
+    env->SetProtoMethod(tmpl, "disconnect", JSBindingsConnection::Disconnect);
+    target->Set(env->context(),
+                class_name,
+                tmpl->GetFunction(env->context()).ToLocalChecked())
+        .ToChecked();
+  }
+
   static void New(const FunctionCallbackInfo<Value>& info) {
     Environment* env = Environment::GetCurrent(info);
     CHECK(info[0]->IsFunction());
@@ -300,18 +338,8 @@ void Initialize(Local<Object> target, Local<Value> unused,
   env->SetMethod(target, "registerAsyncHook", RegisterAsyncHookWrapper);
   env->SetMethodNoSideEffect(target, "isEnabled", IsEnabled);
 
-  auto conn_str = FIXED_ONE_BYTE_STRING(env->isolate(), "Connection");
-  Local<FunctionTemplate> tmpl =
-      env->NewFunctionTemplate(JSBindingsConnection::New);
-  tmpl->InstanceTemplate()->SetInternalFieldCount(1);
-  tmpl->SetClassName(conn_str);
-  tmpl->Inherit(AsyncWrap::GetConstructorTemplate(env));
-  env->SetProtoMethod(tmpl, "dispatch", JSBindingsConnection::Dispatch);
-  env->SetProtoMethod(tmpl, "disconnect", JSBindingsConnection::Disconnect);
-  target
-      ->Set(env->context(), conn_str,
-            tmpl->GetFunction(env->context()).ToLocalChecked())
-      .ToChecked();
+  JSBindingsConnection<LocalConnection>::Bind(env, target);
+  JSBindingsConnection<MainThreadConnection>::Bind(env, target);
 }
 
 }  // namespace
diff --git a/test/sequential/test-inspector-bindings.js b/test/parallel/test-inspector-bindings.js
similarity index 98%
rename from test/sequential/test-inspector-bindings.js
rename to test/parallel/test-inspector-bindings.js
index f5583c1d7bc6d6..c632bb8840fa30 100644
--- a/test/sequential/test-inspector-bindings.js
+++ b/test/parallel/test-inspector-bindings.js
@@ -1,4 +1,3 @@
-// Flags: --expose-internals
 'use strict';
 const common = require('../common');
 common.skipIfInspectorDisabled();
@@ -86,7 +85,7 @@ function testSampleDebugSession() {
   }, TypeError);
   session.post('Debugger.enable', () => cbAsSecondArgCalled = true);
   session.post('Debugger.setBreakpointByUrl', {
-    'lineNumber': 14,
+    'lineNumber': 13,
     'url': pathToFileURL(path.resolve(__dirname, __filename)).toString(),
     'columnNumber': 0,
     'condition': ''
diff --git a/test/parallel/test-inspector-connect-main-thread.js b/test/parallel/test-inspector-connect-main-thread.js
new file mode 100644
index 00000000000000..0b072ecf598c32
--- /dev/null
+++ b/test/parallel/test-inspector-connect-main-thread.js
@@ -0,0 +1,179 @@
+'use strict';
+const common = require('../common');
+
+common.skipIfInspectorDisabled();
+
+const assert = require('assert');
+const { Session } = require('inspector');
+const path = require('path');
+const { pathToFileURL } = require('url');
+const { isMainThread, parentPort, Worker, workerData } =
+    require('worker_threads');
+
+if (!workerData) {
+  common.skipIfWorker();
+}
+
+function toDebug() {
+  let a = 1;
+  a = a + 1;
+  return a * 200;
+}
+
+const messagesSent = [];
+
+async function post(session, method, params) {
+  return new Promise((resolve, reject) => {
+    session.post(method, params, (error, success) => {
+      messagesSent.push(method);
+      if (error) {
+        console.log(`Message ${method} produced an error`);
+        reject(error);
+      } else {
+        console.log(`Message ${method} was sent`);
+        resolve(success);
+      }
+    });
+  });
+}
+
+async function waitForNotification(session, notification) {
+  return new Promise((resolve) => session.once(notification, resolve));
+}
+
+function startWorker(skipChild, sharedBuffer) {
+  return new Promise((resolve) => {
+    const worker = new Worker(__filename, {
+      workerData: { skipChild, sharedBuffer }
+    });
+    worker.on('error', (e) => {
+      console.error(e);
+      throw e;
+    });
+    worker.once('message', (m) => {
+      resolve(worker);
+    });
+  });
+}
+
+function waitForConsoleRequest(worker) {
+  return new Promise((resolve) => {
+    worker.on('message', ({ doConsoleLog }) => {
+      if (doConsoleLog) {
+        resolve();
+      }
+    });
+  });
+}
+
+function waitForMessagesSent(worker) {
+  return new Promise((resolve) => {
+    worker.on('message', ({ messagesSent }) => {
+      if (messagesSent) {
+        resolve(messagesSent);
+      }
+    });
+  });
+}
+
+function doConsoleLog(arrayBuffer) {
+  console.log('Message for a test');
+  arrayBuffer[0] = 128;
+}
+
+// This tests that inspector callbacks are called in a microtask
+// and do not interrupt the main code. Interrupting the code flow
+// can lead to unexpected behaviors.
+async function ensureListenerDoesNotInterrupt(session) {
+  const currentTime = Date.now();
+  let consoleLogHappened = false;
+  session.once('Runtime.consoleAPICalled',
+               () => { consoleLogHappened = true; });
+  const buf = new Uint8Array(workerData.sharedBuffer);
+  parentPort.postMessage({ doConsoleLog: true });
+  while (buf[0] === 1) {
+    // Making sure the console.log was executed
+  }
+  while ((Date.now() - currentTime) < 50) {
+    // Spin wait for 50ms, assume that was enough to get inspector message
+  }
+  assert.strictEqual(consoleLogHappened, false);
+  await new Promise(queueMicrotask);
+  assert.strictEqual(consoleLogHappened, true);
+}
+
+async function main() {
+  const sharedBuffer = new SharedArrayBuffer(1);
+  const arrayBuffer = new Uint8Array(sharedBuffer);
+  arrayBuffer[0] = 1;
+  const worker = await startWorker(false, sharedBuffer);
+  waitForConsoleRequest(worker).then(doConsoleLog.bind(null, arrayBuffer));
+  const workerDonePromise = waitForMessagesSent(worker);
+  assert.strictEqual(toDebug(), 400);
+  assert.deepStrictEqual(await workerDonePromise, [
+    'Debugger.enable',
+    'Runtime.enable',
+    'Debugger.setBreakpointByUrl',
+    'Debugger.evaluateOnCallFrame',
+    'Debugger.resume'
+  ]);
+}
+
+async function childMain() {
+  // Ensures the worker does not terminate too soon
+  parentPort.on('message', () => { });
+  await waitForMessagesSent(await startWorker(true));
+  const session = new Session();
+  session.connectToMainThread();
+  await post(session, 'Debugger.enable');
+  await post(session, 'Runtime.enable');
+  await post(session, 'Debugger.setBreakpointByUrl', {
+    'lineNumber': 18,
+    'url': pathToFileURL(path.resolve(__dirname, __filename)).toString(),
+    'columnNumber': 0,
+    'condition': ''
+  });
+  const pausedPromise = waitForNotification(session, 'Debugger.paused');
+  parentPort.postMessage('Ready');
+  const callFrameId = (await pausedPromise).params.callFrames[0].callFrameId;
+
+  // Delay to ensure main thread is truly suspended
+  await new Promise((resolve) => setTimeout(resolve, 50));
+
+  const { result: { value } } =
+      await post(session,
+                 'Debugger.evaluateOnCallFrame',
+                 { callFrameId, expression: 'a * 100' });
+  assert.strictEqual(value, 100);
+  await post(session, 'Debugger.resume');
+  await ensureListenerDoesNotInterrupt(session);
+  parentPort.postMessage({ messagesSent });
+  parentPort.close();
+  console.log('Worker is done');
+}
+
+async function skipChildMain() {
+  // Ensures the worker does not terminate too soon
+  parentPort.on('message', () => { });
+
+  const session = new Session();
+  session.connectToMainThread();
+  const notifications = [];
+  session.on('NodeWorker.attachedToWorker', (n) => notifications.push(n));
+  await post(session, 'NodeWorker.enable', { waitForDebuggerOnStart: false });
+  // 2 notifications mean there are 2 workers so we are connected to a main
+  // thread
+  assert.strictEqual(notifications.length, 2);
+  parentPort.postMessage('Ready');
+  parentPort.postMessage({ messagesSent });
+  parentPort.close();
+  console.log('Skip child is done');
+}
+
+if (isMainThread) {
+  main().then(common.mustCall());
+} else if (workerData.skipChild) {
+  skipChildMain().then(common.mustCall());
+} else {
+  childMain().then(common.mustCall());
+}