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

stream: use the ring buffer for WHATWG stream request queues · nodejs/node@0ddfc6a · GitHub

/ node Public

Commit 0ddfc6a

Browse files
mcollina
authored and
aduh95
committed
stream: use the ring buffer for WHATWG stream request queues
The [[queue]] backing the controllers became a ring buffer, but the read and write request queues (readRequests, readIntoRequests, writeRequests) were left as plain arrays consumed with ArrayPrototypeShift, which is O(n) and, even for the single pending request of the await-each regime, far slower than an indexed head advance. Back them with the same Queue, materialized lazily from the shared empty queue so acquiring a reader or constructing a writer allocates no request storage until a read or write actually parks. pipe-to: +4.7% to +9.4% (all 16 configs, ***) readable-read type=byob: +2.3% (**) parked read loop: +15%, write loop: +11% (local harness) Follow-up to #64312. Signed-off-by: Matteo Collina <hello@matteocollina.com> PR-URL: #64431 Reviewed-By: James M Snell <jasnell@gmail.com> Reviewed-By: Antoine du Hamel <duhamelantoine1995@gmail.com>
1 parent 8fe5807 commit 0ddfc6a

3 files changed

Lines changed: 54 additions & 38 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 39 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@ const {
9797
ArrayBufferViewGetByteLength,
9898
ArrayBufferViewGetByteOffset,
9999
AsyncIterator,
100+
Queue,
100101
canCopyArrayBuffer,
101102
cloneAsUint8Array,
102103
copyArrayBuffer,
@@ -2192,9 +2193,9 @@ function readableStreamCancel(stream, reason) {
21922193
reader,
21932194
} = stream[kState];
21942195
if (reader !== undefined && readableStreamHasBYOBReader(stream)) {
2195-
for (let n = 0; n < reader[kState].readIntoRequests.length; n++)
2196-
reader[kState].readIntoRequests[n][kClose]();
2197-
reader[kState].readIntoRequests = [];
2196+
const readIntoRequests = reader[kState].readIntoRequests;
2197+
while (readIntoRequests.length)
2198+
readIntoRequests.shift()[kClose]();
21982199
}
21992200

22002201
return PromisePrototypeThen(
@@ -2216,9 +2217,9 @@ function readableStreamClose(stream) {
22162217
reader[kState].close?.resolve();
22172218

22182219
if (readableStreamHasDefaultReader(stream)) {
2219-
for (let n = 0; n < reader[kState].readRequests.length; n++)
2220-
reader[kState].readRequests[n][kClose]();
2221-
reader[kState].readRequests = [];
2220+
const readRequests = reader[kState].readRequests;
2221+
while (readRequests.length)
2222+
readRequests.shift()[kClose]();
22222223
}
22232224
}
22242225

@@ -2246,14 +2247,14 @@ function readableStreamError(stream, error) {
22462247
}
22472248

22482249
if (readableStreamHasDefaultReader(stream)) {
2249-
for (let n = 0; n < reader[kState].readRequests.length; n++)
2250-
reader[kState].readRequests[n][kError](error);
2251-
reader[kState].readRequests = [];
2250+
const readRequests = reader[kState].readRequests;
2251+
while (readRequests.length)
2252+
readRequests.shift()[kError](error);
22522253
} else {
22532254
assert(readableStreamHasBYOBReader(stream));
2254-
for (let n = 0; n < reader[kState].readIntoRequests.length; n++)
2255-
reader[kState].readIntoRequests[n][kError](error);
2256-
reader[kState].readIntoRequests = [];
2255+
const readIntoRequests = reader[kState].readIntoRequests;
2256+
while (readIntoRequests.length)
2257+
readIntoRequests.shift()[kError](error);
22572258
}
22582259
}
22592260

@@ -2297,7 +2298,7 @@ function readableStreamFulfillReadRequest(stream, chunk, done) {
22972298
reader,
22982299
} = stream[kState];
22992300
assert(reader[kState].readRequests.length);
2300-
const readRequest = ArrayPrototypeShift(reader[kState].readRequests);
2301+
const readRequest = reader[kState].readRequests.shift();
23012302

23022303
// TODO(@jasnell): It's not clear under what exact conditions done
23032304
// will be true here. The spec requires this check but none of the
@@ -2315,7 +2316,7 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done) {
23152316
reader,
23162317
} = stream[kState];
23172318
assert(reader[kState].readIntoRequests.length);
2318-
const readIntoRequest = ArrayPrototypeShift(reader[kState].readIntoRequests);
2319+
const readIntoRequest = reader[kState].readIntoRequests.shift();
23192320
if (done)
23202321
readIntoRequest[kClose](chunk);
23212322
else
@@ -2325,15 +2326,21 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done) {
23252326
function readableStreamAddReadRequest(stream, readRequest) {
23262327
assert(readableStreamHasDefaultReader(stream));
23272328
assert(stream[kState].state === 'readable');
2328-
ArrayPrototypePush(stream[kState].reader[kState].readRequests, readRequest);
2329+
const readerState = stream[kState].reader[kState];
2330+
let readRequests = readerState.readRequests;
2331+
if (readRequests === kEmptyQueue)
2332+
readRequests = readerState.readRequests = new Queue();
2333+
readRequests.push(readRequest);
23292334
}
23302335

23312336
function readableStreamAddReadIntoRequest(stream, readIntoRequest) {
23322337
assert(readableStreamHasBYOBReader(stream));
23332338
assert(stream[kState].state !== 'errored');
2334-
ArrayPrototypePush(
2335-
stream[kState].reader[kState].readIntoRequests,
2336-
readIntoRequest);
2339+
const readerState = stream[kState].reader[kState];
2340+
let readIntoRequests = readerState.readIntoRequests;
2341+
if (readIntoRequests === kEmptyQueue)
2342+
readIntoRequests = readerState.readIntoRequests = new Queue();
2343+
readIntoRequests.push(readIntoRequest);
23372344
}
23382345

23392346
function readableStreamReaderGenericCancel(reader, reason) {
@@ -2405,10 +2412,9 @@ function readableStreamDefaultReaderRelease(reader) {
24052412
}
24062413

24072414
function readableStreamDefaultReaderErrorReadRequests(reader, e) {
2408-
for (let n = 0; n < reader[kState].readRequests.length; ++n) {
2409-
reader[kState].readRequests[n][kError](e);
2410-
}
2411-
reader[kState].readRequests = [];
2415+
const readRequests = reader[kState].readRequests;
2416+
while (readRequests.length)
2417+
readRequests.shift()[kError](e);
24122418
}
24132419

24142420
function readableStreamBYOBReaderRelease(reader) {
@@ -2420,10 +2426,9 @@ function readableStreamBYOBReaderRelease(reader) {
24202426
}
24212427

24222428
function readableStreamBYOBReaderErrorReadIntoRequests(reader, e) {
2423-
for (let n = 0; n < reader[kState].readIntoRequests.length; ++n) {
2424-
reader[kState].readIntoRequests[n][kError](e);
2425-
}
2426-
reader[kState].readIntoRequests = [];
2429+
const readIntoRequests = reader[kState].readIntoRequests;
2430+
while (readIntoRequests.length)
2431+
readIntoRequests.shift()[kError](e);
24272432
}
24282433

24292434
function readableStreamReaderGenericRelease(reader) {
@@ -2495,14 +2500,18 @@ function setupReadableStreamBYOBReader(reader, stream) {
24952500
if (!isReadableByteStreamController(controller))
24962501
throw new ERR_INVALID_ARG_VALUE('stream', stream, 'must be a byte stream');
24972502
readableStreamReaderGenericInitialize(reader, stream);
2498-
reader[kState].readIntoRequests = [];
2503+
// The read-request queues use the same ring buffer as [[queue]], drained
2504+
// from a moving head rather than with ArrayPrototypeShift. Start from the
2505+
// shared immutable empty queue so acquiring a reader allocates no request
2506+
// storage until a read actually parks.
2507+
reader[kState].readIntoRequests = kEmptyQueue;
24992508
}
25002509

25012510
function setupReadableStreamDefaultReader(reader, stream) {
25022511
if (isReadableStreamLocked(stream))
25032512
throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
25042513
readableStreamReaderGenericInitialize(reader, stream);
2505-
reader[kState].readRequests = [];
2514+
reader[kState].readRequests = kEmptyQueue;
25062515
}
25072516

25082517
function readableStreamDefaultControllerClose(controller) {
@@ -3147,7 +3156,7 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
31473156
}
31483157
const transferredView =
31493158
new Uint8Array(transferredBuffer, byteOffset, byteLength);
3150-
const readRequest = ArrayPrototypeShift(readRequests);
3159+
const readRequest = readRequests.shift();
31513160
readRequest[kChunk](transferredView);
31523161
}
31533162
} else if (readableStreamHasBYOBReader(stream)) {
@@ -3520,7 +3529,7 @@ function readableByteStreamControllerProcessReadRequestsUsingQueue(controller) {
35203529
}
35213530
readableByteStreamControllerFillReadRequestFromQueue(
35223531
controller,
3523-
ArrayPrototypeShift(reader[kState].readRequests),
3532+
reader[kState].readRequests.shift(),
35243533
);
35253534
}
35263535
}

‎lib/internal/webstreams/util.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -404,6 +404,7 @@ module.exports = {
404404
ArrayBufferViewGetByteLength,
405405
ArrayBufferViewGetByteOffset,
406406
AsyncIterator,
407+
Queue,
407408
canCopyArrayBuffer,
408409
cloneAsUint8Array,
409410
copyArrayBuffer,

‎lib/internal/webstreams/writablestream.js‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
'use strict';
22

33
const {
4-
ArrayPrototypePush,
5-
ArrayPrototypeShift,
64
FunctionPrototypeBind,
75
FunctionPrototypeCall,
86
ObjectDefineProperties,
@@ -55,6 +53,7 @@ const {
5553
} = require('internal/worker/js_transferable');
5654

5755
const {
56+
Queue,
5857
createPromiseCallbackNoParams,
5958
createPromiseCallback1Param,
6059
createPromiseCallback2Params,
@@ -613,7 +612,10 @@ function createWritableStreamState() {
613612
controller: undefined,
614613
state: 'writable',
615614
storedError: undefined,
616-
writeRequests: [],
615+
// Ring-buffer request queue, materialized lazily on the first pending
616+
// write (see writableStreamAddWriteRequest) so construction allocates
617+
// no request storage.
618+
writeRequests: kEmptyQueue,
617619
writer: undefined,
618620
transfer: {
619621
__proto__: null,
@@ -823,7 +825,7 @@ function writableStreamRejectCloseAndClosedPromiseIfNeeded(stream) {
823825
function writableStreamMarkFirstWriteRequestInFlight(stream) {
824826
assert(stream[kState].inFlightWriteRequest.promise === undefined);
825827
assert(stream[kState].writeRequests.length);
826-
const writeRequest = ArrayPrototypeShift(stream[kState].writeRequests);
828+
const writeRequest = stream[kState].writeRequests.shift();
827829
stream[kState].inFlightWriteRequest = writeRequest;
828830
}
829831

@@ -901,9 +903,9 @@ function writableStreamFinishErroring(stream) {
901903
stream[kState].state = 'errored';
902904
stream[kState].controller[kError]();
903905
const storedError = stream[kState].storedError;
904-
for (let n = 0; n < stream[kState].writeRequests.length; n++)
905-
stream[kState].writeRequests[n].reject(storedError);
906-
stream[kState].writeRequests = [];
906+
const writeRequests = stream[kState].writeRequests;
907+
while (writeRequests.length)
908+
writeRequests.shift().reject(storedError);
907909

908910
if (stream[kState].pendingAbortRequest.abort.promise === undefined) {
909911
writableStreamRejectCloseAndClosedPromiseIfNeeded(stream);
@@ -952,7 +954,11 @@ function writableStreamAddWriteRequest(stream) {
952954
// PromiseWithResolvers() already returns a { promise, resolve, reject }
953955
// record, so push it as-is instead of rebuilding an identical object.
954956
const writeRequest = PromiseWithResolvers();
955-
ArrayPrototypePush(stream[kState].writeRequests, writeRequest);
957+
const streamState = stream[kState];
958+
let writeRequests = streamState.writeRequests;
959+
if (writeRequests === kEmptyQueue)
960+
writeRequests = streamState.writeRequests = new Queue();
961+
writeRequests.push(writeRequest);
956962
return writeRequest.promise;
957963
}
958964

0 commit comments

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