stream: apply more perf improvements to stream/new

Author: jasnell

lib/internal/streams/new/broadcast.js

@@ -56,6 +56,9 @@ const {
 
 const encoder = new TextEncoder();
 
+// Cached resolved promise to avoid allocating a new one on every sync fast-path.
+const kResolvedPromise = PromiseResolve();
+
 // Non-exported symbol for internal cancel notification from BroadcastImpl
 // to BroadcastWriter. Because this symbol is not exported, external code
 // cannot call it.
@@ -388,11 +391,37 @@ class BroadcastWriter {
     return this._broadcast._getDesiredSize();
   }
 
-  async write(chunk, options) {
+  write(chunk, options) {
+    // Fast path: no signal, writer open, buffer has space
+    if (!options?.signal && !this._closed && !this._aborted &&
+        this._broadcast._canWrite()) {
+      const converted =
+        typeof chunk === 'string' ? encoder.encode(chunk) : chunk;
+      this._broadcast._write([converted]);
+      this._totalBytes += converted.byteLength;
+      return kResolvedPromise;
+    }
     return this.writev([chunk], options);
   }
 
-  async writev(chunks, options) {
+  writev(chunks, options) {
+    // Fast path: no signal, writer open, buffer has space
+    if (!options?.signal && !this._closed && !this._aborted &&
+        this._broadcast._canWrite()) {
+      const converted = allUint8Array(chunks) ?
+        ArrayPrototypeSlice(chunks) :
+        ArrayPrototypeMap(chunks, (c) =>
+          (typeof c === 'string' ? encoder.encode(c) : c));
+      this._broadcast._write(converted);
+      for (let i = 0; i < converted.length; i++) {
+        this._totalBytes += converted[i].byteLength;
+      }
+      return kResolvedPromise;
+    }
+    return this._writevSlow(chunks, options);
+  }
+
+  async _writevSlow(chunks, options) {
     const signal = options?.signal;
 
     // Check for pre-aborted signal
@@ -404,9 +433,9 @@ class BroadcastWriter {
       throw new ERR_INVALID_STATE('Writer is closed');
     }
 
-    const converted = allUint8Array(chunks)
-      ? ArrayPrototypeSlice(chunks)
-      : ArrayPrototypeMap(chunks, (c) =>
+    const converted = allUint8Array(chunks) ?
+      ArrayPrototypeSlice(chunks) :
+      ArrayPrototypeMap(chunks, (c) =>
         (typeof c === 'string' ? encoder.encode(c) : c));
 
     if (this._broadcast._write(converted)) {
@@ -447,9 +476,9 @@ class BroadcastWriter {
   writevSync(chunks) {
     if (this._closed || this._aborted) return false;
     if (!this._broadcast._canWrite()) return false;
-    const converted = allUint8Array(chunks)
-      ? ArrayPrototypeSlice(chunks)
-      : ArrayPrototypeMap(chunks, (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++) {
@@ -461,12 +490,12 @@ class BroadcastWriter {
   }
 
   // end() is synchronous internally - signal accepted for interface compliance.
-  async end(options) {
-    if (this._closed) return this._totalBytes;
+  end(options) {
+    if (this._closed) return PromiseResolve(this._totalBytes);
     this._closed = true;
     this._broadcast._end();
     this._resolvePendingDrains(false);
-    return this._totalBytes;
+    return PromiseResolve(this._totalBytes);
   }
 
   endSync() {
@@ -477,14 +506,15 @@ class BroadcastWriter {
     return this._totalBytes;
   }
 
-  async fail(reason) {
-    if (this._aborted) return;
+  fail(reason) {
+    if (this._aborted) return kResolvedPromise;
     this._aborted = true;
     this._closed = true;
     const error = reason ?? new ERR_INVALID_STATE('Failed');
     this._rejectPendingWrites(error);
     this._rejectPendingDrains(error);
     this._broadcast._abort(error);
+    return kResolvedPromise;
   }
 
   failSync(reason) {

lib/internal/streams/new/push.js

@@ -38,6 +38,9 @@ const {
   pull: pullWithTransforms,
 } = require('internal/streams/new/pull');
 
+// Cached resolved promise to avoid allocating a new one on every sync fast-path.
+const kResolvedPromise = PromiseResolve();
+
 // =============================================================================
 // PushQueue - Internal Queue with Chunk-Based Backpressure
 // =============================================================================
@@ -432,12 +435,30 @@ class PushWriter {
     return this._queue.desiredSize;
   }
 
-  async write(chunk, options) {
+  write(chunk, options) {
+    if (!options?.signal && this._queue.canWriteSync()) {
+      const bytes = toUint8Array(chunk);
+      this._queue.writeSync([bytes]);
+      return kResolvedPromise;
+    }
     const bytes = toUint8Array(chunk);
-    await this._queue.writeAsync([bytes], options?.signal);
+    return this._queue.writeAsync([bytes], options?.signal);
   }
 
-  async writev(chunks, options) {
+  writev(chunks, options) {
+    if (!options?.signal && this._queue.canWriteSync()) {
+      let bytes;
+      if (allUint8Array(chunks)) {
+        bytes = ArrayPrototypeSlice(chunks);
+      } else {
+        bytes = [];
+        for (let i = 0; i < chunks.length; i++) {
+          ArrayPrototypePush(bytes, toUint8Array(chunks[i]));
+        }
+      }
+      this._queue.writeSync(bytes);
+      return kResolvedPromise;
+    }
     let bytes;
     if (allUint8Array(chunks)) {
       bytes = ArrayPrototypeSlice(chunks);
@@ -447,7 +468,7 @@ class PushWriter {
         ArrayPrototypePush(bytes, toUint8Array(chunks[i]));
       }
     }
-    await this._queue.writeAsync(bytes, options?.signal);
+    return this._queue.writeAsync(bytes, options?.signal);
   }
 
   writeSync(chunk) {
@@ -470,18 +491,19 @@ class PushWriter {
     return this._queue.writeSync(bytes);
   }
 
-  async end(options) {
+  end(options) {
     // end() on PushQueue is synchronous (sets state, resolves pending reads).
     // Signal accepted for interface compliance but there is nothing to cancel.
-    return this._queue.end();
+    return PromiseResolve(this._queue.end());
   }
 
   endSync() {
     return this._queue.end();
   }
 
-  async fail(reason) {
+  fail(reason) {
     this._queue.fail(reason);
+    return kResolvedPromise;
   }
 
   failSync(reason) {