worker: flush stdout and stderr on exit

Signed-off-by: Matteo Collina <hello@matteocollina.com> PR-URL: #56428 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Paolo Insogna <paolo@cowtech.it>

Author: mcollina

lib/internal/bootstrap/switches/is_not_main_thread.js

@@ -31,14 +31,24 @@ process.removeListener('removeListener', stopListeningIfSignal);
 
 const {
  createWorkerStdio,
+ kStdioWantsMoreDataCallback,
 } = require('internal/worker/io');
 
 let workerStdio;
 function lazyWorkerStdio() {
- if (!workerStdio) workerStdio = createWorkerStdio();
+ if (workerStdio === undefined) {
+ workerStdio = createWorkerStdio();
+ process.on('exit', flushSync);
+ }
+
  return workerStdio;
 }
 
+function flushSync() {
+ workerStdio.stdout[kStdioWantsMoreDataCallback]();
+ workerStdio.stderr[kStdioWantsMoreDataCallback]();
+}
+
 function getStdout() { return lazyWorkerStdio().stdout; }
 
 function getStderr() { return lazyWorkerStdio().stderr; }

lib/internal/worker/io.js

@@ -373,9 +373,13 @@ class WritableWorkerStdio extends Writable {
  chunks: ArrayPrototypeMap(chunks,
  ({ chunk, encoding }) => ({ chunk, encoding })),
  });
- ArrayPrototypePush(this[kWritableCallbacks], cb);
- if (this[kPort][kWaitingStreams]++ === 0)
- this[kPort].ref();
+ if (process._exiting) {
+ cb();
+ } else {
+ ArrayPrototypePush(this[kWritableCallbacks], cb);
+ if (this[kPort][kWaitingStreams]++ === 0)
+ this[kPort].ref();
+ }
  }
 
  _final(cb) {

test/parallel/test-worker-stdio-flush-inflight.js

@@ -0,0 +1,24 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { Worker, isMainThread } = require('worker_threads');
+
+if (isMainThread) {
+ const w = new Worker(__filename, { stdout: true });
+ const expected = 'hello world';
+
+ let data = '';
+ w.stdout.setEncoding('utf8');
+ w.stdout.on('data', (chunk) => {
+ data += chunk;
+ });
+
+ w.on('exit', common.mustCall(() => {
+ assert.strictEqual(data, expected);
+ }));
+} else {
+ process.stdout.write('hello');
+ process.stdout.write(' ');
+ process.stdout.write('world');
+ process.exit(0);
+}

test/parallel/test-worker-stdio-flush.js

@@ -0,0 +1,25 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const { Worker, isMainThread } = require('worker_threads');
+
+if (isMainThread) {
+ const w = new Worker(__filename, { stdout: true });
+ const expected = 'hello world';
+
+ let data = '';
+ w.stdout.setEncoding('utf8');
+ w.stdout.on('data', (chunk) => {
+ data += chunk;
+ });
+
+ w.on('exit', common.mustCall(() => {
+ assert.strictEqual(data, expected);
+ }));
+} else {
+ process.on('exit', () => {
+ process.stdout.write(' ');
+ process.stdout.write('world');
+ });
+ process.stdout.write('hello');
+}