From 9f69ca14e1148647250bb924c54969ebe5762b21 Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Thu, 13 Aug 2026 19:23:47 -0400 Subject: [PATCH 1/3] stream: speed up WHATWG web streams Avoid per-chunk async wrappers for sync pull/write/start, fill default readable queues in pipeTo, 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. Assisted-by: Grok Signed-off-by: Yagiz Nizipli --- lib/internal/webstreams/readablestream.js | 84 ++++++++++- lib/internal/webstreams/transformstream.js | 31 ++-- lib/internal/webstreams/util.js | 74 +++++++-- lib/internal/webstreams/writablestream.js | 104 ++++++++++--- node.gyp | 1 + src/node_binding.cc | 1 + src/node_external_reference.h | 1 + src/node_webstreams.cc | 106 +++++++++++++ .../test-whatwg-webstreams-hotpath.js | 142 ++++++++++++++++++ test/parallel/test-whatwg-writablestream.js | 14 ++ 10 files changed, 504 insertions(+), 54 deletions(-) create mode 100644 src/node_webstreams.cc create mode 100644 test/parallel/test-whatwg-webstreams-hotpath.js diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index b05cb7c257eb..3c479f638048 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -109,7 +109,9 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, kEmptyQueue, + promiseFromAlgorithmResult, kState, kType, lazyTransfer, @@ -250,6 +252,22 @@ class ReadableStream { */ constructor(source = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + // The empty-argument constructor is the creation.js / `new + // ReadableStream()` hot path: skip validateObject and strategy/source + // extraction when both arguments are the shared default sentinel. + if (source === kEmptyObject && strategy === kEmptyObject) { + this[kState] = createReadableStreamState(); + setupReadableStreamDefaultController( + this, + // eslint-disable-next-line no-use-before-define + new ReadableStreamDefaultController(kSkipThrow), + nonOpStart, + nonOpPull, + nonOpCancel, + 1, + defaultSizeAlgorithm); + return; + } validateObject(source, 'source', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); this[kState] = createReadableStreamState(); @@ -1689,6 +1707,11 @@ function readableStreamPipeTo( const controller = source[kState].controller; + if (source[kState].state === 'readable' && + isReadableStreamDefaultController(controller)) { + readableStreamDefaultControllerFillSync(controller); + } + // Fast path: batch reads when data is buffered in a default controller. // This avoids parking read requests and reduces promise allocation // overhead. @@ -2699,12 +2722,54 @@ function readableStreamDefaultControllerPull(controller) { controller[kState].pullRejected = (error) => readableStreamDefaultControllerError(controller, error); } + const result = controller[kState].pullAlgorithm(controller); + if (isNonThenable(result)) { + queueMicrotask(controller[kState].pullFulfilled); + return; + } PromisePrototypeThen( - controller[kState].pullAlgorithm(controller), + promiseFromAlgorithmResult(result), controller[kState].pullFulfilled, controller[kState].pullRejected); } +// pipeTo-only: fill a default controller's queue by repeatedly invoking a +// synchronous pull. Not used by the spec pull-fulfillment path (tee and +// WPT require one pull per microtask there). +function readableStreamDefaultControllerFillSync(controller) { + const state = controller[kState]; + if (state.pulling || state.pullAlgorithm === undefined) + return; + if (state.pullFulfilled === undefined) { + state.pullFulfilled = () => { + state.pulling = false; + if (state.pullAgain) { + state.pullAgain = false; + readableStreamDefaultControllerCallPullIfNeeded(controller); + } + }; + state.pullRejected = + (error) => readableStreamDefaultControllerError(controller, error); + } + while (readableStreamDefaultControllerShouldCallPull(controller)) { + state.pulling = true; + const before = state.queue.length; + const result = state.pullAlgorithm(controller); + if (isNonThenable(result)) { + state.pulling = false; + state.pullAgain = false; + if (state.queue.length === before && !state.closeRequested) + return; + continue; + } + PromisePrototypeThen( + promiseFromAlgorithmResult(result), + state.pullFulfilled, + state.pullRejected); + return; + } +} + function readableStreamDefaultControllerClearAlgorithms(controller) { controller[kState].pullAlgorithm = undefined; controller[kState].cancelAlgorithm = undefined; @@ -2726,7 +2791,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) { resetQueue(controller); const result = controller[kState].cancelAlgorithm(reason); readableStreamDefaultControllerClearAlgorithms(controller); - return result; + return promiseFromAlgorithmResult(result); } function readableStreamDefaultControllerPullSteps(controller, readRequest) { @@ -2787,8 +2852,7 @@ function setupReadableStreamDefaultController( const startResult = startAlgorithm(); - if (startResult === null || - (typeof startResult !== 'object' && typeof startResult !== 'function')) { + if (isNonThenable(startResult)) { // Non-thenable start result: fulfillment is guaranteed and no .then // lookup on the result is observable, so run the post-start step // directly at the exact microtask position the promise reaction @@ -3519,8 +3583,13 @@ function readableByteStreamControllerCallPullIfNeeded(controller) { controller[kState].pullRejected = (error) => readableByteStreamControllerError(controller, error); } + const result = controller[kState].pullAlgorithm(controller); + if (isNonThenable(result)) { + queueMicrotask(controller[kState].pullFulfilled); + return; + } PromisePrototypeThen( - controller[kState].pullAlgorithm(controller), + promiseFromAlgorithmResult(result), controller[kState].pullFulfilled, controller[kState].pullRejected); } @@ -3542,7 +3611,7 @@ function readableByteStreamControllerCancelSteps(controller, reason) { resetQueue(controller); const result = controller[kState].cancelAlgorithm(reason); readableByteStreamControllerClearAlgorithms(controller); - return result; + return promiseFromAlgorithmResult(result); } // Dequeues the first chunk of the byte queue as a Uint8Array view, @@ -3664,8 +3733,7 @@ function setupReadableByteStreamController( const startResult = startAlgorithm(); - if (startResult === null || - (typeof startResult !== 'object' && typeof startResult !== 'function')) { + if (isNonThenable(startResult)) { // See setupReadableStreamDefaultController. queueMicrotask(() => { controller[kState].started = true; diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 535c783a3a31..4ed55cd599f2 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -54,6 +54,7 @@ const { kType, nonOpCancel, nonOpFlush, + delayedAlgorithmResult, } = require('internal/webstreams/util'); const { @@ -123,9 +124,14 @@ class TransformStream { writableStrategy = kEmptyObject, readableStrategy = kEmptyObject) { markTransferMode(this, false, true); - validateObject(transformer, 'transformer', kValidateObjectAllowObjects); - validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull); - validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull); + if (transformer !== kEmptyObject) + validateObject(transformer, 'transformer', kValidateObjectAllowObjects); + if (writableStrategy !== kEmptyObject) { + validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull); + } + if (readableStrategy !== kEmptyObject) { + validateObject(readableStrategy, 'readableStrategy', kValidateObjectAllowObjectsAndNull); + } const readableType = transformer?.readableType; const writableType = transformer?.writableType; const start = transformer?.start; @@ -348,7 +354,7 @@ const isTransformStream = const isTransformStreamDefaultController = isBrandCheck('TransformStreamDefaultController'); -async function defaultTransformAlgorithm(chunk, controller) { +function defaultTransformAlgorithm(chunk, controller) { transformStreamDefaultControllerEnqueue(controller, chunk); } @@ -589,15 +595,16 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) { const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const cancelPromise = controller[kState].cancelAlgorithm(reason); + const cancelPromise = + delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason)); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( cancelPromise, () => { - if (readable[kState].state === 'errored') + if (readable[kState].state === 'errored') { reject(readable[kState].storedError); - else { + } else { readableStreamDefaultControllerError(readable[kState].controller, reason); resolve(); } @@ -622,7 +629,8 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) { } const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const flushPromise = controller[kState].flushAlgorithm(controller); + const flushPromise = + delayedAlgorithmResult(controller[kState].flushAlgorithm(controller)); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( flushPromise, @@ -659,15 +667,16 @@ function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const cancelPromise = controller[kState].cancelAlgorithm(reason); + const cancelPromise = + delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason)); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( cancelPromise, () => { - if (writable[kState].state === 'errored') + if (writable[kState].state === 'errored') { reject(writable[kState].storedError); - else { + } else { writableStreamDefaultControllerErrorIfNeeded( writable[kState].controller, reason); diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 8bc4c02be31e..67163c8fbe4c 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -4,7 +4,6 @@ const { Array, ArrayBufferPrototypeGetByteLength, ArrayBufferPrototypeGetDetached, - ArrayBufferPrototypeSlice, AsyncIteratorPrototype, DataViewPrototypeGetBuffer, DataViewPrototypeGetByteLength, @@ -20,7 +19,6 @@ const { TypedArrayPrototypeGetBuffer, TypedArrayPrototypeGetByteLength, TypedArrayPrototypeGetByteOffset, - Uint8Array, } = primordials; const { @@ -33,6 +31,11 @@ const { copyArrayBuffer, } = internalBinding('buffer'); +const { + isNonThenable, + cloneAsUint8Array: nativeCloneAsUint8Array, +} = internalBinding('webstreams'); + const { inspect, } = require('util'); @@ -128,12 +131,7 @@ function ArrayBufferViewGetByteOffset(view) { } function cloneAsUint8Array(view) { - const buffer = ArrayBufferViewGetBuffer(view); - const byteOffset = ArrayBufferViewGetByteOffset(view); - const byteLength = ArrayBufferViewGetByteLength(view); - return new Uint8Array( - ArrayBufferPrototypeSlice(buffer, byteOffset, byteOffset + byteLength), - ); + return nativeCloneAsUint8Array(view); } function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) { @@ -333,19 +331,43 @@ function enqueueValueWithSize(controller, value, size) { // each known call-site arity gets its own wrapper. The exact number of // arguments passed through to the user callback is observable and must be // preserved. +// +// These are intentionally not `async` functions. An `async` wrapper always +// allocates a Promise even when the user callback is synchronous and +// returns a non-thenable; callers use `isNonThenable()` (or +// `PromisePrototypeThen` for thenables) to settle the result, which matches +// the spec's promise-returning conversion without the extra allocation. function createPromiseCallbackNoParams(name, fn, thisArg) { validateFunction(fn, name); - return async () => FunctionPrototypeCall(fn, thisArg); + return () => { + try { + return FunctionPrototypeCall(fn, thisArg); + } catch (error) { + return PromiseReject(error); + } + }; } function createPromiseCallback1Param(name, fn, thisArg) { validateFunction(fn, name); - return async (arg) => FunctionPrototypeCall(fn, thisArg, arg); + return (arg) => { + try { + return FunctionPrototypeCall(fn, thisArg, arg); + } catch (error) { + return PromiseReject(error); + } + }; } function createPromiseCallback2Params(name, fn, thisArg) { validateFunction(fn, name); - return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2); + return (arg1, arg2) => { + try { + return FunctionPrototypeCall(fn, thisArg, arg1, arg2); + } catch (error) { + return PromiseReject(error); + } + }; } function isPromisePending(promise) { @@ -354,6 +376,25 @@ function isPromisePending(promise) { return details?.[0] === kPending; } +// Convert a promise-returning algorithm's raw result into a Promise. A +// non-thenable (the common sync-callback case) becomes the shared +// resolved promise; a user thenable is wrapped so Promise.prototype.then +// can be called on it. +function promiseFromAlgorithmResult(result) { + if (isNonThenable(result)) + return PromiseResolve(); + return PromiseResolve(result); +} + +// 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. +function delayedAlgorithmResult(result) { + if (isNonThenable(result)) + return PromiseResolve(); + return PromisePrototypeThen(PromiseResolve(), () => result); +} + // Shared shapes for lazily-materialized { promise, resolve, reject } // records whose settlement is already known. function resolvedRecord() { @@ -382,15 +423,15 @@ function setPromiseHandled(promise) { PromisePrototypeThen(promise, undefined, () => {}); } -async function nonOpFlush() {} +function nonOpFlush() {} function nonOpStart() {} -async function nonOpPull() {} +function nonOpPull() {} -async function nonOpCancel() {} +function nonOpCancel() {} -async function nonOpWrite() {} +function nonOpWrite() {} let transfer; function lazyTransfer() { @@ -407,6 +448,7 @@ module.exports = { Queue, canCopyArrayBuffer, cloneAsUint8Array, + isNonThenable, copyArrayBuffer, createPromiseCallbackNoParams, createPromiseCallback1Param, @@ -427,6 +469,8 @@ module.exports = { materializeQueue, nonOpCancel, nonOpFlush, + promiseFromAlgorithmResult, + delayedAlgorithmResult, nonOpPull, nonOpStart, nonOpWrite, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 10b7dbcf277c..c35108e0a7a3 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -65,6 +65,7 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, isPromisePending, kEmptyQueue, kState, @@ -74,6 +75,7 @@ const { nonOpStart, nonOpWrite, peekQueueValue, + promiseFromAlgorithmResult, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -183,6 +185,15 @@ class WritableStream { */ constructor(sink = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + if (sink === kEmptyObject && strategy === kEmptyObject) { + this[kState] = createWritableStreamState(); + setupWritableStreamDefaultControllerFromSink( + this, + sink, + 1, + defaultSizeAlgorithm); + return; + } validateObject(sink, 'sink', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); const type = sink?.type; @@ -519,7 +530,7 @@ class WritableStreamDefaultController { [kAbort](reason) { const result = this[kState].abortAlgorithm(reason); writableStreamDefaultControllerClearAlgorithms(this); - return result; + return promiseFromAlgorithmResult(result); } [kError]() { @@ -532,7 +543,7 @@ class WritableStreamDefaultController { get signal() { if (!isWritableStreamDefaultController(this)) throw new ERR_INVALID_THIS('WritableStreamDefaultController'); - return this[kState].abortController.signal; + return (this[kState].abortController ??= new AbortController()).signal; } /** @@ -707,7 +718,9 @@ function writableStreamAbort(stream, reason) { if (state === 'closed' || state === 'errored') return PromiseResolve(); - controller[kState].abortController.abort(reason); + // Materialize lazily so construction stays cheap, but abort() must + // still abort the same signal later observed via controller.signal. + (controller[kState].abortController ??= new AbortController()).abort(reason); state = stream[kState].state; if (state === 'closed' || state === 'errored') @@ -1165,6 +1178,61 @@ function writableStreamDefaultControllerWrite(controller, chunk, chunkSize) { writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); } +function writableStreamDefaultControllerCompleteWrite(controller) { + const stream = controller[kState].stream; + writableStreamFinishInFlightWrite(stream); + const streamState = stream[kState]; + const { + state, + } = streamState; + assert(state === 'writable' || state === 'erroring'); + dequeueValue(controller); + if (!streamState.closeQueuedOrInFlight && + state === 'writable') { + writableStreamUpdateBackpressure(controller, streamState); + } +} + +function writableStreamDefaultControllerDrainWriteQueue(controller) { + const controllerState = controller[kState]; + const stream = controllerState.stream; + for (;;) { + if (!controllerState.started || + stream[kState].inFlightWriteRequest.promise !== undefined) + return; + if (stream[kState].state === 'erroring') { + writableStreamFinishErroring(stream); + return; + } + if (!controllerState.queue.length) + return; + const value = peekQueueValue(controller); + if (value === kCloseSentinel) { + writableStreamDefaultControllerProcessClose(controller); + return; + } + writableStreamMarkFirstWriteRequestInFlight(stream); + const result = controllerState.writeAlgorithm(value, controller); + if (isNonThenable(result)) { + // pipeTo's shared write tracker uses `promise: null` and has no + // per-write then-callback that must interleave with the next sink + // write. Regular writer.write() requests carry a real Promise and + // must keep the spec's one-completion-per-microtask order. + if (stream[kState].inFlightWriteRequest.promise === null) { + writableStreamDefaultControllerCompleteWrite(controller); + continue; + } + queueMicrotask(controllerState.writeFulfilled); + return; + } + PromisePrototypeThen( + promiseFromAlgorithmResult(result), + controllerState.writeFulfilled, + controllerState.writeRejected); + return; + } +} + function writableStreamDefaultControllerProcessWrite(controller, chunk) { const { stream, @@ -1177,18 +1245,10 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { // so they are created once on the first write and reused for every // subsequent write instead of allocating two fresh closures per chunk. controller[kState].writeFulfilled = () => { - writableStreamFinishInFlightWrite(stream); - const streamState = stream[kState]; - const { - state, - } = streamState; - assert(state === 'writable' || state === 'erroring'); - dequeueValue(controller); - if (!streamState.closeQueuedOrInFlight && - state === 'writable') { - writableStreamUpdateBackpressure(controller, streamState); - } - writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); + writableStreamDefaultControllerCompleteWrite(controller); + // Already in the spec's "upon fulfillment" turn: drain further + // synchronous writes here instead of one-write-per-microtask. + writableStreamDefaultControllerDrainWriteQueue(controller); }; controller[kState].writeRejected = (error) => { if (stream[kState].state === 'writable') @@ -1197,8 +1257,13 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { }; } + const result = writeAlgorithm(chunk, controller); + if (isNonThenable(result)) { + queueMicrotask(controller[kState].writeFulfilled); + return; + } PromisePrototypeThen( - writeAlgorithm(chunk, controller), + promiseFromAlgorithmResult(result), controller[kState].writeFulfilled, controller[kState].writeRejected); } @@ -1212,7 +1277,7 @@ function writableStreamDefaultControllerProcessClose(controller) { writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); assert(!queue.length); - const sinkClosePromise = closeAlgorithm(); + const sinkClosePromise = promiseFromAlgorithmResult(closeAlgorithm()); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( sinkClosePromise, @@ -1359,7 +1424,7 @@ function setupWritableStreamDefaultController( highWaterMark, queue: kEmptyQueue, queueTotalSize: 0, - abortController: new AbortController(), + abortController: undefined, sizeAlgorithm, started: false, stream, @@ -1373,8 +1438,7 @@ function setupWritableStreamDefaultController( const startResult = startAlgorithm(); - if (startResult === null || - (typeof startResult !== 'object' && typeof startResult !== 'function')) { + if (isNonThenable(startResult)) { // Non-thenable start result: fulfillment is guaranteed and no .then // lookup on the result is observable, so run the post-start step // directly at the exact microtask position the promise reaction diff --git a/node.gyp b/node.gyp index 4f7a3d1ff634..ff00f73476a9 100644 --- a/node.gyp +++ b/node.gyp @@ -174,6 +174,7 @@ 'src/node_v8.cc', 'src/node_wasi.cc', 'src/node_wasm_web_api.cc', + 'src/node_webstreams.cc', 'src/node_watchdog.cc', 'src/node_worker.cc', 'src/node_zlib.cc', diff --git a/src/node_binding.cc b/src/node_binding.cc index 330c7f167105..48ce8d86ac1e 100644 --- a/src/node_binding.cc +++ b/src/node_binding.cc @@ -99,6 +99,7 @@ V(wasi) \ V(wasm_web_api) \ V(watchdog) \ + V(webstreams) \ V(worker) \ V(zlib) diff --git a/src/node_external_reference.h b/src/node_external_reference.h index 1e987ce2d4f3..89be54a19cc0 100644 --- a/src/node_external_reference.h +++ b/src/node_external_reference.h @@ -119,6 +119,7 @@ class ExternalReferenceRegistry { V(v8) \ V(zlib) \ V(wasm_web_api) \ + V(webstreams) \ V(worker) #if NODE_HAVE_I18N_SUPPORT diff --git a/src/node_webstreams.cc b/src/node_webstreams.cc new file mode 100644 index 000000000000..92547311cdba --- /dev/null +++ b/src/node_webstreams.cc @@ -0,0 +1,106 @@ +#include "env-inl.h" +#include "node.h" +#include "node_debug.h" +#include "node_errors.h" +#include "node_external_reference.h" + +using v8::ArrayBuffer; +using v8::ArrayBufferView; +using v8::BackingStore; +using v8::BackingStoreInitializationMode; +using v8::BackingStoreOnFailureMode; +using v8::CFunction; +using v8::Context; +using v8::FunctionCallbackInfo; +using v8::Isolate; +using v8::Local; +using v8::Object; +using v8::Uint8Array; +using v8::Value; + +namespace node { +namespace webstreams { + +// A value is a thenable only if it is a non-null object or a function +// (the spec's GetV(result, "then") is observable on those). Primitives, +// null, and undefined can never be thenables, so pull/write/start +// algorithms that return them can be settled without allocating a +// promise or looking up `.then`. +static bool IsNonThenableValue(Local value) { + return value->IsNullOrUndefined() || + (!value->IsObject() && !value->IsFunction()); +} + +static void IsNonThenable(const FunctionCallbackInfo& args) { + args.GetReturnValue().Set(IsNonThenableValue(args[0])); +} + +static bool FastIsNonThenable(Local unused, Local value) { + TRACK_V8_FAST_API_CALL("webstreams.isNonThenable"); + return IsNonThenableValue(value); +} + +static CFunction fast_is_non_thenable(CFunction::Make(FastIsNonThenable)); + +// Clone an ArrayBufferView into a fresh Uint8Array. Used by the +// byte-stream / tee paths in place of ArrayBuffer.prototype.slice + +// `new Uint8Array`, so the copy is a single memcpy. +static void CloneAsUint8Array(const FunctionCallbackInfo& args) { + Environment* env = Environment::GetCurrent(args); + Isolate* isolate = env->isolate(); + if (!args[0]->IsArrayBufferView()) { + THROW_ERR_INVALID_ARG_TYPE( + env, "The \"view\" argument must be an ArrayBufferView"); + return; + } + + Local view = args[0].As(); + Local source = view->Buffer(); + if (source->WasDetached()) { + THROW_ERR_INVALID_STATE(env, "Cannot clone a detached ArrayBuffer"); + return; + } + + const size_t byte_length = view->ByteLength(); + std::unique_ptr store = ArrayBuffer::NewBackingStore( + isolate, + byte_length, + BackingStoreInitializationMode::kUninitialized, + BackingStoreOnFailureMode::kReturnNull); + if (!store) { + THROW_ERR_MEMORY_ALLOCATION_FAILED(isolate); + return; + } + + if (byte_length > 0) { + view->CopyContents(store->Data(), byte_length); + } + + Local ab = ArrayBuffer::New(isolate, std::move(store)); + args.GetReturnValue().Set(Uint8Array::New(ab, 0, byte_length)); +} + +static void Initialize(Local target, + Local unused, + Local context, + void* priv) { + SetFastMethodNoSideEffect(context, + target, + "isNonThenable", + IsNonThenable, + &fast_is_non_thenable); + SetMethod(context, target, "cloneAsUint8Array", CloneAsUint8Array); +} + +static void RegisterExternalReferences(ExternalReferenceRegistry* registry) { + registry->Register(IsNonThenable); + registry->Register(fast_is_non_thenable); + registry->Register(CloneAsUint8Array); +} + +} // namespace webstreams +} // namespace node + +NODE_BINDING_CONTEXT_AWARE_INTERNAL(webstreams, node::webstreams::Initialize) +NODE_BINDING_EXTERNAL_REFERENCE(webstreams, + node::webstreams::RegisterExternalReferences) diff --git a/test/parallel/test-whatwg-webstreams-hotpath.js b/test/parallel/test-whatwg-webstreams-hotpath.js new file mode 100644 index 000000000000..98249c7a4041 --- /dev/null +++ b/test/parallel/test-whatwg-webstreams-hotpath.js @@ -0,0 +1,142 @@ +// Flags: --expose-internals --no-warnings +'use strict'; + +const common = require('../common'); +const assert = require('assert'); +const { + ReadableStream, + WritableStream, +} = require('node:stream/web'); +const { internalBinding } = require('internal/test/binding'); +const { + isNonThenable, + cloneAsUint8Array, +} = internalBinding('webstreams'); + +// The native helpers must be the ones the JS implementation actually calls. +assert.strictEqual(typeof isNonThenable, 'function'); +assert.strictEqual(typeof cloneAsUint8Array, 'function'); + +assert.strictEqual(isNonThenable(undefined), true); +assert.strictEqual(isNonThenable(null), true); +assert.strictEqual(isNonThenable(1), true); +assert.strictEqual(isNonThenable('x'), true); +assert.strictEqual(isNonThenable(true), true); +assert.strictEqual(isNonThenable({}), false); +assert.strictEqual(isNonThenable(() => {}), false); +assert.strictEqual(isNonThenable(Promise.resolve()), false); + +{ + const src = new Uint8Array([1, 2, 3, 4]); + const cloned = cloneAsUint8Array(src); + assert.ok(cloned instanceof Uint8Array); + assert.deepStrictEqual([...cloned], [1, 2, 3, 4]); + src[0] = 9; + assert.strictEqual(cloned[0], 1); +} + +{ + assert.throws(() => cloneAsUint8Array(1), { + code: 'ERR_INVALID_ARG_TYPE', + }); +} + +// Public API: pull-driven ReadableStream + read(). +(async () => { + const rs = new ReadableStream({ + start(controller) { + controller.enqueue('a'); + controller.enqueue('b'); + controller.close(); + }, + }); + const reader = rs.getReader(); + { + const { value, done } = await reader.read(); + assert.strictEqual(value, 'a'); + assert.strictEqual(done, false); + } + { + const { value, done } = await reader.read(); + assert.strictEqual(value, 'b'); + assert.strictEqual(done, false); + } + { + const { value, done } = await reader.read(); + assert.strictEqual(value, undefined); + assert.strictEqual(done, true); + } +})().then(common.mustCall()); + +// Public API: pipeTo with a sync sink — the optimized write drain path. +(async () => { + const expected = []; + const received = []; + const rs = new ReadableStream({ + start(controller) { + for (let i = 0; i < 32; i++) { + expected.push(i); + controller.enqueue(i); + } + controller.close(); + }, + }); + await rs.pipeTo(new WritableStream({ + write(chunk) { + received.push(chunk); + }, + })); + assert.deepStrictEqual(received, expected); +})().then(common.mustCall()); + +// pipeTo of a sync pull source must still fill and deliver every chunk +// (the pipeTo-only sync fill path). +{ + let calls = 0; + new ReadableStream({ + pull(controller) { + controller.enqueue(++calls); + }, + }, { + highWaterMark: 4, + }); + queueMicrotask(common.mustCall(() => { + // Spec path: one pull on start. Further pulls wait for fulfillment. + assert.strictEqual(calls, 1); + })); +} + +// pipeTo of a pull-driven source must deliver every chunk. +(async () => { + const n = 64; + let i = 0; + const received = []; + const rs = new ReadableStream({ + pull(controller) { + if (i < n) + controller.enqueue(i++); + else + controller.close(); + }, + }, { highWaterMark: 8 }); + await rs.pipeTo(new WritableStream({ + write(chunk) { + received.push(chunk); + }, + }, { highWaterMark: 8 })); + assert.strictEqual(received.length, n); + assert.deepStrictEqual(received, Array.from({ length: n }, (_, k) => k)); +})().then(common.mustCall()); + +{ + // Do not read controller.signal before abort(): the lazy AbortController + // must still report the abort reason on first access. + let ctrl; + const err = new Error('hotpath-abort-before-signal'); + const ws = new WritableStream({ + start(c) { ctrl = c; }, + }); + ws.abort(err); + assert.strictEqual(ctrl.signal.aborted, true); + assert.strictEqual(ctrl.signal.reason, err); +} diff --git a/test/parallel/test-whatwg-writablestream.js b/test/parallel/test-whatwg-writablestream.js index 88d9c57b9de7..66b181c8eb56 100644 --- a/test/parallel/test-whatwg-writablestream.js +++ b/test/parallel/test-whatwg-writablestream.js @@ -248,6 +248,20 @@ class Sink { }); } +{ + // abort() must abort the controller signal even if .signal was never + // observed before the abort (lazy AbortController materialization). + let ctrl; + const err = new Error('abort-before-signal'); + const ws = new WritableStream({ + start(c) { ctrl = c; }, + }); + assert.ok(ctrl); + ws.abort(err); + assert.strictEqual(ctrl.signal.aborted, true); + assert.strictEqual(ctrl.signal.reason, err); +} + { let controller; const writable = new WritableStream({ From c796760214cacf69ebe597f8d9b52f351253af10 Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Thu, 13 Aug 2026 19:23:47 -0400 Subject: [PATCH 2/3] typings: add webstreams internalBinding types Declare isNonThenable and cloneAsUint8Array on the new webstreams binding and register it in InternalBindingMap. Assisted-by: Grok Signed-off-by: Yagiz Nizipli --- typings/globals.d.ts | 2 ++ typings/internalBinding/webstreams.d.ts | 4 ++++ 2 files changed, 6 insertions(+) create mode 100644 typings/internalBinding/webstreams.d.ts diff --git a/typings/globals.d.ts b/typings/globals.d.ts index 536a8c4c2822..6a4464f18afa 100644 --- a/typings/globals.d.ts +++ b/typings/globals.d.ts @@ -34,6 +34,7 @@ import { URLPatternBinding } from "./internalBinding/url_pattern"; import { UtilBinding } from './internalBinding/util'; import { UVBinding } from './internalBinding/uv'; import { WASIBinding } from './internalBinding/wasi'; +import { WebstreamsBinding } from './internalBinding/webstreams'; import { WorkerBinding } from './internalBinding/worker'; import { ZlibBinding } from './internalBinding/zlib'; @@ -74,6 +75,7 @@ interface InternalBindingMap { util: UtilBinding; uv: UVBinding; wasi: WASIBinding; + webstreams: WebstreamsBinding; worker: WorkerBinding; zlib: ZlibBinding; } diff --git a/typings/internalBinding/webstreams.d.ts b/typings/internalBinding/webstreams.d.ts new file mode 100644 index 000000000000..814c4fb77481 --- /dev/null +++ b/typings/internalBinding/webstreams.d.ts @@ -0,0 +1,4 @@ +export interface WebstreamsBinding { + isNonThenable(value: unknown): boolean; + cloneAsUint8Array(view: ArrayBufferView): Uint8Array; +} From 1cf731985c3be802e752ff49e6cc645babd09cac Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Fri, 14 Aug 2026 10:06:00 -0400 Subject: [PATCH 3/3] stream: address webstreams review feedback Defer the default controller on new ReadableStream() until first use. Settle non-thenable pull/write with Promise.resolve().then so a throw in the fulfillment callback is an unhandled rejection, as on the promise path. Share a single no-op algorithm, format the native binding, and cover Proxy thenables. Re-run configure on the benchmark CI merge-commit build so new sources such as src/node_webstreams.cc are linked. Assisted-by: Grok Signed-off-by: Yagiz Nizipli --- .github/workflows/benchmark.yml | 2 +- lib/internal/webstreams/readablestream.js | 64 +++++++++++++++---- lib/internal/webstreams/util.js | 37 +++++------ lib/internal/webstreams/writablestream.js | 10 ++- src/node_webstreams.cc | 17 ++--- .../test-whatwg-webstreams-hotpath.js | 41 ++++++++++++ 6 files changed, 122 insertions(+), 49 deletions(-) diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 276cf738f8a6..561f8d56813f 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -141,7 +141,7 @@ jobs: --arg devTools '[]' \ --arg benchmarkTools '[]' \ --run ' - make -j4 V=1 + make build-ci -j4 V=1 ' - name: Run benchmark diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 3c479f638048..53b55959c992 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -252,20 +252,11 @@ class ReadableStream { */ constructor(source = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); - // The empty-argument constructor is the creation.js / `new - // ReadableStream()` hot path: skip validateObject and strategy/source - // extraction when both arguments are the shared default sentinel. + // Empty-argument `new ReadableStream()`: no source, no strategy, and + // no controller. Reads never deliver data, so skip those allocations + // until getReader/cancel/error first need a default controller. if (source === kEmptyObject && strategy === kEmptyObject) { this[kState] = createReadableStreamState(); - setupReadableStreamDefaultController( - this, - // eslint-disable-next-line no-use-before-define - new ReadableStreamDefaultController(kSkipThrow), - nonOpStart, - nonOpPull, - nonOpCancel, - 1, - defaultSizeAlgorithm); return; } validateObject(source, 'source', kValidateObjectAllowObjects); @@ -317,6 +308,11 @@ class ReadableStream { // only default controllers were wired here; byte stream controllers // keep the previous no-op behavior. const controller = this[kState].controller; + if (controller === undefined) { + if (this[kState].state === 'readable') + readableStreamError(this, error); + return; + } if (isReadableStreamDefaultController(controller)) controller.error(error); } @@ -367,6 +363,7 @@ class ReadableStream { return PromiseReject( new ERR_INVALID_STATE.TypeError('ReadableStream is locked')); } + ensureEmptyDefaultController(this); return readableStreamCancel(this, reason); } @@ -2576,6 +2573,7 @@ function setupReadableStreamBYOBReader(reader, stream) { function setupReadableStreamDefaultReader(reader, stream) { if (isReadableStreamLocked(stream)) throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked'); + ensureEmptyDefaultController(stream); readableStreamReaderGenericInitialize(reader, stream); reader[kState].readRequests = kEmptyQueue; } @@ -2722,9 +2720,19 @@ function readableStreamDefaultControllerPull(controller) { controller[kState].pullRejected = (error) => readableStreamDefaultControllerError(controller, error); } + // Non-thenable results use PromiseResolve().then so a throw in + // pullFulfilled becomes an unhandled rejection (same as the Then + // path below), not an uncaught exception from queueMicrotask. + // That reaction is one microtask, matching Then on a fulfilled + // async-wrapper promise. This is not a sync pull-to-pull hop; + // pipeTo's FillSync is the only path that skips the microtask + // between consecutive user pulls. const result = controller[kState].pullAlgorithm(controller); if (isNonThenable(result)) { - queueMicrotask(controller[kState].pullFulfilled); + PromisePrototypeThen( + PromiseResolve(), + controller[kState].pullFulfilled, + controller[kState].pullRejected); return; } PromisePrototypeThen( @@ -2824,6 +2832,31 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) { readableStreamDefaultControllerPull(controller); } +// 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) { + if (stream[kState].controller !== undefined) + return stream[kState].controller; + const controller = new ReadableStreamDefaultController(kSkipThrow); + controller[kState] = { + cancelAlgorithm: nonOpCancel, + closeRequested: false, + highWaterMark: 1, + pullAgain: false, + pullAlgorithm: nonOpPull, + pulling: false, + pullFulfilled: undefined, + pullRejected: undefined, + queue: kEmptyQueue, + queueTotalSize: 0, + started: true, + sizeAlgorithm: defaultSizeAlgorithm, + stream, + }; + stream[kState].controller = controller; + return controller; +} + function setupReadableStreamDefaultController( stream, controller, @@ -3585,7 +3618,10 @@ function readableByteStreamControllerCallPullIfNeeded(controller) { } const result = controller[kState].pullAlgorithm(controller); if (isNonThenable(result)) { - queueMicrotask(controller[kState].pullFulfilled); + PromisePrototypeThen( + PromiseResolve(), + controller[kState].pullFulfilled, + controller[kState].pullRejected); return; } PromisePrototypeThen( diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js index 67163c8fbe4c..b9570f1d5c7e 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -332,11 +332,10 @@ function enqueueValueWithSize(controller, value, size) { // arguments passed through to the user callback is observable and must be // preserved. // -// These are intentionally not `async` functions. An `async` wrapper always -// allocates a Promise even when the user callback is synchronous and -// returns a non-thenable; callers use `isNonThenable()` (or -// `PromisePrototypeThen` for thenables) to settle the result, which matches -// the spec's promise-returning conversion without the extra allocation. +// These are intentionally not `async` functions and not `Promise.try`. +// Both always allocate a Promise, even when the user callback is +// synchronous and returns a non-thenable. Callers use `isNonThenable()` +// (or `PromisePrototypeThen` for thenables) to settle the result. function createPromiseCallbackNoParams(name, fn, thisArg) { validateFunction(fn, name); return () => { @@ -377,9 +376,10 @@ function isPromisePending(promise) { } // Convert a promise-returning algorithm's raw result into a Promise. A -// non-thenable (the common sync-callback case) becomes the shared -// resolved promise; a user thenable is wrapped so Promise.prototype.then -// can be called on it. +// value that cannot be a thenable (null, undefined, or a non-object +// non-function primitive) becomes the shared resolved promise. Objects +// and functions go through PromiseResolve so a `.then` lookup, if any, +// stays observable. function promiseFromAlgorithmResult(result) { if (isNonThenable(result)) return PromiseResolve(); @@ -423,15 +423,7 @@ function setPromiseHandled(promise) { PromisePrototypeThen(promise, undefined, () => {}); } -function nonOpFlush() {} - -function nonOpStart() {} - -function nonOpPull() {} - -function nonOpCancel() {} - -function nonOpWrite() {} +function nonOp() {} let transfer; function lazyTransfer() { @@ -467,13 +459,14 @@ module.exports = { kType, lazyTransfer, materializeQueue, - nonOpCancel, - nonOpFlush, + nonOp, + nonOpCancel: nonOp, + nonOpFlush: nonOp, promiseFromAlgorithmResult, delayedAlgorithmResult, - nonOpPull, - nonOpStart, - nonOpWrite, + nonOpPull: nonOp, + nonOpStart: nonOp, + nonOpWrite: nonOp, peekQueueValue, rejectedHandledRecord, resetQueue, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index c35108e0a7a3..47d44ddf4947 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -1222,7 +1222,10 @@ function writableStreamDefaultControllerDrainWriteQueue(controller) { writableStreamDefaultControllerCompleteWrite(controller); continue; } - queueMicrotask(controllerState.writeFulfilled); + PromisePrototypeThen( + PromiseResolve(), + controllerState.writeFulfilled, + controllerState.writeRejected); return; } PromisePrototypeThen( @@ -1259,7 +1262,10 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { const result = writeAlgorithm(chunk, controller); if (isNonThenable(result)) { - queueMicrotask(controller[kState].writeFulfilled); + PromisePrototypeThen( + PromiseResolve(), + controller[kState].writeFulfilled, + controller[kState].writeRejected); return; } PromisePrototypeThen( diff --git a/src/node_webstreams.cc b/src/node_webstreams.cc index 92547311cdba..87d49724ed8d 100644 --- a/src/node_webstreams.cc +++ b/src/node_webstreams.cc @@ -21,11 +21,11 @@ using v8::Value; namespace node { namespace webstreams { -// A value is a thenable only if it is a non-null object or a function -// (the spec's GetV(result, "then") is observable on those). Primitives, -// null, and undefined can never be thenables, so pull/write/start -// algorithms that return them can be settled without allocating a -// promise or looking up `.then`. +// True when `value` cannot be a thenable: null, undefined, or a +// non-object non-function primitive. Objects and functions are treated +// as maybe-thenable without looking up `.then` (that lookup is +// observable). Proxies of objects/functions take the maybe-thenable +// path; a Proxy around a primitive is still an object. static bool IsNonThenableValue(Local value) { return value->IsNullOrUndefined() || (!value->IsObject() && !value->IsFunction()); @@ -84,11 +84,8 @@ static void Initialize(Local target, Local unused, Local context, void* priv) { - SetFastMethodNoSideEffect(context, - target, - "isNonThenable", - IsNonThenable, - &fast_is_non_thenable); + SetFastMethodNoSideEffect( + context, target, "isNonThenable", IsNonThenable, &fast_is_non_thenable); SetMethod(context, target, "cloneAsUint8Array", CloneAsUint8Array); } diff --git a/test/parallel/test-whatwg-webstreams-hotpath.js b/test/parallel/test-whatwg-webstreams-hotpath.js index 98249c7a4041..88efcf9b5511 100644 --- a/test/parallel/test-whatwg-webstreams-hotpath.js +++ b/test/parallel/test-whatwg-webstreams-hotpath.js @@ -25,6 +25,9 @@ assert.strictEqual(isNonThenable(true), true); assert.strictEqual(isNonThenable({}), false); assert.strictEqual(isNonThenable(() => {}), false); assert.strictEqual(isNonThenable(Promise.resolve()), false); +assert.strictEqual(isNonThenable(new Proxy({}, {})), false); +assert.strictEqual(isNonThenable(new Proxy(Object(1), {})), false); +assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); { const src = new Uint8Array([1, 2, 3, 4]); @@ -128,6 +131,44 @@ assert.strictEqual(isNonThenable(Promise.resolve()), false); assert.deepStrictEqual(received, Array.from({ length: n }, (_, k) => k)); })().then(common.mustCall()); +{ + // Empty-argument construction defers the controller. cancel() and + // getReader() must still work on the public API. + const rs = new ReadableStream(); + rs.cancel().then(common.mustCall()); +} + +{ + const rs = new ReadableStream(); + const reader = rs.getReader(); + reader.cancel().then(common.mustCall()); +} + +{ + // A Proxy around a thenable must not take the non-thenable shortcut. + let pulled = false; + const thenable = new Proxy({ + then(resolve) { + resolve(); + }, + }, {}); + const rs = new ReadableStream({ + pull(controller) { + if (pulled) { + controller.close(); + return thenable; + } + pulled = true; + controller.enqueue('proxied'); + return thenable; + }, + }); + rs.getReader().read().then(common.mustCall(({ value, done }) => { + assert.strictEqual(value, 'proxied'); + assert.strictEqual(done, false); + })); +} + { // Do not read controller.signal before abort(): the lazy AbortController // must still report the abort reason on first access.