diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index ec3d76a55648..0ab26af0251e 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -1354,6 +1354,7 @@ impl BlobExt for Blob { mkdirp_if_not_exists, extra_options: options, mode: None, + handoff_fd: None, }, ) } @@ -4456,6 +4457,9 @@ pub struct WriteFileOptions { pub mkdirp_if_not_exists: Option, pub extra_options: Option, pub mode: Option, + /// Destination fd already opened by the sync fast path, for `WriteFile` + /// to adopt instead of reopening. + pub handoff_fd: Option, } /// Write an empty string to a file by truncating it. @@ -4674,6 +4678,14 @@ pub fn write_file_with_source_destination( destination_blob: &mut Blob, options: &WriteFileOptions, ) -> JsResult { + // Close the handoff fd on any branch that doesn't adopt it. + let handoff_fd = core::cell::Cell::new(options.handoff_fd); + let _close_handoff_fd = scopeguard::guard((), |()| { + if let Some(fd) = handoff_fd.get() { + let _ = bun_sys::close(fd); + } + }); + let destination_store = destination_blob .store .get() @@ -4734,6 +4746,10 @@ pub fn write_file_with_source_destination( options.mkdirp_if_not_exists.unwrap_or(true), ) .expect("unreachable"); + if let Some(fd) = handoff_fd.take() { + // SAFETY: file_copier was just produced by heap::into_raw; sole owner. + unsafe { (*file_copier).opened_fd = fd }; + } let task = write_file_mod::WriteFileTask::create_on_js_thread(ctx, file_copier); // Defer promise creation until we're just about to schedule the task. // `JSPromiseStrong.strong` is private in `bun_jsc`, so use `init` (which @@ -5040,6 +5056,16 @@ pub fn write_file_internal( } } + #[cfg(not(windows))] + let handoff_fd = core::cell::Cell::new(None::); + // Close the fast-path fd on any early return before WriteFile adopts it. + #[cfg(not(windows))] + let _close_handoff_fd = scopeguard::guard((), |()| { + if let Some(fd) = handoff_fd.get() { + let _ = bun_sys::close(fd); + } + }); + // If you're doing Bun.write(), try to go fast by writing short input on the main thread. // This is a heuristic, but it's a good one. // @@ -5047,6 +5073,7 @@ pub fn write_file_internal( #[cfg(not(windows))] { let mut needs_async = false; + let mut fast_path_fd: Option = None; let fast_path_ok = matches!(*path_or_blob, PathOrBlob::Path(_)) || (matches!(*path_or_blob, PathOrBlob::Blob(ref b) if b.offset.get() == 0 && !b.is_s3() @@ -5075,6 +5102,7 @@ pub fn write_file_internal( &pathlike, str.get(), &mut needs_async, + &mut fast_path_fd, ) } else { write_string_to_file_fast::( @@ -5082,6 +5110,7 @@ pub fn write_file_internal( &pathlike, str.get(), &mut needs_async, + &mut fast_path_fd, ) }; if !needs_async { @@ -5106,6 +5135,7 @@ pub fn write_file_internal( &pathlike, buffer_view.byte_slice(), &mut needs_async, + &mut fast_path_fd, ) } else { write_bytes_to_file_fast::( @@ -5113,6 +5143,7 @@ pub fn write_file_internal( &pathlike, buffer_view.byte_slice(), &mut needs_async, + &mut fast_path_fd, ) }; if !needs_async { @@ -5121,6 +5152,7 @@ pub fn write_file_internal( } } } + handoff_fd.set(fast_path_fd); } // if path_or_blob is a path, convert it into a file blob @@ -5285,6 +5317,12 @@ pub fn write_file_internal( // StoreRef clone+drop keeps the destination store alive across the call. let _dest_hold = destination_store; + #[cfg(not(windows))] + let options = WriteFileOptions { + handoff_fd: handoff_fd.take(), + ..options + }; + write_file_with_source_destination( global_this, &mut *source_blob, @@ -5375,6 +5413,7 @@ pub fn write_file(global_this: &JSGlobalObject, callframe: &CallFrame) -> JsResu mkdirp_if_not_exists, extra_options: options, mode, + handoff_fd: None, }, ) } @@ -5387,6 +5426,7 @@ fn write_string_to_file_fast( pathlike: &PathOrFileDescriptor, str: BunString, needs_async: &mut bool, + handoff_fd: &mut Option, ) -> JSValue { let fd: Fd = if !NEEDS_OPEN { pathlike.fd() @@ -5413,6 +5453,20 @@ fn write_string_to_file_fast( } }; + // Non-regular files (FIFO/socket/chardev) can EAGAIN; route them to the + // async WriteFile path before any bytes go out so it can wait for POLLOUT. + if !str.is_empty() { + if let bun_sys::Result::Ok(st) = bun_sys::fstat(fd) { + if !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode) { + *needs_async = true; + if NEEDS_OPEN { + *handoff_fd = Some(fd); + } + return JSValue::ZERO; + } + } + } + // Declared before the truncate guard so it drops *after* it (close runs last). let _close = NEEDS_OPEN.then(|| bun_sys::CloseOnDrop::new(fd)); @@ -5470,7 +5524,8 @@ fn write_bytes_to_file_fast( global_this: &JSGlobalObject, pathlike: &PathOrFileDescriptor, bytes: &[u8], - _needs_async: &mut bool, + needs_async: &mut bool, + handoff_fd: &mut Option, ) -> JSValue { let fd: Fd = if !NEEDS_OPEN { pathlike.fd() @@ -5490,7 +5545,7 @@ fn write_bytes_to_file_fast( bun_sys::Result::Err(err) => { #[cfg(not(windows))] if err.get_errno() == bun_sys::E::ENOENT { - *_needs_async = true; + *needs_async = true; return JSValue::ZERO; } return JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm( @@ -5501,6 +5556,20 @@ fn write_bytes_to_file_fast( } }; + // Non-regular files (FIFO/socket/chardev) can EAGAIN; route them to the + // async WriteFile path before any bytes go out so it can wait for POLLOUT. + if !bytes.is_empty() { + if let bun_sys::Result::Ok(st) = bun_sys::fstat(fd) { + if !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode) { + *needs_async = true; + if NEEDS_OPEN { + *handoff_fd = Some(fd); + } + return JSValue::ZERO; + } + } + } + // TODO: on windows this is always synchronous let truncate = NEEDS_OPEN || bytes.is_empty(); @@ -5520,7 +5589,7 @@ fn write_bytes_to_file_fast( bun_sys::Result::Err(err) => { #[cfg(not(windows))] if err.get_errno() == bun_sys::E::EAGAIN { - *_needs_async = true; + *needs_async = true; return JSValue::ZERO; } let err_js = if !NEEDS_OPEN { diff --git a/src/runtime/webcore/S3Client.rs b/src/runtime/webcore/S3Client.rs index 02f6a6345b00..d2607e5a1097 100644 --- a/src/runtime/webcore/S3Client.rs +++ b/src/runtime/webcore/S3Client.rs @@ -594,6 +594,7 @@ impl S3Client { mkdirp_if_not_exists: Some(false), extra_options: options, mode: None, + handoff_fd: None, }, ) } diff --git a/src/runtime/webcore/blob/write_file.rs b/src/runtime/webcore/blob/write_file.rs index 1f6795af3729..3dad1e1704fb 100644 --- a/src/runtime/webcore/blob/write_file.rs +++ b/src/runtime/webcore/blob/write_file.rs @@ -333,15 +333,15 @@ impl WriteFile { let fd = self.opened_fd; debug_assert!(fd != Fd::INVALID); - // We do not use pwrite() because the file may not be - // seekable (such as stdout) - // - // On macOS, it is an error to use pwrite() on a - // non-seekable file. - let result: bun_sys::Result = - sys::write(fd, &self.bytes_blob.shared_view()[off..off + len]); - loop { + // We do not use pwrite() because the file may not be + // seekable (such as stdout) + // + // On macOS, it is an error to use pwrite() on a + // non-seekable file. + let result: bun_sys::Result = + sys::write(fd, &self.bytes_blob.shared_view()[off..off + len]); + match &result { bun_sys::Result::Ok(res) => { *wrote = *res; @@ -464,10 +464,11 @@ impl WriteFile { } } - // We opened the file descriptor with O_NONBLOCK, so we - // shouldn't have to worry about blocking reads/writes - // - // We do not call fstat() because that is very expensive. + // Path-opened fds have no cached mode; fstat so a FIFO/socket + // reaches wait_for_writable() on EAGAIN. + if let bun_sys::Result::Ok(st) = sys::fstat(fd) { + break 'brk !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode); + } false }; diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index a4c3fb2f551b..0f3919feea33 100644 --- a/test/js/bun/io/bun-write.test.js +++ b/test/js/bun/io/bun-write.test.js @@ -1,6 +1,7 @@ import { describe, expect, it, test } from "bun:test"; import fs, { mkdirSync } from "fs"; import { bunEnv, bunExe, exampleHtml, exampleSite, gcTick, isWindows, tempDir, withoutAggressiveGC } from "harness"; +import { mkfifo } from "mkfifo"; import path, { join } from "path"; let i = 0; @@ -727,4 +728,68 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { expect(f.name).toBe(filePath); }); + + // Writing more than the kernel pipe buffer to a FIFO used to deliver only the + // first partial write to the reader and then either reject ENXIO (fast path + // closed and reopened the fifo) or never settle (async path spun on EAGAIN + // with could_block=false). Both must now deliver the full payload. + describe.skipIf(isWindows).each([ + ["string", payload => payload], + ["Uint8Array", payload => new TextEncoder().encode(payload)], + ["Blob", payload => new Blob([payload])], + ["Response", payload => new Response(payload)], + ])("Bun.write(fifo, %s) larger than the pipe buffer", (label, toSource) => { + // 200003 bytes: well over the 64 KiB Linux pipe buffer (and the 8 KiB + // minimum some CI kernels use), with a distinct tail so a torn prefix + // is visible. + const N = 200003; + const payload = Buffer.alloc(N - 3, "x").toString() + "END"; + + async function exercise(suffix, writeTo) { + using dir = tempDir(`bun-write-fifo-${label}-${suffix}`, {}); + const fifo = join(String(dir), "f.fifo"); + mkfifo(fifo); + + // Open the read side synchronously with O_NONBLOCK so it returns + // immediately (no writer required) and the writer's O_WRONLY|O_NONBLOCK + // open cannot observe ENXIO. Reading through Bun.file(fd).stream() polls + // on EAGAIN instead of parking a thread-pool worker, which matters under + // describe.concurrent. + const readFd = fs.openSync(fifo, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + const drained = (async () => { + const chunks = []; + for await (const chunk of Bun.file(readFd).stream()) chunks.push(chunk); + return Buffer.concat(chunks); + })(); + drained.catch(() => {}); + + let got; + let written; + try { + written = await writeTo(fifo, toSource(payload)); + got = await drained; + } finally { + fs.closeSync(readFd); + await drained.catch(() => {}); + } + + expect({ bytes: got.length, tail: got.subarray(-3).toString(), written }).toEqual({ + bytes: N, + tail: "END", + written: N, + }); + } + + it("delivers every byte to the reader (path dest)", () => exercise("path", (fifo, src) => Bun.write(fifo, src))); + + it("delivers every byte to the reader (O_NONBLOCK fd dest)", () => + exercise("fd", async (fifo, src) => { + const wfd = fs.openSync(fifo, fs.constants.O_WRONLY | fs.constants.O_NONBLOCK); + try { + return await Bun.write(Bun.file(wfd), src); + } finally { + fs.closeSync(wfd); + } + })); + }); });