Visitar URL original
stream: speed up async iteration over WHATWG byte streams · nodejs/node@f249614 · GitHub
Skip to content

Commit f249614

Browse files
mcollinaRafaelGSS
mcollina
authored and
RafaelGSS
committed
stream: speed up async iteration over WHATWG byte streams
for await / reader.read() loops over byte streams were ~4x slower than over default streams. Three per-chunk costs, none required by the spec: - ArrayBufferViewGetBuffer/ByteLength/ByteOffset went through ReflectGet(view.constructor.prototype, ...), a reflective get that is ~3.5x slower than the original prototype getters from primordials and spoofable through a user-defined .constructor to boot. - The buffered fast paths in ReadableStreamDefaultReader.read() and the async iterator only covered default controllers, so byte streams with queued data still allocated a read request and PromiseWithResolvers per chunk. Byte-queue dequeue is fully synchronous (it is the queue-filled arm of the byte controller's pull steps), so both fast paths now resolve directly from the byte queue. - readableByteStreamControllerEnqueue re-ran the reader brand check and re-loaded the read request list four times per chunk across HasDefaultReader / ProcessReadRequestsUsingQueue / GetNumReadRequests / FulfillReadRequest; it now does a single pass. The async iterator also reuses its read request object across reads (at most one is ever in flight). benchmark/webstreams interleaved same-day A/B, --runs 10: readable-async-iterator bytes +16.3% (***), readable-read byob +9.1% (***), all other rows neutral. Profiler harness: parked byte iteration +14%, buffered byte iteration +37%, buffered byte read loop +18%, default-stream rows at parity. WPT streams/compression/encoding subtests identical to baseline. Signed-off-by: Matteo Collina <hello@matteocollina.com> PR-URL: #64291 Reviewed-By: Stephen Belanger <admin@stephenbelanger.com> Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com> Reviewed-By: Yagiz Nizipli <yagiz@nizipli.com> Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 933e5f1 commit f249614

2 files changed

Lines changed: 118 additions & 51 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 93 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -501,6 +501,12 @@ class ReadableStream {
501501
current: undefined,
502502
};
503503
let started = false;
504+
// A single reusable read request: at most one read is ever in flight
505+
// (next() chains through state.current), and the request is consumed
506+
// before the next read starts, so only its promise record changes
507+
// per read.
508+
// eslint-disable-next-line no-use-before-define
509+
const readRequest = new ReadableStreamAsyncIteratorReadRequest(reader, state, undefined);
504510

505511
// The nextSteps function is not an async function in order
506512
// to make it more efficient. Because nextSteps explicitly
@@ -519,8 +525,8 @@ class ReadableStream {
519525
}
520526
const promise = PromiseWithResolvers();
521527

522-
// eslint-disable-next-line no-use-before-define
523-
readableStreamDefaultReaderRead(reader, new ReadableStreamAsyncIteratorReadRequest(reader, state, promise));
528+
readRequest.promise = promise;
529+
readableStreamDefaultReaderRead(reader, readRequest);
524530
return promise.promise;
525531
}
526532

