fetch network resource on worker thread

Author: islandryu

lib/internal/inspector/fall_through_handle.js

@@ -0,0 +1,29 @@
+'use strict';
+const { addFallThroughListener } = internalBinding('inspector');
+const { InternalWorker } = require('internal/worker');
+
+function spawnLoadNetworkResourceWorker() {
+  const worker = new InternalWorker('internal/inspector/load_network_resource_worker', {
+    stderr: false,
+    stdin: false,
+    stdout: false,
+    trackUnmanagedFds: false,
+  });
+
+  worker.unref();
+
+  addFallThroughListener((sessionId, callId, method, message) => {
+    if (method === 'Network.loadNetworkResource') {
+      worker.postMessage({
+        sessionId,
+        callId,
+        method,
+        message,
+      });
+    }
+  });
+}
+
+module.exports = {
+  spawnLoadNetworkResourceWorker,
+};

lib/internal/inspector/load_network_resource_worker.js

@@ -0,0 +1,54 @@
+'use strict';
+const {
+  JSONParse,
+  JSONStringify,
+} = primordials;
+const {
+  emitProtocolResponseInParent,
+  addIoData,
+} = internalBinding('inspector');
+const { AbortController } = require('internal/abort_controller');
+const { setTimeout, clearTimeout } = require('timers');
+const { fetch } = require('internal/deps/undici/undici');
+
+module.exports = function setupInspectorWorker(_, port) {
+  function emitResponse(callId, sessionId, success, streamId = null) {
+    const result = {
+      id: callId,
+      result: {
+        resource: {
+          success,
+        },
+      },
+    };
+    if (success && streamId !== null) {
+      result.result.resource.stream = streamId.toString();
+    }
+    emitProtocolResponseInParent(callId, JSONStringify(result), sessionId);
+  }
+
+  port.on('message', (msg) => {
+    const { sessionId, callId, message } = msg;
+    const url = JSONParse(message).params.url;
+
+    const controller = new AbortController();
+    const timeoutId = setTimeout(() => controller.abort(), 2000);
+
+    fetch(url, { signal: controller.signal })
+      .then((response) => {
+        clearTimeout(timeoutId);
+        if (!response.ok) {
+          emitResponse(callId, sessionId, false);
+          return null;
+        }
+        return response.text().then((text) => {
+          const streamId = addIoData(text);
+          emitResponse(callId, sessionId, true, streamId);
+        });
+      })
+      .catch(() => {
+        clearTimeout(timeoutId);
+        emitResponse(callId, sessionId, false);
+      });
+  });
+};

lib/internal/modules/run_main.js

