GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing github.com — stream: prevent share from eagerly draining source · nodejs/node@0afda47

File tree

Expand file treeCollapse file tree

Open diff view settings

Filter options

lib/internal/streams/iter
- share.js
test/parallel
- test-stream-iter-share-from.js

Expand file treeCollapse file tree

Open diff view settings

Collapse file

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

Copy file name to clipboardExpand all lines: lib/internal/streams/iter/share.js

+14Lines changed: 14 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -199,6 +199,9 @@ class ShareImpl {`
`199`	`199`	`}`
`200`	`200`
`201`	`201`	`await self.#pullFromSource(!shouldBuffer);`
	`202`	`+ if (!shouldBuffer) {`
	`203`	`+ await self.#waitForBufferSpaceAfterDrop();`
	`204`	`+ }`
`202`	`205`	`}`
`203`	`206`	`};`
`204`	`207`
`@@ -317,6 +320,17 @@ class ShareImpl {`
`317`	`320`	`return true;`
`318`	`321`	`}`
`319`	`322`
	`323`	`+ async #waitForBufferSpaceAfterDrop() {`
	`324`	`+ while (this.#bufferedBytes >= this.#options.budget &&`
	`325`	`+ !this.#cancelled &&`
	`326`	`+ this.#sourceError === undefined &&`
	`327`	`+ !this.#sourceExhausted) {`
	`328`	`+ const { promise, resolve } = PromiseWithResolvers();`
	`329`	`+ ArrayPrototypePush(this.#pullWaiters, resolve);`
	`330`	`+ await promise;`
	`331`	`+ }`
	`332`	`+ }`
	`333`	`+`
`320`	`334`	`#pullFromSource(discard = false) {`
`321`	`335`	`if (this.#sourceExhausted \|\| this.#cancelled) {`
`322`	`336`	`return PromiseResolve();`

Collapse file

`‎test/parallel/test-stream-iter-share-from.js‎`

Copy file name to clipboardExpand all lines: test/parallel/test-stream-iter-share-from.js

+36-26Lines changed: 36 additions & 26 deletions

Original file line number	Diff line number	Diff line change
`@@ -170,40 +170,50 @@ async function testShareDropOldest() {`
`170`	`170`	`}`
`171`	`171`
`172`	`172`	`async function testShareDropNewest() {`
`173`		`- // With drop-newest and a stalled consumer, the async path allows the`
`174`		`- // buffer to grow beyond budget (the "drop" applies to the`
`175`		`- // backpressure signal, not the buffer contents). Both consumers`
`176`		`- // ultimately see all items.`
	`173`	`+ let pulls = 0;`
	`174`	`+ let secondPull;`
	`175`	`+ const secondPullStarted = new Promise((resolve) => {`
	`176`	`+ secondPull = resolve;`
	`177`	`+ });`
	`178`	`+`
`177`	`179`	`async function* source() {`
`178`		`- for (let i = 0; i < 4; i++) {`
	`180`	`+ for (let i = 0; i < 7; i++) {`
	`181`	`+ pulls++;`
	`182`	`+ if (pulls === 2) secondPull();`
`179`	`183`	`const chunk = new Uint8Array(16384);`
`180`	`184`	`chunk[0] = i;`
`181`	`185`	`yield [chunk];`
`182`	`186`	`}`
`183`	`187`	`}`
`184`		`- const shared = share(source(), { budget: 32768, backpressure: 'drop-newest' });`
`185`		`- const fast = shared.pull();`
`186`		`- const slow = shared.pull();`
	`188`	`+ const shared = share(source(), {`
	`189`	`+ budget: 16384,`
	`190`	`+ backpressure: 'drop-newest',`
	`191`	`+ });`
	`192`	`+ const fast = shared.pull()[Symbol.asyncIterator]();`
	`193`	`+ const slow = shared.pull()[Symbol.asyncIterator]();`
`187`	`194`
`188`		`- // Fast consumer reads all items`
`189`		`- const fastIndices = [];`
`190`		`- for await (const batch of fast) {`
`191`		`- for (const chunk of batch) {`
`192`		`- fastIndices.push(chunk[0]);`
`193`		`- }`
`194`		`- }`
`195`		`- assert.strictEqual(fastIndices.length, 2);`
	`195`	`+ const first = await fast.next();`
	`196`	`+ assert.strictEqual(first.value[0][0], 0);`
`196`	`197`
`197`		`- // Slow consumer also sees all items (buffer grew past budget)`
`198`		`- const slowIndices = [];`
`199`		`- for await (const batch of slow) {`
`200`		`- for (const chunk of batch) {`
`201`		`- slowIndices.push(chunk[0]);`
`202`		`- }`
`203`		`- }`
`204`		`- assert.strictEqual(slowIndices.length, 2);`
`205`		`- assert.strictEqual(slowIndices[0], 0);`
`206`		`- assert.strictEqual(slowIndices[1], 1);`
	`198`	`+ let nextSettled = false;`
	`199`	`+ const next = fast.next().then((result) => {`
	`200`	`+ nextSettled = true;`
	`201`	`+ return result;`
	`202`	`+ });`
	`203`	`+`
	`204`	`+ await secondPullStarted;`
	`205`	`+ await new Promise(setImmediate);`
	`206`	`+ assert.strictEqual(pulls, 2);`
	`207`	`+ assert.strictEqual(nextSettled, false);`
	`208`	`+`
	`209`	`+ const slowResult = await slow.next();`
	`210`	`+ assert.strictEqual(slowResult.value[0][0], 0);`
	`211`	`+`
	`212`	`+ const nextResult = await next;`
	`213`	`+ assert.strictEqual(nextResult.value[0][0], 2);`
	`214`	`+ assert.strictEqual(pulls, 3);`
	`215`	`+`
	`216`	`+ shared.cancel();`
`207`	`217`	`}`
`208`	`218`
`209`	`219`	`// =============================================================================`

0 commit comments

Comments

0 (0)

You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.