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

stream: use ring buffer for WHATWG stream queues · nodejs/node@cbb2568 · GitHub

/ node Public

Commit cbb2568

Browse files
anonrig
authored and
aduh95
committed
stream: use ring buffer for WHATWG stream queues
The [[queue]] backing every default readable/writable controller was a plain array of { value, size } wrappers consumed with ArrayPrototypeShift, so each buffered chunk allocated a wrapper object and each dequeue moved (or forced the engine to re-linearize) the remaining elements; the byte controller queue paid the same shift cost for its chunk descriptor records. Replace the array with a power-of-two ring buffer. Default controller queues store each entry as (value, size) in two consecutive slots, so the per-chunk wrapper allocation disappears; the byte controller keeps its descriptor records (they are mutated in place at the head) in single slots. Controllers start from (and are reset to) a shared immutable empty queue, so constructing a stream allocates no queue storage until a chunk is actually buffered. Enqueues measured by the internal default size algorithm (never observable by user code, always returns 1, cannot throw) skip the algorithm call and its try/catch entirely. The layout mirrors what Bun/WebKit use for the same spec structure: [[queue]] as a ring-buffer deque (WTF::Deque in Bun's src/jsc/bindings/webcore/streams/StreamQueue.h), the pure-JS ring buffer in Bun's src/js/internal/fifo.ts, and the trivial-size-algorithm bypass in src/jsc/bindings/webcore/streams/JSReadableStreamDefaultController.cpp. benchmark/compare.js against the unmodified baseline (30-run capture plus an independent 15-run repeat, Welch t-test, all p < 1e-5): webstreams/pipe-to.js +12-17% across all sixteen high-water-mark configurations, readable-read-buffered +20% (bufferSize=1) to +49% (bufferSize=1000), readable-async-iterator +21%. No stable significant regression across the rest of the webstreams suite: the creation.js and readable-read.js deltas seen in the full-suite capture disappear in isolated 60-run rechecks. Refs: https://github.com/oven-sh/bun/blob/main/src/js/internal/fifo.ts Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com> Assisted-by: Grok (Grok Build) PR-URL: #64312 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 67688cc commit cbb2568

4 files changed

Lines changed: 352 additions & 25 deletions

File tree

‎lib/internal/webstreams/readablestream.js‎

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -109,9 +109,11 @@ const {
109109
extractSizeAlgorithm,
110110
getNonWritablePropertyDescriptor,
111111
isBrandCheck,
112+
kEmptyQueue,
112113
kState,
113114
kType,
114115
lazyTransfer,
116+
materializeQueue,
115117
nonOpCancel,
116118
nonOpPull,
117119
nonOpStart,
@@ -2519,6 +2521,11 @@ function readableStreamDefaultControllerEnqueue(controller, chunk) {
25192521
reader[kType] === 'ReadableStreamDefaultReader' &&
25202522
reader[kState].readRequests.length) {
25212523
readableStreamFulfillReadRequest(stream, chunk, false);
2524+
} else if (controllerState.sizeAlgorithm === defaultSizeAlgorithm) {
2525+
// The internal default size algorithm is never observable by user
2526+
// code, always returns 1, and cannot throw: enqueue with the
2527+
// constant instead of calling it.
2528+
enqueueValueWithSize(controller, chunk, 1);
25222529
} else {
25232530
try {
25242531
const chunkSize =
@@ -2676,7 +2683,7 @@ function setupReadableStreamDefaultController(
26762683
pulling: false,
26772684
pullFulfilled: undefined,
26782685
pullRejected: undefined,
2679-
queue: [],
2686+
queue: kEmptyQueue,
26802687
queueTotalSize: 0,
26812688
started: false,
26822689
sizeAlgorithm,
@@ -3151,14 +3158,13 @@ function readableByteStreamControllerEnqueueChunkToQueue(
31513158
buffer,
31523159
byteOffset,
31533160
byteLength) {
3154-
ArrayPrototypePush(
3155-
controller[kState].queue,
3156-
{
3157-
buffer,
3158-
byteOffset,
3159-
byteLength,
3160-
});
3161-
controller[kState].queueTotalSize += byteLength;
3161+
const state = controller[kState];
3162+
materializeQueue(state).push({
3163+
buffer,
3164+
byteOffset,
3165+
byteLength,
3166+
});
3167+
state.queueTotalSize += byteLength;
31623168
}
31633169

31643170
function readableByteStreamControllerEnqueueDetachedPullIntoToQueue(
@@ -3213,7 +3219,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
32133219
} = controller[kState];
32143220

32153221
while (totalBytesToCopyRemaining) {
3216-
const headOfQueue = queue[0];
3222+
const headOfQueue = queue.peek();
32173223
const bytesToCopy = MathMin(
32183224
totalBytesToCopyRemaining,
32193225
headOfQueue.byteLength);
@@ -3231,7 +3237,7 @@ function readableByteStreamControllerFillPullIntoDescriptorFromQueue(
32313237
headOfQueue.byteOffset,
32323238
bytesToCopy);
32333239
if (headOfQueue.byteLength === bytesToCopy) {
3234-
ArrayPrototypeShift(queue);
3240+
queue.shift();
32353241
} else {
32363242
headOfQueue.byteOffset += bytesToCopy;
32373243
headOfQueue.byteLength -= bytesToCopy;
@@ -3447,7 +3453,7 @@ function readableByteStreamControllerDequeueChunk(controller) {
34473453
buffer,
34483454
byteOffset,
34493455
byteLength,
3450-
} = ArrayPrototypeShift(controller[kState].queue);
3456+
} = controller[kState].queue.shift();
34513457

34523458
controller[kState].queueTotalSize -= byteLength;
34533459
readableByteStreamControllerHandleQueueDrain(controller);
@@ -3543,7 +3549,7 @@ function setupReadableByteStreamController(
35433549
pullRejected: undefined,
35443550
started: false,
35453551
stream,
3546-
queue: [],
3552+
queue: kEmptyQueue,
35473553
queueTotalSize: 0,
35483554
highWaterMark,
35493555
pullAlgorithm,

‎lib/internal/webstreams/util.js‎

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

33
const {
4+
Array,
45
ArrayBufferPrototypeGetByteLength,
56
ArrayBufferPrototypeGetDetached,
67
ArrayBufferPrototypeSlice,
7-
ArrayPrototypePush,
8-
ArrayPrototypeShift,
98
AsyncIteratorPrototype,
109
DataViewPrototypeGetBuffer,
1110
DataViewPrototypeGetByteLength,
1211
DataViewPrototypeGetByteOffset,
1312
FunctionPrototypeCall,
1413
MathMax,
1514
NumberIsNaN,
15+
ObjectFreeze,
1616
PromisePrototypeThen,
1717
PromiseReject,
1818
PromiseResolve,
@@ -152,32 +152,167 @@ function isBrandCheck(brand) {
152152
};
153153
}
154154

155+
// Backing store for the spec's [[queue]]: a power-of-two ring buffer
156+
// instead of a plain array. Entries are pushed at the tail and consumed
157+
// at the head, so a plain array either moves every element or forces the
158+
// engine to re-linearize on each shift, and the default controllers would
159+
// additionally have to allocate a { value, size } wrapper object per
160+
// chunk just to keep the pair together. Default readable/writable
161+
// controller queues store each entry as (value, size) in two consecutive
162+
// slots via the *Pair methods; the readable byte controller queue stores
163+
// its chunk descriptor records in single slots via push/shift/peek. A
164+
// given instance only ever uses one of the two access patterns, so
165+
// head/tail stay aligned to the entry stride.
166+
class Queue {
167+
constructor(listLength = 8) {
168+
this.head = 0;
169+
this.tail = 0;
170+
// Number of logical entries currently in the queue: (value, size)
171+
// pairs for default controller queues, descriptor records for byte
172+
// controller queues.
173+
this.length = 0;
174+
this.capacityMask = listLength - 1;
175+
this.list = new Array(listLength);
176+
this.dequeuedSize = 0;
177+
}
178+
179+
// Single-slot entries (readable byte controller chunk records).
180+
181+
push(entry) {
182+
const tail = this.tail;
183+
this.list[tail] = entry;
184+
this.tail = (tail + 1) & this.capacityMask;
185+
this.length++;
186+
if (this.tail === this.head)
187+
this.grow();
188+
}
189+
190+
shift() {
191+
const head = this.head;
192+
const list = this.list;
193+
const entry = list[head];
194+
list[head] = undefined;
195+
this.head = (head + 1) & this.capacityMask;
196+
if (--this.length === 0)
197+
this.rewind();
198+
return entry;
199+
}
200+
201+
peek() {
202+
return this.list[this.head];
203+
}
204+
205+
// Two-slot (value, size) entries (default controller queues). The
206+
// stride is always 2 and capacities are even, so `tail + 1`/`head + 1`
207+
// never need to wrap.
208+
209+
pushPair(value, size) {
210+
const tail = this.tail;
211+
const list = this.list;
212+
list[tail] = value;
213+
list[tail + 1] = size;
214+
this.tail = (tail + 2) & this.capacityMask;
215+
this.length++;
216+
if (this.tail === this.head)
217+
this.grow();
218+
}
219+
220+
// Returns the dequeued value; the size of the same entry is left in
221+
// `this.dequeuedSize` so that callers can update [[queueTotalSize]]
222+
// without a per-entry wrapper object having to exist.
223+
shiftPair() {
224+
const head = this.head;
225+
const list = this.list;
226+
const value = list[head];
227+
this.dequeuedSize = list[head + 1];
228+
list[head] = undefined;
229+
list[head + 1] = undefined;
230+
this.head = (head + 2) & this.capacityMask;
231+
if (--this.length === 0)
232+
this.rewind();
233+
return value;
234+
}
235+
236+
peekPairValue() {
237+
return this.list[this.head];
238+
}
239+
240+
// The ring is completely full (the post-push tail caught up with the
241+
// head): double the capacity, re-linearizing from the head so index
242+
// arithmetic stays trivial.
243+
grow() {
244+
const list = this.list;
245+
const capacity = list.length;
246+
const head = this.head;
247+
if (head !== 0) {
248+
const relinearized = new Array(capacity * 2);
249+
let n = 0;
250+
for (let i = head; i < capacity; i++)
251+
relinearized[n++] = list[i];
252+
for (let i = 0; i < head; i++)
253+
relinearized[n++] = list[i];
254+
this.list = relinearized;
255+
this.head = 0;
256+
} else {
257+
list.length = capacity * 2;
258+
}
259+
this.tail = capacity;
260+
this.capacityMask = (capacity * 2) - 1;
261+
}
262+
263+
// The queue just became empty: restart at slot 0 so shallow queues net
264+
// sequential slot access, and drop the enlarged backing store after a
265+
// large burst has fully drained.
266+
rewind() {
267+
this.head = 0;
268+
this.tail = 0;
269+
if (this.list.length > 1024) {
270+
this.list.length = 8;
271+
this.capacityMask = 0b111;
272+
}
273+
}
274+
}
275+
276+
// Controllers start out with (and are reset to) this shared immutable
277+
// empty queue, so constructing a stream never allocates queue storage;
278+
// a real Queue is materialized by the enqueue paths on first use. All
279+
// dequeue/peek paths are guarded by `.length` (or the equivalent
280+
// [[queueTotalSize]]) checks, so they can never observe the sentinel in
281+
// a mutating way; it never stores entries, so it gets a zero-length
282+
// backing list.
283+
const kEmptyQueue = ObjectFreeze(new Queue(0));
284+
285+
function materializeQueue(state) {
286+
const queue = state.queue;
287+
if (queue === kEmptyQueue)
288+
return state.queue = new Queue();
289+
return queue;
290+
}
291+
155292
// The queue helpers below run once per chunk on the hot paths of every
156293
// default readable/writable stream, so they load the controller state a
157294
// single time and don't assert the existence of the queue fields (both
158295
// are unconditionally initialized during controller setup and only ever
159296
// replaced wholesale).
160297
function dequeueValue(controller) {
161298
const state = controller[kState];
162-
assert(state.queue.length);
163-
const {
164-
value,
165-
size,
166-
} = ArrayPrototypeShift(state.queue);
167-
state.queueTotalSize = MathMax(0, state.queueTotalSize - size);
299+
const queue = state.queue;
300+
assert(queue.length);
301+
const value = queue.shiftPair();
302+
state.queueTotalSize = MathMax(0, state.queueTotalSize - queue.dequeuedSize);
168303
return value;
169304
}
170305

171306
function resetQueue(controller) {
172307
const state = controller[kState];
173-
state.queue = [];
308+
state.queue = kEmptyQueue;
174309
state.queueTotalSize = 0;
175310
}
176311

177312
function peekQueueValue(controller) {
178313
const state = controller[kState];
179314
assert(state.queue.length);
180-
return state.queue[0].value;
315+
return state.queue.peekPairValue();
181316
}
182317

183318
function enqueueValueWithSize(controller, value, size) {
@@ -188,7 +323,7 @@ function enqueueValueWithSize(controller, value, size) {
188323
coercedSize === Infinity) {
189324
throw new ERR_INVALID_ARG_VALUE.RangeError('size', size);
190325
}
191-
ArrayPrototypePush(state.queue, { value, size: coercedSize });
326+
materializeQueue(state).pushPair(value, coercedSize);
192327
state.queueTotalSize += coercedSize;
193328
}
194329

@@ -284,9 +419,11 @@ module.exports = {
284419
getNonWritablePropertyDescriptor,
285420
isBrandCheck,
286421
isPromisePending,
422+
kEmptyQueue,
287423
kState,
288424
kType,
289425
lazyTransfer,
426+
materializeQueue,
290427
nonOpCancel,
291428
nonOpFlush,
292429
nonOpPull,

‎lib/internal/webstreams/writablestream.js‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ const {
6767
getNonWritablePropertyDescriptor,
6868
isBrandCheck,
6969
isPromisePending,
70+
kEmptyQueue,
7071
kState,
7172
kType,
7273
lazyTransfer,
@@ -1177,6 +1178,11 @@ function writableStreamDefaultControllerGetChunkSize(controller, chunk) {
11771178
return 1;
11781179
}
11791180

1181+
// The internal default size algorithm is never observable by user
1182+
// code, always returns 1, and cannot throw: skip the call.
1183+
if (sizeAlgorithm === defaultSizeAlgorithm)
1184+
return 1;
1185+
11801186
try {
11811187
return FunctionPrototypeCall(
11821188
sizeAlgorithm,
@@ -1293,7 +1299,7 @@ function setupWritableStreamDefaultController(
12931299
abortAlgorithm,
12941300
closeAlgorithm,
12951301
highWaterMark,
1296-
queue: [],
1302+
queue: kEmptyQueue,
12971303
queueTotalSize: 0,
12981304
abortController: new AbortController(),
12991305
sizeAlgorithm,

0 commit comments

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