From 5999da326d1fbcf1fe6236b55965fca992059882 Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Mon, 17 Aug 2026 23:14:53 +0000 Subject: [PATCH 1/2] stream: speed up WHATWG web streams 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 --- lib/internal/webstreams/readablestream.js | 71 +++++- lib/internal/webstreams/transformstream.js | 31 ++- lib/internal/webstreams/util.js | 89 +++++-- lib/internal/webstreams/writablestream.js | 109 ++++++-- node.gyp | 1 + src/node_binding.cc | 1 + src/node_external_reference.h | 1 + src/node_webstreams.cc | 103 ++++++++ .../test-whatwg-webstreams-hotpath.js | 237 ++++++++++++++++++ test/parallel/test-whatwg-writablestream.js | 14 ++ typings/globals.d.ts | 2 + typings/internalBinding/webstreams.d.ts | 4 + 12 files changed, 599 insertions(+), 64 deletions(-) create mode 100644 src/node_webstreams.cc create mode 100644 test/parallel/test-whatwg-webstreams-hotpath.js create mode 100644 typings/internalBinding/webstreams.d.ts diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index 84f16dc7b4a1..be2d5ff178ac 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -111,8 +111,10 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, kEmptyQueue, kResolvedPromise, + promiseFromAlgorithmResult, kState, kType, lazyTransfer, @@ -252,6 +254,16 @@ class ReadableStream { */ constructor(source = kEmptyObject, strategy = kEmptyObject) { markTransferMode(this, false, true); + // 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. + // Subclasses that call those methods after super() materialize the + // controller in the subclass constructor; see + // ensureEmptyDefaultController. + if (source === kEmptyObject && strategy === kEmptyObject) { + this[kState] = createReadableStreamState(); + return; + } validateObject(source, 'source', kValidateObjectAllowObjects); validateObject(strategy, 'strategy', kValidateObjectAllowObjectsAndNull); this[kState] = createReadableStreamState(); @@ -301,8 +313,13 @@ 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); + readableStreamDefaultControllerError(controller, error); } // Used by the internal stream interop (end-of-stream). Materialized @@ -351,6 +368,7 @@ class ReadableStream { return PromiseReject( new ERR_INVALID_STATE.TypeError('ReadableStream is locked')); } + ensureEmptyDefaultController(this); return readableStreamCancel(this, reason); } @@ -2580,6 +2598,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; } @@ -2729,7 +2748,8 @@ function readableStreamDefaultControllerPull(controller) { // The pull algorithm may be a raw callback (a wrapped user source.pull // returns its result uncoerced; a synchronous throw surfaces here) or an // internal algorithm that always returns a promise; thenAlgorithmResult - // handles both. + // handles both. Non-thenable results react on kResolvedPromise so each + // pull is still separated by a microtask, matching the spec. let result; try { result = controller[kState].pullAlgorithm(controller); @@ -2763,7 +2783,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) { resetQueue(controller); const result = controller[kState].cancelAlgorithm(reason); readableStreamDefaultControllerClearAlgorithms(controller); - return result; + return promiseFromAlgorithmResult(result); } function readableStreamDefaultControllerPullSteps(controller, readRequest) { @@ -2796,6 +2816,43 @@ function readableStreamDefaultControllerPullSteps(controller, readRequest) { readableStreamDefaultControllerPull(controller); } +// Materialize the deferred default controller for `new ReadableStream()`. +// +// started is true immediately: the empty-argument start algorithm is a +// no-op, so there is no initial pull and nothing can observe an unstarted +// controller without first calling getReader/cancel/pipeTo/tee/values, +// all of which come through here. That is also why this still matches +// WPT: those tests either pass a source (leaving this path) or wait for +// start, which is already complete for a no-op start. +// +// Subclasses that call cancel(), getReader(), pipeTo(), tee(), or +// values() in the constructor body after super() will materialize the +// controller before the subclass constructor finishes. Passing a source +// (for example to install start/pull) leaves the empty-argument path +// and creates the controller during super() as usual. +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: nonOpCallback, + 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, @@ -2824,8 +2881,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 @@ -3586,7 +3642,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, @@ -3708,8 +3764,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 05439a25dcb5..ab1360b746d9 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,9 +331,20 @@ 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 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 async () => FunctionPrototypeCall(fn, thisArg); + return () => { + try { + return FunctionPrototypeCall(fn, thisArg); + } catch (error) { + return PromiseReject(error); + } + }; } // Raw variants that skip the async wrapper's implicit result promise. @@ -364,8 +373,7 @@ const kResolvedPromise = PromiseResolve(); // matches the spec's "a promise resolved with" conversion (identity for // native promises). function thenAlgorithmResult(result, onFulfilled, onRejected) { - if (result === null || - (typeof result !== 'object' && typeof result !== 'function')) { + if (isNonThenable(result)) { PromisePrototypeThen(kResolvedPromise, onFulfilled); } else { PromisePrototypeThen(PromiseResolve(result), onFulfilled, onRejected); @@ -374,12 +382,24 @@ function thenAlgorithmResult(result, onFulfilled, onRejected) { 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) { @@ -388,11 +408,31 @@ function isPromisePending(promise) { return details?.[0] === kPending; } +// Convert a promise-returning algorithm's raw result into a Promise. A +// 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 kResolvedPromise; + 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 kResolvedPromise; + return PromisePrototypeThen(kResolvedPromise, () => result); +} + // Shared shapes for lazily-materialized { promise, resolve, reject } // records whose settlement is already known. function resolvedRecord() { return { - promise: PromiseResolve(), + promise: kResolvedPromise, resolve: undefined, reject: undefined, }; @@ -416,16 +456,13 @@ function setPromiseHandled(promise) { PromisePrototypeThen(promise, undefined, () => {}); } -async function nonOpFlush() {} - -// Shared non-op for the start/pull/write algorithm callbacks, which all -// follow the raw-callback contract (see createRawCallback*): the -// non-thenable return takes the allocation-free fast path in -// thenAlgorithmResult(). +// Shared no-op. Start/pull/write use the raw-callback contract (see +// createRawCallback*): a non-thenable return takes the allocation-free +// path in thenAlgorithmResult(). Cancel/flush/abort wrap the result +// with promiseFromAlgorithmResult/delayedAlgorithmResult, so a sync +// no-op is equivalent to the previous async empty functions. function nonOpCallback() {} -async function nonOpCancel() {} - let transfer; function lazyTransfer() { if (transfer === undefined) @@ -441,6 +478,7 @@ module.exports = { Queue, canCopyArrayBuffer, cloneAsUint8Array, + isNonThenable, copyArrayBuffer, createPromiseCallbackNoParams, createPromiseCallback1Param, @@ -463,9 +501,10 @@ module.exports = { lazyTransfer, materializeQueue, nonOpCallback, - nonOpCancel, - nonOpFlush, - + nonOpCancel: nonOpCallback, + nonOpFlush: nonOpCallback, + promiseFromAlgorithmResult, + delayedAlgorithmResult, peekQueueValue, rejectedHandledRecord, resetQueue, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index 1e9ca02cfe96..c406d116f07b 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -65,14 +65,17 @@ const { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, isPromisePending, kEmptyQueue, + kResolvedPromise, kState, kType, lazyTransfer, nonOpCallback, nonOpCancel, peekQueueValue, + promiseFromAlgorithmResult, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -183,6 +186,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 +531,7 @@ class WritableStreamDefaultController { [kAbort](reason) { const result = this[kState].abortAlgorithm(reason); writableStreamDefaultControllerClearAlgorithms(this); - return result; + return promiseFromAlgorithmResult(result); } [kError]() { @@ -532,7 +544,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 +719,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') @@ -1169,6 +1183,69 @@ 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); + let result; + try { + result = controllerState.writeAlgorithm(value, controller); + } catch (error) { + result = PromiseReject(error); + } + 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; + } + PromisePrototypeThen( + kResolvedPromise, + controllerState.writeFulfilled, + controllerState.writeRejected); + return; + } + thenAlgorithmResult( + result, + controllerState.writeFulfilled, + controllerState.writeRejected); + return; + } +} + function writableStreamDefaultControllerProcessWrite(controller, chunk) { const { stream, @@ -1181,18 +1258,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') @@ -1204,7 +1273,8 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { // The write algorithm may be a raw callback (a wrapped user sink.write // returns its result uncoerced; a synchronous throw surfaces here) or an // internal algorithm that always returns a promise; thenAlgorithmResult - // handles both. + // handles both. Non-thenable results react on kResolvedPromise so each + // writer.write() completion is still separated by a microtask. let result; try { result = writeAlgorithm(chunk, controller); @@ -1226,7 +1296,7 @@ function writableStreamDefaultControllerProcessClose(controller) { writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); assert(!queue.length); - const sinkClosePromise = closeAlgorithm(); + const sinkClosePromise = promiseFromAlgorithmResult(closeAlgorithm()); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( sinkClosePromise, @@ -1373,7 +1443,7 @@ function setupWritableStreamDefaultController( highWaterMark, queue: kEmptyQueue, queueTotalSize: 0, - abortController: new AbortController(), + abortController: undefined, sizeAlgorithm, started: false, stream, @@ -1387,8 +1457,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..87d49724ed8d --- /dev/null +++ b/src/node_webstreams.cc @@ -0,0 +1,103 @@ +#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 { + +// 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()); +} + +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..929c35cde166 --- /dev/null +++ b/test/parallel/test-whatwg-webstreams-hotpath.js @@ -0,0 +1,237 @@ +// 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); +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]); + 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()); + +// Spec path: each pull is separated by a microtask. Start schedules one +// pull; further pulls wait for that fulfillment and do not run in the +// same turn. +{ + let calls = 0; + new ReadableStream({ + pull(controller) { + controller.enqueue(++calls); + }, + }, { + highWaterMark: 4, + }); + queueMicrotask(common.mustCall(() => { + assert.strictEqual(calls, 1); + // The next pull is queued only after fulfillment, so it is not + // invoked in this same turn. + queueMicrotask(common.mustCall(() => { + assert.strictEqual(calls, 2); + })); + })); +} + +// 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()); + +{ + // 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()); +} + +{ + // Subclass that calls getReader() before the subclass constructor + // finishes: the deferred controller is materialized then. + class Sub extends ReadableStream { + constructor() { + super(); + this.reader = this.getReader(); + } + } + const rs = new Sub(); + assert.ok(rs.locked); + rs.reader.cancel().then(common.mustCall()); +} + +{ + // Subclass that calls cancel() before the subclass constructor finishes. + class Sub extends ReadableStream { + constructor() { + super(); + this.closed = this.cancel(); + } + } + const rs = new Sub(); + rs.closed.then(common.mustCall()); +} + +{ + // Passing a source leaves the empty-argument path, so start() receives + // a controller during super() even if the subclass constructor later + // calls getReader(). + let sawController = false; + class Sub extends ReadableStream { + constructor() { + super({ + start(controller) { + sawController = controller != null; + }, + }); + this.reader = this.getReader(); + } + } + const rs = new Sub(); + assert.ok(rs.locked); + queueMicrotask(common.mustCall(() => { + assert.strictEqual(sawController, true); + rs.reader.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. + 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({ diff --git a/typings/globals.d.ts b/typings/globals.d.ts index e64b8e6d89fc..13b947c3e285 100644 --- a/typings/globals.d.ts +++ b/typings/globals.d.ts @@ -38,6 +38,7 @@ import { UtilBinding } from './internalBinding/util'; import { UVBinding } from './internalBinding/uv'; import { WASIBinding } from './internalBinding/wasi'; import { WatchdogBinding } from './internalBinding/watchdog'; +import { WebstreamsBinding } from './internalBinding/webstreams'; import { WorkerBinding } from './internalBinding/worker'; import { ZlibBinding } from './internalBinding/zlib'; @@ -82,6 +83,7 @@ interface InternalBindingMap { uv: UVBinding; wasi: WASIBinding; watchdog: WatchdogBinding; + 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 cea5388bc13c358d6177e455120cea7f530aab2b Mon Sep 17 00:00:00 2001 From: Yagiz Nizipli Date: Tue, 18 Aug 2026 01:26:08 +0000 Subject: [PATCH 2/2] stream: address webstreams review blockers 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 --- lib/internal/webstreams/readablestream.js | 14 ++- lib/internal/webstreams/transformstream.js | 23 ++-- lib/internal/webstreams/util.js | 102 +++++++---------- lib/internal/webstreams/writablestream.js | 60 +--------- node.gyp | 1 - src/node_binding.cc | 1 - src/node_external_reference.h | 1 - src/node_webstreams.cc | 103 ------------------ .../test-whatwg-webstreams-hotpath.js | 101 ++++++++++++++--- typings/globals.d.ts | 2 - typings/internalBinding/webstreams.d.ts | 4 - 11 files changed, 147 insertions(+), 265 deletions(-) delete mode 100644 src/node_webstreams.cc delete mode 100644 typings/internalBinding/webstreams.d.ts diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index be2d5ff178ac..71317f9f8170 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -114,7 +114,6 @@ const { isNonThenable, kEmptyQueue, kResolvedPromise, - promiseFromAlgorithmResult, kState, kType, lazyTransfer, @@ -314,8 +313,9 @@ class ReadableStream { // keep the previous no-op behavior. const controller = this[kState].controller; if (controller === undefined) { - if (this[kState].state === 'readable') + if (this[kState].state === 'readable') { readableStreamError(this, error); + } return; } if (isReadableStreamDefaultController(controller)) @@ -368,7 +368,10 @@ class ReadableStream { return PromiseReject( new ERR_INVALID_STATE.TypeError('ReadableStream is locked')); } - ensureEmptyDefaultController(this); + // Only materialize the deferred empty controller when cancel will + // actually run cancel steps. closed/errored streams return immediately. + if (this[kState].state === 'readable') + ensureEmptyDefaultController(this); return readableStreamCancel(this, reason); } @@ -1440,6 +1443,7 @@ function createReadableStreamState() { return { __proto__: null, closedPromise: undefined, + controller: undefined, disturbed: false, reader: undefined, state: 'readable', @@ -2783,7 +2787,7 @@ function readableStreamDefaultControllerCancelSteps(controller, reason) { resetQueue(controller); const result = controller[kState].cancelAlgorithm(reason); readableStreamDefaultControllerClearAlgorithms(controller); - return promiseFromAlgorithmResult(result); + return result; } function readableStreamDefaultControllerPullSteps(controller, readRequest) { @@ -3642,7 +3646,7 @@ function readableByteStreamControllerCancelSteps(controller, reason) { resetQueue(controller); const result = controller[kState].cancelAlgorithm(reason); readableByteStreamControllerClearAlgorithms(controller); - return promiseFromAlgorithmResult(result); + return result; } // Dequeues the first chunk of the byte queue as a Uint8Array view, diff --git a/lib/internal/webstreams/transformstream.js b/lib/internal/webstreams/transformstream.js index 4ed55cd599f2..6cfe15ca7e85 100644 --- a/lib/internal/webstreams/transformstream.js +++ b/lib/internal/webstreams/transformstream.js @@ -54,7 +54,6 @@ const { kType, nonOpCancel, nonOpFlush, - delayedAlgorithmResult, } = require('internal/webstreams/util'); const { @@ -124,8 +123,9 @@ class TransformStream { writableStrategy = kEmptyObject, readableStrategy = kEmptyObject) { markTransferMode(this, false, true); - if (transformer !== kEmptyObject) + if (transformer !== kEmptyObject) { validateObject(transformer, 'transformer', kValidateObjectAllowObjects); + } if (writableStrategy !== kEmptyObject) { validateObject(writableStrategy, 'writableStrategy', kValidateObjectAllowObjectsAndNull); } @@ -354,7 +354,7 @@ const isTransformStream = const isTransformStreamDefaultController = isBrandCheck('TransformStreamDefaultController'); -function defaultTransformAlgorithm(chunk, controller) { +async function defaultTransformAlgorithm(chunk, controller) { transformStreamDefaultControllerEnqueue(controller, chunk); } @@ -595,16 +595,15 @@ async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) { const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const cancelPromise = - delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason)); + const cancelPromise = 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(); } @@ -629,8 +628,7 @@ function transformStreamDefaultSinkCloseAlgorithm(stream) { } const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const flushPromise = - delayedAlgorithmResult(controller[kState].flushAlgorithm(controller)); + const flushPromise = controller[kState].flushAlgorithm(controller); transformStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( flushPromise, @@ -667,16 +665,15 @@ function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { const { promise, resolve, reject } = PromiseWithResolvers(); controller[kState].finishPromise = promise; - const cancelPromise = - delayedAlgorithmResult(controller[kState].cancelAlgorithm(reason)); + const cancelPromise = 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 ab1360b746d9..8fba954bd52c 100644 --- a/lib/internal/webstreams/util.js +++ b/lib/internal/webstreams/util.js @@ -4,6 +4,7 @@ const { Array, ArrayBufferPrototypeGetByteLength, ArrayBufferPrototypeGetDetached, + ArrayBufferPrototypeSlice, AsyncIteratorPrototype, DataViewPrototypeGetBuffer, DataViewPrototypeGetByteLength, @@ -19,6 +20,7 @@ const { TypedArrayPrototypeGetBuffer, TypedArrayPrototypeGetByteLength, TypedArrayPrototypeGetByteOffset, + Uint8Array, } = primordials; const { @@ -31,11 +33,6 @@ const { copyArrayBuffer, } = internalBinding('buffer'); -const { - isNonThenable, - cloneAsUint8Array: nativeCloneAsUint8Array, -} = internalBinding('webstreams'); - const { inspect, } = require('util'); @@ -131,7 +128,22 @@ function ArrayBufferViewGetByteOffset(view) { } function cloneAsUint8Array(view) { - return nativeCloneAsUint8Array(view); + const buffer = ArrayBufferViewGetBuffer(view); + const byteOffset = ArrayBufferViewGetByteOffset(view); + const byteLength = ArrayBufferViewGetByteLength(view); + return new Uint8Array( + ArrayBufferPrototypeSlice(buffer, byteOffset, byteOffset + byteLength), + ); +} + +// 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. +function isNonThenable(value) { + return value === null || + (typeof value !== 'object' && typeof value !== 'function'); } function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) { @@ -332,19 +344,13 @@ function enqueueValueWithSize(controller, value, size) { // arguments passed through to the user callback is observable and must be // preserved. // -// 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. +// Cold algorithms (cancel/close/abort/flush/transform) stay `async` so +// a user thenable is adopted with the same microtask count as before. +// Pull/write use the raw-callback contract instead (see +// createRawCallback*) and route results through thenAlgorithmResult(). function createPromiseCallbackNoParams(name, fn, thisArg) { validateFunction(fn, name); - return () => { - try { - return FunctionPrototypeCall(fn, thisArg); - } catch (error) { - return PromiseReject(error); - } - }; + return async () => FunctionPrototypeCall(fn, thisArg); } // Raw variants that skip the async wrapper's implicit result promise. @@ -382,24 +388,12 @@ function thenAlgorithmResult(result, onFulfilled, onRejected) { function createPromiseCallback1Param(name, fn, thisArg) { validateFunction(fn, name); - return (arg) => { - try { - return FunctionPrototypeCall(fn, thisArg, arg); - } catch (error) { - return PromiseReject(error); - } - }; + return async (arg) => FunctionPrototypeCall(fn, thisArg, arg); } function createPromiseCallback2Params(name, fn, thisArg) { validateFunction(fn, name); - return (arg1, arg2) => { - try { - return FunctionPrototypeCall(fn, thisArg, arg1, arg2); - } catch (error) { - return PromiseReject(error); - } - }; + return async (arg1, arg2) => FunctionPrototypeCall(fn, thisArg, arg1, arg2); } function isPromisePending(promise) { @@ -408,31 +402,12 @@ function isPromisePending(promise) { return details?.[0] === kPending; } -// Convert a promise-returning algorithm's raw result into a Promise. A -// 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 kResolvedPromise; - 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 kResolvedPromise; - return PromisePrototypeThen(kResolvedPromise, () => result); -} - // Shared shapes for lazily-materialized { promise, resolve, reject } -// records whose settlement is already known. +// records whose settlement is already known. Each call mints a fresh +// promise so public slots (writer.ready / writer.closed) stay distinct. function resolvedRecord() { return { - promise: kResolvedPromise, + promise: PromiseResolve(), resolve: undefined, reject: undefined, }; @@ -456,13 +431,16 @@ function setPromiseHandled(promise) { PromisePrototypeThen(promise, undefined, () => {}); } -// Shared no-op. Start/pull/write use the raw-callback contract (see -// createRawCallback*): a non-thenable return takes the allocation-free -// path in thenAlgorithmResult(). Cancel/flush/abort wrap the result -// with promiseFromAlgorithmResult/delayedAlgorithmResult, so a sync -// no-op is equivalent to the previous async empty functions. +async function nonOpFlush() {} + +// Shared non-op for the start/pull/write algorithm callbacks, which all +// follow the raw-callback contract (see createRawCallback*): the +// non-thenable return takes the allocation-free fast path in +// thenAlgorithmResult(). function nonOpCallback() {} +async function nonOpCancel() {} + let transfer; function lazyTransfer() { if (transfer === undefined) @@ -478,7 +456,6 @@ module.exports = { Queue, canCopyArrayBuffer, cloneAsUint8Array, - isNonThenable, copyArrayBuffer, createPromiseCallbackNoParams, createPromiseCallback1Param, @@ -493,6 +470,7 @@ module.exports = { extractSizeAlgorithm, getNonWritablePropertyDescriptor, isBrandCheck, + isNonThenable, isPromisePending, kEmptyQueue, kResolvedPromise, @@ -501,10 +479,8 @@ module.exports = { lazyTransfer, materializeQueue, nonOpCallback, - nonOpCancel: nonOpCallback, - nonOpFlush: nonOpCallback, - promiseFromAlgorithmResult, - delayedAlgorithmResult, + nonOpCancel, + nonOpFlush, peekQueueValue, rejectedHandledRecord, resetQueue, diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index c406d116f07b..c0bb70ca397a 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -68,14 +68,12 @@ const { isNonThenable, isPromisePending, kEmptyQueue, - kResolvedPromise, kState, kType, lazyTransfer, nonOpCallback, nonOpCancel, peekQueueValue, - promiseFromAlgorithmResult, rejectedHandledRecord, resetQueue, resolvedRecord, @@ -531,7 +529,7 @@ class WritableStreamDefaultController { [kAbort](reason) { const result = this[kState].abortAlgorithm(reason); writableStreamDefaultControllerClearAlgorithms(this); - return promiseFromAlgorithmResult(result); + return result; } [kError]() { @@ -1198,54 +1196,6 @@ function writableStreamDefaultControllerCompleteWrite(controller) { } } -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); - let result; - try { - result = controllerState.writeAlgorithm(value, controller); - } catch (error) { - result = PromiseReject(error); - } - 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; - } - PromisePrototypeThen( - kResolvedPromise, - controllerState.writeFulfilled, - controllerState.writeRejected); - return; - } - thenAlgorithmResult( - result, - controllerState.writeFulfilled, - controllerState.writeRejected); - return; - } -} - function writableStreamDefaultControllerProcessWrite(controller, chunk) { const { stream, @@ -1259,9 +1209,7 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { // subsequent write instead of allocating two fresh closures per chunk. controller[kState].writeFulfilled = () => { writableStreamDefaultControllerCompleteWrite(controller); - // Already in the spec's "upon fulfillment" turn: drain further - // synchronous writes here instead of one-write-per-microtask. - writableStreamDefaultControllerDrainWriteQueue(controller); + writableStreamDefaultControllerAdvanceQueueIfNeeded(controller); }; controller[kState].writeRejected = (error) => { if (stream[kState].state === 'writable') @@ -1274,7 +1222,7 @@ function writableStreamDefaultControllerProcessWrite(controller, chunk) { // returns its result uncoerced; a synchronous throw surfaces here) or an // internal algorithm that always returns a promise; thenAlgorithmResult // handles both. Non-thenable results react on kResolvedPromise so each - // writer.write() completion is still separated by a microtask. + // write completion is still separated by a microtask. let result; try { result = writeAlgorithm(chunk, controller); @@ -1296,7 +1244,7 @@ function writableStreamDefaultControllerProcessClose(controller) { writableStreamMarkCloseRequestInFlight(stream); dequeueValue(controller); assert(!queue.length); - const sinkClosePromise = promiseFromAlgorithmResult(closeAlgorithm()); + const sinkClosePromise = closeAlgorithm(); writableStreamDefaultControllerClearAlgorithms(controller); PromisePrototypeThen( sinkClosePromise, diff --git a/node.gyp b/node.gyp index ff00f73476a9..4f7a3d1ff634 100644 --- a/node.gyp +++ b/node.gyp @@ -174,7 +174,6 @@ '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 48ce8d86ac1e..330c7f167105 100644 --- a/src/node_binding.cc +++ b/src/node_binding.cc @@ -99,7 +99,6 @@ 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 89be54a19cc0..1e987ce2d4f3 100644 --- a/src/node_external_reference.h +++ b/src/node_external_reference.h @@ -119,7 +119,6 @@ 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 deleted file mode 100644 index 87d49724ed8d..000000000000 --- a/src/node_webstreams.cc +++ /dev/null @@ -1,103 +0,0 @@ -#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 { - -// 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()); -} - -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 index 929c35cde166..3cff7f555e9f 100644 --- a/test/parallel/test-whatwg-webstreams-hotpath.js +++ b/test/parallel/test-whatwg-webstreams-hotpath.js @@ -7,13 +7,15 @@ const { ReadableStream, WritableStream, } = require('node:stream/web'); -const { internalBinding } = require('internal/test/binding'); const { - isNonThenable, cloneAsUint8Array, -} = internalBinding('webstreams'); + isNonThenable, + kState, +} = require('internal/webstreams/util'); +const { + kControllerErrorFunction, +} = require('internal/streams/utils'); -// The native helpers must be the ones the JS implementation actually calls. assert.strictEqual(typeof isNonThenable, 'function'); assert.strictEqual(typeof cloneAsUint8Array, 'function'); @@ -39,8 +41,35 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); } { - assert.throws(() => cloneAsUint8Array(1), { - code: 'ERR_INVALID_ARG_TYPE', + const view = new DataView(new ArrayBuffer(4)); + new Uint8Array(view.buffer).set([9, 8, 7, 6]); + const cloned = cloneAsUint8Array(view); + assert.ok(cloned instanceof Uint8Array); + assert.deepStrictEqual([...cloned], [9, 8, 7, 6]); +} + +{ + const cloned = cloneAsUint8Array(new Uint8Array()); + assert.ok(cloned instanceof Uint8Array); + assert.strictEqual(cloned.byteLength, 0); +} + +{ + const buf = Buffer.alloc(16); + buf[4] = 7; + buf[5] = 8; + const sliced = buf.subarray(4, 8); + const cloned = cloneAsUint8Array(sliced); + assert.deepStrictEqual([...cloned], [7, 8, 0, 0]); + assert.strictEqual(cloned.buffer.byteLength, 4); +} + +{ + const ab = new ArrayBuffer(8); + const view = new Uint8Array(ab); + ab.transfer(); + assert.throws(() => cloneAsUint8Array(view), { + name: 'TypeError', }); } @@ -69,9 +98,9 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); assert.strictEqual(value, undefined); assert.strictEqual(done, true); } -})().then(common.mustCall()); +})().then(common.mustCall(), common.mustNotCall()); -// Public API: pipeTo with a sync sink — the optimized write drain path. +// Public API: pipeTo with a sync sink. (async () => { const expected = []; const received = []; @@ -90,11 +119,12 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); }, })); assert.deepStrictEqual(received, expected); -})().then(common.mustCall()); +})().then(common.mustCall(), common.mustNotCall()); // Spec path: each pull is separated by a microtask. Start schedules one // pull; further pulls wait for that fulfillment and do not run in the -// same turn. +// same turn. Nested queueMicrotask checks encode that exact schedule +// and will fail if an unrelated change shifts pull timing. { let calls = 0; new ReadableStream({ @@ -134,13 +164,13 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); }, { highWaterMark: 8 })); assert.strictEqual(received.length, n); assert.deepStrictEqual(received, Array.from({ length: n }, (_, k) => k)); -})().then(common.mustCall()); +})().then(common.mustCall(), common.mustNotCall()); { // 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()); + rs.cancel().then(common.mustCall(), common.mustNotCall()); } { @@ -154,7 +184,7 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); } const rs = new Sub(); assert.ok(rs.locked); - rs.reader.cancel().then(common.mustCall()); + rs.reader.cancel().then(common.mustCall(), common.mustNotCall()); } { @@ -166,7 +196,7 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); } } const rs = new Sub(); - rs.closed.then(common.mustCall()); + rs.closed.then(common.mustCall(), common.mustNotCall()); } { @@ -188,14 +218,14 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); assert.ok(rs.locked); queueMicrotask(common.mustCall(() => { assert.strictEqual(sawController, true); - rs.reader.cancel().then(common.mustCall()); + rs.reader.cancel().then(common.mustCall(), common.mustNotCall()); })); } { const rs = new ReadableStream(); const reader = rs.getReader(); - reader.cancel().then(common.mustCall()); + reader.cancel().then(common.mustCall(), common.mustNotCall()); } { @@ -235,3 +265,42 @@ assert.strictEqual(isNonThenable(new Proxy(() => {}, {})), false); assert.strictEqual(ctrl.signal.aborted, true); assert.strictEqual(ctrl.signal.reason, err); } + +{ + // writer.ready and writer.closed are distinct spec slots, and two + // writers must not share a process-wide resolved promise. + const a = new WritableStream().getWriter(); + const b = new WritableStream().getWriter(); + assert.notStrictEqual(a.ready, b.ready); + assert.notStrictEqual(a.closed, b.closed); + assert.notStrictEqual(a.ready, a.closed); +} + +(async () => { + const ws = new WritableStream(); + await ws.close(); + const writer = ws.getWriter(); + assert.notStrictEqual(writer.ready, writer.closed); + await Promise.all([writer.ready, writer.closed]); +})().then(common.mustCall(), common.mustNotCall()); + +{ + // stream.cancel() must not return a process-shared resolved promise. + const a = new ReadableStream(); + const b = new ReadableStream(); + const pa = a.cancel(); + const pb = b.cancel(); + assert.notStrictEqual(pa, pb); + pa.then(common.mustCall(), common.mustNotCall()); + pb.then(common.mustCall(), common.mustNotCall()); +} + +{ + // Erroring an empty stream via interop must not allocate a controller, + // and a later cancel() must not allocate one either. + const rs = new ReadableStream(); + rs[kControllerErrorFunction](new Error('empty-error')); + assert.strictEqual(rs[kState].controller, undefined); + rs.cancel().then(common.mustNotCall(), common.mustCall()); + assert.strictEqual(rs[kState].controller, undefined); +} diff --git a/typings/globals.d.ts b/typings/globals.d.ts index 13b947c3e285..e64b8e6d89fc 100644 --- a/typings/globals.d.ts +++ b/typings/globals.d.ts @@ -38,7 +38,6 @@ import { UtilBinding } from './internalBinding/util'; import { UVBinding } from './internalBinding/uv'; import { WASIBinding } from './internalBinding/wasi'; import { WatchdogBinding } from './internalBinding/watchdog'; -import { WebstreamsBinding } from './internalBinding/webstreams'; import { WorkerBinding } from './internalBinding/worker'; import { ZlibBinding } from './internalBinding/zlib'; @@ -83,7 +82,6 @@ interface InternalBindingMap { uv: UVBinding; wasi: WASIBinding; watchdog: WatchdogBinding; - webstreams: WebstreamsBinding; worker: WorkerBinding; zlib: ZlibBinding; } diff --git a/typings/internalBinding/webstreams.d.ts b/typings/internalBinding/webstreams.d.ts deleted file mode 100644 index 814c4fb77481..000000000000 --- a/typings/internalBinding/webstreams.d.ts +++ /dev/null @@ -1,4 +0,0 @@ -export interface WebstreamsBinding { - isNonThenable(value: unknown): boolean; - cloneAsUint8Array(view: ArrayBufferView): Uint8Array; -}