diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts index 46096c814092..c6ce9a0bd904 100644 --- a/packages/bun-types/bun.d.ts +++ b/packages/bun-types/bun.d.ts @@ -1938,6 +1938,11 @@ declare module "bun" { }): void; write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number; + /** + * Write multiple chunks of data in order. + * @returns Number of bytes written + */ + writev(chunks: ArrayBufferView[]): number; /** * Flush the internal buffer. * diff --git a/packages/bun-types/s3.d.ts b/packages/bun-types/s3.d.ts index 1d1d62ec494c..5d5646c8d9e8 100644 --- a/packages/bun-types/s3.d.ts +++ b/packages/bun-types/s3.d.ts @@ -14,6 +14,15 @@ declare module "bun" { * @returns Number of bytes written or, if the write is pending, a Promise resolving to the number of bytes */ write(chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer): number | Promise; + /** + * Write multiple chunks of data to the file in order. + * + * If the file descriptor is not writable yet, the data is buffered. + * + * @param chunks The data to write + * @returns Number of bytes written or, if the write is pending, a Promise resolving to the number of bytes + */ + writev(chunks: ArrayBufferView[]): number | Promise; /** * Flush the internal buffer, committing the data to disk or the pipe. * diff --git a/src/codegen/generate-jssink.ts b/src/codegen/generate-jssink.ts index dea2a643525c..cb32068d0b8f 100644 --- a/src/codegen/generate-jssink.ts +++ b/src/codegen/generate-jssink.ts @@ -1072,7 +1072,7 @@ pub use ${rustPath} as ${name}; `; - const hostFns = ["construct", "write", "end", "flush", "start"] as const; + const hostFns = ["construct", "write", "writev", "end", "flush", "start"] as const; for (const fn of hostFns) { const sym = `${name}__${fn}`; symbols.push(sym); @@ -1180,6 +1180,7 @@ function lutInput() { end ${`${name}__end`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0 start ${`${name}__start`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 1 write ${`${name}__write`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 1 + writev ${`${name}__writev`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 1 ref ${`${name}__ref`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0 unref ${`${name}__unref`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0 _getFd ${`${name}__getFd`.padEnd(padding + 8)} ReadOnly|DontDelete|Function 0 @@ -1194,6 +1195,7 @@ function lutInput() { end ${`${controller}__end`.padEnd(protopad + 4)} ReadOnly|DontDelete|Function 0 start ${`${name}__start`.padEnd(protopad + 4)} ReadOnly|DontDelete|Function 1 write ${`${name}__write`.padEnd(protopad + 4)} ReadOnly|DontDelete|Function 1 + writev ${`${name}__writev`.padEnd(protopad + 4)} ReadOnly|DontDelete|Function 1 @end */ `; diff --git a/src/io/PipeWriter.rs b/src/io/PipeWriter.rs index 0291ff5c4e07..bd11511e6ed0 100644 --- a/src/io/PipeWriter.rs +++ b/src/io/PipeWriter.rs @@ -67,15 +67,12 @@ pub trait PosixPipeWriter { fn try_write(&self, force_sync: bool, buf: &[u8]) -> WriteResult { // PERF: try_write_with_write_fn is not monomorphized per FileType — // profile if hot. - let ft = if !force_sync { - self.get_file_type() - } else { - FileType::File - }; - match ft { - FileType::NonblockingPipe | FileType::File => { - self.try_write_with_write_fn(buf, sys::write) - } + if force_sync { + return self.try_write_with_write_fn(buf, sys::write); + } + match self.get_file_type() { + FileType::NonblockingPipe => self.try_write_with_write_fn(buf, sys::write), + FileType::File => self.try_write_with_write_fn(buf, sys::write_nonblocking), FileType::Pipe => self.try_write_with_write_fn(buf, write_to_blocking_pipe), FileType::Socket => self.try_write_with_write_fn(buf, sys::send_non_block), } @@ -604,6 +601,7 @@ pub struct PosixStreamingWriter { pub is_done: bool, pub closed_without_reporting: bool, pub force_sync: bool, + pub close_fd: bool, } impl Default for PosixStreamingWriter { @@ -615,6 +613,7 @@ impl Default for PosixStreamingWriter PosixStreamingWriter { if self.get_fd() != Fd::INVALID { debug_assert!(!self.closed_without_reporting); self.closed_without_reporting = true; - self.handle.close(None, None::); + self.handle + .close_impl(None, None::, self.close_fd); } } @@ -946,6 +946,125 @@ impl PosixStreamingWriter { rc } + fn writev_buffered(&mut self, bufs: &[&[u8]], total: usize) -> WriteResult { + if self.outgoing.ensure_unused_capacity(total).is_err() { + return WriteResult::Err(sys::Error::oom()); + } + for b in bufs { + self.outgoing.write_assume_capacity(b); + } + self.maybe_write_newly_buffered_data(total) + } + + #[cfg(unix)] + fn buffer_tail(&mut self, bufs: &[&[u8]], mut skip: usize, remaining: usize) -> Result<(), ()> { + if self.outgoing.ensure_unused_capacity(remaining).is_err() { + return Err(()); + } + for b in bufs { + if skip >= b.len() { + skip -= b.len(); + continue; + } + self.outgoing.write_assume_capacity(&b[skip..]); + skip = 0; + } + Ok(()) + } + + pub fn writev(&mut self, bufs: &[&[u8]]) -> WriteResult { + if self.is_done || self.closed_without_reporting { + return WriteResult::Done(0); + } + let mut total: usize = 0; + for b in bufs { + total += b.len(); + } + if total == 0 { + return WriteResult::Wrote(0); + } + #[cfg(not(unix))] + { + self.writev_buffered(bufs, total) + } + #[cfg(unix)] + { + const IOV_MAX: usize = 1024; + if self.outgoing.size() > 0 || self.should_buffer(total) || bufs.len() > IOV_MAX { + return self.writev_buffered(bufs, total); + } + + let fd = self.get_fd(); + if fd == Fd::INVALID { + return WriteResult::Done(0); + } + let file_type = if self.force_sync { + FileType::File + } else { + self.get_file_type() + }; + if matches!(file_type, FileType::Pipe) { + return self.writev_buffered(bufs, total); + } + + let mut iov: Vec = Vec::with_capacity(bufs.len()); + for b in bufs { + if b.is_empty() { + continue; + } + iov.push(libc::iovec { + iov_base: b.as_ptr().cast_mut().cast::(), + iov_len: b.len(), + }); + } + + let mut offset: usize = 0; + let mut start: usize = 0; + loop { + let rc = if matches!(file_type, FileType::File) && !self.force_sync { + sys::writev_nonblocking(fd, &iov[start..]) + } else { + sys::writev(fd, &iov[start..]) + }; + let wrote = match rc { + sys::Result::Err(err) => { + if err.is_retry() { + if self.buffer_tail(bufs, offset, total - offset).is_err() { + return WriteResult::Err(sys::Error::oom()); + } + self.parent_on_write(offset, WriteStatus::Pending); + Self::register_poll(self); + return WriteResult::Pending(offset); + } + return WriteResult::Err(err); + } + sys::Result::Ok(n) => n, + }; + offset += wrote; + if wrote == 0 { + self.parent_on_write(offset, WriteStatus::EndOfFile); + return WriteResult::Done(offset); + } + if offset == total { + self.parent_on_write(offset, WriteStatus::Drained); + return WriteResult::Wrote(offset); + } + let mut consumed = wrote; + while start < iov.len() && consumed >= iov[start].iov_len { + consumed -= iov[start].iov_len; + start += 1; + } + if start < iov.len() && consumed > 0 { + // SAFETY: `consumed < iov[start].iov_len`, so the advanced + // base stays within the original live buffer. + iov[start].iov_base = + unsafe { iov[start].iov_base.cast::().add(consumed) }.cast(); + iov[start].iov_len -= consumed; + } + } + } + } + pub fn flush(&mut self) -> WriteResult { if self.closed_without_reporting || self.is_done { return WriteResult::Done(0); @@ -1026,10 +1145,11 @@ impl PosixStreamingWriter { } let parent = self.parent; - self.handle.close( + self.handle.close_impl( Some(parent.cast()), // SAFETY: parent was set via set_parent with a *mut Parent. Some(|ctx: *mut c_void| unsafe { Parent::on_close(ctx.cast::()) }), + self.close_fd, ); } @@ -2443,6 +2563,62 @@ impl WindowsStreamingWriter { self.write_internal_u8(buffer, WriteKind::Bytes) } + pub fn writev(&mut self, bufs: &[&[u8]]) -> WriteResult { + if self.is_done { + return WriteResult::Done(0); + } + let mut total: usize = 0; + for b in bufs { + total += b.len(); + } + if total == 0 { + return WriteResult::Wrote(0); + } + let had_buffered_data = self.outgoing.is_not_empty(); + if self.outgoing.ensure_unused_capacity(total).is_err() { + return WriteResult::Err(sys::Error::oom()); + } + for b in bufs { + self.outgoing.write_assume_capacity(b); + } + + if matches!(self.source, Some(Source::SyncFile(_))) { + let result = (|| { + let remain = self.outgoing.slice(); + let initial_len = remain.len(); + let mut remain = remain; + let fd = Fd::from_uv(match &self.source { + Some(Source::SyncFile(f)) => f.file, + _ => unreachable!(), + }); + while remain.len() > 0 { + match sys::write(fd, remain) { + sys::Result::Err(err) => return WriteResult::Err(err), + sys::Result::Ok(wrote) => { + remain = &remain[wrote..]; + if wrote == 0 { + break; + } + } + } + } + let wrote = initial_len - remain.len(); + if wrote == 0 { + return WriteResult::Done(wrote); + } + WriteResult::Wrote(wrote) + })(); + self.outgoing.reset(); + return result; + } + + if had_buffered_data { + return WriteResult::Pending(0); + } + self.process_send(); + self.last_write_result.clone() + } + pub fn flush(&mut self) -> WriteResult { if self.is_done { return WriteResult::Done(0); @@ -2464,7 +2640,8 @@ impl WindowsStreamingWriter { self.is_done = true; if !self.has_pending_data() { - if !self.owns_fd { + if !self.owns_fd && !matches!(self.source, Some(Source::File(_) | Source::SyncFile(_))) + { return; } self.close(); diff --git a/src/io/openForWriting.rs b/src/io/openForWriting.rs index ac898e686bea..cd64a3ab5016 100644 --- a/src/io/openForWriting.rs +++ b/src/io/openForWriting.rs @@ -14,6 +14,9 @@ pub trait OpenForWritingInput { is_nonblocking: &mut bool, openat: &dyn Fn(Fd, &ZStr, i32, Mode) -> bun_sys::Result, ) -> bun_sys::Result; + fn borrowed_fd(&self) -> Option { + None + } } impl OpenForWritingInput for crate::PathOrFileDescriptor<'_> { @@ -31,7 +34,13 @@ impl OpenForWritingInput for crate::PathOrFileDescriptor<'_> { *is_nonblocking = true; bun_sys::openat_a(dir, path, input_flags, mode) } - Fd(fd_) => bun_sys::dup_with_flags(*fd_, 0), + Fd(_) => unreachable!("borrowed_fd() short-circuits Fd in open_for_writing_impl"), + } + } + fn borrowed_fd(&self) -> Option { + match self { + crate::PathOrFileDescriptor::Fd(fd) => Some(*fd), + crate::PathOrFileDescriptor::Path(_) => None, } } } @@ -112,15 +121,30 @@ where #[cfg(unix)] let mut isatty = false; let mut is_nonblocking = false; - let result = - input_path.open_for_writing_result(dir, input_flags, mode, &mut is_nonblocking, &openat); - let fd = result?; + let borrowed = input_path.borrowed_fd(); + let fd = match borrowed { + Some(fd) => fd, + None => input_path.open_for_writing_result( + dir, + input_flags, + mode, + &mut is_nonblocking, + &openat, + )?, + }; + #[cfg(windows)] + let _ = borrowed; #[cfg(unix)] { + let close_on_err = |fd: Fd| { + if borrowed.is_none() { + fd.close(); + } + }; match bun_sys::fstat(fd) { Err(err) => { - fd.close(); + close_on_err(fd); return Err(err); } Ok(stat) => { @@ -151,7 +175,7 @@ where let flags = match bun_sys::get_fcntl_flags(fd) { Ok(flags) => flags, Err(err) => { - fd.close(); + close_on_err(fd); return Err(err); } }; diff --git a/src/js/internal/fs/streams.ts b/src/js/internal/fs/streams.ts index 44e8d040fd7b..fb58a397e2ee 100644 --- a/src/js/internal/fs/streams.ts +++ b/src/js/internal/fs/streams.ts @@ -37,6 +37,7 @@ const { validateInteger, validateInt32, validateFunction } = require("internal/v const kIsPerformingIO = Symbol("kIsPerformingIO"); const kIoDone = Symbol("kIoDone"); +const kFileSink = Symbol("kFileSink"); // Bun supports a fast path for `createWriteStream("path.txt")` where instead of // using `node:fs`, `Bun.file(...).writer()` is used instead. const kWriteStreamFastPath = Symbol("kWriteStreamFastPath"); @@ -275,6 +276,15 @@ function streamConstruct(this: FSStream, callback: (e?: any) => void) { callback(err); } else { this.fd = fd; + if (this[kFileSink] === true && fs.write === write) { + try { + this[kFileSink] = Bun.file(fd).writer(); + this._write = fileSinkWrite; + this._writev = fileSinkWritev; + } catch { + this[kFileSink] = undefined; + } + } callback(); this.emit("open", this.fd); this.emit("ready"); @@ -369,6 +379,24 @@ function close(stream, err, cb) { return; } + const sink = stream[kFileSink]; + if (sink && sink !== true) { + stream[kFileSink] = undefined; + let rc; + try { + rc = sink.end(err); + } catch (sinkErr) { + return close(stream, err || sinkErr, cb); + } + if ($isPromise(rc)) { + rc.then( + () => close(stream, err, cb), + sinkErr => close(stream, err || sinkErr, cb), + ); + return; + } + } + if (!stream.fd) { cb(err); } else if (stream.flush) { @@ -449,6 +477,7 @@ function WriteStream(this: FSStream, path: string | null, options?: any): void { if (!write) this._write = null; if (!writev) this._writev = null; } else { + if (!fastPath && fd == null && start === undefined && autoClose !== false) this[kFileSink] = true; this._writev = undefined; $assert(this[kFs].write, "assuming user does not delete fs.write!"); } @@ -571,6 +600,74 @@ function writevAll(chunks, size, pos, cb, retries = 0) { }); } +function fileSinkWrite(data, encoding, cb) { + if (this.destroyed) return cb($ERR_STREAM_DESTROYED("write")); + const sink = this[kFileSink]; + let rc; + try { + rc = sink.write(data); + if (!$isPromise(rc)) rc = sink.flush(); + } catch (e) { + return cb(e); + } + if ($isPromise(rc)) { + this[kIsPerformingIO] = true; + rc.then( + () => afterFileSinkWriteSettled(this, cb, null, data.length), + err => afterFileSinkWriteSettled(this, cb, err, 0), + ); + } else { + this.bytesWritten += data.length; + process.nextTick(afterFileSinkWrite, this, cb); + } +} + +function fileSinkWritev(data, cb) { + if (this.destroyed) return cb($ERR_STREAM_DESTROYED("write")); + const len = data.length; + let size = 0; + const chunks = new Array(len); + for (let i = 0; i < len; i++) { + const chunk = data[i].chunk; + chunks[i] = chunk; + size += chunk.length; + } + const sink = this[kFileSink]; + let rc; + try { + rc = sink.writev(chunks); + if (!$isPromise(rc)) rc = sink.flush(); + } catch (e) { + return cb(e); + } + if ($isPromise(rc)) { + this[kIsPerformingIO] = true; + rc.then( + () => afterFileSinkWriteSettled(this, cb, null, size), + err => afterFileSinkWriteSettled(this, cb, err, 0), + ); + } else { + this.bytesWritten += size; + process.nextTick(afterFileSinkWrite, this, cb); + } +} + +function afterFileSinkWriteSettled(stream, cb, err, bytes) { + stream[kIsPerformingIO] = false; + if (stream.destroyed) { + cb(err || $ERR_STREAM_DESTROYED("write")); + return stream.emit(kIoDone, err); + } + if (err) return cb(err); + stream.bytesWritten += bytes; + cb(null); +} + +function afterFileSinkWrite(stream, cb) { + if (stream.destroyed) return cb($ERR_STREAM_DESTROYED("write")); + cb(null); +} + function _write(data, encoding, cb) { const fileSink = this[kWriteStreamFastPath]; diff --git a/src/jsc/bindings/headers.h b/src/jsc/bindings/headers.h index ab056ebc60bd..bd8373e57849 100644 --- a/src/jsc/bindings/headers.h +++ b/src/jsc/bindings/headers.h @@ -479,6 +479,7 @@ BUN_DECLARE_HOST_FUNCTION(ArrayBufferSink__flush); BUN_DECLARE_HOST_FUNCTION(ArrayBufferSink__start); ZIG_DECL void ArrayBufferSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(ArrayBufferSink__write); +BUN_DECLARE_HOST_FUNCTION(ArrayBufferSink__writev); #endif CPP_DECL JSC::EncodedJSValue HTTPSResponseSink__assignToStream(JSC::JSGlobalObject* arg0, JSC::EncodedJSValue JSValue1, void* arg2, void** arg3); @@ -498,6 +499,7 @@ BUN_DECLARE_HOST_FUNCTION(HTTPSResponseSink__flush); BUN_DECLARE_HOST_FUNCTION(HTTPSResponseSink__start); ZIG_DECL void HTTPSResponseSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(HTTPSResponseSink__write); +BUN_DECLARE_HOST_FUNCTION(HTTPSResponseSink__writev); #endif CPP_DECL JSC::EncodedJSValue HTTPResponseSink__assignToStream(JSC::JSGlobalObject* arg0, JSC::EncodedJSValue JSValue1, void* arg2, void** arg3); @@ -517,6 +519,7 @@ BUN_DECLARE_HOST_FUNCTION(HTTPResponseSink__flush); BUN_DECLARE_HOST_FUNCTION(HTTPResponseSink__start); ZIG_DECL void HTTPResponseSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(HTTPResponseSink__write); +BUN_DECLARE_HOST_FUNCTION(HTTPResponseSink__writev); #endif CPP_DECL JSC::EncodedJSValue FileSink__assignToStream(JSC::JSGlobalObject* arg0, JSC::EncodedJSValue JSValue1, void* arg2, void** arg3); @@ -536,6 +539,7 @@ BUN_DECLARE_HOST_FUNCTION(FileSink__flush); BUN_DECLARE_HOST_FUNCTION(FileSink__start); ZIG_DECL void FileSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(FileSink__write); +BUN_DECLARE_HOST_FUNCTION(FileSink__writev); #endif @@ -556,6 +560,7 @@ BUN_DECLARE_HOST_FUNCTION(FileSink__flush); BUN_DECLARE_HOST_FUNCTION(FileSink__start); ZIG_DECL void FileSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(FileSink__write); +BUN_DECLARE_HOST_FUNCTION(FileSink__writev); #endif CPP_DECL JSC::EncodedJSValue NetworkSink__assignToStream(JSC::JSGlobalObject* arg0, JSC::EncodedJSValue JSValue1, void* arg2, void** arg3); @@ -576,6 +581,7 @@ BUN_DECLARE_HOST_FUNCTION(NetworkSink__flush); BUN_DECLARE_HOST_FUNCTION(NetworkSink__start); ZIG_DECL void NetworkSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(NetworkSink__write); +BUN_DECLARE_HOST_FUNCTION(NetworkSink__writev); #endif CPP_DECL JSC::EncodedJSValue H3ResponseSink__assignToStream(JSC::JSGlobalObject* arg0, JSC::EncodedJSValue JSValue1, void* arg2, void** arg3); @@ -596,6 +602,7 @@ BUN_DECLARE_HOST_FUNCTION(H3ResponseSink__flush); BUN_DECLARE_HOST_FUNCTION(H3ResponseSink__start); ZIG_DECL void H3ResponseSink__updateRef(void* arg0, bool arg1); BUN_DECLARE_HOST_FUNCTION(H3ResponseSink__write); +BUN_DECLARE_HOST_FUNCTION(H3ResponseSink__writev); #endif #ifdef __cplusplus diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index ec3d76a55648..8e503d8dc86b 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -1816,6 +1816,7 @@ impl BlobExt for Blob { #[cfg(windows)] { use bun_io::pipe_writer::BaseWindowsPipeWriter as _; + use bun_sys::FdExt as _; let pathlike = &store.data.as_file().pathlike; // SAFETY: bun_vm() never returns null for a Bun-owned global. @@ -1857,6 +1858,30 @@ impl BlobExt for Blob { ) }; + let borrowed = matches!(pathlike, PathOrFileDescriptor::Fd(_)); + // uv_pipe_open adopts the handle; dup so the caller keeps their fd. + let (writer_fd, owns_fd) = if borrowed + && !is_stdout_or_stderr + && matches!( + bun_sys::windows::libuv::uv_guess_handle(fd.uv()), + bun_sys::windows::libuv::HandleType::NamedPipe + | bun_sys::windows::libuv::HandleType::Tty + ) { + match bun_sys::dup(fd).and_then(|d| { + d.make_lib_uv_owned_for_syscall( + bun_sys::Tag::dup, + bun_sys::ErrorCase::CloseOnFail, + ) + }) { + bun_sys::Result::Ok(dup) => (dup, true), + bun_sys::Result::Err(err) => { + return Err(global_this.throw_value(err.to_js(global_this))); + } + } + } else { + (fd, !borrowed) + }; + let sink = webcore::FileSink::init( fd, jsc::EventLoopHandle::init( @@ -1869,18 +1894,19 @@ impl BlobExt for Blob { ); // SAFETY: `init` returns a freshly-allocated +1 *mut FileSink; sole owner here. let sink_mut = unsafe { &mut *sink }; - sink_mut - .writer - .with_mut(|w| w.owns_fd = !matches!(pathlike, PathOrFileDescriptor::Fd(_))); + sink_mut.writer.with_mut(|w| w.owns_fd = owns_fd); let start_result = sink_mut.writer.with_mut(|w| { if is_stdout_or_stderr { - w.start_sync(fd, false) + w.start_sync(writer_fd, false) } else { - w.start(fd, true) + w.start(writer_fd, true) } }); if let bun_sys::Result::Err(err) = start_result { + if owns_fd { + writer_fd.close(); + } // SAFETY: release the +1 ref from `init`. unsafe { webcore::FileSink::deref(sink) }; return Err(global_this.throw_value(err.to_js(global_this))); diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index f0a207de17bf..6e80297477ea 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -6,7 +6,9 @@ use core::sync::atomic::{AtomicI32, Ordering}; use bun_io::pipe_writer::BaseWindowsPipeWriter as _; use bun_io::{self, WriteResult, WriteStatus}; use bun_jsc::JsCell; -use bun_sys::{self as sys, Fd, FdExt as _}; +#[cfg(windows)] +use bun_sys::FdExt as _; +use bun_sys::{self as sys, Fd}; use crate::api::bun::process::Status as SpawnStatus; use crate::webcore::jsc::{CallFrame, EventLoopHandle, JSGlobalObject, JSValue, JsResult}; @@ -644,6 +646,37 @@ impl FileSink { sys::Result::Ok(fd) => fd, }; + let borrowed = matches!(&options.input_path, PathOrFileDescriptor::Fd(_)); + #[allow(unused_mut)] + let mut owns_fd = !borrowed; + self.fd.set(fd); + // Pollable fds need a per-sink epoll entry; dup those, adopt the rest. + #[cfg(unix)] + let fd = if borrowed && self.pollable.get() { + let dup = bun_sys::dup_with_flags(fd, 0)?; + owns_fd = true; + dup + } else { + fd + }; + // uv_pipe_open adopts the handle; dup so the caller keeps their fd. + #[cfg(windows)] + let fd = if borrowed + && !self.force_sync.get() + && matches!( + uv::uv_guess_handle(fd.uv()), + uv::HandleType::NamedPipe | uv::HandleType::Tty + ) { + use bun_sys::FdExt as _; + let dup = bun_sys::dup(fd)?.make_lib_uv_owned_for_syscall( + bun_sys::Tag::dup, + bun_sys::ErrorCase::CloseOnFail, + )?; + owns_fd = true; + dup + } else { + fd + }; #[cfg(windows)] { if self.force_sync.get() { @@ -653,10 +686,13 @@ impl FileSink { .with_mut(|w| w.start_sync(fd, self.pollable.get())) { sys::Result::Err(err) => { - fd.close(); + if owns_fd { + fd.close(); + } return sys::Result::Err(err); } sys::Result::Ok(()) => { + self.writer.with_mut(|w| w.owns_fd = owns_fd); self.writer .with_mut(|w| w.update_ref(self.io_evtloop(), false)); } @@ -668,10 +704,21 @@ impl FileSink { // SAFETY(JsCell): `start` is pure I/O setup; no JS. match self.writer.with_mut(|w| w.start(fd, self.pollable.get())) { sys::Result::Err(err) => { - fd.close(); + // POSIX start() may have set handle; let Drop own the close. + #[cfg(unix)] + self.writer.with_mut(|w| w.close_fd = owns_fd); + // Windows start() leaves source = None on failure; close here. + #[cfg(windows)] + if owns_fd { + fd.close(); + } return sys::Result::Err(err); } sys::Result::Ok(()) => { + #[cfg(unix)] + self.writer.with_mut(|w| w.close_fd = owns_fd); + #[cfg(windows)] + self.writer.with_mut(|w| w.owns_fd = owns_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 @@ -816,7 +863,7 @@ impl FileSink { pub unsafe fn on_auto_flush(this: *mut FileSink) -> bool { // SAFETY: caller contract — `this` is live with write+dealloc provenance. unsafe { - if (*this).done.get() || !(*this).writer.get().has_pending_data() { + if !(*this).writer.get().has_pending_data() { (*this).update_ref(false); (*this).auto_flusher.with_mut(|a| a.registered.set(false)); return false; @@ -824,6 +871,7 @@ impl FileSink { let _guard = FileSinkRef::new_ref(this); + let done = (*this).done.get(); let amount_buffered = (*this).writer.get().outgoing.size(); // SAFETY(JsCell): `IOWriter::flush` is pure I/O; the `on_write` @@ -850,11 +898,17 @@ impl FileSink { } WriteResult::Done(_) => { (*this).update_ref(false); + if done { + (*this).writer.with_mut(|w| w.end()); + } (*this).run_pending_later(); } WriteResult::Wrote(amount_drained) => { if amount_drained == amount_buffered { (*this).update_ref(false); + if done { + (*this).writer.with_mut(|w| w.end()); + } (*this).run_pending_later(); } } @@ -1017,6 +1071,16 @@ impl FileSink { self.write(data) } + pub fn writev_bytes(&self, bufs: &[&[u8]]) -> streams::Writable { + if self.done.get() { + return streams::Writable::Done; + } + let buffered_before = self.writer.get().buffered_len(); + let rc = self.writer.with_mut(|w| w.writev(bufs)); + let accepted = self.bytes_accepted(buffered_before, &rc); + self.to_result(rc, accepted) + } + pub fn write_latin1(&self, data: &streams::Result) -> streams::Writable { if self.done.get() { return streams::Writable::Done; @@ -1274,6 +1338,9 @@ impl crate::webcore::sink::JsSinkType for FileSink { fn write_bytes(&mut self, data: &streams::Result) -> streams::result::Writable { Self::write(self, data) } + fn writev_bytes(&mut self, bufs: &[&[u8]]) -> streams::result::Writable { + Self::writev_bytes(self, bufs) + } fn write_utf16(&mut self, data: &streams::Result) -> streams::result::Writable { Self::write_utf16(self, data) } diff --git a/src/runtime/webcore/Sink.rs b/src/runtime/webcore/Sink.rs index 2d33b1e0b7f6..ab040b01e68d 100644 --- a/src/runtime/webcore/Sink.rs +++ b/src/runtime/webcore/Sink.rs @@ -330,6 +330,34 @@ pub trait JsSinkType: Sized { fn write_bytes(&mut self, data: &streams::Result) -> streams::result::Writable; fn write_utf16(&mut self, data: &streams::Result) -> streams::result::Writable; fn write_latin1(&mut self, data: &streams::Result) -> streams::result::Writable; + fn writev_bytes(&mut self, bufs: &[&[u8]]) -> streams::result::Writable { + use streams::result::Writable; + let mut total: u64 = 0; + let mut backpressure = false; + for b in bufs { + if b.is_empty() { + continue; + } + let data = bun_ptr::RawSlice::new(b); + match self.write_bytes(&streams::Result::Temporary(data)) { + Writable::Owned(n) | Writable::Temporary(n) => total += n, + Writable::Backpressure(n) => { + total += n; + backpressure = true; + } + Writable::OwnedAndDone(n) | Writable::TemporaryAndDone(n) => { + return Writable::OwnedAndDone(total + n); + } + Writable::Done => return Writable::OwnedAndDone(total), + other => return other, + } + } + if backpressure { + Writable::Backpressure(total) + } else { + Writable::Owned(total) + } + } fn end(&mut self, err: Option) -> sys::Result<()>; fn end_from_js(&mut self, global: &JSGlobalObject) -> sys::Result; fn flush(&mut self) -> sys::Result<()>; @@ -509,6 +537,66 @@ impl JSSink { .to_js(global)) } + /// `${abi_name}__writev` host-fn body. + pub fn js_writev( + global: &crate::webcore::jsc::JSGlobalObject, + frame: &crate::webcore::jsc::CallFrame, + ) -> crate::webcore::jsc::JsResult { + use crate::webcore::jsc::JSValue; + bun_core::mark_binding!(); + + let arg = frame.argument(0); + arg.ensure_still_alive(); + let _keep = bun_jsc::EnsureStillAlive(arg); + if !arg.is_array() { + return Err(global.throw_value(global.to_type_error( + bun_jsc::ErrorCode::INVALID_ARG_TYPE, + format_args!("writev() expects an array of ArrayBufferView"), + ))); + } + + let len = arg.get_length(global)? as usize; + if len == 0 { + return Ok(JSValue::js_number(0.0)); + } + const MAX_CHUNKS: usize = 1 << 20; + if len > MAX_CHUNKS { + return Err(global.throw_value(global.to_type_error( + bun_jsc::ErrorCode::OUT_OF_RANGE, + format_args!("writev() chunk count {} exceeds {}", len, MAX_CHUNKS), + ))); + } + + bun_jsc::MarkedArgumentBuffer::new(|roots| { + let mut items: Vec = Vec::with_capacity(len); + for i in 0..len { + let item = arg.get_index(global, i as u32)?; + roots.append(item); + items.push(item); + } + let mut slices: Vec<&[u8]> = Vec::with_capacity(len); + for item in &items { + let Some(buffer) = item.as_array_buffer(global) else { + return Err(global.throw_value(global.to_type_error( + bun_jsc::ErrorCode::INVALID_ARG_TYPE, + format_args!("writev() expects an array of ArrayBufferView"), + ))); + }; + let slice = buffer.slice(); + // SAFETY: `roots` keeps every cell GC-live and no further user + // JS runs between here and `writev_bytes`, so the backing + // store cannot be detached out from under the slice. + slices.push(unsafe { core::slice::from_raw_parts(slice.as_ptr(), slice.len()) }); + } + // Acquire `&mut sink` only after all accessor JS has run. + let this = Self::get_this(global, frame)?; + if let Some(err) = this.sink.get_pending_error() { + return Err(global.throw_value(err)); + } + Ok(this.sink.writev_bytes(&slices).to_js(global)) + }) + } + /// `${abi_name}__flush` host-fn body. pub fn js_flush( global: &crate::webcore::jsc::JSGlobalObject, diff --git a/src/sys/lib.rs b/src/sys/lib.rs index 90d4cdbac33f..04c3fe11ce63 100644 --- a/src/sys/lib.rs +++ b/src/sys/lib.rs @@ -7486,6 +7486,36 @@ pub fn read_nonblocking(fd: Fd, buf: &mut [u8]) -> Maybe { } read(fd, buf) } +/// Linux: `pwritev2(iov, -1, RWF_NOWAIT)`; else plain `writev`. +pub fn writev_nonblocking(fd: Fd, vecs: &[PlatformIoVec]) -> Maybe { + #[cfg(any(target_os = "linux", target_os = "android"))] + while linux::RWFFlagSupport::is_maybe_supported() { + // SAFETY: fd valid; vecs is a live slice of iovec. + let rc = unsafe { + sys_pwritev2( + fd.native(), + vecs.as_ptr(), + vecs.len() as c_int, + -1, + RWF_NOWAIT, + ) + }; + if rc < 0 { + let e = last_errno(); + match e { + libc::EOPNOTSUPP | libc::ENOSYS | libc::EPERM | libc::EACCES => { + linux::RWFFlagSupport::disable(); + break; + } + libc::EINTR => continue, + _ => return Err(Error::from_code_int(e, Tag::writev).with_fd(fd)), + } + } + return Ok(rc as usize); + } + writev(fd, vecs) +} + /// Linux: `pwritev2(.., RWF_NOWAIT)`; else plain `write`. pub fn write_nonblocking(fd: Fd, buf: &[u8]) -> Maybe { #[cfg(any(target_os = "linux", target_os = "android"))] diff --git a/test/js/bun/util/filesink.test.ts b/test/js/bun/util/filesink.test.ts index e78190256084..77d908315bfd 100644 --- a/test/js/bun/util/filesink.test.ts +++ b/test/js/bun/util/filesink.test.ts @@ -165,9 +165,65 @@ describe("FileSink", () => { }); }); } + + it("writev writes each chunk in order", async () => { + const path = join(tmpdirSync(), "writev.txt"); + const sink = Bun.file(path).writer(); + const rc = sink.writev([ + Buffer.from("one "), + new Uint8Array([0x74, 0x77, 0x6f, 0x20]), + new TextEncoder().encode("three"), + ]); + expect(typeof rc === "number" || rc instanceof Promise).toBe(true); + await sink.end(); + expect(await Bun.file(path).text()).toBe("one two three"); + }); + + it("writev rejects non-ArrayBufferView entries", async () => { + const path = join(tmpdirSync(), "writev-bad.txt"); + const sink = Bun.file(path).writer(); + try { + expect(() => sink.writev(["string" as any])).toThrow(expect.objectContaining({ code: "ERR_INVALID_ARG_TYPE" })); + expect(() => (sink as any).writev("not an array")).toThrow( + expect.objectContaining({ code: "ERR_INVALID_ARG_TYPE" }), + ); + } finally { + await sink.end(); + } + }); + + it("writev does not dereference a buffer detached by an accessor getter", async () => { + const path = join(tmpdirSync(), "writev-detach.txt"); + const sink = Bun.file(path).writer(); + const ab = new ArrayBuffer(64); + const u8 = new Uint8Array(ab); + const arr: any[] = [u8, null]; + Object.defineProperty(arr, 1, { + get() { + (ab as any).transfer(); + return new Uint8Array([0x6f, 0x6b]); + }, + }); + // The first element is validated after all getters have run, so it is + // seen as detached (byteLength 0) and contributes no bytes; the second + // element's two bytes are written. + sink.writev(arr); + await sink.end(); + expect(await Bun.file(path).text()).toBe("ok"); + }); + + it("writev on ArrayBufferSink falls through to write()", () => { + const s = new Bun.ArrayBufferSink(); + s.start(); + s.writev([Buffer.from("hello"), Buffer.from(" "), Buffer.from("world")]); + const out = s.end(); + expect(new TextDecoder().decode(out)).toBe("hello world"); + }); }); +import { once } from "node:events"; import fs from "node:fs"; +import net from "node:net"; import path from "node:path"; import util from "node:util"; @@ -472,6 +528,43 @@ if (isWindows) { }), ); }); + + it("Bun.file(fd).writer() on a named pipe dups so end() releases the sink", async () => { + const pipePath = `\\\\.\\pipe\\bun-filesink-${process.pid}-${Date.now()}`; + const { promise: gotData, resolve: onData, reject: onErr } = Promise.withResolvers(); + const server = net.createServer(c => { + c.once("data", d => onData(String(d))); + c.once("error", onErr); + }); + server.once("error", onErr); + server.listen(pipePath); + await once(server, "listening"); + + const fd = fs.openSync(pipePath, "w"); + try { + const baseline = fileSinkInternals.liveCount(); + const sink = Bun.file(fd).writer(); + const rc = sink.write(Buffer.from("hello")); + if (rc instanceof Promise) await rc; + await sink.end(); + + // end() must not close the caller's fd: fstat still works. + expect(() => fs.fstatSync(fd)).not.toThrow(); + expect(await gotData).toBe("hello"); + + for (let i = 0; i < 50; i++) { + Bun.gc(true); + if (fileSinkInternals.liveCount() <= baseline) break; + await Bun.sleep(10); + } + expect(fileSinkInternals.liveCount()).toBeLessThanOrEqual(baseline); + } finally { + try { + fs.closeSync(fd); + } catch {} + await new Promise(r => server.close(() => r())); + } + }); } // When a write to a pollable fd returns `.pending`, FileSink takes a diff --git a/test/js/node/fs/fs.test.ts b/test/js/node/fs/fs.test.ts index 16396f0dd211..d56b750c8290 100644 --- a/test/js/node/fs/fs.test.ts +++ b/test/js/node/fs/fs.test.ts @@ -14,6 +14,7 @@ import { tempDirWithFiles, tmpdirSync, } from "harness"; +import { once } from "node:events"; import fs, { closeSync, constants, @@ -3671,6 +3672,138 @@ describe("createWriteStream", () => { } }); }); + + async function writeLines(ws: fs.WriteStream, n: number, line: string | Buffer) { + for (let i = 0; i < n; i++) { + if (!ws.write(line)) await once(ws, "drain"); + } + await new Promise((resolve, reject) => ws.end(err => (err ? reject(err) : resolve()))); + await once(ws, "close"); + } + + it.skipIf(!isLinux)("coalesces many small writes instead of dispatching one syscall per chunk", async () => { + const syscw = () => +readFileSync("/proc/self/io", "utf8").match(/syscw: (\d+)/)![1]; + const N = 5000; + const line = Buffer.alloc(81, "x"); + using dir = tempDir("ws-coalesce", {}); + const streamPath = join(String(dir), "out.txt"); + + const ws = createWriteStream(streamPath); + await once(ws, "ready"); + const fd = ws.fd as number; + expect(fd).toBeGreaterThan(0); + + const before = syscw(); + await writeLines(ws, N, line); + const writeSyscalls = syscw() - before; + + expect({ + size: statSync(streamPath).size, + bytesWritten: ws.bytesWritten, + fdClosed: (() => { + try { + fstatSync(fd); + return false; + } catch { + return true; + } + })(), + }).toEqual({ size: N * line.length, bytesWritten: N * line.length, fdClosed: true }); + + // Without coalescing each chunk is a separate thread-pool dispatch (one + // write(2) to the file plus one 8-byte eventfd wake), so 5000 chunks is + // ~10000 write syscalls. Coalescing brings it well under N/4. + expect(writeSyscalls).toBeLessThan(N / 4); + }); + + it.skipIf(!isLinux)("holds a single fd for the stream's lifetime", async () => { + using dir = tempDir("ws-fd-count", {}); + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const fs = require("node:fs"); + const { once } = require("node:events"); + const countFds = () => fs.readdirSync("/proc/self/fd").length; + const before = countFds(); + const ws = fs.createWriteStream(process.argv[1]); + await once(ws, "ready"); + const open = countFds() - before; + ws.write("x"); + await new Promise((res, rej) => ws.end(e => (e ? rej(e) : res()))); + await once(ws, "close"); + console.log(JSON.stringify({ open, closed: countFds() - before }));`, + join(String(dir), "out.txt"), + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ open: 1, closed: 0 }); + expect(exitCode).toBe(0); + }); + + it("many small writes produce byte-exact output and bytesWritten", async () => { + const N = isDebug ? 2000 : 10000; + const chunk = "\u00e9#"; // 2-byte UTF-8 char + 1 ASCII byte => 3 bytes/chunk + const byteLen = Buffer.byteLength(chunk); + using dir = tempDir("ws-bytes", {}); + const streamPath = join(String(dir), "out.txt"); + + const ws = createWriteStream(streamPath); + await writeLines(ws, N, chunk); + + expect({ + size: statSync(streamPath).size, + bytesWritten: ws.bytesWritten, + head: readFileSync(streamPath, "utf8").slice(0, chunk.length * 3), + }).toEqual({ size: N * byteLen, bytesWritten: N * byteLen, head: chunk + chunk + chunk }); + }); + + it.skipIf(!isLinux)("surfaces a synchronous write(2) failure via the stream's 'error' event", async () => { + const ws = createWriteStream("/dev/full"); + await once(ws, "ready"); + let unhandled = 0; + const onUnhandled = () => unhandled++; + process.on("unhandledRejection", onUnhandled); + try { + const errorPromise = once(ws, "error"); + ws.write(Buffer.alloc(8192)); + const [err] = (await errorPromise) as [NodeJS.ErrnoException]; + await new Promise(r => setImmediate(r)); + expect({ code: err.code, bytesWritten: ws.bytesWritten, unhandled }).toEqual({ + code: "ENOSPC", + bytesWritten: 0, + unhandled: 0, + }); + } finally { + process.off("unhandledRejection", onUnhandled); + } + }); + + it("routes writes through fs.write so a monkey-patch is honoured", async () => { + using dir = tempDir("ws-patch", {}); + const streamPath = join(String(dir), "out.txt"); + const ws = createWriteStream(streamPath); + const original = fs.write; + let calls = 0; + // @ts-ignore + fs.write = function () { + calls++; + return original.apply(fs, arguments); + }; + try { + ws.write("hello"); + ws.write(" world"); + await new Promise((resolve, reject) => ws.end(err => (err ? reject(err) : resolve()))); + await once(ws, "close"); + } finally { + fs.write = original; + } + expect({ calls, contents: readFileSync(streamPath, "utf8") }).toEqual({ calls: 2, contents: "hello world" }); + }); }); describe("fs/promises", () => {