stream: release only stream/iter reads that can be cancelled while pending

Since b809ad3 every kNextUncancellable read of a from() async source
registered a hook that releases the read when the normalization is
cancelled while it is pending. pipeTo() without a signal never cancels
then, and paid for the hook on every read: 2-5% on async pipes to a
write-only sink, measured in isolation.

Releasing is now asked for with kNextUncancellable(true), by the callers
that can cancel while a read is pending (pull() pipelines, pipeTo() with
a signal, the consumers with a signal). pipeTo() without a signal reads
as it did before b809ad3, with the same handlers.

pipeTo() of an async source to a write-only sink, no signal, 16-byte
chunks, isolated: 0.95-0.97x before this change, 0.98-1.01x after,
against the parent of b809ad3.

Assisted-by: OpenCode

Author: jasnell

lib/internal/streams/iter/consumers.js

@@ -178,7 +178,7 @@ async function collectAsync(source, signal, limit) {
     const uncancellable = iterator[kNextUncancellable] !== undefined;
     for (;;) {
       const result = await (uncancellable ?
-        iterator[kNextUncancellable]() : iterator.next());
+        iterator[kNextUncancellable](true) : iterator.next());
       if (result.done || state.aborted) return;
       try {
         recordBatch(result.value);

lib/internal/streams/iter/from.js

@@ -153,12 +153,14 @@ const kNextSyncBatch = Symbol('kNextSyncBatch');
 
 // The method of an iterator returned by from() for an async iterable that
 // reads like next(), for internal callers that stop reading when they
-// cancel the normalization (pipeTo(), pull() pipelines, consumers): a
-// return() or throw() while a read is pending closes the source at once,
-// as for next(), but does not settle the pending read early. That read
-// settles when the source's next() does, rejecting with the cancellation
-// reason, and the caller ignores it. It saves the read a promise of its
-// own: see createAsyncSourceNormalizer().
+// cancel the normalization (pipeTo(), pull() pipelines, consumers). It
+// saves the read a promise of its own: see createAsyncSourceNormalizer().
+// With `releasable` true, a return() or throw() while a read is pending
+// closes the source at once, as for next(), but does not settle the
+// pending read early: the read settles when the source's next() does,
+// rejecting with the cancellation reason, and the caller ignores it.
+// Without it, the caller must not cancel while a read is pending (pipeTo()
+// without a signal). The first call decides for all reads.
 const kNextUncancellable = Symbol('kNextUncancellable');
 
 // Methods of the iterator returned by yieldNormalizationAbortable(), used by
@@ -190,8 +192,8 @@ function createNormalizationIterator(createIterator) {
         cancelNormalization(context, error, true);
         return FunctionPrototypeCall(iterator.throw, iterator, error);
       },
-      [kNextUncancellable]() {
-        return iterator[kNextUncancellable]();
+      [kNextUncancellable](releasable) {
+        return iterator[kNextUncancellable](releasable);
       },
       [SymbolAsyncIterator]() {
         return this;
@@ -843,9 +845,13 @@ function createAsyncSourceNormalizer(source, context) {
   let valueBatches = null;
   // Whether the source is read with kNextUncancellable: set by the
   // kNextUncancellable method, for callers that ignore a read pending when
-  // they cancel the normalization. readReleased: whether a cancellation has
-  // released such a read (see onUncancellableCancelled()).
+  // they cancel the normalization. releasableReads: whether the caller
+  // can cancel while a read is pending, so that a cancellation releases it
+  // (pipeTo() without a signal cannot, and its reads skip that).
+  // readReleased: whether a cancellation has released such a read (see
+  // onUncancellableCancelled()).
   let uncancellable = false;
+  let releasableReads = false;
   let readReleased = false;
   // Whether the source is read with kRead, and the settling functions of
   // the read's promise while it is pending.
@@ -923,8 +929,6 @@ function createAsyncSourceNormalizer(source, context) {
   // A kReadUncancellable result: handled as the result of the source's
   // next() is, in the same reaction.
   function onUncancellableResult(result) {
-    if (readReleased) throw context.reason;
-    context.resolve = null;
     let iterResult;
     try {
       iterResult = iterator[kToIterResult](result);
@@ -935,18 +939,29 @@ function createAsyncSourceNormalizer(source, context) {
   }
 
   function onUncancellableError(error) {
-    if (readReleased) throw context.reason;
-    context.resolve = null;
     iterator[kReadFailed]();
     return fail(error);
   }
 
-  // context.resolve while a kNextUncancellable read is pending: the caller
-  // ignores that read, so it is released rather than waited for. The
-  // normalization ends, the source is closed at once, as for a cancelled
-  // kRead (errors ignored), and the operations queued behind the read,
-  // such as the return() that cancelled it, run now. The read settles when
-  // the source's next() does, rejecting with the cancellation reason.
+  // The same, for a releasable read.
+  function onReleasableResult(result) {
+    if (readReleased) throw context.reason;
+    context.resolve = null;
+    return onUncancellableResult(result);
+  }
+
+  function onReleasableError(error) {
+    if (readReleased) throw context.reason;
+    context.resolve = null;
+    return onUncancellableError(error);
+  }
+
+  // context.resolve while a releasable kNextUncancellable read is pending:
+  // the caller ignores that read, so it is released rather than waited for.
+  // The normalization ends, the source is closed at once, as for a
+  // cancelled kRead (errors ignored), and the operations queued behind the
+  // read, such as the return() that cancelled it, run now. The read settles
+  // when the source's next() does, rejecting with the cancellation reason.
   function onUncancellableCancelled() {
     context.resolve = null;
     readReleased = true;
@@ -955,7 +970,7 @@ function createAsyncSourceNormalizer(source, context) {
     settled();
   }
 
-  function readUncancellable() {
+  function readReleasable() {
     const cancelledBefore = context.cancelled;
     const read = iterator[kReadUncancellable]();
     if (!cancelledBefore) {
@@ -966,8 +981,7 @@ function createAsyncSourceNormalizer(source, context) {
         context.resolve = onUncancellableCancelled;
       }
     }
-    return PromisePrototypeThen(read, onUncancellableResult,
-                                onUncancellableError);
+    return PromisePrototypeThen(read, onReleasableResult, onReleasableError);
   }
 
   // The kRead handlers: the result is handled in the reaction to the
@@ -995,7 +1009,11 @@ function createAsyncSourceNormalizer(source, context) {
   }
 
   function pullSource() {
-    if (uncancellable) return readUncancellable();
+    if (uncancellable) {
+      if (releasableReads) return readReleasable();
+      return PromisePrototypeThen(iterator[kReadUncancellable](),
+                                  onUncancellableResult, onUncancellableError);
+    }
     if (readable) {
       // One promise for the read, which a cancellation rejects, instead of
       // next()'s promise and a reaction to it.
@@ -1090,8 +1108,9 @@ function createAsyncSourceNormalizer(source, context) {
     next() { return run(doNext); },
     return(value) { return run(doReturn, value); },
     throw(error) { return run(doThrow, error); },
-    [kNextUncancellable]() {
+    [kNextUncancellable](releasable = false) {
       uncancellable = true;
+      releasableReads = releasable;
       return run(doNext);
     },
   }, null);

lib/internal/streams/iter/pull.js

@@ -1270,7 +1270,7 @@ class PipelineSource {
         this.#uncancellable = false;
       }
     }
-    if (this.#uncancellable) return this.#iterator[kNextUncancellable]();
+    if (this.#uncancellable) return this.#iterator[kNextUncancellable](true);
     return this.#iterator.next();
   }
 
@@ -2099,7 +2099,7 @@ async function pipeTo(source, ...args) {
     const uncancellable = iterator[kNextUncancellable] !== undefined;
     for (;;) {
       const result = await (uncancellable ?
-        iterator[kNextUncancellable]() : iterator.next());
+        iterator[kNextUncancellable](true) : iterator.next());
       if (result.done || abortState.aborted) return;
       try {
         const p = writeBatch(result.value);