From 0f485172ce3b1ab858ceb8afb272e2bef6d594c3 Mon Sep 17 00:00:00 2001 From: Erick Wendel Date: Fri, 17 Dec 2021 12:04:29 -0300 Subject: [PATCH] lib: child process queue pending messages It fixes the problem for the child process not receiving messages. Fixes: https://github.com/nodejs/node/issues/41134 --- lib/internal/child_process.js | 25 ++++++++++++++++++- .../test-esm-child-process-fork-main.mjs | 20 +++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) create mode 100644 test/es-module/test-esm-child-process-fork-main.mjs diff --git a/lib/internal/child_process.js b/lib/internal/child_process.js index 40592146ada6de..1ce3dc776dfbe8 100644 --- a/lib/internal/child_process.js +++ b/lib/internal/child_process.js @@ -10,6 +10,7 @@ const { ObjectDefineProperty, ObjectSetPrototypeOf, ReflectApply, + SafeSet, StringPrototypeSlice, Symbol, Uint8Array, @@ -81,6 +82,7 @@ let HTTPParser; const MAX_HANDLE_RETRANSMISSIONS = 3; const kChannelHandle = Symbol('kChannelHandle'); const kIsUsedAsStdio = Symbol('kIsUsedAsStdio'); +const kPendingMessages = Symbol('pendingMessages'); // This object contain function to convert TCP objects to native handle objects // and back again. @@ -526,6 +528,7 @@ class Control extends EventEmitter { constructor(channel) { super(); this.#channel = channel; + this[kPendingMessages] = new SafeSet(); } // The methods keeping track of the counter are being used to track the @@ -699,6 +702,19 @@ function setupChannel(target, channel, serializationMode) { }); }); + target.on('newListener', function() { + if (!target.channel) return; + + const messages = target.channel[kPendingMessages]; + if (!messages.size) return; + + for (const messageParams of messages) { + process.nextTick(() => ReflectApply(target.emit, target, messageParams)); + } + + messages.clear(); + }); + target.send = function(message, handle, options, callback) { if (typeof handle === 'function') { callback = handle; @@ -912,7 +928,14 @@ function setupChannel(target, channel, serializationMode) { }; function emit(event, message, handle) { - target.emit(event, message, handle); + const isInternalMessage = 'internalMessage' === event; + const hasListenersInstalled = target.listenerCount('message'); + if (hasListenersInstalled || isInternalMessage) { + target.emit(event, message, handle); + return; + } + + target.channel[kPendingMessages].add([event, message, handle]); } function handleMessage(message, handle, internal) { diff --git a/test/es-module/test-esm-child-process-fork-main.mjs b/test/es-module/test-esm-child-process-fork-main.mjs new file mode 100644 index 00000000000000..c7961086494186 --- /dev/null +++ b/test/es-module/test-esm-child-process-fork-main.mjs @@ -0,0 +1,20 @@ +import '../common/index.mjs'; +import assert from 'assert'; +import { fork } from 'child_process'; +import { once } from 'events'; + +if (process.argv[2] !== 'child') { + const currentFile = 'test-esm-child-process-fork-main.mjs'; + const { pathname: filename } = new URL(currentFile, import.meta.url); + const cp = fork(filename, ['child']); + const message = 'Hello World'; + cp.send(message); + + const [received] = await once(cp, 'message'); + + assert.deepStrictEqual(received, message); + + cp.disconnect(); +} else { + process.on('message', (msg) => process.send(msg)); +}