worker: refactor stdio to improve performance

Signed-off-by: Matteo Collina <[email protected]>
PR-URL: #56630
Reviewed-By: Yagiz Nizipli <[email protected]>
Reviewed-By: Robert Nagy <[email protected]>
Reviewed-By: Paolo Insogna <[email protected]>
Reviewed-By: Luigi Pinca <[email protected]>

Author: mcollina

lib/internal/worker.js

@@ -340,9 +340,11 @@ class Worker extends EventEmitter {
       {
         const { stream, chunks } = message;
         const readable = this[kParentSideStdio][stream];
-        ArrayPrototypeForEach(chunks, ({ chunk, encoding }) => {
+        // This is a hot path, use a for(;;) loop
+        for (let i = 0; i < chunks.length; i++) {
+          const { chunk, encoding } = chunks[i];
           readable.push(chunk, encoding);
-        });
+        }
         return;
       }
       case messageTypes.STDIO_WANTS_MORE_DATA:

lib/internal/worker/io.js

@@ -1,9 +1,7 @@
 'use strict';
 
 const {
-  ArrayPrototypeForEach,
-  ArrayPrototypeMap,
-  ArrayPrototypePush,
+  Array,
   FunctionPrototypeBind,
   FunctionPrototypeCall,
   ObjectAssign,
@@ -77,7 +75,7 @@ const kOnMessage = Symbol('kOnMessage');
 const kOnMessageError = Symbol('kOnMessageError');
 const kPort = Symbol('kPort');
 const kWaitingStreams = Symbol('kWaitingStreams');
-const kWritableCallbacks = Symbol('kWritableCallbacks');
+const kWritableCallback = Symbol('kWritableCallback');
 const kStartedReading = Symbol('kStartedReading');
 const kStdioWantsMoreDataCallback = Symbol('kStdioWantsMoreDataCallback');
 const kCurrentlyReceivingPorts =
@@ -282,20 +280,29 @@ class WritableWorkerStdio extends Writable {
     super({ decodeStrings: false });
     this[kPort] = port;
     this[kName] = name;
-    this[kWritableCallbacks] = [];
+    this[kWritableCallback] = null;
   }
 
   _writev(chunks, cb) {
+    const toSend = new Array(chunks.length);
+
+    // We avoid .map() because it's a hot path
+    for (let i = 0; i < chunks.length; i++) {
+      const { chunk, encoding } = chunks[i];
+      toSend[i] = { chunk, encoding };
+    }
+
     this[kPort].postMessage({
       type: messageTypes.STDIO_PAYLOAD,
       stream: this[kName],
-      chunks: ArrayPrototypeMap(chunks,
-                                ({ chunk, encoding }) => ({ chunk, encoding })),
+      chunks: toSend,
     });
     if (process._exiting) {
       cb();
     } else {
-      ArrayPrototypePush(this[kWritableCallbacks], cb);
+      // Only one writev happens at any given time,
+      // so we can safely overwrite the callback.
+      this[kWritableCallback] = cb;
       if (this[kPort][kWaitingStreams]++ === 0)
         this[kPort].ref();
     }
@@ -311,11 +318,13 @@ class WritableWorkerStdio extends Writable {
   }
 
   [kStdioWantsMoreDataCallback]() {
-    const cbs = this[kWritableCallbacks];
-    this[kWritableCallbacks] = [];
-    ArrayPrototypeForEach(cbs, (cb) => cb());
-    if ((this[kPort][kWaitingStreams] -= cbs.length) === 0)
-      this[kPort].unref();
+    const cb = this[kWritableCallback];
+    if (cb) {
+      this[kWritableCallback] = null;
+      cb();
+      if (--this[kPort][kWaitingStreams] === 0)
+        this[kPort].unref();
+    }
   }
 }