@@ -574,28 +580,38 @@ class ReadableStream {
574580
}
575581
// No read is in flight. Mirror the buffered fast path of
576582
// ReadableStreamDefaultReader.read(): when data is already queued
577-
// in a default controller, resolve immediately without allocating
578-
// a read request. The result settles synchronously, so leaving
583+
// in the controller, resolve immediately without allocating a
584+
// read request. The result settles synchronously, so leaving
579585
// state.current undefined matches the state the slow path reaches
580586
// once its read request callbacks have settled.
581587
const stream = reader[kState].stream;
582-
if (!state.done && stream !== undefined) {
588+
if (!state.done && stream !== undefined &&
589+
stream[kState].state === 'readable') {
583590
const controller = stream[kState].controller;
584-
if (stream[kState].state === 'readable' &&
585-
isReadableStreamDefaultController(controller) &&
586-
controller[kState].queue.length > 0) {
587-
stream[kState].disturbed = true;
588-
const chunk = dequeueValue(controller);
589-
590-
if (controller[kState].closeRequested &&
591-
!controller[kState].queue.length) {
592-
readableStreamDefaultControllerClearAlgorithms(controller);
593-
readableStreamClose(stream);
594-
} else {
595-
readableStreamDefaultControllerCallPullIfNeeded(controller);
591+
if (isReadableStreamDefaultController(controller)) {
592+
if (controller[kState].queue.length > 0) {
593+
stream[kState].disturbed = true;
594+
const chunk = dequeueValue(controller);
595+
596+
if (controller[kState].closeRequested &&
597+
!controller[kState].queue.length) {
598+
readableStreamDefaultControllerClearAlgorithms(controller);
599+
readableStreamClose(stream);
600+
} else {
601+
readableStreamDefaultControllerCallPullIfNeeded(controller);
602+
}
603+
604+
return PromiseResolve({ done: false, value: chunk });
596605
}
606+
} else if (controller[kState].queueTotalSize > 0) {
607+
// Byte controller with buffered data: same shape as above via
608+
// the queue-filled arm of the byte controller's pull steps.
609+
stream[kState].disturbed = true;
610+
return PromiseResolve({
611+
done: false,
597612

598-
return PromiseResolve({ done: false, value: chunk });
613+
value: readableByteStreamControllerDequeueChunk(controller),
614+
});
599615
}
600616
}
601617
state.current = nextSteps();
@@ -918,24 +934,36 @@ class ReadableStreamDefaultReader {
918934
const stream = this[kState].stream;
919935
const controller = stream[kState].controller;
920936

921-
// Fast path: if data is already buffered in a default controller,
937+
// Fast path: if data is already buffered in the controller's queue,
922938
// return a resolved promise immediately without creating a read request.
923939
// This is spec-compliant because read() returns a Promise, and
924940
// Promise.resolve() callbacks still run in the microtask queue.
925-
if (stream[kState].state === 'readable' &&
926-
isReadableStreamDefaultController(controller) &&
927-
controller[kState].queue.length > 0) {
928-
stream[kState].disturbed = true;
929-
const chunk = dequeueValue(controller);
941+
if (stream[kState].state === 'readable') {
942+
if (isReadableStreamDefaultController(controller)) {
943+
if (controller[kState].queue.length > 0) {
944+
stream[kState].disturbed = true;
945+
const chunk = dequeueValue(controller);
946+
947+
if (controller[kState].closeRequested && !controller[kState].queue.length) {
948+
readableStreamDefaultControllerClearAlgorithms(controller);
949+
readableStreamClose(stream);
950+
} else {
951+
readableStreamDefaultControllerCallPullIfNeeded(controller);
952+
}
930953

931-
if (controller[kState].closeRequested && !controller[kState].queue.length) {
932-
readableStreamDefaultControllerClearAlgorithms(controller);
933-
readableStreamClose(stream);
934-
} else {
935-
readableStreamDefaultControllerCallPullIfNeeded(controller);
954+
return PromiseResolve({ done: false, value: chunk });
955+
}
956+
} else if (controller[kState].queueTotalSize > 0) {
957+
// Byte controller with buffered data: mirror the queue-filled arm
958+
// of its pull steps (which never consults pendingPullIntos) minus
959+
// the read request.
960+
stream[kState].disturbed = true;
961+
return PromiseResolve({
962+
done: false,
963+
964+
value: readableByteStreamControllerDequeueChunk(controller),
965+
});
936966
}
937-
938-
return PromiseResolve({ done: false, value: chunk });
939967
}
940968

941969
// Slow path: create request and go through normal flow
@@ -3044,9 +3072,23 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
30443072
}
30453073
}
30463074

3047-
if (readableStreamHasDefaultReader(stream)) {
3048-
readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
3049-
if (!readableStreamGetNumReadRequests(stream)) {
3075+
// Single consolidated pass over the reader state. The spec routes this
3076+
// through HasDefaultReader / ProcessReadRequestsUsingQueue /
3077+
// GetNumReadRequests / FulfillReadRequest, which would re-run the same
3078+
// reader brand check and re-load the read request list four times on
3079+
// this per-chunk path.
3080+
const { reader } = stream[kState];
3081+
if (reader !== undefined &&
3082+
reader[kState] !== undefined &&
3083+
reader[kType] === 'ReadableStreamDefaultReader') {
3084+
const { readRequests } = reader[kState];
3085+
if (readRequests.length && controller[kState].queueTotalSize > 0) {
3086+
// Only possible when data was enqueued while the stream was not
3087+
// being read; read requests otherwise never coexist with a
3088+
// non-empty queue.
3089+
readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
3090+
}
3091+
if (!readRequests.length) {
30503092
readableByteStreamControllerEnqueueChunkToQueue(
30513093
controller,
30523094
transferredBuffer,
@@ -3060,7 +3102,8 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
30603102
}
30613103
const transferredView =
30623104
new Uint8Array(transferredBuffer, byteOffset, byteLength);
3063-
readableStreamFulfillReadRequest(stream, transferredView, false);
3105+
const readRequest = ArrayPrototypeShift(readRequests);
3106+
readRequest[kChunk](transferredView);
30643107
}
30653108
} else if (readableStreamHasBYOBReader(stream)) {
30663109
readableByteStreamControllerEnqueueChunkToQueue(
@@ -3395,22 +3438,28 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
33953438
return result;
33963439
}
33973440

3398-
function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
3399-
const {
3400-
queue,
3401-
queueTotalSize,
3402-
} = controller[kState];
3403-
assert(queueTotalSize > 0);
3441+
// Dequeues the first chunk of the byte queue as a Uint8Array view,
3442+
// handling queue drain (close-on-empty or pull) before the view is
3443+
// created. This is the [[queueTotalSize]] > 0 arm of the byte
3444+
// controller's pull steps; it is also called directly from the
3445+
// buffered fast paths in ReadableStreamDefaultReader.read() and the
3446+
// async iterator, which resolve with the view without allocating a
3447+
// read request.
3448+
function readableByteStreamControllerDequeueChunk(controller) {
3449+
assert(controller[kState].queueTotalSize > 0);
34043450
const {
34053451
buffer,
34063452
byteOffset,
34073453
byteLength,
3408-
} = ArrayPrototypeShift(queue);
3454+
} = ArrayPrototypeShift(controller[kState].queue);
34093455

34103456
controller[kState].queueTotalSize -= byteLength;
34113457
readableByteStreamControllerHandleQueueDrain(controller);
3412-
const view = new Uint8Array(buffer, byteOffset, byteLength);
3413-
readRequest[kChunk](view);
3458+
return new Uint8Array(buffer, byteOffset, byteLength);
3459+
}
3460+
3461+
function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
3462+
readRequest[kChunk](readableByteStreamControllerDequeueChunk(controller));
34143463
}
34153464

34163465
function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {

‎lib/internal/webstreams/util.js‎

Lines changed: 25 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,14 +7,19 @@ const {
77
ArrayPrototypePush,
88
ArrayPrototypeShift,
99
AsyncIteratorPrototype,
10+
DataViewPrototypeGetBuffer,
11+
DataViewPrototypeGetByteLength,
12+
DataViewPrototypeGetByteOffset,
1013
FunctionPrototypeCall,
1114
MathMax,
1215
NumberIsNaN,
1316
PromisePrototypeThen,
1417
PromiseReject,
1518
PromiseResolve,
16-
ReflectGet,
1719
Symbol,
20+
TypedArrayPrototypeGetBuffer,
21+
TypedArrayPrototypeGetByteLength,
22+
TypedArrayPrototypeGetByteOffset,
1823
Uint8Array,
1924
} = primordials;
2025

@@ -41,6 +46,10 @@ const {
4146

4247
const assert = require('internal/assert');
4348

49+
const {
50+
isDataView,
51+
} = require('internal/util/types');
52+
4453
const {
4554
validateFunction,
4655
} = require('internal/validators');
@@ -93,20 +102,29 @@ function customInspect(depth, options, name, data) {
93102
return `${name} ${inspect(data, opts)}`;
94103
}
95104

96-
// These are defensive to work around the possibility that
97-
// the buffer, byteLength, and byteOffset properties on
98-
// ArrayBuffer and ArrayBufferView's may have been tampered with.
105+
// These use the original prototype getters so that user tampering with
106+
// the buffer, byteLength, and byteOffset properties on ArrayBuffer and
107+
// ArrayBufferView's is not observed. They run once or more per chunk on
108+
// every byte-stream path, so they must not go through a reflective get
109+
// (the previous view.constructor.prototype lookup was both slower and
110+
// spoofable via a user-defined .constructor).
99111

100112
function ArrayBufferViewGetBuffer(view) {
101-
return ReflectGet(view.constructor.prototype, 'buffer', view);
113+
return isDataView(view) ?
114+
DataViewPrototypeGetBuffer(view) :
115+
TypedArrayPrototypeGetBuffer(view);
102116
}
103117

104118
function ArrayBufferViewGetByteLength(view) {
105-
return ReflectGet(view.constructor.prototype, 'byteLength', view);
119+
return isDataView(view) ?
120+
DataViewPrototypeGetByteLength(view) :
121+
TypedArrayPrototypeGetByteLength(view);
106122
}
107123

108124
function ArrayBufferViewGetByteOffset(view) {
109-
return ReflectGet(view.constructor.prototype, 'byteOffset', view);
125+
return isDataView(view) ?
126+
DataViewPrototypeGetByteOffset(view) :
127+
TypedArrayPrototypeGetByteOffset(view);
110128
}
111129

112130
function cloneAsUint8Array(view) {

0 commit comments

Comments
 (0)