worker: flush stdout and stderr on exit

Signed-off-by: Matteo Collina <[email protected]>
PR-URL: #56428
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Paolo Insogna <[email protected]>

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');
+}