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

stream: cut per-chunk allocations in pipeTo · nodejs/node@76baf80 · GitHub

/ node Public

Commit 76baf80

Browse files
mcollina
authored and
aduh95
committed
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 c543cfb. 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 <hello@matteocollina.com> PR-URL: #64890 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com>
1 parent 95d27e2 commit 76baf80

3 files changed

Lines changed: 310 additions & 65 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 92 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -46,10 +46,6 @@ const {
4646
DOMException,
4747
} = internalBinding('messaging');
4848

49-
const {
50-
markPromiseAsHandled,
51-
} = internalBinding('util');
52-
5349
const {
5450
isArrayBufferView,
5551
isDataView,
@@ -139,7 +135,7 @@ const {
139135
writableStreamCloseQueuedOrInFlight,
140136
writableStreamDefaultWriterCloseWithErrorPropagation,
141137
writableStreamDefaultWriterRelease,
142-
writableStreamDefaultWriterWrite,
138+
writableStreamDefaultWriterWriteWithRequest,
143139
writerClosedPromise,
144140
writerReadyPromise,
145141
} = require('internal/webstreams/writablestream');
@@ -1530,8 +1526,38 @@ function readableStreamPipeTo(
15301526

15311527
const promise = PromiseWithResolvers();
15321528

1533-
const state = {
1534-
currentWrite: PromiseResolve(),
1529+
// One shared write request tracks every chunk written to the
1530+
// destination, instead of a { promise, resolve, reject } record per
1531+
// write. `stall` is armed by waitForPendingWrites() during shutdown;
1532+
// `failed`/`failure` latch a write that could not proceed.
1533+
const writeTracker = {
1534+
// Non-undefined: queue entries are discriminated from kNilRequest
1535+
// by `promise === undefined`.
1536+
promise: null,
1537+
pending: 0,
1538+
failed: false,
1539+
failure: undefined,
1540+
stall: null,
1541+
resolve() {
1542+
if (--this.pending === 0 && this.stall !== null) {
1543+
const stall = this.stall;
1544+
this.stall = null;
1545+
if (this.failed)
1546+
stall.reject(this.failure);
1547+
else
1548+
stall.resolve();
1549+
}
1550+
},
1551+
reject(error) {
1552+
this.pending--;
1553+
this.failed = true;
1554+
this.failure = error;
1555+
if (this.stall !== null) {
1556+
const stall = this.stall;
1557+
this.stall = null;
1558+
stall.reject(error);
1559+
}
1560+
},
15351561
};
15361562

15371563
// The error here can be undefined. The rejected arg
@@ -1548,11 +1574,14 @@ function readableStreamPipeTo(
15481574
promise.resolve();
15491575
}
15501576

1551-
async function waitForCurrentWrite() {
1552-
const write = state.currentWrite;
1553-
await write;
1554-
if (write !== state.currentWrite)
1555-
await waitForCurrentWrite();
1577+
function waitForPendingWrites() {
1578+
if (writeTracker.pending === 0) {
1579+
return writeTracker.failed ?
1580+
PromiseReject(writeTracker.failure) :
1581+
PromiseResolve();
1582+
}
1583+
writeTracker.stall = PromiseWithResolvers();
1584+
return writeTracker.stall.promise;
15561585
}
15571586

15581587
function shutdownWithAnAction(action, rejected, originalError) {
@@ -1561,7 +1590,7 @@ function readableStreamPipeTo(
15611590
if (dest[kState].state === 'writable' &&
15621591
!writableStreamCloseQueuedOrInFlight(dest)) {
15631592
PromisePrototypeThen(
1564-
waitForCurrentWrite(),
1593+
waitForPendingWrites(),
15651594
complete,
15661595
(error) => finalize(true, error));
15671596
return;
@@ -1582,7 +1611,7 @@ function readableStreamPipeTo(
15821611
if (dest[kState].state === 'writable' &&
15831612
!writableStreamCloseQueuedOrInFlight(dest)) {
15841613
PromisePrototypeThen(
1585-
waitForCurrentWrite(),
1614+
waitForPendingWrites(),
15861615
() => finalize(rejected, error),
15871616
(error) => finalize(true, error));
15881617
return;
@@ -1639,25 +1668,46 @@ function readableStreamPipeTo(
16391668
PromisePrototypeThen(promise, action, () => {});
16401669
}
16411670

1642-
async function step() {
1643-
if (shuttingDown) return true;
1671+
// The pump loop is callback-driven to avoid per-iteration promise
1672+
// allocations. At most one read is in flight at a time, so one read
1673+
// request and one forwarding function are reused for every chunk;
1674+
// the chunk travels through `pendingChunk`.
1675+
let pendingChunk;
1676+
let readRequest;
1677+
1678+
// Ready promise rejection is handled by the destination-errored
1679+
// watcher.
1680+
function ignoreReadyRejection() {}
1681+
1682+
function forwardChunk() {
1683+
const chunk = pendingChunk;
1684+
pendingChunk = undefined;
1685+
writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
1686+
pump();
1687+
}
1688+
1689+
function pump() {
1690+
if (shuttingDown) return;
16441691

16451692
if (dest[kState].backpressure) {
1646-
await writerReadyPromise(writer).promise;
1647-
if (shuttingDown) return true;
1693+
PromisePrototypeThen(
1694+
writerReadyPromise(writer).promise,
1695+
pump,
1696+
ignoreReadyRejection);
1697+
return;
16481698
}
16491699

16501700
const controller = source[kState].controller;
16511701

16521702
// Fast path: batch reads when data is buffered in a default controller.
1653-
// This avoids creating PipeToReadableStreamReadRequest objects and
1654-
// reduces promise allocation overhead.
1703+
// This avoids parking read requests and reduces promise allocation
1704+
// overhead.
16551705
if (source[kState].state === 'readable' &&
16561706
isReadableStreamDefaultController(controller) &&
16571707
controller[kState].queue.length > 0) {
16581708

16591709
while (controller[kState].queue.length > 0) {
1660-
if (shuttingDown) return true;
1710+
if (shuttingDown) return;
16611711

16621712
const chunk = dequeueValue(controller);
16631713

@@ -1668,8 +1718,7 @@ function readableStreamPipeTo(
16681718

16691719
// Write the chunk - we're already in a separate microtask from enqueue
16701720
// because we awaited the writer ready promise above.
1671-
state.currentWrite = writableStreamDefaultWriterWrite(writer, chunk);
1672-
markPromiseAsHandled(state.currentWrite);
1721+
writableStreamDefaultWriterWriteWithRequest(writer, chunk, writeTracker);
16731722

16741723
// Check backpressure after each write
16751724
if (dest[kState].backpressure) {
@@ -1686,24 +1735,29 @@ function readableStreamPipeTo(
16861735

16871736
// Check if stream closed during batch
16881737
if (source[kState].state === 'closed') {
1689-
return true;
1738+
return;
16901739
}
16911740

1692-
// Yield to microtask queue between batches to allow events/signals to fire
1693-
return false;
1741+
// Yield to microtask queue between batches to allow events/signals
1742+
// to fire
1743+
queueMicrotask(pump);
1744+
return;
16941745
}
16951746

1696-
// Slow path: use read request for async reads
1697-
const promise = PromiseWithResolvers();
1698-
// eslint-disable-next-line no-use-before-define
1699-
readableStreamDefaultReaderRead(reader, new PipeToReadableStreamReadRequest(writer, state, promise));
1700-
1701-
return promise.promise;
1702-
}
1703-
1704-
async function run() {
1705-
// Run until step resolves as true
1706-
while (!await step());
1747+
// Slow path: park a lazily materialized read request. Close and
1748+
// error are handled by the source watchers.
1749+
readRequest ??= {
1750+
[kChunk](chunk) {
1751+
// Per spec, pipeTo must queue a microtask for the write to avoid
1752+
// synchronous write during enqueue(). See WHATWG Streams spec
1753+
// "ReadableStreamPipeTo" step 15's "chunk steps".
1754+
pendingChunk = chunk;
1755+
queueMicrotask(forwardChunk);
1756+
},
1757+
[kClose]() {},
1758+
[kError]() {},
1759+
};
1760+
readableStreamDefaultReaderRead(reader, readRequest);
17071761
}
17081762

17091763
if (signal !== undefined) {
@@ -1715,7 +1769,7 @@ function readableStreamPipeTo(
17151769
disposable = addAbortListener(signal, abortAlgorithm);
17161770
}
17171771

1718-
setPromiseHandled(run());
1772+
pump();
17191773

17201774
watchErrored(source, readerClosedPromise(reader).promise, (error) => {
17211775
if (!preventAbort) {
@@ -1760,33 +1814,6 @@ function readableStreamPipeTo(
17601814
return promise.promise;
17611815
}
17621816

1763-
class PipeToReadableStreamReadRequest {
1764-
constructor(writer, state, promise) {
1765-
this.writer = writer;
1766-
this.state = state;
1767-
this.promise = promise;
1768-
}
1769-
1770-
[kChunk](chunk) {
1771-
// Per spec, pipeTo must queue a microtask for the write to avoid
1772-
// synchronous write during enqueue(). See WHATWG Streams spec
1773-
// "ReadableStreamPipeTo" step 15's "chunk steps".
1774-
queueMicrotask(() => {
1775-
this.state.currentWrite = writableStreamDefaultWriterWrite(this.writer, chunk);
1776-
markPromiseAsHandled(this.state.currentWrite);
1777-
this.promise.resolve(false);
1778-
});
1779-
}
1780-
1781-
[kClose]() {
1782-
this.promise.resolve(true);
1783-
}
1784-
1785-
[kError](error) {
1786-
this.promise.reject(error);
1787-
}
1788-
}
1789-
17901817
function readableStreamTee(stream, cloneForBranch2) {
17911818
if (isReadableByteStreamController(stream[kState].controller)) {
17921819
return readableByteStreamTee(stream);

‎lib/internal/webstreams/writablestream.js‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1001,6 +1001,61 @@ function writableStreamDefaultWriterWrite(writer, chunk) {
10011001
return promise;
10021002
}
10031003

1004+
// Variant of writableStreamDefaultWriterWrite for pipeTo: the caller
1005+
// provides a shared request object instead of a per-write promise record.
1006+
// `pending` is incremented before the controller write, which can settle
1007+
// requests synchronously when the stream starts erroring; precondition
1008+
// failures are latched on `failed`/`failure`.
1009+
function writableStreamDefaultWriterWriteWithRequest(writer, chunk, request) {
1010+
const writerState = writer[kState];
1011+
const stream = writerState.stream;
1012+
assert(stream !== undefined);
1013+
const streamState = stream[kState];
1014+
const {
1015+
controller,
1016+
} = streamState;
1017+
const chunkSize = writableStreamDefaultControllerGetChunkSize(
1018+
controller,
1019+
chunk);
1020+
if (stream !== writerState.stream) {
1021+
request.failed = true;
1022+
request.failure =
1023+
new ERR_INVALID_STATE.TypeError('Mismatched WritableStreams');
1024+
return;
1025+
}
1026+
const {
1027+
state,
1028+
} = streamState;
1029+
1030+
if (state === 'errored') {
1031+
request.failed = true;
1032+
request.failure = streamState.storedError;
1033+
return;
1034+
}
1035+
1036+
if (streamState.closeQueuedOrInFlight || state === 'closed') {
1037+
request.failed = true;
1038+
request.failure =
1039+
new ERR_INVALID_STATE.TypeError('WritableStream is closed');
1040+
return;
1041+
}
1042+
1043+
if (state === 'erroring') {
1044+
request.failed = true;
1045+
request.failure = streamState.storedError;
1046+
return;
1047+
}
1048+
1049+
assert(state === 'writable');
1050+
1051+
let writeRequests = streamState.writeRequests;
1052+
if (writeRequests === kEmptyQueue)
1053+
writeRequests = streamState.writeRequests = new Queue();
1054+
writeRequests.push(request);
1055+
request.pending++;
1056+
writableStreamDefaultControllerWrite(controller, chunk, chunkSize);
1057+
}
1058+
10041059
function writableStreamDefaultWriterRelease(writer) {
10051060
const {
10061061
stream,
@@ -1376,6 +1431,7 @@ module.exports = {
13761431
writableStreamCloseQueuedOrInFlight,
13771432
writableStreamAddWriteRequest,
13781433
writableStreamDefaultWriterWrite,
1434+
writableStreamDefaultWriterWriteWithRequest,
13791435
writableStreamDefaultWriterRelease,
13801436
writableStreamDefaultWriterGetDesiredSize,
13811437
writableStreamDefaultWriterEnsureReadyPromiseRejected,

0 commit comments

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