Skip to content

stream: replace queue and readRequests with buffer list for performance improvement#49554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rluvaton wants to merge 3 commits into from
Closed

stream: replace queue and readRequests with buffer list for performance improvement #49554

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
240542c
stream: replace queue with buffer list for performance improvement
rluvaton Sep 8, 2023
4e39d86
format
rluvaton Sep 8, 2023
04f02a1
reduce number of chunks in pipe benchmark
rluvaton Sep 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion benchmark/webstreams/pipe-to.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -6,7 +6,7 @@ const{
} = require('node:stream/web');

const bench = common.createBenchmark(main,{
n: [5e6],
n: [1e5],
highWaterMarkR: [512, 1024, 2048, 4096],
highWaterMarkW: [512, 1024, 2048, 4096],
});
Expand Down
62 changes: 36 additions & 26 deletions lib/internal/webstreams/readablestream.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@ const{
SymbolToStringTag,
Uint8Array,
} = primordials;
const BufferList = require('internal/streams/buffer_list');

const{
AbortError,
Expand DownExpand Up@@ -813,6 +814,7 @@ class ReadIntoRequest{
get promise(){return this[kState].promise}
}


class ReadableStreamDefaultReader{
[kType] = 'ReadableStreamDefaultReader'

Expand All@@ -823,7 +825,7 @@ class ReadableStreamDefaultReader{
if (!isReadableStream(stream))
throw new ERR_INVALID_ARG_TYPE('stream', 'ReadableStream', stream);
this[kState] ={
readRequests: [],
readRequests: new BufferList(),
stream: undefined,
close:{
promise: undefined,
Expand DownExpand Up@@ -1958,9 +1960,12 @@ function readableStreamClose(stream){
reader[kState].close.resolve();

if (readableStreamHasDefaultReader(stream)){
for (let n = 0; n < reader[kState].readRequests.length; n++)
reader[kState].readRequests[n][kClose]();
reader[kState].readRequests = [];
let start = reader[kState].readRequests.head;
while (start !== null){
start.data[kClose]();
start = start.next;
}
reader[kState].readRequests.clear();
}
}

Expand All@@ -1982,9 +1987,12 @@ function readableStreamError(stream, error){
setPromiseHandled(reader[kState].close.promise);

if (readableStreamHasDefaultReader(stream)){
for (let n = 0; n < reader[kState].readRequests.length; n++)
reader[kState].readRequests[n][kError](error);
reader[kState].readRequests = [];
let start = reader[kState].readRequests.head;
while (start !== null){
start.data[kError](error);
start = start.next;
}
reader[kState].readRequests.clear();
} else{
assert(readableStreamHasBYOBReader(stream));
for (let n = 0; n < reader[kState].readIntoRequests.length; n++)
Expand DownExpand Up@@ -2033,7 +2041,7 @@ function readableStreamFulfillReadRequest(stream, chunk, done){
reader,
} = stream[kState];
assert(reader[kState].readRequests.length);
const readRequest = ArrayPrototypeShift(reader[kState].readRequests);
const readRequest = reader[kState].readRequests.shift();

// TODO(@jasnell): It's not clear under what exact conditions done
// will be true here. The spec requires this check but none of the
Expand DownExpand Up@@ -2061,7 +2069,7 @@ function readableStreamFulfillReadIntoRequest(stream, chunk, done){
function readableStreamAddReadRequest(stream, readRequest){
assert(readableStreamHasDefaultReader(stream));
assert(stream[kState].state === 'readable');
ArrayPrototypePush(stream[kState].reader[kState].readRequests, readRequest);
stream[kState].reader[kState].readRequests.push(readRequest);
}

function readableStreamAddReadIntoRequest(stream, readIntoRequest){
Expand DownExpand Up@@ -2114,10 +2122,12 @@ function readableStreamDefaultReaderRelease(reader){
}

function readableStreamDefaultReaderErrorReadRequests(reader, e){
for (let n = 0; n < reader[kState].readRequests.length; ++n){
reader[kState].readRequests[n][kError](e);
let start = reader[kState].readRequests.head;
while (start !== null){
start.data[kError](e);
start = start.next;
}
reader[kState].readRequests = [];
reader[kState].readRequests.clear();
}

function readableStreamBYOBReaderRelease(reader){
Expand DownExpand Up@@ -2210,7 +2220,7 @@ function setupReadableStreamDefaultReader(reader, stream){
if (isReadableStreamLocked(stream))
throw new ERR_INVALID_STATE.TypeError('ReadableStream is locked');
readableStreamReaderGenericInitialize(reader, stream);
reader[kState].readRequests = [];
reader[kState].readRequests.clear();
}

function readableStreamDefaultControllerClose(controller){
Expand DownExpand Up@@ -2379,7 +2389,7 @@ function setupReadableStreamDefaultController(
pullAgain: false,
pullAlgorithm,
pulling: false,
queue: [],
queue: new BufferList(),
queueTotalSize: 0,
started: false,
sizeAlgorithm,
Expand DownExpand Up@@ -2808,13 +2818,11 @@ function readableByteStreamControllerEnqueueChunkToQueue(
buffer,
byteOffset,
byteLength){
ArrayPrototypePush(
controller[kState].queue,
{
buffer,
byteOffset,
byteLength,
});
controller[kState].queue.push({
buffer,
byteOffset,
byteLength,
});
controller[kState].queueTotalSize += byteLength;
}

Expand DownExpand Up@@ -2868,7 +2876,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
} = controller[kState];

while (totalBytesToCopyRemaining){
const headOfQueue = queue[0];
const headOfQueue = queue.head.data;
const bytesToCopy = MathMin(
totalBytesToCopyRemaining,
headOfQueue.byteLength);
Expand All@@ -2886,7 +2894,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
headOfQueue.byteOffset,
bytesToCopy);
if (headOfQueue.byteLength === bytesToCopy){
ArrayPrototypeShift(queue);
queue.shift();
} else{
headOfQueue.byteOffset += bytesToCopy;
headOfQueue.byteLength -= bytesToCopy;
Expand DownExpand Up@@ -3087,7 +3095,7 @@ function readableByteStreamControllerFillReadRequestFromQueue(controller, readRe
buffer,
byteOffset,
byteLength,
} = ArrayPrototypeShift(queue);
} = queue.shift();

controller[kState].queueTotalSize -= byteLength;
readableByteStreamControllerHandleQueueDrain(controller);
Expand All@@ -3103,13 +3111,15 @@ function readableByteStreamControllerProcessReadRequestsUsingQueue(controller){
const{reader } = stream[kState];
assert(isReadableStreamDefaultReader(reader));

// TODO - may be able to change this to next next next
while (reader[kState].readRequests.length > 0){
// TODO - why this is in the while loop?
Comment on lines +3114 to +3116
Copy link
MemberAuthor

@rluvatonrluvatonSep 8, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

remove todos

if (queueTotalSize === 0){
return;
}
readableByteStreamControllerFillReadRequestFromQueue(
controller,
ArrayPrototypeShift(reader[kState].readRequests),
reader[kState].readRequests.shift(),
);
}
}
Expand DownExpand Up@@ -3177,7 +3187,7 @@ function setupReadableByteStreamController(
pulling: false,
started: false,
stream,
queue: [],
queue: new BufferList(),
queueTotalSize: 0,
highWaterMark,
pullAlgorithm,
Expand Down
11 changes: 5 additions & 6 deletions lib/internal/webstreams/util.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2,8 +2,6 @@

const{
ArrayBufferPrototypeSlice,
ArrayPrototypePush,
ArrayPrototypeShift,
AsyncIteratorPrototype,
FunctionPrototypeCall,
MathMax,
Expand DownExpand Up@@ -146,7 +144,7 @@ function dequeueValue(controller){
const{
value,
size,
} = ArrayPrototypeShift(controller[kState].queue);
} = controller[kState].queue.shift();
controller[kState].queueTotalSize =
MathMax(0, controller[kState].queueTotalSize - size);
return value;
Expand All@@ -155,15 +153,16 @@ function dequeueValue(controller){
function resetQueue(controller){
assert(controller[kState].queue !== undefined);
assert(controller[kState].queueTotalSize !== undefined);
controller[kState].queue = [];
controller[kState].queue.clear();
controller[kState].queueTotalSize = 0;
}

function peekQueueValue(controller){
assert(controller[kState].queue !== undefined);
assert(controller[kState].queueTotalSize !== undefined);
assert(controller[kState].queue.length);
return controller[kState].queue[0].value;
debugger;
Copy link
MemberAuthor

@rluvatonrluvatonSep 8, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

remove

return controller[kState].queue.head.data.value;
}

function enqueueValueWithSize(controller, value, size){
Expand All@@ -176,7 +175,7 @@ function enqueueValueWithSize(controller, value, size){
size === Infinity){
throw new ERR_INVALID_ARG_VALUE.RangeError('size', size);
}
ArrayPrototypePush(controller[kState].queue, {value, size });
controller[kState].queue.push({value, size });
controller[kState].queueTotalSize += size;
}

Expand Down
4 changes: 3 additions & 1 deletion lib/internal/webstreams/writablestream.js
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -78,6 +78,8 @@ const{

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

const BufferList = require('internal/streams/buffer_list');

const kAbort = Symbol('kAbort');
const kCloseSentinel = Symbol('kCloseSentinel');
const kError = Symbol('kError');
Expand DownExpand Up@@ -1269,7 +1271,7 @@ function setupWritableStreamDefaultController(
abortAlgorithm,
closeAlgorithm,
highWaterMark,
queue: [],
queue: new BufferList(),
queueTotalSize: 0,
abortController: new AbortController(),
sizeAlgorithm,
Expand Down