From d838534e52202ae55f29409c7e09c425e3bca93a Mon Sep 17 00:00:00 2001 From: Matteo Collina Date: Sat, 1 Aug 2026 08:50:38 +0200 Subject: [PATCH] stream: cut per-chunk allocations in pipeTo readableStreamPipeTo allocated, for every chunk written to the destination, a { promise, resolve, reject } write request record that it immediately marked as handled, and drove its loop with an async step()/run() pair whose implicit promises cost one allocation and one reaction per iteration. The parked-read path additionally allocated a read request object, a PromiseWithResolvers record, and a microtask closure per chunk; this is the steady state for pipeThrough, since a TransformStream's readable side has a high water mark of zero. Replace the per-write records with a single per-pipe tracker that the write request queue holds once per pending write and whose resolve()/reject() methods maintain a pending-write count, drive the pump loop with plain callbacks instead of async functions, and reuse one read request and one forwarding function across all chunks, the same pattern tee uses since c543cfb72b7. Benchmark results (benchmark/compare.js --runs 20): webstreams/pipe-to.js +29.9% to +35.8% across all 16 configurations (all 99.9% confidence); a pipeThrough(TransformStream) passthrough loop improves ~17%; every other webstreams benchmark is unchanged. Signed-off-by: Matteo Collina --- lib/internal/webstreams/readablestream.js | 157 ++++++++++------- lib/internal/webstreams/writablestream.js | 56 ++++++ .../test-webstreams-pipeto-write-request.js | 162 ++++++++++++++++++ 3 files changed, 310 insertions(+), 65 deletions(-) create mode 100644 test/parallel/test-webstreams-pipeto-write-request.js diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js index b25166957614..e1e80eb953c0 100644 --- a/lib/internal/webstreams/readablestream.js +++ b/lib/internal/webstreams/readablestream.js @@ -46,10 +46,6 @@ const { DOMException, } = internalBinding('messaging'); -const { - markPromiseAsHandled, -} = internalBinding('util'); - const { isArrayBufferView, isDataView, @@ -139,7 +135,7 @@ const { writableStreamCloseQueuedOrInFlight, writableStreamDefaultWriterCloseWithErrorPropagation, writableStreamDefaultWriterRelease, - writableStreamDefaultWriterWrite, + writableStreamDefaultWriterWriteWithRequest, writerClosedPromise, writerReadyPromise, } = require('internal/webstreams/writablestream'); @@ -1530,8 +1526,38 @@ function readableStreamPipeTo( const promise = PromiseWithResolvers(); - const state = { - currentWrite: PromiseResolve(), + // One shared write request tracks every chunk written to the + // destination, instead of a { promise, resolve, reject } record per + // write. `stall` is armed by waitForPendingWrites() during shutdown; + // `failed`/`failure` latch a write that could not proceed. + const writeTracker = { + // Non-undefined: queue entries are discriminated from kNilRequest + // by `promise === undefined`. + promise: null, + pending: 0, + failed: false, + failure: undefined, + stall: null, + resolve() { + if (--this.pending === 0 && this.stall !== null) { + const stall = this.stall; + this.stall = null; + if (this.failed) + stall.reject(this.failure); + else + stall.resolve(); + } + }, + reject(error) { + this.pending--; + this.failed = true; + this.failure = error; + if (this.stall !== null) { + const stall = this.stall; + this.stall = null; + stall.reject(error); + } + }, }; // The error here can be undefined. The rejected arg @@ -1548,11 +1574,14 @@ function readableStreamPipeTo( promise.resolve(); } - async function waitForCurrentWrite() { - const write = state.currentWrite; - await write; - if (write !== state.currentWrite) - await waitForCurrentWrite(); + function waitForPendingWrites() { + if (writeTracker.pending === 0) { + return writeTracker.failed ? + PromiseReject(writeTracker.failure) : + PromiseResolve(); + } + writeTracker.stall = PromiseWithResolvers(); + return writeTracker.stall.promise; } function shutdownWithAnAction(action, rejected, originalError) { @@ -1561,7 +1590,7 @@ function readableStreamPipeTo( if (dest[kState].state === 'writable' && !writableStreamCloseQueuedOrInFlight(dest)) { PromisePrototypeThen( - waitForCurrentWrite(), + waitForPendingWrites(), complete, (error) => finalize(true, error)); return; @@ -1582,7 +1611,7 @@ function readableStreamPipeTo( if (dest[kState].state === 'writable' && !writableStreamCloseQueuedOrInFlight(dest)) { PromisePrototypeThen( - waitForCurrentWrite(), + waitForPendingWrites(), () => finalize(rejected, error), (error) => finalize(true, error)); return; @@ -1639,25 +1668,46 @@ function readableStreamPipeTo( PromisePrototypeThen(promise, action, () => {}); } - async function step() { - if (shuttingDown) return true; + // The pump loop is callback-driven to avoid per-iteration promise + // allocations. At most one read is in flight at a time, so one read + // request and one forwarding function are reused for every chunk; + // the chunk travels through `pendingChunk`. + let pendingChunk; + let readRequest; + + // Ready promise rejection is handled by the destination-errored + // watcher. + function ignoreReadyRejection() {} + + function forwardChunk() { + const chunk = pendingChunk; + pendingChunk = undefined; + writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker); + pump(); + } + + function pump() { + if (shuttingDown) return; if (dest[kState].backpressure) { - await writerReadyPromise(writer).promise; - if (shuttingDown) return true; + PromisePrototypeThen( + writerReadyPromise(writer).promise, + pump, + ignoreReadyRejection); + return; } const controller = source[kState].controller; // Fast path: batch reads when data is buffered in a default controller. - // This avoids creating PipeToReadableStreamReadRequest objects and - // reduces promise allocation overhead. + // This avoids parking read requests and reduces promise allocation + // overhead. if (source[kState].state === 'readable' && isReadableStreamDefaultController(controller) && controller[kState].queue.length > 0) { while (controller[kState].queue.length > 0) { - if (shuttingDown) return true; + if (shuttingDown) return; const chunk = dequeueValue(controller); @@ -1668,8 +1718,7 @@ function readableStreamPipeTo( // Write the chunk - we're already in a separate microtask from enqueue // because we awaited the writer ready promise above. - state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk); - markPromiseAsHandled(state.currentWrite); + writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker); // Check backpressure after each write if (dest[kState].backpressure) { @@ -1686,24 +1735,29 @@ function readableStreamPipeTo( // Check if stream closed during batch if (source[kState].state === 'closed') { - return true; + return; } - // Yield to microtask queue between batches to allow events/signals to fire - return false; + // Yield to microtask queue between batches to allow events/signals + // to fire + queueMicrotask(pump); + return; } - // Slow path: use read request for async reads - const promise = PromiseWithResolvers(); - // eslint-disable-next-line no-use-before-define - readableStreamDefaultReaderRead(reader, new PipeToReadableStreamReadRequest(writer, state, promise)); - - return promise.promise; - } - - async function run() { - // Run until step resolves as true - while (!await step()); + // Slow path: park a lazily materialized read request. Close and + // error are handled by the source watchers. + readRequest ??= { + [kChunk](chunk) { + // Per spec, pipeTo must queue a microtask for the write to avoid + // synchronous write during enqueue(). See WHATWG Streams spec + // "ReadableStreamPipeTo" step 15's "chunk steps". + pendingChunk = chunk; + queueMicrotask(forwardChunk); + }, + [kClose]() {}, + [kError]() {}, + }; + readableStreamDefaultReaderRead(reader, readRequest); } if (signal !== undefined) { @@ -1715,7 +1769,7 @@ function readableStreamPipeTo( disposable = addAbortListener(signal, abortAlgorithm); } - setPromiseHandled(run()); + pump(); watchErrored(source, readerClosedPromise(reader).promise, (error) => { if (!preventAbort) { @@ -1760,33 +1814,6 @@ function readableStreamPipeTo( return promise.promise; } -class PipeToReadableStreamReadRequest { - constructor(writer, state, promise) { - this.writer = writer; - this.state = state; - this.promise = promise; - } - - [kChunk](chunk) { - // Per spec, pipeTo must queue a microtask for the write to avoid - // synchronous write during enqueue(). See WHATWG Streams spec - // "ReadableStreamPipeTo" step 15's "chunk steps". - queueMicrotask(() => { - this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk); - markPromiseAsHandled(this.state.currentWrite); - this.promise.resolve(false); - }); - } - - [kClose]() { - this.promise.resolve(true); - } - - [kError](error) { - this.promise.reject(error); - } -} - function readableStreamTee(stream, cloneForBranch2) { if (isReadableByteStreamController(stream[kState].controller)) { return readableByteStreamTee(stream); diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index f0eb687be129..87e3bcaa2850 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -1001,6 +1001,61 @@ function writableStreamDefaultWriterWrite(writer, chunk) { return promise; } +// Variant of writableStreamDefaultWriterWrite for pipeTo: the caller +// provides a shared request object instead of a per-write promise record. +// `pending` is incremented before the controller write, which can settle +// requests synchronously when the stream starts erroring; precondition +// failures are latched on `failed`/`failure`. +function writableStreamDefaultWriterWriteWithRequest(writer, chunk, request) { + const writerState = writer[kState]; + const stream = writerState.stream; + assert(stream !== undefined); + const streamState = stream[kState]; + const { + controller, + } = streamState; + const chunkSize = writableStreamDefaultControllerGetChunkSize( + controller, + chunk); + if (stream !== writerState.stream) { + request.failed = true; + request.failure = + new ERR_INVALID_STATE.TypeError('Mismatched WritableStreams'); + return; + } + const { + state, + } = streamState; + + if (state === 'errored') { + request.failed = true; + request.failure = streamState.storedError; + return; + } + + if (streamState.closeQueuedOrInFlight || state === 'closed') { + request.failed = true; + request.failure = + new ERR_INVALID_STATE.TypeError('WritableStream is closed'); + return; + } + + if (state === 'erroring') { + request.failed = true; + request.failure = streamState.storedError; + return; + } + + assert(state === 'writable'); + + let writeRequests = streamState.writeRequests; + if (writeRequests === kEmptyQueue) + writeRequests = streamState.writeRequests = new Queue(); + writeRequests.push(request); + request.pending++; + writableStreamDefaultControllerWrite(controller, chunk, chunkSize); +} + function writableStreamDefaultWriterRelease(writer) { const { stream, @@ -1376,6 +1431,7 @@ module.exports = { writableStreamCloseQueuedOrInFlight, writableStreamAddWriteRequest, writableStreamDefaultWriterWrite, + writableStreamDefaultWriterWriteWithRequest, writableStreamDefaultWriterRelease, writableStreamDefaultWriterGetDesiredSize, writableStreamDefaultWriterEnsureReadyPromiseRejected, diff --git a/test/parallel/test-webstreams-pipeto-write-request.js b/test/parallel/test-webstreams-pipeto-write-request.js new file mode 100644 index 000000000000..23e973a400da --- /dev/null +++ b/test/parallel/test-webstreams-pipeto-write-request.js @@ -0,0 +1,162 @@ +// Flags: --expose-internals --no-warnings +'use strict'; + +// Exercises writableStreamDefaultWriterWriteWithRequest: settle paths +// and the latched precondition failures. + +const common = require('../common'); +const assert = require('assert'); + +const { + WritableStream, + WritableStreamDefaultWriter, + ReadableStream, +} = require('stream/web'); + +const { + writableStreamDefaultWriterWriteWithRequest, +} = require('internal/webstreams/writablestream'); + +function makeRequest(overrides = {}) { + return { + promise: null, + pending: 0, + failed: false, + failure: undefined, + resolve: common.mustNotCall('resolve'), + reject: common.mustNotCall('reject'), + ...overrides, + }; +} + +{ + // A write to a writable destination bumps `pending` and settles through + // request.resolve(). + const request = makeRequest({ + resolve: common.mustCall(function() { + assert.strictEqual(this, request); + assert.strictEqual(this.pending, 1); + assert.strictEqual(this.failed, false); + }), + }); + const ws = new WritableStream({ + write: common.mustCall((chunk) => { + assert.strictEqual(chunk, 'chunk'); + }), + }); + const writer = new WritableStreamDefaultWriter(ws); + writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request); + assert.strictEqual(request.pending, 1); +} + +{ + // A write whose sink rejects settles through request.reject() with the + // sink's error. + const error = new Error('sink failure'); + const request = makeRequest({ + reject: common.mustCall(function(reason) { + assert.strictEqual(this, request); + assert.strictEqual(reason, error); + }), + }); + const ws = new WritableStream({ + write: common.mustCall(() => Promise.reject(error)), + }); + const writer = new WritableStreamDefaultWriter(ws); + writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request); + assert.strictEqual(request.pending, 1); +} + +{ + // Writing to an errored destination latches the stored error without + // queueing the write. + const error = new Error('start failure'); + const ws = new WritableStream({ + start(controller) { controller.error(error); }, + }); + queueMicrotask(common.mustCall(() => { + const writer = new WritableStreamDefaultWriter(ws); + const request = makeRequest(); + writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request); + assert.strictEqual(request.pending, 0); + assert.strictEqual(request.failed, true); + assert.strictEqual(request.failure, error); + })); +} + +{ + // Writing to a destination with a queued close latches an + // ERR_INVALID_STATE TypeError. + const ws = new WritableStream({}); + const writer = new WritableStreamDefaultWriter(ws); + writer.close().then(common.mustCall()); + const request = makeRequest(); + writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request); + assert.strictEqual(request.pending, 0); + assert.strictEqual(request.failed, true); + assert.match(request.failure.message, /WritableStream is closed/); +} + +{ + // Writing to an erroring destination latches the abort reason. The + // stream stays in the 'erroring' state until its start algorithm + // settles, so aborting right after construction reaches it + // deterministically. + const reason = new Error('abort reason'); + const ws = new WritableStream({}); + const writer = new WritableStreamDefaultWriter(ws); + writer.abort(reason).then(common.mustCall()); + const request = makeRequest(); + writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request); + assert.strictEqual(request.pending, 0); + assert.strictEqual(request.failed, true); + assert.strictEqual(request.failure, reason); +} + +{ + // A size algorithm that detaches the writer makes the write latch a + // mismatched-streams error. + let writer; + const ws = new WritableStream({}, { + size: common.mustCall(() => { + writer.releaseLock(); + return 1; + }), + highWaterMark: 1, + }); + writer = new WritableStreamDefaultWriter(ws); + const request = makeRequest(); + writableStreamDefaultWriterWriteWithRequest(writer, 'chunk', request); + assert.strictEqual(request.pending, 0); + assert.strictEqual(request.failed, true); + assert.match(request.failure.message, /Mismatched WritableStreams/); +} + +{ + // End to end: aborting a pipe with a write in flight still waits for + // the in-flight write before aborting the destination, and the pipe + // rejects with an AbortError. + const ac = new AbortController(); + const order = []; + const { promise: gate, resolve: openGate } = Promise.withResolvers(); + let i = 0; + const rs = new ReadableStream({ + pull(controller) { controller.enqueue(i++); }, + }); + const pipe = rs.pipeTo(new WritableStream({ + write: common.mustCall((chunk) => { + order.push(`write:${chunk}`); + ac.abort(); + queueMicrotask(() => { + order.push('settle:0'); + openGate(); + }); + return gate; + }), + abort: common.mustCall(() => { + order.push('abort'); + assert.deepStrictEqual(order, ['write:0', 'settle:0', 'abort']); + }), + }), { signal: ac.signal }); + assert.rejects(pipe, { name: 'AbortError' }).then(common.mustCall()); +}