Refactor delete peer_close_watching_, try_catch, add kPeerCloseCallbackField

Tseian · Tseian · commit ba9e1800e59d · 2026-03-25T11:21:14.000+08:00
diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js
@@ -176,7 +176,7 @@ const handleConversion = {
           // waiting for the NODE_HANDLE_ACK of the current passing handle.
           assert(!target._pendingMessage);
           target._pendingMessage =
-              { callback, message, handle, options, retransmissions: 0 };
+            { callback, message, handle, options, retransmissions: 0 };
         } else {
           handle.close();
         }
@@ -335,8 +335,8 @@ function flushStdio(subprocess) {
 function createSocket(pipe, readable, watchPeerClose) {
   const sock = net.Socket({ handle: pipe, readable });
   if (watchPeerClose &&
-      process.platform !== 'win32' &&
-      typeof pipe?.watchPeerClose === 'function') {
+    process.platform !== 'win32' &&
+    typeof pipe?.watchPeerClose === 'function') {
     pipe.watchPeerClose(true, () => sock.destroy());
     sock.once('close', () => pipe.watchPeerClose(false));
   }
@@ -375,7 +375,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
 
   validateOneOf(options.serialization, 'options.serialization',
-                [undefined, 'json', 'advanced']);
+    [undefined, 'json', 'advanced']);
   const serialization = options.serialization || 'json';
 
   if (ipc !== undefined) {
@@ -387,7 +387,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
     ArrayPrototypePush(options.envPairs, `NODE_CHANNEL_FD=${ipcFd}`);
     ArrayPrototypePush(options.envPairs,
-                       `NODE_CHANNEL_SERIALIZATION_MODE=${serialization}`);
+      `NODE_CHANNEL_SERIALIZATION_MODE=${serialization}`);
   }
 
   validateString(options.file, 'options.file');
