stream: cut promise churn in webstreams hot paths · nodejs/node@f7e0c81 · GitHub
Skip to content

Commit f7e0c81

Browse files
mcollinaaduh95
mcollina
authored and
aduh95
committed
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> PR-URL: #65138 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day> Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com> Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
1 parent 562168f commit f7e0c81

3 files changed

Lines changed: 126 additions & 25 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 62 additions & 17 deletions

‎lib/internal/webstreams/util.js‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -338,6 +338,40 @@ function createPromiseCallbackNoParams(name, fn, thisArg) {
338338
return async () => FunctionPrototypeCall(fn, thisArg);
339339
}
340340

341+
// Raw variants that skip the async wrapper's implicit result promise.
342+
// Consumers of a raw callback invoke it inside try/catch and route the
343+
// result through thenAlgorithmResult() below.
344+
function createRawCallback1Param(name, fn, thisArg) {
345+
validateFunction(fn, name);
346+
return (arg) => FunctionPrototypeCall(fn, thisArg, arg);
347+
}
348+
349+
function createRawCallback2Params(name, fn, thisArg) {
350+
validateFunction(fn, name);
351+
return (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2);
352+
}
353+
354+
// A single shared, forever-resolved promise used to enqueue a reaction at
355+
// the next microtask checkpoint without allocating a fresh promise.
356+
const kResolvedPromise = PromiseResolve();
357+
358+
// Wires the (possibly non-thenable) result of an underlying algorithm
359+
// callback to its fulfilled/rejected continuations. A non-thenable result
360+
// means fulfillment is guaranteed and no then() lookup is observable, so
361+
// the fulfillment step is enqueued directly at the exact microtask
362+
// position the coerced promise's reaction would have had, skipping the
363+
// per-chunk promise allocation. For thenable results PromiseResolve()
364+
// matches the spec's "a promise resolved with" conversion (identity for
365+
// native promises).
366+
function thenAlgorithmResult(result, onFulfilled, onRejected) {
367+
if (result === null ||
368+
(typeof result !== 'object' && typeof result !== 'function')) {
369+
PromisePrototypeThen(kResolvedPromise, onFulfilled);
370+
} else {
371+
PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected);
372+
}
373+
}
374+
341375
function createPromiseCallback1Param(name, fn, thisArg) {
342376
validateFunction(fn, name);
343377
return async (arg) => FunctionPrototypeCall(fn, thisArg, arg);
@@ -386,11 +420,14 @@ async function nonOpFlush() {}
386420

387421
function nonOpStart() {}
388422

389-
async function nonOpPull() {}
423+
// nonOpPull and nonOpWrite are raw callbacks (see createRawCallback*):
424+
// their non-thenable return takes the allocation-free fast path in
425+
// thenAlgorithmResult().
426+
function nonOpPull() {}
390427

391428
async function nonOpCancel() {}
392429

393-
async function nonOpWrite() {}
430+
function nonOpWrite() {}
394431

395432
let transfer;
396433
function lazyTransfer() {
@@ -411,6 +448,8 @@ module.exports = {
411448
createPromiseCallbackNoParams,
412449
createPromiseCallback1Param,
413450
createPromiseCallback2Params,
451+
createRawCallback1Param,
452+
createRawCallback2Params,
414453
customInspect,
415454
defaultSizeAlgorithm,
416455
dequeueValue,
@@ -421,6 +460,7 @@ module.exports = {
421460
isBrandCheck,
422461
isPromisePending,
423462
kEmptyQueue,
463+
kResolvedPromise,
424464
kState,
425465
kType,
426466
lazyTransfer,
@@ -435,4 +475,5 @@ module.exports = {
435475
resetQueue,
436476
resolvedRecord,
437477
setPromiseHandled,
478+
thenAlgorithmResult,
438479
};

‎lib/internal/webstreams/writablestream.js‎

Lines changed: 21 additions & 6 deletions

0 commit comments

Comments
 (0)