8000 worker: enable stdio · nodejs/node@6b1a887 · GitHub
[go: up one dir, main page]
Skip to content

Commit 6b1a887

Browse files
addaleaxtargos
addaleax
authored and
targos
committed
worker: enable stdio
Provide `stdin`, `stdout` and `stderr` options for the `Worker` constructor, and make these available to the worker thread under their usual names. The default for `stdin` is an empty stream, the default for `stdout` and `stderr` is redirecting to the parent thread’s corresponding stdio streams. PR-URL: #20876 Reviewed-By: Gireesh Punathil <gpunathi@in.ibm.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com> Reviewed-By: Shingo Inoue <leko.noor@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Tiancheng "Timothy" Gu <timothygu99@gmail.com> Reviewed-By: John-David Dalton <john.david.dalton@gmail.com> Reviewed-By: Gus Caplan <me@gus.host>
1 parent c97fb91 commit 6b1a887

File tree

4 files changed

+256
-11
lines changed

4 files changed

+256
-11
lines changed

‎doc/api/worker.md‎

Lines changed: 43 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -240,7 +240,7 @@ Most Node.js APIs are available inside of it.
240240
Notable differences inside a Worker environment are:
241241

242242
- The [`process.stdin`][], [`process.stdout`][] and [`process.stderr`][]
243-
properties are set to `null`.
243+
may be redirected by the parent thread.
244244
- The [`require('worker').isMainThread`][] property is set to `false`.
245245
- The [`require('worker').parentPort`][] message port is available,
246246
- [`process.exit()`][] does not stop the whole program, just the single thread,
@@ -313,6 +313,13 @@ if (isMainThread) {
313313
described in the [HTML structured clone algorithm][], and an error will be
314314
thrown if the object cannot be cloned (e.g. because it contains
315315
`function`s).
316+
* stdin {boolean} If this is set to `true`, then `worker.stdin` will
317+
provide a writable stream whose contents will appear as `process.stdin`
318+
inside the Worker. By default, no data is provided.
319+
* stdout {boolean} If this is set to `true`, then `worker.stdout` will
320+
not automatically be piped through to `process.stdout` in the parent.
321+
* stderr {boolean} If this is set to `true`, then `worker.stderr` will
322+
not automatically be piped through to `process.stderr` in the parent.
316323

317324
### Event: 'error'
318325
<!-- YAML
@@ -377,6 +384,41 @@ Opposite of `unref()`, calling `ref()` on a previously `unref()`ed worker will
377384
behavior). If the worker is `ref()`ed, calling `ref()` again will have
378385
no effect.
379386

387+
### worker.stderr
388+
<!-- YAML
389+
added: REPLACEME
390+
-->
391+
392+
* {stream.Readable}
393+
394+
This is a readable stream which contains data written to [`process.stderr`][]
395+
inside the worker thread. If `stderr: true` was not passed to the
396+
[`Worker`][] constructor, then data will be piped to the parent thread's
397+
[`process.stderr`][] stream.
398+
399+
### worker.stdin
400+
<!-- YAML
401+
added: REPLACEME
402+
-->
403+
404+
* {null|stream.Writable}
405+
406+
If `stdin: true` was passed to the [`Worker`][] constructor, this is a
407+
writable stream. The data written to this stream will be made available in
408+
the worker thread as [`process.stdin`][].
409+
410+
### worker.stdout
411+
<!-- YAML
412+
added: REPLACEME
413+
-->
414+
415+
* {stream.Readable}
416+
417+
This is a readable stream which contains data written to [`process.stdout`][]
418+
inside the worker thread. If `stdout: true` was not passed to the
419+
[`Worker`][] constructor, then data will be piped to the parent thread's
420+
[`process.stdout`][] stream.
421+
380422
### worker.terminate([callback])
381423
<!-- YAML
382424
added: REPLACEME

