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

stream: decouple transform backpressure changes · nodejs/node@9ec9383 · GitHub

/ node Public

Commit 9ec9383

Browse files
mcollina
authored and
aduh95
committed
stream: decouple transform backpressure changes
The spec's [[backpressureChangePromise]] is only ever observed by the transform source pull algorithm (settles when backpressure next becomes true) and by a sink write arriving under backpressure (settles when it next becomes false), both internal. Replace the promise record with direct continuation delivery: a parked pull is completed by enqueueing the readable controller's pull-fulfilled step on the shared resolved promise, and a parked write by a cached per-stream continuation that resolves the sink promise with the perform-transform promise, so adoption reproduces the previous derived-chain settle depth exactly. Each delivery lands at the same microtask position as the old record's reaction. transformStreamDefaultControllerPerformTransform now mirrors the reference implementation's promiseCall().then(undefined, rejection steps) directly instead of running an async wrapper pair per chunk: the transformer.transform callback is wrapped raw, a non-thenable result reuses the shared resolved promise, and a synchronous throw is delivered through a rejected promise, keeping the error timing inside a reaction. A pipe-through benchmark is added since the suite had no transform-throughput row. pipeThrough passthrough improves by ~8% and a writer-driven transform loop by ~14%; the pipe-to and read families are unchanged. Signed-off-by: Matteo Collina <hello@matteocollina.com> PR-URL: #65143 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>
1 parent d286423 commit 9ec9383

3 files changed

Lines changed: 156 additions & 44 deletions

File tree

‎benchmark/webstreams/pipe-through.js‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
'use strict';
2+
const common = require('../common.js');
3+
const {
4+
ReadableStream,
5+
TransformStream,
6+
} = require('node:stream/web');
7+
8+
const bench = common.createBenchmark(main, {
9+
n: [5e5],
10+
kind: ['default', 'transform'],
11+
});
12+
13+
async function main({ n, kind }) {
14+
const b = Buffer.alloc(64);
15+
let i = 0;
16+
const rs = new ReadableStream({
17+
pull(controller) {
18+
if (i++ < n) {
19+
controller.enqueue(b);
20+
} else {
21+
controller.close();
22+
}
23+
},
24+
});
25+
const ts = kind === 'default' ?
26+
new TransformStream() :
27+
new TransformStream({
28+
transform(chunk, controller) { controller.enqueue(chunk); },
29+
});
30+
31+
const reader = rs.pipeThrough(ts).getReader();
32+
bench.start();
33+
for (;;) {
34+
const { done } = await reader.read();
35+
if (done) break;
36+
}
37+
bench.end(n);
38+
}

‎lib/internal/webstreams/transformstream.js‎

