stream: read stream/iter from() sources in pipelines without a promise per read

A pull() pipeline, pipeTo() with a signal and the consumers with a signal
reject their own pending pull, pipe or read when they are aborted, and
ignore the read of the source that is still pending then. Yet they read a
from() async source with its cancellable next(), which gives every read a
promise of its own so that a cancellation can reject it at once: about
230 bytes per batch, for a rejection nobody waits for.

They now read such sources with kNextUncancellable, as pipeTo() without a
signal already did, when nothing but stream/iter code waits for the
reads: always for pipeTo() and the consumers, and for pull() pipelines
without stateful transforms. A stateful transform's generator waits for
the source in user code, so with one the reads stay cancellable and its
finally block still runs at once.

What makes this possible is that a cancellation now releases a pending
kNextUncancellable read instead of waiting for it: the normalization
ends, the source is closed at once (as for a cancelled next(), errors
ignored), and the operations queued behind the read, such as the return()
that cancelled it, run. The read settles when the source's next() does,
rejecting with the cancellation reason, and is ignored. A pipeline
stopped with return() or throw() while a pull is pending closes its
(stateless) transform layers without waiting for them, since they wait
for that read.

Allocation per batch, 16-byte chunks, async source:
- pipeTo() with a signal: 967 -> 738 bytes (as without a signal)
- pull() with one transform: 1518 -> 1289 bytes
- pipeTo() with a signal and one transform: 1829 -> 1601 bytes

test-stream-iter-abort-pending-read.js checks, for pull(), pipeTo() and
bytes(), with and without transforms and stopped by a signal, return()
or throw() while the source's next() is pending, that the call rejects
at once, the source is closed before that next() settles, and nothing
happens when it does. It also passes before this change.

Assisted-by: OpenCode

Author: jasnell

lib/internal/streams/iter/consumers.js

