stream: construct stream/iter wait and merge records

The records queued when a stream/iter read, write or drain has to wait,
and merge()'s ready-queue entries, were `{ __proto__: null, ... }`
literals, which V8 creates in dictionary mode. They can be created once
per chunk: whenever the consumer is ahead of the producer, every read
waits, and with a full budget every write does.

Create them with constructors whose prototype is an empty, frozen,
null-prototype object: PendingRequest and PendingWrite (push(),
broadcast() and fromWritable()), QueuedWrite (fromWritable()) and
MergeEntry (merge()). fromWritable() drain waiters now settle through
resolve(false) instead of a per-waiter close() closure. merge() tells
error entries apart by their missing iterator rather than by a kind
string.

When every push() read waits for data, or every write waits behind a
full budget, reading or writing 16-byte chunks is about 30% faster.
fromWritable() with queued writes is about 18% faster, and merge() of
two sources about 9% faster.

Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: #66648
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -58,6 +58,8 @@ const {
 
 const {
   IterResult,
+  PendingRequest,
+  PendingWrite,
   kMultiConsumerDefaultBudget,
   kResolvedPromise,
   convertChunks,
@@ -264,7 +266,7 @@ class BroadcastImpl {
             if (state.resolve) {
               const { promise, resolve, reject } = PromiseWithResolvers();
               ArrayPrototypePush(state.pending,
-                                 { __proto__: null, resolve, reject });
+                                 new PendingRequest(resolve, reject));
               return promise;
             }
 
@@ -625,7 +627,7 @@ class BroadcastWriter {
     if (canWrite === null) return null;
     if (canWrite) return PromiseResolve(true);
     const { promise, resolve, reject } = PromiseWithResolvers();
-    ArrayPrototypePush(this.#pendingDrains, { __proto__: null, resolve, reject });
+    ArrayPrototypePush(this.#pendingDrains, new PendingRequest(resolve, reject));
     return promise;
   }
 
@@ -802,7 +804,7 @@ class BroadcastWriter {
    */
   #createPendingWrite(batch, signal) {
     const { promise, resolve, reject } = PromiseWithResolvers();
-    const entry = { __proto__: null, batch, resolve, reject };
+    const entry = new PendingWrite(batch, resolve, reject);
     this.#pendingWrites.push(entry);
     if (signal) {
       wireBroadcastWriteSignal(entry, signal, resolve, reject, this);

lib/internal/streams/iter/classic.js

@@ -14,6 +14,7 @@
 const {
   ArrayPrototypePush,
   FunctionPrototypeCall,
+  ObjectFreeze,
   Promise,
   PromisePrototypeThen,
   PromiseReject,
@@ -61,6 +62,7 @@ const {
 } = require('internal/streams/iter/types');
 
 const {
+  PendingRequest,
   convertChunks,
   getWriterSignal,
   onSignalAbort,
@@ -522,6 +524,18 @@ function toReadableSync(source, options = kNullPrototype) {
 // Cache: one Writer adapter per Writable instance.
 const fromWritableCache = new SafeWeakMap();
 
+// A write queued by fromWritable() until the writable can accept it. Like
+// the other per-write records, constructed rather than created as a
+// `{ __proto__: null, ... }` literal (a dictionary-mode object).
+function QueuedWrite(chunks, resolve, reject) {
+  this.chunks = chunks;
+  this.resolve = resolve;
+  this.reject = reject;
+  this.signal = undefined;
+  this.onAbort = undefined;
+}
+QueuedWrite.prototype = ObjectFreeze({ __proto__: null });
+
 /**
  * Create a stream/iter Writer adapter from a classic Writable (or duck-type).
  *
@@ -643,7 +657,7 @@ function fromWritable(writable, options = kNullPrototype) {
       if (preserveReason) {
         pending[i].reject(reason);
       } else {
-        pending[i].close();
+        pending[i].resolve(false);
       }
     }
 
@@ -755,14 +769,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
   function queueWrite(chunks, signal) {
     const { promise, resolve, reject } = PromiseWithResolvers();
-    const entry = {
-      __proto__: null,
-      chunks,
-      resolve,
-      reject,
-      signal: undefined,
-      onAbort: undefined,
-    };
+    const entry = new QueuedWrite(chunks, resolve, reject);
     pendingWrites.push(entry);
     installDrainListener();
 
@@ -1038,12 +1045,7 @@ function fromWritable(writable, options = kNullPrototype) {
       return PromiseResolve(true);
     }
     const { promise, resolve, reject } = PromiseWithResolvers();
-    ArrayPrototypePush(drainWaiters, {
-      __proto__: null,
-      resolve,
-      reject,
-      close() { resolve(false); },
-    });
+    ArrayPrototypePush(drainWaiters, new PendingRequest(resolve, reject));
     installDrainListener();
     return promise;
   };

lib/internal/streams/iter/consumers.js

@@ -17,6 +17,7 @@ const {
   ArrayPrototypeShift,
   ArrayPrototypeSlice,
   FunctionPrototypeCall,
+  ObjectFreeze,
   Promise,
   PromisePrototypeThen,
   SafePromiseAllReturnVoid,
@@ -456,6 +457,16 @@ function ondrain(drainable) {
 
 const kNoMergeError = Symbol('kNoMergeError');
 
+// An entry in merge()'s ready queue: a value from `iterator`, or, for a
+// source that failed, `reason` with no iterator. Created for every merged
+// chunk, so constructed rather than created as a dictionary-mode literal.
+function MergeEntry(iterator, value, reason) {
+  this.iterator = iterator;
+  this.value = value;
+  this.reason = reason;
+}
+MergeEntry.prototype = ObjectFreeze({ __proto__: null });
+
 /**
  * Merge multiple async iterables by yielding values in temporal order.
  * @param {...(AsyncIterable<Uint8Array[]>|object)} args
@@ -531,12 +542,8 @@ function merge(...args) {
         if (result.done) {
           activeCount--;
         } else {
-          ArrayPrototypePush(ready, {
-            __proto__: null,
-            kind: 'value',
-            iterator,
-            value: result.value,
-          });
+          ArrayPrototypePush(ready,
+                             new MergeEntry(iterator, result.value, undefined));
         }
         if (waitResolve) {
           waitResolve();
@@ -547,11 +554,7 @@ function merge(...args) {
       const onRejected = (iterator, reason) => {
         pendingPulls.delete(iterator);
         if (stopped) return;
-        ArrayPrototypePush(ready, {
-          __proto__: null,
-          kind: 'error',
-          reason,
-        });
+        ArrayPrototypePush(ready, new MergeEntry(undefined, undefined, reason));
         if (waitResolve) {
           waitResolve();
           waitResolve = null;
@@ -579,7 +582,7 @@ function merge(...args) {
           // Drain ready queue synchronously
           while (ready.length > 0) {
             const item = ArrayPrototypeShift(ready);
-            if (item.kind === 'error') {
+            if (item.iterator === undefined) {
               throw item.reason;
             }
             yield item.value;

lib/internal/streams/iter/push.js

@@ -30,6 +30,8 @@ const {
 
 const {
   IterResult,
+  PendingRequest,
+  PendingWrite,
   kPushDefaultBudget,
   kResolvedPromise,
   createBatchEntry,
@@ -294,7 +296,7 @@ class PushQueue {
    */
   #createPendingWrite(batch, signal) {
     const { promise, resolve, reject } = PromiseWithResolvers();
-    const entry = { __proto__: null, batch, resolve, reject };
+    const entry = new PendingWrite(batch, resolve, reject);
     this.#pendingWrites.push(entry);
 
     if (signal) {
@@ -429,7 +431,7 @@ class PushQueue {
    */
   waitForDrain() {
     const { promise, resolve, reject } = PromiseWithResolvers();
-    ArrayPrototypePush(this.#pendingDrains, { __proto__: null, resolve, reject });
+    ArrayPrototypePush(this.#pendingDrains, new PendingRequest(resolve, reject));
     return promise;
   }
 
@@ -467,7 +469,7 @@ class PushQueue {
     }
 
     const { promise, resolve, reject } = PromiseWithResolvers();
-    this.#pendingReads.push({ __proto__: null, resolve, reject });
+    this.#pendingReads.push(new PendingRequest(resolve, reject));
     return promise;
   }
 

lib/internal/streams/iter/utils.js

@@ -241,6 +241,21 @@ function BatchEntry(views, byteLength) {
 }
 BatchEntry.prototype = ObjectFreeze({ __proto__: null });
 
+// Waiters for reads, writes and drains are queued whenever a stream has
+// to wait, which can be once per chunk.
+function PendingRequest(resolve, reject) {
+  this.resolve = resolve;
+  this.reject = reject;
+}
+PendingRequest.prototype = ObjectFreeze({ __proto__: null });
+
+function PendingWrite(batch, resolve, reject) {
+  this.batch = batch;
+  this.resolve = resolve;
+  this.reject = reject;
+}
+PendingWrite.prototype = ObjectFreeze({ __proto__: null });
+
 function snapshotByteView(value) {
   const buffer = TypedArrayPrototypeGetBuffer(value);
   const sharedBufferView = isSharedArrayBuffer(buffer) ?
@@ -576,6 +591,8 @@ function validateBackpressure(value) {
 
 module.exports = {
   IterResult,
+  PendingRequest,
+  PendingWrite,
   kMultiConsumerDefaultBudget,
   kPushDefaultBudget,
   kResolvedPromise,