stream: speed up WHATWG web streams - #65273
Conversation
|
Review requested:
|
nodejs-github-bot
added
lib / src
6a3f89b to
c796760
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #65273 +/- ##
==========================================
- Coverage 91.85% 90.13% -1.73%
==========================================
Files 400 752 +352
Lines 178855 251981 +73126
Branches 27331 47380 +20049
==========================================
+ Hits 164292 227118 +62826
- Misses 14234 16178 +1944
- Partials 329 8685 +8356
🚀 New features to boost your workflow:
|
jasnell
added
the
semver-major
|
Defensively marking this semver-major. If you can show that the optimization does not change observable behavior, that can be dropped, but the change in microtask timing from one pull to the next is likely observable. |
anonrig
added
the
commit-queue-squash
|
@jasnell I believe |
|
Benchmark GHA (webstreams): https://github.com/nodejs/node/actions/runs/31808780686 |
27c4cf3 to
1cf7319
Compare
|
@jasnell can you rereview please |
|
|
||
| // Materialize the deferred default controller for `new ReadableStream()`. | ||
| // started is true immediately: start is a no-op and there is no initial pull. | ||
| function ensureEmptyDefaultController(stream) { |
There was a problem hiding this comment.
Is this correctly handled in subclasses? A subclass could end up calling cancel, getReader, etc before the constructor finishes, causing the controller to be materialized. Worth documenting and tests.
There was a problem hiding this comment.
Looks okay to me? As explained in anonrig's previous comment, the controller is materialized in all those methods:
cancelcallsensureEmptyDefaultControllerdirectlygetReadercalls it indirectly throughsetupReadableStreamDefaultReaderpipeTo/tee/valuescreate an internal reader, so they also go throughsetupReadableStreamDefaultReader
Subclasses don't really affect this: if a subclass wanted to get access to the controller, they'd still need to pass a source object with a start/pull method, which pushes them off the empty-argument path.
jasnell
left a comment
jasnell
left a comment
There was a problem hiding this comment.
Some additional review comments. Will review again once merge conflicts are resolved and I'd like @MattiasBuelens to review before this proceeds.
| } | ||
|
|
||
| // Materialize the deferred default controller for `new ReadableStream()`. | ||
| // started is true immediately: start is a no-op and there is no initial pull. |
There was a problem hiding this comment.
I'm actually impressed that this still passes WPT? 😅 I know I've found a lot of subtle issues in the tests where the behavior changed slightly depending on whether or not the test waits for the stream to be started.
| return; | ||
| } | ||
| if (isReadableStreamDefaultController(controller)) | ||
| controller.error(error); |
There was a problem hiding this comment.
Off-topic, but this should really call the abstract op instead of going through a method lookup.
| controller.error(error); | |
| readableStreamDefaultControllerError(controller, error); |
|
|
||
| // Materialize the deferred default controller for `new ReadableStream()`. | ||
| // started is true immediately: start is a no-op and there is no initial pull. | ||
| function ensureEmptyDefaultController(stream) { |
There was a problem hiding this comment.
Looks okay to me? As explained in anonrig's previous comment, the controller is materialized in all those methods:
cancelcallsensureEmptyDefaultControllerdirectlygetReadercalls it indirectly throughsetupReadableStreamDefaultReaderpipeTo/tee/valuescreate an internal reader, so they also go throughsetupReadableStreamDefaultReader
Subclasses don't really affect this: if a subclass wanted to get access to the controller, they'd still need to pass a source object with a start/pull method, which pushes them off the empty-argument path.
d3ff6d1 to
c7125f4
Compare
|
Since I'm buried in a few other things and it looks like this is mostly being updated by AI anyway, I had my agent draft up a review. I skimmed it over and can't disagree with any part of it: Details0. Reconstructed history (since the squash hid it)I fetched all four force-pushed heads. The PR was never a single commit until the last push:
The rebase matters more than the squash: a large fraction of this PR already landed on main as #65138 Three things happened in the squash that reviewers can't see:
Please push the review responses as fixup commits and let the commit-queue squash them. The 1. The PR description no longer describes the PR
The commit message says "Behavior-preserving" while the PR carries 2. Breaking changes2.1
|
| site | reached via | user-visible? |
|---|---|---|
writablestream.js:675 |
writerClosedPromise() → get closed() (:411) |
yes — writer.closed, when stream state is closed |
writablestream.js:701 |
writerReadyPromise() → get ready() (:435) |
yes — writer.ready, when writable and no backpressure |
writablestream.js:704 |
same | yes — writer.ready, when stream state is closed |
writablestream.js:240 |
get [kIsClosedPromise]() |
internal; only internal/streams/end-of-stream.js:374 reads it¹ |
readablestream.js:336 |
get [kIsClosedPromise]() |
internal; same¹ |
¹ kIsClosedPromise is SymbolFor('nodejs.webstream.isClosedPromise') — a registered symbol, so
stream[Symbol.for('nodejs.webstream.isClosedPromise')].promise reaches it from userland too. Not
public API, but not sealed either.
There are two directly reachable spec violations:
// 1. cross-stream identity leak — no closing or backpressure needed
const a = new WritableStream().getWriter();
const b = new WritableStream().getWriter();
a.ready === b.ready // true after this PR; false on main
// 2. two distinct internal slots collapse to one object
const w = /* writer on a closed WritableStream */;
w.closed === w.ready // true after this PR; false on mainPer spec [[closedPromise]] and [[readyPromise]] are separate slots, each initialised to "a new
promise", so (2) is unambiguously wrong. A third case exists but is narrower than I first wrote: a
writer.ready read before a backpressure cycle is === to one read after it only if the user
never observes ready during the backpressure window (otherwise the pending
PromiseWithResolvers() record is cached and reused). Still a deviation from
WritableStreamDefaultWriterEnsureReadyPromiseInitialized, just not unconditional.
WPT's aborting.any.js:43 / :1128 compare resolved-vs-pending and resolved-vs-rejected promises,
so neither catches this.
One more thing worth confirming (I have not executed it): kResolvedPromise is also the object the
implementation schedules on — PromisePrototypeThen(kResolvedPromise, pump) and friends. Handing
userland a reference to it means an own constructor property with a poisoned Symbol.species can be
installed on that specific object, and Promise.prototype.then does SpeciesConstructor(this, …)
before creating its result promise. This is not a new class of exposure — Node is already
susceptible to Promise.prototype.constructor poisoning for every internal promise — but it turns a
global-mutation attack into a targeted one that needs no global writes.
Suggested fix. The constraint is narrow: writerClosedPromise() and writerReadyPromise() back
two distinct spec slots, so they must hand out distinct promises. That doesn't require reverting the
whole helper. Two reasonable options:
- revert
resolvedRecord()topromise: PromiseResolve()— one line, restores main's behaviour; or - keep the shared instance but scope it to the two
[kIsClosedPromise]sites, which aren't
spec-observable slots, and mint a fresh promise in the writer getters.
I'd expect either to be unmeasurable. Every consumer is cold: the two public getters, [kInspect]
(writablestream.js:504-505), and one watchErrored() per pipeTo (readablestream.js:1819).
Notably writerReadyPromise() is not consulted per chunk any more — pipeTo parks via
parkOnReady() — so there's no per-chunk allocation being saved here. If there is a benchmark
showing otherwise I'd want to see it, but absent one this looks like it can go without argument.
Directly exposing the shared promise is not only breaking, it is potentially exploitable.
2.2 BLOCKER — two mutually inconsistent timing-compensation helpers
function promiseFromAlgorithmResult(result) { // 0 extra ticks
if (isNonThenable(result)) return kResolvedPromise;
return PromiseResolve(result);
}
function delayedAlgorithmResult(result) { // +1 tick vs. old
if (isNonThenable(result)) return kResolvedPromise;
return PromisePrototypeThen(kResolvedPromise, () => result);
}createPromiseCallback{NoParams,1Param,2Params} stopped being async. For a non-thenable user
return both helpers reproduce the old 1-tick settlement. For a thenable return they do not:
- old
async () => R:Ris adopted viaNewPromiseResolveThenableJob→ consumer reaction at tick 3. promiseFromAlgorithmResult(R)=PromiseResolve(R)= identity → consumer at tick 1
(2 ticks earlier).delayedAlgorithmResult(R)=kResolvedPromise.then(() => R)→ still adoptsR, plus one →
consumer at tick 4 (1 tick later).
Reachable, user-observable, on public API:
| call site | user callback | delta |
|---|---|---|
readableStreamDefaultControllerCancelSteps |
source.cancel() returns a promise |
stream.cancel() settles 2 ticks early |
readableByteStreamControllerCancelSteps |
same | 2 ticks early |
writableStreamDefaultControllerProcessClose |
sink.close() returns a promise |
2 ticks early |
WritableStreamDefaultController[kAbort] |
sink.abort() returns a promise |
2 ticks early |
transformStreamDefaultSink{Abort,Close}Algorithm, …SourceCancelAlgorithm |
transformer.cancel/flush |
1 tick late |
transformStreamDefaultControllerPerformTransform |
transformer.transform |
2 ticks early (return await raw vs return await asyncWrapper()) |
The comment on delayedAlgorithmResult gives the game away:
// Cancel/flush/abort only: insert an extra microtask so "upon fulfillment" of an already-settled
// user promise runs after start-settlement reactions queued during construction. Pull/write must not use this.
That's a compensation reverse-engineered from a failing test, not derived from the spec. Two helpers
that differ only by a microtask, applied per-call-site by hand, is exactly the kind of thing that
rots. Either (a) keep the async wrapper for the cold algorithms (cancel/close/abort/flush/transform
— these run once per stream, not per chunk, so there is no measurable win to give up) and only use
the raw-callback contract for pull/write, or (b) show the tick accounting for each of the six
sites above in a test. This is Mattias's r3798423135 point in a nutshell.
2.3 The write-side batching is the same deviation fillSync was removed for
writableStreamDefaultControllerDrainWriteQueue() (writablestream.js:1201) processes consecutive
sink writes in one microtask whenever the in-flight request is pipeTo's shared tracker:
if (stream[kState].inFlightWriteRequest.promise === null) {
writableStreamDefaultControllerCompleteWrite(controller);
continue; // <-- no microtask between writes
}Spec WritableStreamDefaultControllerProcessWrite ends with "Upon fulfillment of sinkWritePromise …
Perform WritableStreamDefaultControllerAdvanceQueueIfNeeded" — one microtask per write,
unconditionally. jasnell marked the readable-side equivalent semver-major and it was removed; the
writable-side equivalent survived and was never discussed. It is observable via the interleaving of
any concurrently-scheduled microtask, the settle position of pipeTo()'s promise, and fairness
between two concurrent pipes.
It's also a layering violation: writablestream.js now branches on promise === null, a sentinel
defined by the writeTracker object literal inside readableStreamPipeTo in
readablestream.js:1545. Nothing links the two but a comment.
2.4 cloneAsUint8Array changes the detached-buffer error
Old path threw V8's TypeError: Cannot perform ArrayBuffer.prototype.slice on a detached ArrayBuffer.
New path throws ERR_INVALID_STATE, which is mapped to Error, not TypeError
(src/node_errors.h:106). This is reachable: readableByteStreamTee's forwardChunk
(readablestream.js:2012) catches it and errors both branches with it, so it lands in user code as
the stream's stored error. If a byte tee is racing a buffer transfer, the observable error class
changes. Not covered by WPT or by the new test.
2.5 Lazy AbortController — looks correct, but check [kControllerErrorFunction]
(this[kState].abortController ??= new AbortController()) in both the getter (writablestream.js:547)
and writableStreamAbort (:724) is fine, and the abort-before-signal test covers the ordering. Note
WritableStream[kControllerErrorFunction] (writablestream.js:224) still does
this[kState].controller.error(error) unguarded — that's fine today because the writable always
materialises a controller, but it's now the odd one out versus the readable's new undefined guard.
3. Complexity vs. gain
3.1 The benchmarks measure the special cases this PR adds
benchmark/webstreams/creation.js times new ReadableStream() / new WritableStream() with no
arguments — a stream with no underlying source, which cannot produce data and has no real-world use.
The reported "creation: ReadableStream 1.49x, WritableStream 2.05x" is measuring the microbenchmark,
not a workload. The cost of that number is ensureEmptyDefaultController() plus guards in cancel(),
setupReadableStreamDefaultReader() and [kControllerErrorFunction], and a permanent obligation on
every future stream[kState].controller access to consider undefined.
Worse, it likely pessimizes the real path: createReadableStreamState() (readablestream.js:1439)
does not declare a controller field, so controller is added transitionally. Before this PR every
ReadableStream state object reached the same map before escaping the constructor. Now there are two
maps, and the hot read() fast path (readablestream.js:957 and :608) reads
stream[kState].controller — so those ICs go polymorphic as soon as a process mixes empty and
non-empty streams. If the empty-construction optimization is kept, createReadableStreamState()
should at minimum initialise controller: undefined.
benchmark/webstreams/pipe-to.js uses write(chunk, controller) {} — a fully synchronous sink. That
is precisely the shape §2.3's drain loop targets. Real sinks (fs, net, zlib, fetch) return promises and
take the thenable branch, getting zero benefit from the drain loop while paying its complexity
forever.
3.2 writableStreamDefaultControllerDrainWriteQueue duplicates two spec algorithms
It is a copy of writableStreamDefaultControllerAdvanceQueueIfNeeded (:1372) plus an inlined
writableStreamDefaultControllerProcessWrite (:1249), in a for(;;) with four returns and one
continue. Future spec fixes must now be applied in two places. Concrete problems in it:
- It reads
controllerState.writeFulfilled/writeRejectedbut never initialises them. It happens to
be safe because the only caller iswriteFulfilleditself — an undocumented invariant with no
assert. If anyone ever calls it from the start-completion path (which is what
advanceQueueIfNeededdoes),PromisePrototypeThen(kResolvedPromise, undefined, undefined)silently
swallows the write completion and the stream hangs. Please add
assert(controllerState.writeFulfilled !== undefined). - Line 1235 passes
writeRejectedas the rejection handler ofkResolvedPromise, which can never
reject. Dead argument, and inconsistent withthenAlgorithmResult, which deliberately omits it. - Line 1241 calls
thenAlgorithmResult(result, …)afterisNonThenable(result)has already returned
false — a redundant second check on the hot path. - The re-entrancy is subtle and undocumented:
completeWrite()→writableStreamUpdateBackpressure()
→writer[kState].ready.resolve()— and for pipeTo thatresolveispump, called
synchronously from inside the loop.pumpthen re-enterswritableStreamDefaultControllerWrite→
advanceQueueIfNeeded→processWrite, and the loop's next iteration bails because a request is
back in flight. I believe this is bounded and correct, but it needs a comment explaining why, and it
means thecontinuefires far less often than the design implies — which makes me want the
per-change benchmark attribution below even more.
3.3 The native binding is not justified
src/node_webstreams.cc exists for two functions.
isNonThenable replaces this inline JS:
result === null || (typeof result !== 'object' && typeof result !== 'function')TurboFan compiles that to a couple of map/instance-type checks. A Fast API call cannot beat it, and in
unoptimized code (Ignition/Sparkplug — i.e. startup, and any stream that never gets hot) it degrades to
a full C++ call, which is strictly slower than the JS it replaced. jasnell asked this directly
(r3780428766: "I don't understand why this needs to be a C++ function"); the answer given —
Kept the Fast API so the JIT can call the predicate without a slow C++/JS call on every chunk
— assumes the alternative is a C++ call. It isn't; the alternative is two inlined typeofs. Please
post an isolated microbenchmark of the predicate alone (JS inline vs. Fast API) before keeping this.
My expectation is it's a regression.
Also, the implementation is dead-code-y and not quite equivalent to the JS:
return value->IsNullOrUndefined() || (!value->IsObject() && !value->IsFunction());v8::Value::IsObject() is IsJSReceiver(), which is already true for functions — !value->IsObject()
suffices. And for an undetectable object (typeof x === 'undefined' but IsObject() true) the JS
predicate says "non-thenable" while the C++ says "maybe-thenable". Not reachable from Node userland
today, but it shows the two are not the same function.
cloneAsUint8Array is a more plausible win (one binding call instead of
ArrayBuffer.prototype.slice + new Uint8Array), but slice is already a fast V8 builtin doing a
memcpy, so the saving is one JS-level call and one intermediate object — on the byte-tee path only.
Given §2.4's error-behaviour change, I'd want a number for this specifically too.
Either way, adding a whole new internalBinding + node.gyp entry + external-reference registration +
typings for two predicates is a permanent cost. If isNonThenable really must be native, it belongs in
src/node_types.cc alongside the other fast type predicates rather than in a new webstreams binding.
3.4 Ask: per-change benchmark attribution
This PR bundles at least six independent optimizations behind one aggregate ratio. Given that #65138
already captured the shared parts, please split and measure each separately against current main:
- raw callbacks +
promiseFromAlgorithmResult/delayedAlgorithmResultfor cancel/close/abort/flush writableStreamDefaultControllerDrainWriteQueue- deferred
ReadableStreamcontroller - lazy writable
AbortController validateObjectskips (readable/writable/transform)- native
isNonThenable - native
cloneAsUint8Array
My prior is that (2) is the only one with a defensible number, (1) is the only one with a real
breaking-change cost, and (3)–(6) are benchmark-shaped. Items with no measurable win on current main
should be dropped — that alone would remove most of the complexity and most of the risk.
4. Test gaps
test/parallel/test-whatwg-webstreams-hotpath.js is decent on isNonThenable/Proxy and the subclass
cases, but nothing tests the risky behaviour:
- No test that
writer.ready !== writer.closed, or that two streams'readypromises are distinct
(would fail today — see §2.1). - No tick-ordering test for any of the six call sites in §2.2. These are exactly the cases WPT
under-covers. - No test for the pipeTo sync-write drain interleaving with an unrelated
queueMicrotask, nor for the
re-entrantpumppath in §3.2. cloneAsUint8Array: no detached-buffer,DataView, zero-length, resizable/length-tracking, or
Buffer(offset ≠ 0) case. The offset case matters — the old code sliced
[byteOffset, byteOffset+byteLength); the new code relies onCopyContents. A
Buffer.from(pool).subarray(k)test would pin that.(async () => {…})().then(common.mustCall())(×4) has no rejection handler; use
.then(common.mustCall(), common.mustNotCall())so a failure reports the actual error rather than a
mustCall miss.- The "each pull is separated by a microtask" test encodes an exact tick schedule via nested
queueMicrotask. That's the right intent, but it will break on unrelated changes; a comment saying
so would help the next person. - No doc changes. jasnell asked in r3780357430 for the non-standard behaviour to be documented;
doc/api/webstreams.mdis untouched.
5. Nits
node.gyp:177—src/node_webstreams.ccinserted betweennode_wasm_web_api.ccand
node_watchdog.cc, breaking the sort. Move afternode_watchdog.cc.- Import/export ordering:
promiseFromAlgorithmResultbetweenkResolvedPromiseandkState
(readablestream.js:112),isNonThenablebetweencloneAsUint8ArrayandcopyArrayBuffer
(util.js exports),delayedAlgorithmResultafternonOpFlush(transformstream.js:57), and the
promiseFromAlgorithmResult/delayedAlgorithmResultpair dropped into the middle of
module.exports(util.js:502-506) along with a removed blank line. transformstream.js:127-135— inconsistent braces (if (transformer !== kEmptyObject)unbraced, the
next two braced).transformstream.js:601-611/669-679— pure brace reformatting unrelated to the change; drop it
to keep the diff reviewable.THROW_ERR_INVALID_ARG_TYPE(env, "The \"view\" argument must be an ArrayBufferView")doesn't follow
the usualERR_INVALID_ARG_TYPEphrasing (must be an instance of X. Received …).src/node_webstreams.ccusesstd::unique_ptrwithout including<memory>.ReadableStream.prototype.cancel()callsensureEmptyDefaultController(this)unconditionally,
including when the stream is alreadyclosed/errored— wherereadableStreamCancelreturns before
touching the controller. That allocates a controller purely to throw it away, which contradicts the
stated goal.- jasnell's r3799526914 (braces in
[kControllerErrorFunction]) is still open, though the substance is
already there.
6. Recommendation
Request changes. Concretely, before this can be re-reviewed:
- Push review responses as fixup commits, not a squash.
- Re-run the full benchmark suite against current main (post-stream: cut promise churn in webstreams hot paths #65138) and publish all configs with
confidence intervals; restore thebenchmark.ymlbuild-cifix so CI can verify it. - Rewrite the PR body and commit message to describe what the code actually does, and either justify
or dropsemver-majorwith the §2 list in hand. - Make
writer.readyandwriter.closeddistinct promises again (§2.1) — reverting
resolvedRecord()to a freshPromiseResolve()is the one-line version; scoping the shared
instance to the internal[kIsClosedPromise]sites also works.
|
@jasnell do you think that this is still semver major or can we remove the label? |
|
Still semver-major |
|
@nodejs/tsc since this is semver major, it requires your review. |
|
@anonrig ... see the details in #65273 (comment) for a longer review. i'll try to do a line-by-line review later this week. |
ad88fa7 to
08914f5
Compare
| if (readable[kState].state === 'errored') | ||
| reject(readable[kState].storedError); | ||
| } else { | ||
| else { |
| if (writable[kState].state === 'errored') | ||
| reject(writable[kState].storedError); | ||
| } else { | ||
| else { |
|
Note: Bots are not permitted to attest with |
Avoid per-chunk async wrappers for sync pull/write/start and complete pipeTo writes without one microtask per chunk. Add a native webstreams binding with a Fast API isNonThenable check on the data plane and a memcpy clone for byte views. Empty stream construction skips redundant validation and lazily creates the writable AbortController, materializing it on abort() so controller.signal still reflects the abort reason. Use the shared kResolvedPromise on the pull/write hot path instead of allocating PromiseResolve(). Assisted-by: Grok Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com>
Stop sharing kResolvedPromise on writer.ready/closed and cancel(). Restore async wrappers for cancel/close/abort/flush/transform so thenable results keep the previous microtask count. Remove the write-queue drain loop so each write stays one microtask apart. Drop the native webstreams binding. isNonThenable and cloneAsUint8Array stay in JS so a detached buffer still throws TypeError. Initialize the deferred controller field and skip materializing it on cancel of a non-readable empty stream. Assisted-by: Grok Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com>
08914f5 to
cea5388
Compare
4 participants
Behavior-preserving performance work on
node:stream/web. Specorchestration and brand checks stay in JS; the per-chunk data plane
drops Promise/microtask churn for the common sync pull/write case
and gets a small native helper on the hot path.
asyncwrappers on sync source/sink algorithms.createPromiseCallback*now calls the user function and returnsthe raw result. Non-thenable results settle via
queueMicrotask(same position as
Promise.resolve().then) instead of allocatinga Promise per pull/write/start.
pipeTofills a default readable queue from sync pulls andstill batches already-queued chunks into the destination. Further
spec pull-fulfillment (tee, WPT) stays one pull per microtask.
the fulfillment turn. Regular
writer.write()keeps the specone-completion-per-microtask order.
internalBinding('webstreams'): Fast APIisNonThenable()on every pull/write/start result, andcloneAsUint8Array()as a single memcpy for byte-stream / teeclones.
new ReadableStream()/new WritableStream()skipvalidateObjecton the shared emptysentinels. The writable
AbortControlleris created lazily andmaterialized on
abort(), socontroller.signalobserved afterabort is still aborted with that reason.
Public constructors, methods, and WHATWG Streams behavior
(backpressure, BYOB, pipeTo, tee, errors, transfer) are unchanged.
Benchmarks
benchmark/compare.js --runs 10of the in-repowebstreams/suiteon the same machine, same
out/Release/nodefamily (pre-changebinary vs this tree). Rates are ops/sec.
Hot-path geometric mean of
new/oldacross all configs ofpipe-to.js,readable-read.js,readable-read-buffered.js,creation.js,readable-async-iterator.js, andtee.js:1.94x (32 configs). Full suite including
js_transfer.js:1.84x (35 configs). No config has
mean(new)/mean(old) < 1.0(min 1.03 on
js_transferReadableStream).Tests
test/wpt/test-streams.jstest/parallel/test-whatwg-readable*,writable*,transform*,webstreams*,test-webstreams*,test-global-webstreams.jstest/parallel/test-whatwg-webstreams-hotpath.js(publicread()/pipeTo, native helpers, abort-before-signal)test-whatwg-writablestream.jsAI assistance
This change was developed with assistance from Grok. I reviewed,
tested, and take responsibility for the submitted code.