stream: fixup some perf bugs in stream/new

Author: jasnell

lib/internal/streams/new/broadcast.js

@@ -50,6 +50,10 @@ const {
   pull: pullWithTransforms,
 } = require('internal/streams/new/pull');
 
+const {
+  allUint8Array,
+} = require('internal/streams/new/utils');
+
 const encoder = new TextEncoder();
 
 // Non-exported symbol for internal cancel notification from BroadcastImpl
@@ -400,8 +404,10 @@ class BroadcastWriter {
       throw new ERR_INVALID_STATE('Writer is closed');
     }
 
-    const converted = ArrayPrototypeMap(chunks, (c) =>
-      (typeof c === 'string' ? encoder.encode(c) : c));
+    const converted = allUint8Array(chunks)
+      ? ArrayPrototypeSlice(chunks)
+      : ArrayPrototypeMap(chunks, (c) =>
+        (typeof c === 'string' ? encoder.encode(c) : c));
 
     if (this._broadcast._write(converted)) {
       for (let i = 0; i < converted.length; i++) {
@@ -441,8 +447,10 @@ class BroadcastWriter {
   writevSync(chunks) {
     if (this._closed || this._aborted) return false;
     if (!this._broadcast._canWrite()) return false;
-    const converted = ArrayPrototypeMap(chunks, (c) =>
-      (typeof c === 'string' ? encoder.encode(c) : c));
+    const converted = allUint8Array(chunks)
+      ? ArrayPrototypeSlice(chunks)
+      : ArrayPrototypeMap(chunks, (c) =>
+        (typeof c === 'string' ? encoder.encode(c) : c));
     if (this._broadcast._write(converted)) {
       for (let i = 0; i < converted.length; i++) {
         this._totalBytes += converted[i].byteLength;

lib/internal/streams/new/push.js

@@ -31,6 +31,7 @@ const {
 
 const {
   toUint8Array,
+  allUint8Array,
 } = require('internal/streams/new/utils');
 
 const {
@@ -437,9 +438,14 @@ class PushWriter {
   }
 
   async writev(chunks, options) {
-    const bytes = [];
-    for (let i = 0; i < chunks.length; i++) {
-      ArrayPrototypePush(bytes, toUint8Array(chunks[i]));
+    let bytes;
+    if (allUint8Array(chunks)) {
+      bytes = ArrayPrototypeSlice(chunks);
+    } else {
+      bytes = [];
+      for (let i = 0; i < chunks.length; i++) {
+        ArrayPrototypePush(bytes, toUint8Array(chunks[i]));
+      }
     }
     await this._queue.writeAsync(bytes, options?.signal);
   }
@@ -452,9 +458,14 @@ class PushWriter {
 
   writevSync(chunks) {
     if (!this._queue.canWriteSync()) return false;
-    const bytes = [];
-    for (let i = 0; i < chunks.length; i++) {
-      ArrayPrototypePush(bytes, toUint8Array(chunks[i]));
+    let bytes;
+    if (allUint8Array(chunks)) {
+      bytes = ArrayPrototypeSlice(chunks);
+    } else {
+      bytes = [];
+      for (let i = 0; i < chunks.length; i++) {
+        ArrayPrototypePush(bytes, toUint8Array(chunks[i]));
+      }
     }
     return this._queue.writeSync(bytes);
   }

lib/internal/streams/new/utils.js

@@ -26,6 +26,19 @@ function toUint8Array(chunk) {
   return chunk;
 }
 
+/**
+ * Check if all chunks in an array are already Uint8Array (no strings).
+ * Short-circuits on the first string found.
+ * @param {Array<Uint8Array|string>} chunks
+ * @returns {boolean}
+ */
+function allUint8Array(chunks) {
+  for (let i = 0; i < chunks.length; i++) {
+    if (typeof chunks[i] === 'string') return false;
+  }
+  return true;
+}
+
 /**
  * Calculate total byte length of an array of chunks.
  * @param {Uint8Array[]} chunks
@@ -100,6 +113,7 @@ function parsePullArgs(args) {
 
 module.exports = {
   toUint8Array,
+  allUint8Array,
   concatBytes,
   isPullOptions,
   parsePullArgs,