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

buffer: fix Blob.stream() leaking source buffer · nodejs/node@ddacb3f · GitHub

/ node Public

Commit ddacb3f

Browse files
semimikoh
authored and
aduh95
committed
buffer: fix Blob.stream() leaking source buffer
Blob.prototype.stream() registered a wakeup callback on the underlying source's start() and never released it. The strong Reader::wakeup_ handle kept the reader -- and through it the blob's DataQueue and backing store -- reachable as a GC root, so the source buffer leaked on every stream() call. On Node 26+, streaming a 1 MiB blob 300 times retained ~300 MiB in process.memoryUsage().arrayBuffers while the V8 heap stayed small. Register the wakeup lazily in pull() and clear it on every terminal or idle path (EOS, error, cancel, backpressure), mirroring the cleanup already done by the async iterator path. The strong handle now only lives while a pull is in flight, so the reader and its backing store become collectable once the stream finishes, errors, is cancelled, or goes idle under backpressure. Fixes: #63574 Signed-off-by: semimikoh <ejffjeosms@gmail.com> PR-URL: #63577 Fixes: #63574 Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 271c745 commit ddacb3f

2 files changed

Lines changed: 73 additions & 4 deletions

File tree

‎lib/internal/blob.js‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -467,30 +467,40 @@ function createBlobReaderStream(reader) {
467467
// There really should only be one read at a time so using an
468468
// array here is purely defensive.
469469
this.pendingPulls = [];
470-
// Register a wakeup callback that the C++ side can invoke
470+
// Lazily register a wakeup callback that the C++ side can invoke
471471
// when new data is available after a STATUS_BLOCK.
472-
reader.setWakeup(() => {
472+
this.wakeup = () => {
473473
if (this.pendingPulls.length > 0) {
474474
this.readNext(c);
475475
}
476-
});
476+
};
477477
},
478478
pull(c) {
479479
const { promise, resolve, reject } = PromiseWithResolvers();
480+
if (this.pendingPulls.length === 0) {
481+
reader.setWakeup(this.wakeup);
482+
}
480483
this.pendingPulls.push({ resolve, reject });
481484
this.readNext(c);
482485
return promise;
483486
},
487+
clearWakeupIfIdle() {
488+
if (this.pendingPulls.length === 0) {
489+
reader.setWakeup(undefined);
490+
}
491+
},
484492
readNext(c) {
485493
reader.pull((status, buffer) => {
486494
// If pendingPulls is empty here, the stream had to have
487495
// been canceled, and we don't really care about the result.
488496
// We can simply exit.
489497
if (this.pendingPulls.length === 0) {
498+
reader.setWakeup(undefined);
490499
return;
491500
}
492501
if (status === 0) {
493502
// EOS
503+
reader.setWakeup(undefined);
494504
c.close();
495505
// This is to signal the end for byob readers
496506
// see https://streams.spec.whatwg.org/#example-rbs-pull
@@ -502,6 +512,7 @@ function createBlobReaderStream(reader) {
502512
// The read could fail for many different reasons when reading
503513
// from a non-memory resident blob part (e.g. file-backed blob).
504514
// The error details the system error code.
515+
reader.setWakeup(undefined);
505516
const error =
506517
lazyDOMException('The blob could not be read',
507518
'NotReadableError');
@@ -511,7 +522,7 @@ function createBlobReaderStream(reader) {
511522
return;
512523
} else if (status === 2) {
513524
// STATUS_BLOCK: No data available yet. The wakeup callback
514-
// registered in start() will re-invoke readNext when data
525+
// registered in pull() will re-invoke readNext when data
515526
// arrives.
516527
return;
517528
}
@@ -531,6 +542,7 @@ function createBlobReaderStream(reader) {
531542
if (this.pendingPulls.length !== 0) {
532543
const pending = this.pendingPulls.shift();
533544
pending.resolve();
545+
this.clearWakeupIfIdle();
534546
}
535547
return;
536548
}
@@ -539,6 +551,7 @@ function createBlobReaderStream(reader) {
539551
});
540552
},
541553
cancel(reason) {
554+
reader.setWakeup(undefined);
542555
// Reject any currently pending pulls here.
543556
for (const pending of this.pendingPulls) {
544557
pending.reject(reason);

‎test/parallel/test-blob-stream-gc.js‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
// Flags: --expose-gc --no-concurrent-array-buffer-sweeping
2+
'use strict';
3+
4+
const common = require('../common');
5+
const assert = require('assert');
6+
const { setImmediate: setImmediatePromise } = require('timers/promises');
7+
8+
const MiB = 1024 * 1024;
9+
const iterations = 64;
10+
const maxRetained = 16 * MiB;
11+
12+
async function collectArrayBuffers() {
13+
for (let i = 0; i < 3; i++) {
14+
global.gc();
15+
await setImmediatePromise();
16+
}
17+
}
18+
19+
async function assertNoBlobStreamRetention(name, fn) {
20+
const buffer = Buffer.alloc(MiB);
21+
22+
await collectArrayBuffers();
23+
const before = process.memoryUsage().arrayBuffers;
24+
25+
for (let i = 0; i < iterations; i++) {
26+
await fn(buffer);
27+
}
28+
29+
await collectArrayBuffers();
30+
const retained = process.memoryUsage().arrayBuffers - before;
31+
32+
assert(
33+
retained < maxRetained,
34+
`${name} retained ${retained} bytes in arrayBuffers`,
35+
);
36+
}
37+
38+
(async () => {
39+
await assertNoBlobStreamRetention('unused Blob streams',
40+
common.mustCall(async (buffer) => {
41+
new Blob([buffer]).stream();
42+
}, iterations));
43+
44+
await assertNoBlobStreamRetention('cancelled Blob streams',
45+
common.mustCall(async (buffer) => {
46+
await new Blob([buffer]).stream()
47+
.cancel();
48+
}, iterations));
49+
50+
await assertNoBlobStreamRetention('drained Blob streams',
51+
common.mustCall(async (buffer) => {
52+
await new Response(
53+
new Blob([buffer]).stream(),
54+
).arrayBuffer();
55+
}, iterations));
56+
})().then(common.mustCall());

0 commit comments

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