stream: cut promise churn in webstreams hot paths

Three related reductions on the per-chunk paths:

Wrap user sink.write and source.pull callbacks without coercing their
result into a promise. When the callback returns a non-thenable (the
common synchronous case), fulfillment is guaranteed and no then() lookup
is observable, so the fulfilled reaction is enqueued through a single
shared resolved promise at the exact microtask position the coerced
promise's reaction would have had, skipping the implicit async-wrapper
promise per chunk. Thenable results go through PromiseResolve(), which
matches the spec's "a promise resolved with" conversion (identity for
native promises).

Park pipeTo's pump on backpressure by installing a record that
duck-types the writer's lazily-materialized [[readyPromise]] record and
whose resolve function is the pump continuation itself. Backpressure
clearing then resumes the pump directly instead of materializing a fresh
promise record plus reaction per flip, and the pump no longer schedules
a microtask per batch. writableStreamUpdateBackpressure publishes the
new backpressure state before resolving the ready record so the pump
observes the updated value.

Replace queueMicrotask() on the pipeTo and tee chunk-forwarding paths
with a reaction on the shared resolved promise, which enqueues the
continuation at the same position without the per-call scheduling
overhead.

pipe-to improves by 8-14% across all benchmark configurations, with
readable-read and tee also improving in spot runs.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

lib/internal/webstreams/readablestream.js

