stream: buffer single-chunk stream/iter broadcast() and share() batches lightly

As for push() writes, broadcast() writes and the source batches share()
buffers were each kept as a BatchEntry with an array of views, also for
the common case of a single chunk, and broadcast's write() first wrapped
its chunk in an array.

createEntry() returns the chunk's FixedByteView for a single-chunk batch
and a BatchEntry otherwise; validateBatchEntry() and splitBatchEntry()
take either. broadcast() and share() buffer their batches with it, and
push()'s writev() too.

Allocation per chunk, two consumers, 16-byte chunks:
- share(), sync source: 722 -> 626 bytes; async source: 1944 -> 1848
- Broadcast.from(), sync source: 868 -> 752 bytes; async: 1873 -> 1783

testBufferedBatchViewResizeRejected checks that a view resized in a
buffered multi-chunk batch still rejects the read, for broadcast() and
share(). It also passes before this change.

Assisted-by: OpenCode

Author: jasnell

lib/internal/streams/iter/broadcast.js

@@ -64,7 +64,8 @@ const {
   kNullOnceOption,
   kResolvedPromise,
   convertChunks,
-  createBatchEntry,
+  createEntry,
+  snapshotByteView,
   getProtocolMethod,
   getWriterSignal,
   getMinCursor,
@@ -684,18 +685,18 @@ class BroadcastWriter {
     const signal = getWriterSignal(options);
     // Fast path: no signal, writer open, buffer has space
     if (this.#canUseWriteFastPath(signal)) {
-      const batch = createBatchEntry([converted]);
+      const batch = snapshotByteView(converted);
       this.#broadcast[kWrite](batch);
       this.#totalBytes += batch.byteLength;
       return kResolvedPromise;
     }
-    return this.#writeBatchSlow(createBatchEntry([converted]), signal);
+    return this.#writeBatchSlow(snapshotByteView(converted), signal);
   }
 
   writev(chunks, options) {
     const converted = convertChunks(chunks);
     const signal = getWriterSignal(options);
-    const batch = createBatchEntry(converted);
+    const batch = createEntry(converted);
     // Fast path: no signal, writer open, buffer has space
     if (this.#canUseWriteFastPath(signal)) {
       if (this.#state === 'open' && this.#broadcast[kWrite](batch)) {
@@ -740,7 +741,7 @@ class BroadcastWriter {
   writeSync(chunk) {
     const converted = toWriterUint8Array(chunk);
     if (this.#state !== 'open') return false;
-    const batch = createBatchEntry([converted]);
+    const batch = snapshotByteView(converted);
     if (this.#broadcast[kWrite](batch)) {
       this.#totalBytes += batch.byteLength;
       return true;
@@ -751,7 +752,7 @@ class BroadcastWriter {
   writevSync(chunks) {
     const converted = convertChunks(chunks);
     if (this.#state !== 'open') return false;
-    const batch = createBatchEntry(converted);
+    const batch = createEntry(converted);
     if (this.#broadcast[kWrite](batch)) {
       this.#totalBytes += batch.byteLength;
       return true;
@@ -761,7 +762,7 @@ class BroadcastWriter {
 
   [kWriteBatchSync](chunks) {
     if (this.#state !== 'open') return false;
-    const batch = createBatchEntry(chunks);
+    const batch = createEntry(chunks);
     if (this.#broadcast[kWrite](batch)) {
       this.#totalBytes += batch.byteLength;
       return true;
@@ -770,7 +771,7 @@ class BroadcastWriter {
   }
 
   [kWriteBatch](chunks, signal) {
-    return this.#writeBatchSlow(createBatchEntry(chunks), signal);
+    return this.#writeBatchSlow(createEntry(chunks), signal);
   }
 
   end(options) {

lib/internal/streams/iter/push.js

@@ -36,7 +36,7 @@ const {
   kNullOnceOption,
   kPushDefaultBudget,
   kResolvedPromise,
-  createBatchEntry,
+  createEntry,
   onSignalAbort,
   toWriterUint8Array,
   convertChunks,
@@ -196,7 +196,7 @@ class PushQueue {
    * @returns {boolean}
    */
   // `entry` is a BatchEntry, or for a single chunk a FixedByteView (see
-  // entryOfChunk()).
+  // createEntry()).
   writeSync(entry) {
     if (this.#writerState !== 'open') return false;
     if (this.#consumerState !== 'active') return false;
@@ -651,15 +651,6 @@ 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) {
@@ -694,7 +685,7 @@ class PushWriter {
   }
 
   write(chunk, options) {
-    const entry = entryOfChunk(toWriterUint8Array(chunk));
+    const entry = snapshotByteView(toWriterUint8Array(chunk));
     const signal = getWriterSignal(options);
     if (!signal && this.#queue.canWriteSync()) {
       this.#queue.writeSync(entry);
@@ -704,7 +695,7 @@ class PushWriter {
   }
 
   writev(chunks, options) {
-    const entry = createBatchEntry(convertChunks(chunks));
+    const entry = createEntry(convertChunks(chunks));
     const signal = getWriterSignal(options);
     if (!signal && this.#queue.writeSync(entry)) {
       return kResolvedPromise;
@@ -713,11 +704,11 @@ class PushWriter {
   }
 
   writeSync(chunk) {
-    return this.#queue.writeSync(entryOfChunk(toWriterUint8Array(chunk)));
+    return this.#queue.writeSync(snapshotByteView(toWriterUint8Array(chunk)));
   }
 
   writevSync(chunks) {
-    return this.#queue.writeSync(createBatchEntry(convertChunks(chunks)));
+    return this.#queue.writeSync(createEntry(convertChunks(chunks)));
   }
 
   end(options) {

lib/internal/streams/iter/share.js

@@ -40,7 +40,7 @@ const {
 const {
   IterResult,
   kMultiConsumerDefaultBudget,
-  createBatchEntry,
+  createEntry,
   getProtocolMethod,
   getMinCursor,
   onSignalAbort,
@@ -620,7 +620,7 @@ class ShareImpl {
   }
 
   #bufferBatch(batch) {
-    const entry = createBatchEntry(batch);
+    const entry = createEntry(batch);
     // 'drop-oldest' evicts whole entries. A single pulled batch can be much
     // larger than the budget (for example when from() combines many values
     // of a sync source), and evicting it would discard every chunk in it,
@@ -948,7 +948,7 @@ class SyncShareImpl {
   }
 
   #bufferBatch(batch) {
-    const entry = createBatchEntry(batch);
+    const entry = createEntry(batch);
     // 'drop-oldest' evicts whole entries. A single pulled batch can be much
     // larger than the budget (for example when from() combines many values
     // of a sync source), and evicting it would discard every chunk in it,

lib/internal/streams/iter/utils.js

@@ -789,6 +789,17 @@ function createBatchEntry(chunks) {
   return new BatchEntry(views, byteLength);
 }
 
+// The entry a buffer (push(), broadcast(), share()) keeps for a written or
+// read batch: for a single chunk, its FixedByteView, which has the
+// byteLength a BatchEntry has, rather than a BatchEntry and an array of
+// views. A batch of one chunk is the common case. The two are told apart
+// by `views`, which a FixedByteView lacks; validateBatchEntry() and
+// splitBatchEntry() take either.
+function createEntry(chunks) {
+  if (chunks.length === 1) return snapshotByteView(chunks[0]);
+  return createBatchEntry(chunks);
+}
+
 /**
  * Split a batch entry into consecutive entries that are each smaller than
  * `limit` bytes, preserving chunk order. A single chunk of `limit` bytes or
@@ -799,7 +810,9 @@ function createBatchEntry(chunks) {
  */
 function splitBatchEntry(entry, limit) {
   const { views } = entry;
-  if (entry.byteLength < limit || views.length < 2) return undefined;
+  if (entry.byteLength < limit || views === undefined || views.length < 2) {
+    return undefined;
+  }
   const entries = [];
   let current = [];
   let byteLength = 0;
@@ -818,9 +831,11 @@ function splitBatchEntry(entry, limit) {
 }
 
 function validateBatchEntry(entry) {
-  const chunks = new Array(entry.views.length);
-  for (let i = 0; i < entry.views.length; i++) {
-    chunks[i] = validateByteView(entry.views[i]);
+  const views = entry.views;
+  if (views === undefined) return [validateByteView(entry)];
+  const chunks = new Array(views.length);
+  for (let i = 0; i < views.length; i++) {
+    chunks[i] = validateByteView(views[i]);
   }
   return chunks;
 }
@@ -1048,6 +1063,7 @@ module.exports = {
   convertChunks,
   checkFixedBatchChunk,
   createBatchEntry,
+  createEntry,
   fixedBatchToEntry,
   snapshotFixedBatch,
   recordChunk,

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

@@ -134,6 +134,29 @@ async function testBroadcastRejectsResizedBufferedView() {
   await assert.rejects(writer.end(), kResizeError);
 }
 
+// The same, for a view buffered in a batch of several chunks.
+async function testBufferedBatchViewResizeRejected() {
+  {
+    const buffer = new ArrayBuffer(1, { maxByteLength: 2 });
+    const { writer, broadcast: bc } = broadcast();
+    const iterator = bc.push()[Symbol.asyncIterator]();
+    assert.strictEqual(
+      writer.writevSync([Uint8Array.of(1), new Uint8Array(buffer)]), true);
+    buffer.resize(2);
+    await assert.rejects(iterator.next(), kResizeError);
+    await assert.rejects(writer.end(), kResizeError);
+  }
+  {
+    const buffer = new ArrayBuffer(1, { maxByteLength: 2 });
+    const shared = share([[Uint8Array.of(1), new Uint8Array(buffer)]]);
+    const first = shared.pull()[Symbol.asyncIterator]();
+    const second = shared.pull()[Symbol.asyncIterator]();
+    assert.strictEqual((await first.next()).value.length, 2);
+    buffer.resize(2);
+    await assert.rejects(second.next(), kResizeError);
+  }
+}
+
 async function testShareRejectsResizedBufferedView() {
   const asyncBuffer = new ArrayBuffer(1, { maxByteLength: 2 });
   const shared = share([[new Uint8Array(asyncBuffer)]]);
@@ -372,6 +395,7 @@ Promise.all([
   testPendingWritesRejectResizedViews(),
   testBroadcastRejectsResizedBufferedView(),
   testShareRejectsResizedBufferedView(),
+  testBufferedBatchViewResizeRejected(),
   testConsumersRejectResizedViews(),
   testConsumersRejectDetachedViews(),
   testPipeRejectsWriterResize(),