stream: read sync sources synchronously in stream/iter pipeTo()

pipeTo() iterated the from() normalization of a sync iterable source
with for await...of, which costs three promises and several ticks per
batch, even though the batches are read synchronously.

Let the iterator returned by from() for a sync iterable read the next
batch synchronously when no operation is running or queued and the
value read needs no asynchronous normalization, and have pipeTo() use it
when there are no transforms and no signal. Other values are normalized
through next() as before, a write error still closes the source, and an
error reading the source still does not.

The source and the writer see the same calls in the same order; the
batches are no longer written on separate ticks unless a write is
asynchronous. Piping a sync generator yielding 16-byte chunks, one per
batch, is about 2.5 times faster, the same as pipeToSync().

Assisted-by: OpenCode

Author: jasnell

lib/internal/streams/iter/from.js

@@ -139,9 +139,36 @@ function waitForNormalization(value, context) {
   return promise;
 }
 
+// The method of an iterator returned by from() for a sync iterable that reads
+// the next batch synchronously when it can, for pipeTo(): see nextSyncBatch()
+// in createSyncSourceNormalizer().
+const kNextSyncBatch = Symbol('kNextSyncBatch');
+
 function createNormalizationIterator(createIterator) {
   const context = createNormalizationContext();
   const iterator = createIterator(context);
+  if (iterator[kNextSyncBatch] !== undefined) {
+    return ObjectSetPrototypeOf({
+      next(value) {
+        return FunctionPrototypeCall(iterator.next, iterator, value);
+      },
+      return(value) {
+        cancelNormalization(
+          context, lazyDOMException('Aborted', 'AbortError'));
+        return FunctionPrototypeCall(iterator.return, iterator, value);
+      },
+      throw(error) {
+        cancelNormalization(context, error, true);
+        return FunctionPrototypeCall(iterator.throw, iterator, error);
+      },
+      [kNextSyncBatch]() {
+        return iterator[kNextSyncBatch]();
+      },
+      [SymbolAsyncIterator]() {
+        return this;
+      },
+    }, null);
+  }
   return ObjectSetPrototypeOf({
     next(value) {
       return FunctionPrototypeCall(iterator.next, iterator, value);
@@ -938,7 +965,7 @@ function createSyncSourceNormalizer(source, context) {
   let done = false;
   // A normalizeAsyncSourceValue() generator for the value being normalized.
   let valueBatches = null;
-  const { run, settled, release } = createOperationQueue();
+  const { run, settled, release, idle } = createOperationQueue();
 
   // Results are often produced synchronously. An async generator stays busy
   // until the tick after a yield or return (both await their operand), so
@@ -1063,10 +1090,36 @@ function createSyncSourceNormalizer(source, context) {
       });
   }
 
+  // Read the next batch like next() but synchronously, when it is read
+  // synchronously from the source and no operation is running or queued:
+  // returns the batch, or null when done. Returns undefined, having started
+  // nothing but the normalization of a value, when next() must be used.
+  // Throws on error, like next() rejects.
+  function nextSyncBatch() {
+    if (!idle() || valueBatches !== null) return undefined;
+    if (done) return null;
+    let result;
+    try {
+      result = reader.next();
+    } catch (error) {
+      done = true;
+      throw error;
+    }
+    if (result.done) {
+      done = true;
+      return null;
+    }
+    const value = result.value;
+    if (ArrayIsArray(value)) return value;
+    valueBatches = normalizeAsyncSourceValue(value.value, context, false);
+    return undefined;
+  }
+
   return ObjectSetPrototypeOf({
     next() { return run(doNext); },
     return(value) { return run(doReturn, value); },
     throw(error) { return run(doThrow, error); },
+    [kNextSyncBatch]: nextSyncBatch,
   }, null);
 }
 
@@ -1290,6 +1343,7 @@ module.exports = {
   isPrimitiveChunk,
   isSyncIterable,
   isUint8ArrayBatch,
+  kNextSyncBatch,
   normalizeAsyncSource,
   normalizeAsyncValue,
   normalizeSyncSource,

lib/internal/streams/iter/pull.js

@@ -52,6 +52,7 @@ const {
   isSyncIterable,
   isAsyncIterable,
   isUint8ArrayBatch,
+  kNextSyncBatch,
 } = require('internal/streams/iter/from');
 
 const {
@@ -1708,6 +1709,34 @@ async function pipeTo(source, ...args) {
     }
   }
 
+  // Write the batches of a sync source as the for await...of loop below
+  // does, reading them synchronously when possible: an error reading a
+  // batch ends the loop, and an error writing one closes the source first,
+  // ignoring errors closing it.
+  async function pipeSyncSource(iterator) {
+    for (;;) {
+      let batch = iterator[kNextSyncBatch]();
+      if (batch === undefined) {
+        const result = await iterator.next();
+        if (result.done) return;
+        batch = result.value;
+      } else if (batch === null) {
+        return;
+      }
+      try {
+        const p = writeBatch(batch);
+        if (p) await p;
+      } catch (error) {
+        try {
+          await iterator.return();
+        } catch {
+          // The error writing the batch is thrown.
+        }
+        throw error;
+      }
+    }
+  }
+
   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
@@ -1767,6 +1796,8 @@ async function pipeTo(source, ...args) {
           const p = writeBatch(batch);
           if (p) await p;
         }
+      } else if (normalized[kNextSyncBatch] !== undefined) {
+        await pipeSyncSource(normalized);
       } else {
         for await (const batch of normalized) {
           const p = writeBatch(batch);

lib/internal/streams/iter/utils.js

@@ -564,6 +564,10 @@ function createOperationQueue() {
       busy = false;
       drain();
     },
+    // Whether no operation is running or queued.
+    idle() {
+      return !busy && (queue === null || queue.length === 0);
+    },
   };
 }
 

test/parallel/test-stream-iter-pipeto.js

@@ -345,6 +345,52 @@ async function testPipeToSyncIterableAsyncValue() {
   assert.strictEqual(result, 'ab');
 }
 
+// pipeTo() reads sync iterables synchronously when it can. An error writing
+// a batch must still close the source, ignoring an error closing it, and an
+// error reading the source must not close it.
+async function testPipeToSyncIterableWriteError() {
+  for (const returnThrows of [false, true]) {
+    const error = new Error('write');
+    let closed = false;
+    const source = {
+      [Symbol.iterator]() {
+        let i = 0;
+        return {
+          next() {
+            return { done: false, value: [new Uint8Array([i++])] };
+          },
+          return() {
+            closed = true;
+            if (returnThrows) throw new Error('return');
+            return { done: true };
+          },
+        };
+      },
+    };
+    await assert.rejects(pipeTo(source, {
+      write: common.mustNotCall(),
+      writeSync: common.mustCall(() => { throw error; }),
+      fail: common.mustCall((reason) => assert.strictEqual(reason, error)),
+    }), error);
+    assert.strictEqual(closed, true);
+  }
+
+  const error = new Error('source');
+  const source = {
+    [Symbol.iterator]() {
+      return {
+        next() { throw error; },
+        return: common.mustNotCall(),
+      };
+    },
+  };
+  await assert.rejects(pipeTo(source, {
+    write: common.mustNotCall(),
+    writeSync: common.mustNotCall(),
+    fail: common.mustCall((reason) => assert.strictEqual(reason, error)),
+  }), error);
+}
+
 Promise.all([
   testPipeToSync(),
   testPipeTo(),
@@ -365,5 +411,6 @@ Promise.all([
   testPipeToSyncIterableUsesFromBatching(),
   testPipeToSyncIterableWriteFallback(),
   testPipeToSyncIterableAsyncValue(),
+  testPipeToSyncIterableWriteError(),
   testPipeToSourceNormalizationIndependentOfWriter(),
 ]).then(common.mustCall());