-
Notifications
You must be signed in to change notification settings - Fork 5k
Bun.write: deliver the full payload to a FIFO named by path #36028
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.
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
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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<JSValue> { | ||
| let destination_store = destination_blob | ||
|
|
@@ -4688,10 +4689,19 @@ 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(); | ||
|
|
||
| // Only the fast path in `write_file_internal` supplies a pre-opened fd, and | ||
| // it only runs for string/ArrayBuffer (→ Bytes) sources into a path-backed | ||
| // File destination, so the fd is consumed on the File+Bytes arm below. | ||
| debug_assert!( | ||
| dest_opened_fd == Fd::INVALID | ||
| || (destination_type == store::DataTag::File && source_type == store::DataTag::Bytes) | ||
| ); | ||
|
|
||
| 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 +4712,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 +4741,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 +5057,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,13 +5090,15 @@ pub fn write_file_internal( | |
| &pathlike, | ||
| str.get(), | ||
| &mut needs_async, | ||
| &mut dest_opened_fd, | ||
| ) | ||
| } else { | ||
| write_string_to_file_fast::<false>( | ||
| global_this, | ||
| &pathlike, | ||
| str.get(), | ||
| &mut needs_async, | ||
| &mut dest_opened_fd, | ||
| ) | ||
| }; | ||
| if !needs_async { | ||
|
|
@@ -5106,13 +5123,15 @@ pub fn write_file_internal( | |
| &pathlike, | ||
| buffer_view.byte_slice(), | ||
| &mut needs_async, | ||
| &mut dest_opened_fd, | ||
| ) | ||
| } else { | ||
| write_bytes_to_file_fast::<false>( | ||
| global_this, | ||
| &pathlike, | ||
| buffer_view.byte_slice(), | ||
| &mut needs_async, | ||
| &mut dest_opened_fd, | ||
| ) | ||
| }; | ||
| if !needs_async { | ||
|
|
@@ -5123,6 +5142,14 @@ pub fn write_file_internal( | |
| } | ||
| } | ||
|
|
||
| // If the fast path handed us an open fd and anything below throws before | ||
| // `WriteFile` adopts it, close it instead of leaking. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code |
||
| 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 +5316,7 @@ pub fn write_file_internal( | |
| global_this, | ||
| &mut *source_blob, | ||
| &mut destination_blob, | ||
| scopeguard::ScopeGuard::into_inner(dest_opened_fd), | ||
| &options, | ||
| ) | ||
| } | ||
|
|
@@ -5387,6 +5415,7 @@ fn write_string_to_file_fast<const NEEDS_OPEN: bool>( | |
| pathlike: &PathOrFileDescriptor, | ||
| str: BunString, | ||
| needs_async: &mut bool, | ||
| handoff_fd: &mut Fd, | ||
| ) -> JSValue { | ||
| let fd: Fd = if !NEEDS_OPEN { | ||
| pathlike.fd() | ||
|
|
@@ -5413,6 +5442,20 @@ fn write_string_to_file_fast<const NEEDS_OPEN: bool>( | |
| } | ||
| }; | ||
|
|
||
| // A FIFO/socket/chardev opened O_NONBLOCK will EAGAIN once the pipe buffer | ||
| // fills. Closing after a partial write delivers EOF to the reader and the | ||
| // async fallback's reopen then fails ENXIO, so hand the open fd to the | ||
| // async `WriteFile` (which owns the POLLOUT drain) instead of writing here. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code |
||
| 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 +5514,7 @@ fn write_bytes_to_file_fast<const NEEDS_OPEN: bool>( | |
| pathlike: &PathOrFileDescriptor, | ||
| bytes: &[u8], | ||
| _needs_async: &mut bool, | ||
| handoff_fd: &mut Fd, | ||
| ) -> JSValue { | ||
| let fd: Fd = if !NEEDS_OPEN { | ||
| pathlike.fd() | ||
|
|
@@ -5501,6 +5545,18 @@ fn write_bytes_to_file_fast<const NEEDS_OPEN: bool>( | |
| } | ||
| }; | ||
|
|
||
| // See `write_string_to_file_fast`: hand a non-regular destination's open | ||
| // fd to the async `WriteFile` instead of risking a torn partial write. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code |
||
| 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(); | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -276,14 +276,15 @@ 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, | ||
| ) -> Result<*mut WriteFile, Error> { | ||
| 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<C>( | ||
| 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::<c_void>(), | ||
| callback, | ||
| mkdirp_if_not_exists, | ||
|
|
@@ -338,35 +341,26 @@ 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; | ||
| true | ||
| } | ||
| bun_sys::Result::Err(err) => { | ||
| if err.get_errno() == io::RETRY { | ||
| // write(2) on a regular file never returns EAGAIN, so reaching | ||
| // this arm means the fd is pollable regardless of what the | ||
| // open-time classification guessed. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code |
||
| 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<WriteFile>, _global: &JSGlobalObject) -> Result<(), JsTerminated> { | ||
|
|
@@ -453,21 +447,26 @@ 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. | ||
|
Comment on lines
+452
to
+453
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code |
||
| break 'brk !bun_sys::is_regular_file(file.mode); | ||
| } | ||
| } | ||
| PathOrFileDescriptor::Path(_) => { | ||
| // We just opened the path with O_NONBLOCK. For a FIFO/socket/ | ||
| // character device that open succeeds and every write can | ||
| // EAGAIN, so classify from fstat instead of assuming regular. | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code |
||
| 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 +1380,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() | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -728,3 +728,64 @@ | |
| 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], | ||
|
Check warning on line 743 in test/js/bun/io/bun-write.test.js
|
||
|
Comment on lines
+738
to
+743
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🟡 The test matrix has no <256 KiB Uint8Array case, so the new fd-handoff block in Extended reasoning...What the gap isThis PR adds identical fd-handoff logic to two sibling fast-path functions:
Both blocks do the same thing: after opening a path with The new test's variant matrix at bun-write.test.js:738-743 is: 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],
])Why the second handoff block is never reachedThe ArrayBuffer fast path is gated at Blob.rs:5109: } else if let Some(buffer_view) = data.as_array_buffer(global_this) {
if buffer_view.byte_len < 256 * 1024 {
...
write_bytes_to_file_fast::<true>(...)The only Uint8Array case in the matrix is 1 MiB ( Step-by-step: what happens if the handoff block at 5548-5558 is deletedTake a hypothetical 200 KiB Uint8Array to a FIFO with a slow reader:
This is exactly the torn-write bug the PR fixes for strings — silently unfixed for the ArrayBuffer entry point if that block were removed. But no test in this PR would catch it: the only <256 KiB case is a string. Why this matters per REVIEW.mdREVIEW.md's test rules (which per the doc's own header "have blocked merges"):
Deleting the load-bearing handoff block in FixOne line in the it.each([
["200 KiB string", "string", 200 * 1024],
["200 KiB Uint8Array", "u8", 200 * 1024], // ← add: exercises write_bytes_to_file_fast handoff
["1 MiB string", "string", 1 << 20],
["1 MiB Uint8Array", "u8", 1 << 20],
["1 MiB Blob", "blob", 1 << 20],
])The comment above the matrix ("200 KiB exercises the <256 KiB sync fast path") already states the intent — it just wasn't applied to the SeverityMarking this nit: the production fix is present and correct on both code paths, so merging as-is causes no user-facing failure. The gap is purely test-coverage — a one-line matrix addition — but it does leave one of the two load-bearing hunks in this PR untested, which REVIEW.md explicitly calls out. |
||
| ["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); | ||
| }); | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If you need a paragraph-long comment to justify why the workaround is OK, the code is wrong — fix the code