Lines changed: 109 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@ const {
55
ObjectDefineProperties,
66
ObjectSetPrototypeOf,
77
PromisePrototypeThen,
8+
PromiseReject,
9+
PromiseResolve,
810
PromiseWithResolvers,
911
Symbol,
1012
SymbolToStringTag,
@@ -44,12 +46,14 @@ const {
4446

4547
const {
4648
createPromiseCallback1Param,
47-
createPromiseCallback2Params,
49+
createRawCallback2Params,
4850
customInspect,
4951
extractHighWaterMark,
5052
extractSizeAlgorithm,
5153
getNonWritablePropertyDescriptor,
5254
isBrandCheck,
55+
kParkedAlgorithmResult,
56+
kResolvedPromise,
5357
kState,
5458
kType,
5559
nonOpCancel,
@@ -258,7 +262,10 @@ function InternalTransferredTransformStream() {
258262
readable: undefined,
259263
writable: undefined,
260264
backpressure: undefined,
261-
backpressureChange: undefined,
265+
pullPending: false,
266+
pendingWrite: undefined,
267+
pendingWriteChunk: undefined,
268+
writeContinuation: undefined,
262269
controller: undefined,
263270
};
264271
}
@@ -348,7 +355,9 @@ const isTransformStream =
348355
const isTransformStreamDefaultController =
349356
isBrandCheck('TransformStreamDefaultController');
350357

351-
async function defaultTransformAlgorithm(chunk, controller) {
358+
// Raw callback (see createRawCallback*): invoked inside the try/catch of
359+
// transformStreamDefaultControllerPerformTransform.
360+
function defaultTransformAlgorithm(chunk, controller) {
352361
transformStreamDefaultControllerEnqueue(controller, chunk);
353362
}
354363

@@ -385,7 +394,12 @@ function initializeTransformStream(
385394
writable,
386395
controller: undefined,
387396
backpressure: undefined,
388-
backpressureChange: undefined,
397+
// Continuation slots replacing the spec's
398+
// [[backpressureChangePromise]]; see transformStreamSetBackpressure.
399+
pullPending: false,
400+
pendingWrite: undefined,
401+
pendingWriteChunk: undefined,
402+
writeContinuation: undefined,
389403
};
390404

391405
transformStreamSetBackpressure(stream, true);
@@ -422,24 +436,30 @@ function transformStreamUnblockWrite(stream) {
422436
// The spec's [[backpressureChangePromise]] is only ever observed by the
423437
// source pull algorithm (settles when backpressure next becomes true) and
424438
// by a sink write arriving while backpressure is set (settles when
425-
// backpressure next becomes false). Instead of allocating a fresh promise
426-
// record on every flip, the record is materialized lazily on first
427-
// observation and dropped once settled; flips nobody is waiting on
428-
// allocate nothing.
429-
function transformStreamBackpressureChangePromise(stream) {
430-
const state = stream[kState];
431-
return (state.backpressureChange ??= PromiseWithResolvers()).promise;
432-
}
433-
439+
// backpressure next becomes false). Both observers are internal, so the
440+
// promise record is replaced by continuation slots: a parked pull is
441+
// completed by delivering the readable controller's pull-fulfilled step,
442+
// and a parked write by the cached write continuation (see
443+
// transformStreamDefaultSinkWriteAlgorithm). Each is enqueued on the
444+
// shared resolved promise at the exact microtask position the old
445+
// record's reaction would have had.
434446
function transformStreamSetBackpressure(stream, backpressure) {
435447
const state = stream[kState];
436448
assert(state.backpressure !== backpressure);
437-
const backpressureChange = state.backpressureChange;
438-
if (backpressureChange !== undefined) {
439-
state.backpressureChange = undefined;
440-
backpressureChange.resolve();
441-
}
442449
state.backpressure = backpressure;
450+
if (backpressure) {
451+
if (state.pullPending) {
452+
state.pullPending = false;
453+
// The pull-fulfilled step exists: a pull parked it (see
454+
// transformStreamDefaultSourcePullAlgorithm), and the readable
455+
// controller creates it before invoking the pull algorithm.
456+
PromisePrototypeThen(
457+
kResolvedPromise,
458+
state.readable[kState].controller[kState].pullFulfilled);
459+
}
460+
} else if (state.pendingWrite !== undefined) {
461+
PromisePrototypeThen(kResolvedPromise, state.writeContinuation);
462+
}
443463
}
444464

445465
function setupTransformStreamDefaultController(
@@ -456,6 +476,7 @@ function setupTransformStreamDefaultController(
456476
transformAlgorithm,
457477
flushAlgorithm,
458478
cancelAlgorithm,
479+
performTransformRejected: undefined,
459480
};
460481
stream[kState].controller = controller;
461482
}
@@ -468,7 +489,7 @@ function setupTransformStreamDefaultControllerFromTransformer(
468489
const flush = transformer?.flush;
469490
const cancel = transformer?.cancel;
470491
const transformAlgorithm = transform ?
471-
createPromiseCallback2Params('transformer.transform', transform, transformer) :
492+
createRawCallback2Params('transformer.transform', transform, transformer) :
472493
defaultTransformAlgorithm;
473494
const flushAlgorithm = flush ?
474495
createPromiseCallback1Param('transformer.flush', flush, transformer) :
@@ -521,18 +542,40 @@ function transformStreamDefaultControllerError(controller, error) {
521542
transformStreamError(controller[kState].stream, error);
522543
}
523544

524-
async function transformStreamDefaultControllerPerformTransform(controller, chunk) {
545+
// Mirrors the reference implementation's
546+
// `promiseCall(transformAlgorithm, ...).then(undefined, rejectionSteps)`:
547+
// the returned promise settles one microtask after the (coerced) result
548+
// does, and a rejection errors the transform stream before propagating.
549+
// The raw transform callback plus the shared resolved promise for
550+
// non-thenable results replace the previous async wrapper's two implicit
551+
// promises per chunk.
552+
function transformStreamDefaultControllerPerformTransform(controller, chunk) {
553+
const controllerState = controller[kState];
554+
const transformAlgorithm = controllerState.transformAlgorithm;
555+
if (transformAlgorithm === undefined) {
556+
// Algorithms were cleared by a concurrent cancel/abort/close.
557+
return kResolvedPromise;
558+
}
559+
let result;
525560
try {
526-
const transformAlgorithm = controller[kState].transformAlgorithm;
527-
if (transformAlgorithm === undefined) {
528-
// Algorithms were cleared by a concurrent cancel/abort/close.
529-
return;
530-
}
531-
return await transformAlgorithm(chunk, controller);
561+
result = transformAlgorithm(chunk, controller);
532562
} catch (error) {
563+
result = PromiseReject(error);
564+
}
565+
if (result === null ||
566+
(typeof result !== 'object' && typeof result !== 'function')) {
567+
result = kResolvedPromise;
568+
} else {
569+
result = PromiseResolve(result);
570+
}
571+
controllerState.performTransformRejected ??= (error) => {
533572
transformStreamError(controller[kState].stream, error);
534573
throw error;
535-
}
574+
};
575+
return PromisePrototypeThen(
576+
result,
577+
undefined,
578+
controllerState.performTransformRejected);
536579
}
537580

538581
function transformStreamDefaultControllerTerminate(controller) {
@@ -553,26 +596,42 @@ function transformStreamDefaultControllerTerminate(controller) {
553596
}
554597

555598
function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
599+
const state = stream[kState];
556600
const {
557601
writable,
558602
controller,
559-
} = stream[kState];
603+
} = state;
560604
assert(writable[kState].state === 'writable');
561-
if (stream[kState].backpressure) {
562-
const backpressureChange = transformStreamBackpressureChangePromise(stream);
563-
return PromisePrototypeThen(
564-
backpressureChange,
565-
() => {
566-
const {
567-
writable,
568-
} = stream[kState];
569-
if (writable[kState].state === 'erroring')
570-
throw writable[kState].storedError;
571-
assert(writable[kState].state === 'writable');
572-
return transformStreamDefaultControllerPerformTransform(
605+
if (state.backpressure) {
606+
// Park the chunk and one promise record; the backpressure -> false
607+
// flip delivers the cached continuation (see
608+
// transformStreamSetBackpressure) at the same microtask position as
609+
// the old [[backpressureChangePromise]] reaction. The continuation
610+
// resolves the sink promise with the perform-transform promise, so
611+
// adoption reproduces the old derived-chain settle depth exactly.
612+
// The writable dispatches a single write at a time, so one pending
613+
// slot suffices.
614+
assert(state.pendingWrite === undefined);
615+
const pendingWrite = PromiseWithResolvers();
616+
state.pendingWrite = pendingWrite;
617+
state.pendingWriteChunk = chunk;
618+
state.writeContinuation ??= () => {
619+
const pending = state.pendingWrite;
620+
const pendingChunk = state.pendingWriteChunk;
621+
state.pendingWrite = undefined;
622+
state.pendingWriteChunk = undefined;
623+
const writableState = state.writable[kState];
624+
if (writableState.state === 'erroring') {
625+
pending.reject(writableState.storedError);
626+
return;
627+
}
628+
assert(writableState.state === 'writable');
629+
pending.resolve(
630+
transformStreamDefaultControllerPerformTransform(
573631
controller,
574-
chunk);
575-
});
632+
pendingChunk));
633+
};
634+
return pendingWrite.promise;
576635
}
577636
return transformStreamDefaultControllerPerformTransform(controller, chunk);
578637
}
@@ -642,9 +701,15 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) {
642701
}
643702

644703
function transformStreamDefaultSourcePullAlgorithm(stream) {
645-
assert(stream[kState].backpressure);
704+
const state = stream[kState];
705+
assert(state.backpressure);
646706
transformStreamSetBackpressure(stream, false);
647-
return transformStreamBackpressureChangePromise(stream);
707+
// Park the pull: the next backpressure -> true flip delivers the
708+
// pull-fulfilled step (see transformStreamSetBackpressure). The old
709+
// [[backpressureChangePromise]] this replaces was only ever resolved,
710+
// so the parked pull needs no rejection delivery.
711+
state.pullPending = true;
712+
return kParkedAlgorithmResult;
648713
}
649714

650715
function transformStreamDefaultSourceCancelAlgorithm(stream, reason) {

‎lib/internal/webstreams/util.js‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -355,6 +355,12 @@ function createRawCallback2Params(name, fn, thisArg) {
355355
// the next microtask checkpoint without allocating a fresh promise.
356356
const kResolvedPromise = PromiseResolve();
357357

358+
// Returned by an internal algorithm to signal that it parked the
359+
// operation and takes responsibility for delivering the fulfilled (or
360+
// rejected) continuation itself later, instead of settling a promise
361+
// (see the transform stream source pull algorithm).
362+
const kParkedAlgorithmResult = { __proto__: null };
363+
358364
// Wires the (possibly non-thenable) result of an underlying algorithm
359365
// callback to its fulfilled/rejected continuations. A non-thenable result
360366
// means fulfillment is guaranteed and no then() lookup is observable, so
@@ -364,6 +370,8 @@ const kResolvedPromise = PromiseResolve();
364370
// matches the spec's "a promise resolved with" conversion (identity for
365371
// native promises).
366372
function thenAlgorithmResult(result, onFulfilled, onRejected) {
373+
if (result === kParkedAlgorithmResult)
374+
return;
367375
if (result === null ||
368376
(typeof result !== 'object' && typeof result !== 'function')) {
369377
PromisePrototypeThen(kResolvedPromise, onFulfilled);
@@ -457,6 +465,7 @@ module.exports = {
457465
isBrandCheck,
458466
isPromisePending,
459467
kEmptyQueue,
468+
kParkedAlgorithmResult,
460469
kResolvedPromise,
461470
kState,
462471
kType,

0 commit comments

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