@@ -136,6 +136,10 @@ function runEntryPointWithESMLoader(callback) {
  * @param {string} main - First positional CLI argument, such as `'entry.js'` from `node entry.js`
  */
 function executeUserEntryPoint(main = process.argv[1]) {
+  if (getOptionValue('--experimental-inspector-network-resource')) {
+    require('internal/inspector/fall_through_handle').spawnLoadNetworkResourceWorker();
+  }
+
   let useESMLoader;
   let resolvedMain;
   if (getOptionValue('--entry-url')) {

src/inspector/io_agent.cc

@@ -1,5 +1,7 @@
 #include "io_agent.h"
+#include <string>
 #include "crdtp/dispatch.h"
+#include "node_mutex.h"
 
 namespace node::inspector::protocol {
 
@@ -8,17 +10,47 @@ void IoAgent::Wire(UberDispatcher* dispatcher) {
   IO::Dispatcher::wire(dispatcher, this);
 }
 
+std::unordered_map<int, int> IoAgent::offset_map_;
+std::unordered_map<int, std::string> IoAgent::data_map_;
+std::atomic<int> IoAgent::stream_counter_{1};
+Mutex IoAgent::data_mutex_;
+
+int IoAgent::setData(const std::string& value) {
+  int key = getNextStreamId();
+  Mutex::ScopedLock lock(data_mutex_);
+  data_map_[key] = value;
+
+  return key;
+}
+
+int IoAgent::getNextStreamId() {
+  return stream_counter_++;
+}
+
 DispatchResponse IoAgent::read(const String& in_handle,
                                Maybe<int> in_offset,
                                Maybe<int> in_size,
                                String* out_data,
                                bool* out_eof) {
-  std::string txt = data_map_[in_handle];
+  Mutex::ScopedReadLock lock(data_mutex_);
+  std::string in_handle_str = in_handle;
+  int stream_id = 0;
+  bool is_number = std::all_of(in_handle_str.begin(),
+                                in_handle_str.end(),
+                                ::isdigit);
+  if (is_number) {
+    stream_id = std::stoi(in_handle_str);
+  } else {
+    out_data = new String("");
+    *out_eof = true;
+  }
+
+  std::string txt = data_map_[stream_id];
   int offset = 0;
   if (in_offset.isJust()) {
     offset = in_offset.fromJust();
-  } else if (offset_map_.find(in_handle) != offset_map_.end()) {
-    offset = offset_map_[in_handle];
+  } else if (offset_map_.find(stream_id) != offset_map_.end()) {
+    offset = offset_map_[stream_id];
   }
   int size = 1 << 20;
   if (in_size.isJust()) {
@@ -28,18 +60,28 @@ DispatchResponse IoAgent::read(const String& in_handle,
   if (static_cast<std::size_t>(offset) < txt.length()) {
     std::string out_txt = txt.substr(offset, size);
     out_data->assign(out_txt);
+    *out_eof = false;
   } else {
     *out_eof = true;
   }
 
-  offset_map_[in_handle] = offset + size;
+  offset_map_[stream_id] = offset + size;
 
   return DispatchResponse::Success();
 }
 
 DispatchResponse IoAgent::close(const String& in_handle) {
-  offset_map_.erase(in_handle);
-  data_map_.erase(in_handle);
+  Mutex::ScopedWriteLock lock(data_mutex_);
+  std::string in_handle_str = in_handle;
+  int stream_id = 0;
+  bool is_number = std::all_of(in_handle_str.begin(),
+                               in_handle_str.end(),
+                               ::isdigit);
+  if (is_number) {
+    stream_id = std::stoi(in_handle_str);
+    offset_map_.erase(stream_id);
+    data_map_.erase(stream_id);
+  }
   return DispatchResponse::Success();
 }
 }  // namespace node::inspector::protocol

src/inspector/io_agent.h

@@ -3,6 +3,7 @@
 
 #include <unordered_map>
 #include "node/inspector/protocol/IO.h"
+#include "node_mutex.h"
 
 namespace node::inspector::protocol {
 
@@ -17,15 +18,16 @@ class IoAgent : public IO::Backend {
                         bool* out_eof) override;
   DispatchResponse close(const String& in_handle) override;
 
-  void setData(const std::string& key, const std::string value) {
-    data_map_[key] = value;
-  }
+  static int setData(const std::string& value);
 
  private:
   std::shared_ptr<IO::Frontend> frontend_;
-  std::unordered_map<std::string, int> offset_map_;
-  std::unordered_map<std::string, std::string> data_map_;
+  static int getNextStreamId();
+
+  static std::unordered_map<int, std::string> data_map_;
+  static std::unordered_map<int, int> offset_map_;
+  static std::atomic<int> stream_counter_;
+  static Mutex data_mutex_;
 };
 }  // namespace node::inspector::protocol
-
 #endif  // SRC_INSPECTOR_IO_AGENT_H_

src/inspector/network_agent.cc

@@ -183,12 +183,10 @@ std::unique_ptr<protocol::Network::Response> createResponseFromObject(
 
 NetworkAgent::NetworkAgent(NetworkInspector* inspector,
                            v8_inspector::V8Inspector* v8_inspector,
-                           Environment* env,
-                           std::shared_ptr<protocol::IoAgent> io_agent)
+                           Environment* env)
     : inspector_(inspector),
       v8_inspector_(v8_inspector),
-      env_(env),
-      io_agent_(io_agent) {
+      env_(env) {
   event_notifier_map_["requestWillBeSent"] = &NetworkAgent::requestWillBeSent;
   event_notifier_map_["responseReceived"] = &NetworkAgent::responseReceived;
   event_notifier_map_["loadingFailed"] = &NetworkAgent::loadingFailed;
@@ -221,144 +219,11 @@ protocol::DispatchResponse NetworkAgent::disable() {
   return protocol::DispatchResponse::Success();
 }
 
-std::tuple<int, std::string, std::string> NetworkAgent::spawnFetchProcess(
-    std::string_view code, Environment* env, std::string_view url) {
-  std::string stdout_result;
-  std::string stderr_result;
-  uv_loop_t* loop = new uv_loop_t;
-  uv_loop_init(loop);
-  uv_process_t child;
-  uv_pipe_t stdout_pipe;
-  uv_pipe_init(loop, &stdout_pipe, 0);
-  uv_pipe_t stderr_pipe;
-  uv_pipe_init(loop, &stderr_pipe, 0);
-
-  uv_process_options_t uv_process_options;
-  std::string command =
-      env->exec_path() + " --eval \"" + code.data() + "\" -- " + url.data();
-
-  const char* file = env->exec_path().c_str();
-  char* args[] = {const_cast<char*>(file),
-                  const_cast<char*>("--eval"),
-                  reinterpret_cast<char*>(const_cast<char*>(code.data())),
-                  reinterpret_cast<char*>(const_cast<char*>(url.data())),
-                  nullptr};
-
-  uv_stdio_container_t stdio[3];
-  uv_process_options.file = file;
-  uv_process_options.args = args;
-  uv_process_options.flags = 0;
-  uv_process_options.stdio_count = 3;
-  uv_process_options.stdio = stdio;
-  uv_process_options.cwd = nullptr;
-  uv_process_options.env = nullptr;
-
-  uv_process_options.exit_cb =
-      [](uv_process_t* req, int64_t exit_status, int term_signal) {
-        uv_close(reinterpret_cast<uv_handle_t*>(req), nullptr);
-      };
-
-  stdio[0].flags = UV_INHERIT_FD;
-  stdio[0].data.fd = 0;
-  stdio[1].flags =
-      static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
-  stdio[1].data.stream = reinterpret_cast<uv_stream_t*>(&stdout_pipe);
-  stdio[2].flags =
-      static_cast<uv_stdio_flags>(UV_CREATE_PIPE | UV_WRITABLE_PIPE);
-  stdio[2].data.stream = reinterpret_cast<uv_stream_t*>(&stderr_pipe);
-
-  int r = uv_spawn(loop, &child, &uv_process_options);
-
-  if (r != 0) {
-    uv_loop_close(loop);
-    delete loop;
-    return {r, stdout_result, stderr_result};
-  }
-
-  auto alloc_cb =
-      [](uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
-        buf->base = static_cast<char*>(malloc(suggested_size));
-        buf->len = suggested_size;
-      };
-
-  auto read_cb = [](uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) {
-    auto* response = static_cast<std::string*>(stream->data);
-    if (nread > 0) {
-      response->append(buf->base, nread);
-    } else if (nread < 0) {
-      if (!response->empty() && response->back() == '\n') {
-        response->pop_back();
-      }
-      uv_close(reinterpret_cast<uv_handle_t*>(stream), nullptr);
-    }
-    if (buf->base) free(buf->base);
-  };
-
-  stdout_pipe.data = &stdout_result;
-  uv_read_start(
-      reinterpret_cast<uv_stream_t*>(&stdout_pipe), alloc_cb, read_cb);
-
-  stderr_pipe.data = &stderr_result;
-  uv_read_start(
-      reinterpret_cast<uv_stream_t*>(&stderr_pipe), alloc_cb, read_cb);
-
-  uv_run(loop, UV_RUN_DEFAULT);
-
-  uv_walk(
-      loop,
-      [](uv_handle_t* handle, void*) {
-        if (!uv_is_closing(handle)) {
-          uv_close(handle, nullptr);
-        }
-      },
-      nullptr);
-
-  uv_run(loop, UV_RUN_DEFAULT);
-
-  uv_loop_close(loop);
-  delete loop;
-  return {r, stdout_result, stderr_result};
-}
-
 protocol::DispatchResponse NetworkAgent::loadNetworkResource(
     const protocol::String& in_url,
     std::unique_ptr<protocol::Network::LoadNetworkResourcePageResult>*
         out_resource) {
-  if (!env_->options()->experimental_inspector_network_resource) {
-    return protocol::DispatchResponse::MethodNotFound(
-        "Network.loadNetworkResource is not supported in this environment. "
-        "Please enable the experimental-inspector-network-resource option.");
-  }
-  DCHECK(io_agent_);
-
-  std::string code = R"(
-      fetch(process.argv[1], {signal: AbortSignal.timeout(2000) }).then(res => {
-        if (res.ok) {
-          res.text().then(console.log)
-        } else {
-          throw new Error('Network error: ' + res.status);
-        }
-      })
-    )";
-
-  auto [r, response, err] = spawnFetchProcess(code, env_, in_url);
-  if (r == 0 && err.empty()) {
-    std::string uuid = std::to_string(load_id_counter_);
-    load_id_counter_++;
-    io_agent_->setData(uuid, response);
-    auto result = protocol::Network::LoadNetworkResourcePageResult::create()
-                      .setSuccess(true)
-                      .setStream(uuid)
-                      .build();
-    out_resource->reset(result.release());
-  } else {
-    auto result = protocol::Network::LoadNetworkResourcePageResult::create()
-                      .setSuccess(false)
-                      .build();
-    out_resource->reset(result.release());
-  }
-
-  return protocol::DispatchResponse::Success();
+  return protocol::DispatchResponse::FallThrough();
 }
 
 void NetworkAgent::requestWillBeSent(v8::Local<v8::Context> context,

src/inspector/network_agent.h

@@ -16,8 +16,7 @@ class NetworkAgent : public protocol::Network::Backend {
  public:
   explicit NetworkAgent(NetworkInspector* inspector,
                         v8_inspector::V8Inspector* v8_inspector,
-                        Environment* env,
-                        std::shared_ptr<protocol::IoAgent> io_agent);
+                        Environment* env);
 
   void Wire(protocol::UberDispatcher* dispatcher);
 
@@ -56,8 +55,6 @@ class NetworkAgent : public protocol::Network::Backend {
   using EventNotifier = void (NetworkAgent::*)(v8::Local<v8::Context> context,
                                                v8::Local<v8::Object>);
   std::unordered_map<protocol::String, EventNotifier> event_notifier_map_;
-  std::shared_ptr<protocol::IoAgent> io_agent_;
-  int load_id_counter_ = 1;
 };
 
 }  // namespace inspector