@@ -54,6 +54,7 @@ const {
   fromSync,
   isAsyncIterable,
   isSyncIterable,
+  kNextUncancellable,
 } = require('internal/streams/iter/from');
 
 const {
@@ -170,10 +171,15 @@ async function collectAsync(source, signal, limit) {
 
   // Read as for await...of does, stopping once the signal aborts; see
   // raceSignal().
+  // A from() source is read with kNextUncancellable: on an abort,
+  // raceSignal() rejects and closes the source at once, and a read still
+  // pending is ignored.
   const state = { __proto__: null, aborted: false };
   async function consume(iterator) {
+    const uncancellable = iterator[kNextUncancellable] !== undefined;
     for (;;) {
-      const result = await iterator.next();
+      const result = await (uncancellable ?
+        iterator[kNextUncancellable]() : iterator.next());
       if (result.done || state.aborted) return;
       try {
         recordBatch(result.value);

lib/internal/streams/iter/from.js

@@ -152,8 +152,13 @@ function waitForNormalization(value, context) {
 const kNextSyncBatch = Symbol('kNextSyncBatch');
 
 // The method of an iterator returned by from() for an async iterable that
-// reads like next() when nothing can cancel the normalization while a read
-// is pending, for pipeTo(): see createAsyncSourceNormalizer().
+// 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().
 const kNextUncancellable = Symbol('kNextUncancellable');
 
 // Methods of the iterator returned by yieldNormalizationAbortable(), used by
@@ -591,11 +596,12 @@ function yieldNormalizationAbortable(source, context) {
             });
           return promise;
         },
-        // Like next(), when nothing cancels the normalization while the read
-        // is pending, in two parts so that the caller handles the result in
-        // the same reaction: kReadUncancellable starts the read, returning a
-        // promise for the source's result, and the caller passes that result
-        // to kToIterResult, or calls kReadFailed if the promise rejects.
+        // Like next(), for a caller that ignores the read if it cancels the
+        // normalization while the read is pending (see kNextUncancellable),
+        // in two parts so that the caller handles the result in the same
+        // reaction: kReadUncancellable starts the read, returning a promise
+        // for the source's result, and the caller passes that result to
+        // kToIterResult, or calls kReadFailed if the promise rejects.
         [kReadUncancellable]() {
           if (completed) {
             return PromiseResolve(kDoneSourceResult);
@@ -836,9 +842,11 @@ function createAsyncSourceNormalizer(source, context) {
   let boundedBatches = null;
   let valueBatches = null;
   // Whether the source is read with kNextUncancellable: set by the
-  // kNextUncancellable method, for callers that never cancel the
-  // normalization while a read is pending.
+  // 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()).
   let uncancellable = false;
+  let readReleased = false;
   // Whether the source is read with kRead, and the settling functions of
   // the read's promise while it is pending.
   let readable = false;
@@ -915,6 +923,8 @@ 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);
@@ -925,10 +935,41 @@ 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.
+  function onUncancellableCancelled() {
+    context.resolve = null;
+    readReleased = true;
+    state = kDone;
+    markPromiseAsHandled(iterator.return());
+    settled();
+  }
+
+  function readUncancellable() {
+    const cancelledBefore = context.cancelled;
+    const read = iterator[kReadUncancellable]();
+    if (!cancelledBefore) {
+      // Cancelled while the source's next() was called, or later.
+      if (context.cancelled) {
+        onUncancellableCancelled();
+      } else {
+        context.resolve = onUncancellableCancelled;
+      }
+    }
+    return PromisePrototypeThen(read, onUncancellableResult,
+                                onUncancellableError);
+  }
+
   // The kRead handlers: the result is handled in the reaction to the
   // source's result, and settles the read's promise.
   function onRead(result) {
@@ -954,10 +995,7 @@ function createAsyncSourceNormalizer(source, context) {
   }
 
   function pullSource() {
-    if (uncancellable) {
-      return PromisePrototypeThen(iterator[kReadUncancellable](),
-                                  onUncancellableResult, onUncancellableError);
-    }
+    if (uncancellable) return readUncancellable();
     if (readable) {
       // One promise for the read, which a cancellation rejects, instead of
       // next()'s promise and a reaction to it.

lib/internal/streams/iter/pull.js

@@ -1233,20 +1233,44 @@ class PipelineSource {
   #abort;
   #iterator = null;
   #closed = false;
-
-  constructor(source, abort) {
+  // Whether a from() source is read with kNextUncancellable (see the
+  // constructor).
+  #uncancellable;
+
+  // `uncancellable`: whether nothing but stream/iter code waits for the
+  // reads, so that a read pending when the pipeline aborts can be left to
+  // settle when the source's next() does. The pipeline rejects its own
+  // pending pull and ignores the read, and closing the source still calls
+  // its return() at once. This saves every read of a from() source a
+  // promise of its own. A stateful transform reads the source in user code
+  // (its generator waits for the read), so with one the reads stay
+  // cancellable, and an abort rejects them at once.
+  constructor(source, abort, uncancellable) {
     this.#source = source;
     this.#abort = abort;
+    this.#uncancellable = uncancellable;
   }
 
   [SymbolAsyncIterator]() {
     return this;
   }
 
+  // Whether the source is read with kNextUncancellable (known once it has
+  // been read).
+  get uncancellable() {
+    return this.#uncancellable && this.#iterator !== null;
+  }
+
   next() {
     if (this.#abort.aborted) return PromiseReject(this.#abort.reason);
     if (this.#closed) return PromiseResolve(new IterResult(true, undefined));
-    this.#iterator ??= this.#source[SymbolAsyncIterator]();
+    if (this.#iterator === null) {
+      this.#iterator = this.#source[SymbolAsyncIterator]();
+      if (this.#iterator[kNextUncancellable] === undefined) {
+        this.#uncancellable = false;
+      }
+    }
+    if (this.#uncancellable) return this.#iterator[kNextUncancellable]();
     return this.#iterator.next();
   }
 
@@ -1290,6 +1314,13 @@ class PipelineSource {
   }
 }
 
+function hasTransformObject(transforms) {
+  for (let i = 0; i < transforms.length; i++) {
+    if (isTransformObject(transforms[i])) return true;
+  }
+  return false;
+}
+
 /**
  * Build the chain of transform layers of a pipeline.
  * @param {AsyncIterable<Uint8Array[]>} normalized
@@ -1377,6 +1408,8 @@ function createAsyncPipeline(source, transforms, signal,
   // runs one pull at a time.
   let resolvePull = null;
   let rejectPull = null;
+  // See stop().
+  let closeWithoutWaiting = false;
   const operations = createOperationQueue();
 
   const self = ObjectSetPrototypeOf({
@@ -1437,6 +1470,14 @@ function createAsyncPipeline(source, transforms, signal,
     if (rejectPull !== null) {
       abortPipeline(reason, false);
       state = kActive;
+      // The pending pull is waiting for a read of the source that is now
+      // ignored, and with uncancellable reads nothing settles that read
+      // early: closing the transforms would wait for it, as a transform
+      // layer's return() waits for its pending next(). The transforms are
+      // all stateless then (see PipelineSource), so nothing but stream/iter
+      // code waits for them, and doReturn() / doThrow() close them without
+      // waiting. The source has already been closed.
+      closeWithoutWaiting = pipelineSource.uncancellable;
     } else {
       abort.abort(reason);
     }
@@ -1542,7 +1583,8 @@ function createAsyncPipeline(source, transforms, signal,
     if (state === kStart) {
       state = kActive;
       abort = new PipelineAbort();
-      pipelineSource = new PipelineSource(source, abort);
+      pipelineSource = new PipelineSource(
+        source, abort, !hasTransformObject(transforms));
       if (signal !== undefined) {
         signal.addEventListener('abort', onSignalAbort,
                                 { __proto__: null, once: true,
@@ -1600,6 +1642,11 @@ function createAsyncPipeline(source, transforms, signal,
     }
     state = kDone;
     removeSignalListener();
+    if (closeWithoutWaiting) {
+      markPromiseAsHandled(closeAsyncIterator(iterator, false));
+      operations.settled();
+      return PromiseResolve(result);
+    }
     return PromisePrototypeThen(
       closeAsyncIterator(iterator, false), () => {
         operations.settled();
@@ -1618,6 +1665,11 @@ function createAsyncPipeline(source, transforms, signal,
     }
     state = kDone;
     removeSignalListener();
+    if (closeWithoutWaiting) {
+      markPromiseAsHandled(closeAsyncIterator(iterator, true));
+      operations.settled();
+      return PromiseReject(error);
+    }
     return PromisePrototypeThen(
       closeAsyncIterator(iterator, true), () => {
         operations.settled();
@@ -2040,10 +2092,14 @@ async function pipeTo(source, ...args) {
   }
 
   // Write the batches of a source read with next(), as the for await...of
-  // loop below does, stopping once `signal` aborts.
+  // loop below does, stopping once `signal` aborts. A from() source is read
+  // with kNextUncancellable: on an abort, raceSignal() rejects and closes
+  // the source at once, and a read still pending is ignored.
   async function pipeSourceWithSignal(iterator) {
+    const uncancellable = iterator[kNextUncancellable] !== undefined;
     for (;;) {
-      const result = await iterator.next();
+      const result = await (uncancellable ?
+        iterator[kNextUncancellable]() : iterator.next());
       if (result.done || abortState.aborted) return;
       try {
         const p = writeBatch(result.value);
@@ -2144,9 +2200,8 @@ async function pipeTo(source, ...args) {
     if (transforms.length === 0) {
       // Fast path: no transforms - iterate normalized source directly
       if (signal) {
-        // Batches are read synchronously when possible, as without a signal.
-        // Async reads go through next(), which return() can cancel while it
-        // is pending, unlike kNextUncancellable.
+        // Batches are read synchronously when possible, as without a signal,
+        // otherwise as pipeSourceWithSignal() says.
         const iterator = normalized[SymbolAsyncIterator]();
         await raceSignal(iterator, signal, abortState,
                          iterator[kNextSyncBatch] !== undefined ?