child_process: create proper public API for channel
Instead of exposing the C++ bindings object as `subprocess.channel` or `process.channel`, provide the “control” object that was previously used internally as the public-facing variant of it. This should be better than returning the raw pipe object, and matches the original intention (when the `channel` property was first added) of providing a proper way to `.ref()` or `.unref()` the channel. PR-URL: https://github.com/nodejs/node/pull/30165 Refs: https://github.com/nodejs/node/pull/9322 Refs: https://github.com/nodejs/node/issues/9313 Reviewed-By: Jeremiah Senkpiel <fishrock123@rocketmail.com> Reviewed-By: Denys Otrishko <shishugi@gmail.com> Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl> Reviewed-By: James M Snell <jasnell@gmail.com>
This commit is contained in:
parent
d6a2badef7
commit
e65bed1b7e
@ -1018,6 +1018,10 @@ See [Advanced Serialization][] for more details.
|
|||||||
### `subprocess.channel`
|
### `subprocess.channel`
|
||||||
<!-- YAML
|
<!-- YAML
|
||||||
added: v7.1.0
|
added: v7.1.0
|
||||||
|
changes:
|
||||||
|
- version: REPLACEME
|
||||||
|
pr-url: https://github.com/nodejs/node/pull/30165
|
||||||
|
description: The object no longer accidentally exposes native C++ bindings.
|
||||||
-->
|
-->
|
||||||
|
|
||||||
* {Object} A pipe representing the IPC channel to the child process.
|
* {Object} A pipe representing the IPC channel to the child process.
|
||||||
@ -1025,6 +1029,22 @@ added: v7.1.0
|
|||||||
The `subprocess.channel` property is a reference to the child's IPC channel. If
|
The `subprocess.channel` property is a reference to the child's IPC channel. If
|
||||||
no IPC channel currently exists, this property is `undefined`.
|
no IPC channel currently exists, this property is `undefined`.
|
||||||
|
|
||||||
|
#### `subprocess.channel.ref()`
|
||||||
|
<!-- YAML
|
||||||
|
added: v7.1.0
|
||||||
|
-->
|
||||||
|
|
||||||
|
This method makes the IPC channel keep the event loop of the parent process
|
||||||
|
running if `.unref()` has been called before.
|
||||||
|
|
||||||
|
#### `subprocess.channel.unref()`
|
||||||
|
<!-- YAML
|
||||||
|
added: v7.1.0
|
||||||
|
-->
|
||||||
|
|
||||||
|
This method makes the IPC channel not keep the event loop of the parent process
|
||||||
|
running, and lets it finish even while the channel is open.
|
||||||
|
|
||||||
### `subprocess.connected`
|
### `subprocess.connected`
|
||||||
<!-- YAML
|
<!-- YAML
|
||||||
added: v0.7.2
|
added: v0.7.2
|
||||||
|
@ -626,6 +626,10 @@ $ bash -c 'exec -a customArgv0 ./node'
|
|||||||
## `process.channel`
|
## `process.channel`
|
||||||
<!-- YAML
|
<!-- YAML
|
||||||
added: v7.1.0
|
added: v7.1.0
|
||||||
|
changes:
|
||||||
|
- version: REPLACEME
|
||||||
|
pr-url: https://github.com/nodejs/node/pull/30165
|
||||||
|
description: The object no longer accidentally exposes native C++ bindings.
|
||||||
-->
|
-->
|
||||||
|
|
||||||
* {Object}
|
* {Object}
|
||||||
@ -635,6 +639,30 @@ If the Node.js process was spawned with an IPC channel (see the
|
|||||||
property is a reference to the IPC channel. If no IPC channel exists, this
|
property is a reference to the IPC channel. If no IPC channel exists, this
|
||||||
property is `undefined`.
|
property is `undefined`.
|
||||||
|
|
||||||
|
### `process.channel.ref()`
|
||||||
|
<!-- YAML
|
||||||
|
added: v7.1.0
|
||||||
|
-->
|
||||||
|
|
||||||
|
This method makes the IPC channel keep the event loop of the process
|
||||||
|
running if `.unref()` has been called before.
|
||||||
|
|
||||||
|
Typically, this is managed through the number of `'disconnect'` and `'message'`
|
||||||
|
listeners on the `process` object. However, this method can be used to
|
||||||
|
explicitly request a specific behavior.
|
||||||
|
|
||||||
|
### `process.channel.unref()`
|
||||||
|
<!-- YAML
|
||||||
|
added: v7.1.0
|
||||||
|
-->
|
||||||
|
|
||||||
|
This method makes the IPC channel not keep the event loop of the process
|
||||||
|
running, and lets it finish even while the channel is open.
|
||||||
|
|
||||||
|
Typically, this is managed through the number of `'disconnect'` and `'message'`
|
||||||
|
listeners on the `process` object. However, this method can be used to
|
||||||
|
explicitly request a specific behavior.
|
||||||
|
|
||||||
## `process.chdir(directory)`
|
## `process.chdir(directory)`
|
||||||
<!-- YAML
|
<!-- YAML
|
||||||
added: v0.1.17
|
added: v0.1.17
|
||||||
|
@ -122,10 +122,10 @@ function _forkChild(fd, serializationMode) {
|
|||||||
p.unref();
|
p.unref();
|
||||||
const control = setupChannel(process, p, serializationMode);
|
const control = setupChannel(process, p, serializationMode);
|
||||||
process.on('newListener', function onNewListener(name) {
|
process.on('newListener', function onNewListener(name) {
|
||||||
if (name === 'message' || name === 'disconnect') control.ref();
|
if (name === 'message' || name === 'disconnect') control.refCounted();
|
||||||
});
|
});
|
||||||
process.on('removeListener', function onRemoveListener(name) {
|
process.on('removeListener', function onRemoveListener(name) {
|
||||||
if (name === 'message' || name === 'disconnect') control.unref();
|
if (name === 'message' || name === 'disconnect') control.unrefCounted();
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -91,8 +91,9 @@ function createWritableStdioStream(fd) {
|
|||||||
// an error when trying to use it again. In that case, create the socket
|
// an error when trying to use it again. In that case, create the socket
|
||||||
// using the existing handle instead of the fd.
|
// using the existing handle instead of the fd.
|
||||||
if (process.channel && process.channel.fd === fd) {
|
if (process.channel && process.channel.fd === fd) {
|
||||||
|
const { kChannelHandle } = require('internal/child_process');
|
||||||
stream = new net.Socket({
|
stream = new net.Socket({
|
||||||
handle: process.channel,
|
handle: process[kChannelHandle],
|
||||||
readable: false,
|
readable: false,
|
||||||
writable: true
|
writable: true
|
||||||
});
|
});
|
||||||
|
@ -66,6 +66,7 @@ let freeParser;
|
|||||||
let HTTPParser;
|
let HTTPParser;
|
||||||
|
|
||||||
const MAX_HANDLE_RETRANSMISSIONS = 3;
|
const MAX_HANDLE_RETRANSMISSIONS = 3;
|
||||||
|
const kChannelHandle = Symbol('kChannelHandle');
|
||||||
const kIsUsedAsStdio = Symbol('kIsUsedAsStdio');
|
const kIsUsedAsStdio = Symbol('kIsUsedAsStdio');
|
||||||
|
|
||||||
// This object contain function to convert TCP objects to native handle objects
|
// This object contain function to convert TCP objects to native handle objects
|
||||||
@ -108,7 +109,7 @@ const handleConversion = {
|
|||||||
// The worker should keep track of the socket
|
// The worker should keep track of the socket
|
||||||
message.key = socket.server._connectionKey;
|
message.key = socket.server._connectionKey;
|
||||||
|
|
||||||
const firstTime = !this.channel.sockets.send[message.key];
|
const firstTime = !this[kChannelHandle].sockets.send[message.key];
|
||||||
const socketList = getSocketList('send', this, message.key);
|
const socketList = getSocketList('send', this, message.key);
|
||||||
|
|
||||||
// The server should no longer expose a .connection property
|
// The server should no longer expose a .connection property
|
||||||
@ -508,22 +509,45 @@ ChildProcess.prototype.unref = function() {
|
|||||||
};
|
};
|
||||||
|
|
||||||
class Control extends EventEmitter {
|
class Control extends EventEmitter {
|
||||||
|
#channel = null;
|
||||||
|
#refs = 0;
|
||||||
|
#refExplicitlySet = false;
|
||||||
|
|
||||||
constructor(channel) {
|
constructor(channel) {
|
||||||
super();
|
super();
|
||||||
this.channel = channel;
|
this.#channel = channel;
|
||||||
this.refs = 0;
|
|
||||||
}
|
}
|
||||||
ref() {
|
|
||||||
if (++this.refs === 1) {
|
// The methods keeping track of the counter are being used to track the
|
||||||
this.channel.ref();
|
// listener count on the child process object as well as when writes are
|
||||||
|
// in progress. Once the user has explicitly requested a certain state, these
|
||||||
|
// methods become no-ops in order to not interfere with the user's intentions.
|
||||||
|
refCounted() {
|
||||||
|
if (++this.#refs === 1 && !this.#refExplicitlySet) {
|
||||||
|
this.#channel.ref();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
unref() {
|
|
||||||
if (--this.refs === 0) {
|
unrefCounted() {
|
||||||
this.channel.unref();
|
if (--this.#refs === 0 && !this.#refExplicitlySet) {
|
||||||
|
this.#channel.unref();
|
||||||
this.emit('unref');
|
this.emit('unref');
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
ref() {
|
||||||
|
this.#refExplicitlySet = true;
|
||||||
|
this.#channel.ref();
|
||||||
|
}
|
||||||
|
|
||||||
|
unref() {
|
||||||
|
this.#refExplicitlySet = true;
|
||||||
|
this.#channel.unref();
|
||||||
|
}
|
||||||
|
|
||||||
|
get fd() {
|
||||||
|
return this.#channel ? this.#channel.fd : undefined;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const channelDeprecationMsg = '_channel is deprecated. ' +
|
const channelDeprecationMsg = '_channel is deprecated. ' +
|
||||||
@ -531,7 +555,9 @@ const channelDeprecationMsg = '_channel is deprecated. ' +
|
|||||||
|
|
||||||
let serialization;
|
let serialization;
|
||||||
function setupChannel(target, channel, serializationMode) {
|
function setupChannel(target, channel, serializationMode) {
|
||||||
target.channel = channel;
|
const control = new Control(channel);
|
||||||
|
target.channel = control;
|
||||||
|
target[kChannelHandle] = channel;
|
||||||
|
|
||||||
ObjectDefineProperty(target, '_channel', {
|
ObjectDefineProperty(target, '_channel', {
|
||||||
get: deprecate(() => {
|
get: deprecate(() => {
|
||||||
@ -547,8 +573,6 @@ function setupChannel(target, channel, serializationMode) {
|
|||||||
target._handleQueue = null;
|
target._handleQueue = null;
|
||||||
target._pendingMessage = null;
|
target._pendingMessage = null;
|
||||||
|
|
||||||
const control = new Control(channel);
|
|
||||||
|
|
||||||
if (serialization === undefined)
|
if (serialization === undefined)
|
||||||
serialization = require('internal/child_process/serialization');
|
serialization = require('internal/child_process/serialization');
|
||||||
const {
|
const {
|
||||||
@ -796,11 +820,11 @@ function setupChannel(target, channel, serializationMode) {
|
|||||||
|
|
||||||
if (wasAsyncWrite) {
|
if (wasAsyncWrite) {
|
||||||
req.oncomplete = () => {
|
req.oncomplete = () => {
|
||||||
control.unref();
|
control.unrefCounted();
|
||||||
if (typeof callback === 'function')
|
if (typeof callback === 'function')
|
||||||
callback(null);
|
callback(null);
|
||||||
};
|
};
|
||||||
control.ref();
|
control.refCounted();
|
||||||
} else if (typeof callback === 'function') {
|
} else if (typeof callback === 'function') {
|
||||||
process.nextTick(callback, null);
|
process.nextTick(callback, null);
|
||||||
}
|
}
|
||||||
@ -855,6 +879,7 @@ function setupChannel(target, channel, serializationMode) {
|
|||||||
|
|
||||||
// This marks the fact that the channel is actually disconnected.
|
// This marks the fact that the channel is actually disconnected.
|
||||||
this.channel = null;
|
this.channel = null;
|
||||||
|
this[kChannelHandle] = null;
|
||||||
|
|
||||||
if (this._pendingMessage)
|
if (this._pendingMessage)
|
||||||
closePendingHandle(this);
|
closePendingHandle(this);
|
||||||
@ -1011,7 +1036,7 @@ function getValidStdio(stdio, sync) {
|
|||||||
|
|
||||||
|
|
||||||
function getSocketList(type, worker, key) {
|
function getSocketList(type, worker, key) {
|
||||||
const sockets = worker.channel.sockets[type];
|
const sockets = worker[kChannelHandle].sockets[type];
|
||||||
let socketList = sockets[key];
|
let socketList = sockets[key];
|
||||||
if (!socketList) {
|
if (!socketList) {
|
||||||
const Construct = type === 'send' ? SocketListSend : SocketListReceive;
|
const Construct = type === 'send' ? SocketListSend : SocketListReceive;
|
||||||
@ -1054,6 +1079,7 @@ function spawnSync(options) {
|
|||||||
|
|
||||||
module.exports = {
|
module.exports = {
|
||||||
ChildProcess,
|
ChildProcess,
|
||||||
|
kChannelHandle,
|
||||||
setupChannel,
|
setupChannel,
|
||||||
getValidStdio,
|
getValidStdio,
|
||||||
stdioStringToArray,
|
stdioStringToArray,
|
||||||
|
@ -36,7 +36,9 @@ else
|
|||||||
function master() {
|
function master() {
|
||||||
// spawn() can only create one IPC channel so we use stdin/stdout as an
|
// spawn() can only create one IPC channel so we use stdin/stdout as an
|
||||||
// ad-hoc command channel.
|
// ad-hoc command channel.
|
||||||
const proc = spawn(process.execPath, [__filename, 'worker'], {
|
const proc = spawn(process.execPath, [
|
||||||
|
'--expose-internals', __filename, 'worker'
|
||||||
|
], {
|
||||||
stdio: ['pipe', 'pipe', 'pipe', 'ipc']
|
stdio: ['pipe', 'pipe', 'pipe', 'ipc']
|
||||||
});
|
});
|
||||||
let handle = null;
|
let handle = null;
|
||||||
@ -57,12 +59,13 @@ function master() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function worker() {
|
function worker() {
|
||||||
process.channel.readStop(); // Make messages batch up.
|
const { kChannelHandle } = require('internal/child_process');
|
||||||
|
process[kChannelHandle].readStop(); // Make messages batch up.
|
||||||
process.stdout.ref();
|
process.stdout.ref();
|
||||||
process.stdout.write('ok\r\n');
|
process.stdout.write('ok\r\n');
|
||||||
process.stdin.once('data', common.mustCall((data) => {
|
process.stdin.once('data', common.mustCall((data) => {
|
||||||
assert.strictEqual(data.toString(), 'ok\r\n');
|
assert.strictEqual(data.toString(), 'ok\r\n');
|
||||||
process.channel.readStart();
|
process[kChannelHandle].readStart();
|
||||||
}));
|
}));
|
||||||
let n = 0;
|
let n = 0;
|
||||||
process.on('message', common.mustCall((msg, handle) => {
|
process.on('message', common.mustCall((msg, handle) => {
|
||||||
|
@ -42,8 +42,7 @@ if (process.argv[2] === 'pipe') {
|
|||||||
const child = childProcess.fork(process.argv[1], ['pipe'], { silent: true });
|
const child = childProcess.fork(process.argv[1], ['pipe'], { silent: true });
|
||||||
|
|
||||||
// Allow child process to self terminate
|
// Allow child process to self terminate
|
||||||
child.channel.close();
|
child.disconnect();
|
||||||
child.channel = null;
|
|
||||||
|
|
||||||
child.on('exit', function() {
|
child.on('exit', function() {
|
||||||
process.exit(0);
|
process.exit(0);
|
||||||
|
Loading…
x
Reference in New Issue
Block a user