@@ -408,10 +408,10 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
   // Run-time errors should emit an error, not throw an exception.
   if (err === UV_EACCES ||
-      err === UV_EAGAIN ||
-      err === UV_EMFILE ||
-      err === UV_ENFILE ||
-      err === UV_ENOENT) {
+    err === UV_EAGAIN ||
+    err === UV_EMFILE ||
+    err === UV_ENFILE ||
+    err === UV_ENOENT) {
     if (childProcessSpawn.hasSubscribers) {
       childProcessSpawn.error.publish({
         process: this,
@@ -501,7 +501,7 @@ ChildProcess.prototype.spawn = function spawn(options) {
 
   for (i = 0; i < stdio.length; i++)
     ArrayPrototypePush(this.stdio,
-                       stdio[i].socket === undefined ? null : stdio[i].socket);
+      stdio[i].socket === undefined ? null : stdio[i].socket);
 
   // Add .send() method and start listening for IPC data
   if (ipc !== undefined) setupChannel(this, ipc, serialization);
@@ -667,16 +667,16 @@ function setupChannel(target, channel, serializationMode) {
   target.on('internalMessage', function(message, handle) {
     // Once acknowledged - continue sending handles.
     if (message.cmd === 'NODE_HANDLE_ACK' ||
-        message.cmd === 'NODE_HANDLE_NACK') {
+      message.cmd === 'NODE_HANDLE_NACK') {
 
       if (target._pendingMessage) {
         if (message.cmd === 'NODE_HANDLE_ACK') {
           closePendingHandle(target);
         } else if (target._pendingMessage.retransmissions++ ===
-                   MAX_HANDLE_RETRANSMISSIONS) {
+          MAX_HANDLE_RETRANSMISSIONS) {
           closePendingHandle(target);
           process.emitWarning('Handle did not reach the receiving process ' +
-                              'correctly', 'SentHandleNotReceivedWarning');
+            'correctly', 'SentHandleNotReceivedWarning');
         }
       }
 
@@ -686,9 +686,9 @@ function setupChannel(target, channel, serializationMode) {
 
       if (target._pendingMessage) {
         target._send(target._pendingMessage.message,
-                     target._pendingMessage.handle,
-                     target._pendingMessage.options,
-                     target._pendingMessage.callback);
+          target._pendingMessage.handle,
+          target._pendingMessage.options,
+          target._pendingMessage.callback);
       }
 
       for (let i = 0; i < queue.length; i++) {
@@ -785,9 +785,9 @@ function setupChannel(target, channel, serializationMode) {
     // will result in error message that is weakly consumable.
     // So perform a final check on message prior to sending.
     if (typeof message !== 'string' &&
-        typeof message !== 'object' &&
-        typeof message !== 'number' &&
-        typeof message !== 'boolean') {
+      typeof message !== 'object' &&
+      typeof message !== 'number' &&
+      typeof message !== 'boolean') {
       throw new ERR_INVALID_ARG_TYPE(
         'message', ['string', 'object', 'number', 'boolean'], message);
     }
@@ -848,8 +848,8 @@ function setupChannel(target, channel, serializationMode) {
         handle.setSimultaneousAccepts(true);
       }
     } else if (this._handleQueue &&
-               !(message && (message.cmd === 'NODE_HANDLE_ACK' ||
-                             message.cmd === 'NODE_HANDLE_NACK'))) {
+      !(message && (message.cmd === 'NODE_HANDLE_ACK' ||
+        message.cmd === 'NODE_HANDLE_NACK'))) {
       // Queue request anyway to avoid out-of-order messages.
       ArrayPrototypePush(this._handleQueue, {
         callback: callback,
@@ -986,11 +986,11 @@ function setupChannel(target, channel, serializationMode) {
 const INTERNAL_PREFIX = 'NODE_';
 function isInternal(message) {
   return (message !== null &&
-          typeof message === 'object' &&
-          typeof message.cmd === 'string' &&
-          message.cmd.length > INTERNAL_PREFIX.length &&
-          StringPrototypeSlice(message.cmd, 0, INTERNAL_PREFIX.length) ===
-            INTERNAL_PREFIX);
+    typeof message === 'object' &&
+    typeof message.cmd === 'string' &&
+    message.cmd.length > INTERNAL_PREFIX.length &&
+    StringPrototypeSlice(message.cmd, 0, INTERNAL_PREFIX.length) ===
+    INTERNAL_PREFIX);
 }
 
 const nop = FunctionPrototype;
@@ -1028,7 +1028,7 @@ function getValidStdio(stdio, sync) {
     if (stdio === 'ignore') {
       ArrayPrototypePush(acc, { type: 'ignore' });
     } else if (stdio === 'pipe' || stdio === 'overlapped' ||
-               (typeof stdio === 'number' && stdio < 0)) {
+      (typeof stdio === 'number' && stdio < 0)) {
       const a = {
         type: stdio === 'overlapped' ? 'overlapped' : 'pipe',
         readable: i === 0,
@@ -1068,7 +1068,7 @@ function getValidStdio(stdio, sync) {
         fd: typeof stdio === 'number' ? stdio : stdio.fd,
       });
     } else if (getHandleWrapType(stdio) || getHandleWrapType(stdio.handle) ||
-               getHandleWrapType(stdio._handle)) {
+      getHandleWrapType(stdio._handle)) {
       const handle = getHandleWrapType(stdio) ?
         stdio :
         getHandleWrapType(stdio.handle) ? stdio.handle : stdio._handle;
diff --git a/src/pipe_wrap.cc b/src/pipe_wrap.cc
@@ -162,11 +162,6 @@ PipeWrap::PipeWrap(Environment* env,
                    // Suggestion: uv_pipe_init() returns void.
 }
 
-PipeWrap::~PipeWrap() {
-  peer_close_watching_ = false;
-  peer_close_cb_.Reset();
-}
-
 void PipeWrap::Bind(const FunctionCallbackInfo<Value>& args) {
   PipeWrap* wrap;
   ASSIGN_OR_RETURN_UNWRAP(&wrap, args.This());
@@ -228,16 +223,20 @@ void PipeWrap::WatchPeerClose(const FunctionCallbackInfo<Value>& args) {
   CHECK_GT(args.Length(), 0);
   CHECK(args[0]->IsBoolean());
   const bool enable = args[0].As<v8::Boolean>()->Value();
+  
+  Environment* env = wrap->env();
+  Isolate* isolate = env->isolate();
+  v8::HandleScope handle_scope(isolate);
+  v8::Context::Scope context_scope(env->context());
+  Local<Object> obj = wrap->object();
 
   // UnwatchPeerClose
   if (!enable) {
-    if (!wrap->peer_close_watching_) {
-      wrap->peer_close_cb_.Reset();
+    if (obj->GetInternalField(kPeerCloseCallbackField).As<Value>()->IsUndefined()) {
       return;
     }
 
-    wrap->peer_close_watching_ = false;
-    wrap->peer_close_cb_.Reset();
+    obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate));
     uv_read_stop(wrap->stream());
     return;
   }
@@ -246,26 +245,21 @@ void PipeWrap::WatchPeerClose(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
-  if (wrap->peer_close_watching_) {
+  if (!obj->GetInternalField(kPeerCloseCallbackField).As<Value>()->IsUndefined()) {
     return;
   }
 
   CHECK_GT(args.Length(), 1);
   CHECK(args[1]->IsFunction());
 
-  Environment* env = wrap->env();
-  Isolate* isolate = env->isolate();
-
-  // Store the JS callback securely so it isn't garbage collected.
-  wrap->peer_close_cb_.Reset(isolate, args[1].As<Function>());
-  wrap->peer_close_watching_ = true;
+  // Store the JS callback in an internal field.
+  obj->SetInternalField(kPeerCloseCallbackField, args[1]);
 
   // Start reading to detect EOF/ECONNRESET from the peer.
   // We use our custom allocator and reader, ignoring actual data.
   int err = uv_read_start(wrap->stream(), PeerCloseAlloc, PeerCloseRead);
   if (err != 0) {
-    wrap->peer_close_watching_ = false;
-    wrap->peer_close_cb_.Reset();
+    obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate));
   }
 }
 
@@ -282,7 +276,7 @@ void PipeWrap::PeerCloseRead(uv_stream_t* stream,
                              ssize_t nread,
                              const uv_buf_t* buf) {
   PipeWrap* wrap = static_cast<PipeWrap*>(stream->data);
-  if (wrap == nullptr || !wrap->peer_close_watching_) return;
+  if (wrap == nullptr) return;
 
   // Ignore actual data reads or EAGAIN (0). We only watch for disconnects.
   if (nread > 0 || nread == 0) return;
@@ -291,22 +285,25 @@ void PipeWrap::PeerCloseRead(uv_stream_t* stream,
   if (nread != UV_EOF && nread != UV_ECONNRESET) return;
 
   // Peer has closed the connection. Stop reading immediately.
-  wrap->peer_close_watching_ = false;
   uv_read_stop(stream);
 
-  if (wrap->peer_close_cb_.IsEmpty()) return;
   Environment* env = wrap->env();
   Isolate* isolate = env->isolate();
 
   // Set up V8 context and handles to safely execute the JS callback.
   v8::HandleScope handle_scope(isolate);
   v8::Context::Scope context_scope(env->context());
-  Local<Function> cb = wrap->peer_close_cb_.Get(isolate);
-  // Reset before calling to prevent re-entrancy issues
-  wrap->peer_close_cb_.Reset();
+  Local<Object> obj = wrap->object();
 
-  errors::TryCatchScope try_catch(env);
-  try_catch.SetVerbose(true);
+  // Check if callback is set
+  if (obj->GetInternalField(kPeerCloseCallbackField).As<Value>()->IsUndefined()) {
+    return;
+  }
+
+  Local<Value> cb_value = obj->GetInternalField(kPeerCloseCallbackField).As<Value>();
+  Local<Function> cb = cb_value.As<Function>();
+  // Reset before calling to prevent re-entrancy issues
+  obj->SetInternalField(kPeerCloseCallbackField, v8::Undefined(isolate));
 
   // MakeCallback properly tracks AsyncHooks context and flushes microtasks.
   wrap->MakeCallback(cb, 0, nullptr);
diff --git a/src/pipe_wrap.h b/src/pipe_wrap.h
@@ -40,6 +40,11 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
     IPC
   };
 
+  enum InternalFields {
+    kPeerCloseCallbackField = LibuvStreamWrap::kInternalFieldCount,
+    kInternalFieldCount
+  };
+
   static v8::MaybeLocal<v8::Object> Instantiate(Environment* env,
                                                 AsyncWrap* parent,
                                                 SocketType type);
@@ -54,7 +59,6 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
   SET_SELF_SIZE(PipeWrap)
 
  private:
-  ~PipeWrap() override;
   PipeWrap(Environment* env,
            v8::Local<v8::Object> object,
            ProviderType provider,
@@ -78,9 +82,6 @@ class PipeWrap : public ConnectionWrap<PipeWrap, uv_pipe_t> {
       const v8::FunctionCallbackInfo<v8::Value>& args);
 #endif
   static void Fchmod(const v8::FunctionCallbackInfo<v8::Value>& args);
-
-  bool peer_close_watching_ = false;
-  v8::Global<v8::Function> peer_close_cb_;
 };
 
 
