Skip to content

child_process: fix stdin close event when child closes fd 0 #64626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mhayk wants to merge 1 commit into
base: main
Choose a base branch
from
+78 −37
Open

child_process: fix stdin close event when child closes fd 0 #64626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 49 additions & 37 deletions lib/internal/child_process.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand Down Expand Up @@ -332,8 +332,20 @@ function flushStdio(subprocess) {
}


function createSocket(pipe, readable) {
return net.Socket({ handle: pipe, readable });
function createSocket(pipe, readable, stdin = false) {
// For stdin specifically, we must call read(0) to start the read loop,
// allowing libuv to detect EOF/HUP events when the child closes fd 0.
// See: https://github.com/nodejs/node/issues/25131
if (stdin) {
// stdin: create without specifying readable to trigger read(0),
// then manually set readable=false and writable=true
const socket = net.Socket({ handle: pipe });
socket.readable = false;
socket.writable = true;
return socket;
}
// For all other sockets, use the standard behavior
return net.Socket({ handle: pipe, readable, writable: !readable });
}


Expand Down Expand Up @@ -368,7 +380,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) {
Expand All @@ -380,7 +392,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');
Expand Down Expand Up @@ -418,10 +430,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,
Expand Down Expand Up @@ -489,7 +501,7 @@ ChildProcess.prototype.spawn = function spawn(options) {

if (stream.handle) {
stream.socket = createSocket(this.pid !== 0 ?
stream.handle : null, i > 0);
stream.handle : null, i > 0, i === 0);

if (i > 0 && this.pid !== 0) {
this._closesNeeded++;
Expand All @@ -511,7 +523,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);
Expand Down Expand Up @@ -557,7 +569,7 @@ ChildProcess.prototype.kill = function kill(sig) {
return false;
};

ChildProcess.prototype[SymbolDispose] = function() {
ChildProcess.prototype[SymbolDispose] = function () {
if (!this.killed) {
this.kill();
}
Expand Down Expand Up @@ -635,7 +647,7 @@ function setupChannel(target, channel, serializationMode) {
let pendingHandle = null;
initMessageChannel(channel);
channel.pendingHandle = null;
channel.onread = function(arrayBuffer) {
channel.onread = function (arrayBuffer) {
const recvHandle = channel.pendingHandle;
channel.pendingHandle = null;
if (arrayBuffer) {
Expand Down Expand Up @@ -674,19 +686,19 @@ function setupChannel(target, channel, serializationMode) {
channel.sockets = { got: {}, send: {} };

// Handlers will go through this
target.on('internalMessage', function(message, handle) {
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');
}
}

Expand All @@ -696,9 +708,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++) {
Expand Down Expand Up @@ -740,7 +752,7 @@ function setupChannel(target, channel, serializationMode) {
});
});

target.on('newListener', function() {
target.on('newListener', function () {

process.nextTick(() => {
if (!target.channel || !target.listenerCount('message'))
Expand All @@ -758,7 +770,7 @@ function setupChannel(target, channel, serializationMode) {
});
});

target.send = function(message, handle, options, callback) {
target.send = function (message, handle, options, callback) {
if (typeof handle === 'function') {
callback = handle;
handle = undefined;
Expand All @@ -784,7 +796,7 @@ function setupChannel(target, channel, serializationMode) {
return false;
};

target._send = function(message, handle, options, callback) {
target._send = function (message, handle, options, callback) {
assert(this.connected || this.channel);

if (message === undefined)
Expand All @@ -795,9 +807,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);
}
Expand Down Expand Up @@ -858,8 +870,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,
Expand Down Expand Up @@ -922,7 +934,7 @@ function setupChannel(target, channel, serializationMode) {
// null and connected is false
target.connected = true;

target.disconnect = function() {
target.disconnect = function () {
if (!this.connected) {
this.emit('error', new ERR_IPC_DISCONNECTED());
return;
Expand All @@ -938,7 +950,7 @@ function setupChannel(target, channel, serializationMode) {
this._disconnect();
};

target._disconnect = function() {
target._disconnect = function () {
assert(this.channel);

// This marks the fact that the channel is actually disconnected.
Expand Down Expand Up @@ -996,11 +1008,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;
Expand Down Expand Up @@ -1038,7 +1050,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,
Expand Down Expand Up @@ -1078,7 +1090,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;
Expand Down
29 changes: 29 additions & 0 deletions test/parallel/test-child-process-stdin-close.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
'use strict';

const common = require('../common');
const assert = require('assert');
const { spawn } = require('child_process');

// Test that stdin close event is emitted when child process closes its stdin
const cp = spawn(
'node', [
'-e',
'fs.closeSync(0); setTimeout(() => {}, 2000)'
],
{stdio: ['pipe', 'inherit', 'inherit']}
);

let closeEventEmitted = false;

cp.stdin.on('close', common.mustCall(() => {
closeEventEmitted = true;
}));

setTimeout(() => {
assert(closeEventEmitted, 'stdin close event was not emitted');
cp.kill();
}, 1000);

cp.on('exit', common.mustCall((code, signal) => {
assert(closeEventEmitted, 'stdin close event must be emitted before child exit');
}));
Loading