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

stream: commit pull-into descriptors after filling from queue · nodejs/node@a4a8361 · GitHub

/ node Public

Commit a4a8361

Browse files
MattiasBuelens
authored and
aduh95
committed
stream: commit pull-into descriptors after filling from queue
Fixes: #56044 PR-URL: #56072 Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Matthew Aitken <maitken033380023@gmail.com>
1 parent f1c2d2f commit a4a8361

2 files changed

Lines changed: 47 additions & 19 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 36 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@ const {
9494
ArrayBufferViewGetByteLength,
9595
ArrayBufferViewGetByteOffset,
9696
AsyncIterator,
97+
canCopyArrayBuffer,
9798
cloneAsUint8Array,
9899
copyArrayBuffer,
99100
createPromiseCallback,
@@ -2552,6 +2553,15 @@ function readableByteStreamControllerCommitPullIntoDescriptor(stream, desc) {
25522553
}
25532554
}
25542555

2556+
function readableByteStreamControllerCommitPullIntoDescriptors(stream, descriptors) {
2557+
for (let i = 0; i < descriptors.length; ++i) {
2558+
readableByteStreamControllerCommitPullIntoDescriptor(
2559+
stream,
2560+
descriptors[i],
2561+
);
2562+
}
2563+
}
2564+
25552565
function readableByteStreamControllerInvalidateBYOBRequest(controller) {
25562566
if (controller[kState].byobRequest === null)
25572567
return;
@@ -2758,11 +2768,11 @@ function readableByteStreamControllerRespondInClosedState(controller, desc) {
27582768
stream,
27592769
} = controller[kState];
27602770
if (readableStreamHasBYOBReader(stream)) {
2761-
while (readableStreamGetNumReadIntoRequests(stream) > 0) {
2762-
readableByteStreamControllerCommitPullIntoDescriptor(
2763-
stream,
2764-
readableByteStreamControllerShiftPendingPullInto(controller));
2771+
const filledPullIntos = [];
2772+
for (let i = 0; i < readableStreamGetNumReadIntoRequests(stream); ++i) {
2773+
ArrayPrototypePush(filledPullIntos, readableByteStreamControllerShiftPendingPullInto(controller));
27652774
}
2775+
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
27662776
}
27672777
}
27682778

@@ -2843,8 +2853,9 @@ function readableByteStreamControllerEnqueue(controller, chunk) {
28432853
transferredBuffer,
28442854
byteOffset,
28452855
byteLength);
2846-
readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
2856+
const filledPullIntos = readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
28472857
controller);
2858+
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
28482859
} else {
28492860
assert(!isReadableStreamLocked(stream));
28502861
readableByteStreamControllerEnqueueChunkToQueue(
@@ -2937,6 +2948,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
29372948
const maxAlignedBytes = maxBytesFilled - (maxBytesFilled % elementSize);
29382949
let totalBytesToCopyRemaining = maxBytesToCopy;
29392950
let ready = false;
2951+
assert(!ArrayBufferPrototypeGetDetached(buffer));
29402952
assert(bytesFilled < minimumFill);
29412953
if (maxAlignedBytes >= minimumFill) {
29422954
totalBytesToCopyRemaining = maxAlignedBytes - bytesFilled;
@@ -2952,12 +2964,12 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
29522964
totalBytesToCopyRemaining,
29532965
headOfQueue.byteLength);
29542966
const destStart = byteOffset + desc.bytesFilled;
2955-
const arrayBufferByteLength = ArrayBufferPrototypeGetByteLength(buffer);
2956-
if (arrayBufferByteLength - destStart < bytesToCopy) {
2957-
throw new ERR_INVALID_STATE.RangeError(
2958-
'view ArrayBuffer size is invalid');
2959-
}
2960-
assert(arrayBufferByteLength - destStart >= bytesToCopy);
2967+
assert(canCopyArrayBuffer(
2968+
buffer,
2969+
destStart,
2970+
headOfQueue.buffer,
2971+
headOfQueue.byteOffset,
2972+
bytesToCopy));
29612973
copyArrayBuffer(
29622974
buffer,
29632975
destStart,
@@ -2991,26 +3003,30 @@ function readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
29913003
const {
29923004
closeRequested,
29933005
pendingPullIntos,
2994-
stream,
29953006
} = controller[kState];
29963007
assert(!closeRequested);
3008+
const filledPullIntos = [];
29973009
while (pendingPullIntos.length) {
29983010
if (!controller[kState].queueTotalSize)
2999-
return;
3011+
break;
30003012
const desc = pendingPullIntos[0];
30013013
if (readableByteStreamControllerFillPullIntoDescriptorFromQueue(
30023014
controller,
30033015
desc)) {
30043016
readableByteStreamControllerShiftPendingPullInto(controller);
3005-
readableByteStreamControllerCommitPullIntoDescriptor(stream, desc);
3017+
ArrayPrototypePush(filledPullIntos, desc);
30063018
}
30073019
}
3020+
return filledPullIntos;
30083021
}
30093022

30103023
function readableByteStreamControllerRespondInReadableState(
30113024
controller,
30123025
bytesWritten,
30133026
desc) {
3027+
const {
3028+
stream,
3029+
} = controller[kState];
30143030
const {
30153031
buffer,
30163032
bytesFilled,
@@ -3031,9 +3047,10 @@ function readableByteStreamControllerRespondInReadableState(
30313047
controller,
30323048
desc,
30333049
);
3034-
readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
3050+
const filledPullIntos = readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(
30353051
controller,
30363052
);
3053+
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
30373054
return;
30383055
}
30393056

@@ -3059,10 +3076,10 @@ function readableByteStreamControllerRespondInReadableState(
30593076
ArrayBufferPrototypeGetByteLength(remainder));
30603077
}
30613078
desc.bytesFilled -= remainderSize;
3062-
readableByteStreamControllerCommitPullIntoDescriptor(
3063-
controller[kState].stream,
3064-
desc);
3065-
readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
3079+
const filledPullIntos = readableByteStreamControllerProcessPullIntoDescriptorsUsingQueue(controller);
3080+
3081+
readableByteStreamControllerCommitPullIntoDescriptor(stream, desc);
3082+
readableByteStreamControllerCommitPullIntoDescriptors(stream, filledPullIntos);
30663083
}
30673084

30683085
function readableByteStreamControllerRespondWithNewView(controller, view) {

‎lib/internal/webstreams/util.js‎

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

33
const {
4+
ArrayBufferPrototypeGetByteLength,
5+
ArrayBufferPrototypeGetDetached,
46
ArrayBufferPrototypeSlice,
57
ArrayPrototypePush,
68
ArrayPrototypeShift,
@@ -107,6 +109,14 @@ function cloneAsUint8Array(view) {
107109
);
108110
}
109111

112+
function canCopyArrayBuffer(toBuffer, toIndex, fromBuffer, fromIndex, count) {
113+
return toBuffer !== fromBuffer &&
114+
!ArrayBufferPrototypeGetDetached(toBuffer) &&
115+
!ArrayBufferPrototypeGetDetached(fromBuffer) &&
116+
toIndex + count <= ArrayBufferPrototypeGetByteLength(toBuffer) &&
117+
fromIndex + count <= ArrayBufferPrototypeGetByteLength(fromBuffer);
118+
}
119+
110120
function isBrandCheck(brand) {
111121
return (value) => {
112122
return value != null &&
@@ -261,6 +271,7 @@ module.exports = {
261271
ArrayBufferViewGetByteLength,
262272
ArrayBufferViewGetByteOffset,
263273
AsyncIterator,
274+
canCopyArrayBuffer,
264275
createPromiseCallback,
265276
cloneAsUint8Array,
266277
copyArrayBuffer,

0 commit comments

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