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

@@ -184,12 +184,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;
@@ -349,41 +347,7 @@ 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

@@ -24,8 +24,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);
 

src/inspector/network_inspector.cc

@@ -4,11 +4,10 @@ namespace node {
 namespace inspector {
 
 NetworkInspector::NetworkInspector(Environment* env,
-                                   v8_inspector::V8Inspector* v8_inspector,
-                                   std::shared_ptr<protocol::IoAgent> io_agent)
+                                   v8_inspector::V8Inspector* v8_inspector)
     : enabled_(false), env_(env) {
   network_agent_ =
-      std::make_unique<NetworkAgent>(this, v8_inspector, env, io_agent);
+      std::make_unique<NetworkAgent>(this, v8_inspector, env);
 }
 NetworkInspector::~NetworkInspector() {
   network_agent_.reset();

src/inspector/network_inspector.h

@@ -13,8 +13,7 @@ class NetworkInspector {
  public:
   explicit NetworkInspector(
       Environment* env,
-      v8_inspector::V8Inspector* v8_inspector,
-      std::shared_ptr<protocol::IoAgent> io_agent = nullptr);
+      v8_inspector::V8Inspector* v8_inspector);
   ~NetworkInspector();
 
   void Wire(protocol::UberDispatcher* dispatcher);

src/inspector/node_protocol.pdl

@@ -188,7 +188,6 @@ experimental domain Network
   # Enables network tracking, network events will now be delivered to the client.
   command enable
 
-<<<<<<< HEAD
   # Enables streaming of the response for the given requestId.
   # If enabled, the dataReceived event contains the data that was received during streaming.
   experimental command streamResourceContent
@@ -198,15 +197,13 @@ experimental domain Network
     returns
       # Data that has been buffered until streaming is enabled.
       binary bufferedData
-=======
   # Fetches the resource and returns the content.
   command loadNetworkResource
     parameters
       # URL of the resource to get content for.
       string url
     returns
       LoadNetworkResourcePageResult resource
->>>>>>> ba6277dcf1 (inspector: initial support for Network.loadNetworkResource)
 
   # Fired when page is about to send HTTP request.
   event requestWillBeSent