diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index ec3d76a55648..b52a827370b8 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -4672,6 +4672,7 @@ pub fn write_file_with_source_destination( ctx: &JSGlobalObject, source_blob: &mut Blob, destination_blob: &mut Blob, + dest_opened_fd: Fd, options: &WriteFileOptions, ) -> JsResult { let destination_store = destination_blob @@ -4688,10 +4689,17 @@ pub fn write_file_with_source_destination( ); let Some(source_store) = source_blob.store.get().clone() else { + debug_assert!(dest_opened_fd == Fd::INVALID); return write_file_with_empty_source_to_destination(ctx, destination_blob, options); }; let source_type = source_store.data.tag(); + debug_assert!( + dest_opened_fd == Fd::INVALID + || (destination_type == store::DataTag::File && source_type == store::DataTag::Bytes), + "fast path only pre-opens for Bytes->File; other arms would leak the fd", + ); + if destination_type == store::DataTag::File && source_type == store::DataTag::Bytes { let write_file_promise = bun_core::heap::into_raw(Box::new(WriteFilePromise { promise: jsc::JSPromiseStrong::default(), @@ -4702,6 +4710,8 @@ pub fn write_file_with_source_destination( // `WriteFile::create` takes its own ref. #[cfg(windows)] { + debug_assert!(dest_opened_fd == Fd::INVALID); + let _ = dest_opened_fd; let promise = JSPromise::create(ctx); let promise_value = promise.as_value(ctx); promise_value.ensure_still_alive(); @@ -4729,6 +4739,7 @@ pub fn write_file_with_source_destination( let file_copier = write_file_mod::WriteFile::create( destination_blob.borrowed_view(), source_blob.borrowed_view(), + dest_opened_fd, write_file_promise, WriteFilePromise::run, options.mkdirp_if_not_exists.unwrap_or(true), @@ -5044,6 +5055,8 @@ pub fn write_file_internal( // This is a heuristic, but it's a good one. // // except if you're on Windows. Windows I/O is slower. Let's not even try. + #[cfg_attr(windows, allow(unused_mut))] + let mut dest_opened_fd = Fd::INVALID; #[cfg(not(windows))] { let mut needs_async = false; @@ -5075,6 +5088,7 @@ pub fn write_file_internal( &pathlike, str.get(), &mut needs_async, + &mut dest_opened_fd, ) } else { write_string_to_file_fast::( @@ -5082,6 +5096,7 @@ pub fn write_file_internal( &pathlike, str.get(), &mut needs_async, + &mut dest_opened_fd, ) }; if !needs_async { @@ -5106,6 +5121,7 @@ pub fn write_file_internal( &pathlike, buffer_view.byte_slice(), &mut needs_async, + &mut dest_opened_fd, ) } else { write_bytes_to_file_fast::( @@ -5113,6 +5129,7 @@ pub fn write_file_internal( &pathlike, buffer_view.byte_slice(), &mut needs_async, + &mut dest_opened_fd, ) }; if !needs_async { @@ -5123,6 +5140,13 @@ pub fn write_file_internal( } } + // Close the fast path's fd if anything below throws before `WriteFile` adopts it. + let dest_opened_fd = scopeguard::guard(dest_opened_fd, |fd| { + if fd != Fd::INVALID { + let _ = bun_sys::close(fd); + } + }); + // if path_or_blob is a path, convert it into a file blob let mut destination_blob: Blob = match path_or_blob { PathOrBlob::Path(path) => { @@ -5289,6 +5313,7 @@ pub fn write_file_internal( global_this, &mut *source_blob, &mut destination_blob, + scopeguard::ScopeGuard::into_inner(dest_opened_fd), &options, ) } @@ -5387,6 +5412,7 @@ fn write_string_to_file_fast( pathlike: &PathOrFileDescriptor, str: BunString, needs_async: &mut bool, + handoff_fd: &mut Fd, ) -> JSValue { let fd: Fd = if !NEEDS_OPEN { pathlike.fd() @@ -5413,6 +5439,18 @@ fn write_string_to_file_fast( } }; + // Hand a FIFO/socket/chardev fd to the async WriteFile: closing here after a + // partial write would EOF the reader and make the async reopen fail ENXIO. + if NEEDS_OPEN && !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) { + *handoff_fd = fd; + *needs_async = true; + 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)); @@ -5471,6 +5509,7 @@ fn write_bytes_to_file_fast( pathlike: &PathOrFileDescriptor, bytes: &[u8], _needs_async: &mut bool, + handoff_fd: &mut Fd, ) -> JSValue { let fd: Fd = if !NEEDS_OPEN { pathlike.fd() @@ -5501,6 +5540,17 @@ fn write_bytes_to_file_fast( } }; + // See `write_string_to_file_fast`: hand a non-regular fd to the async WriteFile. + if NEEDS_OPEN && !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) { + *handoff_fd = fd; + *_needs_async = true; + return JSValue::ZERO; + } + } + } + // TODO: on windows this is always synchronous let truncate = NEEDS_OPEN || bytes.is_empty(); diff --git a/src/runtime/webcore/blob/write_file.rs b/src/runtime/webcore/blob/write_file.rs index 1f6795af3729..b3587c757fa6 100644 --- a/src/runtime/webcore/blob/write_file.rs +++ b/src/runtime/webcore/blob/write_file.rs @@ -276,6 +276,7 @@ impl WriteFile { pub fn create_with_ctx( file_blob: Blob, bytes_blob: Blob, + opened_fd: Fd, on_write_file_context: *mut c_void, on_complete_callback: WriteFileOnWriteFileCallback, mkdirp_if_not_exists: bool, @@ -283,7 +284,7 @@ impl WriteFile { let write_file = bun_core::heap::into_raw(Box::new(WriteFile { file_blob, bytes_blob, - opened_fd: Fd::INVALID, + opened_fd, system_error: None, errno: None, task: WorkPoolTask { @@ -311,6 +312,7 @@ impl WriteFile { pub fn create( file_blob: Blob, bytes_blob: Blob, + opened_fd: Fd, context: *mut C, callback: WriteFileOnWriteFileCallback, mkdirp_if_not_exists: bool, @@ -321,6 +323,7 @@ impl WriteFile { WriteFile::create_with_ctx( file_blob, bytes_blob, + opened_fd, context.cast::(), callback, mkdirp_if_not_exists, @@ -338,35 +341,24 @@ impl WriteFile { // // 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 { - 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; + true + } + bun_sys::Result::Err(err) => { + if err.get_errno() == io::RETRY { + // Regular-file write(2) never EAGAINs, so the fd is pollable. + self.could_block = true; + self.wait_for_writable(); + } else { + self.errno = Some(bun_errno::from_errno(err.errno as i32).into()); + self.system_error = Some(err.to_system_error().into()); } + false } - break; } - - true } pub fn then(mut this: Box, _global: &JSGlobalObject) -> Result<(), JsTerminated> { @@ -453,21 +445,24 @@ impl WriteFile { self.could_block = 'brk: { if let Some(store) = self.file_blob.store.get().as_ref() { if let blob::store::Data::File(file) = &store.data { - if file.pathlike.is_fd() { - // If seekable was set, then so was mode - if file.seekable.is_some() { - // This is mostly to handle pipes which were passsed to the process somehow - // such as stderr, stdout. Bun.stdin and Bun.stderr will automatically set `mode` for us. - break 'brk !bun_sys::is_regular_file(file.mode); + match file.pathlike { + PathOrFileDescriptor::Fd(_) => { + // If seekable was set, then so was mode + if file.seekable.is_some() { + // This is mostly to handle pipes which were passsed to the process somehow + // such as stderr, stdout. Bun.stdin and Bun.stderr will automatically set `mode` for us. + break 'brk !bun_sys::is_regular_file(file.mode); + } + } + PathOrFileDescriptor::Path(_) => { + // Opened O_NONBLOCK; the path may be a FIFO/socket/chardev. + if let bun_sys::Result::Ok(st) = sys::fstat(fd) { + break 'brk !bun_sys::is_regular_file(st.st_mode as bun_sys::Mode); + } } } } } - - // 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. false }; @@ -1381,6 +1376,7 @@ impl WriteFileWaitFromLockedValueTask { global_this, &mut blob, &mut file_blob, + Fd::INVALID, &blob::WriteFileOptions { mkdirp_if_not_exists: Some(this_ref.mkdirp_if_not_exists), ..Default::default() diff --git a/test/js/bun/io/bun-write.test.js b/test/js/bun/io/bun-write.test.js index a4c3fb2f551b..6c4f4771a2f3 100644 --- a/test/js/bun/io/bun-write.test.js +++ b/test/js/bun/io/bun-write.test.js @@ -728,3 +728,68 @@ int posix_fadvise(int fd, off_t offset, off_t len, int advice) { expect(f.name).toBe(filePath); }); }); + +// Bun.write(path, ...) on a FIFO used to either (a) tear the payload at the +// pipe-buffer boundary because the sync fast path closed the fd on EAGAIN and +// the async fallback's reopen got ENXIO after the reader saw EOF, or (b) never +// settle because the thread-pool WriteFile kept re-matching a cached EAGAIN +// without ever calling write() again. Run outside describe.concurrent so an +// unfixed build's spin doesn't starve neighbours into their default timeout. +describe.skipIf(isWindows)("Bun.write to a FIFO by path", () => { + // 200 KiB exercises the <256 KiB sync fast path; 1 MiB exercises the + // thread-pool WriteFile path (string/ArrayBuffer) and the Blob path. + it.each([ + ["200 KiB string", "string", 200 * 1024], + ["1 MiB string", "string", 1 << 20], + ["1 MiB Uint8Array", "u8", 1 << 20], + ["1 MiB Blob", "blob", 1 << 20], + ])( + "delivers a %s in full", + async (_label, kind, size) => { + using dir = tempDir("bun-write-fifo", {}); + const script = ` + const fs = require("fs"); + const { FIFO, KIND, SIZE } = process.env; + const size = Number(SIZE); + require("child_process").execFileSync("mkfifo", [FIFO]); + // Open the read end synchronously so Bun.write's O_WRONLY|O_NONBLOCK + // open has a reader and doesn't ENXIO; hand it to a child to drain so + // the write side actually back-pressures. + const readFd = fs.openSync(FIFO, fs.constants.O_RDONLY | fs.constants.O_NONBLOCK); + const reader = Bun.spawn({ + cmd: [process.execPath, "-e", + "let n=0; for await (const c of Bun.stdin.stream()) n+=c.length; process.stdout.write(String(n))"], + stdin: readFd, + stdout: "pipe", + stderr: "inherit", + }); + fs.closeSync(readFd); + const body = Buffer.alloc(size, 0x61); + const src = + KIND === "string" ? body.toString() + : KIND === "u8" ? new Uint8Array(body) + : new Blob([body]); + const wrote = await Bun.write(FIFO, src); + const got = Number(await reader.stdout.text()); + await reader.exited; + console.log(JSON.stringify({ wrote, got, size })); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", script], + env: { ...bunEnv, FIFO: join(String(dir), "fifo"), KIND: kind, SIZE: String(size) }, + stdout: "pipe", + stderr: "pipe", + timeout: 20_000, + killSignal: "SIGKILL", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stderr, stdout: stdout.trim(), exitCode, signalCode: proc.signalCode }).toEqual({ + stderr: "", + stdout: JSON.stringify({ wrote: size, got: size, size }), + exitCode: 0, + signalCode: null, + }); + }, + 30_000, + ); +});