stream: reduce per-chunk overhead in stream/iter consumers

bytes(), text(), arrayBuffer(), array() and their sync variants kept a
snapshot object for every collected chunk, plus a batch entry and its
views array, so they can detect chunks that were resized or detached
before the result is assembled. For streams of many small chunks this
dominated memory use: collecting 1,000,000 one-byte chunks peaked at
about 490 MB of heap for 1 MB of data.

A non-empty view of a fixed-length, non-shared ArrayBuffer can only
change by its buffer being detached, which makes its byteLength 0, so
recording its byteLength is enough. Keep a full snapshot only for
empty views and views of resizable or shared buffers. The same input
now peaks at about 60 MB and is collected about five times faster.

Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>
PR-URL: #66483
Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com>

Author: jasnell

lib/internal/streams/iter/consumers.js

@@ -54,9 +54,9 @@ const {
 
 const {
   concatBytes,
-  createBatchEntry,
   getProtocolMethod,
-  validateBatchEntry,
+  recordChunk,
+  validateRecordedChunks,
   yieldAbortable,
 } = require('internal/streams/iter/utils');
 
@@ -95,17 +95,6 @@ function isMergeOptions(value) {
 // Shared chunk collection helpers
 // =============================================================================
 
-function flattenBatchEntries(entries) {
-  const chunks = [];
-  for (let i = 0; i < entries.length; i++) {
-    const batch = validateBatchEntry(entries[i]);
-    for (let j = 0; j < batch.length; j++) {
-      ArrayPrototypePush(chunks, batch[j]);
-    }
-  }
-  return chunks;
-}
-
 /**
  * Collect chunks from a sync source into an array.
  * @param {Iterable<Uint8Array[]>} source
@@ -115,23 +104,20 @@ function flattenBatchEntries(entries) {
 function collectSync(source, limit) {
   // Normalize source via fromSync() - accepts strings, ArrayBuffers, protocols, etc.
   const normalized = fromSync(source);
-  const entries = [];
+  const chunks = [];
+  const checks = [];
   let totalBytes = 0;
 
   for (const batch of normalized) {
-    const entry = createBatchEntry(batch);
-    if (limit !== undefined) {
-      for (let i = 0; i < entry.views.length; i++) {
-        totalBytes += entry.views[i].byteLength;
-        if (totalBytes > limit) {
-          throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
-        }
+    for (let i = 0; i < batch.length; i++) {
+      totalBytes += recordChunk(chunks, checks, batch[i]);
+      if (limit !== undefined && totalBytes > limit) {
+        throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
       }
     }
-    ArrayPrototypePush(entries, entry);
   }
 
-  return flattenBatchEntries(entries);
+  return validateRecordedChunks(chunks, checks);
 }
 
 /**
@@ -146,14 +132,17 @@ async function collectAsync(source, signal, limit) {
 
   // Normalize source via from() - accepts strings, ArrayBuffers, protocols, etc.
   const normalized = from(source);
-  const entries = [];
+  const chunks = [];
+  const checks = [];
 
   // Fast path: no signal and no limit
   if (!signal && limit === undefined) {
     for await (const batch of normalized) {
-      ArrayPrototypePush(entries, createBatchEntry(batch));
+      for (let i = 0; i < batch.length; i++) {
+        recordChunk(chunks, checks, batch[i]);
+      }
     }
-    return flattenBatchEntries(entries);
+    return validateRecordedChunks(chunks, checks);
   }
 
   // Slow path: with signal or limit checks
@@ -162,19 +151,15 @@ async function collectAsync(source, signal, limit) {
 
   for await (const batch of iterable) {
     signal?.throwIfAborted();
-    const entry = createBatchEntry(batch);
-    if (limit !== undefined) {
-      for (let i = 0; i < entry.views.length; i++) {
-        totalBytes += entry.views[i].byteLength;
-        if (totalBytes > limit) {
-          throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
-        }
+    for (let i = 0; i < batch.length; i++) {
+      totalBytes += recordChunk(chunks, checks, batch[i]);
+      if (limit !== undefined && totalBytes > limit) {
+        throw new ERR_OUT_OF_RANGE('totalBytes', `<= ${limit}`, totalBytes);
       }
     }
-    ArrayPrototypePush(entries, entry);
   }
 
-  return flattenBatchEntries(entries);
+  return validateRecordedChunks(chunks, checks);
 }
 
 /**

lib/internal/streams/iter/utils.js

@@ -4,6 +4,7 @@ const {
   Array,
   ArrayBufferPrototypeGetByteLength,
   ArrayBufferPrototypeGetDetached,
+  ArrayBufferPrototypeGetResizable,
   ArrayPrototypePush,
   ArrayPrototypeSlice,
   PromiseResolve,
@@ -235,6 +236,51 @@ function validateByteView(snapshot) {
   return value;
 }
 
+/**
+ * Append a chunk to `chunks` for later concatenation, and the information
+ * needed to detect that it was resized or detached in the meantime to
+ * `checks`. This provides the same guarantee as snapshotByteView() and
+ * validateByteView() without allocating a snapshot for every chunk in the
+ * common case: a non-empty view of a fixed-length, non-shared ArrayBuffer can
+ * only change by the buffer being detached, which makes its byteLength 0, so
+ * its byteLength is all that needs to be recorded.
+ * @param {Uint8Array[]} chunks
+ * @param {Array<number|object>} checks
+ * @param {Uint8Array} value
+ * @returns {number} The byteLength of `value`.
+ */
+function recordChunk(chunks, checks, value) {
+  const buffer = TypedArrayPrototypeGetBuffer(value);
+  const byteLength = TypedArrayPrototypeGetByteLength(value);
+  ArrayPrototypePush(chunks, value);
+  if (byteLength === 0 || isSharedArrayBuffer(buffer) ||
+      ArrayBufferPrototypeGetResizable(buffer)) {
+    ArrayPrototypePush(checks, snapshotByteView(value));
+  } else {
+    ArrayPrototypePush(checks, byteLength);
+  }
+  return byteLength;
+}
+
+/**
+ * Validate chunks recorded with recordChunk().
+ * @param {Uint8Array[]} chunks
+ * @param {Array<number|object>} checks
+ * @returns {Uint8Array[]} `chunks`
+ */
+function validateRecordedChunks(chunks, checks) {
+  for (let i = 0; i < chunks.length; i++) {
+    const check = checks[i];
+    if (typeof check !== 'number') {
+      validateByteView(check);
+    } else if (TypedArrayPrototypeGetByteLength(chunks[i]) !== check) {
+      throw new ERR_INVALID_STATE.TypeError(
+        'Byte view was resized or detached after being accepted');
+    }
+  }
+  return chunks;
+}
+
 function createBatchEntry(chunks) {
   const views = new Array(chunks.length);
   let byteLength = 0;
@@ -489,6 +535,7 @@ module.exports = {
   concatBytes,
   convertChunks,
   createBatchEntry,
+  recordChunk,
   splitBatchEntry,
   getProtocolMethod,
   getWriterSignal,
@@ -503,6 +550,7 @@ module.exports = {
   toWriterUint8Array,
   validateBackpressure,
   validateBatchEntry,
+  validateRecordedChunks,
   validateByteView,
   yieldAbortable,
 };

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

@@ -167,12 +167,42 @@ async function testPipeRejectsWriterResize() {
   );
 }
 
+async function testConsumersRejectDetachedViews() {
+  // Views of fixed-length buffers are tracked without a full snapshot; they
+  // must still be rejected when detached after being accepted.
+  const asyncBuffer = new ArrayBuffer(2);
+  async function* asyncSource() {
+    yield [new Uint8Array(asyncBuffer), Uint8Array.of(1)];
+    asyncBuffer.transfer();
+  }
+  await assert.rejects(array(asyncSource()), kResizeError);
+  const limitedBuffer = new ArrayBuffer(2);
+  async function* limitedSource() {
+    yield [new Uint8Array(limitedBuffer)];
+    limitedBuffer.transfer();
+  }
+  await assert.rejects(array(limitedSource(), { limit: 10 }), kResizeError);
+
+  const syncBuffer = new ArrayBuffer(2);
+  function* syncSource() {
+    yield [new Uint8Array(syncBuffer, 1)];
+    syncBuffer.transfer();
+  }
+  assert.throws(() => arraySync(syncSource()), kResizeError);
+
+  // Unchanged views are returned as-is.
+  const chunk = new Uint8Array(4);
+  const [result] = arraySync([[chunk]]);
+  assert.strictEqual(result, chunk);
+}
+
 Promise.all([
   testBufferedViewMutationRejected(),
   testDropOldestUsesAcceptedByteLength(),
   testPendingWritesRejectResizedViews(),
   testBroadcastRejectsResizedBufferedView(),
   testShareRejectsResizedBufferedView(),
   testConsumersRejectResizedViews(),
+  testConsumersRejectDetachedViews(),
   testPipeRejectsWriterResize(),
 ]).then(common.mustCall());