stream: check stream/iter pipe writes more cheaply

pipeTo() and pipeToSync() check that each chunk is not resized or
detached while the writer uses it: around writeSync() with
callWithByteView() for batches of one chunk, and with a batch entry of
snapshots, one object for every chunk, for larger batches written one
chunk at a time.

A non-empty view of a fixed-length, non-shared ArrayBuffer can only
change by the buffer being detached, which makes its byteLength 0 (see
FixedByteView). For such views, check only the byteLength around
writeSync(), and snapshot batches as a FixedBatch of their chunks and
byteLengths, checking each buffer once when chunks share it. Other views
are checked as before.

Piping a sync generator yielding 16-byte chunks one per batch is about
15% faster with pipeTo() and 19% faster with pipeToSync(); with batches
of 128 chunks, pipeTo() is about 28% faster and pipeToSync() 20%.

Assisted-by: OpenCode
Signed-off-by: James M Snell <jasnell@gmail.com>

Author: jasnell

lib/internal/streams/iter/pull.js

@@ -62,10 +62,13 @@ const {
   kResolvedPromise,
   kStart,
   callWithByteView,
+  checkFixedBatchChunk,
   createBatchEntry,
   createOperationQueue,
+  fixedBatchToEntry,
   isTransformObject,
   parsePullArgs,
+  snapshotFixedBatch,
   snapshotTransform,
   toUint8Array,
   validateBatchEntry,
@@ -1546,6 +1549,24 @@ function pipeToSync(source, ...args) {
         totalBytes += TypedArrayPrototypeGetByteLength(chunk);
         continue;
       }
+      if (!hasWritevSync) {
+        // Chunks written one at a time: snapshot them without an object for
+        // every chunk when possible.
+        const fixed = snapshotFixedBatch(batch);
+        if (fixed !== null) {
+          for (let i = 0; i < batch.length; i++) {
+            const chunk = checkFixedBatchChunk(fixed, i);
+            const accepted = writer.writeSync(chunk);
+            checkFixedBatchChunk(fixed, i);
+            if (accepted === false) {
+              throw new ERR_OUT_OF_RANGE(
+                'write', 'within byte budget', 'budget exhausted');
+            }
+            totalBytes += fixed.byteLengths[i];
+          }
+          continue;
+        }
+      }
       const entry = createBatchEntry(batch);
       if (hasWritevSync && batch.length > 1) {
         const accepted = writer.writevSync(validateBatchEntry(entry));
@@ -1672,6 +1693,21 @@ async function pipeTo(source, ...args) {
   // Write a batch using try-fallback: sync first, async if needed.
   // Returns undefined on sync success, or a Promise when async fallback
   // is required. Callers must check: const p = writeBatch(b); if (p) await p;
+  // Write a FixedBatch with writeSync(), like the loop at the end of
+  // writeBatch(), falling back to the async path for the rest of the batch.
+  function writeFixedBatch(fixed) {
+    const { chunks } = fixed;
+    for (let i = 0; i < chunks.length; i++) {
+      const chunk = checkFixedBatchChunk(fixed, i);
+      if (!writer.writeSync(chunk)) {
+        checkFixedBatchChunk(fixed, i);
+        return writeBatchAsyncFallback(fixedBatchToEntry(fixed), i);
+      }
+      checkFixedBatchChunk(fixed, i);
+      totalBytes += fixed.byteLengths[i];
+    }
+  }
+
   function writeBatch(batch) {
     // Single chunk, the common case: check the view around writeSync()
     // without allocating a batch entry, and create one only to fall back to
@@ -1684,6 +1720,10 @@ async function pipeTo(source, ...args) {
       }
       return writeBatchAsyncFallback(createBatchEntry(batch), 0);
     }
+    if (!hasWritev && hasWriteSync) {
+      const fixed = snapshotFixedBatch(batch);
+      if (fixed !== null) return writeFixedBatch(fixed);
+    }
     const entry = createBatchEntry(batch);
     if (hasWritev && batch.length > 1) {
       if (!hasWritevSync ||

lib/internal/streams/iter/utils.js

@@ -468,6 +468,20 @@ function validateByteView(snapshot) {
  */
 function callWithByteView(value, method, receiver) {
   const buffer = TypedArrayPrototypeGetBuffer(value);
+  if (!isSharedArrayBuffer(buffer) && !ArrayBufferPrototypeGetResizable(buffer)) {
+    // A non-empty view of a fixed-length, non-shared ArrayBuffer can only
+    // change by the buffer being detached, which makes its byteLength 0 (see
+    // FixedByteView), so checking its byteLength is enough.
+    const byteLength = TypedArrayPrototypeGetByteLength(value);
+    if (byteLength !== 0) {
+      const result = FunctionPrototypeCall(method, receiver, value);
+      if (TypedArrayPrototypeGetByteLength(value) !== byteLength) {
+        throw new ERR_INVALID_STATE.TypeError(
+          'Byte view was resized or detached after being accepted');
+      }
+      return result;
+    }
+  }
   if (isSharedArrayBuffer(buffer)) {
     const snapshot = snapshotByteView(value);
     const result = FunctionPrototypeCall(method, receiver, value);
@@ -607,6 +621,68 @@ function validateRecordedChunks(chunks, checks) {
   return chunks;
 }
 
+// A batch accepted for writing one chunk at a time: its chunks, and their
+// byteLengths when accepted (see snapshotFixedBatch()).
+function FixedBatch(chunks, byteLengths, byteLength) {
+  this.chunks = chunks;
+  this.byteLengths = byteLengths;
+  this.byteLength = byteLength;
+}
+FixedBatch.prototype = ObjectFreeze({ __proto__: null });
+
+/**
+ * Snapshot a batch of Uint8Arrays like createBatchEntry(), without an object
+ * for every chunk, when every chunk is a non-empty view of a fixed-length,
+ * non-shared ArrayBuffer: such views can only change by being detached,
+ * which makes their byteLength 0 (see FixedByteView), so their byteLengths
+ * are all that needs to be recorded. Check a chunk with checkFixedBatchChunk()
+ * before and after using it.
+ * @param {Uint8Array[]} chunks
+ * @returns {FixedBatch|null} null if a chunk needs a full snapshot.
+ */
+function snapshotFixedBatch(chunks) {
+  const count = chunks.length;
+  const byteLengths = new Array(count);
+  let byteLength = 0;
+  let checkedBuffer;
+  for (let i = 0; i < count; i++) {
+    const view = chunks[i];
+    const buffer = TypedArrayPrototypeGetBuffer(view);
+    // Chunks often share a buffer: check each buffer once.
+    if (buffer !== checkedBuffer) {
+      if (isSharedArrayBuffer(buffer) ||
+          ArrayBufferPrototypeGetResizable(buffer)) {
+        return null;
+      }
+      checkedBuffer = buffer;
+    }
+    const length = TypedArrayPrototypeGetByteLength(view);
+    if (length === 0) return null;
+    byteLengths[i] = length;
+    byteLength += length;
+  }
+  return new FixedBatch(ArrayPrototypeSlice(chunks), byteLengths, byteLength);
+}
+
+function checkFixedBatchChunk(batch, index) {
+  const chunk = batch.chunks[index];
+  if (TypedArrayPrototypeGetByteLength(chunk) !== batch.byteLengths[index]) {
+    throw new ERR_INVALID_STATE.TypeError(
+      'Byte view was resized or detached after being accepted');
+  }
+  return chunk;
+}
+
+// The batch entry of a FixedBatch, to continue writing it asynchronously.
+function fixedBatchToEntry(batch) {
+  const { chunks, byteLengths } = batch;
+  const views = new Array(chunks.length);
+  for (let i = 0; i < chunks.length; i++) {
+    views[i] = new FixedByteView(chunks[i], byteLengths[i]);
+  }
+  return new BatchEntry(views, batch.byteLength);
+}
+
 function createBatchEntry(chunks) {
   const views = new Array(chunks.length);
   let byteLength = 0;
@@ -870,7 +946,10 @@ module.exports = {
   concatBytes,
   createOperationQueue,
   convertChunks,
+  checkFixedBatchChunk,
   createBatchEntry,
+  fixedBatchToEntry,
+  snapshotFixedBatch,
   recordChunk,
   splitBatchEntry,
   getProtocolMethod,

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

@@ -210,6 +210,53 @@ async function testPipeRejectsSyncWriteDetachOrResize() {
   assert.deepStrictEqual(written, [chunk]);
 }
 
+// Batches of several chunks written one at a time are checked against the
+// byteLengths recorded when the batch was accepted, before and after each
+// writeSync(). Detaching a later chunk, or the chunk being written, must be
+// rejected.
+async function testPipeRejectsDetachInMultiChunkBatch() {
+  for (const detachIndex of [1, 0]) {
+    const asyncBuffers = [new ArrayBuffer(2), new ArrayBuffer(2)];
+    const asyncWritten = [];
+    await assert.rejects(pipeTo([asyncBuffers.map((b) => new Uint8Array(b))], {
+      writeSync(chunk) {
+        asyncWritten.push(chunk.byteLength);
+        if (asyncWritten.length === 1) asyncBuffers[detachIndex].transfer();
+        return true;
+      },
+      write: common.mustNotCall(),
+      fail: common.mustCall(),
+    }), kResizeError);
+    assert.deepStrictEqual(asyncWritten, [2]);
+
+    const syncBuffers = [new ArrayBuffer(2), new ArrayBuffer(2)];
+    const syncWritten = [];
+    assert.throws(() => pipeToSync([syncBuffers.map((b) => new Uint8Array(b))], {
+      writeSync(chunk) {
+        syncWritten.push(chunk.byteLength);
+        if (syncWritten.length === 1) syncBuffers[detachIndex].transfer();
+        return true;
+      },
+      fail: common.mustCall(),
+    }, { preventClose: true }), kResizeError);
+    assert.deepStrictEqual(syncWritten, [2]);
+  }
+
+  // When writeSync() declines a chunk, pipeTo() writes the rest of the batch
+  // with write().
+  const chunks = [Uint8Array.of(1), Uint8Array.of(2, 3), Uint8Array.of(4)];
+  const written = [];
+  assert.strictEqual(await pipeTo([chunks], {
+    writeSync(chunk) {
+      if (chunk === chunks[1]) return false;
+      written.push(chunk);
+      return true;
+    },
+    write(chunk) { written.push(chunk); },
+  }, { preventClose: true }), 4);
+  assert.deepStrictEqual(written, chunks);
+}
+
 async function testConsumersRejectDetachedViews() {
   // Views of fixed-length buffers are tracked without a full snapshot; they
   // must still be rejected when detached after being accepted.
@@ -249,4 +296,5 @@ Promise.all([
   testConsumersRejectDetachedViews(),
   testPipeRejectsWriterResize(),
   testPipeRejectsSyncWriteDetachOrResize(),
+  testPipeRejectsDetachInMultiChunkBatch(),
 ]).then(common.mustCall());