@@ -101,6 +101,7 @@ const {
   cloneAsUint8Array,
   copyArrayBuffer,
   createPromiseCallback1Param,
+  createRawCallback1Param,
   customInspect,
   defaultSizeAlgorithm,
   dequeueValue,
@@ -110,6 +111,7 @@ const {
   getNonWritablePropertyDescriptor,
   isBrandCheck,
   kEmptyQueue,
+  kResolvedPromise,
   kState,
   kType,
   lazyTransfer,
@@ -121,6 +123,7 @@ const {
   resetQueue,
   resolvedRecord,
   setPromiseHandled,
+  thenAlgorithmResult,
 } = require('internal/webstreams/util');
 
 const {
@@ -137,7 +140,6 @@ const {
   writableStreamDefaultWriterRelease,
   writableStreamDefaultWriterWriteWithRequest,
   writerClosedPromise,
-  writerReadyPromise,
 } = require('internal/webstreams/writablestream');
 
 const { Buffer } = require('buffer');
@@ -1674,11 +1676,31 @@ function readableStreamPipeTo(
   // the chunk travels through `pendingChunk`.
   let pendingChunk;
   let readRequest;
+  let readyHook;
 
   // Ready promise rejection is handled by the destination-errored
   // watcher.
   function ignoreReadyRejection() {}
 
+  // Parks the pump on the destination's backpressure by installing a
+  // record that duck-types the writer's lazily-materialized
+  // [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
+  // when backpressure clears (after publishing the new backpressure
+  // state), which re-enters the pump directly instead of rotating a
+  // fresh promise record plus reaction per flip. The pipe holds the only
+  // reference to the writer, so the record is never observable as a real
+  // ready promise; the erroring/release paths probe `promise` via
+  // isPromisePending() and call `reject`, so it carries a real
+  // forever-pending promise and a no-op reject.
+  function parkOnReady() {
+    readyHook ??= {
+      promise: PromiseWithResolvers().promise,
+      resolve: pump,
+      reject: ignoreReadyRejection,
+    };
+    writer[kState].ready = readyHook;
+  }
+
   function forwardChunk() {
     const chunk = pendingChunk;
     pendingChunk = undefined;
@@ -1690,10 +1712,7 @@ function readableStreamPipeTo(
     if (shuttingDown) return;
 
     if (dest[kState].backpressure) {
-      PromisePrototypeThen(
-        writerReadyPromise(writer).promise,
-        pump,
-        ignoreReadyRejection);
+      parkOnReady();
       return;
     }
 
@@ -1738,9 +1757,18 @@ function readableStreamPipeTo(
         return;
       }
 
-      // Yield to microtask queue between batches to allow events/signals
-      // to fire
-      queueMicrotask(pump);
+      // Park on backpressure directly: the ready hook resumes the pump
+      // when a completed write clears it.
+      if (dest[kState].backpressure) {
+        parkOnReady();
+        return;
+      }
+
+      // Yield to the microtask queue between batches so completed-write
+      // reactions and events/signals fire; a shared resolved promise
+      // enqueues the continuation at the same position as queueMicrotask
+      // without the per-batch scheduling overhead.
+      PromisePrototypeThen(kResolvedPromise, pump);
       return;
     }
 
@@ -1752,7 +1780,7 @@ function readableStreamPipeTo(
         // synchronous write during enqueue(). See WHATWG Streams spec
         // "ReadableStreamPipeTo" step 15's "chunk steps".
         pendingChunk = chunk;
-        queueMicrotask(forwardChunk);
+        PromisePrototypeThen(kResolvedPromise, forwardChunk);
       },
       [kClose]() {},
       [kError]() {},
@@ -1867,7 +1895,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
         // The microtask is required by the spec (ReadableStreamTee's
         // "chunk steps" queue one).
         pendingChunk = value;
-        queueMicrotask(forwardChunk);
+        PromisePrototypeThen(kResolvedPromise, forwardChunk);
       },
       [kClose]() {
         // The `process.nextTick()` is not part of the spec.
@@ -2020,7 +2048,7 @@ function readableByteStreamTee(stream) {
     defaultReadRequest ??= {
       [kChunk](chunk) {
         pendingChunk = chunk;
-        queueMicrotask(forwardChunk);
+        PromisePrototypeThen(kResolvedPromise, forwardChunk);
       },
       [kClose]() {
         reading = false;
@@ -2709,8 +2737,18 @@ function readableStreamDefaultControllerPull(controller) {
     controller[kState].pullRejected =
       (error) => readableStreamDefaultControllerError(controller, error);
   }
-  PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
+  // The pull algorithm may be a raw callback (a wrapped user source.pull
+  // returns its result uncoerced; a synchronous throw surfaces here) or an
+  // internal algorithm that always returns a promise; thenAlgorithmResult
+  // handles both.
+  let result;
+  try {
+    result = controller[kState].pullAlgorithm(controller);
+  } catch (error) {
+    result = PromiseReject(error);
+  }
+  thenAlgorithmResult(
+    result,
     controller[kState].pullFulfilled,
     controller[kState].pullRejected);
 }
@@ -2836,7 +2874,7 @@ function setupReadableStreamDefaultControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    createPromiseCallback1Param('source.pull', pull, source) :
+    createRawCallback1Param('source.pull', pull, source) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
     createPromiseCallback1Param('source.cancel', cancel, source) :
@@ -3529,8 +3567,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
     controller[kState].pullRejected =
       (error) => readableByteStreamControllerError(controller, error);
   }
-  PromisePrototypeThen(
-    controller[kState].pullAlgorithm(controller),
+  // See readableStreamDefaultControllerPull for the raw-callback contract.
+  let result;
+  try {
+    result = controller[kState].pullAlgorithm(controller);
+  } catch (error) {
+    result = PromiseReject(error);
+  }
+  thenAlgorithmResult(
+    result,
     controller[kState].pullFulfilled,
     controller[kState].pullRejected);
 }
@@ -3710,7 +3755,7 @@ function setupReadableByteStreamControllerFromSource(
     FunctionPrototypeBind(start, source, controller) :
     nonOpStart;
   const pullAlgorithm = pull ?
-    createPromiseCallback1Param('source.pull', pull, source) :
+    createRawCallback1Param('source.pull', pull, source) :
     nonOpPull;
   const cancelAlgorithm = cancel ?
     createPromiseCallback1Param('source.cancel', cancel, source) :

lib/internal/webstreams/util.js

@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
   return async () => FunctionPrototypeCall(fn, thisArg);
 }
 
+// Raw variants that skip the async wrapper's implicit result promise.
+// Consumers of a raw callback invoke it inside try/catch and route the
+// result through thenAlgorithmResult() below.
+function createRawCallback1Param(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return (arg) => FunctionPrototypeCall(fn, thisArg, arg);
+}
+
+function createRawCallback2Params(name, fn, thisArg) {
+  validateFunction(fn, name);
+  return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
+}
+
+// A single shared, forever-resolved promise used to enqueue a reaction at
+// the next microtask checkpoint without allocating a fresh promise.
+const kResolvedPromise = PromiseResolve();
+
+// Wires the (possibly non-thenable) result of an underlying algorithm
+// callback to its fulfilled/rejected continuations. A non-thenable result
+// means fulfillment is guaranteed and no then() lookup is observable, so
+// the fulfillment step is enqueued directly at the exact microtask
+// position the coerced promise's reaction would have had, skipping the
+// per-chunk promise allocation. For thenable results PromiseResolve()
+// matches the spec's "a promise resolved with" conversion (identity for
+// native promises).
+function thenAlgorithmResult(result, onFulfilled, onRejected) {
+  if (result === null ||
+      (typeof result !== 'object' && typeof result !== 'function')) {
+    PromisePrototypeThen(kResolvedPromise, onFulfilled);
+  } else {
+    PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected);
+  }
+}
+
 function createPromiseCallback1Param(name, fn, thisArg) {
   validateFunction(fn, name);
   return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
 
 function nonOpStart() {}
 
-async function nonOpPull() {}
+// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
+// their non-thenable return takes the allocation-free fast path in
+// thenAlgorithmResult().
+function nonOpPull() {}
 
 async function nonOpCancel() {}
 
-async function nonOpWrite() {}
+function nonOpWrite() {}
 
 let transfer;
 function lazyTransfer() {
@@ -411,6 +448,8 @@ module.exports = {
   createPromiseCallbackNoParams,
   createPromiseCallback1Param,
   createPromiseCallback2Params,
+  createRawCallback1Param,
+  createRawCallback2Params,
   customInspect,
   defaultSizeAlgorithm,
   dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
   isBrandCheck,
   isPromisePending,
   kEmptyQueue,
+  kResolvedPromise,
   kState,
   kType,
   lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
   resetQueue,
   resolvedRecord,
   setPromiseHandled,
+  thenAlgorithmResult,
 };

lib/internal/webstreams/writablestream.js

@@ -56,7 +56,7 @@ const {
   Queue,
   createPromiseCallbackNoParams,
   createPromiseCallback1Param,
-  createPromiseCallback2Params,
+  createRawCallback2Params,
   customInspect,
   defaultSizeAlgorithm,
   dequeueValue,
@@ -78,6 +78,7 @@ const {
   resetQueue,
   resolvedRecord,
   setPromiseHandled,
+  thenAlgorithmResult,
 } = require('internal/webstreams/util');
 
 const {
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
   const backpressure =
     controllerState.highWaterMark - controllerState.queueTotalSize <= 0;
   const writer = streamState.writer;
-  if (writer !== undefined && streamState.backpressure !== backpressure) {
+  const changed = streamState.backpressure !== backpressure;
+  // The state field is published before the ready record is resolved so
+  // that a ready resolve hook (pipeTo's pump continuation) observes the
+  // new value.
+  streamState.backpressure = backpressure;
+  if (writer !== undefined && changed) {
     if (backpressure) {
       // The spec replaces [[readyPromise]] with a fresh pending promise;
       // dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
       writer[kState].ready?.resolve();
     }
   }
-  streamState.backpressure = backpressure;
 }
 
 function writableStreamStartErroring(stream, reason) {
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
     };
   }
 
-  PromisePrototypeThen(
-    writeAlgorithm(chunk, controller),
+  // The write algorithm may be a raw callback (a wrapped user sink.write
+  // returns its result uncoerced; a synchronous throw surfaces here) or an
+  // internal algorithm that always returns a promise; thenAlgorithmResult
+  // handles both.
+  let result;
+  try {
+    result = writeAlgorithm(chunk, controller);
+  } catch (error) {
+    result = PromiseReject(error);
+  }
+  thenAlgorithmResult(
+    result,
     controller[kState].writeFulfilled,
     controller[kState].writeRejected);
 }
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
     FunctionPrototypeBind(start, sink, controller) :
     nonOpStart;
   const writeAlgorithm = write ?
-    createPromiseCallback2Params('sink.write', write, sink) :
+    createRawCallback2Params('sink.write', write, sink) :
     nonOpWrite;
   const closeAlgorithm = close ?
     createPromiseCallbackNoParams('sink.close', close, sink) :