Free a thread-pool job's own copy of a large input when the pool thread is done with it

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 });
+}

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

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<u8>) -> Vec<u8> {
+        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::<u8>()`
     // + `set_len(len+n)` copies: `spare_capacity_mut` returns

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<u8> {
+        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.

src/jsc/bindgen.rs

@@ -880,6 +880,8 @@ pub struct PinnedBytes(ThreadIsolated<Pinned>);
 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<u8> {
+        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(),
         }
     }

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<u8>` and drop that on its own thread.
 pub unsafe trait ThreadIsolatedArg {}
 
 /// A `T` built as [`ThreadIsolatedArg`] describes.

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<PromiseResult> {

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)
         }
 

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,14 +1407,15 @@ 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.
         unsafe impl ThreadIsolatedArg for $ty {}
         impl FsArgument for $ty {
             #[inline] fn from_js(ctx: &JSGlobalObject, arguments: &mut ArgumentsSlice) -> JsResult<Self> { <$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<Self>,
         ) -> Option<bun_jsc::Completion<Self>> {
-            let mut node_fs = NodeFS::default();
-            this.result = NodeFS::dispatch::<R, A, F>(&mut node_fs, &this.args, Flavor::Async);
+            this.result = {
+                let mut node_fs = NodeFS::default();
+                NodeFS::dispatch::<R, A, F>(&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)
         }
 

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<u8> {
+        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) {