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

stream, quic: update iterable streams backpressure · nodejs/node@125c19d · GitHub

/ node Public

Commit 125c19d

Browse files
jasnell
authored and
aduh95
committed
stream, quic: update iterable streams backpressure
Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode/Opus PR-URL: #64464 Reviewed-By: Tim Perry <pimterry@gmail.com> Reviewed-By: Trivikram Kamat <trivikr.dev@gmail.com> Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>
1 parent 193091f commit 125c19d

28 files changed

Lines changed: 294 additions & 282 deletions

‎benchmark/streams/iter-throughput-share.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ const common = require('../common.js');
55
const bench = common.createBenchmark(main, {
66
consumers: [2, 8, 32],
77
batches: [1e4],
8-
backpressure: ['block'],
8+
backpressure: ['unbounded'],
99
n: [5],
1010
}, {
1111
flags: ['--experimental-stream-iter'],
@@ -24,7 +24,7 @@ async function main({ consumers, batches, backpressure, n }) {
2424

2525
bench.start();
2626
for (let i = 0; i < n; i++) {
27-
const shared = share(source(), { highWaterMark: 64, backpressure });
27+
const shared = share(source(), { budget: 65536, backpressure });
2828
const readers = Array.from({ length: consumers }, () => array(shared.pull()));
2929
await Promise.all(readers);
3030
}

‎doc/api/quic.md‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1276,7 +1276,7 @@ added: v23.8.0
12761276
interleaved with data from other streams of the same priority level.
12771277
When `false`, the stream should be completed before same-priority peers.
12781278
**Default:** `false`.
1279-
* `highWaterMark` {number} The maximum number of bytes that the writer
1279+
* `budget` {number} The maximum number of bytes that the writer
12801280
will buffer before `writeSync()` returns `false`. When the buffered
12811281
data exceeds this limit, the caller should wait for drain before
12821282
writing more. **Default:** `65536` (64 KB).
@@ -1317,7 +1317,7 @@ added: v23.8.0
13171317
interleaved with data from other streams of the same priority level.
13181318
When `false`, the stream should be completed before same-priority peers.
13191319
**Default:** `false`.
1320-
* `highWaterMark` {number} The maximum number of bytes that the writer
1320+
* `budget` {number} The maximum number of bytes that the writer
13211321
will buffer before `writeSync()` returns `false`. When the buffered
13221322
data exceeds this limit, the caller should wait for drain before
13231323
writing more. **Default:** `65536` (64 KB).
@@ -1924,7 +1924,7 @@ added: v23.8.0
19241924
The directionality of the stream, or `null` if the stream has been destroyed
19251925
or is still pending. Read only.
19261926

1927-
### `stream.highWaterMark`
1927+
### `stream.budget`
19281928

19291929
<!-- YAML
19301930
added: REPLACEME
@@ -2236,7 +2236,8 @@ The Writer has the following methods:
22362236
the QUIC transport-layer `INTERNAL_ERROR` (`0x1`) for raw QUIC).
22372237
See [`stream.destroy()`][] for a full-stream abort that also resets
22382238
the readable side via `STOP_SENDING`.
2239-
* `desiredSize` — Available capacity in bytes, or `null` if closed/errored.
2239+
* `canWrite` — `true` if writes will be accepted, `false` if at capacity,
2240+
or `null` if closed/errored.
22402241

22412242
The bytes from each `writeSync()` / `writevSync()` / `write()` / `writev()`
22422243
input chunk are copied into an internal buffer, so the caller's source

‎doc/api/stream_iter.md‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -223,7 +223,7 @@ How each policy uses these buffers:
223223

224224
| Policy | Buffer limit | Pending writes limit |
225225
| --------------- | ------------ | -------------------- |
226-
| `'strict'` | `budget` | `budget` |
226+
| `'strict'` | `budget` | 1 |
227227
| `'unbounded'` | `budget` | Unbounded |
228228
| `'drop-oldest'` | `budget` | N/A (never waits) |
229229
| `'drop-newest'` | `budget` | N/A (never waits) |
@@ -232,8 +232,8 @@ How each policy uses these buffers:
232232

233233
Strict mode catches "fire-and-forget" patterns where the producer calls
234234
`write()` without awaiting, which would cause unbounded memory growth.
235-
It limits both the buffer and the pending writes queue to
236-
`budget` bytes.
235+
It limits the buffer to `budget` bytes and the pending writes queue
236+
to a single entry.
237237

238238
If you properly await each write, you can only ever have one pending
239239
write at a time (yours), so you never hit the pending writes limit.

‎lib/internal/quic/quic.js‎

Lines changed: 25 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -297,8 +297,8 @@ const endpointRegistry = new SafeSet();
297297
* (e.g. HTTP/3).
298298
* @property {'high'|'default'|'low'} [priority] The priority level of the stream.
299299
* @property {boolean} [incremental] Whether to interleave data with same-priority streams.
300-
* @property {number} [highWaterMark] The high water mark for write
301-
* backpressure, in bytes. **Default:** `65536`.
300+
* @property {number} [budget] The byte budget for write backpressure.
301+
* **Default:** `65536`.
302302
* @property {OnHeadersCallback} [onheaders] Callback for incoming initial headers
303303
* @property {OnTrailersCallback} [ontrailers] Callback for incoming trailing headers
304304
* @property {OnInfoCallback} [oninfo] Callback for informational (1xx) headers
@@ -1325,7 +1325,7 @@ function applyCallbacks(session, cbs) {
13251325
* @param {QuicStream} stream The JS stream object
13261326
* @param {any} body The body source
13271327
*/
1328-
const kDefaultHighWaterMark = 65536;
1328+
const kDefaultBudget = 65536;
13291329
const kDefaultMaxPendingDatagrams = 128;
13301330

13311331
function configureOutbound(handle, stream, body) {
@@ -1405,20 +1405,20 @@ function configureOutbound(handle, stream, body) {
14051405
);
14061406
}
14071407

1408-
// Sets the high water mark and initial writeDesiredSize for a streaming
1408+
// Sets the budget and initial writeDesiredSize for a streaming
14091409
// outbound source. Called after handle.initStreamingSource() for both
14101410
// body-source and writer paths. One-shot body sources (string, Uint8Array,
14111411
// Blob, FileHandle, etc.) do not use this -- they go through attachSource
14121412
// and are not subject to backpressure.
14131413
function initStreamingBackpressure(stream) {
14141414
const state = getQuicStreamState(stream);
14151415
// Only set defaults if the user hasn't already configured them
1416-
// (e.g., via createBidirectionalStream({ highWaterMark: N })).
1417-
if (state.highWaterMark === 0) {
1418-
state.highWaterMark = kDefaultHighWaterMark;
1416+
// (e.g., via createBidirectionalStream({ budget: N })).
1417+
if (state.budget === 0) {
1418+
state.budget = kDefaultBudget;
14191419
}
14201420
if (state.writeDesiredSize === 0) {
1421-
state.writeDesiredSize = state.highWaterMark;
1421+
state.writeDesiredSize = state.budget;
14221422
}
14231423
}
14241424

@@ -1699,23 +1699,23 @@ class QuicStream {
16991699
}
17001700

17011701
/**
1702-
* The high water mark for write backpressure. When the total queued
1702+
* The byte budget for write backpressure. When the total queued
17031703
* outbound bytes exceeds this value, writeSync returns false and
1704-
* desiredSize drops to 0. Default is 65536 (64KB).
1704+
* canWrite returns false. Default is 65536 (64KB).
17051705
* @type {number}
17061706
*/
1707-
get highWaterMark() {
1707+
get budget() {
17081708
assertIsQuicStream(this);
1709-
return this.#inner.state.highWaterMark;
1709+
return this.#inner.state.budget;
17101710
}
17111711

1712-
set highWaterMark(val) {
1712+
set budget(val) {
17131713
assertIsQuicStream(this);
1714-
validateInteger(val, 'highWaterMark', 0, 0xFFFFFFFF);
1714+
validateInteger(val, 'budget', 0, 0xFFFFFFFF);
17151715
const inner = this.#inner;
1716-
inner.state.highWaterMark = val;
1716+
inner.state.budget = val;
17171717
// If writeDesiredSize hasn't been set yet (still 0 from initialization),
1718-
// initialize it to the highWaterMark so the first write can proceed.
1718+
// initialize it to the budget so the first write can proceed.
17191719
if (inner.state.writeDesiredSize === 0 && val > 0) {
17201720
inner.state.writeDesiredSize = val;
17211721
}
@@ -2163,8 +2163,8 @@ class QuicStream {
21632163
// will accept the data into the DataQueue and
21642164
// UpdateWriteDesiredSize() will drop writeDesiredSize toward 0,
21652165
// at which point the standard drain mechanism takes over.
2166-
// This follows the Web Streams model where writes beyond the HWM
2167-
// succeed and backpressure applies to *subsequent* writes.
2166+
// This follows the iter-streams model where writes beyond the
2167+
// budget succeed and backpressure applies to *subsequent* writes.
21682168
if (stream.#inner.state.writeDesiredSize === 0) return false;
21692169
const result = handle.write([chunk]);
21702170
if (result === undefined) return false;
@@ -2323,9 +2323,9 @@ class QuicStream {
23232323

23242324
const writer = {
23252325
__proto__: null,
2326-
get desiredSize() {
2326+
get canWrite() {
23272327
if (closed || errored || stream.#inner.state.writeEnded) return null;
2328-
return stream.#inner.state.writeDesiredSize;
2328+
return stream.#inner.state.writeDesiredSize > 0;
23292329
},
23302330
writeSync,
23312331
write,
@@ -3254,7 +3254,7 @@ class QuicSession {
32543254
body,
32553255
priority = 'default',
32563256
incremental = false,
3257-
highWaterMark = kDefaultHighWaterMark,
3257+
budget = kDefaultBudget,
32583258
headers,
32593259
onheaders,
32603260
ontrailers,
@@ -3290,8 +3290,8 @@ class QuicSession {
32903290
stream[kAttachFileHandle](body);
32913291
}
32923292

3293-
// Set the high water mark for backpressure.
3294-
stream.highWaterMark = highWaterMark;
3293+
// Set the byte budget for backpressure.
3294+
stream.budget = budget;
32953295

32963296
// Set stream callbacks before sending headers to avoid missing events.
32973297
if (onheaders) stream.onheaders = onheaders;
@@ -4047,8 +4047,8 @@ class QuicSession {
40474047
const stream = new QuicStream(kPrivateConstructor, handle, this, direction,
40484048
false /* isLocal */);
40494049

4050-
// Set the default high water mark for received streams.
4051-
stream.highWaterMark = kDefaultHighWaterMark;
4050+
// Set the default byte budget for received streams.
4051+
stream.budget = kDefaultBudget;
40524052

40534053
// A new stream was received. If we don't have an onstream callback, then
40544054
// there's nothing we can do about it. Destroy the stream in this case.

‎lib/internal/quic/state.js‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ const {
104104
IDX_STATE_STREAM_WANTS_TRAILERS,
105105
IDX_STATE_STREAM_RECEIVED_EARLY_DATA,
106106
IDX_STATE_STREAM_WRITE_DESIRED_SIZE,
107-
IDX_STATE_STREAM_HIGH_WATER_MARK,
107+
IDX_STATE_STREAM_BUDGET,
108108
IDX_STATE_STREAM_RESET_CODE,
109109
} = internalBinding('quic');
110110

@@ -871,18 +871,18 @@ class QuicStreamState {
871871
}
872872

873873
/** @type {number} */
874-
get highWaterMark() {
874+
get budget() {
875875
const handle = this.#handle;
876876
if (handle === undefined) return undefined;
877877
return DataViewPrototypeGetUint32(
878-
handle, this.#offset + IDX_STATE_STREAM_HIGH_WATER_MARK, kIsLittleEndian);
878+
handle, this.#offset + IDX_STATE_STREAM_BUDGET, kIsLittleEndian);
879879
}
880880

881-
set highWaterMark(val) {
881+
set budget(val) {
882882
const handle = this.#handle;
883883
if (handle === undefined) return;
884884
DataViewPrototypeSetUint32(
885-
handle, this.#offset + IDX_STATE_STREAM_HIGH_WATER_MARK, val, kIsLittleEndian);
885+
handle, this.#offset + IDX_STATE_STREAM_BUDGET, val, kIsLittleEndian);
886886
}
887887

888888
toString() {
@@ -908,7 +908,7 @@ class QuicStreamState {
908908
early,
909909
resetCode,
910910
writeDesiredSize,
911-
highWaterMark,
911+
budget,
912912
} = this;
913913
return {
914914
__proto__: null,
@@ -928,7 +928,7 @@ class QuicStreamState {
928928
early,
929929
resetCode,
930930
writeDesiredSize,
931-
highWaterMark,
931+
budget,
932932
};
933933
}
934934

@@ -965,7 +965,7 @@ class QuicStreamState {
965965
early,
966966
resetCode,
967967
writeDesiredSize,
968-
highWaterMark,
968+
budget,
969969
} = this;
970970

971971
return `QuicStreamState ${inspect({
@@ -985,7 +985,7 @@ class QuicStreamState {
985985
early,
986986
resetCode,
987987
writeDesiredSize,
988-
highWaterMark,
988+
budget,
989989
}, opts)}`;
990990
}
991991

‎lib/internal/streams/iter/broadcast.js‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,6 @@ const {
5555
const {
5656
kMultiConsumerDefaultBudget,
5757
kResolvedPromise,
58-
clampBudget,
5958
convertChunks,
6059
getWriterSignal,
6160
getMinCursor,
@@ -779,7 +778,7 @@ function broadcast(options = { __proto__: null }) {
779778

780779
const opts = {
781780
__proto__: null,
782-
budget: clampBudget(budget),
781+
budget,
783782
backpressure,
784783
signal,
785784
};

‎lib/internal/streams/iter/push.js‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@ const {
3737
const {
3838
kPushDefaultBudget,
3939
kResolvedPromise,
40-
clampBudget,
4140
onSignalAbort,
4241
toUint8Array,
4342
convertChunks,
@@ -99,7 +98,7 @@ class PushQueue {
9998
if (signal !== undefined) {
10099
validateAbortSignal(signal, 'options.signal');
101100
}
102-
this.#budget = clampBudget(budget);
101+
this.#budget = budget;
103102
this.#backpressure = backpressure;
104103
this.#signal = signal;
105104
this.#abortHandler = undefined;
@@ -127,7 +126,12 @@ class PushQueue {
127126
if (this.#writerState !== 'open' || this.#consumerState !== 'active') {
128127
return null;
129128
}
130-
return this.#bufferedBytes < this.#budget;
129+
if ((this.#backpressure === 'strict' ||
130+
this.#backpressure === 'unbounded') &&
131+
this.#bufferedBytes >= this.#budget) {
132+
return false;
133+
}
134+
return true;
131135
}
132136

133137
/**
@@ -156,6 +160,10 @@ class PushQueue {
156160

157161
const batchSize = this.#batchByteSize(chunks);
158162

163+
// Skip empty chunks -- zero-byte writes would accumulate infinitely
164+
// without ever triggering backpressure under a byte-budget model.
165+
if (batchSize === 0) return true;
166+
159167
if (this.#bufferedBytes >= this.#budget) {
160168
switch (this.#backpressure) {
161169
case 'strict':
@@ -181,6 +189,11 @@ class PushQueue {
181189
this.#bytesWritten += batchSize;
182190

183191
this.#resolvePendingReads();
192+
// After drop-oldest, evicting a large chunk may bring us under budget.
193+
// Resolve pending drains so writers waiting on backpressure can proceed.
194+
if (this.#bufferedBytes < this.#budget) {
195+
this.#resolvePendingDrains(true);
196+
}
184197
return true;
185198
}
186199

‎lib/internal/streams/iter/share.js‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@ const {
3636

3737
const {
3838
kMultiConsumerDefaultBudget,
39-
clampBudget,
4039
getMinCursor,
4140
hasProtocol,
4241
onSignalAbort,
@@ -696,7 +695,7 @@ function share(source, options = { __proto__: null }) {
696695

697696
const opts = {
698697
__proto__: null,
699-
budget: clampBudget(budget),
698+
budget,
700699
backpressure,
701700
signal,
702701
};
@@ -723,7 +722,7 @@ function shareSync(source, options = { __proto__: null }) {
723722

724723
const opts = {
725724
__proto__: null,
726-
budget: clampBudget(budget),
725+
budget,
727726
backpressure,
728727
};
729728

0 commit comments

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