From f8a99eec0a7d65b741b889dda860b4e2af159133 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 1 Nov 2026 03:48:28 +0000 Subject: [PATCH 1/2] Free a thread-pool job's own copy of a large input when the pool thread is done with it A job that has its own copy of an input kept that copy until its completion ran on the JS thread. A caller that starts many calls before it returns to the event loop held every copy. These jobs now drop a large copy at the end of the pool body: node:fs writeFile, appendFile, write and writev, Bun.Archive.write, Bun.zstdCompress and Bun.zstdDecompress. The copy is the UTF-8 form of a string that is not ASCII or that is an atom, the bytes of a string with an encoding, the bytes that Bun.Archive.write copies or builds, or the copy of a view of a resizable ArrayBuffer or of a WebAssembly.Memory. A smaller copy, a shared string and a pin stay for the JS thread. bun_core::vec::releases_on_pool has the size rule: a copy of more than 512 KiB. Each helper returns a Vec, so a pool thread can drop only plain bytes. --- .../snippets/write-file-copied-data.node.mjs | 89 ++++++++++++++++++ docs/project/bindgen.mdx | 5 +- src/bun_core/lib.rs | 19 ++++ src/jsc/array_buffer.rs | 19 ++++ src/jsc/bindgen.rs | 21 ++++- src/jsc/thread_isolated.rs | 1 + src/runtime/api/Archive.rs | 3 + src/runtime/api/BunObject.rs | 1 + src/runtime/node/node_fs.rs | 26 +++++- src/runtime/node/types.rs | 22 +++++ test/harness.ts | 93 +++++++++++++++++++ test/js/bun/archive.test.ts | 42 ++++++++- test/js/bun/util/zstd.test.ts | 28 +++++- test/js/node/fs/promises.test.js | 88 +++++++++++++++++- 14 files changed, 446 insertions(+), 11 deletions(-) create mode 100644 bench/snippets/write-file-copied-data.node.mjs diff --git a/bench/snippets/write-file-copied-data.node.mjs b/bench/snippets/write-file-copied-data.node.mjs new file mode 100644 index 000000000000..45ec65a88779 --- /dev/null +++ b/bench/snippets/write-file-copied-data.node.mjs @@ -0,0 +1,89 @@ +// @runtime node, bun +// fs.promises.writeFile of data that the write job cannot read in place. bun reads an ASCII +// string or a buffer of the caller in place. It has its own copy of the other sources here. +// The rows of mitata write one file at a time. The bursts start every write before they await +// one. The chains await each write before the next one. Each burst and each chain runs in a new +// process, because a process that has run one has its memory to use again. +import { Buffer } from "node:buffer"; +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync } from "node:fs"; +import { writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import process from "node:process"; +import { fileURLToPath } from "node:url"; + +const KB = 1024; +const MB = 1024 * KB; +// Each gives a source that is `size` bytes in the file. +const sources = { + "ASCII": size => Buffer.alloc(size, "a").toString("latin1"), + "Latin-1": size => Buffer.alloc(size / 2, 0xe9).toString("latin1"), + "UTF-16": size => Buffer.alloc((size / 3) * 2, "\u4e2d", "utf16le").toString("utf16le"), + "base64": size => Buffer.alloc(size, "a").toString("base64"), + "resizable view": size => new Uint8Array(new ArrayBuffer(size, { maxByteLength: 2 * size })).fill(0x61), +}; +const encoding = kind => (kind === "base64" ? "base64" : "utf8"); +const label = size => (size >= MB ? size / MB + " MB" : size / KB + " KB"); + +const dir = mkdtempSync(join(tmpdir(), "bun-bench-write-file-copied-data-")); +try { + const mode = process.argv[2]; + if (mode === "burst" || mode === "chain") { + const [kind, count, size] = [process.argv[3], Number(process.argv[4]), Number(process.argv[5])]; + const source = sources[kind](size); + const paths = Array.from({ length: count }, (_, i) => join(dir, "out" + i)); + const faults = process.resourceUsage().minorPageFault; + const start = performance.now(); + if (mode === "burst") await Promise.all(paths.map(path => writeFile(path, source, encoding(kind)))); + else for (let i = 0; i < count; i++) await writeFile(paths[0], source, encoding(kind)); + const ms = performance.now() - start; + const usage = process.resourceUsage(); + console.log( + `${count} x ${label(size)} ${kind} ${mode === "burst" ? "at once" : "in a chain"}`.padEnd(38) + + `${ms.toFixed(0)} ms`.padStart(9) + + `${usage.minorPageFault - faults} minor faults`.padStart(22) + + `${(usage.maxRSS / KB).toFixed(0)} MB peak RSS`.padStart(18), + ); + } else { + // First, while this process is small: a child starts with the peak RSS of its parent. + if (!process.env.BENCHMARK_RUNNER) { + for (const [mode, kind, count, size] of [ + ["burst", "ASCII", 300, MB], + ["burst", "UTF-16", 1500, 300 * KB], + ...["Latin-1", "UTF-16", "base64", "resizable view"].map(kind => ["burst", kind, 300, MB]), + ["chain", "UTF-16", 16, 32 * MB], + ["chain", "resizable view", 16, 32 * MB], + ]) { + spawnSync(process.execPath, [fileURLToPath(import.meta.url), mode, kind, count, size].map(String), { + stdio: "inherit", + }); + } + } + + const { bench, run } = await import("../runner.mjs"); + const path = join(dir, "out"); + for (const [kind, size] of [ + ["ASCII", 12], + ["UTF-16", 12], + ...Object.keys(sources).flatMap(kind => [16 * KB, 300 * KB, MB, 16 * MB].map(size => [kind, size])), + ]) { + const source = size === 12 ? (kind === "ASCII" ? "short string" : "short \u4e2d\u6587") : sources[kind](size); + bench(`${size === 12 ? "12 B" : label(size)} ${kind}`, async () => { + await writeFile(path, source, encoding(kind)); + }); + } + const chain = sources["UTF-16"](MB); + const paths = Array.from({ length: 16 }, (_, i) => join(dir, "chain" + i)); + bench("16 chains of 8 x 1 MB UTF-16", async () => { + await Promise.all( + paths.map(async path => { + for (let i = 0; i < 8; i++) await writeFile(path, chain); + }), + ); + }); + await run(); + } +} finally { + rmSync(dir, { recursive: true, force: true }); +} diff --git a/docs/project/bindgen.mdx b/docs/project/bindgen.mdx index fea0ab724ca9..e435a8e70b1f 100644 --- a/docs/project/bindgen.mdx +++ b/docs/project/bindgen.mdx @@ -552,7 +552,10 @@ from a buffer. **Copied** bytes are the body's. **Pinned** bytes are for a job on another thread to read after the call returns: a string is thread-isolated, a buffer is a GC-rooted `PinnedArrayBuffer`, and a resizable one is copied. A `PinnedBytes` is `Send`, derefs to `[u8]`, and is dropped on the JS -thread. +thread. A job that is done with the bytes calls `take_owned_for_pool_release()` and drops what it +gets: its own copy of the bytes, if it has one that `bun_core::vec::releases_on_pool` accepts. That +is the UTF-8 form of a string that is not ASCII or that is an atom, or the copy of a resizable +buffer. The copy then does not wait for the JS thread. **A shared buffer** is borrowed like any other, as JSC's own `toBase64` and `indexOf` read one, and another thread that writes it during the call changes what the body reads. `shared: "copied"` is diff --git a/src/bun_core/lib.rs b/src/bun_core/lib.rs index 2ce018548143..0c86dc91afdc 100644 --- a/src/bun_core/lib.rs +++ b/src/bun_core/lib.rs @@ -449,6 +449,25 @@ pub mod vec { v.truncate(len - n); } + /// mimalloc's `MI_LARGE_MAX_OBJ_SIZE`: a larger block has a page of its own, which every thread frees the same way. + const POOL_RELEASE_MIN_BYTES: usize = 512 * 1024; + + /// Whether a work-pool job frees its own copy of an input on the pool thread after its last read. Otherwise the JS thread frees it at completion. + #[inline] + pub fn releases_on_pool(capacity: usize) -> bool { + capacity > POOL_RELEASE_MIN_BYTES + } + + /// The bytes for the job to drop on the pool thread, if [`releases_on_pool`] says so. + #[inline] + pub fn take_for_pool_release(bytes: &mut Vec) -> Vec { + if releases_on_pool(bytes.capacity()) { + core::mem::take(bytes) + } else { + Vec::new() + } + } + // ── Spare-capacity fill helpers ──────────────────────────────────────── // Consolidates 11 hand-rolled `spare_capacity_mut().as_mut_ptr().cast::()` // + `set_len(len+n)` copies: `spare_capacity_mut` returns diff --git a/src/jsc/array_buffer.rs b/src/jsc/array_buffer.rs index 181a0dc7ce30..81fc5014f909 100644 --- a/src/jsc/array_buffer.rs +++ b/src/jsc/array_buffer.rs @@ -795,6 +795,25 @@ impl PinnedArrayBuffer { true } + /// The read-only copy, if [`bun_core::vec::releases_on_pool`] says so. The view is then empty. + #[inline] + pub fn take_copy_for_pool_release(&mut self) -> Vec { + let Some(copy) = self.copy.as_deref_mut() else { + return Vec::new(); + }; + debug_assert!( + copy.write_back_to.is_none(), + "the JS thread writes a read scratch back" + ); + if !bun_core::vec::releases_on_pool(copy.bytes.capacity()) { + return Vec::new(); + } + self.buffer.ptr = ptr::null_mut(); + self.buffer.len = 0; + self.buffer.byte_len = 0; + core::mem::take(&mut copy.bytes) + } + /// Swaps in `len` bytes of scratch space if `grow()` could free the block while a job writes to /// it. If this returns true, the job writes from index 0 and [`write_back`](Self::write_back) /// copies the bytes to `offset`, even if `grow()` detached the view. diff --git a/src/jsc/bindgen.rs b/src/jsc/bindgen.rs index 38b2251a445d..493b1fdaebca 100644 --- a/src/jsc/bindgen.rs +++ b/src/jsc/bindgen.rs @@ -880,6 +880,8 @@ pub struct PinnedBytes(ThreadIsolated); enum Pinned { /// Thread-isolated. String(bun_core::Utf8WithString), + /// The job's own UTF-8 bytes of a string that it cannot share. + Utf8(bun_core::Utf8Bytes<'static>), /// GC-rooted too. A resizable one is copied. Buffer(PinnedArrayBuffer), } @@ -906,15 +908,29 @@ impl PinnedBytes { } StringOrJsBytes::String(string) => { let utf8 = string.into_utf8_with_string_thread_isolated(); - if !utf8.is_shared() { + if utf8.is_shared() { + Pinned::String(utf8) + } else { global.vm().report_extra_memory(utf8.slice().len()); + Pinned::Utf8(utf8.into_utf8()) } - Pinned::String(utf8) } }; // SAFETY: the string is thread-isolated, and the buffer pinned and GC-rooted. Ok(Self(unsafe { ThreadIsolated::new(pinned) })) } + + /// The job's own copy of the bytes, if [`bun_core::vec::take_for_pool_release`] moves it out. The bytes are then empty. + #[inline] + pub fn take_owned_for_pool_release(&mut self) -> Vec { + match &mut *self.0 { + Pinned::Utf8(bun_core::Utf8Bytes::Owned(bytes)) => { + bun_core::vec::take_for_pool_release(bytes) + } + Pinned::Buffer(buffer) => buffer.take_copy_for_pool_release(), + Pinned::String(_) | Pinned::Utf8(_) => Vec::new(), + } + } } impl core::ops::Deref for PinnedBytes { @@ -923,6 +939,7 @@ impl core::ops::Deref for PinnedBytes { fn deref(&self) -> &[u8] { match &*self.0 { Pinned::String(utf8) => utf8.slice(), + Pinned::Utf8(utf8) => utf8.slice(), Pinned::Buffer(buffer) => buffer.slice(), } } diff --git a/src/jsc/thread_isolated.rs b/src/jsc/thread_isolated.rs index 3b17de1ee998..37ae35b7f02a 100644 --- a/src/jsc/thread_isolated.rs +++ b/src/jsc/thread_isolated.rs @@ -1,5 +1,6 @@ /// # Safety /// Its JS buffers are pinned and GC-rooted and its strings thread-isolated: a work-pool job may read it and the JS thread drops it. +/// The job may move bytes that the value alone owns out as a `Vec` and drop that on its own thread. pub unsafe trait ThreadIsolatedArg {} /// A `T` built as [`ThreadIsolatedArg`] describes. diff --git a/src/runtime/api/Archive.rs b/src/runtime/api/Archive.rs index 03c753db79b3..6c72081a861d 100644 --- a/src/runtime/api/Archive.rs +++ b/src/runtime/api/Archive.rs @@ -808,6 +808,9 @@ pub(crate) struct WriteContext { impl TaskContext for WriteContext { fn run(&mut self) { self.result = self.do_run(); + if let WriteData::Owned(data) = &mut self.data { + drop(bun_core::vec::take_for_pool_release(data)); + } } fn run_from_js(&mut self, global: &JSGlobalObject) -> JsResult { diff --git a/src/runtime/api/BunObject.rs b/src/runtime/api/BunObject.rs index a35467fa352d..afd63528133e 100644 --- a/src/runtime/api/BunObject.rs +++ b/src/runtime/api/BunObject.rs @@ -2033,6 +2033,7 @@ pub(crate) mod JSZstd { } else { decompress_to_box(input) }; + drop(this.buffer.take_owned_for_pool_release()); Some(done) } diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index cf6262e3a9ab..94b83ea71bca 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -1389,6 +1389,9 @@ mod _async_tasks { } /// Runs on the JS thread before the result reaches JS. See [`PinnedArrayBuffer::write_back`]. fn write_back(&mut self, _bytes_read: u64) {} + /// Runs on the pool thread after the operation. See [`StringOrBuffer::take_owned_for_pool_release`]. + #[inline] + fn release_data_on_pool(&mut self) {} fn paths(&self) -> [Option<&PathLike<'static>>; 2] { [None, None] } @@ -1404,7 +1407,7 @@ mod _async_tasks { /// Forward [`FsArgument`] to the inherent `from_js` each `args::*` struct /// already defines. macro_rules! impl_fs_argument { - ( $( $ty:ty $( => |$args:ident| [$first:expr, $second:expr] )? ),+ $(,)? ) => { + ( $( $ty:ty $( => |$args:ident| [$first:expr, $second:expr] )? $( ; release = |$this:ident| $release:expr )? ),+ $(,)? ) => { $( // SAFETY: `from_js_async` parses paths and data thread-isolated / pinned // and rooted; the remaining fields are plain data. @@ -1412,6 +1415,7 @@ mod _async_tasks { impl FsArgument for $ty { #[inline] fn from_js(ctx: &JSGlobalObject, arguments: &mut ArgumentsSlice) -> JsResult { <$ty>::from_js(ctx, arguments) } $( fn paths(&self) -> [Option<&PathLike<'static>>; 2] { let $args = self; [$first, $second] } )? + $( #[inline] fn release_data_on_pool(&mut self) { let $this = self; $release } )? } )+ }; } @@ -1440,9 +1444,9 @@ mod _async_tasks { args::Cp<'static> => |args| [Some(&args.src), Some(&args.dest)], ); impl_fs_argument!( - args::Writev, + args::Writev; release = |args| args.buffers.release_copies_on_pool(), args::FTruncate, - args::Write<'static>, + args::Write<'static>; release = |args| drop(args.buffer.take_owned_for_pool_release()), args::Fchown, args::FChmod, args::Fstat, @@ -1510,6 +1514,10 @@ mod _async_tasks { fn signal(&self) -> Option<&AbortSignal> { self.signal.as_deref() } + #[inline] + fn release_data_on_pool(&mut self) { + drop(self.data.take_owned_for_pool_release()); + } fn paths(&self) -> [Option<&PathLike<'static>>; 2] { [path_of(&self.file), None] } @@ -1525,6 +1533,10 @@ mod _async_tasks { fn signal(&self) -> Option<&AbortSignal> { self.0.signal.as_deref() } + #[inline] + fn release_data_on_pool(&mut self) { + drop(self.0.data.take_owned_for_pool_release()); + } fn paths(&self) -> [Option<&PathLike<'static>>; 2] { [path_of(&self.0.file), None] } @@ -1805,8 +1817,12 @@ mod _async_tasks { this: &mut Self, done: bun_jsc::Completion, ) -> Option> { - let mut node_fs = NodeFS::default(); - this.result = NodeFS::dispatch::(&mut node_fs, &this.args, Flavor::Async); + this.result = { + let mut node_fs = NodeFS::default(); + NodeFS::dispatch::(&mut node_fs, &this.args, Flavor::Async) + }; + // After `node_fs` is gone: what a pool thread allocates after the release can take the freed range. + this.args.release_data_on_pool(); Some(done) } diff --git a/src/runtime/node/types.rs b/src/runtime/node/types.rs index 88f2a3f7dc6e..08fcffe82d43 100644 --- a/src/runtime/node/types.rs +++ b/src/runtime/node/types.rs @@ -289,6 +289,16 @@ impl StringOrBuffer<'static> { unsafe { ThreadIsolated::new(Self::owned(bytes)) } } + /// The job's own copy of the bytes, if [`bun_core::vec::take_for_pool_release`] moves it out. The bytes are then empty. + #[inline] + pub(crate) fn take_owned_for_pool_release(&mut self) -> Vec { + match self { + Self::Utf8(Utf8Bytes::Owned(bytes)) => bun_core::vec::take_for_pool_release(bytes), + Self::PinnedBuffer(buffer) => buffer.take_copy_for_pool_release(), + _ => Vec::new(), + } + } + /// `value` is ArrayBuffer-like (the caller checked): borrowed for `Sync`, /// [`PinnedArrayBuffer::root_read_only`] for `Async`. pub(crate) fn buffer_from_js( @@ -1509,6 +1519,18 @@ impl VectorArrayBuffer { } } + /// Runs on the pool thread after a write. See [`PinnedArrayBuffer::take_copy_for_pool_release`]. + pub(crate) fn release_copies_on_pool(&mut self) { + for element in &mut self.wasm { + let Some(pinned) = &mut element.native else { + continue; + }; + if pinned.take_copy_for_pool_release().capacity() != 0 { + self.buffers[element.index] = bun_sys::platform_iovec_create(&mut []); + } + } + } + /// Runs on the JS thread after a read of `bytes_read` bytes, which filled the spans in order. /// See [`PinnedArrayBuffer::write_back`]. pub(crate) fn write_back(&mut self, bytes_read: usize) { diff --git a/test/harness.ts b/test/harness.ts index 999d51e22906..35e93c2cd7bc 100644 --- a/test/harness.ts +++ b/test/harness.ts @@ -625,6 +625,99 @@ export function emptyProcessMaxRSS() { }).then(({ maxRSS }) => maxRSS)); } +/** + * Source for a fixture that keeps its JS thread out of the event loop while + * other threads work: + * - `waitFor(what, done)` polls `done()` and throws when it gives up. + * - `byteAt(path, position)` is that byte of the file, or -1 if the file does + * not have it yet. + * - `noJobRuns()` is true when every other thread of the process sleeps. + * Linux only: the thread states come from /proc. + */ +export const waitForOtherThreadsSource = /* js */ ` + // Each poll sleeps 1 ms, so the limit is 10 seconds or more. + const waitFor = (what, done) => { + for (let polls = 0; !done(); polls++) { + if (polls === 10000) throw new Error("gave up waiting for " + what); + Bun.sleepSync(1); + } + }; + const byteAt = (path, position) => { + const fs = require("fs"); + let fd; + try { + fd = fs.openSync(path, "r"); + const byte = Buffer.alloc(1); + return fs.readSync(fd, byte, 0, 1, position) === 1 ? byte[0] : -1; + } catch { + return -1; + } finally { + if (fd !== undefined) fs.closeSync(fd); + } + }; + const noJobRuns = () => { + const fs = require("fs"); + const self = fs.readlinkSync("/proc/thread-self").split("/").pop(); + return fs.readdirSync("/proc/self/task").every(tid => { + if (tid === self) return true; + try { + // The state follows the thread name, which is in parentheses. + const stat = fs.readFileSync("/proc/self/task/" + tid + "/stat", "utf8"); + return stat[stat.lastIndexOf(")") + 2] === "S"; + } catch { + // The thread is gone. + return true; + } + }); + }; +`; + +/** + * Asserts on the RSS that a bun process gains while `count` calls are pending. + * + * The process runs `setup` and awaits one call. Then it evaluates `call` + * `count` times. `call` is an expression that gives a promise, and `i` is the + * index of the call (-1 for the first one). The JS thread stays in that loop, + * so it completes no call, and it starts a call only when the thread pool is + * done with the call before it. What a call holds from that point until the JS + * thread completes it adds up in the RSS, which the process samples before + * and after the loop. At the end the process awaits the calls and evaluates + * `verify`, which must give true. `results` has what the calls resolved to. + * + * `done` is an expression that is true when the thread pool is done with call + * `i`, for example `byteAt(path, last) === 0x61`. The loop evaluates it many + * times, so it must change nothing. Without it the loop waits for + * `noJobRuns()`, which is Linux only. + */ +export async function expectRssDeltaOfPendingCallsBelow( + options: { setup?: string; call: string; done?: string; verify: string; count: number }, + bounds: { release: number; debug: number }, +) { + const done = options.done ?? (isLinux ? "noJobRuns()" : undefined); + if (!done) throw new Error("expectRssDeltaOfPendingCallsBelow needs `done` on this platform"); + const fixture = /* js */ ` + const fs = require("fs"); + ${waitForOtherThreadsSource} + ${options.setup ?? ""} + // The pool threads start with this call, before the first sample. + { + const i = -1; + await (${options.call}); + } + const pending = []; + const before = process.memoryUsage.rss(); + for (let i = 0; i < ${options.count}; i++) { + pending.push(${options.call}); + waitFor("call " + i, () => ${done}); + } + const deltaMiB = (process.memoryUsage.rss() - before) / 1024 / 1024; + const results = await Promise.all(pending); + if (!(${options.verify})) throw new Error("the calls did not give the expected result"); + console.log(JSON.stringify({ deltaMiB })); + `; + await expectRssDeltaBelow(["-e", fixture], bounds); +} + // we must ensure that finalizers are run // so that the reference-counting logic is exercised export function gcTick(traceForDebugging = false) { diff --git a/test/js/bun/archive.test.ts b/test/js/bun/archive.test.ts index 03eebef5225a..88074706d7b1 100644 --- a/test/js/bun/archive.test.ts +++ b/test/js/bun/archive.test.ts @@ -1,5 +1,14 @@ import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, isLinux, isMacOS, isRoot, isWindows, tempDir } from "harness"; +import { + bunEnv, + bunExe, + expectRssDeltaOfPendingCallsBelow, + isLinux, + isMacOS, + isRoot, + isWindows, + tempDir, +} from "harness"; import { randomBytes } from "node:crypto"; import { chmodSync, @@ -1814,6 +1823,37 @@ describe("Bun.Archive", () => { ); }).toThrow(); }); + + // The write job has its own copy of a typed array, and of the archive it makes from an + // object. The pool thread frees a large copy when it is done with the write. If the copy + // stayed until the JS thread completes the write, a caller that starts many writes before + // it returns to the event loop would hold every copy. + test.concurrent.each([ + ["a typed array", `Bun.Archive.write(out, bytes)`, 0, `fs.readFileSync(out).equals(bytes)`], + [ + "an object of files", + `Bun.Archive.write(out, { "a.bin": bytes })`, + 512, + `Buffer.from(await (await new Bun.Archive(fs.readFileSync(out)).files()).get("a.bin").bytes()).equals(bytes)`, + ], + ])("frees its copy of %s when the write is done", async (_, call, header, verify) => { + using dir = tempDir("archive-write-copy", {}); + // 48 copies of 2 MiB are 96 MiB, if every copy stays. + await expectRssDeltaOfPendingCallsBelow( + { + setup: ` + const out = ${JSON.stringify(join(String(dir), "out.tar"))}; + const bytes = Buffer.alloc(2 * 1024 * 1024, 0x61); + `, + // The file is empty before each call. It has the last byte of the data again when the pool thread has written it. + call: `(fs.writeFileSync(out, ""), ${call})`, + done: `byteAt(out, ${header} + bytes.length - 1) === 0x61`, + verify, + count: 48, + }, + { release: 48, debug: 64 }, + ); + }); }); describe("GC safety", () => { diff --git a/test/js/bun/util/zstd.test.ts b/test/js/bun/util/zstd.test.ts index 880b3bea894b..d0be1cf01b85 100644 --- a/test/js/bun/util/zstd.test.ts +++ b/test/js/bun/util/zstd.test.ts @@ -9,7 +9,7 @@ import { zstdDecompressSync, } from "bun"; import { afterAll, beforeAll, describe, expect, it } from "bun:test"; -import { bunEnv, bunExe, isASAN, isLinux, libcPathForDlopen, rss } from "harness"; +import { bunEnv, bunExe, expectRssDeltaOfPendingCallsBelow, isASAN, isLinux, libcPathForDlopen, rss } from "harness"; import { readFileSync } from "node:fs"; import zlib from "node:zlib"; import path from "path"; @@ -887,6 +887,32 @@ describe.concurrent("async compression of a resizable ArrayBuffer that shrinks a }); }); +// The job reads a buffer or an ASCII string of the caller in place. It has its own copy of a +// string that is not ASCII and of a view of a resizable ArrayBuffer. The pool thread frees a +// large copy when it is done with it. If the copy stayed until the JS thread completes the +// call, a caller that starts many calls before it returns to the event loop would hold every +// copy. +// The job leaves no trace that the JS thread can poll for. So the fixture waits until every +// other thread sleeps, which it can see on Linux only. +it.concurrent.skipIf(!isLinux).each([ + ["a string that is not ASCII", `Buffer.alloc(2 * 1024 * 1024, "a").toString("latin1") + "\\u4e2d"`], + [ + "a view of a resizable ArrayBuffer", + `new Uint8Array(new ArrayBuffer(2 * 1024 * 1024, { maxByteLength: 4 * 1024 * 1024 })).fill(0x61)`, + ], +])("zstdCompress frees its copy of %s when the compression is done", async (_, data) => { + // 48 copies of 2 MiB are 96 MiB, if every copy stays. + await expectRssDeltaOfPendingCallsBelow( + { + setup: `const data = ${data};`, + call: `Bun.zstdCompress(data, { level: 1 })`, + verify: `Buffer.from(Bun.zstdDecompressSync(results.at(-1))).equals(Buffer.from(data))`, + count: 48, + }, + { release: 48, debug: 64 }, + ); +}); + describe.concurrent("Zstandard HTTP compression", () => { // Sample data for HTTP tests const testData = { diff --git a/test/js/node/fs/promises.test.js b/test/js/node/fs/promises.test.js index 47a025595ea0..c4992987065c 100644 --- a/test/js/node/fs/promises.test.js +++ b/test/js/node/fs/promises.test.js @@ -1,4 +1,4 @@ -import { tempDir, tempDirWithFiles } from "harness"; +import { expectRssDeltaOfPendingCallsBelow, isWindows, tempDir, tempDirWithFiles } from "harness"; import { join } from "path"; const assert = require("assert"); const os = require("os"); @@ -552,3 +552,89 @@ describe("AbortSignal rejections use node's AbortError shape", () => { expectNodeAbortError(writeErr, signal.reason); }); }); + +// A write job reads a buffer or an ASCII string of the caller in place. It has its own copy of +// the data when it cannot: a string that is not ASCII, a string literal, a string with an +// encoding, and a view of a resizable ArrayBuffer or of a WebAssembly.Memory. The pool thread +// frees a large copy when it is done with the write. If the copy stayed until the JS thread +// completes the write, a caller that starts many writes before it returns to the event loop +// would hold every copy. +describe.concurrent("the write job's copy of the data", () => { + const size = 2 * 1024 * 1024; + // 48 copies of 2 MiB are 96 MiB, if every copy stays. + const count = 48; + const bounds = { release: 48, debug: 64 }; + const text = `Buffer.alloc(${size}, "a").toString("latin1")`; + const wasmView = `new Uint8Array(new WebAssembly.Memory({ initial: ${size / 65536}, maximum: ${(2 * size) / 65536} }).buffer).fill(0x61)`; + const write = `fs.promises.writeFile(out, data)`; + it.each([ + ["writeFile(path, string)", `${text} + "\\u4e2d"`, `Buffer.from(data)`, write], + ["writeFile(path, string literal)", `new Function('return "' + ${text} + '"')()`, `Buffer.from(data)`, write], + [ + "writeFile(path, string, encoding)", + `Buffer.alloc(${size}, "a").toString("base64")`, + `Buffer.from(data, "base64")`, + `fs.promises.writeFile(out, data, "base64")`, + ], + [ + "writeFile(path, view of a resizable ArrayBuffer)", + `new Uint8Array(new ArrayBuffer(${size}, { maxByteLength: ${2 * size} })).fill(0x61)`, + `Buffer.from(data)`, + write, + ], + ["writeFile(path, view of a WebAssembly.Memory)", wasmView, `Buffer.from(data)`, write], + ["appendFile(path, string)", `${text} + "\\u4e2d"`, `Buffer.from(data)`, `fs.promises.appendFile(out, data)`], + ])("of %s is freed when the write is done", async (_, data, written, call) => { + using dir = tempDir("fs-write-copy", {}); + await expectRssDeltaOfPendingCallsBelow( + { + setup: ` + const out = ${JSON.stringify(join(String(dir), "out"))}; + const data = ${data}; + const written = ${written}; + `, + // The file is empty before each call. It has its last byte again when the pool thread has written the data. + call: `(fs.writeFileSync(out, ""), ${call})`, + done: `byteAt(out, written.length - 1) === written.at(-1)`, + verify: `fs.readFileSync(out).equals(written)`, + count, + }, + bounds, + ); + }); + + // On Windows these writes are libuv requests. They have no step on the thread pool of bun. + it.skipIf(isWindows).each([ + ["FileHandle.write(string, position)", `${text} + "\\u4e2d"`, `Buffer.from(data)`, `handle.write(data, 0)`], + [ + // The write is the part of the view that has the other byte. + "FileHandle.write(view of a resizable ArrayBuffer, offset, length, position)", + `new Uint8Array(new ArrayBuffer(${2 * size}, { maxByteLength: ${4 * size} })).fill(0x61).fill(0x62, ${size / 2}, ${size / 2 + size})`, + `Buffer.alloc(${size}, 0x62)`, + `handle.write(data, ${size / 2}, ${size}, 0)`, + ], + [ + "FileHandle.writev([view of a WebAssembly.Memory], position)", + wasmView, + `Buffer.from(data)`, + `handle.writev([data], 0)`, + ], + ])("of %s is freed when the write is done", async (_, data, written, call) => { + using dir = tempDir("fs-write-copy", {}); + await expectRssDeltaOfPendingCallsBelow( + { + setup: ` + const out = ${JSON.stringify(join(String(dir), "out"))}; + const data = ${data}; + const written = ${written}; + const handle = await fs.promises.open(out, "w"); + `, + call: `(fs.ftruncateSync(handle.fd, 0), ${call})`, + done: `byteAt(out, written.length - 1) === written.at(-1)`, + verify: `(await handle.close(), fs.readFileSync(out).equals(written))`, + count, + }, + bounds, + ); + }); +}); From 7ab833321041f6194f37e0c08f408cf94226348b Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 1 Nov 2026 17:31:35 +0000 Subject: [PATCH 2/2] test: cover Bun.zstdDecompress and check every result of the pending calls - zstd.test.ts: a third case starts Bun.zstdDecompress calls on a view of a resizable ArrayBuffer. The input is a small frame and then a skippable frame, so the job has a large input and a small output. Each case now checks every result, not only the last one. - promises.test.js: the FileHandle cases check bytesWritten of every call. - bench/snippets/write-file-copied-data.node.mjs: the UTF-16 source is exactly `size` bytes as UTF-8 when `size` is not a multiple of 3. --- .../snippets/write-file-copied-data.node.mjs | 7 +++- test/js/bun/util/zstd.test.ts | 37 ++++++++++++++++--- test/js/node/fs/promises.test.js | 2 +- 3 files changed, 38 insertions(+), 8 deletions(-) diff --git a/bench/snippets/write-file-copied-data.node.mjs b/bench/snippets/write-file-copied-data.node.mjs index 45ec65a88779..b52a2c5444bc 100644 --- a/bench/snippets/write-file-copied-data.node.mjs +++ b/bench/snippets/write-file-copied-data.node.mjs @@ -19,7 +19,12 @@ const MB = 1024 * KB; const sources = { "ASCII": size => Buffer.alloc(size, "a").toString("latin1"), "Latin-1": size => Buffer.alloc(size / 2, 0xe9).toString("latin1"), - "UTF-16": size => Buffer.alloc((size / 3) * 2, "\u4e2d", "utf16le").toString("utf16le"), + // 3 bytes for each CJK character, and 1 byte for each "a" that fills the rest. + "UTF-16": size => { + const units = Buffer.alloc((Math.floor(size / 3) + (size % 3)) * 2, "\u4e2d", "utf16le"); + units.fill("a", units.length - (size % 3) * 2, units.length, "utf16le"); + return units.toString("utf16le"); + }, "base64": size => Buffer.alloc(size, "a").toString("base64"), "resizable view": size => new Uint8Array(new ArrayBuffer(size, { maxByteLength: 2 * size })).fill(0x61), }; diff --git a/test/js/bun/util/zstd.test.ts b/test/js/bun/util/zstd.test.ts index d0be1cf01b85..fcc60f941cf9 100644 --- a/test/js/bun/util/zstd.test.ts +++ b/test/js/bun/util/zstd.test.ts @@ -895,18 +895,43 @@ describe.concurrent("async compression of a resizable ArrayBuffer that shrinks a // The job leaves no trace that the JS thread can poll for. So the fixture waits until every // other thread sleeps, which it can see on Linux only. it.concurrent.skipIf(!isLinux).each([ - ["a string that is not ASCII", `Buffer.alloc(2 * 1024 * 1024, "a").toString("latin1") + "\\u4e2d"`], [ + "zstdCompress", + "a string that is not ASCII", + `const data = Buffer.alloc(2 * 1024 * 1024, "a").toString("latin1") + "\\u4e2d"; + const expected = Buffer.from(data);`, + `Bun.zstdCompress(data, { level: 1 })`, + `Bun.zstdDecompressSync(result)`, + ], + [ + "zstdCompress", + "a view of a resizable ArrayBuffer", + `const data = new Uint8Array(new ArrayBuffer(2 * 1024 * 1024, { maxByteLength: 4 * 1024 * 1024 })).fill(0x61); + const expected = Buffer.from(data);`, + `Bun.zstdCompress(data, { level: 1 })`, + `Bun.zstdDecompressSync(result)`, + ], + [ + "zstdDecompress", "a view of a resizable ArrayBuffer", - `new Uint8Array(new ArrayBuffer(2 * 1024 * 1024, { maxByteLength: 4 * 1024 * 1024 })).fill(0x61)`, + // A frame of a few bytes, then a skippable frame that fills the view. So the input is large and the output is small. + `const expected = Buffer.from("hello"); + const frame = Bun.zstdCompressSync(expected); + const data = new Uint8Array(new ArrayBuffer(2 * 1024 * 1024, { maxByteLength: 4 * 1024 * 1024 })); + data.set(frame); + const skippable = new DataView(data.buffer, frame.length, 8); + skippable.setUint32(0, 0x184d2a50, true); + skippable.setUint32(4, data.length - frame.length - 8, true);`, + `Bun.zstdDecompress(data)`, + `result`, ], -])("zstdCompress frees its copy of %s when the compression is done", async (_, data) => { +])("%s frees its copy of %s when the job is done", async (_api, _input, setup, call, output) => { // 48 copies of 2 MiB are 96 MiB, if every copy stays. await expectRssDeltaOfPendingCallsBelow( { - setup: `const data = ${data};`, - call: `Bun.zstdCompress(data, { level: 1 })`, - verify: `Buffer.from(Bun.zstdDecompressSync(results.at(-1))).equals(Buffer.from(data))`, + setup, + call, + verify: `results.every(result => Buffer.from(${output}).equals(expected))`, count: 48, }, { release: 48, debug: 64 }, diff --git a/test/js/node/fs/promises.test.js b/test/js/node/fs/promises.test.js index c4992987065c..f8c0f0fcb667 100644 --- a/test/js/node/fs/promises.test.js +++ b/test/js/node/fs/promises.test.js @@ -631,7 +631,7 @@ describe.concurrent("the write job's copy of the data", () => { `, call: `(fs.ftruncateSync(handle.fd, 0), ${call})`, done: `byteAt(out, written.length - 1) === written.at(-1)`, - verify: `(await handle.close(), fs.readFileSync(out).equals(written))`, + verify: `(await handle.close(), results.every(result => result.bytesWritten === written.length) && fs.readFileSync(out).equals(written))`, count, }, bounds,