diff --git a/doc/api/quic.md b/doc/api/quic.md index a19ff2befb5731..8944de328c7a9f 100644 --- a/doc/api/quic.md +++ b/doc/api/quic.md @@ -1285,14 +1285,14 @@ added: v23.8.0 * `headers` {Object} Initial request or response headers to send. Only used when the session supports headers (e.g. HTTP/3). If `body` is not specified and `headers` is provided, the stream is treated as - headers-only (terminal). + headers-only (terminal). * `priority` {string} The priority level of the stream. One of `'high'`, `'default'`, or `'low'`. **Default:** `'default'`. * `incremental` {boolean} When `true`, data from this stream may be interleaved with data from other streams of the same priority level. When `false`, the stream should be completed before same-priority peers. **Default:** `false`. - * `highWaterMark` {number} The maximum number of bytes that the writer + * `budget` {number} The maximum number of bytes that the writer will buffer before `writeSync()` returns `false`. When the buffered data exceeds this limit, the caller should wait for drain before writing more. **Default:** `65536` (64 KB). @@ -1300,6 +1300,8 @@ added: v23.8.0 Called with `(headers)`. * `ontrailers` {Function} Callback for received trailing headers. Called with `(trailers)`. + + * `oninfo` {Function} Callback for received informational (1xx) headers. Called with `(headers)`. * `onwanttrailers` {Function} Callback when trailers should be sent. @@ -1333,7 +1335,7 @@ added: v23.8.0 interleaved with data from other streams of the same priority level. When `false`, the stream should be completed before same-priority peers. **Default:** `false`. - * `highWaterMark` {number} The maximum number of bytes that the writer + * `budget` {number} The maximum number of bytes that the writer will buffer before `writeSync()` returns `false`. When the buffered data exceeds this limit, the caller should wait for drain before writing more. **Default:** `65536` (64 KB). @@ -1940,7 +1942,7 @@ added: v23.8.0 The directionality of the stream, or `null` if the stream has been destroyed or is still pending. Read only. -### `stream.highWaterMark` +### `stream.budget` | slots |---> Consumer pulls + [ write() ] ----+ +--->| buffer |---> Consumer pulls [ write() ] | | | (bucket)| for await (...) [ write() ] v | +---------+ +--------+ ^ @@ -214,29 +214,29 @@ that closes when the bucket is full: 'strict' mode limits this too! ``` -* **Slots (the bucket)** -- data ready for the consumer, capped at - `highWaterMark`. When the consumer pulls, it drains all slots at once - into a single batch. +* **Buffer (the bucket)** -- data ready for the consumer, capped at + `budget` bytes. When the consumer pulls, it drains all buffered data + at once into a single batch. -* **Pending writes (the hose)** -- writes waiting for slot space. After +* **Pending writes (the hose)** -- writes waiting for buffer space. After the consumer drains, pending writes are promoted into the now-empty - slots and their promises settle. + buffer and their promises settle. How each policy uses these buffers: -| Policy | Slots limit | Pending writes limit | -| --------------- | --------------- | -------------------- | -| `'strict'` | `highWaterMark` | `highWaterMark` | -| `'block'` | `highWaterMark` | Unbounded | -| `'drop-oldest'` | `highWaterMark` | N/A (never waits) | -| `'drop-newest'` | `highWaterMark` | N/A (never waits) | +| Policy | Buffer limit | Pending writes limit | +| --------------- | ------------ | -------------------- | +| `'strict'` | `budget` | `budget` | +| `'unbounded'` | `budget` | Unbounded | +| `'drop-oldest'` | `budget` | N/A (never waits) | +| `'drop-newest'` | `budget` | N/A (never waits) | #### Strict (default) Strict mode catches "fire-and-forget" patterns where the producer calls `write()` without awaiting, which would cause unbounded memory growth. -It limits both the slots buffer and the pending writes queue to -`highWaterMark`. +It limits both the buffer and the pending writes queue to +`budget` bytes. If you properly await each write, you can only ever have one pending write at a time (yours), so you never hit the pending writes limit. @@ -246,7 +246,7 @@ overflows: ```mjs import { push, text } from 'node:stream/iter'; -const { writer, readable } = push({ highWaterMark: 16 }); +const { writer, readable } = push({ budget: 16384 }); // Consumer must run concurrently -- without it, the first write // that fills the buffer blocks the producer forever. @@ -265,7 +265,7 @@ console.log(await consuming); const { push, text } = require('node:stream/iter'); async function run() { - const { writer, readable } = push({ highWaterMark: 16 }); + const { writer, readable } = push({ budget: 16384 }); // Consumer must run concurrently -- without it, the first write // that fills the buffer blocks the producer forever. @@ -293,9 +293,9 @@ for (const item of dataset) { // --> throws "Backpressure violation: too many pending writes" ``` -#### Block +#### Unbounded -Block mode caps slots at `highWaterMark` but places no limit on the +Unbounded mode caps buffered bytes at `budget` but places no limit on the pending writes queue. Awaited writes block until the consumer makes room, just like strict mode. The difference is that unawaited writes silently queue forever instead of throwing -- a potential memory leak if the @@ -309,8 +309,8 @@ properly, or when migrating code from those APIs. import { push, text } from 'node:stream/iter'; const { writer, readable } = push({ - highWaterMark: 16, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consuming = text(readable); @@ -328,8 +328,8 @@ const { push, text } = require('node:stream/iter'); async function run() { const { writer, readable } = push({ - highWaterMark: 16, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consuming = text(readable); @@ -355,9 +355,9 @@ any scenario where stale data is less valuable than current data. ```mjs import { push } from 'node:stream/iter'; -// Keep only the 5 most recent readings +// Keep only the most recent ~16 KB of readings const { writer, readable } = push({ - highWaterMark: 5, + budget: 16384, backpressure: 'drop-oldest', }); ``` @@ -365,9 +365,9 @@ const { writer, readable } = push({ ```cjs const { push } = require('node:stream/iter'); -// Keep only the 5 most recent readings +// Keep only the most recent ~16 KB of readings const { writer, readable } = push({ - highWaterMark: 5, + budget: 16384, backpressure: 'drop-oldest', }); ``` @@ -382,9 +382,9 @@ shedding load under pressure. ```mjs import { push } from 'node:stream/iter'; -// Accept up to 10 buffered items; discard anything beyond that +// Accept up to 16 KB of buffered data; discard anything beyond that const { writer, readable } = push({ - highWaterMark: 10, + budget: 16384, backpressure: 'drop-newest', }); ``` @@ -392,9 +392,9 @@ const { writer, readable } = push({ ```cjs const { push } = require('node:stream/iter'); -// Accept up to 10 buffered items; discard anything beyond that +// Accept up to 16 KB of buffered data; discard anything beyond that const { writer, readable } = push({ - highWaterMark: 10, + budget: 16384, backpressure: 'drop-newest', }); ``` @@ -416,14 +416,16 @@ if (writer.endSync() < 0) await writer.end(); writer.fail(err); // Always synchronous, no fallback needed ``` -#### `writer.desiredSize` +#### `writer.canWrite` -* {number|null} +* {boolean|null} -The number of buffer slots available before the high water mark is reached. -Returns `null` if the writer is closed or the consumer has disconnected. +Returns `true` if the next write is likely to be accepted (buffered data is +below capacity), `false` if backpressure is active, or `null` if the writer +is closed or the consumer has disconnected. -The value is always non-negative. +This is a hint, not a guarantee: the state can change between the check and +the write. Use [`ondrain()`][] to wait for capacity rather than polling. #### `writer.end([options])` @@ -766,10 +768,10 @@ added: * `...transforms` {Function|Object} Optional transforms applied to the readable side. * `options` {Object} - * `highWaterMark` {number} Maximum number of buffered slots before - backpressure is applied. Must be >= 1; values below 1 are clamped to 1. - **Default:** `4`. - * `backpressure` {string} Backpressure policy: `'strict'`, `'block'`, + * `budget` {number} Maximum number of buffered bytes before + backpressure is applied. Must be >= 16384. + **Default:** `16384`. + * `backpressure` {string} Backpressure policy: `'strict'`, `'unbounded'`, `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`. * `signal` {AbortSignal} Abort the stream. * Returns: {Object} @@ -829,18 +831,18 @@ added: --> * `options` {Object} - * `highWaterMark` {number} Buffer size for both directions. - **Default:** `4`. + * `budget` {number} Buffer size in bytes for both directions. + **Default:** `16384`. * `backpressure` {string} Policy for both directions. **Default:** `'strict'`. * `signal` {AbortSignal} Cancellation signal for both channels. * `a` {Object} Options specific to the A-to-B direction. Overrides shared options. - * `highWaterMark` {number} + * `budget` {number} * `backpressure` {string} * `b` {Object} Options specific to the B-to-A direction. Overrides shared options. - * `highWaterMark` {number} + * `budget` {number} * `backpressure` {string} * Returns: {Array} A pair `[channelA, channelB]` of duplex channels. @@ -1079,9 +1081,10 @@ fulfills with `true` when the writer can accept more data. ```mjs import { push, ondrain, text } from 'node:stream/iter'; -const { writer, readable } = push({ highWaterMark: 2 }); -writer.writeSync('a'); -writer.writeSync('b'); +const { writer, readable } = push({ budget: 16384 }); +const chunk = new Uint8Array(8192); // 8 KB +writer.writeSync(chunk); +writer.writeSync(chunk); // 16 KB total -- buffer full // Start consuming so the buffer can actually drain const consuming = text(readable); @@ -1099,9 +1102,10 @@ await consuming; const { push, ondrain, text } = require('node:stream/iter'); async function run() { - const { writer, readable } = push({ highWaterMark: 2 }); - writer.writeSync('a'); - writer.writeSync('b'); + const { writer, readable } = push({ budget: 16384 }); + const chunk = new Uint8Array(8192); // 8 KB + writer.writeSync(chunk); + writer.writeSync(chunk); // 16 KB total -- buffer full // Start consuming so the buffer can actually drain const consuming = text(readable); @@ -1214,9 +1218,9 @@ added: --> * `options` {Object} - * `highWaterMark` {number} Buffer size in slots. Must be >= 1; values - below 1 are clamped to 1. **Default:** `16`. - * `backpressure` {string} `'strict'`, `'block'`, `'drop-oldest'`, or + * `budget` {number} Buffer size in bytes. Must be >= 16384. + **Default:** `65536`. + * `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`. * `signal` {AbortSignal} * Returns: {Object} @@ -1275,12 +1279,6 @@ async function run() { run().catch(console.error); ``` -#### `broadcast.bufferSize` - -* {number} - -The number of chunks currently buffered. - #### `broadcast.cancel([reason])` * `reason` {Error} @@ -1331,9 +1329,9 @@ added: * `source` {AsyncIterable} The source to share. * `options` {Object} - * `highWaterMark` {number} Buffer size. Must be >= 1; values below 1 - are clamped to 1. **Default:** `16`. - * `backpressure` {string} `'strict'`, `'block'`, `'drop-oldest'`, or + * `budget` {number} Buffer size in bytes. Must be >= 16384. + **Default:** `65536`. + * `backpressure` {string} `'strict'`, `'unbounded'`, `'drop-oldest'`, or `'drop-newest'`. **Default:** `'strict'`. * Returns: {Share} @@ -1373,12 +1371,6 @@ async function run() { run().catch(console.error); ``` -#### `share.bufferSize` - -* {number} - -The number of chunks currently buffered. - #### `share.cancel([reason])` * `reason` {Error} @@ -1426,8 +1418,8 @@ added: * `source` {Iterable} The sync source to share. * `options` {Object} - * `highWaterMark` {number} Must be >= 1; values below 1 are clamped - to 1. **Default:** `16`. + * `budget` {number} Must be >= 16384. + **Default:** `65536`. * `backpressure` {string} **Default:** `'strict'`. * Returns: {SyncShare} @@ -1528,7 +1520,7 @@ added: v26.1.0 * `backpressure` {string} Backpressure policy. **Default:** `'strict'`. * `'strict'` -- writes are rejected when the buffer is full. Catches callers that ignore backpressure. - * `'block'` -- writes wait for drain when the buffer is full. Recommended + * `'unbounded'` -- writes wait for drain when the buffer is full. Recommended for use with [`pipeTo()`][]. * `'drop-newest'` -- writes are silently discarded when the buffer is full. * `'drop-oldest'` -- **not supported**. Throws `ERR_INVALID_ARG_VALUE`. @@ -1561,7 +1553,7 @@ const writable = new Writable({ }); await pipeTo(from('hello world'), - fromWritable(writable, { backpressure: 'block' })); + fromWritable(writable, { backpressure: 'unbounded' })); ``` ```cjs @@ -1574,7 +1566,7 @@ async function run() { }); await pipeTo(from('hello world'), - fromWritable(writable, { backpressure: 'block' })); + fromWritable(writable, { backpressure: 'unbounded' })); } run(); ``` @@ -2097,6 +2089,7 @@ console.log(textSync(stream)); // 'hello world' [`from()`]: #frominput [`fromSync()`]: #fromsyncinput [`node:zlib/iter`]: zlib.md#iterable-compression +[`ondrain()`]: #ondraindrainable [`pipeTo()`]: #pipetosource-transforms-writer-options [`pull()`]: #pullsource-transforms-options [`pullSync()`]: #pullsyncsource-transforms-options diff --git a/lib/internal/quic/quic.js b/lib/internal/quic/quic.js index 96362355c1deba..bae19870179d8c 100644 --- a/lib/internal/quic/quic.js +++ b/lib/internal/quic/quic.js @@ -297,8 +297,8 @@ const endpointRegistry = new SafeSet(); * (e.g. HTTP/3). * @property {'high'|'default'|'low'} [priority] The priority level of the stream. * @property {boolean} [incremental] Whether to interleave data with same-priority streams. - * @property {number} [highWaterMark] The high water mark for write - * backpressure, in bytes. **Default:** `65536`. + * @property {number} [budget] The byte budget for write backpressure. + * **Default:** `65536`. * @property {OnHeadersCallback} [onheaders] Callback for incoming initial headers * @property {OnTrailersCallback} [ontrailers] Callback for incoming trailing headers * @property {OnInfoCallback} [oninfo] Callback for informational (1xx) headers @@ -1322,7 +1322,7 @@ function applyCallbacks(session, cbs) { * @param {QuicStream} stream The JS stream object * @param {any} body The body source */ -const kDefaultHighWaterMark = 65536; +const kDefaultBudget = 65536; const kDefaultMaxPendingDatagrams = 128; function configureOutbound(handle, stream, body) { @@ -1402,7 +1402,7 @@ function configureOutbound(handle, stream, body) { ); } -// Sets the high water mark and initial writeDesiredSize for a streaming +// Sets the budget and initial writeDesiredSize for a streaming // outbound source. Called after handle.initStreamingSource() for both // body-source and writer paths. One-shot body sources (string, Uint8Array, // Blob, FileHandle, etc.) do not use this -- they go through attachSource @@ -1410,12 +1410,12 @@ function configureOutbound(handle, stream, body) { function initStreamingBackpressure(stream) { const state = getQuicStreamState(stream); // Only set defaults if the user hasn't already configured them - // (e.g., via createBidirectionalStream({ highWaterMark: N })). - if (state.highWaterMark === 0) { - state.highWaterMark = kDefaultHighWaterMark; + // (e.g., via createBidirectionalStream({ budget: N })). + if (state.budget === 0) { + state.budget = kDefaultBudget; } if (state.writeDesiredSize === 0) { - state.writeDesiredSize = state.highWaterMark; + state.writeDesiredSize = state.budget; } } @@ -1696,23 +1696,23 @@ class QuicStream { } /** - * The high water mark for write backpressure. When the total queued + * The byte budget for write backpressure. When the total queued * outbound bytes exceeds this value, writeSync returns false and - * desiredSize drops to 0. Default is 65536 (64KB). + * canWrite returns false. Default is 65536 (64KB). * @type {number} */ - get highWaterMark() { + get budget() { assertIsQuicStream(this); - return this.#inner.state.highWaterMark; + return this.#inner.state.budget; } - set highWaterMark(val) { + set budget(val) { assertIsQuicStream(this); - validateInteger(val, 'highWaterMark', 0, 0xFFFFFFFF); + validateInteger(val, 'budget', 0, 0xFFFFFFFF); const inner = this.#inner; - inner.state.highWaterMark = val; + inner.state.budget = val; // If writeDesiredSize hasn't been set yet (still 0 from initialization), - // initialize it to the highWaterMark so the first write can proceed. + // initialize it to the budget so the first write can proceed. if (inner.state.writeDesiredSize === 0 && val > 0) { inner.state.writeDesiredSize = val; } @@ -2160,8 +2160,8 @@ class QuicStream { // will accept the data into the DataQueue and // UpdateWriteDesiredSize() will drop writeDesiredSize toward 0, // at which point the standard drain mechanism takes over. - // This follows the Web Streams model where writes beyond the HWM - // succeed and backpressure applies to *subsequent* writes. + // This follows the iter-streams model where writes beyond the + // budget succeed and backpressure applies to *subsequent* writes. if (stream.#inner.state.writeDesiredSize === 0) return false; const result = handle.write([chunk]); if (result === undefined) return false; @@ -2320,9 +2320,9 @@ class QuicStream { const writer = { __proto__: null, - get desiredSize() { + get canWrite() { if (closed || errored || stream.#inner.state.writeEnded) return null; - return stream.#inner.state.writeDesiredSize; + return stream.#inner.state.writeDesiredSize > 0; }, writeSync, write, @@ -3251,7 +3251,7 @@ class QuicSession { body, priority = 'default', incremental = false, - highWaterMark = kDefaultHighWaterMark, + budget = kDefaultBudget, headers, onheaders, ontrailers, @@ -3287,8 +3287,8 @@ class QuicSession { stream[kAttachFileHandle](body); } - // Set the high water mark for backpressure. - stream.highWaterMark = highWaterMark; + // Set the byte budget for backpressure. + stream.budget = budget; // Set stream callbacks before sending headers to avoid missing events. if (onheaders) stream.onheaders = onheaders; @@ -4044,8 +4044,8 @@ class QuicSession { const stream = new QuicStream(kPrivateConstructor, handle, this, direction, false /* isLocal */); - // Set the default high water mark for received streams. - stream.highWaterMark = kDefaultHighWaterMark; + // Set the default byte budget for received streams. + stream.budget = kDefaultBudget; // A new stream was received. If we don't have an onstream callback, then // there's nothing we can do about it. Destroy the stream in this case. diff --git a/lib/internal/quic/state.js b/lib/internal/quic/state.js index 8815b0bb4c32cf..27df0596163abf 100644 --- a/lib/internal/quic/state.js +++ b/lib/internal/quic/state.js @@ -104,7 +104,7 @@ const { IDX_STATE_STREAM_WANTS_TRAILERS, IDX_STATE_STREAM_RECEIVED_EARLY_DATA, IDX_STATE_STREAM_WRITE_DESIRED_SIZE, - IDX_STATE_STREAM_HIGH_WATER_MARK, + IDX_STATE_STREAM_BUDGET, IDX_STATE_STREAM_RESET_CODE, } = internalBinding('quic'); @@ -871,18 +871,18 @@ class QuicStreamState { } /** @type {number} */ - get highWaterMark() { + get budget() { const handle = this.#handle; if (handle === undefined) return undefined; return DataViewPrototypeGetUint32( - handle, this.#offset + IDX_STATE_STREAM_HIGH_WATER_MARK, kIsLittleEndian); + handle, this.#offset + IDX_STATE_STREAM_BUDGET, kIsLittleEndian); } - set highWaterMark(val) { + set budget(val) { const handle = this.#handle; if (handle === undefined) return; DataViewPrototypeSetUint32( - handle, this.#offset + IDX_STATE_STREAM_HIGH_WATER_MARK, val, kIsLittleEndian); + handle, this.#offset + IDX_STATE_STREAM_BUDGET, val, kIsLittleEndian); } toString() { @@ -908,7 +908,7 @@ class QuicStreamState { early, resetCode, writeDesiredSize, - highWaterMark, + budget, } = this; return { __proto__: null, @@ -928,7 +928,7 @@ class QuicStreamState { early, resetCode, writeDesiredSize, - highWaterMark, + budget, }; } @@ -965,7 +965,7 @@ class QuicStreamState { early, resetCode, writeDesiredSize, - highWaterMark, + budget, } = this; return `QuicStreamState ${inspect({ @@ -985,7 +985,7 @@ class QuicStreamState { early, resetCode, writeDesiredSize, - highWaterMark, + budget, }, opts)}`; } diff --git a/lib/internal/streams/iter/broadcast.js b/lib/internal/streams/iter/broadcast.js index 3592453036d303..4707d68298945a 100644 --- a/lib/internal/streams/iter/broadcast.js +++ b/lib/internal/streams/iter/broadcast.js @@ -10,7 +10,6 @@ const { ArrayIsArray, ArrayPrototypePush, ArrayPrototypeShift, - MathMax, PromisePrototypeThen, PromiseReject, PromiseResolve, @@ -54,9 +53,9 @@ const { } = require('internal/streams/iter/pull'); const { - kMultiConsumerDefaultHWM, + kMultiConsumerDefaultBudget, kResolvedPromise, - clampHWM, + clampBudget, convertChunks, getMinCursor, hasProtocol, @@ -75,7 +74,6 @@ const kCancelWriter = Symbol('kCancelWriter'); const kWrite = Symbol('kWrite'); const kEnd = Symbol('kEnd'); const kAbort = Symbol('kAbort'); -const kGetDesiredSize = Symbol('kGetDesiredSize'); const kCanWrite = Symbol('kCanWrite'); const kOnBufferDrained = Symbol('kOnBufferDrained'); @@ -95,6 +93,8 @@ class BroadcastImpl { #writer = null; #cachedMinCursor = 0; #cachedMinCursorConsumers = 0; + /** Cumulative byte size of buffered entries */ + #bufferedBytes = 0; constructor(options) { this.#options = options; @@ -109,18 +109,10 @@ class BroadcastImpl { return this.#options.backpressure; } - get highWaterMark() { - return this.#options.highWaterMark; - } - get consumerCount() { return this.#consumers.size; } - get bufferSize() { - return this.#buffer.length; - } - push(...args) { const { transforms, options } = parsePullArgs(args); const rawConsumer = this.#createRawConsumer(); @@ -282,14 +274,18 @@ class BroadcastImpl { [kWrite](chunk) { if (this.#ended || this.#cancelled) return false; - if (this.#buffer.length >= this.#options.highWaterMark) { + if (this.#bufferedBytes >= this.#options.budget) { switch (this.#options.backpressure) { case 'strict': - case 'block': + case 'unbounded': return false; case 'drop-oldest': - this.#buffer.shift(); - this.#bufferStart++; + while (this.#bufferedBytes >= this.#options.budget && + this.#buffer.length > 0) { + const evicted = this.#buffer.shift(); + this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferStart++; + } for (const consumer of this.#consumers) { if (consumer.cursor < this.#bufferStart) { this.#deleteConsumerFromMin(consumer); @@ -303,7 +299,9 @@ class BroadcastImpl { } } + const batchSize = this.#batchByteSize(chunk); this.#buffer.push(chunk); + this.#bufferedBytes += batchSize; this.#notifyConsumers(); return true; } @@ -358,16 +356,16 @@ class BroadcastImpl { this.#cachedMinCursorConsumers = 0; } - [kGetDesiredSize]() { - if (this.#ended || this.#cancelled) return null; - return MathMax(0, this.#options.highWaterMark - this.#buffer.length); - } - + /** + * Check if the next write is likely to be accepted. + * Returns null if ended/cancelled, true/false otherwise. + * @returns {boolean | null} + */ [kCanWrite]() { - if (this.#ended || this.#cancelled) return false; + if (this.#ended || this.#cancelled) return null; if ((this.#options.backpressure === 'strict' || - this.#options.backpressure === 'block') && - this.#buffer.length >= this.#options.highWaterMark) { + this.#options.backpressure === 'unbounded') && + this.#bufferedBytes >= this.#options.budget) { return false; } return true; @@ -388,16 +386,28 @@ class BroadcastImpl { } const trimCount = this.#cachedMinCursor - this.#bufferStart; if (trimCount > 0) { + for (let i = 0; i < trimCount; i++) { + const evicted = this.#buffer.get(i); + this.#bufferedBytes -= this.#batchByteSize(evicted); + } this.#buffer.trimFront(trimCount); this.#bufferStart = this.#cachedMinCursor; if (this[kOnBufferDrained] && - this.#buffer.length < this.#options.highWaterMark) { + this.#bufferedBytes < this.#options.budget) { this[kOnBufferDrained](); } } } + #batchByteSize(batch) { + let size = 0; + for (let i = 0; i < batch.length; i++) { + size += TypedArrayPrototypeGetByteLength(batch[i]); + } + return size; + } + #notifyConsumers() { const waiters = this.#waiters; if (waiters.length === 0) return; @@ -506,9 +516,9 @@ class BroadcastWriter { // The drainable protocol works with Stream.ondrain to provide a notification // when the writer can accept more data after being backpressured. [drainableProtocol]() { - const desired = this.desiredSize; - if (desired === null) return null; - if (desired > 0) return PromiseResolve(true); + const canWrite = this.canWrite; + if (canWrite === null) return null; + if (canWrite) return PromiseResolve(true); const { promise, resolve, reject } = PromiseWithResolvers(); ArrayPrototypePush(this.#pendingDrains, { __proto__: null, resolve, reject }); return promise; @@ -522,8 +532,8 @@ class BroadcastWriter { return this.#isClosed() || this.#aborted; } - get desiredSize() { - return this.#isClosedOrAborted() ? null : this.#broadcast[kGetDesiredSize](); + get canWrite() { + return this.#isClosedOrAborted() ? null : this.#broadcast[kCanWrite](); } #canUseWriteFastPath(options) { @@ -578,10 +588,9 @@ class BroadcastWriter { } const policy = this.#broadcast.backpressurePolicy; - const hwm = this.#broadcast.highWaterMark; if (policy === 'strict') { - if (this.#pendingWrites.length >= hwm) { + if (this.#pendingWrites.length >= 1) { throw new ERR_INVALID_STATE.TypeError( 'Backpressure violation: too many pending writes. ' + 'Await each write() call to respect backpressure.'); @@ -589,7 +598,7 @@ class BroadcastWriter { return this.#createPendingWrite(converted, signal); } - // 'block' policy + // 'unbounded' policy return this.#createPendingWrite(converted, signal); } @@ -750,17 +759,17 @@ function onBroadcastCancel(broadcastImpl, signal) { /** * Create a broadcast channel for push-model multi-consumer streaming. - * @param {{ highWaterMark?: number, backpressure?: string, signal?: AbortSignal }} [options] + * @param {{ budget?: number, backpressure?: string, signal?: AbortSignal }} [options] * @returns {{ writer: Writer, broadcast: Broadcast }} */ function broadcast(options = { __proto__: null }) { validateObject(options, 'options'); const { - highWaterMark = kMultiConsumerDefaultHWM, + budget = kMultiConsumerDefaultBudget, backpressure = 'strict', signal, } = options; - validateInteger(highWaterMark, 'options.highWaterMark'); + validateInteger(budget, 'options.budget', 16384); validateBackpressure(backpressure); if (signal !== undefined) { validateAbortSignal(signal, 'options.signal'); @@ -768,7 +777,7 @@ function broadcast(options = { __proto__: null }) { const opts = { __proto__: null, - highWaterMark: clampHWM(highWaterMark), + budget: clampBudget(budget), backpressure, signal, }; diff --git a/lib/internal/streams/iter/classic.js b/lib/internal/streams/iter/classic.js index b4401632946f0d..7b2777c56318e1 100644 --- a/lib/internal/streams/iter/classic.js +++ b/lib/internal/streams/iter/classic.js @@ -14,7 +14,6 @@ const { ArrayIsArray, ArrayPrototypePush, - MathMax, NumberMAX_SAFE_INTEGER, Promise, PromisePrototypeThen, @@ -417,7 +416,7 @@ const fromWritableCache = new SafeWeakMap(); * writableHighWaterMark, writableLength, writableObjectMode. * @param {object} writable - A classic Writable or duck-type. * @param {object} [options] - * @param {string} [options.backpressure] - 'strict', 'block', + * @param {string} [options.backpressure] - 'strict', 'unbounded', * 'drop-newest'. 'drop-oldest' is not supported. * @returns {object} A stream/iter Writer adapter. */ @@ -548,9 +547,9 @@ function fromWritable(writable, options = kNullPrototype) { const writer = { __proto__: null, - get desiredSize() { + get canWrite() { if (!isWritable()) return null; - return MathMax(0, hwm - (writable.writableLength ?? 0)); + return (writable.writableLength ?? 0) < hwm; }, writeSync(chunk) { @@ -604,10 +603,10 @@ function fromWritable(writable, options = kNullPrototype) { const ok = writable.write(bytes); if (ok) return PromiseResolve(); - // backpressure === 'block' (or strict with room that filled on + // backpressure === 'unbounded' (or strict with room that filled on // this write -- writable.write() accepted the data but returned // false indicating the buffer is now at/over hwm). - if (backpressure === 'block') { + if (backpressure === 'unbounded') { return waitForDrain(); } @@ -655,7 +654,7 @@ function fromWritable(writable, options = kNullPrototype) { if (ok) return PromiseResolve(); - if (backpressure === 'block') { + if (backpressure === 'unbounded') { return waitForDrain(); } diff --git a/lib/internal/streams/iter/duplex.js b/lib/internal/streams/iter/duplex.js index bd06f37303cfc6..b37b912792328c 100644 --- a/lib/internal/streams/iter/duplex.js +++ b/lib/internal/streams/iter/duplex.js @@ -20,13 +20,13 @@ const { /** * Create a pair of connected duplex channels for bidirectional communication. - * @param {{ highWaterMark?: number, backpressure?: string, signal?: AbortSignal, + * @param {{ budget?: number, backpressure?: string, signal?: AbortSignal, * a?: object, b?: object }} [options] * @returns {[DuplexChannel, DuplexChannel]} */ function duplex(options = { __proto__: null }) { validateObject(options, 'options'); - const { highWaterMark, backpressure, signal, a, b } = options; + const { budget, backpressure, signal, a, b } = options; if (a !== undefined) { validateObject(a, 'options.a'); } @@ -40,13 +40,13 @@ function duplex(options = { __proto__: null }) { // Channel A writes to B's readable (A->B direction). // Signal is NOT passed to push() -- we handle abort via close() below. const { writer: aWriter, readable: bReadable } = push({ - highWaterMark: a?.highWaterMark ?? highWaterMark, + budget: a?.budget ?? budget, backpressure: a?.backpressure ?? backpressure, }); // Channel B writes to A's readable (B->A direction) const { writer: bWriter, readable: aReadable } = push({ - highWaterMark: b?.highWaterMark ?? highWaterMark, + budget: b?.budget ?? budget, backpressure: b?.backpressure ?? backpressure, }); diff --git a/lib/internal/streams/iter/pull.js b/lib/internal/streams/iter/pull.js index 95ec2a084cdaf3..44a046e1b9013c 100644 --- a/lib/internal/streams/iter/pull.js +++ b/lib/internal/streams/iter/pull.js @@ -24,6 +24,7 @@ const { codes: { ERR_INVALID_ARG_TYPE, ERR_INVALID_ARG_VALUE, + ERR_OUT_OF_RANGE, }, } = require('internal/errors'); const { lazyDOMException } = require('internal/util'); @@ -862,14 +863,11 @@ function pipeToSync(source, ...args) { const hasEndSync = typeof writer.endSync === 'function'; try { - let canContinue = true; for (const batch of pipeline) { - if (!canContinue) { - break; - } if (hasWritevSync && batch.length > 1) { if (writer.writevSync(batch) === false) { - break; + throw new ERR_OUT_OF_RANGE( + 'write', 'within byte budget', 'budget exhausted'); } for (let i = 0; i < batch.length; i++) { totalBytes += TypedArrayPrototypeGetByteLength(batch[i]); @@ -878,8 +876,8 @@ function pipeToSync(source, ...args) { for (let i = 0; i < batch.length; i++) { const chunk = batch[i]; if (writer.writeSync(chunk) === false) { - canContinue = false; - break; + throw new ERR_OUT_OF_RANGE( + 'write', 'within byte budget', 'budget exhausted'); } totalBytes += TypedArrayPrototypeGetByteLength(chunk); } diff --git a/lib/internal/streams/iter/push.js b/lib/internal/streams/iter/push.js index 23adc5edc1cd3e..35e4e7f26b1710 100644 --- a/lib/internal/streams/iter/push.js +++ b/lib/internal/streams/iter/push.js @@ -8,7 +8,6 @@ const { ArrayIsArray, ArrayPrototypePush, - MathMax, PromiseReject, PromiseResolve, PromiseWithResolvers, @@ -36,9 +35,9 @@ const { } = require('internal/streams/iter/types'); const { - kPushDefaultHWM, + kPushDefaultBudget, kResolvedPromise, - clampHWM, + clampBudget, onSignalAbort, toUint8Array, convertChunks, @@ -81,23 +80,25 @@ class PushQueue { #pendingEnd = null; /** Configuration */ - #highWaterMark; + #budget; #backpressure; #signal; #abortHandler; + /** Cumulative byte size of buffered batches */ + #bufferedBytes = 0; constructor(options = { __proto__: null }) { const { - highWaterMark = kPushDefaultHWM, + budget = kPushDefaultBudget, backpressure = 'strict', signal, } = options; - validateInteger(highWaterMark, 'options.highWaterMark'); + validateInteger(budget, 'options.budget', 16384); validateBackpressure(backpressure); if (signal !== undefined) { validateAbortSignal(signal, 'options.signal'); } - this.#highWaterMark = clampHWM(highWaterMark); + this.#budget = clampBudget(budget); this.#backpressure = backpressure; this.#signal = signal; this.#abortHandler = undefined; @@ -117,15 +118,15 @@ class PushQueue { // =========================================================================== /** - * Get slots available before hitting highWaterMark. + * Check if the next write is likely to be accepted. * Returns null if writer is closed/errored or consumer has terminated. - * @returns {number | null} + * @returns {boolean | null} */ - get desiredSize() { + get canWrite() { if (this.#writerState !== 'open' || this.#consumerState !== 'active') { return null; } - return MathMax(0, this.#highWaterMark - this.#slots.length); + return this.#bufferedBytes < this.#budget; } /** @@ -136,8 +137,8 @@ class PushQueue { if (this.#writerState !== 'open') return false; if (this.#consumerState !== 'active') return false; if ((this.#backpressure === 'strict' || - this.#backpressure === 'block') && - this.#slots.length >= this.#highWaterMark) { + this.#backpressure === 'unbounded') && + this.#bufferedBytes >= this.#budget) { return false; } return true; @@ -152,30 +153,31 @@ class PushQueue { if (this.#writerState !== 'open') return false; if (this.#consumerState !== 'active') return false; - if (this.#slots.length >= this.#highWaterMark) { + const batchSize = this.#batchByteSize(chunks); + + if (this.#bufferedBytes >= this.#budget) { switch (this.#backpressure) { case 'strict': return false; - case 'block': + case 'unbounded': return false; case 'drop-oldest': - if (this.#slots.length > 0) { - this.#slots.shift(); + while (this.#bufferedBytes >= this.#budget && + this.#slots.length > 0) { + const evicted = this.#slots.shift(); + this.#bufferedBytes -= this.#batchByteSize(evicted); } break; case 'drop-newest': // Discard this write, but return true - for (let i = 0; i < chunks.length; i++) { - this.#bytesWritten += TypedArrayPrototypeGetByteLength(chunks[i]); - } + this.#bytesWritten += batchSize; return true; } } this.#slots.push(chunks); - for (let i = 0; i < chunks.length; i++) { - this.#bytesWritten += TypedArrayPrototypeGetByteLength(chunks[i]); - } + this.#bufferedBytes += batchSize; + this.#bytesWritten += batchSize; this.#resolvePendingReads(); return true; @@ -218,13 +220,13 @@ class PushQueue { // Buffer is full switch (this.#backpressure) { case 'strict': - if (this.#pendingWrites.length >= this.#highWaterMark) { + if (this.#pendingWrites.length >= 1) { throw new ERR_INVALID_STATE.RangeError( 'Backpressure violation: too many pending writes. ' + 'Await each write() call to respect backpressure.'); } return this.#createPendingWrite(chunks, signal); - case 'block': + case 'unbounded': return this.#createPendingWrite(chunks, signal); default: throw new ERR_INVALID_STATE( @@ -368,7 +370,7 @@ class PushQueue { } /** - * Wait for backpressure to clear (desiredSize > 0). + * Wait for backpressure to clear (canWrite becomes true). * @returns {Promise} */ waitForDrain() { @@ -449,6 +451,7 @@ class PushQueue { // =========================================================================== #drain() { + this.#bufferedBytes = 0; if (this.#slots.length === 1) { return this.#slots.shift(); } @@ -464,6 +467,14 @@ class PushQueue { return result; } + #batchByteSize(batch) { + let size = 0; + for (let i = 0; i < batch.length; i++) { + size += TypedArrayPrototypeGetByteLength(batch[i]); + } + return size; + } + #resolvePendingReads() { while (this.#pendingReads.length > 0) { if (this.#slots.length > 0) { @@ -492,16 +503,16 @@ class PushQueue { #resolvePendingWrites() { while (this.#pendingWrites.length > 0 && - this.#slots.length < this.#highWaterMark) { + this.#bufferedBytes < this.#budget) { const pending = this.#pendingWrites.shift(); + const batchSize = this.#batchByteSize(pending.chunks); this.#slots.push(pending.chunks); - for (let i = 0; i < pending.chunks.length; i++) { - this.#bytesWritten += TypedArrayPrototypeGetByteLength(pending.chunks[i]); - } + this.#bufferedBytes += batchSize; + this.#bytesWritten += batchSize; pending.resolve(); } - if (this.#slots.length < this.#highWaterMark) { + if (this.#bufferedBytes < this.#budget) { this.#resolvePendingDrains(true); } } @@ -554,14 +565,14 @@ class PushWriter { } [drainableProtocol]() { - const desired = this.desiredSize; - if (desired === null) return null; - if (desired > 0) return PromiseResolve(true); + const canWrite = this.canWrite; + if (canWrite === null) return null; + if (canWrite) return PromiseResolve(true); return this.#queue.waitForDrain(); } - get desiredSize() { - return this.#queue.desiredSize; + get canWrite() { + return this.#queue.canWrite; } write(chunk, options) { diff --git a/lib/internal/streams/iter/share.js b/lib/internal/streams/iter/share.js index e65a5eb648b620..02adb08d43c85a 100644 --- a/lib/internal/streams/iter/share.js +++ b/lib/internal/streams/iter/share.js @@ -14,6 +14,7 @@ const { SymbolAsyncIterator, SymbolDispose, SymbolIterator, + TypedArrayPrototypeGetByteLength, } = primordials; const { @@ -34,8 +35,8 @@ const { } = require('internal/streams/iter/pull'); const { - kMultiConsumerDefaultHWM, - clampHWM, + kMultiConsumerDefaultBudget, + clampBudget, getMinCursor, hasProtocol, onSignalAbort, @@ -79,6 +80,8 @@ class ShareImpl { #pullWaiters = []; #cachedMinCursor = 0; #cachedMinCursorConsumers = 0; + /** Cumulative byte size of buffered entries */ + #bufferedBytes = 0; constructor(source, options) { this.#source = source; @@ -89,10 +92,6 @@ class ShareImpl { return this.#consumers.size; } - get bufferSize() { - return this.#buffer.length; - } - pull(...args) { const { transforms, options } = parsePullArgs(args); const rawConsumer = this.#createRawConsumer(); @@ -261,7 +260,7 @@ class ShareImpl { // Internal methods async #waitForBufferSpace() { - while (this.#buffer.length >= this.#options.highWaterMark) { + while (this.#bufferedBytes >= this.#options.budget) { if (this.#cancelled || this.#sourceError || this.#sourceExhausted) { return !this.#cancelled; } @@ -269,17 +268,21 @@ class ShareImpl { switch (this.#options.backpressure) { case 'strict': throw new ERR_OUT_OF_RANGE( - 'buffer size', `<= ${this.#options.highWaterMark}`, - this.#buffer.length); - case 'block': { + 'buffered bytes', `< ${this.#options.budget}`, + this.#bufferedBytes); + case 'unbounded': { const { promise, resolve } = PromiseWithResolvers(); ArrayPrototypePush(this.#pullWaiters, resolve); await promise; break; } case 'drop-oldest': - this.#buffer.shift(); - this.#bufferStart++; + while (this.#bufferedBytes >= this.#options.budget && + this.#buffer.length > 0) { + const evicted = this.#buffer.shift(); + this.#bufferedBytes -= this.#batchByteSize(evicted); + this.#bufferStart++; + } for (const consumer of this.#consumers) { if (consumer.cursor < this.#bufferStart) { this.#deleteConsumerFromMin(consumer); @@ -339,6 +342,7 @@ class ShareImpl { this.#sourceExhausted = true; } else { this.#buffer.push(result.value); + this.#bufferedBytes += this.#batchByteSize(result.value); } } catch (error) { this.#sourceError = wrapError(error); @@ -359,6 +363,10 @@ class ShareImpl { } const trimCount = this.#cachedMinCursor - this.#bufferStart; if (trimCount > 0) { + for (let i = 0; i < trimCount; i++) { + const evicted = this.#buffer.get(i); + this.#bufferedBytes -= this.#batchByteSize(evicted); + } this.#buffer.trimFront(trimCount); this.#bufferStart = this.#cachedMinCursor; for (let i = 0; i < this.#pullWaiters.length; i++) { @@ -368,6 +376,14 @@ class ShareImpl { } } + #batchByteSize(batch) { + let size = 0; + for (let i = 0; i < batch.length; i++) { + size += TypedArrayPrototypeGetByteLength(batch[i]); + } + return size; + } + #recomputeMinCursor() { const { minCursor, minCursorConsumers } = getMinCursor( this.#consumers, this.#bufferStart + this.#buffer.length); @@ -407,6 +423,8 @@ class SyncShareImpl { #cancelled = false; #cachedMinCursor = 0; #cachedMinCursorConsumers = 0; + /** Cumulative byte size of buffered entries */ + #bufferedBytes = 0; constructor(source, options) { this.#source = source; @@ -417,10 +435,6 @@ class SyncShareImpl { return this.#consumers.size; } - get bufferSize() { - return this.#buffer.length; - } - pull(...transforms) { const rawConsumer = this.#createRawConsumer(); @@ -487,20 +501,24 @@ class SyncShareImpl { } // Check buffer limit - if (self.#buffer.length >= self.#options.highWaterMark) { + if (self.#bufferedBytes >= self.#options.budget) { switch (self.#options.backpressure) { case 'strict': throw new ERR_OUT_OF_RANGE( - 'buffer size', `<= ${self.#options.highWaterMark}`, - self.#buffer.length); - case 'block': + 'buffered bytes', `< ${self.#options.budget}`, + self.#bufferedBytes); + case 'unbounded': throw new ERR_OUT_OF_RANGE( - 'buffer size', `<= ${self.#options.highWaterMark} ` + - '(blocking not available in sync context)', - self.#buffer.length); + 'buffered bytes', `< ${self.#options.budget} ` + + '(unbounded not available in sync context)', + self.#bufferedBytes); case 'drop-oldest': - self.#buffer.shift(); - self.#bufferStart++; + while (self.#bufferedBytes >= self.#options.budget && + self.#buffer.length > 0) { + const evicted = self.#buffer.shift(); + self.#bufferedBytes -= self.#batchByteSize(evicted); + self.#bufferStart++; + } for (const consumer of self.#consumers) { if (consumer.cursor < self.#bufferStart) { self.#deleteConsumerFromMin(consumer); @@ -599,6 +617,7 @@ class SyncShareImpl { this.#sourceExhausted = true; } else { this.#buffer.push(result.value); + this.#bufferedBytes += this.#batchByteSize(result.value); } } catch (error) { this.#sourceError = wrapError(error); @@ -612,11 +631,23 @@ class SyncShareImpl { } const trimCount = this.#cachedMinCursor - this.#bufferStart; if (trimCount > 0) { + for (let i = 0; i < trimCount; i++) { + const evicted = this.#buffer.get(i); + this.#bufferedBytes -= this.#batchByteSize(evicted); + } this.#buffer.trimFront(trimCount); this.#bufferStart = this.#cachedMinCursor; } } + #batchByteSize(batch) { + let size = 0; + for (let i = 0; i < batch.length; i++) { + size += TypedArrayPrototypeGetByteLength(batch[i]); + } + return size; + } + #recomputeMinCursor() { const { minCursor, minCursorConsumers } = getMinCursor( this.#consumers, this.#bufferStart + this.#buffer.length); @@ -653,11 +684,11 @@ function share(source, options = { __proto__: null }) { const normalized = from(source); validateObject(options, 'options'); const { - highWaterMark = kMultiConsumerDefaultHWM, + budget = kMultiConsumerDefaultBudget, backpressure = 'strict', signal, } = options; - validateInteger(highWaterMark, 'options.highWaterMark'); + validateInteger(budget, 'options.budget', 16384); validateBackpressure(backpressure); if (signal !== undefined) { validateAbortSignal(signal, 'options.signal'); @@ -665,7 +696,7 @@ function share(source, options = { __proto__: null }) { const opts = { __proto__: null, - highWaterMark: clampHWM(highWaterMark), + budget: clampBudget(budget), backpressure, signal, }; @@ -684,15 +715,15 @@ function shareSync(source, options = { __proto__: null }) { const normalized = fromSync(source); validateObject(options, 'options'); const { - highWaterMark = kMultiConsumerDefaultHWM, + budget = kMultiConsumerDefaultBudget, backpressure = 'strict', } = options; - validateInteger(highWaterMark, 'options.highWaterMark'); + validateInteger(budget, 'options.budget', 16384); validateBackpressure(backpressure); const opts = { __proto__: null, - highWaterMark: clampHWM(highWaterMark), + budget: clampBudget(budget), backpressure, }; diff --git a/lib/internal/streams/iter/utils.js b/lib/internal/streams/iter/utils.js index ec02eca43f3385..d7dc9432ff7f44 100644 --- a/lib/internal/streams/iter/utils.js +++ b/lib/internal/streams/iter/utils.js @@ -47,19 +47,19 @@ const encoder = new TextEncoder(); // are somewhat arbitrary but have been tested across various workloads and // appear to yield the best overall throughput/latency balance. -/** Default high water mark for push streams (single-consumer). */ -const kPushDefaultHWM = 4; +/** Minimum and default byte budget for push streams (single-consumer). */ +const kPushDefaultBudget = 16384; -/** Default high water mark for broadcast and share streams (multi-consumer). */ -const kMultiConsumerDefaultHWM = 16; +/** Default byte budget for broadcast and share streams (multi-consumer). */ +const kMultiConsumerDefaultBudget = 65536; /** - * Clamp a high water mark to [1, MAX_SAFE_INTEGER]. + * Clamp a byte budget to [16384, MAX_SAFE_INTEGER]. * @param {number} value * @returns {number} */ -function clampHWM(value) { - return MathMax(1, MathMin(NumberMAX_SAFE_INTEGER, value)); +function clampBudget(value) { + return MathMax(16384, MathMin(NumberMAX_SAFE_INTEGER, value)); } /** @@ -364,18 +364,18 @@ function parsePullArgs(args) { function validateBackpressure(value) { validateOneOf(value, 'options.backpressure', [ 'strict', - 'block', + 'unbounded', 'drop-oldest', 'drop-newest', ]); } module.exports = { - kMultiConsumerDefaultHWM, - kPushDefaultHWM, + kMultiConsumerDefaultBudget, + kPushDefaultBudget, kResolvedPromise, allUint8Array, - clampHWM, + clampBudget, concatBytes, convertChunks, getMinCursor, diff --git a/src/quic/streams.cc b/src/quic/streams.cc index 2670fc87d6bc21..aaa1fe57ffc9bb 100644 --- a/src/quic/streams.cc +++ b/src/quic/streams.cc @@ -66,7 +66,7 @@ namespace quic { /* True when 0-RTT early data was received */ \ V(RECEIVED_EARLY_DATA, received_early_data, uint8_t) \ V(WRITE_DESIRED_SIZE, write_desired_size, uint32_t) \ - V(HIGH_WATER_MARK, high_water_mark, uint32_t) + V(BUDGET, budget, uint32_t) #define STREAM_STATS(V) \ /* Marks the timestamp when the stream object was created. */ \ @@ -1857,13 +1857,13 @@ void Stream::UpdateWriteDesiredSize() { if (!outbound_ || !outbound_->is_streaming()) return; uint64_t available; - uint64_t hwm = state()->high_water_mark; + uint64_t bgt = state()->budget; if (is_pending()) { // Pending streams don't have a stream ID yet, so ngtcp2 can't - // report their flow control window. Use the high water mark as - // the available capacity so writes can proceed while pending. - available = hwm > 0 ? hwm : std::numeric_limits::max(); + // report their flow control window. Use the budget as the + // available capacity so writes can proceed while pending. + available = bgt > 0 ? bgt : std::numeric_limits::max(); } else { // Calculate available capacity based on QUIC flow control. // The effective limit is the minimum of stream-level and @@ -1873,9 +1873,9 @@ void Stream::UpdateWriteDesiredSize() { uint64_t conn_left = ngtcp2_conn_get_max_data_left(conn); available = std::min(stream_left, conn_left); - // Apply the high water mark as an additional ceiling. - if (hwm > 0) { - available = std::min(available, hwm); + // Apply the budget as an additional ceiling. + if (bgt > 0) { + available = std::min(available, bgt); } } diff --git a/test/parallel/test-quic-flow-control-params.mjs b/test/parallel/test-quic-flow-control-params.mjs index 72fe43263a8cd3..a3a4355e383dbe 100644 --- a/test/parallel/test-quic-flow-control-params.mjs +++ b/test/parallel/test-quic-flow-control-params.mjs @@ -50,7 +50,7 @@ const encoder = new TextEncoder(); await clientSession.opened; const stream = await clientSession.createBidirectionalStream({ - highWaterMark: 512, + budget: 512, }); const w = stream.writer; diff --git a/test/parallel/test-quic-flow-control-uni.mjs b/test/parallel/test-quic-flow-control-uni.mjs index 11685700d6da1e..c0b04cb676e6ed 100644 --- a/test/parallel/test-quic-flow-control-uni.mjs +++ b/test/parallel/test-quic-flow-control-uni.mjs @@ -39,7 +39,7 @@ const clientSession = await connect(serverEndpoint.address); await clientSession.opened; const stream = await clientSession.createUnidirectionalStream({ - highWaterMark: 512, + budget: 512, }); const w = stream.writer; diff --git a/test/parallel/test-quic-stream-bidi-writer.mjs b/test/parallel/test-quic-stream-bidi-writer.mjs index 972c376257ec96..9ca636f0875ce7 100644 --- a/test/parallel/test-quic-stream-bidi-writer.mjs +++ b/test/parallel/test-quic-stream-bidi-writer.mjs @@ -41,7 +41,7 @@ const stream = await clientSession.createBidirectionalStream(); const w = stream.writer; // Writer should be open. -strictEqual(typeof w.desiredSize, 'number'); +strictEqual(typeof w.canWrite, 'boolean'); // Write multiple chunks synchronously. strictEqual(w.writeSync(encoder.encode('chunk1')), true); @@ -55,8 +55,8 @@ strictEqual(totalWritten, 18); // 6 * 3 // After end, write should return false. strictEqual(w.writeSync(encoder.encode('nope')), false); -// desiredSize should be null after close. -strictEqual(w.desiredSize, null); +// canWrite should be null after close. +strictEqual(w.canWrite, null); await Promise.all([stream.closed, done.promise]); await clientSession.close(); diff --git a/test/parallel/test-quic-stream-destroy-emits-stop-sending.mjs b/test/parallel/test-quic-stream-destroy-emits-stop-sending.mjs index 47a6d8f1857e4f..24ced6918ff3b4 100644 --- a/test/parallel/test-quic-stream-destroy-emits-stop-sending.mjs +++ b/test/parallel/test-quic-stream-destroy-emits-stop-sending.mjs @@ -6,7 +6,7 @@ // from the peer's perspective and they would keep transmitting until // the session-level idle timer fired. // -// Verified via the server-side writer's `desiredSize` getter, which +// Verified via the server-side writer's `canWrite` getter, which // returns `null` once `state.writeEnded` is set. STOP_SENDING from // the client triggers `Stream::ReceiveStopSending -> EndWritable` on // the server, which sets `state.writeEnded = 1`. @@ -16,7 +16,7 @@ // processes RESET_STREAM first (firing `onreset`) and then // STOP_SENDING. The observation must therefore be deferred until // after the `onreset` callback returns so ngtcp2 can finish -// processing the packet. Capturing `writer.desiredSize` synchronously +// processing the packet. Capturing `writer.canWrite` synchronously // inside `onreset` would always see the pre-STOP_SENDING value. // Throwing inside `onreset` would also crash the process before // STOP_SENDING could be processed. @@ -38,12 +38,12 @@ const serverEndpoint = await listen(mustCall((serverSession) => { serverSession.onstream = mustCall(async (stream) => { const writer = stream.writer; // Sanity: the writer is active before the peer tears the stream - // down, so desiredSize is a number reflecting the initial + // down, so canWrite is a boolean reflecting the initial // flow-control window. - if (typeof writer.desiredSize !== 'number') { + if (typeof writer.canWrite !== 'boolean') { serverObservation.reject(new Error( - `expected initial writer.desiredSize to be a number, ` + - `got ${writer.desiredSize}`)); + `expected initial writer.canWrite to be a boolean, ` + + `got ${writer.canWrite}`)); return; } stream.onreset = mustCall(() => { @@ -53,9 +53,9 @@ const serverEndpoint = await listen(mustCall((serverSession) => { // next event loop tick the rest of the packet has been // processed and `Stream::ReceiveStopSending -> EndWritable` has // flipped `state.writeEnded` to 1, which makes the writer's - // desiredSize getter return null. + // canWrite getter return null. setImmediate(() => { - serverObservation.resolve(writer.desiredSize); + serverObservation.resolve(writer.canWrite); }); }); @@ -78,8 +78,8 @@ const clientClosedAssertion = assert.rejects(stream.closed, err); stream.destroy(err); -const observedDesiredSize = await serverObservation.promise; -strictEqual(observedDesiredSize, null); +const observedCanWrite = await serverObservation.promise; +strictEqual(observedCanWrite, null); await clientClosedAssertion; diff --git a/test/parallel/test-quic-stream-uni-server-initiated.mjs b/test/parallel/test-quic-stream-uni-server-initiated.mjs index 7853938626617d..3233dda4ee7c38 100644 --- a/test/parallel/test-quic-stream-uni-server-initiated.mjs +++ b/test/parallel/test-quic-stream-uni-server-initiated.mjs @@ -47,7 +47,7 @@ clientSession.onstream = mustCall(async (stream) => { // The receiving side of a uni stream should not be writable. // The writer should be pre-closed. const w = stream.writer; - strictEqual(w.desiredSize, null); + strictEqual(w.canWrite, null); await stream.closed; clientSession.close(); diff --git a/test/parallel/test-quic-stream-writer-api.mjs b/test/parallel/test-quic-stream-writer-api.mjs index 6003473584c51d..7beeac0bae1a38 100644 --- a/test/parallel/test-quic-stream-writer-api.mjs +++ b/test/parallel/test-quic-stream-writer-api.mjs @@ -93,12 +93,12 @@ await clientSession.opened; { const stream = await clientSession.createBidirectionalStream(); const w = stream.writer; - // desiredSize should be a number (may be 0 initially before flow - // control window opens, or > 0 if the window is already open). - strictEqual(typeof w.desiredSize, 'number'); - ok(w.desiredSize >= 0, `desiredSize should be >= 0, got ${w.desiredSize}`); - // drainableProtocol should return null when desiredSize > 0 (has capacity), - // or a promise when desiredSize <= 0 (backpressured). Either way, it + // canWrite should be a boolean (false initially before flow + // control window opens, or true if the window is already open). + strictEqual(typeof w.canWrite, 'boolean'); + ok(w.canWrite !== null, `canWrite should not be null, got ${w.canWrite}`); + // drainableProtocol should return null when canWrite is true (has capacity), + // or a promise when canWrite is false (backpressured). Either way, it // should not throw. const { drainableProtocol: dp } = await import('stream/iter'); const drain = w[dp](); @@ -115,8 +115,8 @@ await clientSession.opened; const w = stream.writer; const testError = new Error('writer fail test'); w.fail(testError); - // After fail, desiredSize is null. - strictEqual(w.desiredSize, null); + // After fail, canWrite is null. + strictEqual(w.canWrite, null); // drainableProtocol returns null when errored. const { drainableProtocol: dp } = await import('stream/iter'); strictEqual(w[dp](), null); diff --git a/test/parallel/test-quic-stream-writer-dispose.mjs b/test/parallel/test-quic-stream-writer-dispose.mjs index 3899f2f2973636..38fd2889beeb05 100644 --- a/test/parallel/test-quic-stream-writer-dispose.mjs +++ b/test/parallel/test-quic-stream-writer-dispose.mjs @@ -33,14 +33,14 @@ await clientSession.opened; const stream = await clientSession.createBidirectionalStream(); const w = stream.writer; -// Writer is active — desiredSize should be a number (not null). -strictEqual(typeof w.desiredSize, 'number'); +// Writer is active — canWrite should be a boolean (not null). +strictEqual(typeof w.canWrite, 'boolean'); // Symbol.dispose calls fail() if not already closed/errored. w[Symbol.dispose](); // After dispose, writer should be errored. -strictEqual(w.desiredSize, null); +strictEqual(w.canWrite, null); strictEqual(w.writeSync(encoder.encode('x')), false); // stream.closed resolves because fail() with default code 0 diff --git a/test/parallel/test-quic-writer-backpressure.mjs b/test/parallel/test-quic-writer-backpressure.mjs index 4d74f029929e53..a32f3d6c75d162 100644 --- a/test/parallel/test-quic-writer-backpressure.mjs +++ b/test/parallel/test-quic-writer-backpressure.mjs @@ -1,8 +1,8 @@ // Flags: --experimental-quic --experimental-stream-iter --no-warnings // Test: writer backpressure. -// writeSync returns false when highWaterMark is exceeded. -// drainableProtocol returns promise when desiredSize <= 0. +// writeSync returns false when budget is exceeded. +// drainableProtocol returns promise when canWrite is false. // drainableProtocol promise resolves when drain fires. // Try-fallback pattern: writeSync false, await drain, retry. @@ -18,7 +18,7 @@ if (!hasQuic) { const { listen, connect } = await import('../common/quic.mjs'); const { bytes, drainableProtocol: dp } = await import('stream/iter'); -// Total data: 8 x 1KB = 8KB. highWaterMark: 2KB. +// Total data: 8 x 1KB = 8KB. budget: 2KB. const numChunks = 8; const chunkSize = 1024; const totalSize = numChunks * chunkSize; @@ -40,13 +40,13 @@ const clientSession = await connect(serverEndpoint.address); await clientSession.opened; const stream = await clientSession.createBidirectionalStream({ - highWaterMark: 2048, + budget: 2048, }); const w = stream.writer; -// Initial desiredSize should be the highWaterMark. -strictEqual(w.desiredSize, 2048); -strictEqual(stream.highWaterMark, 2048); +// Initial canWrite should be true. +strictEqual(w.canWrite, true); +strictEqual(stream.budget, 2048); let backpressureCount = 0; @@ -64,14 +64,14 @@ for (let i = 0; i < numChunks; i++) { // The promise resolves when drain fires. await drain; - // After drain, desiredSize should be > 0. - ok(w.desiredSize > 0, `desiredSize after drain should be > 0, got ${w.desiredSize}`); + // After drain, canWrite should be true. + ok(w.canWrite === true, `canWrite after drain should be true, got ${w.canWrite}`); } } w.endSync(); -// Backpressure should have been hit with a 2KB highWaterMark +// Backpressure should have been hit with a 2KB budget // and 1KB chunks (every 2 chunks fills the buffer). ok(backpressureCount > 0, 'backpressure should have been hit'); diff --git a/test/parallel/test-quic-writer-write-rejects.mjs b/test/parallel/test-quic-writer-write-rejects.mjs index 864246b9e80e70..1b4d2c3a03373e 100644 --- a/test/parallel/test-quic-writer-write-rejects.mjs +++ b/test/parallel/test-quic-writer-write-rejects.mjs @@ -2,7 +2,7 @@ // Test: write() rejects when flow-controlled. // The async write() method rejects with ERR_INVALID_STATE when the -// chunk exceeds desiredSize. +// chunk exceeds capacity (canWrite is false). import { hasQuic, skip, mustCall } from '../common/index.mjs'; import assert from 'node:assert'; @@ -31,17 +31,17 @@ const serverEndpoint = await listen(mustCall((serverSession) => { const clientSession = await connect(serverEndpoint.address); await clientSession.opened; -// Use a small highWaterMark to trigger backpressure easily. +// Use a small budget to trigger backpressure easily. const stream = await clientSession.createBidirectionalStream({ - highWaterMark: 1024, + budget: 1024, }); const w = stream.writer; // Fill the buffer. strictEqual(w.writeSync(new Uint8Array(1024)), true); -// desiredSize should now be 0 or very small. -strictEqual(w.desiredSize, 0); +// canWrite should now be false. +strictEqual(w.canWrite, false); // Async write() should reject when buffer is full. await rejects( @@ -53,7 +53,7 @@ await rejects( const drain = w[dp](); ok(drain instanceof Promise); await drain; -ok(w.desiredSize > 0); +ok(w.canWrite === true); // Now write succeeds. await w.write(new Uint8Array(100)); diff --git a/test/parallel/test-stream-iter-broadcast-backpressure.js b/test/parallel/test-stream-iter-broadcast-backpressure.js index d1e466c3ac44cf..ab3a41827ad069 100644 --- a/test/parallel/test-stream-iter-broadcast-backpressure.js +++ b/test/parallel/test-stream-iter-broadcast-backpressure.js @@ -10,36 +10,41 @@ const { broadcast, text } = require('stream/iter'); // ============================================================================= async function testDropOldest() { + const chunk1 = new Uint8Array(16384).fill(49); // '1' + const chunk2 = new Uint8Array(16384).fill(50); // '2' + const chunk3 = new Uint8Array(16384).fill(51); // '3' const { writer, broadcast: bc } = broadcast({ - highWaterMark: 2, + budget: 32768, backpressure: 'drop-oldest', }); const consumer = bc.push(); - writer.writeSync('first'); - writer.writeSync('second'); - // This should drop 'first' - writer.writeSync('third'); + writer.writeSync(chunk1); // 16384 < 32768 + writer.writeSync(chunk2); // 32768 >= 32768 + // Buffer full: this drops chunk1, adds chunk3 + writer.writeSync(chunk3); writer.endSync(); const data = await text(consumer); - assert.strictEqual(data, 'secondthird'); + assert.strictEqual(data, '2'.repeat(16384) + '3'.repeat(16384)); } async function testDropNewest() { + const kept = new Uint8Array(16384).fill(75); // 'K' + const dropped = new Uint8Array(16384).fill(68); // 'D' const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, + budget: 16384, backpressure: 'drop-newest', }); const consumer = bc.push(); - writer.writeSync('kept'); - // This should be silently dropped - writer.writeSync('dropped'); + writer.writeSync(kept); + // Buffer full: new write is silently discarded + writer.writeSync(dropped); writer.endSync(); const data = await text(consumer); - assert.strictEqual(data, 'kept'); + assert.strictEqual(data, 'K'.repeat(16384)); } // ============================================================================= @@ -47,16 +52,17 @@ async function testDropNewest() { // ============================================================================= async function testBlockBackpressure() { + const kChunk = new Uint8Array(16384); const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consumer = bc.push(); - writer.writeSync('a'); + writer.writeSync(kChunk); // Next write should block let writeResolved = false; - const writePromise = writer.write('b').then(() => { writeResolved = true; }); + const writePromise = writer.write(kChunk).then(() => { writeResolved = true; }); await new Promise(setImmediate); assert.strictEqual(writeResolved, false); @@ -76,30 +82,32 @@ async function testBlockBackpressure() { // Verify block backpressure data flows correctly end-to-end async function testBlockBackpressureContent() { + const chunk1 = new Uint8Array(16384).fill(65); // 'A' + const chunk2 = new Uint8Array(16384).fill(66); // 'B' const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consumer = bc.push(); - writer.writeSync('a'); - const writePromise = writer.write('b'); + writer.writeSync(chunk1); + const writePromise = writer.write(chunk2); await new Promise(setImmediate); // Read all and verify content const iter = consumer[Symbol.asyncIterator](); const first = await iter.next(); assert.strictEqual(first.done, false); - const firstStr = new TextDecoder().decode(first.value[0]); - assert.strictEqual(firstStr, 'a'); + assert.strictEqual(first.value[0].byteLength, 16384); + assert.strictEqual(first.value[0][0], 65); // 'A' await writePromise; writer.endSync(); const second = await iter.next(); assert.strictEqual(second.done, false); - const secondStr = new TextDecoder().decode(second.value[0]); - assert.strictEqual(secondStr, 'b'); + assert.strictEqual(second.value[0].byteLength, 16384); + assert.strictEqual(second.value[0][0], 66); // 'B' const done = await iter.next(); assert.strictEqual(done.done, true); @@ -107,7 +115,7 @@ async function testBlockBackpressureContent() { // Writev async path async function testWritevAsync() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 10 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); const consumer = bc.push(); await writer.writev(['hello', ' ', 'world']); @@ -119,7 +127,7 @@ async function testWritevAsync() { // endSync returns the total byte count async function testEndSyncReturnValue() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 10 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); bc.push(); // Need a consumer to write to writer.writeSync('hello'); // 5 bytes diff --git a/test/parallel/test-stream-iter-broadcast-basic.js b/test/parallel/test-stream-iter-broadcast-basic.js index bdecd7d5beb9b1..6f1efe4625da19 100644 --- a/test/parallel/test-stream-iter-broadcast-basic.js +++ b/test/parallel/test-stream-iter-broadcast-basic.js @@ -32,7 +32,7 @@ async function testBasicBroadcast() { } async function testMultipleWrites() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 10 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); const consumer = bc.push(); @@ -74,22 +74,22 @@ async function testConsumerCount() { // ============================================================================= async function testWriteSync() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 2 }); + const kChunk = new Uint8Array(16384); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); const consumer = bc.push(); - assert.strictEqual(writer.writeSync('a'), true); - assert.strictEqual(writer.writeSync('b'), true); - // Buffer full (highWaterMark=2, strict policy) - assert.strictEqual(writer.writeSync('c'), false); + assert.strictEqual(writer.writeSync(kChunk), true); + // Buffer full (16384 >= budget), strict policy rejects + assert.strictEqual(writer.writeSync(kChunk), false); writer.endSync(); const data = await text(consumer); - assert.strictEqual(data, 'ab'); + assert.strictEqual(data.length, 16384); } async function testWritevSync() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 10 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); const consumer = bc.push(); assert.strictEqual(writer.writevSync(['hello', ' ', 'world']), true); @@ -257,7 +257,7 @@ async function testCancelWithFalsyReason() { // Late-joining consumer should read from oldest buffered entry async function testLateJoinerSeesBufferedData() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 16 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); // Write data before any consumer joins writer.writeSync('before-join'); diff --git a/test/parallel/test-stream-iter-broadcast-coverage.js b/test/parallel/test-stream-iter-broadcast-coverage.js index 7aa71062ab197b..b8dec044c30720 100644 --- a/test/parallel/test-stream-iter-broadcast-coverage.js +++ b/test/parallel/test-stream-iter-broadcast-coverage.js @@ -15,17 +15,17 @@ const { // Signal abort on pending write (covers wireBroadcastWriteSignal + removeAt) async function testBroadcastWriteAbort() { const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consumer = bc.push(); // Fill the buffer to capacity - writer.writeSync(new Uint8Array([1])); + writer.writeSync(new Uint8Array(16384).fill(1)); // Next write will block — pass a signal const ac = new AbortController(); - const writePromise = writer.write(new Uint8Array([2]), + const writePromise = writer.write(new Uint8Array(16384).fill(2), { signal: ac.signal }); // Abort the signal @@ -74,7 +74,7 @@ async function testBroadcastFromSyncIterableStrings() { // Ringbuffer grow — push > 16 items without consumer draining async function testRingbufferGrow() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 32 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); const consumer = bc.push(); // Push 20 items (exceeds default ringbuffer capacity of 16) @@ -99,25 +99,20 @@ async function testRingbufferGrow() { // Multiple consumers at the minimum cursor should trim only after the last // one advances or detaches. async function testFanOutMinCursorTrimming() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 4 }); + const { writer, broadcast: bc } = broadcast({ budget: 16384 }); const iter1 = bc.push()[Symbol.asyncIterator](); const iter2 = bc.push()[Symbol.asyncIterator](); writer.writeSync(new Uint8Array([1])); writer.writeSync(new Uint8Array([2])); - assert.strictEqual(bc.bufferSize, 2); assert.strictEqual((await iter1.next()).done, false); - assert.strictEqual(bc.bufferSize, 2); assert.strictEqual((await iter2.next()).done, false); - assert.strictEqual(bc.bufferSize, 1); await iter1.return(); - assert.strictEqual(bc.bufferSize, 1); assert.strictEqual((await iter2.next()).done, false); - assert.strictEqual(bc.bufferSize, 0); writer.endSync(); assert.strictEqual((await iter2.next()).done, true); diff --git a/test/parallel/test-stream-iter-duplex.js b/test/parallel/test-stream-iter-duplex.js index 83c85d7be00816..502c1f88ebb6ac 100644 --- a/test/parallel/test-stream-iter-duplex.js +++ b/test/parallel/test-stream-iter-duplex.js @@ -46,7 +46,7 @@ async function testBidirectional() { } async function testMultipleWrites() { - const [channelA, channelB] = duplex({ highWaterMark: 10 }); + const [channelA, channelB] = duplex({ budget: 16384 }); await channelA.writer.write('one'); await channelA.writer.write('two'); @@ -75,7 +75,7 @@ async function testChannelClose() { async function testWithOptions() { const [channelA, channelB] = duplex({ - highWaterMark: 2, + budget: 16384, backpressure: 'strict', }); @@ -88,8 +88,8 @@ async function testWithOptions() { async function testPerChannelOptions() { const [channelA, channelB] = duplex({ - a: { highWaterMark: 1 }, - b: { highWaterMark: 4 }, + a: { budget: 16384 }, + b: { budget: 16384 }, }); // Channel A -> B direction uses A's options diff --git a/test/parallel/test-stream-iter-pipeto-writev.js b/test/parallel/test-stream-iter-pipeto-writev.js index c466b007dc4a75..cc102cc45a706d 100644 --- a/test/parallel/test-stream-iter-pipeto-writev.js +++ b/test/parallel/test-stream-iter-pipeto-writev.js @@ -129,8 +129,8 @@ async function testWriteSyncAlwaysFails() { // backpressure. pipeTo must wait for drain, not retry the same write. async function assertPushWriterBlockPipeTo(source, expected, expectedTotal) { const { writer, readable } = push({ - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const pipe = pipeTo(source, writer); @@ -155,26 +155,18 @@ async function testPushWriterBlockSyncFalseAccepted() { } async function testPipeToSyncPushWriterStrictFalseRejected() { - const decoder = new TextDecoder(); - const { writer, readable } = push({ highWaterMark: 1 }); - - const total = pipeToSync(['a', 'b'], writer, { preventClose: true }); - assert.strictEqual(total, 1); - - const iter = readable[Symbol.asyncIterator](); - const first = await iter.next(); - assert.strictEqual(first.done, false); - assert.strictEqual(decoder.decode(first.value[0]), 'a'); - - const second = await Promise.race([ - iter.next().then((result) => { - return result.done ? '' : decoder.decode(result.value[0]); - }), - setImmediatePromise().then(() => ''), - ]); - assert.strictEqual(second, ''); - - await iter.return?.(); + const kChunk = new Uint8Array(16384); + const { writer } = push({ budget: 16384 }); + + // Pre-fill the buffer so it's at capacity + writer.writeSync(kChunk); + + // pipeToSync should throw when writeSync returns false (budget exhausted) + assert.throws( + () => pipeToSync([kChunk], writer, + { preventClose: true, preventFail: true }), + { code: 'ERR_OUT_OF_RANGE' }, + ); } async function testPipeToSyncWritevFalseNotCounted() { @@ -187,8 +179,11 @@ async function testPipeToSyncWritevFalseNotCounted() { yield [new Uint8Array([1]), new Uint8Array([2])]; } - const total = pipeToSync(source(), writer); - assert.strictEqual(total, 0); + // pipeToSync throws when writevSync returns false (budget exhausted) + assert.throws( + () => pipeToSync(source(), writer), + { code: 'ERR_OUT_OF_RANGE' }, + ); } // pipeToSync with writevSync diff --git a/test/parallel/test-stream-iter-pull-async.js b/test/parallel/test-stream-iter-pull-async.js index b1d614294328d8..6fa3192e5855d0 100644 --- a/test/parallel/test-stream-iter-pull-async.js +++ b/test/parallel/test-stream-iter-pull-async.js @@ -358,7 +358,7 @@ async function testTransformReturnsArrayBuffer() { // pipeTo() accepts a string source directly (normalized via from()) async function testPipeToStringSource() { const { pipeTo, push: pushFn, text: textFn } = require('stream/iter'); - const { writer, readable } = pushFn({ highWaterMark: 10 }); + const { writer, readable } = pushFn({ budget: 16384 }); const consume = (async () => textFn(readable))(); await pipeTo('hello-pipe', writer); const data = await consume; diff --git a/test/parallel/test-stream-iter-push-backpressure.js b/test/parallel/test-stream-iter-push-backpressure.js index 7196b7cca5da71..31be8eb46baa7d 100644 --- a/test/parallel/test-stream-iter-push-backpressure.js +++ b/test/parallel/test-stream-iter-push-backpressure.js @@ -6,74 +6,71 @@ const assert = require('assert'); const { push, text } = require('stream/iter'); async function testStrictBackpressure() { + const kChunk = new Uint8Array(16384).fill(65); // 'A' const { writer, readable } = push({ - highWaterMark: 1, + budget: 16384, backpressure: 'strict', }); - // First write should succeed synchronously - assert.strictEqual(writer.writeSync('a'), true); - // Second write should fail synchronously (buffer full) - assert.strictEqual(writer.writeSync('b'), false); + // First write fills the budget + assert.strictEqual(writer.writeSync(kChunk), true); + // Second write rejected (buffer full) + assert.strictEqual(writer.writeSync(kChunk), false); // Consume to free space, then end const resultPromise = text(readable); writer.end(); const data = await resultPromise; - assert.strictEqual(data, 'a'); + assert.strictEqual(data, 'A'.repeat(16384)); } async function testDropOldest() { + const chunk1 = new Uint8Array(16384).fill(49); // '1' + const chunk2 = new Uint8Array(16384).fill(50); // '2' + const chunk3 = new Uint8Array(16384).fill(51); // '3' const { writer, readable } = push({ - highWaterMark: 2, + budget: 32768, backpressure: 'drop-oldest', }); - assert.strictEqual(writer.writeSync('first'), true); - assert.strictEqual(writer.writeSync('second'), true); - // This should drop 'first' — return value is true (write accepted via drop) - assert.strictEqual(writer.writeSync('third'), true); + assert.strictEqual(writer.writeSync(chunk1), true); // 16384 < 32768 + assert.strictEqual(writer.writeSync(chunk2), true); // 32768 >= 32768 + // Buffer full: this drops chunk1, adds chunk3 + assert.strictEqual(writer.writeSync(chunk3), true); writer.end(); - const batches = []; - for await (const batch of readable) { - batches.push(batch); - } - // Should have 'second' and 'third' - const allBytes = []; - for (const batch of batches) { - for (const chunk of batch) { - allBytes.push(...chunk); - } - } - const result = new TextDecoder().decode(new Uint8Array(allBytes)); - assert.strictEqual(result, 'secondthird'); + const data = await text(readable); + // chunk2 ('2' x 16384) + chunk3 ('3' x 16384) survived + assert.strictEqual(data, '2'.repeat(16384) + '3'.repeat(16384)); } async function testDropNewest() { + const kept = new Uint8Array(16384).fill(75); // 'K' + const dropped = new Uint8Array(16384).fill(68); // 'D' const { writer, readable } = push({ - highWaterMark: 1, + budget: 16384, backpressure: 'drop-newest', }); - assert.strictEqual(writer.writeSync('kept'), true); - // This is silently dropped — return value is true (accepted but discarded) - assert.strictEqual(writer.writeSync('dropped'), true); + assert.strictEqual(writer.writeSync(kept), true); + // Buffer full: new write is silently discarded + assert.strictEqual(writer.writeSync(dropped), true); writer.end(); const data = await text(readable); - assert.strictEqual(data, 'kept'); + assert.strictEqual(data, 'K'.repeat(16384)); } async function testBlockBackpressure() { - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); // Fill the buffer - writer.writeSync('a'); + writer.writeSync(kChunk); // Next write should block (not throw, not drop) let writeState = 'pending'; - const writePromise = writer.write('b').then(() => { writeState = 'resolved'; }); + const writePromise = writer.write(kChunk).then(() => { writeState = 'resolved'; }); // The write cannot resolve until the buffer is drained, so a microtask // tick is sufficient to confirm it is still blocked. @@ -82,7 +79,7 @@ async function testBlockBackpressure() { // Read from the consumer to drain const iter = readable[Symbol.asyncIterator](); - const first = await iter.next(); // Drains 'a' + const first = await iter.next(); assert.strictEqual(first.done, false); // After draining, the pending write resolves as a microtask @@ -90,7 +87,7 @@ async function testBlockBackpressure() { assert.strictEqual(writeState, 'resolved'); // Now unblocked writer.endSync(); - const second = await iter.next(); // Read 'b' + const second = await iter.next(); assert.strictEqual(second.done, false); await writePromise; } @@ -99,42 +96,44 @@ async function testBlockWriteSyncDoesNotEnqueue() { // With block policy, writeSync returns false when the buffer is full. // The data is NOT accepted — writeSync only operates on the slots buffer. // The caller should fall back to write() which uses the pending queue. - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); // Fill the buffer - assert.strictEqual(writer.writeSync('a'), true); + assert.strictEqual(writer.writeSync(kChunk), true); // Buffer full: writeSync returns false, data NOT enqueued - assert.strictEqual(writer.writeSync('b'), false); + assert.strictEqual(writer.writeSync(kChunk), false); // Use the async write (try-fallback pattern) — this goes to pending queue const consuming = text(readable); - await writer.write('b'); + await writer.write(kChunk); await writer.end(); - // Both chunks should be delivered + // Both chunks should be delivered (sync + async) const result = await consuming; - assert.strictEqual(result, 'ab'); + assert.strictEqual(result.length, 32768); } async function testStrictPendingQueueOverflow() { - // With highWaterMark: 1 and strict, the pending writes queue is also limited to 1. + // With strict mode, the pending writes queue is limited to 1. // Filling the buffer (1 sync write) + filling the pending queue (1 async write) // should leave no room. A third write must reject with a RangeError. + const kChunk = new Uint8Array(16384); const { writer, readable } = push({ - highWaterMark: 1, + budget: 16384, backpressure: 'strict', }); // Fill the buffer - assert.strictEqual(writer.writeSync('a'), true); + assert.strictEqual(writer.writeSync(kChunk), true); // This async write goes into the pending queue (buffer full, queue has room) - const pendingWrite = writer.write('b'); + const pendingWrite = writer.write(kChunk); // This write should reject: buffer full AND pending queue at capacity await assert.rejects( - () => writer.write('c'), + () => writer.write(kChunk), { code: 'ERR_INVALID_STATE', name: 'RangeError', diff --git a/test/parallel/test-stream-iter-push-basic.js b/test/parallel/test-stream-iter-push-basic.js index bd6944e9492575..1daa2fa16d8c75 100644 --- a/test/parallel/test-stream-iter-push-basic.js +++ b/test/parallel/test-stream-iter-push-basic.js @@ -16,7 +16,7 @@ async function testBasicWriteRead() { } async function testMultipleWrites() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); writer.write('a'); writer.write('b'); @@ -27,19 +27,18 @@ async function testMultipleWrites() { assert.strictEqual(data, 'abc'); } -async function testDesiredSize() { - const { writer } = push({ highWaterMark: 3 }); +async function testCanWrite() { + const kHalf = new Uint8Array(8192); + const { writer } = push({ budget: 16384 }); - assert.strictEqual(writer.desiredSize, 3); - writer.writeSync('a'); - assert.strictEqual(writer.desiredSize, 2); - writer.writeSync('b'); - assert.strictEqual(writer.desiredSize, 1); - writer.writeSync('c'); - assert.strictEqual(writer.desiredSize, 0); + assert.strictEqual(writer.canWrite, true); + writer.writeSync(kHalf); + assert.strictEqual(writer.canWrite, true); // 8192 < 16384 + writer.writeSync(kHalf); + assert.strictEqual(writer.canWrite, false); // 16384 >= 16384 writer.end(); - assert.strictEqual(writer.desiredSize, null); + assert.strictEqual(writer.canWrite, null); } async function testWriterEnd() { @@ -75,7 +74,7 @@ async function testWriterFail() { } async function testConsumerBreak() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); writer.writeSync('a'); writer.writeSync('b'); @@ -87,8 +86,8 @@ async function testConsumerBreak() { break; } - // Writer should now see null desiredSize - assert.strictEqual(writer.desiredSize, null); + // Writer should now see null canWrite + assert.strictEqual(writer.canWrite, null); } async function testAbortSignal() { @@ -119,7 +118,7 @@ async function testPreAbortedSignal() { } async function testConsumerBreakWriteSyncReturnsFalse() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); writer.writeSync('a'); // Break after first batch @@ -130,7 +129,7 @@ async function testConsumerBreakWriteSyncReturnsFalse() { // After consumer break, writeSync should return false assert.strictEqual(writer.writeSync('b'), false); - assert.strictEqual(writer.desiredSize, null); + assert.strictEqual(writer.canWrite, null); } async function testPushWithTransforms() { @@ -160,7 +159,7 @@ async function testInvalidBackpressure() { }); // Valid values should not throw - for (const bp of ['strict', 'block', 'drop-oldest', 'drop-newest']) { + for (const bp of ['strict', 'unbounded', 'drop-oldest', 'drop-newest']) { push({ backpressure: bp }); } } @@ -168,7 +167,7 @@ async function testInvalidBackpressure() { Promise.all([ testBasicWriteRead(), testMultipleWrites(), - testDesiredSize(), + testCanWrite(), testWriterEnd(), testWriterFail(), testConsumerBreak(), diff --git a/test/parallel/test-stream-iter-push-writer.js b/test/parallel/test-stream-iter-push-writer.js index f085f465951449..ddfe89e34784bd 100644 --- a/test/parallel/test-stream-iter-push-writer.js +++ b/test/parallel/test-stream-iter-push-writer.js @@ -6,7 +6,7 @@ const assert = require('assert'); const { push, ondrain, text } = require('stream/iter'); async function testOndrain() { - const { writer } = push({ highWaterMark: 1 }); + const { writer } = push({ budget: 16384 }); // With space available, ondrain resolves immediately const drainResult = ondrain(writer); @@ -39,13 +39,14 @@ async function testOndrainProtocolErrorPropagates() { } async function testWriteWithSignalRejects() { - const { writer, readable } = push({ highWaterMark: 1 }); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384 }); // Fill the buffer so write will block - writer.writeSync('a'); + writer.writeSync(kChunk); const ac = new AbortController(); - const writePromise = writer.write('b', { signal: ac.signal }); + const writePromise = writer.write(kChunk, { signal: ac.signal }); // Signal fires while write is pending ac.abort(); @@ -59,7 +60,7 @@ async function testWriteWithSignalRejects() { } async function testWriteWithPreAbortedSignal() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); // Pre-aborted signal should reject immediately await assert.rejects( @@ -75,48 +76,49 @@ async function testWriteWithPreAbortedSignal() { } async function testCancelledWriteRemovedFromQueue() { - const { writer, readable } = push({ highWaterMark: 1 }); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384 }); // Fill the buffer - writer.writeSync('first'); + writer.writeSync(kChunk); const ac = new AbortController(); // This write should be queued since buffer is full - const cancelledWrite = writer.write('cancelled', { signal: ac.signal }); + const cancelledWrite = writer.write(kChunk, { signal: ac.signal }); // Cancel it ac.abort(); await cancelledWrite.catch(() => {}); - // Drain 'first' to make room for the replacement write + // Drain to make room for the replacement write const iter = readable[Symbol.asyncIterator](); await iter.next(); // The cancelled write should NOT occupy a pending slot. // A new write should succeed now that the buffer has room. - await writer.write('second'); + await writer.write(kChunk); writer.end(); const result = await iter.next(); - // 'second' should be the next (and only remaining) chunk - const decoder = new TextDecoder(); - let data = ''; + assert.ok(!result.done); + let totalBytes = 0; for (const chunk of result.value) { - data += decoder.decode(chunk, { stream: true }); + totalBytes += chunk.byteLength; } - assert.strictEqual(data, 'second'); + assert.strictEqual(totalBytes, 16384); await iter.return(); } async function testOndrainResolvesFalseOnConsumerBreak() { - const { writer, readable } = push({ highWaterMark: 1 }); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384 }); - // Fill the buffer so desiredSize = 0 - writer.writeSync('a'); + // Fill the buffer so canWrite = false + writer.writeSync(kChunk); // Also queue a pending write so that reading one chunk - // doesn't clear backpressure (the pending write refills the slot) - const pendingWrite = writer.write('b'); + // doesn't clear backpressure (the pending write refills the buffer) + const pendingWrite = writer.write(kChunk); // Start a drain wait - still at capacity const drainPromise = ondrain(writer); @@ -132,14 +134,15 @@ async function testOndrainResolvesFalseOnConsumerBreak() { } async function testOndrainRejectsOnConsumerThrow() { - const { writer, readable } = push({ highWaterMark: 1 }); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384 }); - // Fill the buffer so desiredSize = 0 - writer.writeSync('a'); + // Fill the buffer so canWrite = false + writer.writeSync(kChunk); // Also queue a pending write so that reading one chunk - // doesn't clear backpressure (the pending write refills the slot) - const pendingWrite = writer.write('b'); + // doesn't clear backpressure (the pending write refills the buffer) + const pendingWrite = writer.write(kChunk); // Start a drain wait - still at capacity const drainPromise = ondrain(writer); @@ -154,7 +157,7 @@ async function testOndrainRejectsOnConsumerThrow() { } async function testWritev() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); const enc = new TextEncoder(); writer.writev([enc.encode('hel'), enc.encode('lo')]); writer.endSync(); @@ -163,7 +166,7 @@ async function testWritev() { } async function testWritevSync() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); const enc = new TextEncoder(); assert.strictEqual(writer.writevSync([enc.encode('hel'), enc.encode('lo')]), true); writer.endSync(); @@ -172,7 +175,7 @@ async function testWritevSync() { } async function testWritevSyncInvalidChunkDoesNotQueue() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); assert.throws( () => writer.writevSync([1]), @@ -194,7 +197,7 @@ async function testWritevSyncInvalidChunkDoesNotQueue() { } async function testWritevMixedTypes() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); // Mix strings and Uint8Arrays writer.writev(['hel', new TextEncoder().encode('lo')]); writer.endSync(); @@ -279,8 +282,9 @@ async function testWriteUint8Array() { } async function testOndrainWaitsForDrain() { - const { writer, readable } = push({ highWaterMark: 1 }); - writer.writeSync('a'); // Fills buffer + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384 }); + writer.writeSync(kChunk); // Fills budget let drainState = 'pending'; const drainPromise = ondrain(writer).then((v) => { drainState = v; }); @@ -299,8 +303,9 @@ async function testOndrainWaitsForDrain() { // Consumer throw causes subsequent writes to reject with consumer's error async function testConsumerThrowRejectsWrites() { - const { writer, readable } = push({ highWaterMark: 1 }); - writer.writeSync('a'); + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384 }); + writer.writeSync(kChunk); const iter = readable[Symbol.asyncIterator](); await iter.throw(new Error('consumer boom')); @@ -385,11 +390,12 @@ async function testConsumerThrowRejectsPendingRead() { // end() while writes are pending rejects those writes async function testEndRejectsPendingWrites() { - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); - writer.writeSync('a'); // fill buffer + const kChunk = new Uint8Array(16384); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); + writer.writeSync(kChunk); // fill budget // This write blocks on backpressure - const writePromise = writer.write('b'); + const writePromise = writer.write(kChunk); await new Promise(setImmediate); @@ -407,7 +413,7 @@ async function testEndRejectsPendingWrites() { } async function testEndIdempotentWhenClosed() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); await writer.write('hello'); // Start consuming concurrently (end() waits for drain) const consume = (async () => { @@ -423,7 +429,7 @@ async function testEndIdempotentWhenClosed() { } async function testAsyncDispose() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); writer.writeSync('hello'); // Symbol.asyncDispose calls fail() with no argument await writer[Symbol.asyncDispose](); @@ -439,7 +445,7 @@ async function testAsyncDispose() { } async function testSyncDispose() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); writer.writeSync('hello'); // Symbol.dispose calls fail() with no argument writer[Symbol.dispose](); @@ -455,7 +461,7 @@ async function testSyncDispose() { } async function testEndRejectsWhenErrored() { - const { writer, readable } = push({ highWaterMark: 10 }); + const { writer, readable } = push({ budget: 16384 }); await writer.write('hello'); const err = new Error('boom'); await writer.fail(err); diff --git a/test/parallel/test-stream-iter-share-async.js b/test/parallel/test-stream-iter-share-async.js index ad382c97e03cf7..f55ca35f69778c 100644 --- a/test/parallel/test-stream-iter-share-async.js +++ b/test/parallel/test-stream-iter-share-async.js @@ -30,7 +30,7 @@ async function testShareMultipleConsumers() { yield [new TextEncoder().encode('chunk3')]; } - const shared = share(gen(), { highWaterMark: 16 }); + const shared = share(gen(), { budget: 16384 }); const c1 = shared.pull(); const c2 = shared.pull(); @@ -97,7 +97,7 @@ async function testShareCancelMidIteration() { sourceReturnCalled = true; } } - const shared = share(gen(), { highWaterMark: 16 }); + const shared = share(gen(), { budget: 16384 }); const consumer = shared.pull(); const items = []; @@ -141,8 +141,8 @@ async function testShareAbortSignal() { yield [enc.encode('b')]; } const shared = share(source(), { - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', signal: ac.signal, }); const fast = shared.pull()[Symbol.asyncIterator](); @@ -259,7 +259,7 @@ async function testShareLateJoiningConsumer() { yield [enc.encode('b')]; yield [enc.encode('c')]; } - const shared = share(gen(), { highWaterMark: 16 }); + const shared = share(gen(), { budget: 16384 }); // First consumer reads all data const c1 = shared.pull(); @@ -281,7 +281,7 @@ async function testShareConsumerBreak() { yield [enc.encode('b')]; yield [enc.encode('c')]; } - const shared = share(gen(), { highWaterMark: 16 }); + const shared = share(gen(), { budget: 16384 }); const c1 = shared.pull(); const c2 = shared.pull(); diff --git a/test/parallel/test-stream-iter-share-from.js b/test/parallel/test-stream-iter-share-from.js index 362bfef78f1a2e..295760bfb2e935 100644 --- a/test/parallel/test-stream-iter-share-from.js +++ b/test/parallel/test-stream-iter-share-from.js @@ -97,13 +97,13 @@ function testSyncShareProtocolReturnsNonObject() { } // ============================================================================= -// Block backpressure: two consumers, slow consumer blocks the source +// Unbounded backpressure: two consumers, slow consumer blocks the source // ============================================================================= async function testShareBlockBackpressure() { - // A source that yields 5 items. With two consumers and highWaterMark: 2, + // A source that yields 5 items. With two consumers and budget: 16384, // the fast consumer drives the source forward. The slow consumer holds back - // trimming, causing the buffer to fill. 'block' mode should stall the + // trimming, causing the buffer to fill. 'unbounded' mode should stall the // source pull until the slow consumer catches up. const enc = new TextEncoder(); async function* source() { @@ -111,7 +111,7 @@ async function testShareBlockBackpressure() { yield [enc.encode(`item${i}`)]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'block' }); + const shared = share(source(), { budget: 16384, backpressure: 'unbounded' }); const fast = shared.pull(); const slow = shared.pull(); @@ -135,71 +135,75 @@ async function testShareDropOldest() { // misses old data and only sees recent items. async function* source() { for (let i = 0; i < 4; i++) { - yield [new TextEncoder().encode(`${i}`)]; + const chunk = new Uint8Array(16384); + chunk[0] = i; // Tag first byte with index + yield [chunk]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'drop-oldest' }); + const shared = share(source(), { budget: 32768, backpressure: 'drop-oldest' }); const fast = shared.pull(); const slow = shared.pull(); // Fast consumer reads all items - const fastItems = []; + const fastIndices = []; for await (const batch of fast) { for (const chunk of batch) { - fastItems.push(new TextDecoder().decode(chunk)); + fastIndices.push(chunk[0]); } } - assert.strictEqual(fastItems.length, 4); + assert.strictEqual(fastIndices.length, 4); // Slow consumer reads after fast is done — old items were dropped - const slowItems = []; + const slowIndices = []; for await (const batch of slow) { for (const chunk of batch) { - slowItems.push(new TextDecoder().decode(chunk)); + slowIndices.push(chunk[0]); } } // The slow consumer should see fewer items than were produced - assert.ok(slowItems.length < 4, - `Expected < 4 items after drop-oldest, got ${slowItems.length}`); - assert.ok(slowItems.length > 0, + assert.ok(slowIndices.length < 4, + `Expected < 4 items after drop-oldest, got ${slowIndices.length}`); + assert.ok(slowIndices.length > 0, 'Expected at least some items after drop-oldest'); // The last item should always be present (most recent items kept) - assert.strictEqual(slowItems[slowItems.length - 1], '3'); + assert.strictEqual(slowIndices[slowIndices.length - 1], 3); } async function testShareDropNewest() { // With drop-newest and a stalled consumer, the async path allows the - // buffer to grow beyond highWaterMark (the "drop" applies to the + // buffer to grow beyond budget (the "drop" applies to the // backpressure signal, not the buffer contents). Both consumers // ultimately see all items. async function* source() { for (let i = 0; i < 4; i++) { - yield [new TextEncoder().encode(`${i}`)]; + const chunk = new Uint8Array(16384); + chunk[0] = i; + yield [chunk]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'drop-newest' }); + const shared = share(source(), { budget: 32768, backpressure: 'drop-newest' }); const fast = shared.pull(); const slow = shared.pull(); // Fast consumer reads all items - const fastItems = []; + const fastIndices = []; for await (const batch of fast) { for (const chunk of batch) { - fastItems.push(new TextDecoder().decode(chunk)); + fastIndices.push(chunk[0]); } } - assert.strictEqual(fastItems.length, 4); + assert.strictEqual(fastIndices.length, 4); - // Slow consumer also sees all items (buffer grew past hwm) - const slowItems = []; + // Slow consumer also sees all items (buffer grew past budget) + const slowIndices = []; for await (const batch of slow) { for (const chunk of batch) { - slowItems.push(new TextDecoder().decode(chunk)); + slowIndices.push(chunk[0]); } } - assert.strictEqual(slowItems.length, 4); - assert.strictEqual(slowItems[0], '0'); - assert.strictEqual(slowItems[3], '3'); + assert.strictEqual(slowIndices.length, 4); + assert.strictEqual(slowIndices[0], 0); + assert.strictEqual(slowIndices[3], 3); } // ============================================================================= @@ -209,16 +213,16 @@ async function testShareDropNewest() { async function testShareStrictBackpressure() { async function* source() { for (let i = 0; i < 10; i++) { - yield [new TextEncoder().encode(`${i}`)]; + yield [new Uint8Array(16384)]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'strict' }); + const shared = share(source(), { budget: 32768, backpressure: 'strict' }); const fast = shared.pull(); // Create a second consumer that never reads — this prevents buffer trimming shared.pull(); // The fast consumer's pulls will eventually cause the buffer to exceed - // the highWaterMark (since the slow consumer prevents trimming), + // the budget (since the slow consumer prevents trimming), // triggering an ERR_OUT_OF_RANGE error. await assert.rejects(async () => { // eslint-disable-next-line no-unused-vars diff --git a/test/parallel/test-stream-iter-share-sync.js b/test/parallel/test-stream-iter-share-sync.js index 98ad74fe4f1c72..2b0b1944a7ff80 100644 --- a/test/parallel/test-stream-iter-share-sync.js +++ b/test/parallel/test-stream-iter-share-sync.js @@ -30,7 +30,7 @@ async function testShareSyncMultipleConsumers() { yield [enc.encode('c')]; } - const shared = shareSync(gen(), { highWaterMark: 16 }); + const shared = shareSync(gen(), { budget: 16384 }); const c1 = shared.pull(); const c2 = shared.pull(); @@ -70,7 +70,7 @@ function testShareSyncCancelMidIteration() { sourceReturnCalled = true; } } - const shared = shareSync(gen(), { highWaterMark: 16 }); + const shared = shareSync(gen(), { budget: 16384 }); const consumer = shared.pull(); const items = []; @@ -97,7 +97,7 @@ function testShareSyncCancelWithReason() { yield [enc.encode('b')]; yield [enc.encode('c')]; } - const shared = shareSync(gen(), { highWaterMark: 16 }); + const shared = shareSync(gen(), { budget: 16384 }); const c1 = shared.pull(); const c2 = shared.pull(); diff --git a/test/parallel/test-stream-iter-validation.js b/test/parallel/test-stream-iter-validation.js index ecefaeb171ec41..fabc70d2a3765a 100644 --- a/test/parallel/test-stream-iter-validation.js +++ b/test/parallel/test-stream-iter-validation.js @@ -20,18 +20,21 @@ const { // push() validation // ============================================================================= -// HighWaterMark must be integer >= 1 -assert.throws(() => push({ highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => push({ highWaterMark: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); -// Values < 1 are clamped to 1 -assert.strictEqual(push({ highWaterMark: 0 }).writer.desiredSize, 1); -assert.strictEqual(push({ highWaterMark: -1 }).writer.desiredSize, 1); -assert.strictEqual(push({ highWaterMark: -100 }).writer.desiredSize, 1); +// Budget must be integer >= 16384 +assert.throws(() => push({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => push({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); +// Values < 16384 are rejected +assert.throws(() => push({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => push({ budget: -1 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => push({ budget: -100 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => push({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' }); +// 16384 is the minimum accepted value +assert.strictEqual(push({ budget: 16384 }).writer.canWrite, true); // MAX_SAFE_INTEGER is accepted -assert.strictEqual(push({ highWaterMark: Number.MAX_SAFE_INTEGER }).writer.desiredSize, - Number.MAX_SAFE_INTEGER); +assert.strictEqual(push({ budget: Number.MAX_SAFE_INTEGER }).writer.canWrite, + true); // Values above MAX_SAFE_INTEGER are rejected by validateInteger -assert.throws(() => push({ highWaterMark: Number.MAX_SAFE_INTEGER + 1 }), +assert.throws(() => push({ budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); // Signal must be AbortSignal @@ -69,33 +72,27 @@ assert.throws(() => duplex('bad'), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => duplex({ a: 42 }), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => duplex({ b: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); -// highWaterMark validation (cascades through to push()) -assert.throws(() => duplex({ highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => duplex({ highWaterMark: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); -assert.throws(() => duplex({ highWaterMark: Number.MAX_SAFE_INTEGER + 1 }), +// Budget validation (cascades through to push()) +assert.throws(() => duplex({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => duplex({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => duplex({ budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); -// Values < 1 are clamped to 1 (both directions) -{ - const [a, b] = duplex({ highWaterMark: 0 }); - assert.strictEqual(a.writer.desiredSize, 1); - assert.strictEqual(b.writer.desiredSize, 1); - a.close(); - b.close(); -} +// Values < 16384 are rejected (both directions) +assert.throws(() => duplex({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' }); // MAX_SAFE_INTEGER is accepted { - const [a, b] = duplex({ highWaterMark: Number.MAX_SAFE_INTEGER }); - assert.strictEqual(a.writer.desiredSize, Number.MAX_SAFE_INTEGER); - assert.strictEqual(b.writer.desiredSize, Number.MAX_SAFE_INTEGER); + const [a, b] = duplex({ budget: Number.MAX_SAFE_INTEGER }); + assert.strictEqual(a.writer.canWrite, true); + assert.strictEqual(b.writer.canWrite, true); a.close(); b.close(); } // Per-direction overrides { - const [a, b] = duplex({ a: { highWaterMark: 0 }, b: { highWaterMark: 5 } }); - assert.strictEqual(a.writer.desiredSize, 1); // clamped - assert.strictEqual(b.writer.desiredSize, 5); + const [a, b] = duplex({ a: { budget: 16384 }, b: { budget: 32768 } }); + assert.strictEqual(a.writer.canWrite, true); + assert.strictEqual(b.writer.canWrite, true); a.close(); b.close(); } @@ -118,29 +115,27 @@ assert.throws(() => pullSync(fromSync('a'), 42), { code: 'ERR_INVALID_ARG_TYPE' // broadcast() validation // ============================================================================= -assert.throws(() => broadcast({ highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => broadcast({ highWaterMark: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); -assert.throws(() => broadcast({ highWaterMark: Number.MAX_SAFE_INTEGER + 1 }), +assert.throws(() => broadcast({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => broadcast({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => broadcast({ budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); -// Values < 1 are clamped to 1 (need a consumer for desiredSize to work) -{ - const bc = broadcast({ highWaterMark: 0 }); - bc.broadcast.push(); - assert.strictEqual(bc.writer.desiredSize, 1); - bc.writer.endSync(); -} +// Values < 16384 are rejected +assert.throws(() => broadcast({ budget: 0 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => broadcast({ budget: -1 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => broadcast({ budget: 16383 }), { code: 'ERR_OUT_OF_RANGE' }); +// 16384 is the minimum accepted value { - const bc = broadcast({ highWaterMark: -1 }); + const bc = broadcast({ budget: 16384 }); bc.broadcast.push(); - assert.strictEqual(bc.writer.desiredSize, 1); + assert.strictEqual(bc.writer.canWrite, true); bc.writer.endSync(); } // MAX_SAFE_INTEGER is accepted { - const bc = broadcast({ highWaterMark: Number.MAX_SAFE_INTEGER }); + const bc = broadcast({ budget: Number.MAX_SAFE_INTEGER }); bc.broadcast.push(); - assert.strictEqual(bc.writer.desiredSize, Number.MAX_SAFE_INTEGER); + assert.strictEqual(bc.writer.canWrite, true); bc.writer.endSync(); } @@ -165,30 +160,36 @@ assert.throws(() => Broadcast.from(42), { code: 'ERR_INVALID_ARG_TYPE' }); // ============================================================================= assert.throws(() => share(42), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => share(from('a'), { highWaterMark: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => share(from('a'), { highWaterMark: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); -assert.throws(() => share(from('a'), { highWaterMark: Number.MAX_SAFE_INTEGER + 1 }), +assert.throws(() => share(from('a'), { budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => share(from('a'), { budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => share(from('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); assert.throws(() => share(from('a'), { signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' }); assert.throws(() => share(from('a'), { backpressure: 'bad' }), { code: 'ERR_INVALID_ARG_VALUE' }); -// share() values < 1 are clamped (no desiredSize, but accepts the value) -share(from('a'), { highWaterMark: 0 }).cancel(); -share(from('a'), { highWaterMark: -1 }).cancel(); -share(from('a'), { highWaterMark: Number.MAX_SAFE_INTEGER }).cancel(); +// share() values < 16384 are rejected +assert.throws(() => share(from('a'), { budget: 0 }), { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => share(from('a'), { budget: -1 }), { code: 'ERR_OUT_OF_RANGE' }); +// 16384 is the minimum, MAX_SAFE_INTEGER is accepted +share(from('a'), { budget: 16384 }).cancel(); +share(from('a'), { budget: Number.MAX_SAFE_INTEGER }).cancel(); assert.throws(() => shareSync(42), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => shareSync(fromSync('a'), { highWaterMark: 'bad' }), +assert.throws(() => shareSync(fromSync('a'), { budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); -assert.throws(() => shareSync(fromSync('a'), { highWaterMark: 1.5 }), +assert.throws(() => shareSync(fromSync('a'), { budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); -assert.throws(() => shareSync(fromSync('a'), { highWaterMark: Number.MAX_SAFE_INTEGER + 1 }), +assert.throws(() => shareSync(fromSync('a'), { budget: Number.MAX_SAFE_INTEGER + 1 }), { code: 'ERR_OUT_OF_RANGE' }); -// shareSync() values < 1 are clamped (accepts the value) -shareSync(fromSync('a'), { highWaterMark: 0 }).cancel(); -shareSync(fromSync('a'), { highWaterMark: -1 }).cancel(); -shareSync(fromSync('a'), { highWaterMark: Number.MAX_SAFE_INTEGER }).cancel(); +// shareSync() values < 16384 are rejected +assert.throws(() => shareSync(fromSync('a'), { budget: 0 }), + { code: 'ERR_OUT_OF_RANGE' }); +assert.throws(() => shareSync(fromSync('a'), { budget: -1 }), + { code: 'ERR_OUT_OF_RANGE' }); +// 16384 is the minimum, MAX_SAFE_INTEGER is accepted +shareSync(fromSync('a'), { budget: 16384 }).cancel(); +shareSync(fromSync('a'), { budget: Number.MAX_SAFE_INTEGER }).cancel(); // Share.from / SyncShare.fromSync reject non-iterable assert.throws(() => Share.from(42), { code: 'ERR_INVALID_ARG_TYPE' }); @@ -316,27 +317,27 @@ async function testAsyncValidation() { // Push with valid options { - const { writer } = push({ highWaterMark: 2 }); + const { writer } = push({ budget: 16384 }); writer.writeSync('hello'); writer.endSync(); } // Duplex with valid options { - const [a, b] = duplex({ highWaterMark: 2 }); + const [a, b] = duplex({ budget: 16384 }); a.close(); b.close(); } // Broadcast with valid options { - const { writer } = broadcast({ highWaterMark: 4 }); + const { writer } = broadcast({ budget: 16384 }); writer.endSync(); } // Share with valid options { - const shared = share(from('hello'), { highWaterMark: 4 }); + const shared = share(from('hello'), { budget: 16384 }); shared.cancel(); } diff --git a/test/parallel/test-stream-iter-writable-from.js b/test/parallel/test-stream-iter-writable-from.js index 529c88ad3662f5..46cfb627cb4a1b 100644 --- a/test/parallel/test-stream-iter-writable-from.js +++ b/test/parallel/test-stream-iter-writable-from.js @@ -17,7 +17,7 @@ const { // ============================================================================= async function testBasicWrite() { - const { writer, readable } = push({ backpressure: 'block' }); + const { writer, readable } = push({ backpressure: 'unbounded' }); const writable = toWritable(writer); writable.write('hello'); @@ -324,7 +324,7 @@ function testInvalidWriterThrows() { // ============================================================================= async function testRoundTrip() { - const { writer, readable } = push({ backpressure: 'block' }); + const { writer, readable } = push({ backpressure: 'unbounded' }); const writable = toWritable(writer); const data = 'round trip test data'; @@ -340,7 +340,7 @@ async function testRoundTrip() { // ============================================================================= async function testPushWriterBlockBackpressureNoDuplicate() { - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); const writable = toWritable(writer); await new Promise((resolve, reject) => { @@ -362,7 +362,7 @@ async function testPushWriterBlockBackpressureNoDuplicate() { // ============================================================================= async function testPushWriterBlockBackpressureWritevNoDuplicate() { - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); const writable = toWritable(writer); await new Promise((resolve, reject) => { @@ -387,7 +387,7 @@ async function testPushWriterBlockBackpressureWritevNoDuplicate() { // ============================================================================= async function testSequentialWrites() { - const { writer, readable } = push({ backpressure: 'block' }); + const { writer, readable } = push({ backpressure: 'unbounded' }); const writable = toWritable(writer); for (let i = 0; i < 10; i++) { diff --git a/test/parallel/test-stream-iter-writable-interop.js b/test/parallel/test-stream-iter-writable-interop.js index e7b83ac22841ce..110e8e0a7d2eee 100644 --- a/test/parallel/test-stream-iter-writable-interop.js +++ b/test/parallel/test-stream-iter-writable-interop.js @@ -52,7 +52,7 @@ async function testBasicWrite() { }, }); - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); await pipeTo(from('hello world'), writer); assert.strictEqual(Buffer.concat(chunks).toString(), 'hello world'); @@ -95,7 +95,7 @@ async function testBlockWaitsForDrain() { }, }); - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); await writer.write('a'); await writer.write('b'); @@ -117,7 +117,7 @@ async function testBlockErrorRejectsPendingWrite() { }, }); - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); // First write fills the buffer, waits for drain const writePromise = writer.write('data that will block'); @@ -251,8 +251,8 @@ async function testDropNewestCountsBytes() { await writer.write('12345'); // 5 bytes, accepted await writer.write('67890'); // 5 bytes, dropped - // desiredSize should be 0 (buffer is full) - assert.strictEqual(writer.desiredSize, 0); + // canWrite should be false (buffer is full) + assert.strictEqual(writer.canWrite, false); } // ============================================================================= @@ -299,7 +299,7 @@ async function testWritev() { }, }); - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); await writer.writev([ new TextEncoder().encode('hello'), new TextEncoder().encode(' '), @@ -365,10 +365,10 @@ async function testFail() { } // ============================================================================= -// desiredSize reflects buffer state +// canWrite reflects buffer state // ============================================================================= -function testDesiredSize() { +function testCanWrite() { const writable = new Writable({ highWaterMark: 100, write(chunk, enc, cb) { @@ -377,19 +377,19 @@ function testDesiredSize() { }); const writer = fromWritable(writable); - assert.strictEqual(writer.desiredSize, 100); + assert.strictEqual(writer.canWrite, true); } // ============================================================================= -// desiredSize is null when destroyed +// canWrite is null when destroyed // ============================================================================= -function testDesiredSizeNull() { +function testCanWriteNull() { const writable = new Writable({ write(chunk, enc, cb) { cb(); } }); const writer = fromWritable(writable); writable.destroy(); - assert.strictEqual(writer.desiredSize, null); + assert.strictEqual(writer.canWrite, null); } // ============================================================================= @@ -481,7 +481,7 @@ async function testPipeToWithTransform() { }, }); - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); await pipeTo(from('hello via transform'), compressGzip(), writer); const decompressed = await text( @@ -590,7 +590,7 @@ async function testFailRejectsPendingWaiters() { }); writable.on('error', () => {}); // Prevent unhandled error - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); // This write will block on drain const writePromise = writer.write('blocked data'); @@ -613,7 +613,7 @@ async function testDisposeRejectsPendingWaiters() { }, }); - const writer = fromWritable(writable, { backpressure: 'block' }); + const writer = fromWritable(writable, { backpressure: 'unbounded' }); // This write will block on drain const writePromise = writer.write('blocked data'); @@ -647,8 +647,8 @@ function testObjectModeThrows() { testFunctionExists(); testSyncMethodsReturnFalse(); testEndSyncReturnsNegativeOne(); -testDesiredSize(); -testDesiredSizeNull(); +testCanWrite(); +testCanWriteNull(); testDrainableNull(); testDropOldestThrows(); testInvalidBackpressureThrows();