Bun.write(file, file): wait for a non-blocking fd instead of rejecting with EAGAIN

src/runtime/webcore/blob/copy_file.rs

@@ -383,7 +383,6 @@ impl CopyFile {
         }
 
         loop {
-            // TODO: this should use non-blocking I/O.
             let written: isize = match USE {
                 TryWith::CopyFileRange => {
                     // SAFETY: raw copy_file_range(2); both fds owned by caller, null offsets.
@@ -427,6 +426,17 @@ impl CopyFile {
             match bun_sys::get_errno(written) {
                 bun_sys::E::SUCCESS => {}
 
+                // A pipe-to-pipe splice is non-blocking if either pipe is, so wait on both sides.
+                bun_sys::E::EAGAIN => {
+                    if let Err(err) = bun_sys::block_until_readable(src_fd)
+                        .and_then(|()| bun_sys::block_until_writable(dest_fd))
+                    {
+                        self.system_error = Some(err.to_system_error());
+                        return Err(bun_errno::from_errno(err.errno as i32).into());
+                    }
+                    continue;
+                }
+
                 // XDEV: cross-device copy not supported
                 // NOSYS: syscall not available
                 // OPNOTSUPP: filesystem doesn't support this operation
@@ -992,19 +1002,28 @@ fn read_write_loop_capped(
     let mut remaining = cap;
     while remaining > 0 {
         let want = (buf.len() as SizeType).min(remaining) as usize;
-        let amt = bun_sys::read(src_fd, &mut buf[..want])?;
+        let amt = match bun_sys::read(src_fd, &mut buf[..want]) {
+            Ok(amt) => amt,
+            Err(err) if err.is_retry() => {
+                bun_sys::block_until_readable(src_fd)?;
+                continue;
+            }
+            Err(err) => return Err(err),
+        };
         if amt == 0 {
             break;
         }
         remaining -= amt as SizeType;
         let mut slice = &buf[..amt];
         while !slice.is_empty() {
-            match bun_sys::write(dest_fd, slice)? {
-                0 => return Ok(()),
-                n => {
+            match bun_sys::write(dest_fd, slice) {
+                Ok(0) => return Ok(()),
+                Ok(n) => {
                     *total += n as u64;
                     slice = &slice[n..];
                 }
+                Err(err) if err.is_retry() => bun_sys::block_until_writable(dest_fd)?,
+                Err(err) => return Err(err),
             }
         }
     }

src/sys/lib.rs

@@ -1411,14 +1411,16 @@ impl Tag {
     #[cfg(not(windows))]
     pub(crate) const setrlimit: Tag = Tag(106);
     pub const clone3: Tag = Tag(107);
+    #[cfg(target_os = "macos")]
+    pub(crate) const select: Tag = Tag(108);
     // `inotify_init1`/`inotify_add_watch` fold under the generic `.watch`
     // tag; `INotifyWatcher.rs` spells it `.inotify`. Alias to `.watch`
     // so the JS-facing `err.syscall == "watch"` string stays node-compatible.
 
     /// The tag name — spelling is frozen (JS-facing
     /// `err.syscall` string; node-compat code matches on it).
     pub fn name(self) -> &'static str {
-        const NAMES: [&str; 108] = [
+        const NAMES: [&str; 109] = [
             "TODO",
             "dup",
             "access",
@@ -1528,6 +1530,7 @@ impl Tag {
             "getrlimit",
             "setrlimit",
             "clone3",
+            "select",
         ];
         NAMES.get(self.0 as usize).copied().unwrap_or("unknown")
     }
@@ -1708,6 +1711,15 @@ mod nocancel {
         ) -> isize;
         #[link_name = "poll$NOCANCEL"]
         pub(crate) fn poll(fds: *mut libc::pollfd, nfds: libc::nfds_t, timeout: c_int) -> c_int;
+        // `_DARWIN_UNLIMITED_SELECT`: no FD_SETSIZE cap; a set is ceil(nfds / 32) `u32`s.
+        #[link_name = "select$DARWIN_EXTSN$NOCANCEL"]
+        pub(crate) fn select(
+            nfds: c_int,
+            readfds: *mut u32,
+            writefds: *mut u32,
+            errorfds: *mut u32,
+            timeout: *mut libc::timeval,
+        ) -> c_int;
         // Remaining `$NOCANCEL` variants Bun links against.
         // safe: by-value `c_int` fd; bad fd → -1/EBADF, no UB.
         #[link_name = "close$NOCANCEL"]
@@ -7630,6 +7642,86 @@ pub fn kevent(
     }
 }
 
+/// Blocks until a read from `fd` would not return EAGAIN (there is data, EOF or an error).
+#[cfg(all(unix, not(target_os = "macos")))]
+pub fn block_until_readable(fd: Fd) -> Maybe<()> {
+    block_until(fd, posix::POLL_IN)
+}
+
+/// Blocks until a write to `fd` would not return EAGAIN (there is room, or an error like EPIPE).
+#[cfg(all(unix, not(target_os = "macos")))]
+pub fn block_until_writable(fd: Fd) -> Maybe<()> {
+    block_until(fd, posix::POLL_OUT)
+}
+
+/// POLLHUP / POLLERR / POLLNVAL wake this without being requested; EINTR is retried.
+#[cfg(all(unix, not(target_os = "macos")))]
+fn block_until(fd: Fd, events: i16) -> Maybe<()> {
+    debug_assert!(fd.is_valid());
+    let mut fds = [posix::PollFd {
+        fd: fd.native(),
+        events,
+        revents: 0,
+    }];
+    match posix::poll(&mut fds, -1) {
+        Ok(_) => Ok(()),
+        Err(err) => Err(err.with_fd(fd)),
+    }
+}
+
+/// Blocks until a read from `fd` would not return EAGAIN (there is data, EOF or an error).
+#[cfg(target_os = "macos")]
+pub fn block_until_readable(fd: Fd) -> Maybe<()> {
+    select_one(fd, SelectFor::Read)
+}
+
+/// Blocks until a write to `fd` would not return EAGAIN (there is room, or an error like EPIPE).
+#[cfg(target_os = "macos")]
+pub fn block_until_writable(fd: Fd) -> Maybe<()> {
+    select_one(fd, SelectFor::Write)
+}
+
+#[cfg(target_os = "macos")]
+#[derive(Clone, Copy)]
+enum SelectFor {
+    Read,
+    Write,
+}
+
+/// On XNU, poll(2) on a named pipe never wakes for the other end closing; select(2) does.
+#[cfg(target_os = "macos")]
+fn select_one(fd: Fd, what: SelectFor) -> Maybe<()> {
+    debug_assert!(fd.is_valid());
+    let index = fd.native() as usize;
+    let word = index / 32;
+    let mut set = vec![0u32; word + 1];
+    loop {
+        set.fill(0);
+        set[word] = 1 << (index % 32);
+        let (read_set, write_set) = match what {
+            SelectFor::Read => (set.as_mut_ptr(), core::ptr::null_mut()),
+            SelectFor::Write => (core::ptr::null_mut(), set.as_mut_ptr()),
+        };
+        // SAFETY: `set` holds the ceil(nfds / 32) words the kernel reads and
+        // writes back for `nfds = fd + 1`; the other sets and the timeout
+        // (wait indefinitely) may be null.
+        let rc = unsafe {
+            nocancel::select(
+                fd.native() + 1,
+                read_set,
+                write_set,
+                core::ptr::null_mut(),
+                core::ptr::null_mut(),
+            )
+        };
+        match get_errno(rc) {
+            E::SUCCESS => return Ok(()),
+            E::EINTR => continue,
+            e => return Err(Error::from_code(e, Tag::select).with_fd(fd)),
+        }
+    }
+}
+
 /// `clonefile` — macOS-only CoW copy. On non-Darwin returns ENOTSUP so
 /// callers can fall back to `copy_file`.
 #[cfg(not(target_os = "macos"))]