Skip to content

Bun.file().text(): decode in the read job on the work pool (POSIX) #47509

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robobun wants to merge 3 commits into
base: main
Choose a base branch
from
+305 −47
Open

Bun.file().text(): decode in the read job on the work pool (POSIX) #47509

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
44530c8
Bun.file().text(): decode in the read job on the work pool (POSIX)
robobun Oct 30, 2026
70d5c15
bench: add the eight-read readFile case to bun-file-text and remove i…
robobun Oct 31, 2026
03759bc
bench: bun-file-text removes its fixtures when a fixture write fails
robobun Oct 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 46 additions & 0 deletions bench/snippets/bun-file-text.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
// Bun.file(path).text() against fs.promises.readFile(path, "utf8"), one read
// at a time and eight at once. Node has no Bun.file: (await fs.openAsBlob(path)).text() stands in.
import { mkdtempSync, openAsBlob, rmSync, writeFileSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { bench, group, run } from "../runner.mjs";

const KiB = 1024;
const text = typeof Bun !== "undefined" ? path => Bun.file(path).text() : async path => (await openAsBlob(path)).text();
const eight = read => Promise.all(Array.from({ length: 8 }, read));

const units = {
"ascii": Buffer.from("The quick brown fox jumps over the lazy dog.\n"),
"utf8": Buffer.from("héllo wörld, 世界 💕\n"),
// Every byte is an invalid sequence: the decoder replaces each one.
"invalid utf8": Buffer.from([0xff]),
};

const dir = mkdtempSync(join(tmpdir(), "bun-file-text-bench-"));

function fixture(name, bytes) {
const unit = units[name];
const path = join(dir, `${name.replaceAll(" ", "-")}-${bytes}.txt`);
Comment thread
robobun marked this conversation as resolved.
Resolved
// Whole units only, so the utf8 fixtures stay valid.
writeFileSync(path, Buffer.alloc(bytes - (bytes % unit.length), unit));
return path;
}

try {
for (const bytes of [64, 4 * KiB, 256 * KiB, 8 * KiB * KiB]) {
for (const name of Object.keys(units)) {
const path = fixture(name, bytes);
group(`${bytes} bytes, ${name}`, () => {
bench("text()", async () => (await text(path)).length);
bench('readFile(path, "utf8")', async () => (await readFile(path, "utf8")).length);
bench("8 x text()", async () => (await eight(() => text(path))).length);
bench('8 x readFile(path, "utf8")', async () => (await eight(() => readFile(path, "utf8"))).length);
});
}
}

await run();
} finally {
rmSync(dir, { recursive: true, force: true });
}
2 changes: 1 addition & 1 deletion src/bun_core/string/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -582,7 +582,7 @@ impl String {
/// atom, symbol or substring), so the value may be handed to one other
/// thread. Non-WTF tags are inert and always qualify.
#[inline]
pub(crate) fn is_thread_isolated(&self) -> bool {
pub fn is_thread_isolated(&self) -> bool {
self.tag != Tag::WTFStringImpl || self.as_wtf().is_thread_isolated()
}
/// What [`make_thread_shareable`] yields: isolated, pre-hashed and never
Expand Down
81 changes: 46 additions & 35 deletions src/runtime/webcore/Blob.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,6 +451,7 @@ impl BlobExt for Blob {
self.store().expect("infallible: store present").clone(),
self.offset.get(),
self.size.get(),
F::READ_AS,
)
.unwrap_or_else(|e| bun_core::handle_oom(Err(e)));
// Create the Promise only after the store has been ref()'d.
Expand Down Expand Up @@ -511,6 +512,9 @@ impl BlobExt for Blob {
let buf = unsafe { bun_core::heap::take(b.buf) };
ReadBytesResult::Ok(buf.into_vec())
}
read_file::ReadFileResultType::Text(_) => {
unreachable!("do_read_file_internal reads as ReadAs::Bytes")
}
read_file::ReadFileResultType::Err(e) => ReadBytesResult::Err(Box::new(e)),
};
// SAFETY: `c` is the `ctx` handed to `read_bytes_to_handler`,
Expand Down Expand Up @@ -683,6 +687,7 @@ impl BlobExt for Blob {
self.store().expect("infallible: store present").clone(),
self.offset.get(),
self.size.get(),
read_file::ReadAs::Bytes,
)
.unwrap_or_else(|e| bun_core::handle_oom(Err(e)));
read_file::ReadFile::schedule(
Expand Down Expand Up @@ -2353,22 +2358,20 @@ impl BlobExt for Blob {
global: &JSGlobalObject,
raw_bytes: *mut [u8],
) -> JsResult<JSValue> {
// SAFETY: `raw_bytes` is valid for reads for the duration of this call
// (either a leaked Box for `Temporary` or a store-backed view otherwise).
if LIFETIME == Lifetime::Temporary {
// SAFETY: `Temporary` ⇒ caller passed a leaked `Box<[u8]>`; reclaim it.
let bytes = unsafe { bun_core::heap::take(raw_bytes) };
return text_from_owned_bytes(bytes).into_js(global);
}
// SAFETY: `raw_bytes` is a store-backed view, valid for reads during this call.
let raw_slice: &[u8] = unsafe { &*raw_bytes };
let (bom, buf) = strings::BOM::detect_and_split(raw_slice);

if buf.is_empty() {
// If all it contained was the bom, we need to free the bytes
if LIFETIME == Lifetime::Temporary {
// SAFETY: temporary lifetime means raw_bytes is a leaked default-allocator buffer.
unsafe { drop(bun_core::heap::take(raw_bytes)) };
}
return Ok(JSValue::js_empty_string(global));
}

if bom == Some(strings::BOM::Utf16Le) {
let _free = (LIFETIME == Lifetime::Temporary).then(|| TemporaryBytes(raw_bytes));
// This branch intentionally does NOT `self.detach()` for
// `Lifetime::Transfer`, unlike the toJSON path.
return BunString::clone_utf16_bytes(buf).into_js(global);
Expand All @@ -2385,31 +2388,18 @@ impl BlobExt for Blob {
let converted = match strings::to_utf16_alloc(buf, false, false) {
Ok(converted) => converted,
Err(_) => {
let err = bun_string_jsc::throw_utf16_transcode_failure(global, buf);
if LIFETIME == Lifetime::Temporary {
// SAFETY: `Temporary` ⇒ caller passed a leaked `Box<[u8]>`; reclaim it.
unsafe { drop(bun_core::heap::take(raw_bytes)) };
}
return Err(err);
return Err(bun_string_jsc::throw_utf16_transcode_failure(global, buf));
}
};
if let Some(external) = converted {
if LIFETIME != Lifetime::Temporary {
self.set_is_ascii_flag(false);
}
self.set_is_ascii_flag(false);
if LIFETIME == Lifetime::Transfer {
self.detach();
}
if LIFETIME == Lifetime::Temporary {
// SAFETY: `Temporary` ⇒ caller passed a leaked `Box<[u8]>`; reclaim it.
unsafe { drop(bun_core::heap::take(raw_bytes)) };
}
return bun_string_jsc::owned_utf16_into_js(global, external);
}

if LIFETIME != Lifetime::Temporary {
self.set_is_ascii_flag(true);
}
self.set_is_ascii_flag(true);
}

match LIFETIME {
Expand Down Expand Up @@ -2457,17 +2447,7 @@ impl BlobExt for Blob {
)
}
}
Lifetime::Temporary => {
// if there was a UTF-8 BOM, we need to clone the buffer because
// external doesn't support this case here yet.
if buf.len() != raw_slice.len() {
let out = BunString::clone_latin1(buf);
// SAFETY: `Temporary` ⇒ caller passed a leaked `Box<[u8]>`.
unsafe { drop(bun_core::heap::take(raw_bytes)) };
return out.into_js(global);
}
EncodedSlice::latin1(buf).to_external_value(global)
}
Lifetime::Temporary => unreachable!("returned above"),
}
}

Expand Down Expand Up @@ -5934,6 +5914,25 @@ impl Drop for TemporaryBytes {
}
}

/// `Blob.text()` of bytes that an ASCII text adopts. It needs no JS heap, so it runs on any thread.
pub(crate) fn text_from_owned_bytes(bytes: Box<[u8]>) -> BunString {
let mut bytes = bytes.into_vec();
let (bom, rest) = strings::BOM::detect_and_split(&bytes);
match bom {
Some(strings::BOM::Utf16Le) => BunString::clone_utf16_bytes(rest),
None => BunString::from_owned_utf8(bytes),
Some(strings::BOM::Utf8) => match strings::to_utf16_alloc(rest, false, false) {
Ok(Some(utf16)) => BunString::create_external_globally_allocated_utf16(utf16),
Ok(None) => {
let bom_len = bytes.len() - rest.len();
bytes.drain(..bom_len);
BunString::create_external_globally_allocated_latin1(bytes)
}
Err(_) => BunString::utf16_transcode_failure(rest),
},
}
}

// Marker types for static fn dispatch through do_read_file/do_read_from_s3.
// Each implements `ReadFileToJs` so a plain fn-pointer monomorphizes per `*WithBytes` body.
pub(crate) struct ToStringWithBytesFn;
Expand All @@ -5944,6 +5943,8 @@ pub(crate) struct ToUint8ArrayWithBytesFn;
pub(crate) struct ToFormDataWithBytesFn;

impl read_file::ReadFileToJs for ToStringWithBytesFn {
#[cfg(not(windows))]
const READ_AS: read_file::ReadAs = read_file::ReadAs::Text;
fn call(b: &Blob, g: &JSGlobalObject, by: *mut [u8], l: Lifetime) -> JsResult<JSValue> {
// SAFETY: `by` upholds the `ReadFileToJs::call` contract — a leaked
// `Box<[u8]>` for `Temporary`, store-backed otherwise.
Expand All @@ -5958,6 +5959,8 @@ impl read_file::ReadFileToJs for ToStringWithBytesFn {
}
}
impl read_file::ReadFileToJs for ToJsonWithBytesFn {
#[cfg(not(windows))]
const READ_AS: read_file::ReadAs = read_file::ReadAs::Bytes;
fn call(b: &Blob, g: &JSGlobalObject, by: *mut [u8], l: Lifetime) -> JsResult<JSValue> {
// SAFETY: see `ToStringWithBytesFn::call`.
unsafe {
Expand All @@ -5971,6 +5974,8 @@ impl read_file::ReadFileToJs for ToJsonWithBytesFn {
}
}
impl read_file::ReadFileToJs for ToJsonSourceWithBytesFn {
#[cfg(not(windows))]
const READ_AS: read_file::ReadAs = read_file::ReadAs::Bytes;
fn call(b: &Blob, g: &JSGlobalObject, by: *mut [u8], l: Lifetime) -> JsResult<JSValue> {
// SAFETY: see `ToStringWithBytesFn::call`.
let Some(text) = json_text_behind_mark(g, unsafe { &*by }) else {
Expand All @@ -5984,6 +5989,8 @@ impl read_file::ReadFileToJs for ToJsonSourceWithBytesFn {
}
}
impl read_file::ReadFileToJs for ToArrayBufferWithBytesFn {
#[cfg(not(windows))]
const READ_AS: read_file::ReadAs = read_file::ReadAs::Bytes;
fn call(b: &Blob, g: &JSGlobalObject, by: *mut [u8], l: Lifetime) -> JsResult<JSValue> {
// SAFETY: see `ToStringWithBytesFn::call`.
unsafe {
Expand All @@ -5999,6 +6006,8 @@ impl read_file::ReadFileToJs for ToArrayBufferWithBytesFn {
}
}
impl read_file::ReadFileToJs for ToUint8ArrayWithBytesFn {
#[cfg(not(windows))]
const READ_AS: read_file::ReadAs = read_file::ReadAs::Bytes;
fn call(b: &Blob, g: &JSGlobalObject, by: *mut [u8], l: Lifetime) -> JsResult<JSValue> {
// SAFETY: see `ToStringWithBytesFn::call`.
unsafe {
Expand All @@ -6014,6 +6023,8 @@ impl read_file::ReadFileToJs for ToUint8ArrayWithBytesFn {
}
}
impl read_file::ReadFileToJs for ToFormDataWithBytesFn {
#[cfg(not(windows))]
const READ_AS: read_file::ReadAs = read_file::ReadAs::Bytes;
fn call(b: &Blob, g: &JSGlobalObject, by: *mut [u8], l: Lifetime) -> JsResult<JSValue> {
let _ = l; // FormData ignores lifetime — bytes are read-only.
// SAFETY: `by` is valid for reads per the `ReadFileToJs::call` contract.
Expand Down
52 changes: 46 additions & 6 deletions src/runtime/webcore/blob/read_file.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,12 @@ use core::sync::atomic::Ordering;

use crate::Error;
use crate::webcore::Lifetime;
#[cfg(not(windows))]
use crate::webcore::blob::ClosingState;
#[cfg(windows)]
use crate::webcore::blob::store::Bytes as ByteStore;
use crate::webcore::blob::store::{Data, File as FileStore};
use crate::webcore::blob::{Blob, FileCloser, FileOpener, MAX_SIZE, SizeType, Store};
#[cfg(not(windows))]
use crate::webcore::blob::{ClosingState, text_from_owned_bytes};
use crate::webcore::node_types::PathOrFileDescriptor;
#[cfg(windows)]
use bun_collections::ByteVecExt as _;
Expand All @@ -25,7 +25,8 @@ use bun_io::FileAction;
// `bun_jsc::EventLoop` is the *module*; the struct is one level deeper.
use bun_jsc::event_loop::EventLoop;
use bun_jsc::{
self as jsc, AnyPromise, JSGlobalObject, JSPromiseStrong, JSValue, JsResult, SystemError,
self as jsc, AnyPromise, JSGlobalObject, JSPromiseStrong, JSValue, JsResult, StringJsc as _,
SystemError,
};
use bun_ptr::RefPtr;
#[cfg(windows)]
Expand All @@ -52,9 +53,21 @@ macro_rules! log {
// NewReadFileHandler
// ──────────────────────────────────────────────────────────────────────────

/// What a [`ReadFile`] job hands its completion.
#[cfg(not(windows))]
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum ReadAs {
Bytes,
/// Blob text, decoded on the work pool: [`ReadFileResultType::Text`].
Text,
}

/// `F` provides the callback that converts the read bytes to a JSValue.
/// Modelled as a trait so each instantiation monomorphizes.
pub(crate) trait ReadFileToJs {
/// What the job hands `then`. Windows has no pool stage until `ReadFileUV` goes (#42819).
#[cfg(not(windows))]
const READ_AS: ReadAs;
/// `by` carries the caller's allocation provenance unchanged:
/// `Lifetime::Temporary` ⇒ a `Box::<[u8]>::into_raw` the callee MUST take
/// ownership of (every `to_*_with_bytes::<Temporary>` arm reclaims it);
Expand Down Expand Up @@ -131,6 +144,9 @@ impl<'a, F: ReadFileToJs> ReadFileCompletion for NewReadFileHandler<'a, F> {
)
})?;
}
ReadFileResultType::Text(text) => {
AnyPromise::Normal(promise).wrap(global_this, move |g| text.into_js(g))?;
}
ReadFileResultType::Err(err) => {
// SAFETY: `promise` was just swapped out of `handler.promise`,
// the `JSPromiseStrong` that rooted it across the async read;
Expand Down Expand Up @@ -215,15 +231,18 @@ pub(crate) struct ReadFileRead {
#[allow(clippy::large_enum_variant)]
pub(crate) enum ReadFileResultType {
Result(ReadFileRead),
/// Blob text that the job decoded on the work pool.
Text(BunString),
Err(SystemError),
}

/// The completion token a `ReadFile` keeps across its async I/O.
pub(crate) type ReadFileTask = bun_jsc::Completion<ReadFile>;

// SAFETY: file store / byte store / blob store ref (atomic), the read buffer and io-loop
// registration state — nothing thread-affine. What the bytes are delivered to lives in the job's
// JS side (`ReadFileCompletionFns`), never here.
// SAFETY: file store / byte store / blob store ref (atomic), the read buffer, the text decoded
// from it (a string this job built and no other thread has seen) and io-loop registration
// state: nothing thread-affine. What the bytes are delivered to lives in the job's JS side
// (`ReadFileCompletionFns`), never here.
unsafe impl Send for ReadFile {}

impl bun_jsc::JobContext for ReadFile {
Expand Down Expand Up @@ -293,6 +312,10 @@ pub(crate) struct ReadFile {
#[cfg(not(windows))]
pub(crate) size: SizeType,
pub(crate) buffer: Vec<u8>,
#[cfg(not(windows))]
pub(crate) read_as: ReadAs,
/// `buffer`, decoded where the job finishes. Only a `ReadAs::Text` job sets it.
pub(crate) text: Option<BunString>,
pub task: WorkPoolTask,
pub(crate) system_error: Option<SystemError>,
pub(crate) errno: Option<Error>,
Expand Down Expand Up @@ -370,6 +393,7 @@ impl ReadFile {
store: RefPtr<Store>,
off: SizeType,
max_len: SizeType,
read_as: ReadAs,
) -> Result<ReadFile, Error> {
let file_store = store.data.as_file().clone();
let read_file = ReadFile {
Expand All @@ -383,6 +407,8 @@ impl ReadFile {
read_eof: false,
size: 0,
buffer: Vec::new(),
read_as,
text: None,
task: WorkPoolTask {
node: Default::default(),
callback: Self::do_read_loop_task,
Expand Down Expand Up @@ -635,6 +661,7 @@ impl ReadFile {
let _store = this.store.take().unwrap();
// reshaped for borrowck — take buffer out so it survives `drop(this)`.
let buf = core::mem::take(&mut this.buffer);
let text = this.text.take();

// `_store` is dropped at end of scope (= store.deref()).
let system_error = this.system_error.take();
Expand All @@ -643,6 +670,9 @@ impl ReadFile {
if let Some(err) = system_error {
return completion.complete(ReadFileResultType::Err(err));
}
if let Some(text) = text {
return completion.complete(ReadFileResultType::Text(text));
}

// The receiver takes ownership. Normalize to `Box<[u8]>` so every
// consumer can reclaim via `heap::take` with a matching layout.
Expand Down Expand Up @@ -694,6 +724,16 @@ impl ReadFile {
if !close_after_io {
if let Some(io_task) = self.io_task.take() {
bloblog!("ReadFile.onFinish() = immediately");
// A stopping VM releases this job unrun, and waits for this thread first.
if self.read_as == ReadAs::Text
&& self.system_error.is_none()
&& io_task.ticket().script_allowed()
{
let bytes = core::mem::take(&mut self.buffer).into_boxed_slice();
let text = text_from_owned_bytes(bytes);
Comment on lines +728 to +733

@claude claude Bot Oct 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Servers that await several large .text() reads at once now peak at up to ~50% more RSS than on the base branch and can be OOM-killed under a memory limit. read_file.rs:733 decodes on the pool thread, so each of up to pool-size reads holds its bytes plus its UTF-16 output at the same moment, and every finished read waits for the JS thread as UTF-16 (2x for multibyte) instead of as bytes. The base decoded one read after another on the JS thread. Fix: bound the memory added per overlapping read, e.g. cap concurrent decodes with a semaphore or decode only when the completion is about to run, and document the new peak formula.

Why this was flagged

Trigger: k overlapping Bun.file(p).text() calls on large non-ASCII files from a Bun.serve handler or a Promise.all. read_file.rs:728-735 runs text_from_owned_bytes inside on_finish on the work pool; Blob.rs:5918-5933 keeps the input bytes alive while to_utf16_alloc (string/mod.rs:487) allocates the UTF-16 output, so each pool thread holds both at once, and the result sits in self.text (read_file.rs:735) until the JS thread drains the completion. On the base branch then() (read_file.rs:680) handed raw bytes to the JS thread and to_string_with_bytes::<Temporary> decoded serially, so at most one bytes+UTF-16 pair existed and queued results were 1x bytes. The extra peak scales with min(overlapping reads, pool threads) plus 2x per queued multibyte result; a container with a cgroup memory limit sized on the base's peak gets OOM-killed. No safeguard bounds it. Remedy: cap concurrent pool decodes or keep queued results as bytes until the JS thread is ready.

Verification: src/runtime/webcore/blob/read_file.rs:728-735 on_finish runs text_from_owned_bytes on the pool thread; src/bun_core/string/mod.rs:486-492 runs to_utf16_alloc while bytes is still owned, so each pool thread holds the input bytes plus the UTF-16 buffer. On the base branch the same transient occurred one read at a time on the JS thread, and pending results waited as raw bytes rather than as UTF-16.

@robobun robobun Oct 31, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the cost in the first Downsides bullet, and the numbers agree. I did not add a cap in this PR. The PR body asks a maintainer to decide it.

What the measurements in the notes show:

  • Twelve overlapping 32 MiB reads peak at 1,028 MB against 675 MB on main, with all twelve results kept. The extra memory is the input of the decodes that run at the same time: at most min(overlapping reads, pool threads) x file size.
  • fs.promises.readFile(path, "utf8") decodes on the pool in the same way today. It peaks at 1,030 MB for the same twelve reads.

What a cap costs:

  • A cap makes decodes wait. A pool thread must not block for that, so the reads that finished need a process-wide queue.
  • With one large decode at a time, the event loop would stay free and the peak would return to about the value on main. The wall time of overlapping large reads would also return to about the value on main (344 ms for the twelve reads, 98 ms in this PR). I did not build or measure this variant.
  • .text() then behaves differently from readFile.

If a maintainer wants the bound, this block in on_finish is the place for it: a byte budget for the decodes in flight, and a read over the budget waits. I can add that to this PR.

debug_assert!(text.is_thread_isolated());
self.text = Some(text);
}
io_task.finish();
Comment thread
robobun marked this conversation as resolved.
Resolved
}
}
Expand Down
Loading
Loading