stream: buffer single-chunk stream/iter push() writes without a batch entry

Every push() write allocated, before the chunk was buffered, a [chunk]
array, a BatchEntry with a views array, and a FixedByteView; a read of
several buffered writes then validated each into an array of its own and
copied those into the batch it returns. That was most of what a write
costs (192 of 223 bytes for a 16-byte chunk, without pointer compression).

A single-chunk write() or writeSync() now buffers the chunk's
FixedByteView itself, which has the byteLength a BatchEntry has; entries
are told apart by `views`, which a FixedByteView lacks. writev() keeps its
BatchEntry. A read validates the buffered views straight into the batch
it returns, and a pending write is checked without collecting its chunks.
The checks are the same as before.

Allocation per write (16-byte chunks; the consumer reads the buffered
writes as they come):
- writeSync(), falling back to write(): 223 -> 70 bytes
- await write(): 397 -> 246 bytes
- strings (pooled encoding): 315 -> 185 bytes

testBufferedEntriesReadTogether checks that single-chunk and batch writes
read together are delivered in order, and that a view resized among them
rejects the read. It also passes before this change.

Assisted-by: OpenCode

Author: jasnell

lib/internal/streams/iter/push.js

@@ -42,8 +42,10 @@ const {
   convertChunks,
   getWriterSignal,
   parsePullArgs,
+  snapshotByteView,
   validateBatchEntry,
   validateBudget,
+  validateByteView,
   newPendingPromise,
   takeReject,
   takeResolve,
@@ -193,11 +195,13 @@ class PushQueue {
    * Returns true if write completed, false if buffer is full.
    * @returns {boolean}
    */
-  writeSync(chunks) {
+  // `entry` is a BatchEntry, or for a single chunk a FixedByteView (see
+  // entryOfChunk()).
+  writeSync(entry) {
     if (this.#writerState !== 'open') return false;
     if (this.#consumerState !== 'active') return false;
 
-    return this.#writeEntry(createBatchEntry(chunks));
+    return this.#writeEntry(entry);
   }
 
   #writeEntry(entry) {
@@ -249,7 +253,7 @@ class PushQueue {
    * failure.
    * @returns {Promise<void>}
    */
-  async writeAsync(chunks, signal) {
+  async writeAsync(entry, signal) {
     // Check writer state before signal (spec order: state, then signal)
     if (this.#writerState === 'closed') {
       throw new ERR_INVALID_STATE.TypeError('Writer is closed');
@@ -268,7 +272,6 @@ class PushQueue {
     // Check for pre-aborted signal (after state checks per spec)
     signal?.throwIfAborted();
 
-    const entry = createBatchEntry(chunks);
     if (this.#writeEntry(entry)) {
       return;
     }
@@ -509,16 +512,24 @@ class PushQueue {
   #drain() {
     try {
       if (this.#slots.length === 1) {
-        const result = validateBatchEntry(this.#slots.shift());
+        const entry = this.#slots.shift();
+        const result = entry.views === undefined ?
+          [validateByteView(entry)] : validateBatchEntry(entry);
         this.#bufferedBytes = 0;
         return result;
       }
 
+      // The chunks of every entry, validated straight into one batch.
       const result = [];
       for (let i = 0; i < this.#slots.length; i++) {
-        const batch = validateBatchEntry(this.#slots.get(i));
-        for (let j = 0; j < batch.length; j++) {
-          result[result.length] = batch[j];
+        const entry = this.#slots.get(i);
+        const views = entry.views;
+        if (views === undefined) {
+          result[result.length] = validateByteView(entry);
+          continue;
+        }
+        for (let j = 0; j < views.length; j++) {
+          result[result.length] = validateByteView(views[j]);
         }
       }
       this.#slots.clear();
@@ -587,7 +598,7 @@ class PushQueue {
           this.#bufferedBytes < this.#budget) {
       const pending = this.#pendingWrites.shift();
       try {
-        validateBatchEntry(pending.batch);
+        checkEntry(pending.batch);
         this.#slots.push(pending.batch);
         this.#bufferedBytes += pending.batch.byteLength;
         this.#bytesWritten += pending.batch.byteLength;
@@ -640,6 +651,26 @@ class PushQueue {
   }
 }
 
+// The buffered entry of a single-chunk write: the chunk's FixedByteView, which
+// has the byteLength a BatchEntry has, instead of a BatchEntry holding an
+// array with it. Writing one chunk is the common case, and this saves the
+// write the BatchEntry, its views array, and the [chunk] array it was made
+// from. Entries are told apart by `views`, which a FixedByteView lacks.
+function entryOfChunk(chunk) {
+  return snapshotByteView(chunk);
+}
+
+// Check that the chunks of a buffered entry still have the byteLength they
+// were accepted with, as validateBatchEntry() does, without collecting them.
+function checkEntry(entry) {
+  const views = entry.views;
+  if (views === undefined) {
+    validateByteView(entry);
+    return;
+  }
+  for (let i = 0; i < views.length; i++) validateByteView(views[i]);
+}
+
 // =============================================================================
 // PushWriter Implementation
 // =============================================================================
@@ -663,32 +694,30 @@ class PushWriter {
   }
 
   write(chunk, options) {
-    const bytes = toWriterUint8Array(chunk);
+    const entry = entryOfChunk(toWriterUint8Array(chunk));
     const signal = getWriterSignal(options);
     if (!signal && this.#queue.canWriteSync()) {
-      this.#queue.writeSync([bytes]);
+      this.#queue.writeSync(entry);
       return kResolvedPromise;
     }
-    return this.#queue.writeAsync([bytes], signal);
+    return this.#queue.writeAsync(entry, signal);
   }
 
   writev(chunks, options) {
-    const bytes = convertChunks(chunks);
+    const entry = createBatchEntry(convertChunks(chunks));
     const signal = getWriterSignal(options);
-    if (!signal && this.#queue.writeSync(bytes)) {
+    if (!signal && this.#queue.writeSync(entry)) {
       return kResolvedPromise;
     }
-    return this.#queue.writeAsync(bytes, signal);
+    return this.#queue.writeAsync(entry, signal);
   }
 
   writeSync(chunk) {
-    const bytes = toWriterUint8Array(chunk);
-    return this.#queue.writeSync([bytes]);
+    return this.#queue.writeSync(entryOfChunk(toWriterUint8Array(chunk)));
   }
 
   writevSync(chunks) {
-    const bytes = convertChunks(chunks);
-    return this.#queue.writeSync(bytes);
+    return this.#queue.writeSync(createBatchEntry(convertChunks(chunks)));
   }
 
   end(options) {

lib/internal/streams/iter/utils.js

@@ -1070,5 +1070,6 @@ module.exports = {
   validateRecordedChunks,
   throwByteViewChanged,
   validateByteView,
+  snapshotByteView,
   yieldAbortable,
 };

test/parallel/test-stream-iter-resizable-buffers.js

@@ -42,6 +42,37 @@ async function testBufferedViewMutationRejected() {
   }
 }
 
+// Several buffered writes, single chunks and batches, read as one batch:
+// they are delivered in order, and a view resized after it was written
+// among them rejects the read.
+async function testBufferedEntriesReadTogether() {
+  {
+    const { writer, readable } = push();
+    writer.writeSync(Uint8Array.of(1));
+    writer.writevSync([Uint8Array.of(2), Uint8Array.of(3)]);
+    writer.writeSync(Uint8Array.of(4));
+    writer.endSync();
+    const batches = await array(readable);
+    assert.deepStrictEqual(batches.map((c) => c[0]), [1, 2, 3, 4]);
+  }
+  for (const single of [true, false]) {
+    const buffer = new ArrayBuffer(1, { maxByteLength: 2 });
+    const { writer, readable } = push();
+    writer.writeSync(Uint8Array.of(1));
+    writer.writevSync([Uint8Array.of(2), Uint8Array.of(3)]);
+    if (single) {
+      writer.writeSync(new Uint8Array(buffer));
+    } else {
+      writer.writevSync([Uint8Array.of(4), new Uint8Array(buffer)]);
+    }
+    buffer.resize(2);
+    await assert.rejects(
+      readable[Symbol.asyncIterator]().next(),
+      kResizeError,
+    );
+  }
+}
+
 async function testDropOldestUsesAcceptedByteLength() {
   const buffer = new ArrayBuffer(16384, { maxByteLength: 16384 });
   const { writer, broadcast: bc } = broadcast({
@@ -336,6 +367,7 @@ Promise.all([
   testPipeRejectsDetachWithWriteOnly(),
   testFixedLengthViewOfResizedBuffer(),
   testBufferedViewMutationRejected(),
+  testBufferedEntriesReadTogether(),
   testDropOldestUsesAcceptedByteLength(),
   testPendingWritesRejectResizedViews(),
   testBroadcastRejectsResizedBufferedView(),