‎lib/internal/process/stdio.js‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,10 @@ const {
66
ERR_UNKNOWN_STDIN_TYPE,
77
ERR_UNKNOWN_STREAM_TYPE
88
} = require('internal/errors').codes;
9-
const { isMainThread } = require('internal/worker');
9+
const {
10+
isMainThread,
11+
workerStdio
12+
} = require('internal/worker');
1013

1114
exports.setup = setupStdio;
1215

@@ -17,8 +20,7 @@ function setupStdio() {
1720

1821
function getStdout() {
1922
if (stdout) return stdout;
20-
if (!isMainThread)
21-
return new (require('stream').Writable)({ write(b, e, cb) { cb(); } });
23+
if (!isMainThread) return workerStdio.stdout;
2224
stdout = createWritableStdioStream(1);
2325
stdout.destroySoon = stdout.destroy;
2426
stdout._destroy = function(er, cb) {
@@ -34,8 +36,7 @@ function setupStdio() {
3436

3537
function getStderr() {
3638
if (stderr) return stderr;
37-
if (!isMainThread)
38-
return new (require('stream').Writable)({ write(b, e, cb) { cb(); } });
39+
if (!isMainThread) return workerStdio.stderr;
3940
stderr = createWritableStdioStream(2);
4041
stderr.destroySoon = stderr.destroy;
4142
stderr._destroy = function(er, cb) {
@@ -51,8 +52,7 @@ function setupStdio() {
5152

5253
function getStdin() {
5354
if (stdin) return stdin;
54-
if (!isMainThread)
55-
return new (require('stream').Readable)({ read() { this.push(null); } });
55+
if (!isMainThread) return workerStdio.stdin;
5656

5757
const tty_wrap = process.binding('tty_wrap');
5858
const fd = 0;

‎lib/internal/worker.js‎

Lines changed: 163 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ const EventEmitter = require('events');
55
const assert = require('assert');
66
const path = require('path');
77
const util = require('util');
8+
const { Readable, Writable } = require('stream');
89
const {
910
ERR_INVALID_ARG_TYPE,
1011
ERR_WORKER_NEED_ABSOLUTE_PATH,
@@ -29,13 +30,20 @@ const isMainThread = threadId === 0;
2930

3031
const kOnMessageListener = Symbol('kOnMessageListener');
3132
const kHandle = Symbol('kHandle');
33+
const kName = Symbol('kName');
3234
const kPort = Symbol('kPort');
3335
const kPublicPort = Symbol('kPublicPort');
3436
const kDispose = Symbol('kDispose');
3537
const kOnExit = Symbol('kOnExit');
3638
const kOnMessage = Symbol('kOnMessage');
3739
const kOnCouldNotSerializeErr = Symbol('kOnCouldNotSerializeErr');
3840
const kOnErrorMessage = Symbol('kOnErrorMessage');
41+
const kParentSideStdio = Symbol('kParentSideStdio');
42+
const kWritableCallbacks = Symbol('kWritableCallbacks');
43+
const kStdioWantsMoreDataCallback = Symbol('kStdioWantsMoreDataCallback');
44+
const kStartedReading = Symbol('kStartedReading');
45+
const kWaitingStreams = Symbol('kWaitingStreams');
46+
const kIncrementsPortRef = Symbol('kIncrementsPortRef');
3947

4048
const debug = util.debuglog('worker');
4149

@@ -129,6 +137,72 @@ function setupPortReferencing(port, eventEmitter, eventName) {
129137
}
130138

131139

140+
class ReadableWorkerStdio extends Readable {
141+
constructor(port, name) {
142+
super();
143+
this[kPort] = port;
144+
this[kName] = name;
145+
this[kIncrementsPortRef] = true;
146+
this[kStartedReading] = false;
147+
this.on('end', () => {
148+
if (this[kIncrementsPortRef] && --this[kPort][kWaitingStreams] === 0)
149+
this[kPort].unref();
150+
});
151+
}
152+
153+
_read() {
154+
if (!this[kStartedReading] && this[kIncrementsPortRef]) {
155+
this[kStartedReading] = true;
156+
if (this[kPort][kWaitingStreams]++ === 0)
157+
this[kPort].ref();
158+
}
159+
160+
this[kPort].postMessage({
161+
type: 'stdioWantsMoreData',
162+
stream: this[kName]
163+
});
164+
}
165+
}
166+
167+
class WritableWorkerStdio extends Writable {
168+
constructor(port, name) {
169+
super({ decodeStrings: false });
170+
this[kPort] = port;
171+
this[kName] = name;
172+
this[kWritableCallbacks] = [];
173+
}
174+
175+
_write(chunk, encoding, cb) {
176+
this[kPort].postMessage({
177+
type: 'stdioPayload',
178+
stream: this[kName],
179+
chunk,
180+
encoding
181+
});
182+
this[kWritableCallbacks].push(cb);
183+
if (this[kPort][kWaitingStreams]++ === 0)
184+
this[kPort].ref();
185+
}
186+
187+
_final(cb) {
188+
this[kPort].postMessage({
189+
type: 'stdioPayload',
190+
stream: this[kName],
191+
chunk: null
192+
});
193+
cb();
194+
}
195+
196+
[kStdioWantsMoreDataCallback]() {
197+
const cbs = this[kWritableCallbacks];
198+
this[kWritableCallbacks] = [];
199+
for (const cb of cbs)
200+
cb();
201+
if ((this[kPort][kWaitingStreams] -= cbs.length) === 0)
202+
this[kPort].unref();
203+
}
204+
}
205+
132206
class Worker extends EventEmitter {
133207
constructor(filename, options = {}) {
134208
super();
@@ -154,8 +228,25 @@ class Worker extends EventEmitter {
154228
this[kPort].on('message', (data) => this[kOnMessage](data));
155229
this[kPort].start();
156230
this[kPort].unref();
231+
this[kPort][kWaitingStreams] = 0;
157232
debug(`[${threadId}] created Worker with ID ${this.threadId}`);
158233

234+
let stdin = null;
235+
if (options.stdin)
236+
stdin = new WritableWorkerStdio(this[kPort], 'stdin');
237+
const stdout = new ReadableWorkerStdio(this[kPort], 'stdout');
238+
if (!options.stdout) {
239+
stdout[kIncrementsPortRef] = false;
240+
pipeWithoutWarning(stdout, process.stdout);
241+
}
242+
const stderr = new ReadableWorkerStdio(this[kPort], 'stderr');
243+
if (!options.stderr) {
244+
stderr[kIncrementsPortRef] = false;
245+
pipeWithoutWarning(stderr, process.stderr);
246+
}
247+
248+
this[kParentSideStdio] = { stdin, stdout, stderr };
249+
159250
const { port1, port2 } = new MessageChannel();
160251
this[kPublicPort] = port1;
161252
this[kPublicPort].on('message', (message) => this.emit('message', message));
@@ -165,7 +256,8 @@ class Worker extends EventEmitter {
165256
filename,
166257
doEval: !!options.eval,
167258
workerData: options.workerData,
168-
publicPort: port2
259+
publicPort: port2,
260+
hasStdin: !!options.stdin
169261
}, [port2]);
170262
// Actually start the new thread now that everything is in place.
171263
this[kHandle].startThread();
@@ -197,6 +289,16 @@ class Worker extends EventEmitter {
197289
return this[kOnCouldNotSerializeErr]();
198290
case 'errorMessage':
199291
return this[kOnErrorMessage](message.error);
292+
case 'stdioPayload':
293+
{
294+
const { stream, chunk, encoding } = message;
295+
return this[kParentSideStdio][stream].push(chunk, encoding);
296+
}
297+
case 'stdioWantsMoreData':
298+
{
299+
const { stream } = message;
300+
return this[kParentSideStdio][stream][kStdioWantsMoreDataCallback]();
301+
}
200302
}
201303

202304
assert.fail(`Unknown worker message type ${message.type}`);
@@ -207,6 +309,18 @@ class Worker extends EventEmitter {
207309
this[kHandle] = null;
208310
this[kPort] = null;
209311
this[kPublicPort] = null;
312+
313+
const { stdout, stderr } = this[kParentSideStdio];
314+
this[kParentSideStdio] = null;
315+
316+
if (!stdout._readableState.ended) {
317+
debug(`[${threadId}] explicitly closes stdout for ${this.threadId}`);
318+
stdout.push(null);
319+
}
320+
if (!stderr._readableState.ended) {
321+
debug(`[${threadId}] explicitly closes stderr for ${this.threadId}`);
322+
stderr.push(null);
323+
}
210324
}
211325

212326
postMessage(...args) {
@@ -243,6 +357,27 @@ class Worker extends EventEmitter {
243357

244358
return this[kHandle].threadId;
245359
}
360+
361+
get stdin() {
362+
return this[kParentSideStdio].stdin;
363+
}
364+
365+
get stdout() {
366+
return this[kParentSideStdio].stdout;
367+
}
368+
369+
get stderr() {
370+
return this[kParentSideStdio].stderr;
371+
}
372+
}
373+
374+
const workerStdio = {};
375+
if (!isMainThread) {
376+
const port = getEnvMessagePort();
377+
port[kWaitingStreams] = 0;
378+
workerStdio.stdin = new ReadableWorkerStdio(port, 'stdin');
379+
workerStdio.stdout = new WritableWorkerStdio(port, 'stdout');
380+
workerStdio.stderr = new WritableWorkerStdio(port, 'stderr');
246381
}
247382

248383
let originalFatalException;
@@ -256,10 +391,14 @@ function setupChild(evalScript) {
256391

257392
port.on('message', (message) => {
258393
if (message.type === 'loadScript') {
259-
const { filename, doEval, workerData, publicPort } = message;
394+
const { filename, doEval, workerData, publicPort, hasStdin } = message;
260395
publicWorker.parentPort = publicPort;
261396
setupPortReferencing(publicPort, publicPort, 'message');
262397
publicWorker.workerData = workerData;
398+
399+
if (!hasStdin)
400+
workerStdio.stdin.push(null);
401+
263402
debug(`[${threadId}] starts worker script ${filename} ` +
264403
`(eval = ${eval}) at cwd = ${process.cwd()}`);
265404
port.unref();
@@ -271,6 +410,14 @@ function setupChild(evalScript) {
271410
require('module').runMain();
272411
}
273412
return;
413+
} else if (message.type === 'stdioPayload') {
414+
const { stream, chunk, encoding } = message;
415+
workerStdio[stream].push(chunk, encoding);
416+
return;
417+
} else if (message.type === 'stdioWantsMoreData') {
418+
const { stream } = message;
419+
workerStdio[stream][kStdioWantsMoreDataCallback]();
420+
return;
274421
}
275422

276423
assert.fail(`Unknown worker message type ${message.type}`);
@@ -317,11 +464,24 @@ function deserializeError(error) {
317464
error.byteLength).toString('utf8');
318465
}
319466

467+
function pipeWithoutWarning(source, dest) {
468+
const sourceMaxListeners = source._maxListeners;
469+
const destMaxListeners = dest._maxListeners;
470+
source.setMaxListeners(Infinity);
471+
dest.setMaxListeners(Infinity);
472+
473+
source.pipe(dest);
474+
475+
source._maxListeners = sourceMaxListeners;
476+
dest._maxListeners = destMaxListeners;
477+
}
478+
320479
module.exports = {
321480
MessagePort,
322481
MessageChannel,
323482
threadId,
324483
Worker,
325484
setupChild,
326-
isMainThread
485+
isMainThread,
486+
workerStdio
327487
};

0 commit comments

Comments
 (0)
0