Skip to content
Open
9 changes: 5 additions & 4 deletions src/js/internal/fs/streams.ts
Original file line number Diff line number Diff line change
Expand Up @@ -449,7 +449,6 @@ function WriteStream(this: FSStream, path: string | null, options?: any): void {
if (!write) this._write = null;
if (!writev) this._writev = null;
} else {
this._writev = undefined;
$assert(this[kFs].write, "assuming user does not delete fs.write!");
}

Expand Down Expand Up @@ -527,7 +526,8 @@ function writeAll(data, size, pos, cb, retries = 0) {

retries = bytesWritten ? 0 : retries + 1;
size -= bytesWritten;
pos += bytesWritten;
// An undefined `pos` means "current file offset"; adding would make it NaN.
if (pos !== undefined) pos += bytesWritten;

// Try writing non-zero number of bytes up to 5 times.
if (retries > 5) {
Expand All @@ -542,7 +542,7 @@ function writeAll(data, size, pos, cb, retries = 0) {
}

function writevAll(chunks, size, pos, cb, retries = 0) {
this[kFs].writev(this.fd, chunks, this.pos, (er, bytesWritten, buffers) => {
this[kFs].writev(this.fd, chunks, pos, (er, bytesWritten, buffers) => {
Comment thread
robobun marked this conversation as resolved.
// No data currently available and operation should be retried later.
if (er?.code === "EAGAIN") {
er = null;
Expand All @@ -557,7 +557,8 @@ function writevAll(chunks, size, pos, cb, retries = 0) {

retries = bytesWritten ? 0 : retries + 1;
size -= bytesWritten;
pos += bytesWritten;
// An undefined `pos` means "current file offset"; adding would make it NaN.
if (pos !== undefined) pos += bytesWritten;

// Try writing non-zero number of bytes up to 5 times.
if (retries > 5) {
Expand Down
242 changes: 168 additions & 74 deletions src/sys/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4567,65 +4567,115 @@ pub fn platform_iovec_const_create(buf: &[u8]) -> PlatformIoVecConst {
}
}

/// `IOV_MAX`/`UIO_MAXIOV` on every POSIX target Bun supports. The kernel
/// rejects larger iovec arrays with `EINVAL`; `writev`/`pwritev` batch and
/// loop, `readv`/`preadv` cap at one batch (short read, matching libuv).
Comment thread
robobun marked this conversation as resolved.
Outdated
#[cfg(unix)]
const POSIX_IOV_MAX: usize = 1024;

/// Single `pwritev(2)` call — no IOV_MAX batching. Callers that may exceed
/// `POSIX_IOV_MAX` should use [`pwritev`].
Comment thread
robobun marked this conversation as resolved.
Outdated
#[cfg(unix)]
#[inline]
fn pwritev_one(fd: Fd, vecs: &[PlatformIoVecConst], offset: i64) -> Maybe<usize> {
#[cfg(target_os = "macos")]
{
// SAFETY: `PlatformIoVecConst` is layout-compatible with `libc::iovec`
// (asserted above). `pwritev$NOCANCEL`: single shot, surfaces EINTR.
let rc = unsafe {
nocancel::pwritev(
fd.native(),
vecs.as_ptr().cast::<libc::iovec>(),
vecs.len() as core::ffi::c_int,
offset,
)
};
if rc < 0 {
return Err(Error::from_code_int(last_errno(), Tag::pwritev));
}
Ok(rc as usize)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
{
// SAFETY: `PlatformIoVecConst` is layout-identical to `libc::iovec`.
unsafe {
linux_syscall::pwritev(fd, vecs.as_ptr().cast::<libc::iovec>(), vecs.len(), offset)
}
.map_err(|e| Error::from_code_int(e, Tag::pwritev))
}
#[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android")))]
loop {
let rc = unsafe {
libc::pwritev(
fd.native(),
vecs.as_ptr().cast::<libc::iovec>(),
vecs.len() as core::ffi::c_int,
offset,
)
};
if rc < 0 {
let e = last_errno();
if e == libc::EINTR {
continue;
}
return Err(Error::from_code_int(e, Tag::pwritev));
}
return Ok(rc as usize);
}
}

/// `bun.sys.pwritev` — gather-write at `offset`. Returns bytes written
/// (may be less than the sum of `vecs` lengths on a short write).
/// Batches iovecs past `POSIX_IOV_MAX`; a short write stops the loop.
pub fn pwritev(fd: Fd, vecs: &[PlatformIoVecConst], offset: i64) -> Maybe<usize> {
#[cfg(unix)]
{
// SAFETY: `PlatformIoVecConst` is layout-compatible with `libc::iovec`
// (asserted above); `pwritev(2)` only reads through `iov_base`.
// Darwin uses `pwritev$NOCANCEL` (avoid cancellation point).
#[cfg(target_os = "macos")]
{
// macOS: single `pwritev$NOCANCEL`, no
// EINTR retry (surfaces EINTR to caller).
// SAFETY: `fd` is a live descriptor; `vecs` gives an exact
// (ptr, len) pair of layout-compatible iovecs (asserted above).
let rc = unsafe {
nocancel::pwritev(
fd.native(),
vecs.as_ptr().cast::<libc::iovec>(),
vecs.len() as core::ffi::c_int,
offset,
)
if vecs.len() <= POSIX_IOV_MAX {
return pwritev_one(fd, vecs, offset);
}

let mut total_written: usize = 0;
let mut remaining = vecs;
let mut position = offset;

while !remaining.is_empty() {
let chunk_len = remaining.len().min(POSIX_IOV_MAX);
let chunk = &remaining[..chunk_len];
let chunk_capacity: usize = chunk.iter().map(|v| v.len).sum();

let bytes_written = match pwritev_one(fd, chunk, position) {
Ok(n) => n,
// Surface the error only if nothing has been written yet;
// otherwise report the partial progress so the caller retries.
Comment thread
robobun marked this conversation as resolved.
Outdated
Err(e) => {
if total_written == 0 {
return Err(e);
}
break;
}
};
if rc < 0 {
return Err(Error::from_code_int(last_errno(), Tag::pwritev));
}
return Ok(rc as usize);
}
#[cfg(any(target_os = "linux", target_os = "android"))]
{
// SAFETY: `PlatformIoVecConst` is layout-identical to `libc::iovec`.
return unsafe {
linux_syscall::pwritev(fd, vecs.as_ptr().cast::<libc::iovec>(), vecs.len(), offset)
total_written += bytes_written;

// Short write: stop so the caller resumes from the right offset.
if bytes_written < chunk_capacity {
break;
}
.map_err(|e| Error::from_code_int(e, Tag::pwritev));
}
#[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android")))]
loop {
let rc = unsafe {
libc::pwritev(
fd.native(),
vecs.as_ptr().cast::<libc::iovec>(),
vecs.len() as core::ffi::c_int,
offset,
)
};
if rc < 0 {
let e = last_errno();
if e == libc::EINTR {
continue;
}
return Err(Error::from_code_int(e, Tag::pwritev));

remaining = &remaining[chunk_len..];
// Negative `offset` means "current file offset"; keep the
// sentinel across batches (matches `sys_uv::pwritev`).
Comment thread
robobun marked this conversation as resolved.
Outdated
if position >= 0 {
position += bytes_written as i64;
}
return Ok(rc as usize);
}

Ok(total_written)
}
#[cfg(windows)]
{
// `PlatformIoVecConst` is layout-identical to `uv_buf_t` on Windows
// (asserted below), so the slice forwards as-is.
// (asserted below), so the slice forwards as-is. `sys_uv::pwritev`
// already batches by `MAX_IOVEC_COUNT`.
Comment thread
robobun marked this conversation as resolved.
Outdated
sys_uv::pwritev(fd, vecs, offset)
}
}
Expand Down Expand Up @@ -4709,38 +4759,39 @@ pub fn platform_iovec_const_create(buf: &[u8]) -> PlatformIoVecConst {
pub fn writev(fd: Fd, vecs: &[PlatformIoVec]) -> Maybe<usize> {
#[cfg(unix)]
{
#[cfg(target_os = "macos")]
{
// SAFETY: `PlatformIoVec` is `libc::iovec`; writev(2) only reads
// the descriptor table. Single shot, surfaces EINTR.
let rc = unsafe {
nocancel::writev(fd.native(), vecs.as_ptr(), vecs.len() as core::ffi::c_int)
};
if rc < 0 {
return Err(Error::from_code_int(last_errno(), Tag::writev).with_fd(fd));
}
return Ok(rc as usize);
}
#[cfg(any(target_os = "linux", target_os = "android"))]
{
// SAFETY: `PlatformIoVec` is `libc::iovec`.
return unsafe { linux_syscall::writev(fd, vecs.as_ptr(), vecs.len()) }
.map_err(|e| Error::from_code_int(e, Tag::writev).with_fd(fd));
if vecs.len() <= POSIX_IOV_MAX {
return writev_one(fd, vecs);
}
#[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android")))]
loop {
// SAFETY: see above.
let rc =
unsafe { libc::writev(fd.native(), vecs.as_ptr(), vecs.len() as core::ffi::c_int) };
if rc < 0 {
let e = last_errno();
if e == libc::EINTR {
continue;

// Batch past the kernel's IOV_MAX limit (matching libuv); a short
// write stops the loop to preserve write ordering.
Comment thread
robobun marked this conversation as resolved.
Outdated
let mut total_written: usize = 0;
let mut remaining = vecs;

while !remaining.is_empty() {
let chunk_len = remaining.len().min(POSIX_IOV_MAX);
let chunk = &remaining[..chunk_len];
let chunk_capacity: usize = chunk.iter().map(|v| v.iov_len).sum();

let bytes_written = match writev_one(fd, chunk) {
Ok(n) => n,
Err(e) => {
if total_written == 0 {
return Err(e);
}
break;
}
return Err(Error::from_code_int(e, Tag::writev).with_fd(fd));
};
total_written += bytes_written;

if bytes_written < chunk_capacity {
break;
}
return Ok(rc as usize);

remaining = &remaining[chunk_len..];
}

Ok(total_written)
}
#[cfg(not(unix))]
{
Expand All @@ -4750,6 +4801,44 @@ pub fn writev(fd: Fd, vecs: &[PlatformIoVec]) -> Maybe<usize> {
}
}

/// Single `writev(2)` call — no IOV_MAX batching. Callers that may exceed
/// `POSIX_IOV_MAX` should use [`writev`].
Comment thread
robobun marked this conversation as resolved.
Outdated
#[cfg(unix)]
#[inline]
fn writev_one(fd: Fd, vecs: &[PlatformIoVec]) -> Maybe<usize> {
#[cfg(target_os = "macos")]
{
// SAFETY: `PlatformIoVec` is `libc::iovec`; writev(2) only reads
// the descriptor table. Single shot, surfaces EINTR.
let rc =
unsafe { nocancel::writev(fd.native(), vecs.as_ptr(), vecs.len() as core::ffi::c_int) };
if rc < 0 {
return Err(Error::from_code_int(last_errno(), Tag::writev).with_fd(fd));
}
Ok(rc as usize)
}
#[cfg(any(target_os = "linux", target_os = "android"))]
{
// SAFETY: `PlatformIoVec` is `libc::iovec`.
unsafe { linux_syscall::writev(fd, vecs.as_ptr(), vecs.len()) }
.map_err(|e| Error::from_code_int(e, Tag::writev).with_fd(fd))
}
#[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android")))]
loop {
// SAFETY: see above.
let rc =
unsafe { libc::writev(fd.native(), vecs.as_ptr(), vecs.len() as core::ffi::c_int) };
if rc < 0 {
let e = last_errno();
if e == libc::EINTR {
continue;
}
return Err(Error::from_code_int(e, Tag::writev).with_fd(fd));
}
return Ok(rc as usize);
}
}

/// `bun.sys.readv` — scatter-read. macOS uses `readv$NOCANCEL` with no
/// EINTR retry; other POSIX retries on EINTR.
pub fn readv(fd: Fd, vecs: &[PlatformIoVec]) -> Maybe<usize> {
Expand All @@ -4759,6 +4848,9 @@ pub fn readv(fd: Fd, vecs: &[PlatformIoVec]) -> Maybe<usize> {
}
#[cfg(unix)]
{
// readv(2) rejects more than IOV_MAX iovecs with EINVAL; cap to one
// batch and return a short read, matching libuv's `uv__fs_read`.
Comment thread
robobun marked this conversation as resolved.
Outdated
let vecs = &vecs[..vecs.len().min(POSIX_IOV_MAX)];
#[cfg(target_os = "macos")]
{
// SAFETY: vecs.ptr is `*const iovec`; the kernel writes through
Expand Down Expand Up @@ -4808,6 +4900,8 @@ pub fn preadv(fd: Fd, vecs: &[PlatformIoVec], position: i64) -> Maybe<usize> {
}
#[cfg(unix)]
{
// See `readv`: cap at IOV_MAX and return a short read.
let vecs = &vecs[..vecs.len().min(POSIX_IOV_MAX)];
#[cfg(target_os = "macos")]
{
// SAFETY: see `readv`. Single shot.
Expand Down
Loading
Loading