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

stream: speed up async iteration over WHATWG byte streams · nodejs/node@3043b2a · GitHub

/ node Public

Commit 3043b2a

Browse files
aduh95
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 9445e27 commit 3043b2a

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
@@ -497,6 +497,12 @@ class ReadableStream {
497497
current: undefined,
498498
};
499499
let started = false;
500+
// A single reusable read request: at most one read is ever in flight
501+
// (next() chains through state.current), and the request is consumed
502+
// before the next read starts, so only its promise record changes
503+
// per read.
504+
// eslint-disable-next-line no-use-before-define
505+
const readRequest = new ReadableStreamAsyncIteratorReadRequest(reader, state, undefined);
500506

501507
// The nextSteps function is not an async function in order
502508
// to make it more efficient. Because nextSteps explicitly
@@ -515,8 +521,8 @@ class ReadableStream {
515521
}
516522
const promise = PromiseWithResolvers();
517523

518-
// eslint-disable-next-line no-use-before-define
519-
readableStreamDefaultReaderRead(reader, new ReadableStreamAsyncIteratorReadRequest(reader, state, promise));
524+
readRequest.promise = promise;
525+
readableStreamDefaultReaderRead(reader, readRequest);
520526
return promise.promise;
521527
}
522528

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

594-
return PromiseResolve({ done: false, value: chunk });
609+
value: readableByteStreamControllerDequeueChunk(controller),
610+
});
595611
}
596612
}
597613
state.current = nextSteps();
@@ -914,24 +930,36 @@ class ReadableStreamDefaultReader {
914930
const stream = this[kState].stream;
915931
const controller = stream[kState].controller;
916932

917-
// Fast path: if data is already buffered in a default controller,
933+
// Fast path: if data is already buffered in the controller's queue,
918934
// return a resolved promise immediately without creating a read request.
919935
// This is spec-compliant because read() returns a Promise, and
920936
// Promise.resolve() callbacks still run in the microtask queue.
921-
if (stream[kState].state === 'readable' &&
922-
isReadableStreamDefaultController(controller) &&
923-
controller[kState].queue.length > 0) {
924-
stream[kState].disturbed = true;
925-
const chunk = dequeueValue(controller);
937+
if (stream[kState].state === 'readable') {
938+
if (isReadableStreamDefaultController(controller)) {
939+
if (controller[kState].queue.length > 0) {
940+
stream[kState].disturbed = true;
941+
const chunk = dequeueValue(controller);
942+
943+
if (controller[kState].closeRequested && !controller[kState].queue.length) {
944+
readableStreamDefaultControllerClearAlgorithms(controller);
945+
readableStreamClose(stream);
946+
} else {
947+
readableStreamDefaultControllerCallPullIfNeeded(controller);
948+
}
926949

927-
if (controller[kState].closeRequested && !controller[kState].queue.length) {
928-
readableStreamDefaultControllerClearAlgorithms(controller);
929-
readableStreamClose(stream);
930-
} else {
931-
readableStreamDefaultControllerCallPullIfNeeded(controller);
950+
return PromiseResolve({ done: false, value: chunk });
951+
}
952+
} else if (controller[kState].queueTotalSize > 0) {
953+
// Byte controller with buffered data: mirror the queue-filled arm
954+
// of its pull steps (which never consults pendingPullIntos) minus
955+
// the read request.
956+
stream[kState].disturbed = true;
957+
return PromiseResolve({
958+
done: false,
959+
960+
value: readableByteStreamControllerDequeueChunk(controller),
961+
});
932962
}
933-
934-
return PromiseResolve({ value: chunk, done: false });
935963
}
936964

937965
// Slow path: create request and go through normal flow
@@ -3040,9 +3068,23 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
30403068
}
30413069
}
30423070

3043-
if (readableStreamHasDefaultReader(stream)) {
3044-
readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
3045-
if (!readableStreamGetNumReadRequests(stream)) {
3071+
// Single consolidated pass over the reader state. The spec routes this
3072+
// through HasDefaultReader / ProcessReadRequestsUsingQueue /
3073+
// GetNumReadRequests / FulfillReadRequest, which would re-run the same
3074+
// reader brand check and re-load the read request list four times on
3075+
// this per-chunk path.
3076+
const { reader } = stream[kState];
3077+
if (reader !== undefined &&
3078+
reader[kState] !== undefined &&
3079+
reader[kType] === 'ReadableStreamDefaultReader') {
3080+
const { readRequests } = reader[kState];
3081+
if (readRequests.length && controller[kState].queueTotalSize > 0) {
3082+
// Only possible when data was enqueued while the stream was not
3083+
// being read; read requests otherwise never coexist with a
3084+
// non-empty queue.
3085+
readableByteStreamControllerProcessReadRequestsUsingQueue(controller);
3086+
}
3087+
if (!readRequests.length) {
30463088
readableByteStreamControllerEnqueueChunkToQueue(
30473089
controller,
30483090
transferredBuffer,
@@ -3056,7 +3098,8 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
30563098
}
30573099
const transferredView =
30583100
new Uint8Array(transferredBuffer, byteOffset, byteLength);
3059-
readableStreamFulfillReadRequest(stream, transferredView, false);
3101+
const readRequest = ArrayPrototypeShift(readRequests);
3102+
readRequest[kChunk](transferredView);
30603103
}
30613104
} else if (readableStreamHasBYOBReader(stream)) {
30623105
readableByteStreamControllerEnqueueChunkToQueue(
@@ -3391,22 +3434,28 @@ function readableByteStreamControllerCancelSteps(controller, reason) {
33913434
return result;
33923435
}
33933436

3394-
function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
3395-
const {
3396-
queue,
3397-
queueTotalSize,
3398-
} = controller[kState];
3399-
assert(queueTotalSize > 0);
3437+
// Dequeues the first chunk of the byte queue as a Uint8Array view,
3438+
// handling queue drain (close-on-empty or pull) before the view is
3439+
// created. This is the [[queueTotalSize]] > 0 arm of the byte
3440+
// controller's pull steps; it is also called directly from the
3441+
// buffered fast paths in ReadableStreamDefaultReader.read() and the
3442+
// async iterator, which resolve with the view without allocating a
3443+
// read request.
3444+
function readableByteStreamControllerDequeueChunk(controller) {
3445+
assert(controller[kState].queueTotalSize > 0);
34003446
const {
34013447
buffer,
34023448
byteOffset,
34033449
byteLength,
3404-
} = ArrayPrototypeShift(queue);
3450+
} = ArrayPrototypeShift(controller[kState].queue);
34053451

34063452
controller[kState].queueTotalSize -= byteLength;
34073453
readableByteStreamControllerHandleQueueDrain(controller);
3408-
const view = new Uint8Array(buffer, byteOffset, byteLength);
3409-
readRequest[kChunk](view);
3454+
return new Uint8Array(buffer, byteOffset, byteLength);
3455+
}
3456+
3457+
function readableByteStreamControllerFillReadRequestFromQueue(controller, readRequest) {
3458+
readRequest[kChunk](readableByteStreamControllerDequeueChunk(controller));
34103459
}
34113460

34123461
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)
Back | FazBrowse Home | New Git URL