From 25f2b44d1ecbd6109ce56ad9b85790b2025f9063 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 12 Jul 2026 16:41:45 -0700 Subject: [PATCH 1/2] stream: update iterable streams to use budget backpressure Updates the implementation to use the updated backpressure model that landed here: https://github.com/WinterTC55/iter-streams/pull/23 Signed-off-by: James M Snell Assisted-by: Opencode/Opus --- doc/api/stream_iter.md | 143 +++++++++--------- lib/internal/streams/iter/broadcast.js | 83 +++++----- lib/internal/streams/iter/classic.js | 13 +- lib/internal/streams/iter/duplex.js | 8 +- lib/internal/streams/iter/pull.js | 12 +- lib/internal/streams/iter/push.js | 83 +++++----- lib/internal/streams/iter/share.js | 93 ++++++++---- lib/internal/streams/iter/utils.js | 22 +-- ...test-stream-iter-broadcast-backpressure.js | 16 +- .../test-stream-iter-broadcast-basic.js | 10 +- .../test-stream-iter-broadcast-coverage.js | 13 +- test/parallel/test-stream-iter-duplex.js | 8 +- .../test-stream-iter-pipeto-writev.js | 6 +- test/parallel/test-stream-iter-pull-async.js | 2 +- .../test-stream-iter-push-backpressure.js | 14 +- test/parallel/test-stream-iter-push-basic.js | 30 ++-- test/parallel/test-stream-iter-push-writer.js | 38 ++--- test/parallel/test-stream-iter-share-async.js | 12 +- test/parallel/test-stream-iter-share-from.js | 18 +-- test/parallel/test-stream-iter-share-sync.js | 6 +- test/parallel/test-stream-iter-validation.js | 104 ++++++------- .../test-stream-iter-writable-from.js | 10 +- .../test-stream-iter-writable-interop.js | 34 ++--- 23 files changed, 407 insertions(+), 371 deletions(-) diff --git a/doc/api/stream_iter.md b/doc/api/stream_iter.md index 0641246d9f013a..a71c7474369651 100644 --- a/doc/api/stream_iter.md +++ b/doc/api/stream_iter.md @@ -186,22 +186,22 @@ The API supports two models: Pull streams have natural backpressure -- the consumer drives the pace, so the source is never read faster than the consumer can process. Push streams need explicit backpressure because the producer and consumer run -independently. The `highWaterMark` and `backpressure` options on `push()`, +independently. The `budget` and `backpressure` options on `push()`, `broadcast()`, and `share()` control how this works. #### The two-buffer model Push streams use a two-part buffering system. Think of it like a bucket -(slots) being filled through a hose (pending writes), with a float valve +(buffer) being filled through a hose (pending writes), with a float valve that closes when the bucket is full: ```text - highWaterMark (e.g., 3) + budget (e.g., 16384) | Producer v | +---------+ v | | - [ write() ] ----+ +--->| 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/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/test/parallel/test-stream-iter-broadcast-backpressure.js b/test/parallel/test-stream-iter-broadcast-backpressure.js index d1e466c3ac44cf..597b37de96f6be 100644 --- a/test/parallel/test-stream-iter-broadcast-backpressure.js +++ b/test/parallel/test-stream-iter-broadcast-backpressure.js @@ -11,7 +11,7 @@ const { broadcast, text } = require('stream/iter'); async function testDropOldest() { const { writer, broadcast: bc } = broadcast({ - highWaterMark: 2, + budget: 16384, backpressure: 'drop-oldest', }); const consumer = bc.push(); @@ -28,7 +28,7 @@ async function testDropOldest() { async function testDropNewest() { const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, + budget: 16384, backpressure: 'drop-newest', }); const consumer = bc.push(); @@ -48,8 +48,8 @@ async function testDropNewest() { async function testBlockBackpressure() { const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consumer = bc.push(); writer.writeSync('a'); @@ -77,8 +77,8 @@ async function testBlockBackpressure() { // Verify block backpressure data flows correctly end-to-end async function testBlockBackpressureContent() { const { writer, broadcast: bc } = broadcast({ - highWaterMark: 1, - backpressure: 'block', + budget: 16384, + backpressure: 'unbounded', }); const consumer = bc.push(); @@ -107,7 +107,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 +119,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..530f2ef8030279 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,12 +74,12 @@ async function testConsumerCount() { // ============================================================================= async function testWriteSync() { - const { writer, broadcast: bc } = broadcast({ highWaterMark: 2 }); + 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) + // Buffer full (budget=16384, strict policy) assert.strictEqual(writer.writeSync('c'), false); writer.endSync(); @@ -89,7 +89,7 @@ async function testWriteSync() { } 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..aa264331b2d0ae 100644 --- a/test/parallel/test-stream-iter-broadcast-coverage.js +++ b/test/parallel/test-stream-iter-broadcast-coverage.js @@ -15,8 +15,8 @@ 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(); @@ -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..bc6ed2d1c5fa48 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); @@ -156,7 +156,7 @@ async function testPushWriterBlockSyncFalseAccepted() { async function testPipeToSyncPushWriterStrictFalseRejected() { const decoder = new TextDecoder(); - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); const total = pipeToSync(['a', 'b'], writer, { preventClose: true }); assert.strictEqual(total, 1); 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..9f528f332dcd08 100644 --- a/test/parallel/test-stream-iter-push-backpressure.js +++ b/test/parallel/test-stream-iter-push-backpressure.js @@ -7,7 +7,7 @@ const { push, text } = require('stream/iter'); async function testStrictBackpressure() { const { writer, readable } = push({ - highWaterMark: 1, + budget: 16384, backpressure: 'strict', }); @@ -25,7 +25,7 @@ async function testStrictBackpressure() { async function testDropOldest() { const { writer, readable } = push({ - highWaterMark: 2, + budget: 16384, backpressure: 'drop-oldest', }); @@ -52,7 +52,7 @@ async function testDropOldest() { async function testDropNewest() { const { writer, readable } = push({ - highWaterMark: 1, + budget: 16384, backpressure: 'drop-newest', }); @@ -66,7 +66,7 @@ async function testDropNewest() { } async function testBlockBackpressure() { - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); // Fill the buffer writer.writeSync('a'); @@ -99,7 +99,7 @@ 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 { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); // Fill the buffer assert.strictEqual(writer.writeSync('a'), true); @@ -118,11 +118,11 @@ async function testBlockWriteSyncDoesNotEnqueue() { } async function testStrictPendingQueueOverflow() { - // With highWaterMark: 1 and strict, the pending writes queue is also limited to 1. + // With budget: 16384 and strict, the pending writes queue is also 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 { writer, readable } = push({ - highWaterMark: 1, + budget: 16384, backpressure: 'strict', }); diff --git a/test/parallel/test-stream-iter-push-basic.js b/test/parallel/test-stream-iter-push-basic.js index bd6944e9492575..859a95fcc1e213 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,19 @@ async function testMultipleWrites() { assert.strictEqual(data, 'abc'); } -async function testDesiredSize() { - const { writer } = push({ highWaterMark: 3 }); +async function testCanWrite() { + const { writer } = push({ budget: 16384 }); - assert.strictEqual(writer.desiredSize, 3); + assert.strictEqual(writer.canWrite, true); writer.writeSync('a'); - assert.strictEqual(writer.desiredSize, 2); + assert.strictEqual(writer.canWrite, true); writer.writeSync('b'); - assert.strictEqual(writer.desiredSize, 1); + assert.strictEqual(writer.canWrite, true); writer.writeSync('c'); - assert.strictEqual(writer.desiredSize, 0); + assert.strictEqual(writer.canWrite, false); writer.end(); - assert.strictEqual(writer.desiredSize, null); + assert.strictEqual(writer.canWrite, null); } async function testWriterEnd() { @@ -75,7 +75,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 +87,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 +119,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 +130,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 +160,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 +168,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..cc797b9d7ffe0c 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,7 +39,7 @@ async function testOndrainProtocolErrorPropagates() { } async function testWriteWithSignalRejects() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); // Fill the buffer so write will block writer.writeSync('a'); @@ -59,7 +59,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,7 +75,7 @@ async function testWriteWithPreAbortedSignal() { } async function testCancelledWriteRemovedFromQueue() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); // Fill the buffer writer.writeSync('first'); @@ -109,9 +109,9 @@ async function testCancelledWriteRemovedFromQueue() { } async function testOndrainResolvesFalseOnConsumerBreak() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); - // Fill the buffer so desiredSize = 0 + // Fill the buffer so canWrite = false writer.writeSync('a'); // Also queue a pending write so that reading one chunk @@ -132,9 +132,9 @@ async function testOndrainResolvesFalseOnConsumerBreak() { } async function testOndrainRejectsOnConsumerThrow() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); - // Fill the buffer so desiredSize = 0 + // Fill the buffer so canWrite = false writer.writeSync('a'); // Also queue a pending write so that reading one chunk @@ -154,7 +154,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 +163,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 +172,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 +194,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,7 +279,7 @@ async function testWriteUint8Array() { } async function testOndrainWaitsForDrain() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); writer.writeSync('a'); // Fills buffer let drainState = 'pending'; @@ -299,7 +299,7 @@ async function testOndrainWaitsForDrain() { // Consumer throw causes subsequent writes to reject with consumer's error async function testConsumerThrowRejectsWrites() { - const { writer, readable } = push({ highWaterMark: 1 }); + const { writer, readable } = push({ budget: 16384 }); writer.writeSync('a'); const iter = readable[Symbol.asyncIterator](); @@ -385,7 +385,7 @@ async function testConsumerThrowRejectsPendingRead() { // end() while writes are pending rejects those writes async function testEndRejectsPendingWrites() { - const { writer, readable } = push({ highWaterMark: 1, backpressure: 'block' }); + const { writer, readable } = push({ budget: 16384, backpressure: 'unbounded' }); writer.writeSync('a'); // fill buffer // This write blocks on backpressure @@ -407,7 +407,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 +423,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 +439,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 +455,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..3ffd632622fb61 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(); @@ -138,7 +138,7 @@ async function testShareDropOldest() { yield [new TextEncoder().encode(`${i}`)]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'drop-oldest' }); + const shared = share(source(), { budget: 16384, backpressure: 'drop-oldest' }); const fast = shared.pull(); const slow = shared.pull(); @@ -169,7 +169,7 @@ async function testShareDropOldest() { 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() { @@ -177,7 +177,7 @@ async function testShareDropNewest() { yield [new TextEncoder().encode(`${i}`)]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'drop-newest' }); + const shared = share(source(), { budget: 16384, backpressure: 'drop-newest' }); const fast = shared.pull(); const slow = shared.pull(); @@ -212,13 +212,13 @@ async function testShareStrictBackpressure() { yield [new TextEncoder().encode(`${i}`)]; } } - const shared = share(source(), { highWaterMark: 2, backpressure: 'strict' }); + const shared = share(source(), { budget: 16384, 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..c8d63a584beb18 100644 --- a/test/parallel/test-stream-iter-validation.js +++ b/test/parallel/test-stream-iter-validation.js @@ -20,18 +20,18 @@ 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 >= 1 +assert.throws(() => push({ budget: 'bad' }), { code: 'ERR_INVALID_ARG_TYPE' }); +assert.throws(() => push({ budget: 1.5 }), { code: 'ERR_OUT_OF_RANGE' }); +// Values < 16384 are clamped to 16384 +assert.strictEqual(push({ budget: 0 }).writer.canWrite, true); +assert.strictEqual(push({ budget: -1 }).writer.canWrite, true); +assert.strictEqual(push({ budget: -100 }).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 +69,33 @@ 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) +// Values < 16384 are clamped to 16384 (both directions) { - const [a, b] = duplex({ highWaterMark: 0 }); - assert.strictEqual(a.writer.desiredSize, 1); - assert.strictEqual(b.writer.desiredSize, 1); + const [a, b] = duplex({ budget: 0 }); + assert.strictEqual(a.writer.canWrite, true); + assert.strictEqual(b.writer.canWrite, true); a.close(); b.close(); } // 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: 0 }, b: { budget: 16384 } }); + assert.strictEqual(a.writer.canWrite, true); // clamped + assert.strictEqual(b.writer.canWrite, true); a.close(); b.close(); } @@ -118,29 +118,29 @@ 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) +// Values < 16384 are clamped to 16384 (need a consumer for canWrite to work) { - const bc = broadcast({ highWaterMark: 0 }); + const bc = broadcast({ budget: 0 }); bc.broadcast.push(); - assert.strictEqual(bc.writer.desiredSize, 1); + assert.strictEqual(bc.writer.canWrite, true); bc.writer.endSync(); } { - const bc = broadcast({ highWaterMark: -1 }); + const bc = broadcast({ budget: -1 }); 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 +165,30 @@ 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 clamped (no canWrite, but accepts the value) +share(from('a'), { budget: 0 }).cancel(); +share(from('a'), { budget: -1 }).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 clamped (accepts the value) +shareSync(fromSync('a'), { budget: 0 }).cancel(); +shareSync(fromSync('a'), { budget: -1 }).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 +316,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(); From 939e46ef50beed023a57b7b821dfbcfc9ad29589 Mon Sep 17 00:00:00 2001 From: James M Snell Date: Sun, 12 Jul 2026 17:22:04 -0700 Subject: [PATCH 2/2] stream, quic: update iterable streams backpressure Signed-off-by: James M Snell Assisted-by: Opencode/Opus --- doc/api/quic.md | 13 +-- lib/internal/quic/quic.js | 50 ++++++------ lib/internal/quic/state.js | 18 ++--- src/quic/streams.cc | 16 ++-- .../test-quic-flow-control-params.mjs | 2 +- test/parallel/test-quic-flow-control-uni.mjs | 2 +- .../parallel/test-quic-stream-bidi-writer.mjs | 6 +- ...quic-stream-destroy-emits-stop-sending.mjs | 20 ++--- .../test-quic-stream-uni-server-initiated.mjs | 2 +- test/parallel/test-quic-stream-writer-api.mjs | 16 ++-- .../test-quic-stream-writer-dispose.mjs | 6 +- .../test-quic-writer-backpressure.mjs | 20 ++--- .../test-quic-writer-write-rejects.mjs | 12 +-- ...test-stream-iter-broadcast-backpressure.js | 44 ++++++----- .../test-stream-iter-broadcast-basic.js | 10 +-- .../test-stream-iter-broadcast-coverage.js | 4 +- .../test-stream-iter-pipeto-writev.js | 39 ++++----- .../test-stream-iter-push-backpressure.js | 79 +++++++++---------- test/parallel/test-stream-iter-push-basic.js | 11 ++- test/parallel/test-stream-iter-push-writer.js | 48 ++++++----- test/parallel/test-stream-iter-share-from.js | 52 ++++++------ test/parallel/test-stream-iter-validation.js | 59 +++++++------- 22 files changed, 272 insertions(+), 257 deletions(-) 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`