Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 2 additions & 0 deletions src/runtime/webcore/FileSink.rs
Original file line number Diff line number Diff line change
Expand Up @@ -657,6 +657,7 @@ impl FileSink {
return sys::Result::Err(err);
}
sys::Result::Ok(()) => {
self.fd.set(fd);
self.writer
.with_mut(|w| w.update_ref(self.io_evtloop(), false));
}
Expand All @@ -672,6 +673,7 @@ impl FileSink {
return sys::Result::Err(err);
}
sys::Result::Ok(()) => {
self.fd.set(fd);
// Only keep the event loop ref'd while there's a pending write in progress.
// If there's no pending write, no need to keep the event loop ref'd.
self.writer
Expand Down
69 changes: 27 additions & 42 deletions src/runtime/webcore/blob/read_file.rs
Original file line number Diff line number Diff line change
Expand Up @@ -508,54 +508,39 @@ impl ReadFile {

/// Never touches `self.buffer`; the caller moves it out for the duration.
pub fn do_read(&mut self, buf: &mut [u8], read_len: &mut usize, retry: &mut bool) -> bool {
let result: bun_sys::Result<usize> = 'brk: {
if bun_sys::S::ISSOCK(self.file_store.mode) {
break 'brk bun_sys::recv_non_block(self.opened_fd, buf);
}
break 'brk bun_sys::read(self.opened_fd, buf);
let result: bun_sys::Result<usize> = if bun_sys::S::ISSOCK(self.file_store.mode) {
bun_sys::recv_non_block(self.opened_fd, buf)
} else {
bun_sys::read(self.opened_fd, buf)
};

loop {
match &result {
Ok(res) => {
*read_len = *res as usize; // @truncate — usize→usize is identity here
self.read_eof = *res == 0;
match result {
Ok(res) => {
*read_len = res;
self.read_eof = res == 0;
true
}
Err(err) => {
if err.get_errno() == io::RETRY {
// EAGAIN is impossible on a regular file, so the fd is pollable.
self.could_block = true;
*retry = true;
self.read_eof = false;
return true;
}
Err(err) => {
match err.get_errno() {
e if e == io::RETRY => {
if !self.could_block {
// regular files cannot use epoll.
// this is fine on kqueue, but not on epoll.
continue;
}
*retry = true;
self.read_eof = false;
return true;
}
_ => {
self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
self.system_error = Some(err.to_system_error().into());
if self.system_error.as_ref().unwrap().path.is_empty() {
self.system_error.as_mut().unwrap().path =
if self.file_store.pathlike.is_path() {
BunString::clone_utf8(
self.file_store.pathlike.path().slice(),
)
.into()
} else {
BunString::EMPTY.into()
};
}
return false;
}
}
self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
self.system_error = Some(err.to_system_error().into());
if self.system_error.as_ref().unwrap().path.is_empty() {
self.system_error.as_mut().unwrap().path = if self.file_store.pathlike.is_path()
{
BunString::clone_utf8(self.file_store.pathlike.path().slice()).into()
} else {
BunString::EMPTY.into()
};
}
false
}
break;
}

true
}

pub fn then(this: Box<Self>, _: &JSGlobalObject) -> jsc::JsTerminatedResult<()> {
Expand Down
38 changes: 14 additions & 24 deletions src/runtime/webcore/blob/write_file.rs
Original file line number Diff line number Diff line change
Expand Up @@ -338,32 +338,22 @@ impl WriteFile {
//
// On macOS, it is an error to use pwrite() on a
// non-seekable file.
let result: bun_sys::Result<usize> =
sys::write(fd, &self.bytes_blob.shared_view()[off..off + len]);

loop {
match &result {
bun_sys::Result::Ok(res) => {
*wrote = *res;
self.total_written += *res;
}
bun_sys::Result::Err(err) => {
if err.get_errno() == io::RETRY {
if !self.could_block {
// regular files cannot use epoll.
// this is fine on kqueue, but not on epoll.
continue;
}
self.wait_for_writable();
return false;
} else {
self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
self.system_error = Some(err.to_system_error().into());
return false;
}
match sys::write(fd, &self.bytes_blob.shared_view()[off..off + len]) {
bun_sys::Result::Ok(res) => {
*wrote = res;
self.total_written += res;
}
bun_sys::Result::Err(err) => {
if err.get_errno() == io::RETRY {
// EAGAIN is impossible on a regular file, so the fd is pollable.
self.could_block = true;
self.wait_for_writable();
return false;
}
Comment thread
claude[bot] marked this conversation as resolved.
Resolved
self.errno = Some(bun_errno::from_errno(err.errno as i32).into());
self.system_error = Some(err.to_system_error().into());
return false;
}
break;
}

true
Expand Down
93 changes: 93 additions & 0 deletions test/js/bun/util/filesink.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -622,3 +622,96 @@ it("fs.promises.writeFile with iterables under GC pressure does not crash", asyn
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({ stdout: "ok", stderr: "", exitCode: 0 });
});

// Accessing process.stdout creates a FileSink on a dup of fd 1. openForWriting
// sets O_NONBLOCK on that dup, which also flips fd 1 itself (shared open file
// description). The sink is then forced into synchronous mode, which is meant
// to clear O_NONBLOCK again, but FileSink.setup() never recorded the fd so the
// undo was a no-op. With fd 1 left non-blocking, console.log silently dropped
// bytes on EAGAIN and Bun.write(Bun.stdout, ...) spun forever in WriteFile.
describe.concurrent("process.stdout leaves fd 1 blocking on a pipe", () => {
it.skipIf(!isLinux)("O_NONBLOCK is cleared after process.stdout is materialized", async () => {
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
`
process.stdout.write("x");
const fs = require("fs");
const fdinfo = fs.readFileSync("/proc/self/fdinfo/1", "utf8");
const flags = parseInt(fdinfo.match(/flags:\\s*(\\d+)/)[1], 8);
process.stderr.write((flags & 0o4000) ? "nonblock" : "blocking");
`,
],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect(stdout).toBe("x");
expect(stderr).toBe("blocking");
expect(exitCode).toBe(0);
});

it.skipIf(isWindows)("Bun.write(Bun.stdout, ...) completes after process.stdout is materialized", async () => {
const SIZE = 1024 * 1024;
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
`
const _ = process.stdout;
await Bun.write(Bun.stdout, Buffer.alloc(${SIZE}, "A").toString());
process.stderr.write("done");
`,
],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect(stderr).toBe("done");
expect(stdout.length).toBe(SIZE);
expect(exitCode).toBe(0);
});

it.skipIf(isWindows)(
"console.log / process.stdout.write / Bun.write(Bun.stdout) interleaved deliver every line",
async () => {
const N = 200;
// Longest line "C199 " + PAD + "\n" stays within macOS PIPE_BUF (512) so
// the un-awaited Bun.write running on a work-pool thread cannot interleave
// mid-line with the main thread's writes.
const PAD = 500;
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
`
const pad = Buffer.alloc(${PAD}, "x").toString();
for (let i = 0; i < ${N}; i++) {
console.log("A" + i + " " + pad);
process.stdout.write("B" + i + " " + pad + "\\n");
Bun.write(Bun.stdout, "C" + i + " " + pad + "\\n");
}
Comment thread
claude[bot] marked this conversation as resolved.
Resolved
await Bun.write(Bun.stdout, "");
process.stderr.write("done");
`,
],
env: bunEnv,
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
expect(stderr).toBe("done");
const lines = stdout.split("\n").filter(Boolean);
expect({
A: lines.filter(l => l.startsWith("A")).length,
B: lines.filter(l => l.startsWith("B")).length,
C: lines.filter(l => l.startsWith("C")).length,
total: lines.length,
}).toEqual({ A: N, B: N, C: N, total: 3 * N });
expect(exitCode).toBe(0);
},
);
});
Loading