stream: update iterable streams to use budget backpressure#64464
stream: update iterable streams to use budget backpressure#64464jasnell wants to merge 2 commits into
Conversation
Updates the implementation to use the updated backpressure model that landed here: WinterTC55/iter-streams#23 Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode/Opus
Signed-off-by: James M Snell <jasnell@gmail.com> Assisted-by: Opencode/Opus
jasnell
added
stream
|
Review requested:
|
nodejs-github-bot
added
lib / src
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #64464 +/- ##
==========================================
+ Coverage 90.23% 90.24% +0.01%
==========================================
Files 741 741
Lines 241604 241652 +48
Branches 45520 45528 +8
==========================================
+ Hits 218010 218079 +69
+ Misses 15120 15101 -19
+ Partials 8474 8472 -2
🚀 New features to boost your workflow:
|
| | `'drop-newest'` | `highWaterMark` | N/A (never waits) | | ||
| | Policy | Buffer limit | Pending writes limit | | ||
| | --------------- | ------------ | -------------------- | | ||
| | `'strict'` | `budget` | `budget` | |
There was a problem hiding this comment.
I think the pending writes limit is now 1 - docs need updating here
| this.#bufferedBytes += batchSize; | ||
| this.#bytesWritten += batchSize; | ||
|
|
||
| this.#resolvePendingReads(); |
There was a problem hiding this comment.
I think this needs to potentially resolve pending drains:
| this.#resolvePendingReads(); | |
| this.#resolvePendingReads(); | |
| if (this.#bufferedBytes < this.#budget) { | |
| this.#resolvePendingDrains(true); | |
| } |
Since we're dropping slot-wise, with drop-oldest, a write can bring us back under the limit (e.g. a 1KB write that makes us drop an 100KB chunk). If we don't resolve drains here, backpressure deadlocks.
| if (this.#consumerState !== 'active') return false; | ||
|
|
||
| if (this.#slots.length >= this.#highWaterMark) { | ||
| const batchSize = this.#batchByteSize(chunks); |
There was a problem hiding this comment.
We currently don't skip empty chunks here. Should we? Changes behaviour a bit, but otherwise bytewise budgets mean you can push infinite empty chunks, and you get weird behaviour like zero-byte writes waiting on backpressure.
Would be easy to check and drop here if we think that's the right thing to do.
| const opts = { | ||
| __proto__: null, | ||
| highWaterMark: clampHWM(highWaterMark), | ||
| budget: clampBudget(budget), |
There was a problem hiding this comment.
Everywhere we run clampBudget we run validateInteger(budget, 'options.budget', 16384) directly beforehand and AFAICT they do exactly the same thing.
| return null; | ||
| } | ||
| return MathMax(0, this.#highWaterMark - this.#slots.length); | ||
| return this.#bufferedBytes < this.#budget; |
There was a problem hiding this comment.
This is false when full on drop-*, while BroadcastWriter streams canWrite returns true in that case (kCanWrite logic differs according to backpressure policy). They should match I think.
| function clampHWM(value) { | ||
| return MathMax(1, MathMin(NumberMAX_SAFE_INTEGER, value)); | ||
| function clampBudget(value) { | ||
| return MathMax(16384, MathMin(NumberMAX_SAFE_INTEGER, value)); |
There was a problem hiding this comment.
Why (here or validate) the hard limit of 16KB as a lower bound? At a glance seems arbitrary - I imagine smaller is less efficient, but if people want to do weird things/have weird constraints do we need to block them? I don't feel strongly but I'd be interested if there's a specific reason.
There was a problem hiding this comment.
The actual limit we end up with (here and in the spec) needs to be fully benchmarked and validated. Went with something that I knew would prompt the question ;-)
4 participants
WinterTC55/iter-streams#23 updated the backpressure model in the iterable streams specification to use a byte-budget model. This PR updates the implementation to match.
Signed-off-by: James M Snell jasnell@gmail.com
Assisted-by: Opencode/Opus