FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

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

/ node Public

Commit f7e0c81

Browse files
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
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,7 @@ const {
101101
cloneAsUint8Array,
102102
copyArrayBuffer,
103103
createPromiseCallback1Param,
104+
createRawCallback1Param,
104105
customInspect,
105106
defaultSizeAlgorithm,
106107
dequeueValue,
@@ -110,6 +111,7 @@ const {
110111
getNonWritablePropertyDescriptor,
111112
isBrandCheck,
112113
kEmptyQueue,
114+
kResolvedPromise,
113115
kState,
114116
kType,
115117
lazyTransfer,
@@ -121,6 +123,7 @@ const {
121123
resetQueue,
122124
resolvedRecord,
123125
setPromiseHandled,
126+
thenAlgorithmResult,
124127
} = require('internal/webstreams/util');
125128

126129
const {
@@ -137,7 +140,6 @@ const {
137140
writableStreamDefaultWriterRelease,
138141
writableStreamDefaultWriterWriteWithRequest,
139142
writerClosedPromise,
140-
writerReadyPromise,
141143
} = require('internal/webstreams/writablestream');
142144

143145
const { Buffer } = require('buffer');
@@ -1664,11 +1666,31 @@ function readableStreamPipeTo(
16641666
// the chunk travels through `pendingChunk`.
16651667
let pendingChunk;
16661668
let readRequest;
1669+
let readyHook;
16671670

16681671
// Ready promise rejection is handled by the destination-errored
16691672
// watcher.
16701673
function ignoreReadyRejection() {}
16711674

1675+
// Parks the pump on the destination's backpressure by installing a
1676+
// record that duck-types the writer's lazily-materialized
1677+
// [[readyPromise]] record: writableStreamUpdateBackpressure resolves it
1678+
// when backpressure clears (after publishing the new backpressure
1679+
// state), which re-enters the pump directly instead of rotating a
1680+
// fresh promise record plus reaction per flip. The pipe holds the only
1681+
// reference to the writer, so the record is never observable as a real
1682+
// ready promise; the erroring/release paths probe `promise` via
1683+
// isPromisePending() and call `reject`, so it carries a real
1684+
// forever-pending promise and a no-op reject.
1685+
function parkOnReady() {
1686+
readyHook ??= {
1687+
promise: new Promise(nonOpCallback),
1688+
resolve: pump,
1689+
reject: ignoreReadyRejection,
1690+
};
1691+
writer[kState].ready = readyHook;
1692+
}
1693+
16721694
function forwardChunk() {
16731695
const chunk = pendingChunk;
16741696
pendingChunk = undefined;
@@ -1680,10 +1702,7 @@ function readableStreamPipeTo(
16801702
if (shuttingDown) return;
16811703

16821704
if (dest[kState].backpressure) {
1683-
PromisePrototypeThen(
1684-
writerReadyPromise(writer).promise,
1685-
pump,
1686-
ignoreReadyRejection);
1705+
parkOnReady();
16871706
return;
16881707
}
16891708

@@ -1728,9 +1747,18 @@ function readableStreamPipeTo(
17281747
return;
17291748
}
17301749

1731-
// Yield to microtask queue between batches to allow events/signals
1732-
// to fire
1733-
queueMicrotask(pump);
1750+
// Park on backpressure directly: the ready hook resumes the pump
1751+
// when a completed write clears it.
1752+
if (dest[kState].backpressure) {
1753+
parkOnReady();
1754+
return;
1755+
}
1756+
1757+
// Yield to the microtask queue between batches so completed-write
1758+
// reactions and events/signals fire; a shared resolved promise
1759+
// enqueues the continuation at the same position as queueMicrotask
1760+
// without the per-batch scheduling overhead.
1761+
PromisePrototypeThen(kResolvedPromise, pump);
17341762
return;
17351763
}
17361764

@@ -1742,7 +1770,7 @@ function readableStreamPipeTo(
17421770
// synchronous write during enqueue(). See WHATWG Streams spec
17431771
// "ReadableStreamPipeTo" step 15's "chunk steps".
17441772
pendingChunk = chunk;
1745-
queueMicrotask(forwardChunk);
1773+
PromisePrototypeThen(kResolvedPromise, forwardChunk);
17461774
},
17471775
[kClose]() {},
17481776
[kError]() {},
@@ -1857,7 +1885,7 @@ function readableStreamDefaultTee(stream, cloneForBranch2) {
18571885
// The microtask is required by the spec (ReadableStreamTee's
18581886
// "chunk steps" queue one).
18591887
pendingChunk = value;
1860-
queueMicrotask(forwardChunk);
1888+
PromisePrototypeThen(kResolvedPromise, forwardChunk);
18611889
},
18621890
[kClose]() {
18631891
// The `process.nextTick()` is not part of the spec.
@@ -2010,7 +2038,7 @@ function readableByteStreamTee(stream) {
20102038
defaultReadRequest ??= {
20112039
[kChunk](chunk) {
20122040
pendingChunk = chunk;
2013-
queueMicrotask(forwardChunk);
2041+
PromisePrototypeThen(kResolvedPromise, forwardChunk);
20142042
},
20152043
[kClose]() {
20162044
reading = false;
@@ -2699,8 +2727,18 @@ function readableStreamDefaultControllerPull(controller) {
26992727
controller[kState].pullRejected =
27002728
(error) => readableStreamDefaultControllerError(controller, error);
27012729
}
2702-
PromisePrototypeThen(
2703-
controller[kState].pullAlgorithm(controller),
2730+
// The pull algorithm may be a raw callback (a wrapped user source.pull
2731+
// returns its result uncoerced; a synchronous throw surfaces here) or an
2732+
// internal algorithm that always returns a promise; thenAlgorithmResult
2733+
// handles both.
2734+
let result;
2735+
try {
2736+
result = controller[kState].pullAlgorithm(controller);
2737+
} catch (error) {
2738+
result = PromiseReject(error);
2739+
}
2740+
thenAlgorithmResult(
2741+
result,
27042742
controller[kState].pullFulfilled,
27052743
controller[kState].pullRejected);
27062744
}
@@ -2826,7 +2864,7 @@ function setupReadableStreamDefaultControllerFromSource(
28262864
FunctionPrototypeBind(start, source, controller) :
28272865
nonOpStart;
28282866
const pullAlgorithm = pull ?
2829-
createPromiseCallback1Param('source.pull', pull, source) :
2867+
createRawCallback1Param('source.pull', pull, source) :
28302868
nonOpPull;
28312869
const cancelAlgorithm = cancel ?
28322870
createPromiseCallback1Param('source.cancel', cancel, source) :
@@ -3519,8 +3557,15 @@ function readableByteStreamControllerCallPullIfNeeded(controller) {
35193557
controller[kState].pullRejected =
35203558
(error) => readableByteStreamControllerError(controller, error);
35213559
}
3522-
PromisePrototypeThen(
3523-
controller[kState].pullAlgorithm(controller),
3560+
// See readableStreamDefaultControllerPull for the raw-callback contract.
3561+
let result;
3562+
try {
3563+
result = controller[kState].pullAlgorithm(controller);
3564+
} catch (error) {
3565+
result = PromiseReject(error);
3566+
}
3567+
thenAlgorithmResult(
3568+
result,
35243569
controller[kState].pullFulfilled,
35253570
controller[kState].pullRejected);
35263571
}
@@ -3700,7 +3745,7 @@ function setupReadableByteStreamControllerFromSource(
37003745
FunctionPrototypeBind(start, source, controller) :
37013746
nonOpStart;
37023747
const pullAlgorithm = pull ?
3703-
createPromiseCallback1Param('source.pull', pull, source) :
3748+
createRawCallback1Param('source.pull', pull, source) :
37043749
nonOpPull;
37053750
const cancelAlgorithm = cancel ?
37063751
createPromiseCallback1Param('source.cancel', cancel, source) :

‎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
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ const {
5656
Queue,
5757
createPromiseCallbackNoParams,
5858
createPromiseCallback1Param,
59-
createPromiseCallback2Params,
59+
createRawCallback2Params,
6060
customInspect,
6161
defaultSizeAlgorithm,
6262
dequeueValue,
@@ -78,6 +78,7 @@ const {
7878
resetQueue,
7979
resolvedRecord,
8080
setPromiseHandled,
81+
thenAlgorithmResult,
8182
} = require('internal/webstreams/util');
8283

8384
const {
@@ -769,7 +770,12 @@ function writableStreamUpdateBackpressure(controller, streamState) {
769770
const backpressure =
770771
controllerState.highWaterMark - controllerState.queueTotalSize <= 0;
771772
const writer = streamState.writer;
772-
if (writer !== undefined && streamState.backpressure !== backpressure) {
773+
const changed = streamState.backpressure !== backpressure;
774+
// The state field is published before the ready record is resolved so
775+
// that a ready resolve hook (pipeTo's pump continuation) observes the
776+
// new value.
777+
streamState.backpressure = backpressure;
778+
if (writer !== undefined && changed) {
773779
if (backpressure) {
774780
// The spec replaces [[readyPromise]] with a fresh pending promise;
775781
// dropping the cache lets the next observation derive it.
@@ -778,7 +784,6 @@ function writableStreamUpdateBackpressure(controller, streamState) {
778784
writer[kState].ready?.resolve();
779785
}
780786
}
781-
streamState.backpressure = backpressure;
782787
}
783788

784789
function writableStreamStartErroring(stream, reason) {
@@ -1197,8 +1202,18 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) {
11971202
};
11981203
}
11991204

1200-
PromisePrototypeThen(
1201-
writeAlgorithm(chunk, controller),
1205+
// The write algorithm may be a raw callback (a wrapped user sink.write
1206+
// returns its result uncoerced; a synchronous throw surfaces here) or an
1207+
// internal algorithm that always returns a promise; thenAlgorithmResult
1208+
// handles both.
1209+
let result;
1210+
try {
1211+
result = writeAlgorithm(chunk, controller);
1212+
} catch (error) {
1213+
result = PromiseReject(error);
1214+
}
1215+
thenAlgorithmResult(
1216+
result,
12021217
controller[kState].writeFulfilled,
12031218
controller[kState].writeRejected);
12041219
}
@@ -1323,7 +1338,7 @@ function setupWritableStreamDefaultControllerFromSink(
13231338
FunctionPrototypeBind(start, sink, controller) :
13241339
nonOpStart;
13251340
const writeAlgorithm = write ?
1326-
createPromiseCallback2Params('sink.write', write, sink) :
1341+
createRawCallback2Params('sink.write', write, sink) :
13271342
nonOpWrite;
13281343
const closeAlgorithm = close ?
13291344
createPromiseCallbackNoParams('sink.close', close, sink) :

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL