diff --git a/src/event_loop/MiniEventLoop.rs b/src/event_loop/MiniEventLoop.rs index 8c3bd7dd6a54..def0230ced69 100644 --- a/src/event_loop/MiniEventLoop.rs +++ b/src/event_loop/MiniEventLoop.rs @@ -47,9 +47,6 @@ unsafe extern "Rust" { } // ──────────────────────────────────────────────────────────────────────────── -const PIPE_READ_BUFFER_SIZE: usize = 256 * 1024; -pub type PipeReadBuffer = [u8; PIPE_READ_BUFFER_SIZE]; - /// Intrusive MPSC queue over `AnyTaskWithExtraContext` linked via its `.next` field. type ConcurrentTaskQueue = UnboundedQueue; @@ -82,7 +79,7 @@ pub struct MiniEventLoop { // Opaque ctx assigned externally; only read/cleared here. pub(crate) after_event_loop_callback_ctx: Option>, pub(crate) after_event_loop_callback: Option, - pub pipe_read_buffer: Option>, + pub pipe_read_scratch: Box, } thread_local! { @@ -215,14 +212,6 @@ impl MiniEventLoop { self.env } - pub fn pipe_read_buffer(&mut self) -> &mut [u8] { - // `boxed_zeroed` avoids the 256 KiB stack temporary `Box::new([0u8; N])` - // would create in debug builds. - &mut self - .pipe_read_buffer - .get_or_insert_with(bun_core::boxed_zeroed::)[..] - } - pub fn on_after_event_loop(&mut self) { if let Some(cb) = self.after_event_loop_callback { let ctx = self.after_event_loop_callback_ctx; @@ -277,7 +266,7 @@ impl MiniEventLoop { top_level_dir: Box::default(), after_event_loop_callback_ctx: None, after_event_loop_callback: None, - pipe_read_buffer: None, + pipe_read_scratch: Box::new(bun_io::PipeReadScratch::new()), } } @@ -438,7 +427,7 @@ bun_io::link_impl_EventLoopCtx! { (*this).after_event_loop_callback = cb; (*this).after_event_loop_callback_ctx = ctx; }, - pipe_read_buffer() => core::ptr::from_mut::<[u8]>((*this).pipe_read_buffer()), + pipe_read_scratch() => &raw const *(*this).pipe_read_scratch, } } diff --git a/src/event_loop/lib.rs b/src/event_loop/lib.rs index afa323575731..9a0be447363f 100644 --- a/src/event_loop/lib.rs +++ b/src/event_loop/lib.rs @@ -35,10 +35,10 @@ pub use ConcurrentTask::{Task, TaskTag, Taskable, task_tag}; // the type/module namespace collision on the PascalCase form. pub use DeferredTaskQueue as deferred_task_queue; -pub use MiniEventLoop::PipeReadBuffer; pub use any_event_loop::{ AnyEventLoop, EventLoopHandle, EventLoopTask, JsPoster, JsPosterVTable, Posted, }; +pub use bun_io::PipeReadScratch; // JS-event-loop arm of `AnyEventLoop` / `EventLoopHandle`. `bun_event_loop` is // a lower tier than `bun_jsc`, so it cannot name `jsc::EventLoop` / @@ -50,7 +50,6 @@ bun_dispatch::link_interface! { fn file_polls() -> *mut bun_io::file_poll::Store; fn put_file_poll(poll: *mut bun_io::FilePoll, was_ever_registered: bool); fn uws_loop() -> *mut bun_uws::Loop; - fn pipe_read_buffer() -> *mut [u8]; fn tick(); fn auto_tick(); fn auto_tick_active(); diff --git a/src/install/PackageManager/security_scanner.rs b/src/install/PackageManager/security_scanner.rs index 4cd539f90369..9d9426118cdd 100644 --- a/src/install/PackageManager/security_scanner.rs +++ b/src/install/PackageManager/security_scanner.rs @@ -1026,7 +1026,7 @@ impl<'a> Drop for SecurityScanSubprocess<'a> { bun_io::impl_buffered_reader_parent! { SecurityScan for SecurityScanSubprocess<'a>; has_on_read_chunk = true; - on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(chunk, has_more); + on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(&chunk, has_more); on_reader_done = |this| (*this).on_reader_done(); on_reader_error = |this, err| (*this).on_reader_error(err); loop_ = |this| (*this).loop_(); diff --git a/src/io/PipeReader.rs b/src/io/PipeReader.rs index 83f8c081e15b..406cb1543f31 100644 --- a/src/io/PipeReader.rs +++ b/src/io/PipeReader.rs @@ -22,7 +22,7 @@ pub type Loop = bun_sys::windows::libuv::Loop; /// dispatch in `bun_runtime::dispatch::__bun_run_file_poll` recovers the type /// from this constant. T2 cannot name `bun_io`, so the value is mirrored. use crate::max_buf::MaxBuf; -use crate::pipes::{FileType, PollOrFd, ReadState}; +use crate::pipes::{Chunk, FileType, PollOrFd, ReadState}; #[cfg(windows)] use crate::source::Source; @@ -74,7 +74,7 @@ pub trait BufferedReaderParent { /// Mirrors `@hasDecl(Type, "onReadChunk")`. const HAS_ON_READ_CHUNK: bool = true; - unsafe fn on_read_chunk(this: *mut Self, chunk: &[u8], has_more: ReadState) -> bool { + unsafe fn on_read_chunk(this: *mut Self, chunk: Chunk<'_>, has_more: ReadState) -> bool { let _ = (this, chunk, has_more); // Default: should not be called when HAS_ON_READ_CHUNK == false. true @@ -121,7 +121,7 @@ impl BufferedReaderVTable { /// When the reader has read a chunk of data /// and hasMore is true, it means that there might be more data to read. /// Returning false prevents the reader from reading more data. - fn on_read_chunk(&self, chunk: &[u8], has_more: ReadState) -> bool { + fn on_read_chunk(&self, chunk: Chunk<'_>, has_more: ReadState) -> bool { self.link().on_read_chunk(chunk, has_more) } @@ -148,30 +148,6 @@ impl Drop for ParentKeepAlive { } } -// The per-loop `pipe_read_buffer` scratch is handed to `on_read_chunk` as the -// chunk itself, and a consumer may keep parsing it while running user code -// that starts a *second* reader synchronously (HTMLRewriter handlers do). A -// nested read loop must not refill the scratch under the outer one, so only -// the outermost loop on the thread borrows it; nested ones read into their -// own `_buffer`. -thread_local! { - static READ_SCRATCH_IN_USE: core::cell::Cell = const { core::cell::Cell::new(false) }; -} - -struct ReadScratchClaim; - -impl ReadScratchClaim { - fn try_claim() -> Option { - READ_SCRATCH_IN_USE.with(|in_use| (!in_use.replace(true)).then_some(Self)) - } -} - -impl Drop for ReadScratchClaim { - fn drop(&mut self) { - READ_SCRATCH_IN_USE.with(|in_use| in_use.set(false)); - } -} - // ────────────────────────────────────────────────────────────────────────── // PosixBufferedReader // ────────────────────────────────────────────────────────────────────────── @@ -587,56 +563,31 @@ impl PosixBufferedReader { /// # Safety /// `this` is the live reader. Raw (not `&mut self`) because - /// `on_read_chunk` dispatched from the read loops re-enters JS, which can + /// `on_read_chunk` dispatched from the read loop re-enters JS, which can /// reach this reader again through its parent — a protected `&mut` /// spanning that re-entry is exactly the aliasing this API avoids. pub unsafe fn read(this: *mut Self) { // SAFETY: caller contract — `this` is live; borrows end at each `;`. - let (paused, fd, file_type, vtable) = unsafe { - ( - (*this).flags.contains(PosixFlags::IS_PAUSED), - (*this).get_fd(), - (*this).get_file_type(), - (*this).vtable, - ) - }; - // Don't initiate new reads if paused - if paused { + let Some((fd, file_type, vtable)) = (unsafe { (*this).begin_read() }) else { return; - } - // As in `on_poll`: the synchronous read loops below dispatch - // `on_read_chunk` and touch `*this` afterwards, so the parent (which - // embeds this reader) must outlive them. + }; + // The read loop dispatches `on_read_chunk` and touches `*this` + // afterwards, so the parent (which embeds this reader) must outlive it. let _parent = vtable.ref_parent(); - - match file_type { - FileType::NonblockingPipe => { - // SAFETY: caller contract. - unsafe { Self::read_pipe(this, fd, 0, false) }; - } - FileType::File => { - // SAFETY: caller contract. - unsafe { Self::read_file(this, fd, 0, false) }; - } - FileType::Socket => { - // SAFETY: caller contract. - unsafe { Self::read_socket(this, fd, 0, false) }; - } - FileType::Pipe => match bun_core::is_readable(fd) { - bun_core::Pollable::Ready => { - // SAFETY: caller contract. - unsafe { Self::read_from_blocking_pipe_without_blocking(this, fd, 0, false) }; - } - bun_core::Pollable::Hup => { - // SAFETY: caller contract. - unsafe { Self::read_from_blocking_pipe_without_blocking(this, fd, 0, true) }; - } + let mut received_hup = false; + if file_type == FileType::Pipe { + match bun_core::is_readable(fd) { + bun_core::Pollable::Ready => {} + bun_core::Pollable::Hup => received_hup = true, bun_core::Pollable::NotReady => { - // SAFETY: caller contract; borrow scoped to the call. + // SAFETY: caller contract; the error dispatch may free the parent. unsafe { Self::register_poll(this) }; + return; } - }, + } } + // SAFETY: caller contract. + unsafe { Self::read_loop(this, file_type, fd, received_hup) }; } /// # Safety @@ -644,164 +595,105 @@ impl PosixBufferedReader { /// [`Self::read`] for why the entry is raw. pub unsafe fn on_poll(this: *mut PosixBufferedReader, size_hint: isize, received_hup: bool) { // SAFETY: caller contract — `this` is live; borrows end at each `;`. - let (paused, fd, file_type, vtable) = unsafe { - ( - (*this).flags.contains(PosixFlags::IS_PAUSED), - (*this).get_fd(), - (*this).get_file_type(), - (*this).vtable, - ) - }; - if paused { + let Some((fd, file_type, vtable)) = (unsafe { (*this).begin_read() }) else { return; - } + }; bun_sys::syslog!("onPoll({}) = {}", fd, size_hint); let _parent = vtable.ref_parent(); - - match file_type { - FileType::NonblockingPipe => { - // SAFETY: caller contract. - unsafe { Self::read_pipe(this, fd, size_hint, received_hup) }; - } - FileType::File => { - // SAFETY: caller contract. - unsafe { Self::read_file(this, fd, size_hint, received_hup) }; - } - FileType::Socket => { - // SAFETY: caller contract. - unsafe { Self::read_socket(this, fd, size_hint, received_hup) }; - } - FileType::Pipe => { - // SAFETY: caller contract. - unsafe { - Self::read_from_blocking_pipe_without_blocking( - this, - fd, - size_hint, - received_hup, - ) - }; - } - } + // SAFETY: caller contract. + unsafe { Self::read_loop(this, file_type, fd, received_hup) }; } - // Takes &vtable instead of &mut Self so - // call sites can pass &parent._buffer alongside without a raw-pointer escape. - #[inline] - fn drain_chunk(vtable: &BufferedReaderVTable, chunk: &[u8], has_more: ReadState) -> bool { - if vtable.is_streaming_enabled() { - if !chunk.is_empty() { - return vtable.on_read_chunk(chunk, has_more); - } + fn begin_read(&self) -> Option<(Fd, FileType, BufferedReaderVTable)> { + if self.flags.contains(PosixFlags::IS_PAUSED) { + return None; } - - false + Some((self.get_fd(), self.get_file_type(), self.vtable)) } - /// Charges `bytes_read` against the `maxBuffer` budget, returning `true` - /// once it is gone. The overflow callback only kills the child, which takes - /// effect asynchronously, so the caller must also stop reading. - #[inline] - fn charge_max_buffer(parent: &mut PosixBufferedReader, bytes_read: usize) -> bool { - let Some(maxbuf) = parent.maxbuf else { + /// Charges `bytes_read` against the `maxBuffer` budget; `true` once it is gone. The overflow callback only kills the child, so the caller must also stop reading. + fn charge_max_buffer(&mut self, bytes_read: usize) -> bool { + let Some(maxbuf) = self.maxbuf else { return false; }; MaxBuf::on_read_bytes(maxbuf, bytes_read as u64) } - /// Closes the handle so the child cannot put more bytes in the pipe, then - /// reports what was buffered. Raw (not `&mut`) like [`Self::done`]: the - /// `done` dispatch may free the parent embedding `*this`, so no receiver - /// protector may be live around it. Callers must already have handed the - /// overflowing chunk to the consumer. - /// - /// # Safety - /// `this` is the live reader. - unsafe fn stop_for_max_buffer(this: *mut PosixBufferedReader) { - // SAFETY: caller contract; the borrow ends at `;`, before the dispatch. - let already_done = unsafe { - (*this).close_without_reporting(); - (*this).flags.contains(PosixFlags::IS_DONE) - }; - if !already_done { - // SAFETY: caller contract; no borrow of `*this` is live. - unsafe { Self::done(this) }; + /// Every kind uses its non-blocking primitive: `RWF_NOWAIT`/poll-guarded reads for pipes, `MSG_DONTWAIT` for sockets; regular files cannot block. + fn sys_read(&self, file_type: FileType, fd: Fd, buf: &mut [u8]) -> sys::Result { + match file_type { + FileType::File if self.flags.contains(PosixFlags::USE_PREAD) => { + sys::pread(fd, buf, i64::try_from(self._offset).expect("int cast")) + } + FileType::File => sys::read(fd, buf), + FileType::Socket => sys::recv_non_block(fd, buf), + FileType::NonblockingPipe | FileType::Pipe => sys::read_nonblocking(fd, buf), } } - /// # Safety - /// Same contract as [`Self::read`]. - unsafe fn read_file( - this: *mut PosixBufferedReader, - fd: Fd, - size_hint: isize, - received_hup: bool, - ) { - fn pread_fn(fd1: Fd, buf: &mut [u8], offset: usize) -> sys::Result { - sys::pread(fd1, buf, i64::try_from(offset).expect("int cast")) - } - // SAFETY: caller contract; borrow ends at `;`. - let use_pread = unsafe { (*this).flags.contains(PosixFlags::USE_PREAD) }; - if use_pread { - // SAFETY: caller contract. - unsafe { - Self::read_with_fn(this, FileType::File, fd, size_hint, received_hup, pread_fn) - }; - } else { - // SAFETY: caller contract. - unsafe { - Self::read_with_fn( - this, - FileType::File, - fd, - size_hint, - received_hup, - |fd, buf, _| sys::read(fd, buf), - ) - }; + /// One syscall into `buf`; charges the byte budget and advances the offset. + fn read_once(&mut self, file_type: FileType, fd: Fd, buf: &mut [u8]) -> ReadOnce { + let buf = MaxBuf::clamp_read_buf(self.maxbuf, buf); + match self.sys_read(file_type, fd, buf) { + sys::Result::Ok(0) => ReadOnce::Stop(Stop::Eof), + sys::Result::Ok(n) => { + self._offset += n; + if self.charge_max_buffer(n) { + ReadOnce::Read(n, Some(Stop::OverBudget)) + } else { + ReadOnce::Read(n, None) + } + } + sys::Result::Err(err) if err.is_retry() => ReadOnce::Stop(Stop::WouldBlock), + sys::Result::Err(err) => ReadOnce::Stop(Stop::Error(err)), } } - /// # Safety - /// Same contract as [`Self::read`]. - unsafe fn read_socket( - this: *mut PosixBufferedReader, + /// Reads into `scratch` until it is worth delivering; returns bytes filled and why it stopped (`None`: deliver and keep going). + fn fill_scratch( + &mut self, + file_type: FileType, fd: Fd, - size_hint: isize, - received_hup: bool, - ) { - // SAFETY: caller contract. - unsafe { - Self::read_with_fn( - this, - FileType::Socket, - fd, - size_hint, - received_hup, - |fd, buf, _| sys::recv_non_block(fd, buf), - ) - }; + scratch: &mut [u8], + ) -> (usize, Option) { + let mut filled = 0; + while scratch.len() - filled > 16 * 1024 && filled < scratch.len() / 2 { + match self.read_once(file_type, fd, &mut scratch[filled..]) { + ReadOnce::Read(n, stop) => { + filled += n; + if stop.is_some() || file_type == FileType::Pipe { + return (filled, stop); + } + } + ReadOnce::Stop(stop) => return (filled, Some(stop)), + } + } + (filled, None) } - /// # Safety - /// Same contract as [`Self::read`]. - unsafe fn read_pipe( - this: *mut PosixBufferedReader, - fd: Fd, - size_hint: isize, - received_hup: bool, - ) { - // SAFETY: caller contract. - unsafe { - Self::read_with_fn( - this, - FileType::NonblockingPipe, - fd, - size_hint, - received_hup, - |fd, buf, _| sys::read_nonblocking(fd, buf), - ) - }; + /// Reads into `_buffer` until it is worth delivering (streaming) or exhausted (buffering). + fn fill_buffer(&mut self, file_type: FileType, fd: Fd, streaming: bool) -> Option { + loop { + self._buffer.reserve(16 * 1024); + // SAFETY: the syscall writes only initialized bytes into the prefix it reports and `commit_spare` exposes exactly that prefix. + let read = unsafe { + let spare: *mut [u8] = bun_core::vec::spare_bytes_mut(&mut self._buffer); + self.read_once(file_type, fd, &mut *spare) + }; + match read { + ReadOnce::Read(n, stop) => { + // SAFETY: `read_once` initialized `n` bytes of the spare capacity. + unsafe { bun_core::vec::commit_spare(&mut self._buffer, n) }; + if stop.is_some() + || file_type == FileType::Pipe + || (streaming && self._buffer.len() >= 128 * 1024) + { + return stop; + } + } + ReadOnce::Stop(stop) => return Some(stop), + } + } } /// # Safety @@ -810,231 +702,128 @@ impl PosixBufferedReader { /// contract) but may mutate it — no borrow of `*this` is held across any /// dispatch below. `on_error()` / `done()` MAY free the parent, so both /// are dispatched in tail position. - unsafe fn read_blocking_pipe( + unsafe fn read_loop( this: *mut PosixBufferedReader, + file_type: FileType, fd: Fd, - _size_hint: isize, - received_hup_initially: bool, + mut received_hup: bool, ) { - // The vtable is two Copy scalars set once at `start()`; copying it out - // lets every `on_read_chunk` dispatch run with no borrow of `*this`. // SAFETY: caller contract — `this` is live. let vtable = unsafe { (*this).vtable }; - let mut received_hup = received_hup_initially; - let scratch = ReadScratchClaim::try_claim(); + let streaming = vtable.is_streaming_enabled(); + let mut scratch = vtable.event_loop().claim_pipe_read_scratch(); loop { - let streaming = vtable.is_streaming_enabled(); - let mut got_retry = false; - // SAFETY: caller contract; borrow ends at `;`. - let unbuffered = scratch.is_some() && unsafe { (*this)._buffer.is_empty() }; - if unbuffered { - // Use stack buffer for streaming — per-loop scratch buffer; - // single-threaded event loop (see `EventLoopCtx::pipe_read_buffer_mut`). - // SAFETY: caller contract; `maxbuf` is Copy. - let maxbuf = unsafe { (*this).maxbuf }; - let stack_buffer = vtable.event_loop().pipe_read_buffer_mut(); - let stack_buffer = MaxBuf::clamp_read_buf(maxbuf, stack_buffer); - - match sys::read_nonblocking(fd, stack_buffer) { - sys::Result::Ok(bytes_read) => { - // SAFETY: caller contract; borrow scoped to the call. - let over_budget = - Self::charge_max_buffer(unsafe { &mut *this }, bytes_read); - - if bytes_read == 0 { - // EOF - finished and closed pipe - // SAFETY: caller contract; `done()` is the tail. - unsafe { - (*this).close_without_reporting(); - if !(*this).flags.contains(PosixFlags::IS_DONE) { - Self::done(this); - } - } - return; - } - - if streaming { - // Stream this chunk and register for next cycle - let keep_going = vtable.on_read_chunk( - &stack_buffer[..bytes_read], - if received_hup && bytes_read < stack_buffer.len() { - ReadState::Eof - } else { - ReadState::Progress - }, - ); - // Re-entrant JS inside on_read_chunk can close the - // reader (nested on_pull -> read -> EOF); the - // captured `fd` is then stale regardless of HUP. - // SAFETY: caller contract (re-entry never frees `*this`). - if unsafe { (*this).is_done() } { - return; - } - if !keep_going && !received_hup && !over_budget { - return; - } - } else { - // SAFETY: caller contract; borrow ends at `;`. - unsafe { - (*this) - ._buffer - .extend_from_slice(&stack_buffer[..bytes_read]); - } - } - - if over_budget { - // SAFETY: caller contract; tail position, raw entry. - unsafe { Self::stop_for_max_buffer(this) }; - return; - } - } - sys::Result::Err(err) => { - if !err.is_retry() { - // SAFETY: caller contract; `on_error` is the tail. - unsafe { Self::on_error(this, err) }; - return; - } - // EAGAIN - fall through to register for next poll - got_retry = true; - } + let use_scratch = unsafe { + (*this)._buffer.is_empty() && (streaming || (*this)._buffer.capacity() == 0) + }; + let (stop, keep_going) = match (use_scratch, scratch.as_mut()) { + (true, Some(scratch)) => { + // SAFETY: caller contract; the borrow ends before the dispatch. + let (filled, stop) = unsafe { (*this).fill_scratch(file_type, fd, scratch) }; + // SAFETY: caller contract; borrow ends at `;`. + unsafe { Self::close_if_final(this, stop.as_ref()) }; + let keep_going = if filled == 0 { + true + } else if streaming { + vtable.on_read_chunk( + Chunk::Scratch(&scratch[..filled]), + Self::read_state(stop.as_ref(), received_hup), + ) + } else { + // SAFETY: caller contract; borrow ends at `;`. + unsafe { (*this)._buffer.extend_from_slice(&scratch[..filled]) }; + true + }; + (stop, keep_going) } - } else { - // SAFETY: caller contract; `maxbuf` is Copy, borrow ends at `;`. - let maxbuf = unsafe { (*this).maxbuf }; - // SAFETY: caller contract; borrow ends at `;`. - unsafe { (*this)._buffer.reserve(16 * 1024) }; - // SAFETY: caller contract. `sys::read_nonblocking` writes only - // initialized bytes into the prefix it reports; `commit_spare` - // exposes exactly that prefix. The `_buffer` borrow ends before - // any dispatch. - let read_result = unsafe { - let buf = bun_core::vec::spare_bytes_mut(&mut (*this)._buffer); - let buf = MaxBuf::clamp_read_buf(maxbuf, buf); - let buf_len = buf.len(); - (sys::read_nonblocking(fd, buf), buf_len) - }; - match read_result { - (sys::Result::Ok(bytes_read), buf_len) => { - // SAFETY: caller contract; borrow scoped to the call. - let over_budget = - Self::charge_max_buffer(unsafe { &mut *this }, bytes_read); - // SAFETY: caller contract; `bytes_read` bytes were just - // initialized by the syscall; borrows end at each `;`. - unsafe { - (*this)._offset += bytes_read; - bun_core::vec::commit_spare(&mut (*this)._buffer, bytes_read); - } - - if bytes_read == 0 { - // SAFETY: caller contract; `done()` is the tail. - unsafe { - (*this).close_without_reporting(); - if !(*this).flags.contains(PosixFlags::IS_DONE) { - Self::done(this); - } - } - return; - } - - if streaming { - // Move the buffer out for the dispatch so re-entrant - // access to the reader cannot alias or reallocate it - // under the chunk slice. - // SAFETY: caller contract; borrow ends at `;`. - let buffer = unsafe { core::mem::take(&mut (*this)._buffer) }; - let new_len = buffer.len(); - let keep_going = vtable.on_read_chunk( - &buffer[new_len - bytes_read..new_len], - if received_hup && bytes_read < buf_len { - ReadState::Eof - } else { - ReadState::Progress - }, - ); - // Delivered bytes are consumed by `on_read_chunk`; keep only what re-entry buffered. + _ => { + // SAFETY: caller contract; the borrow ends before the dispatch. + let stop = unsafe { (*this).fill_buffer(file_type, fd, streaming) }; + // SAFETY: caller contract; borrow ends at `;`. + unsafe { Self::close_if_final(this, stop.as_ref()) }; + // SAFETY: caller contract; borrow ends at `;`. + let keep_going = if !streaming || unsafe { (*this)._buffer.is_empty() } { + true + } else { + // Moved out so a re-entrant read cannot alias or reallocate it under the consumer. + // SAFETY: caller contract; borrow ends at `;`. + let mut buffer = unsafe { mem::take(&mut (*this)._buffer) }; + let state = Self::read_state(stop.as_ref(), received_hup); + if matches!(stop, Some(Stop::Eof | Stop::OverBudget | Stop::Error(_))) { + vtable.on_read_chunk(Chunk::Owned(buffer), state) + } else { + let keep_going = + vtable.on_read_chunk(Chunk::Buffer(&mut buffer), state); + buffer.clear(); // SAFETY: caller contract; borrows end at the block. unsafe { - let mut buffer = buffer; - buffer.clear(); - buffer.extend_from_slice(&(*this)._buffer); - (*this)._buffer = buffer; - } - // SAFETY: caller contract. - if unsafe { (*this).is_done() } { - return; - } - // Closing for `over_budget` outranks the - // consumer asking us to stop: it must still - // happen, or nothing ever caps the pipe. - if !keep_going && !over_budget { - return; + if (*this)._buffer.is_empty() { + (*this)._buffer = buffer; + } } + keep_going } + }; + (stop, keep_going) + } + }; - if over_budget { - // SAFETY: caller contract; tail position, raw entry. - unsafe { Self::stop_for_max_buffer(this) }; - return; - } - } - (sys::Result::Err(err), _) => { - if !err.is_retry() { - // SAFETY: caller contract; `on_error` is the tail. - unsafe { Self::on_error(this, err) }; - return; + match stop { + Some(Stop::Eof | Stop::OverBudget) => { + // SAFETY: caller contract; `done()` is the tail. + unsafe { + if !(*this).flags.contains(PosixFlags::IS_DONE) { + Self::done(this); } - got_retry = true; } + return; } + Some(Stop::Error(err)) => { + // SAFETY: caller contract; `on_error` is the tail. + unsafe { Self::on_error(this, err) }; + return; + } + _ => {} } - - // Register for next poll cycle unless we got HUP - if !received_hup { - // SAFETY: caller contract; borrow scoped to the call. - unsafe { Self::register_poll(this) }; + // Re-entrant JS inside on_read_chunk can close the reader (nested on_pull -> read -> EOF); the captured `fd` is then stale. + // SAFETY: caller contract (re-entry never frees `*this`). + if unsafe { (*this).is_done() } { + return; + } + if let Some(Stop::WouldBlock) = stop { + if file_type == FileType::File { + bun_core::debug_warn!( + "Received EAGAIN while reading from a file. This is a bug." + ); + } else { + // SAFETY: caller contract; the error dispatch may free the parent. + unsafe { Self::register_poll(this) }; + } return; } + if streaming && !keep_going && !received_hup { + return; + } + if file_type != FileType::Pipe { + continue; + } - // We have received HUP. Normally that means all writers are gone - // and draining the buffer will eventually hit EOF (read() == 0), - // so we loop locally instead of re-arming the poll (HUP is - // level-triggered and would fire again immediately). - // - // But `received_hup` is a snapshot from when the epoll/kqueue - // event fired. `onReadChunk` above re-enters JS (resolves the - // pending read, drains microtasks, fires the 'data' event), and - // user code there can open a new writer on the same FIFO — after - // which the pipe is no longer hung up. Looping again would then - // either spin forever on EAGAIN (if the fd is O_NONBLOCK) or - // block the event loop in read() (if the fd is blocking and - // RWF_NOWAIT is unavailable — Linux named FIFOs return - // EOPNOTSUPP for it, unlike anonymous pipes). - // - // An explicit EAGAIN proves the HUP is stale, so re-arm. - if got_retry { - // SAFETY: caller contract; borrow scoped to the call. + // A blocking pipe gets one read per wakeup unless it hung up, in + // which case draining locally reaches EOF — but `received_hup` is a + // snapshot, and user JS inside `on_read_chunk` may have opened a new + // writer on the same FIFO. Re-check before committing to a read + // that could block (Linux named FIFOs reject RWF_NOWAIT). + if !received_hup { + // SAFETY: caller contract; the error dispatch may free the parent. unsafe { Self::register_poll(this) }; return; } - // Otherwise we just returned from user JS; re-poll the fd to see - // whether HUP still holds before committing to another blocking - // read. This is one extra poll() per chunk only on the HUP path - // (i.e. while draining the final buffered bytes), not per read. match bun_core::is_readable(fd) { - bun_core::Pollable::Hup => { - // Still hung up; keep draining towards EOF. - } - bun_core::Pollable::Ready => { - // Data is available but HUP cleared — a writer came back. - // Drop the stale HUP so the next iteration takes the - // normal registerPoll() exit once the data is drained. - received_hup = false; - } + bun_core::Pollable::Hup => {} + bun_core::Pollable::Ready => received_hup = false, bun_core::Pollable::NotReady => { - // No data and no HUP: a writer exists. Go back to the - // event loop instead of blocking in read(). - // SAFETY: caller contract; borrow scoped to the call. + // SAFETY: caller contract; the error dispatch may free the parent. unsafe { Self::register_poll(this) }; return; } @@ -1042,379 +831,94 @@ impl PosixBufferedReader { } } - // PERF: `file_type` is a runtime arg (adt_const_params is unstable); `sys_fn` - // is generic so it still monomorphizes — profile if hot. + /// Closes before the final chunk is delivered, so a consumer that pulls again from inside `on_read_chunk` finds the reader done instead of reading past EOF or the byte budget. + /// /// # Safety - /// Same contract as [`Self::read_blocking_pipe`]: `this` is live, re-entry - /// through `on_read_chunk` may mutate but never frees `*this`, and no - /// borrow of `*this` is held across any dispatch; `on_error()` / `done()` - /// are tail-positioned because they may free the parent. - unsafe fn read_with_fn( - this: *mut PosixBufferedReader, - file_type: FileType, - fd: Fd, - _size_hint: isize, - received_hup: bool, - sys_fn: impl Fn(Fd, &mut [u8], usize) -> sys::Result, - ) { - // Copy scalars set once at `start()`; dispatching through the copy - // keeps `*this` unborrowed across every re-entry point. - // SAFETY: caller contract — `this` is live. - let vtable = unsafe { (*this).vtable }; - let streaming = vtable.is_streaming_enabled(); - let scratch = ReadScratchClaim::try_claim(); - - if streaming && scratch.is_some() { - // Per-loop scratch buffer; single-threaded event loop (see - // `EventLoopCtx::pipe_read_buffer_mut`). - let event_loop = vtable.event_loop(); - let stack_buffer_len = event_loop.pipe_read_buffer_mut().len(); - // SAFETY: caller contract; borrow ends at the loop test. - while unsafe { (*this)._buffer.is_empty() } { - let stack_buffer_cutoff = stack_buffer_len / 2; - let mut head_start = 0usize; // index into stack_buffer where the unwritten head begins - while stack_buffer_len - head_start > 16 * 1024 { - // SAFETY: caller contract; the `maxbuf`/`_offset` reads end - // before the syscall's buffer borrow (event-loop scratch, - // not `*this`). - let (maxbuf, offset) = unsafe { ((*this).maxbuf, (*this)._offset) }; - let buf = &mut event_loop.pipe_read_buffer_mut()[head_start..]; - let buf = MaxBuf::clamp_read_buf(maxbuf, buf); - - match sys_fn(fd, buf, offset) { - sys::Result::Ok(bytes_read) => { - // SAFETY: caller contract; borrow scoped to the call. - let over_budget = - Self::charge_max_buffer(unsafe { &mut *this }, bytes_read); - // SAFETY: caller contract; borrow ends at `;`. - unsafe { (*this)._offset += bytes_read }; - head_start += bytes_read; - - // `over_budget` is terminal for the same reason EOF - // is: the child was killed and nothing past the cap - // may reach the consumer. - if bytes_read == 0 || over_budget { - // SAFETY: caller contract; borrow ends at `;`. - unsafe { (*this).close_without_reporting() }; - if head_start > 0 { - let _ = vtable.on_read_chunk( - &event_loop.pipe_read_buffer_mut()[..head_start], - ReadState::Eof, - ); - } - // SAFETY: caller contract; `done()` is the tail. - unsafe { - if !(*this).flags.contains(PosixFlags::IS_DONE) { - Self::done(this); - } - } - return; - } - - // Keep reading as much as we can - if (stack_buffer_len - head_start) < stack_buffer_cutoff { - // `&& !received_hup` mirrors the - // after-inner-loop flush below (line ~855). - // Without it, a peer close (HUP) with >cutoff - // bytes still buffered makes a parent that - // returns `false` on `.eof` (e.g. shell - // `PipeReader::on_read_chunk`) early-return - // here with data left in the kernel and no - // `register_poll`/`done()` → 90s hang in - // shell-blocking-pipe.test.ts. - // Once HUP is set the kernel - // returns the remaining bytes then 0, so - // draining to `bytes_read == 0` is bounded. - let keep_going = vtable.on_read_chunk( - &event_loop.pipe_read_buffer_mut()[..head_start], - if received_hup { - ReadState::Eof - } else { - ReadState::Progress - }, - ); - // Re-entrant close (nested on_pull -> read -> - // EOF) invalidates the captured `fd`; stop - // before the next recv regardless of HUP. - // SAFETY: caller contract. - if unsafe { (*this).is_done() } { - return; - } - if !keep_going && !received_hup { - return; - } - head_start = 0; - } - } - sys::Result::Err(err) => { - if err.is_retry() { - if file_type == FileType::File { - bun_core::debug_warn!( - "Received EAGAIN while reading from a file. This is a bug.", - ); - } else { - // SAFETY: caller contract; borrow scoped to - // the call. `on_reader_error` from a failed - // re-arm may have freed the struct embedding - // `*this`; the drained head must not be - // delivered. - if !unsafe { Self::register_poll(this) } { - return; - } - } - - if head_start > 0 { - let _ = vtable.on_read_chunk( - &event_loop.pipe_read_buffer_mut()[..head_start], - ReadState::Drained, - ); - } - return; - } + /// `this` is the live reader. + unsafe fn close_if_final(this: *mut Self, stop: Option<&Stop>) { + if matches!(stop, Some(Stop::Eof | Stop::OverBudget)) { + // SAFETY: caller contract; borrow ends at `;`. + unsafe { (*this).close_without_reporting() }; + } + } - if head_start > 0 { - let _ = vtable.on_read_chunk( - &event_loop.pipe_read_buffer_mut()[..head_start], - ReadState::Progress, - ); - } - // SAFETY: caller contract; `on_error` is the tail. - unsafe { Self::on_error(this, err) }; - return; - } - } - } + fn read_state(stop: Option<&Stop>, received_hup: bool) -> ReadState { + match stop { + Some(Stop::Eof | Stop::OverBudget) => ReadState::Eof, + Some(Stop::WouldBlock) => ReadState::Drained, + Some(Stop::Error(_)) => ReadState::Progress, + None if received_hup => ReadState::Eof, + None => ReadState::Progress, + } + } - if head_start > 0 { - let keep_going = vtable.on_read_chunk( - &event_loop.pipe_read_buffer_mut()[..head_start], - if received_hup { - ReadState::Eof - } else { - ReadState::Progress - }, - ); + /// One non-blocking read straight into `dst` for a consumer pulling synchronously; arms the poll when nothing is available. EOF and errors are reported through `on_reader_done` / `on_reader_error` like any other read. + /// + /// # Safety + /// Same contract as [`Self::read`]: those dispatches may free the parent. + pub unsafe fn read_into(this: *mut Self, dst: &mut [u8]) -> (usize, ReadState) { + // SAFETY: caller contract — `this` is live; borrow ends at `;`. + let Some((fd, file_type, vtable)) = (unsafe { (*this).begin_read() }) else { + return (0, ReadState::Progress); + }; + if dst.is_empty() { + return (0, ReadState::Progress); + } + let _parent = vtable.ref_parent(); + if file_type == FileType::Pipe { + match bun_core::is_readable(fd) { + bun_core::Pollable::Ready | bun_core::Pollable::Hup => {} + bun_core::Pollable::NotReady => { // SAFETY: caller contract. - if unsafe { (*this).is_done() } { - return; - } - if !keep_going && !received_hup { - return; - } - } - - if !vtable.is_streaming_enabled() { - break; - } - } - } else { - // SAFETY: caller contract; borrows end at `;`. - let take_stack_path = !streaming - && scratch.is_some() - && unsafe { (*this)._buffer.capacity() == 0 && (*this)._offset == 0 }; - if take_stack_path { - // Avoid a 16 KB dynamic memory allocation when the buffer might very well be empty. - // Per-loop scratch buffer; single-threaded event loop (see - // `EventLoopCtx::pipe_read_buffer_mut`). - // SAFETY: caller contract; `maxbuf` is Copy. - let maxbuf = unsafe { (*this).maxbuf }; - let stack_buffer = vtable.event_loop().pipe_read_buffer_mut(); - let stack_buffer = MaxBuf::clamp_read_buf(maxbuf, stack_buffer); - - // Unlike the block of code following this one, only handle the non-streaming case. - debug_assert!(!streaming); - - match sys_fn(fd, stack_buffer, 0) { - sys::Result::Ok(bytes_read) => { - if bytes_read > 0 { - // SAFETY: caller contract; borrow ends at `;`. - unsafe { - (*this) - ._buffer - .extend_from_slice(&stack_buffer[..bytes_read]); - } - } - // SAFETY: caller contract; borrow scoped to the call. - let over_budget = - Self::charge_max_buffer(unsafe { &mut *this }, bytes_read); - // SAFETY: caller contract; borrow ends at `;`. - unsafe { (*this)._offset += bytes_read }; - - // `over_budget` is terminal for the same reason EOF is: the - // child was killed and nothing past the cap may be buffered. - if bytes_read == 0 || over_budget { - // Move the buffer out so a re-entrant read cannot - // alias it across the drain dispatch. - // SAFETY: caller contract; borrows end at each `;`. - let buffer = unsafe { - (*this).close_without_reporting(); - core::mem::take(&mut (*this)._buffer) - }; - let delivered = vtable.is_streaming_enabled() && !buffer.is_empty(); - let _ = Self::drain_chunk(&vtable, &buffer, ReadState::Eof); - // SAFETY: caller contract; `done()` is the tail. - unsafe { - if !delivered { - let mut buffer = buffer; - buffer.extend_from_slice(&(*this)._buffer); - (*this)._buffer = buffer; - } - if !(*this).flags.contains(PosixFlags::IS_DONE) { - Self::done(this); - } - } - return; - } - } - sys::Result::Err(err) => { - if err.is_retry() { - if file_type == FileType::File { - bun_core::debug_warn!( - "Received EAGAIN while reading from a file. This is a bug.", - ); - } else { - // SAFETY: caller contract; borrow scoped to the call. - unsafe { Self::register_poll(this) }; - } - return; - } - // SAFETY: caller contract; `on_error` is the tail. - unsafe { Self::on_error(this, err) }; - return; - } + unsafe { Self::register_poll(this) }; + return (0, ReadState::Progress); } - - // Allow falling through } } - - loop { - // SAFETY: caller contract. The `_buffer` borrow (reserve + spare - // prefix) and the syscall both end inside this block, before any - // dispatch. - let read_result = unsafe { - let maxbuf = (*this).maxbuf; - (*this)._buffer.reserve(16 * 1024); - let buf = bun_core::vec::spare_bytes_mut(&mut (*this)._buffer); - let buf = MaxBuf::clamp_read_buf(maxbuf, buf); - sys_fn(fd, buf, (*this)._offset) - }; - - match read_result { - sys::Result::Ok(bytes_read) => { - // SAFETY: caller contract; borrow scoped to the call. - let over_budget = Self::charge_max_buffer(unsafe { &mut *this }, bytes_read); - // SAFETY: caller contract; `bytes_read` bytes were just - // initialized by `sys_fn`; borrows end at each `;`. - unsafe { - (*this)._offset += bytes_read; - bun_core::vec::commit_spare(&mut (*this)._buffer, bytes_read); - } - - // `over_budget` is terminal for the same reason EOF is: the - // child was killed and nothing past the cap may be buffered. - if bytes_read == 0 || over_budget { - // SAFETY: caller contract; borrows end at each `;`. - let buffer = unsafe { - (*this).close_without_reporting(); - core::mem::take(&mut (*this)._buffer) - }; - let delivered = vtable.is_streaming_enabled() && !buffer.is_empty(); - let _ = Self::drain_chunk(&vtable, &buffer, ReadState::Eof); - // SAFETY: caller contract; `done()` is the tail. - unsafe { - if !delivered { - let mut buffer = buffer; - buffer.extend_from_slice(&(*this)._buffer); - (*this)._buffer = buffer; - } - if !(*this).flags.contains(PosixFlags::IS_DONE) { - Self::done(this); - } - } - return; - } - - if vtable.is_streaming_enabled() { - // SAFETY: caller contract; borrow ends at `;`. - let over_highwater = unsafe { (*this)._buffer.len() > 128_000 }; - if over_highwater { - // Move the buffer out for the dispatch, then - // reinstall it cleared (matching the pre-existing - // clear-after-dispatch semantics). - // SAFETY: caller contract; borrow ends at `;`. - let mut buffer = unsafe { core::mem::take(&mut (*this)._buffer) }; - let keep_going = vtable.on_read_chunk(&buffer, ReadState::Progress); - buffer.clear(); - // SAFETY: caller contract; borrows end at each `;`. - unsafe { - (*this)._buffer = buffer; - if (*this).is_done() || !keep_going { - return; - } - } - continue; - } + // SAFETY: caller contract; the borrow ends before any dispatch. + let (n, stop) = match unsafe { (*this).read_once(file_type, fd, dst) } { + ReadOnce::Read(n, stop) => (n, stop), + ReadOnce::Stop(stop) => (0, Some(stop)), + }; + match stop { + None => (n, ReadState::Progress), + Some(Stop::Eof | Stop::OverBudget) => { + // SAFETY: caller contract; `done()` may free the parent, nothing of `*this` is touched after. + unsafe { + (*this).close_without_reporting(); + if !(*this).flags.contains(PosixFlags::IS_DONE) { + Self::done(this); } } - sys::Result::Err(err) => { - if vtable.is_streaming_enabled() { - // SAFETY: caller contract; borrow ends at `;`. - let buffer = unsafe { core::mem::take(&mut (*this)._buffer) }; - if !buffer.is_empty() { - let _ = vtable.on_read_chunk(&buffer, ReadState::Drained); - } - // Reinstall cleared (capacity reuse; pre-existing - // clear-after-dispatch semantics). - // SAFETY: caller contract; borrow ends at `;`. - unsafe { - let mut buffer = buffer; - buffer.clear(); - (*this)._buffer = buffer; - } - } - - if err.is_retry() { - if file_type == FileType::File { - bun_core::debug_warn!( - "Received EAGAIN while reading from a file. This is a bug.", - ); - } else { - // SAFETY: caller contract; borrow scoped to the call. - unsafe { Self::register_poll(this) }; - } - return; - } - // SAFETY: caller contract; `on_error` is the tail. - unsafe { Self::on_error(this, err) }; - return; + (n, ReadState::Eof) + } + Some(Stop::WouldBlock) => { + if file_type != FileType::File { + // SAFETY: caller contract. + unsafe { Self::register_poll(this) }; } + (0, ReadState::Drained) } - } - } - - /// # Safety - /// Same contract as [`Self::read`]. - unsafe fn read_from_blocking_pipe_without_blocking( - this: *mut PosixBufferedReader, - fd: Fd, - size_hint: isize, - received_hup: bool, - ) { - // SAFETY: caller contract; borrow ends at `;`. - unsafe { - if (*this).vtable.is_streaming_enabled() { - (*this)._buffer.clear(); + Some(Stop::Error(err)) => { + // SAFETY: caller contract; `on_error` may free the parent. + unsafe { Self::on_error(this, err) }; + (0, ReadState::Progress) } } - - // SAFETY: caller contract. - unsafe { Self::read_blocking_pipe(this, fd, size_hint, received_hup) }; } } -// Keep boolean state in the `PosixFlags` bitflags field — no loose `bool` -// fields on `PosixBufferedReader`. +enum Stop { + Eof, + OverBudget, + WouldBlock, + Error(sys::Error), +} + +enum ReadOnce { + Read(usize, Option), + Stop(Stop), +} impl Drop for PosixBufferedReader { fn drop(&mut self) { @@ -1596,7 +1100,7 @@ impl WindowsBufferedReader { MaxBuf::on_read_bytes(maxbuf, bytes_read as u64) } - fn on_read_chunk(&mut self, buf: &[u8], has_more: ReadState) -> bool { + fn on_read_chunk(&mut self, has_more: ReadState) -> bool { if has_more == ReadState::Eof { self.flags.insert(WindowsFlags::RECEIVED_EOF); } @@ -1605,26 +1109,26 @@ impl WindowsBufferedReader { self.flags.remove(WindowsFlags::HAS_INFLIGHT_READ); return true; } - // PORT_NOTES_PLAN R-2: `&mut self` carries LLVM `noalias`, but - // `vtable.on_read_chunk` re-enters JS and user code can reach this - // reader via a fresh `&mut WindowsBufferedReader` from the parent's - // intrusive `reader` field, writing `self.flags` (e.g. via `pause` / - // `start_reading`). Not currently ASM-cached (noalias-hunt SUSPECT), - // but one inlining change away from caching `self.flags` across the - // call so the trailing `.remove(HAS_INFLIGHT_READ)` RMWs the stale - // pre-call value, clobbering any re-entrant flag change. Launder so - // the post-call RMW reloads through an opaque pointer; mirrors the - // cork fix at b818e70e1c57. + // `on_read_chunk` re-enters JS, which can reach this reader through its parent; go raw across the dispatch so nothing of `self` is cached over it. let this: *mut Self = core::hint::black_box(core::ptr::from_mut(self)); - // SAFETY: `this` aliases the live `&mut self`; single JS thread. The - // reader struct is an inline field of its parent (never freed - // mid-call), so `*this` stays a valid place across re-entry. - let result = unsafe { (*this).vtable.on_read_chunk(buf, has_more) }; - // Re-escape so the trailing RMW cannot reuse a spilled `self.flags` - // from before `on_read_chunk`. + // SAFETY: `this` aliases the live `&mut self`; the reader is an inline field of its parent (never freed mid-call). Borrows end at each `;`. + let (vtable, mut buffer) = unsafe { ((*this).vtable, mem::take(&mut (*this)._buffer)) }; + let result = if buffer.is_empty() { + true + } else if has_more == ReadState::Eof { + vtable.on_read_chunk(Chunk::Owned(buffer), has_more) + } else { + let result = vtable.on_read_chunk(Chunk::Buffer(&mut buffer), has_more); + buffer.clear(); + // SAFETY: `this` is still live (see above). + unsafe { + if (*this)._buffer.is_empty() { + (*this)._buffer = buffer; + } + } + result + }; core::hint::black_box(this); - // Clear has_inflight_read after the callback completes to prevent - // libuv from starting a new read while we're still processing data // SAFETY: `this` is still live (see above). unsafe { (*this).flags.remove(WindowsFlags::HAS_INFLIGHT_READ) }; result @@ -2260,47 +1764,28 @@ impl WindowsBufferedReader { sys::Result::Err(_) => unreachable!(), }; - #[cfg(debug_assertions)] - { - // Pointer-range check against `[ptr, ptr+capacity)` — can't form a - // `&[u8]` over spare capacity (uninit), so do it on addresses. - let base = self._buffer.as_ptr() as usize; - let end = base + self._buffer.capacity(); - let s = slice.as_ptr() as usize; - if !slice.is_empty() && !(s >= base && s + slice.len() <= end) { - panic!("uv_read_cb: buf is not in buffer! This is a bug in bun. Please report it."); - } - } + // Address arithmetic: `slice` covers spare (uninit) capacity, so no `&[u8]` over the Vec may be formed for the check. + debug_assert!( + slice.is_empty() + || (slice.as_ptr() as usize >= self._buffer.as_ptr() as usize + && slice.as_ptr() as usize + slice.len() + <= self._buffer.as_ptr() as usize + self._buffer.capacity()), + "uv_read_cb: buf is not in buffer! This is a bug in bun. Please report it." + ); // move cursor foward // SAFETY: slice is inside _buffer's spare capacity; libuv wrote `amount_result` bytes. unsafe { bun_core::vec::commit_spare(&mut self._buffer, amount_result) }; let over_budget = self.charge_max_buffer(amount_result); + let has_more = if over_budget { + ReadState::Eof + } else { + has_more + }; + // Parents that want the reader paused call `reader().pause()` themselves; stopping here could free a parent whose caller still holds `this` (FileResponseStream on abort). + let _ = self.on_read_chunk(has_more); - let should_continue = self.on_read_chunk(slice, has_more); - - // Streaming parents (shell IOReader, subprocess) cannot re-derive - // `&mut Self` from inside the vtable callback to restart the pipe - // (Stacked-Borrows; see the comment in shell/IOReader.rs). The re-arm - // is already handled by `on_file_read`'s epilogue / `uv_read_start`, - // but clearing the buffer here is load-bearing: without it `_buffer.len` - // grows by `amount_result` every chunk and never resets, so a 1 GB - // `cat` holds 1 GB resident instead of ~64 KB. Clear it here, after - // the streaming consumer has finished with `slice`. - // `should_continue` no longer gates the clear: FileReader may say - // stop at its highwater mark while uv keeps delivering, and leaving - // `_buffer` uncleared would double-buffer (here + FileReader.buffered). - // Parents that want the reader paused call `reader().pause()` - // themselves; stopping here could free a parent whose caller still - // holds `this` (FileResponseStream on abort). - let _ = should_continue; - if has_more != ReadState::Eof && self.vtable.is_streaming_enabled() { - self._buffer.clear(); - } - - // `over_budget` is terminal for the same reason EOF is: the child was - // killed and nothing past the cap may be buffered. if has_more == ReadState::Eof || over_budget { self.close(); } @@ -2322,6 +1807,16 @@ impl WindowsBufferedReader { // SAFETY: caller contract; borrow scoped to the call. unsafe { (*this).unpause() }; } + + /// Windows reads complete through libuv, never synchronously; this just makes sure one is in flight. + /// + /// # Safety + /// `this` is the live reader. + pub unsafe fn read_into(this: *mut Self, _dst: &mut [u8]) -> (usize, ReadState) { + // SAFETY: caller contract; borrow scoped to the call. + unsafe { (*this).unpause() }; + (0, ReadState::Progress) + } } // Keep boolean state in the `WindowsFlags` bitflags field — no loose `bool` diff --git a/src/io/lib.rs b/src/io/lib.rs index 82b8e55ea107..62e083df2b0c 100644 --- a/src/io/lib.rs +++ b/src/io/lib.rs @@ -326,7 +326,7 @@ bun_dispatch::link_interface! { cb: Option, ctx: Option>, ); - fn pipe_read_buffer() -> *mut [u8]; + fn pipe_read_scratch() -> *const PipeReadScratch; } } @@ -381,22 +381,11 @@ impl EventLoopCtx { // discipline above — see block comment. unsafe { &mut *self.file_polls_ptr() } } - /// Single nonnull-asref accessor for the per-loop pipe-read scratch - /// buffer. Same contract as [`loop_mut`]: `pub(crate)`, the buffer is a - /// per-thread set-once allocation owned by the VM/Mini loop, and the - /// event loop is single-threaded, so no second `&mut [u8]` to it can be - /// live. Every in-crate caller (`PipeReader::read_*`) uses it for one - /// blocking syscall and drops the borrow before re-entering the loop. - /// `'static` matches the unbounded lifetime the inline raw-ptr derefs at - /// the call sites already produced; collapses their N identical - /// `&mut *ctx.pipe_read_buffer()` derefs into this one block. + /// Claims the per-loop pipe-read scratch; `None` while a read further up the stack holds it. #[inline] - fn pipe_read_buffer_mut(&self) -> &'static mut [u8] { - // SAFETY: per-thread set-once scratch buffer (`BackRef`-shaped); the - // event loop is single-threaded so this is the sole live `&mut`, and - // every crate-internal caller drops the borrow before any path that - // could re-derive it — see doc comment above. - unsafe { &mut *self.pipe_read_buffer() } + fn claim_pipe_read_scratch(&self) -> Option> { + // SAFETY: per-thread scratch owned by the VM/Mini loop, which outlives every read. + unsafe { (*self.pipe_read_scratch()).claim() } } #[inline] pub(crate) fn loop_ref(&self) { @@ -471,12 +460,14 @@ pub mod heap; pub mod max_buf; #[path = "openForWriting.rs"] pub mod open_for_writing_mod; +pub mod pipe_read_scratch; #[path = "PipeReader.rs"] pub mod pipe_reader; #[path = "PipeWriter.rs"] pub mod pipe_writer; #[path = "pipes.rs"] pub mod pipes; +pub use pipe_read_scratch::{PipeReadScratch, PipeReadScratchGuard}; #[cfg(windows)] #[path = "source.rs"] pub mod source; @@ -490,7 +481,7 @@ pub mod write; pub use write::{AsFmt, DiscardingWriter, FixedBufferStream, FmtAdapter, IntLe, Result, Write}; pub use max_buf as MaxBuf; -pub use pipes::{FileType, ReadState}; +pub use pipes::{Chunk, FileType, ReadState}; // `BufferedReader` parent callback dispatch. Each variant's `link_impl_*!` (in // `bun_runtime`/`bun_install`) forwards to that type's `BufferedReaderParent` @@ -512,7 +503,7 @@ bun_dispatch::link_interface! { SecurityScan, ] { fn has_on_read_chunk() -> bool; - fn on_read_chunk(chunk: &[u8], has_more: pipes::ReadState) -> bool; + fn on_read_chunk(chunk: pipes::Chunk<'_>, has_more: pipes::ReadState) -> bool; fn on_reader_done(); fn on_reader_error(err: bun_sys::Error); fn loop_ptr() -> *mut Loop; @@ -595,7 +586,7 @@ macro_rules! __impl_buffered_reader_parent_body { #[allow(unused_unsafe, clippy::macro_metavars_in_unsafe)] unsafe fn on_read_chunk( $rc_this: *mut Self, - $rc_chunk: &[u8], + $rc_chunk: $crate::Chunk<'_>, $rc_more: $crate::ReadState, ) -> bool { unsafe { $rc } diff --git a/src/io/pipe_read_scratch.rs b/src/io/pipe_read_scratch.rs new file mode 100644 index 000000000000..9fd3667e459f --- /dev/null +++ b/src/io/pipe_read_scratch.rs @@ -0,0 +1,60 @@ +use core::cell::{Cell, UnsafeCell}; +use core::ops::{Deref, DerefMut}; + +pub const PIPE_READ_BUFFER_SIZE: usize = 256 * 1024; +type PipeReadBuffer = [u8; PIPE_READ_BUFFER_SIZE]; + +/// Per-loop scratch for blocking pipe/file reads. Chunks are delivered straight out of it, and a consumer may run user code that starts a nested read while still parsing the chunk, so only one borrower on the thread may hold it at a time. +pub struct PipeReadScratch { + in_use: Cell, + buffer: UnsafeCell>>, +} + +impl PipeReadScratch { + pub const fn new() -> Self { + Self { + in_use: Cell::new(false), + buffer: UnsafeCell::new(None), + } + } + + /// `None` while a borrower further up the stack still holds the guard. + pub fn claim(&self) -> Option> { + if self.in_use.replace(true) { + return None; + } + Some(PipeReadScratchGuard(self)) + } +} + +impl Default for PipeReadScratch { + fn default() -> Self { + Self::new() + } +} + +/// Exclusive claim on the scratch; released on drop. +pub struct PipeReadScratchGuard<'a>(&'a PipeReadScratch); + +impl Deref for PipeReadScratchGuard<'_> { + type Target = [u8]; + #[inline] + fn deref(&self) -> &[u8] { + // SAFETY: `in_use` is set, so this guard is the only accessor of `buffer` until it drops. + unsafe { &(*self.0.buffer.get()).get_or_insert_with(bun_core::boxed_zeroed)[..] } + } +} + +impl DerefMut for PipeReadScratchGuard<'_> { + #[inline] + fn deref_mut(&mut self) -> &mut [u8] { + // SAFETY: as in `deref`. + unsafe { &mut (*self.0.buffer.get()).get_or_insert_with(bun_core::boxed_zeroed)[..] } + } +} + +impl Drop for PipeReadScratchGuard<'_> { + fn drop(&mut self) { + self.0.in_use.set(false); + } +} diff --git a/src/io/pipes.rs b/src/io/pipes.rs index 9aab1226ab09..2afd6348cb3d 100644 --- a/src/io/pipes.rs +++ b/src/io/pipes.rs @@ -146,3 +146,47 @@ pub enum ReadState { /// Received an EAGAIN Drained, } + +/// One delivery from a `BufferedReader`. The variant says who owns the bytes, so a consumer never has to work that out from the pointer. +pub enum Chunk<'a> { + /// The loop's shared scratch: gone once `on_read_chunk` returns. Copy what you keep. + Scratch(&'a [u8]), + /// The reader's own buffer, which it clears and reuses after the call. Copy what you keep, or `take()` it when moving beats copying. + Buffer(&'a mut Vec), + /// The reader is finished with these bytes (EOF, error, budget): yours to move. + Owned(Vec), +} + +impl Chunk<'_> { + /// The bytes as an owned `Vec`, moving rather than copying where the variant allows. + pub fn take(self) -> Vec { + match self { + Chunk::Scratch(bytes) => bytes.to_vec(), + Chunk::Buffer(buffer) => core::mem::take(buffer), + Chunk::Owned(buffer) => buffer, + } + } + + pub fn is_owned(&self) -> bool { + matches!(self, Chunk::Owned(_)) + } + + pub fn truncate(&mut self, len: usize) { + match self { + Chunk::Scratch(bytes) => *bytes = &bytes[..len.min(bytes.len())], + Chunk::Buffer(buffer) => buffer.truncate(len), + Chunk::Owned(buffer) => buffer.truncate(len), + } + } +} + +impl core::ops::Deref for Chunk<'_> { + type Target = [u8]; + fn deref(&self) -> &[u8] { + match self { + Chunk::Scratch(bytes) => bytes, + Chunk::Buffer(buffer) => buffer, + Chunk::Owned(buffer) => buffer, + } + } +} diff --git a/src/jsc/JSValue.rs b/src/jsc/JSValue.rs index 018b1312a480..9bfbf3e85c18 100644 --- a/src/jsc/JSValue.rs +++ b/src/jsc/JSValue.rs @@ -926,10 +926,17 @@ impl JSValue { /// source attached rather than throwing. See `JSC__JSValue__pinArrayBuffer` /// in bindings.cpp for why. Release the pin with `ArrayBuffer::unpin`. pub fn as_pinned_arraybuffer(self, global: &JSGlobalObject) -> Option { - if !JSC__JSValue__pinArrayBuffer(self) { + let kind = JSC__JSValue__pinArrayBuffer(self); + if kind == 0 { return None; } - self.as_array_buffer(global) + let mut buffer = self.as_array_buffer(global); + match &mut buffer { + Some(buffer) => buffer.pinned = kind == 1, + None if kind == 1 => self.unpin_array_buffer(), + None => {} + } + buffer } /// Generic downcast. Dispatches via [`JsClass::from_js`]. #[inline] @@ -2093,7 +2100,8 @@ unsafe extern "C" { global: &JSGlobalObject, out: &mut ArrayBuffer, ) -> bool; - safe fn JSC__JSValue__pinArrayBuffer(this: JSValue) -> bool; + /// 0 = nothing to pin, 1 = pinned an ArrayBuffer (unpin later), 2 = held a bufferless view (nothing to unpin). + safe fn JSC__JSValue__pinArrayBuffer(this: JSValue) -> u8; safe fn JSC__JSValue__asPromise(this: JSValue) -> *mut JSPromise; safe fn JSC__JSValue__asInternalPromise(this: JSValue) -> *mut JSInternalPromise; safe fn Bun__attachAsyncStackFromPromise( diff --git a/src/jsc/VirtualMachine.rs b/src/jsc/VirtualMachine.rs index f7c0c80fb127..04e012f99fa1 100644 --- a/src/jsc/VirtualMachine.rs +++ b/src/jsc/VirtualMachine.rs @@ -1131,6 +1131,17 @@ impl VirtualMachine { self.rare_data.as_mut().unwrap() } + /// Raw projection to the lazily-allocated `RareData`, for callers holding a borrow into it across re-entry (never forms `&mut RareData`). + pub fn rare_data_ptr(&mut self) -> *mut RareData { + if self.rare_data.is_none() { + self.rare_data(); + } + match &mut self.rare_data { + Some(rd) => &raw mut **rd, + None => unreachable!(), + } + } + pub(crate) fn is_main_thread(&self) -> bool { self.worker.is_none() } @@ -2250,9 +2261,7 @@ bun_io::link_impl_EventLoopCtx! { vm.after_event_loop_callback = cb; vm.after_event_loop_callback_ctx = ctx.map(|p| p.as_ptr()); }, - pipe_read_buffer() => { - core::ptr::from_mut::<[u8]>(vm_from_owner(this.cast()).rare_data().pipe_read_buffer()) - }, + pipe_read_scratch() => &raw const *(*vm_from_owner(this.cast()).rare_data_ptr()).pipe_read_scratch, } } diff --git a/src/jsc/array_buffer.rs b/src/jsc/array_buffer.rs index e21dff3f6457..69947bfbd5a1 100644 --- a/src/jsc/array_buffer.rs +++ b/src/jsc/array_buffer.rs @@ -31,6 +31,8 @@ pub struct ArrayBuffer { /// True for resizable ArrayBuffer or growable SharedArrayBuffer — borrowing /// a slice from one is unsafe (it can shrink/reallocate underneath you). pub resizable: bool, + /// Set by [`JSValue::as_pinned_arraybuffer`] when an ArrayBuffer was actually pinned (as opposed to a bufferless view merely held); [`ArrayBuffer::unpin`] is a no-op otherwise. + pub pinned: bool, } impl Default for ArrayBuffer { @@ -43,6 +45,7 @@ impl Default for ArrayBuffer { typed_array_type: JSType::Cell, shared: false, resizable: false, + pinned: false, } } } @@ -138,8 +141,7 @@ unsafe extern "C" { } impl JSValue { - /// Releases a pin taken on this value's backing `JSC::ArrayBuffer` by - /// [`JSValue::as_pinned_arraybuffer`] or a pinning collector. + /// Releases a pin on this value's backing `JSC::ArrayBuffer`. Only for a value whose pin actually pinned a buffer; prefer [`ArrayBuffer::unpin`], which knows. pub fn unpin_array_buffer(self) { JSC__JSValue__unpinArrayBuffer(self); } @@ -150,9 +152,11 @@ impl ArrayBuffer { self.ptr.is_null() } - /// Releases the pin taken by [`JSValue::as_pinned_arraybuffer`]. + /// Releases the pin taken by [`JSValue::as_pinned_arraybuffer`], if it took one. pub fn unpin(&self) { - self.value.unpin_array_buffer(); + if self.pinned { + self.value.unpin_array_buffer(); + } } // require('buffer').kMaxLength. @@ -293,6 +297,7 @@ impl ArrayBuffer { typed_array_type: JSType::Uint8Array, shared: false, resizable: false, + pinned: false, }; // Via `#![feature(adt_const_params)]`: `JSType` derives `ConstParamTy`, so diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index 522e323496ad..d9a18e7c8b64 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/src/jsc/bindings/bindings.cpp @@ -3482,14 +3482,14 @@ bool JSC__JSValue__asArrayBuffer( } out->_value = JSValue::encode(value); out->ptr = static_cast(data); + out->pinned = false; return true; } -// Pin/unpin the backing ArrayBuffer of a JSArrayBuffer or JSArrayBufferView so -// its storage cannot move or be freed while a native borrower holds a slice -// into it. SharedArrayBuffer is never detachable and never moves, so it is left -// unpinned rather than rejected. Returns false if `value` has no ArrayBuffer -// impl. +// Pin/unpin the storage behind a JSArrayBuffer or JSArrayBufferView so it +// cannot move or be freed while a native borrower holds a slice into it. +// SharedArrayBuffer is never detachable and never moves, so it is left +// unpinned rather than rejected. Returns false if `value` has no storage. // // A pin does not make detaching fail, it makes it copy. `pin()` clears // `ArrayBuffer::isDetachable()`, and `ArrayBuffer::transferTo()` answers an @@ -3499,54 +3499,69 @@ bool JSC__JSValue__asArrayBuffer( // `port.postMessage(v, [ab])` each return normally, give the destination an // independent copy, and leave `ab` attached; the bytes being read never move. // -// That departs from ES2024, where transfer() must detach or throw, and from -// Node, which detaches. It is deliberate: the borrow stays zero-copy in the -// common case and memory-safe in every case, at the cost of a transfer that -// silently no-ops for as long as a borrowing op (zlib, fs, crypto, shell, -// Bun.Image, SQL blob binds, ...) happens to be in flight over that buffer. -static JSC::ArrayBuffer* arrayBufferImpl(JSC::JSValue value) -{ +// A view with no ArrayBuffer yet (`Buffer.allocUnsafeSlow`, `new Uint8Array(n)` +// past fastSizeLimit: OversizeTypedArray) is held, not adopted: materializing +// an ArrayBuffer just to pin it registers the bytes with the heap a second +// time and, because ArrayBuffers are only reclaimed by full collections, +// turns every threadpool fs/zlib/crypto op over a fresh Buffer into full-GC +// pressure. Such a view cannot be detached without JS first touching +// `.buffer`; if it does so mid-op the new ArrayBuffer is unpinned and a +// `transfer()` moves (does not free) the storage — the same window Node has. +// The caller keeps the returned kind and only calls unpin for `Pinned`; a +// held view is kept alive by the caller's own root, and nothing here needs +// undoing for it. +enum class PinKind : uint8_t { None = 0, + Pinned = 1, + Held = 2 }; +static PinKind pinStorage(JSC::JSValue value) +{ + JSC::ArrayBuffer* buf = nullptr; if (auto* jb = dynamicDowncast(value)) - return jb->impl(); - if (auto* view = dynamicDowncast(value)) - return view->possiblySharedBuffer(); - return nullptr; + buf = jb->impl(); + else if (auto* view = dynamicDowncast(value)) { + if (view->isDetached()) + return PinKind::None; + if (!view->hasArrayBuffer() && view->mode() == JSC::OversizeTypedArray) + return PinKind::Held; + buf = view->possiblySharedBuffer(); + } + if (!buf) + return PinKind::None; + if (!buf->isShared()) + buf->pin(); + return PinKind::Pinned; } -CPP_DECL bool JSC__JSValue__pinArrayBuffer(JSC::EncodedJSValue v) +CPP_DECL uint8_t JSC__JSValue__pinArrayBuffer(JSC::EncodedJSValue v) { - if (auto* buf = arrayBufferImpl(JSC::JSValue::decode(v))) { - if (!buf->isShared()) - buf->pin(); - return true; - } - return false; + return static_cast(pinStorage(JSC::JSValue::decode(v))); } +// Only for a value `pinStorage` answered `Pinned` for: that buffer still exists (pinned buffers are not detached). CPP_DECL void JSC__JSValue__unpinArrayBuffer(JSC::EncodedJSValue v) { - if (auto* buf = arrayBufferImpl(JSC::JSValue::decode(v))) { - if (!buf->isShared()) - buf->unpin(); - } + auto value = JSC::JSValue::decode(v); + JSC::ArrayBuffer* buf = nullptr; + if (auto* jb = dynamicDowncast(value)) + buf = jb->impl(); + else if (auto* view = dynamicDowncast(value); view && view->hasArrayBuffer()) + buf = view->possiblySharedBuffer(); + if (buf && !buf->isShared()) + buf->unpin(); } // Borrow `v`'s byte storage for off-thread reading. Splits out only the // `FastTypedArray` case from `pinArrayBuffer`, because that's the one mode // where `possiblySharedBuffer()` actually COPIES data // (`ArrayBuffer::tryCreate(span())`) — and it's ≤ fastSizeLimit elements, so -// the caller dupes instead. Every other mode either already has a real -// ArrayBuffer or, for `OversizeTypedArray`, is ADOPTED in-place by -// `slowDownAndWasteMemory()` (`ArrayBuffer::createAdopted` — wraps the -// existing fastMalloc pointer; zero byte copy, just a wrapper + butterfly -// alloc), so `possiblySharedBuffer()` + `pin()` is the right and cheap thing. -// Oversize MUST be pinned: once adopted (which JS can trigger via `.buffer` -// at any moment) it becomes detachable, and a `transfer()` would free the -// storage the worker is reading. +// the caller dupes instead. Every other mode goes through `pinStorage` (pin an +// existing ArrayBuffer, hold an OversizeTypedArray without adopting it). // // 0 Detached/null — nothing to read. // 1 `FastTypedArray` — ≤ fastSizeLimit elements, GC-movable. Caller // should dupe `out_ptr[0..out_len]`; no unpin. -// 2 Everything else — `pin()`ed via `possiblySharedBuffer()`; caller -// MUST `unpinArrayBuffer(v)` when done. +// 2 Pinned an existing ArrayBuffer; caller MUST `unpinArrayBuffer(v)` +// when done. +// 3 Held: a bufferless OversizeTypedArray; nothing to unpin, caller roots +// the value for the duration as it already does for 2. // // `out_ptr`/`out_len` describe the VIEW's byte range (offset+length). CPP_DECL int32_t JSC__JSValue__borrowBytesForOffThread(JSC::EncodedJSValue v, const uint8_t** out_ptr, size_t* out_len) @@ -3559,18 +3574,11 @@ CPP_DECL int32_t JSC__JSValue__borrowBytesForOffThread(JSC::EncodedJSValue v, co *out_len = view->byteLength(); return 1; } - // Oversize/Wasteful/DataView: possiblySharedBuffer() is either a - // getter or an in-place adopt (Oversize → createAdopted) — never a - // byte copy past this point. vector() is read AFTER because adoption - // can in principle repoint m_vector (it doesn't today, but the API - // contract allows it). - auto* buf = view->possiblySharedBuffer(); - if (!buf) return 0; - if (!buf->isShared()) - buf->pin(); + auto kind = pinStorage(view); + if (kind == PinKind::None) return 0; *out_ptr = static_cast(view->vector()); *out_len = view->byteLength(); - return 2; + return kind == PinKind::Held ? 3 : 2; } if (auto* jb = dynamicDowncast(value)) { auto* buf = jb->impl(); diff --git a/src/jsc/bindings/headers-handwritten.h b/src/jsc/bindings/headers-handwritten.h index 67e94e8c3deb..837f9615141b 100644 --- a/src/jsc/bindings/headers-handwritten.h +++ b/src/jsc/bindings/headers-handwritten.h @@ -335,6 +335,7 @@ typedef struct { uint8_t cell_type; bool shared; bool resizable; + bool pinned; } Bun__ArrayBuffer; #include "SyntheticModuleType.h" diff --git a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp index 8789b49d91d5..18dd7f6060ea 100644 --- a/src/jsc/bindings/webcore/streams/BunStreamSource.cpp +++ b/src/jsc/bindings/webcore/streams/BunStreamSource.cpp @@ -225,6 +225,7 @@ namespace WebStreams { using namespace JSC; using WebCore::JSBunStandaloneTextSink; +static constexpr size_t nativeSourceMinChunkSize = 64 * 1024; static constexpr size_t nativeSourceDefaultChunkSize = 256 * 1024; static constexpr size_t nativeSourceMaxChunkSize = 2 * 1024 * 1024; @@ -404,32 +405,37 @@ static void scheduleNativeSourceCallClose(JSGlobalObject* globalObject, JSNative queueStreamsMicrotask(globalObject, WebCore::JSStreamsRuntime::from(globalObject)->onNativeSourceCallCloseMicrotask(), jsUndefined(), adapter); } -static void nativeAdjustChunkSize(JSNativeStreamSourceAdapter* adapter, size_t resultBytes) +// Sizes the next slab to what the source actually delivers: one doubling when a pull fills the slab (files), +// and a shrink to the read size when it does not (pipes and sockets top out at 64-128 KiB per read), so that +// steady-state fills are whole and the slab is handed over rather than copied out of. +static void nativeAdjustChunkSize(JSNativeStreamSourceAdapter* adapter, size_t resultBytes, size_t slabBytes) { const size_t chunkSize = adapter->m_chunkSize; - if (resultBytes >= chunkSize && !adapter->m_hasResized) { + if (resultBytes >= slabBytes) { + if (!adapter->m_hasResized) { + adapter->m_hasResized = true; + adapter->m_chunkSize = std::min(chunkSize * 2, nativeSourceMaxChunkSize); + } + return; + } + if (resultBytes > 0 && resultBytes < chunkSize) { adapter->m_hasResized = true; - adapter->m_chunkSize = std::min(chunkSize * 2, nativeSourceMaxChunkSize); + adapter->m_chunkSize = std::max(WTF::roundUpToPowerOfTwo(resultBytes), nativeSourceMinChunkSize); } } -static JSC::JSUint8Array* uint8Subarray(JSGlobalObject* globalObject, JSC::JSUint8Array* view, size_t offset, size_t length) -{ - RefPtr buffer = view->possiblySharedBuffer(); - return JSC::JSUint8Array::create(globalObject, globalObject->typedArrayStructure(JSC::TypeUint8, false), WTF::move(buffer), view->byteOffset() + offset, length); -} - -// Reuse the pending view only when its BACKING BUFFER is large enough. +// The pending slab is only ever handed to JS whole, so it is reused as long as it is still big enough; the +// bytes are written by the source before anything reads them, so it need not be zeroed. static JSC::JSUint8Array* nativeGetInternalBuffer(JSC::VM& vm, JSGlobalObject* globalObject, JSNativeStreamSourceAdapter* adapter) { auto scope = DECLARE_THROW_SCOPE(vm); const size_t chunkSize = adapter->m_chunkSize; if (JSObject* pending = adapter->pendingView()) { auto* view = uncheckedDowncast(pending); - if (!view->isDetached() && view->possiblySharedBuffer() && view->possiblySharedBuffer()->byteLength() >= chunkSize) + if (!view->isDetached() && view->length() == chunkSize) return view; } - auto* fresh = JSC::JSUint8Array::create(globalObject, globalObject->typedArrayStructure(JSC::TypeUint8, false), chunkSize); + auto* fresh = JSC::JSUint8Array::createUninitialized(globalObject, globalObject->typedArrayStructure(JSC::TypeUint8, false), chunkSize); RETURN_IF_EXCEPTION(scope, nullptr); adapter->setPendingView(vm, fresh); return fresh; @@ -442,7 +448,7 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject, if (result.isNumber()) { double written = result.asNumber(); if (!isClosed) - nativeAdjustChunkSize(adapter, written > 0 ? static_cast(written) : 0); + nativeAdjustChunkSize(adapter, written > 0 ? static_cast(written) : 0, view ? view->length() : adapter->m_chunkSize); if (adapter->m_textMode) { if (written > 0 && view) { size_t count = std::min(static_cast(written), static_cast(view->length())); @@ -460,12 +466,14 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject, if (written > 0 && view) { size_t count = std::min(static_cast(written), static_cast(view->length())); JSC::JSArrayBufferView* toEnqueue = view; - if (view->length() - count > 0) { - toEnqueue = uint8Subarray(globalObject, view, 0, count); + if (count < view->length()) { + // A partial fill (pipes, sockets, the tail of a file) is copied out right-sized and the + // whole slab is reused for the next pull: no subarray views, and the slab is never adopted + // into an ArrayBuffer (which only full collections reclaim). A full fill hands over the slab. + auto* chunk = JSC::JSUint8Array::createUninitialized(globalObject, globalObject->typedArrayStructure(JSC::TypeUint8, false), count); RETURN_IF_EXCEPTION(scope, {}); - auto* tail = uint8Subarray(globalObject, view, count, view->length() - count); - RETURN_IF_EXCEPTION(scope, {}); - newView = tail; + memcpy(chunk->typedVector(), view->typedVector(), count); + toEnqueue = chunk; } else newView = jsUndefined(); if (controller) { @@ -484,8 +492,8 @@ static JSValue nativeDecodePullResult(JSC::VM& vm, JSGlobalObject* globalObject, return jsUndefined(); } if (auto* chunk = dynamicDowncast(result)) { - if (!isClosed) - nativeAdjustChunkSize(adapter, chunk->byteLength()); + if (!isClosed && chunk->byteLength() >= adapter->m_chunkSize) + nativeAdjustChunkSize(adapter, chunk->byteLength(), chunk->byteLength()); if (chunk->byteLength() > 0) { if (adapter->m_textMode) { nativeEnqueueTextChunk(globalObject, controller, adapter->m_textState, chunk->span(), /* flush */ false); diff --git a/src/jsc/event_loop.rs b/src/jsc/event_loop.rs index 4058f8e0b9b4..29da984079e3 100644 --- a/src/jsc/event_loop.rs +++ b/src/jsc/event_loop.rs @@ -332,14 +332,6 @@ impl EventLoop { result } - /// SAFETY: returns `&mut` into VM-owned scratch; two calls alias the same - /// buffer. Caller must not hold another live `&mut` to it. - pub unsafe fn pipe_read_buffer(&mut self) -> &mut [u8] { - // SAFETY: vm() is the live owning VM; rare_data() lazily inits the - // per-VM scratch buffer. Caller contract (see doc): no concurrent &mut. - unsafe { &mut (*self.vm()).rare_data().pipe_read_buffer()[..] } - } - pub fn drain_microtasks_with_global( &mut self, global_object: &JSGlobalObject, @@ -1377,7 +1369,6 @@ bun_event_loop::link_impl_JsEventLoop! { (*store).put(core::ptr::NonNull::new_unchecked(poll), ctx, was_ever_registered); }, uws_loop() => (*this).usockets_loop(), - pipe_read_buffer() => core::ptr::from_mut::<[u8]>((*this).pipe_read_buffer()), tick() => (*this).tick(), auto_tick() => (*this).auto_tick(), auto_tick_active() => (*this).auto_tick_active(), diff --git a/src/jsc/rare_data.rs b/src/jsc/rare_data.rs index 8476ef60cb1b..b51ca67ac4db 100644 --- a/src/jsc/rare_data.rs +++ b/src/jsc/rare_data.rs @@ -266,7 +266,7 @@ pub struct RareData { /// owns the data, no sidecar `Mutex<()>`). pub(crate) listening_sockets_for_watch_mode: Mutex>, - pub(crate) temp_pipe_read_buffer: Option>, + pub pipe_read_scratch: Box, /// `node:http2` PADDED DATA scratch; see [`Self::take_h2_padded_frame_buffer`]. h2_padded_frame_buffer: Option>, @@ -328,7 +328,7 @@ impl Default for RareData { node_fs_stat_watcher_scheduler: None, memory_pressure_watcher: None, listening_sockets_for_watch_mode: Mutex::new(Vec::new()), - temp_pipe_read_buffer: None, + pipe_read_scratch: Box::new(bun_event_loop::PipeReadScratch::new()), h2_padded_frame_buffer: None, s3_default_client: Strong::empty(), node_quic_callbacks: Strong::empty(), @@ -377,15 +377,6 @@ impl PathBuf { // Drop is automatic for Option> fields — no explicit deinit needed. -// ────────────────────────────────────────────────────────────────────────── -// PipeReadBuffer / constants -// ────────────────────────────────────────────────────────────────────────── - -// Canonical definition lives in the lower-tier `bun_event_loop` crate (shared -// with `MiniEventLoop`'s scratch buffer). Re-export so `rare_data::PipeReadBuffer` -// remains a stable path for existing callers. -pub use bun_event_loop::PipeReadBuffer; - /// One max-size HTTP/2 PADDED DATA frame payload (pad-length byte + data + padding). pub type H2PaddedFrameBuffer = [u8; 16384]; @@ -663,10 +654,6 @@ impl RareData { } // ── lazy-init: misc heap slots ──────────────────────────────────────── - pub fn pipe_read_buffer(&mut self) -> &mut PipeReadBuffer { - self.temp_pipe_read_buffer - .get_or_insert_with(bun_core::boxed_zeroed::) - } /// Take the padded-frame scratch out of its slot (lazily allocated). By value rather /// than borrowed: the socket write it feeds can re-enter JS and reach this path @@ -1072,7 +1059,7 @@ fn get_tls_default_ciphers_from_js( impl Drop for RareData { fn drop(&mut self) { - // temp_pipe_read_buffer / h2_padded_frame_buffer / spawn_sync_event_loop_ / + // pipe_read_scratch / h2_padded_frame_buffer / spawn_sync_event_loop_ / // s3_default_client / default_csrf_secret / cleanup_hooks / cron_jobs / // path_buf / tls_default_ciphers: // all dropped automatically via field Drop. diff --git a/src/runtime/api/bun/Terminal.rs b/src/runtime/api/bun/Terminal.rs index 1b890aedbf57..e181da0fbb13 100644 --- a/src/runtime/api/bun/Terminal.rs +++ b/src/runtime/api/bun/Terminal.rs @@ -2045,8 +2045,12 @@ impl BufferedReaderParent for Terminal { bun_io::BufferedReaderParentLinkKind::Terminal; const HAS_ON_READ_CHUNK: bool = true; - unsafe fn on_read_chunk(this: *mut Self, chunk: &[u8], has_more: ReadState) -> bool { - Self::from_parent_ptr(this).on_read_chunk(chunk, has_more) + unsafe fn on_read_chunk( + this: *mut Self, + chunk: bun_io::Chunk<'_>, + has_more: ReadState, + ) -> bool { + Self::from_parent_ptr(this).on_read_chunk(&chunk, has_more) } unsafe fn on_reader_done(this: *mut Self) { Self::from_parent_ptr(this).on_reader_done(); diff --git a/src/runtime/cli/filter_run.rs b/src/runtime/cli/filter_run.rs index fea69d8562b4..ac7a6abfcd33 100644 --- a/src/runtime/cli/filter_run.rs +++ b/src/runtime/cli/filter_run.rs @@ -297,7 +297,7 @@ impl<'a> ProcessHandle<'a> { bun_io::impl_buffered_reader_parent! { FilterRunHandle for ProcessHandle<'a>; has_on_read_chunk = true; - on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(chunk, has_more); + on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(&chunk, has_more); on_reader_done = |this| (*this).on_reader_done(); on_reader_error = |this, err| (*this).on_reader_error(&err); loop_ = |this| (*this).loop_(); diff --git a/src/runtime/cli/multi_run.rs b/src/runtime/cli/multi_run.rs index 23ee11cd59cb..2efd5f6180a8 100644 --- a/src/runtime/cli/multi_run.rs +++ b/src/runtime/cli/multi_run.rs @@ -83,7 +83,7 @@ bun_io::impl_buffered_reader_parent! { has_on_read_chunk = true; on_read_chunk = |this, chunk, _has_more| { let state = &mut *((*(*this).handle).state as *mut State); - let _ = state.read_chunk(&mut *this, chunk); + let _ = state.read_chunk(&mut *this, &chunk); true }; on_reader_done = |this| { diff --git a/src/runtime/cli/test/parallel/Worker.rs b/src/runtime/cli/test/parallel/Worker.rs index b7393f8918b9..7030ed3b3839 100644 --- a/src/runtime/cli/test/parallel/Worker.rs +++ b/src/runtime/cli/test/parallel/Worker.rs @@ -440,7 +440,7 @@ impl Default for WorkerPipe { bun_io::impl_buffered_reader_parent! { TestParallelWorkerPipe for WorkerPipe; has_on_read_chunk = true; - on_read_chunk = |this, chunk, state| (*this).on_read_chunk(chunk, state); + on_read_chunk = |this, chunk, state| (*this).on_read_chunk(&chunk, state); on_reader_done = |this| (*this).on_reader_done(); on_reader_error = |this, err| (*this).on_reader_error(err); // `vm.uv_loop()` is `*mut bun_io::Loop` on every target. diff --git a/src/runtime/image/Image.rs b/src/runtime/image/Image.rs index 5ddad10102cd..802d358f9998 100644 --- a/src/runtime/image/Image.rs +++ b/src/runtime/image/Image.rs @@ -763,10 +763,12 @@ impl Image { // existing fastMalloc storage in-place (zero byte copy); // pinning then keeps it alive even if JS does `.buffer` → // `transfer()` while the worker reads. - 2 => { + kind @ (2 | 3) => { if len == 0 { - // SAFETY: helper pinned `v`; unpin before erroring. - unsafe { JSC__JSValue__unpinArrayBuffer(v) }; + if kind == 2 { + // SAFETY: helper pinned `v`; unpin before erroring. + unsafe { JSC__JSValue__unpinArrayBuffer(v) }; + } Err(PinError::Detached) } else { // SAFETY: pinned until the returned `Pin` drops (with the job's @@ -777,7 +779,7 @@ impl Image { bytes: bun_ptr::RawSlice::new(bytes), ..Default::default() }, - Pin(v), + if kind == 2 { Pin(v) } else { Pin::NONE }, )) } } diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 4919642333e9..8fa740dc3c7b 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -7116,32 +7116,33 @@ impl NodeFS { // If we manage to read the entire file, we don't need to call stat() at all. // This will make it slightly slower to read e.g. 512 KB files, but usually the OS won't return a full 512 KB in one read anyway. // - // The sync case borrows `vm.rareData().pipeReadBuffer()` (a per-VM - // 256 KB heap slab) when a VM is present, otherwise leaves the buffer - // zero-length so the loop is skipped and we fall through to fstat. - // The async path heap-allocates a 256 KB buffer instead (Zig used a - // comptime-sized stack array; Rust cannot size a stack array on - // `flavor`, so heap is forced, but the slab stays uninitialised: it - // is write-only, `Syscall::read` hands it straight to the kernel). + // The sync case claims the per-VM pipe-read scratch when it is free; + // otherwise (async, no VM, or a read further up the stack holds it) + // a heap buffer stands in. It stays uninitialised: it is write-only, + // `Syscall::read` hands it straight to the kernel. use bun_collections::vec_ext::VecExt as _; - let mut async_stack_buffer: Vec = Vec::new(); - if flavor != Flavor::Sync && async_stack_buffer.try_reserve_exact(256 * 1024).is_ok() { + let mut scratch = match self.vm { + // SAFETY: `self.vm` is the live owning `*mut VirtualMachine` (single-threaded VM), which outlives this call. + Some(vm) if flavor == Flavor::Sync => unsafe { + (*(*vm.as_ptr()).rare_data_ptr()).pipe_read_scratch.claim() + }, + _ => None, + }; + let mut heap_buffer: Vec = Vec::new(); + if scratch.is_none() && heap_buffer.try_reserve_exact(256 * 1024).is_ok() { // SAFETY: `u8` has no validity invariant; the buffer is handed // straight to the kernel which only stores into it. Only the // `[..total]` prefix actually filled by `read` is ever observed. - unsafe { async_stack_buffer.expand_to_capacity() }; + unsafe { heap_buffer.expand_to_capacity() }; } - let pre_stat_buf: &mut [u8] = if flavor == Flavor::Sync { - match self.vm { - // SAFETY: `self.vm` is the live owning `*mut VirtualMachine`; - // `rare_data()` lazily inits the heap slab and the returned - // `&mut [u8; 256*1024]` outlives this call (single-threaded VM). - Some(vm) => unsafe { &mut (*vm.as_ptr()).rare_data().pipe_read_buffer()[..] }, - None => &mut [][..], - } - } else { - &mut async_stack_buffer[..] + let pre_stat_buf: &mut [u8] = match scratch.as_mut() { + Some(scratch) => &mut scratch[..], + None => &mut heap_buffer[..], }; + let pre_stat_len = pre_stat_buf + .len() + .min(args.max_size.map_or(usize::MAX, |v| v as usize)); + let pre_stat_buf = &mut pre_stat_buf[..pre_stat_len]; let temporary_read_buffer_before_stat_call: &[u8] = { let mut available: &mut [u8] = &mut pre_stat_buf[..]; while !available.is_empty() { @@ -7163,9 +7164,7 @@ impl NodeFS { if let Some(vm) = self.vm.map(bun_ptr::BackRef::from) { // Attempt to create the buffer in JSC's heap. // This avoids creating a WastefulTypedArray. - // `self.vm` is the live owning `VirtualMachine` - // (per-thread singleton; see `pipe_read_buffer` - // above) — `BackRef` invariant holds. + // `self.vm` is the live owning `VirtualMachine` (per-thread singleton) — `BackRef` invariant holds. let global = vm.global(); let Ok(array_buffer) = bun_jsc::ArrayBuffer::create_buffer( global, diff --git a/src/runtime/node/node_zlib_binding.rs b/src/runtime/node/node_zlib_binding.rs index e03aabe80f99..6b74caaba9e6 100644 --- a/src/runtime/node/node_zlib_binding.rs +++ b/src/runtime/node/node_zlib_binding.rs @@ -424,19 +424,20 @@ impl CompressionStream { // Pin both buffers before mutating any state: materializing a // FastTypedArray's backing store can fail on OOM, and failing here // leaves nothing to unwind. - let in_buf: jsc::ArrayBuffer; - let in_: Option<&[u8]> = if arguments[1].is_null() { + let in_buf: Option = if arguments[1].is_null() { None } else { let Some(buf) = arguments[1].as_pinned_arraybuffer(global_this) else { return Err(global_this.throw_out_of_memory()); }; - in_buf = buf; - Some(&in_buf.byte_slice()[in_off as usize..in_off as usize + in_len as usize]) + Some(buf) }; + let in_: Option<&[u8]> = in_buf + .as_ref() + .map(|b| &b.byte_slice()[in_off as usize..in_off as usize + in_len as usize]); let Some(mut out_buf) = arguments[4].as_pinned_arraybuffer(global_this) else { - if !arguments[1].is_null() { - arguments[1].unpin_array_buffer(); + if let Some(buf) = &in_buf { + buf.unpin(); } return Err(global_this.throw_out_of_memory()); }; diff --git a/src/runtime/server/FileResponseStream.rs b/src/runtime/server/FileResponseStream.rs index af51dd28e600..32278f830b49 100644 --- a/src/runtime/server/FileResponseStream.rs +++ b/src/runtime/server/FileResponseStream.rs @@ -605,7 +605,7 @@ bun_io::impl_buffered_reader_parent! { has_on_read_chunk = true; on_read_chunk = |this, chunk, state| { let _guard = bun_ptr::ScopedRef::::new(this); - (*this).on_read_chunk(chunk, state) + (*this).on_read_chunk(&chunk, state) }; on_reader_done = |this| { let _guard = bun_ptr::ScopedRef::::new(this); diff --git a/src/runtime/server/NodeHTTPResponse.rs b/src/runtime/server/NodeHTTPResponse.rs index 118f74cc473f..aa8cfd2873cc 100644 --- a/src/runtime/server/NodeHTTPResponse.rs +++ b/src/runtime/server/NodeHTTPResponse.rs @@ -2138,11 +2138,11 @@ impl NodeHTTPResponse { let pinned_value = if is_buffer && input_value.is_cell() { match input_value.as_pinned_arraybuffer(global_object) { Some(ab) if ab.resizable && !ab.shared => { - input_value.unpin_array_buffer(); + ab.unpin(); None } - Some(_) => Some(input_value), - None => Some(JSValue::ZERO), + Some(ab) if ab.pinned => Some(input_value), + Some(_) | None => Some(JSValue::ZERO), } } else { Some(JSValue::ZERO) diff --git a/src/runtime/shell/IOReader.rs b/src/runtime/shell/IOReader.rs index 3c02f7ce4759..0484e6ed5c2b 100644 --- a/src/runtime/shell/IOReader.rs +++ b/src/runtime/shell/IOReader.rs @@ -365,7 +365,7 @@ impl IOReader { bun_io::impl_buffered_reader_parent! { ShellIoReader for IOReader; has_on_read_chunk = true; - on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk_cb(chunk, has_more); + on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk_cb(&chunk, has_more); on_reader_done = |this| (*this).on_reader_done_cb(); on_reader_error = |this, err| (*this).on_reader_error(&err); loop_ = |this| (*this).io_evtloop().native_loop(); diff --git a/src/runtime/shell/subproc.rs b/src/runtime/shell/subproc.rs index 2a934c93e0b5..feeef44d8815 100644 --- a/src/runtime/shell/subproc.rs +++ b/src/runtime/shell/subproc.rs @@ -2227,7 +2227,7 @@ impl Drop for PipeReader { bun_io::impl_buffered_reader_parent! { ShellPipeReader for PipeReader; has_on_read_chunk = true; - on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(chunk, has_more); + on_read_chunk = |this, chunk, has_more| (*this).on_read_chunk(&chunk, has_more); on_reader_done = |this| PipeReader::on_reader_done(this); on_reader_error = |this, err| PipeReader::on_reader_error(this, &err); loop_ = |this| (*this).r#loop(); diff --git a/src/runtime/webcore/CompressionStreamCoder.rs b/src/runtime/webcore/CompressionStreamCoder.rs index 2115bcc30545..eaa3394861c9 100644 --- a/src/runtime/webcore/CompressionStreamCoder.rs +++ b/src/runtime/webcore/CompressionStreamCoder.rs @@ -584,13 +584,13 @@ unsafe impl Send for AsyncInput {} /// The pin + GC protection on a chunk whose bytes went to the pool; released /// on drop (JS thread, with the job's Js side). -pub(crate) struct PinnedChunk(JSValue); +pub(crate) struct PinnedChunk(bun_jsc::ArrayBuffer); // SAFETY: pin/protect on a heap cell; gone with the heap. unsafe impl bun_jsc::job::JsAffine for PinnedChunk {} impl Drop for PinnedChunk { fn drop(&mut self) { - self.0.unpin_array_buffer(); - self.0.unprotect(); + self.0.unpin(); + self.0.value.unprotect(); } } @@ -605,7 +605,7 @@ impl AsyncInput { // A resizable non-shared backing can `mprotect()` pages out on // `resize()`; pinning does not block that, so spill to a copy. if buf.resizable && !buf.shared { - chunk.unpin_array_buffer(); + buf.unpin(); return (Self::Owned(fallback.to_vec()), None); } chunk.protect(); @@ -614,7 +614,7 @@ impl AsyncInput { ptr: buf.ptr, len: buf.byte_len, }, - Some(PinnedChunk(chunk)), + Some(PinnedChunk(buf)), ); } (Self::Owned(fallback.to_vec()), None) diff --git a/src/runtime/webcore/FileReader.rs b/src/runtime/webcore/FileReader.rs index 35754fe0caae..6af15b077aa0 100644 --- a/src/runtime/webcore/FileReader.rs +++ b/src/runtime/webcore/FileReader.rs @@ -6,7 +6,7 @@ use bun_collections::VecExt; use bun_io as aio; #[cfg(not(windows))] use bun_io::FileType; -use bun_io::{BufferedReader, ReadState}; +use bun_io::{BufferedReader, Chunk, ReadState}; use bun_jsc::JsCell; use bun_ptr::AsCtxPtr; use bun_sys::{self as sys, Fd, FdExt}; @@ -21,11 +21,6 @@ use crate::webcore::streams; bun_core::declare_scope!(FileReader, visible); -// `pending_view` and the `Js`/`Temporary` variants below borrow into a -// JS-owned typed-array buffer kept alive by `pending_value: Strong` / `ensure_still_alive`. -// Represented as unbounded `&mut [u8]` / `&[u8]` here to keep function bodies -// readable; TODO(refactor): replace with a proper raw-slice wrapper (BACKREF lifetime). - // R-2 (host-fn re-entrancy): every JS-exposed / vtable-reachable method takes // `&self`; per-field interior mutability via `Cell` (Copy) / `JsCell` (non- // Copy). The `SourceContext` trait and `BufferedReaderParent` shims still @@ -62,7 +57,6 @@ pub struct FileReader { pub(crate) event_loop: Cell, pub(crate) lazy: JsCell, pub(crate) buffered: JsCell>, - pub(crate) read_inside_on_pull: JsCell, /// Read-only after construction. pub(crate) highwater_mark: usize, pub(crate) flowing: Cell, @@ -91,7 +85,6 @@ impl Default for FileReader { event_loop: Cell::new(EventLoopHandle::init(core::ptr::null_mut())), lazy: JsCell::new(Lazy::None), buffered: JsCell::new(Vec::new()), - read_inside_on_pull: JsCell::new(ReadDuringJSOnPullResult::None), highwater_mark: 16384, flowing: Cell::new(true), sink: JsCell::new(SinkHandle::None), @@ -102,26 +95,6 @@ impl Default for FileReader { pub type IOReader = BufferedReader; -#[derive(strum::IntoStaticStr)] -pub enum ReadDuringJSOnPullResult { - None, - // TODO(refactor): `&'static mut` forge — sibling `static-widen-mut` pattern; - // see note on `FileReader::pending_view`. - Js(&'static mut [u8]), - AmountRead(usize), - /// Borrows the reader/JS buffer for the duration of one `on_pull` call - /// only. Holder-lifetime, not process-lifetime — `RawSlice` per - /// `bun_ptr::Interned` Population-B triage. - Temporary(bun_ptr::RawSlice), - UseBuffered(usize), -} - -impl ReadDuringJSOnPullResult { - fn is_none(&self) -> bool { - matches!(self, Self::None) - } -} - pub enum Lazy { None, /// Intrusively-refcounted `*Blob.Store`. Uses `StoreRef` (not `Arc`) so the @@ -647,14 +620,12 @@ impl FileReader { true } - pub(crate) fn on_read_chunk(&self, init_buf: &[u8], state: ReadState) -> bool { - let mut buf = init_buf; + pub(crate) fn on_read_chunk(&self, mut chunk: Chunk<'_>, state: ReadState) -> bool { bun_core::scoped_log!( FileReader, - "onReadChunk() = {} ({}) - read_inside_on_pull: {}", - buf.len(), - read_state_tag(state), - <&'static str>::from(self.read_inside_on_pull.get()) + "onReadChunk() = {} ({})", + chunk.len(), + read_state_tag(state) ); if self.done.get() { @@ -662,300 +633,155 @@ impl FileReader { return false; } let mut close = false; - // The close-on-exit is handled at each return - // site below via `close_if_needed` (a scopeguard would alias &mut self). - macro_rules! close_if_needed { - () => { - if close { - self.reader().close(); - } - }; - } let mut has_more = state != ReadState::Eof; - - if !buf.is_empty() { - if let Some(max_size) = self.max_size { - let total_readed = self.total_readed.get(); - if total_readed >= max_size { - return false; - } - let len = (max_size - total_readed).min(buf.len()); - if buf.len() > len { - buf = &buf[0..len]; - } - self.total_readed.set(total_readed + len); - - if buf.is_empty() { - close = true; - has_more = false; - } + if let (Some(max_size), false) = (self.max_size, chunk.is_empty()) { + let total_readed = self.total_readed.get(); + if total_readed >= max_size { + return false; + } + let len = (max_size - total_readed).min(chunk.len()); + chunk.truncate(len); + self.total_readed.set(total_readed + len); + if len == 0 { + close = true; + has_more = false; } } - // Kept as a RAW `*mut Vec` for the lifetime of this fn — never bound to a - // long-lived `&mut Vec`. `reader_buffer` points inside `self.reader` while - // we still hold `&self` and mutate `self.buffered`/`self.pending` etc. - // interleaved with reads/clears of `*reader_buffer`. Holding a `&mut Vec` here - // would be the aliased-&mut forbidden pattern (PORTING.md §Forbidden patterns). - // Use a raw ptr - // and deref only at the exact use sites below. - let reader_buffer: *mut Vec = self.reader().buffer(); - - // Native sink fast-path: bytes go straight to the attached sink, - // bypassing the JS `pending` / `read_inside_on_pull` machinery. let sink = *self.sink.get(); - if sink.is_some() { - if !buf.is_empty() { - let chunk = if has_more { - streams::Result::Temporary(bun_ptr::RawSlice::new(buf)) - } else { - streams::Result::TemporaryAndDone(bun_ptr::RawSlice::new(buf)) - }; - let wrote = sink.write(&chunk); - // SAFETY: see `reader_buffer` decl — tight deref, no `&mut` held. - if is_slice_in_vec_capacity(buf, unsafe { &*reader_buffer }) { - // SAFETY: see `reader_buffer` decl. - unsafe { (*reader_buffer).clear() }; - } - match wrote { - streams::Writable::Backpressure(_) => { - // Returning `false` ends a synchronous read loop; an - // event-driven reader (Windows, pollable fds) has to - // be paused, or its next completion lands here again - // and piles into the sink. `pull_into_sink` unpauses. - self.sink_paused.set(true); - self.reader().pause(); - close_if_needed!(); - return false; - } - streams::Writable::Err(e) => { - self.sink.set(SinkHandle::None); - sink.end(Some(streams::StreamError::Error(e))); - close_if_needed!(); - return false; - } - streams::Writable::Done => { - self.sink.set(SinkHandle::None); - sink.end(None); - close_if_needed!(); - return false; - } - _ => {} - } + let keep_going = if sink.is_some() { + self.write_chunk_to_sink(sink, &chunk, has_more) + } else if self.pending.get().state == streams::PendingState::Pending { + // Pipes may return 0-byte reads short of EOF; keep reading. + if chunk.is_empty() && state == ReadState::Drained { + true + } else { + self.resolve_pending_read(chunk, has_more) } - if !has_more && self.sink.get().is_some() { - self.sink.set(SinkHandle::None); - sink.end(None); + } else { + if self.buffered.get().is_empty() && chunk.is_owned() { + self.buffered.set(chunk.take()); + } else { + self.buffered.with_mut(|b| b.extend_from_slice(&chunk)); } - close_if_needed!(); - return has_more; + // No JS read is waiting; stop at the highwater mark and let onPull restart. `started` gates it: a non-lazy `Bun.spawn` pipe is already reading before any consumer attaches, and throttling then deadlocks a child alternating stdout/stderr writes. + let keep_going = !self.started.get() + || (self.flowing.get() && self.buffered.get().len() < self.highwater_mark); + // A completion-driven reader keeps issuing reads unless stopped; `on_pull` restarts it. + #[cfg(windows)] + if !keep_going { + self.reader().pause(); + } + keep_going + }; + if close { + self.reader().close(); } + keep_going + } - if !self.read_inside_on_pull.get().is_none() { - // R-2: `with_mut` projects `&mut ReadDuringJSOnPullResult` from - // `&self`; `self.buffered` is a disjoint `JsCell` so nested access - // inside the closure is sound. - self.read_inside_on_pull.with_mut(|riop| match riop { - ReadDuringJSOnPullResult::Js(in_progress) => { - if in_progress.len() >= buf.len() && !has_more { - in_progress[0..buf.len()].copy_from_slice(buf); - let remaining: *mut [u8] = &raw mut in_progress[buf.len()..]; - // SAFETY: lifetime laundering — see the `static-widen-mut` note on `ReadDuringJSOnPullResult::Js`. - let remaining = unsafe { &mut *remaining }; - *riop = ReadDuringJSOnPullResult::Js(remaining); - } else if !in_progress.is_empty() && !has_more { - // `buf` outlives the `on_pull` call that consumes this - // variant; holder-lifetime, encoded as `RawSlice`. - *riop = ReadDuringJSOnPullResult::Temporary(bun_ptr::RawSlice::new(buf)); - } else if has_more && !is_slice_in_vec_capacity(buf, self.buffered.get()) { - self.buffered.with_mut(|b| b.extend_from_slice(buf)); - *riop = ReadDuringJSOnPullResult::UseBuffered(buf.len()); - } - } - ReadDuringJSOnPullResult::UseBuffered(original) => { - let original = *original; - self.buffered.with_mut(|b| b.extend_from_slice(buf)); - *riop = ReadDuringJSOnPullResult::UseBuffered(buf.len() + original); - } - ReadDuringJSOnPullResult::None => unreachable!(), - _ => panic!("Invalid state"), + fn write_chunk_to_sink(&self, sink: SinkHandle, chunk: &[u8], has_more: bool) -> bool { + if !chunk.is_empty() { + let chunk = bun_ptr::RawSlice::new(chunk); + let wrote = sink.write(&if has_more { + streams::Result::Temporary(chunk) + } else { + streams::Result::TemporaryAndDone(chunk) }); - } else if self.pending.get().state == streams::PendingState::Pending { - // Certain readers (such as pipes) may return 0-byte reads even when - // not at EOF. Consequently, we need to check whether the reader is - // actually done or not. - if buf.is_empty() && state == ReadState::Drained { - // If the reader is not done, we still want to keep reading. - close_if_needed!(); - return true; - } - - // A labeled block computes `ret`, then cleanup + run + return. - let ret: bool = 'pending: { - let global = self.parent_global(); - let mut pending_array_buffer = self - .pending_value - .get() - .get() - .and_then(|view| view.as_array_buffer(&global)) - .unwrap_or_default(); - let pending_buf = pending_array_buffer.slice_mut(); - if buf.is_empty() { - if self.buffered.get().is_empty() { - self.buffered.set(Vec::new()); // clearAndFree - // SAFETY: see `reader_buffer` decl — tight deref, no &mut held across. - self.buffered.set(unsafe { mem::take(&mut *reader_buffer) }); // moveToUnmanaged - } - - // nested `defer buffer.clearAndFree` folded into the arms. - let buffer = self.buffered.replace(Vec::new()); - if !buffer.is_empty() { - if pending_buf.len() >= buffer.len() { - pending_buf[0..buffer.len()].copy_from_slice(&buffer); - self.pending.with_mut(|p| { - p.result = streams::Result::IntoArrayAndDone(streams::IntoArray { - value: self.pending_value.get().get().unwrap_or_default(), - len: buffer.len() as u64, // @truncate - }) - }); - drop(buffer); // clearAndFree - } else { - self.pending.with_mut(|p| { - p.result = - streams::Result::OwnedAndDone(Vec::::move_from_list(buffer)) - }); - } - } else { - self.pending.with_mut(|p| p.result = streams::Result::Done); - } - break 'pending false; - } - - let was_done = self.reader().is_done(); - - if pending_buf.len() >= buf.len() { - pending_buf[0..buf.len()].copy_from_slice(buf); - // SAFETY: see `reader_buffer` decl. - unsafe { (*reader_buffer).clear() }; - self.buffered.with_mut(|b| b.clear()); - - let into_array = streams::IntoArray { - value: self.pending_value.get().get().unwrap_or_default(), - len: buf.len() as u64, // @truncate - }; - - self.pending.with_mut(|p| { - p.result = if was_done { - streams::Result::IntoArrayAndDone(into_array) - } else { - streams::Result::IntoArray(into_array) - } - }); - break 'pending !was_done; + match wrote { + streams::Writable::Backpressure(_) => { + // Returning `false` ends a synchronous read loop; an event-driven reader (Windows, pollable fds) has to be paused or its next completion piles into the sink. `pull_into_sink` unpauses. + self.sink_paused.set(true); + self.reader().pause(); + return false; } - - // SAFETY: see `reader_buffer` decl — tight deref. - if is_slice_in_vec_capacity(buf, unsafe { &*reader_buffer }) { - if self.reader().is_done() { - // SAFETY: see `reader_buffer` decl. - debug_assert_eq!(buf.as_ptr(), unsafe { (*reader_buffer).as_ptr() }); - // SAFETY: see `reader_buffer` decl — tight deref, no `&mut` held across. - let mut buffer = unsafe { mem::take(&mut *reader_buffer) }; - buffer.truncate(buf.len()); // shrinkRetainingCapacity - self.pending.with_mut(|p| { - p.result = - streams::Result::OwnedAndDone(Vec::::move_from_list(buffer)) - }); - } else { - // SAFETY: see `reader_buffer` decl. - unsafe { (*reader_buffer).clear() }; - self.pending.with_mut(|p| { - p.result = streams::Result::Temporary(bun_ptr::RawSlice::new(buf)) - }); - } - break 'pending !was_done; + streams::Writable::Err(e) => { + self.sink.set(SinkHandle::None); + sink.end(Some(streams::StreamError::Error(e))); + return false; } - - if !is_slice_in_vec_capacity(buf, self.buffered.get()) { - self.pending.with_mut(|p| { - p.result = if self.reader().is_done() { - streams::Result::TemporaryAndDone(bun_ptr::RawSlice::new(buf)) - } else { - streams::Result::Temporary(bun_ptr::RawSlice::new(buf)) - } - }); - break 'pending !was_done; + streams::Writable::Done => { + self.sink.set(SinkHandle::None); + sink.end(None); + return false; } + _ => {} + } + } + if !has_more && self.sink.get().is_some() { + self.sink.set(SinkHandle::None); + sink.end(None); + } + has_more + } - debug_assert_eq!(buf.as_ptr(), self.buffered.get().as_ptr()); - let mut buffered = self.buffered.replace(Vec::new()); - buffered.truncate(buf.len()); // shrinkRetainingCapacity - - self.pending.with_mut(|p| { - p.result = if self.reader().is_done() { - streams::Result::OwnedAndDone(Vec::::move_from_list(buffered)) - } else { - streams::Result::Owned(Vec::::move_from_list(buffered)) - } - }); - break 'pending !was_done; + /// Settles the parked JS read with `chunk` (invariant: a parked read means `buffered` was already drained into it). + fn resolve_pending_read(&self, chunk: Chunk<'_>, has_more: bool) -> bool { + let was_done = self.reader().is_done(); + let global = self.parent_global(); + let mut pending_array_buffer = self + .pending_value + .get() + .get() + .and_then(|view| view.as_array_buffer(&global)) + .unwrap_or_default(); + let pending_buf = pending_array_buffer.slice_mut(); + let ret = if chunk.is_empty() { + let buffered = self.buffered.replace(Vec::new()); + let result = if buffered.is_empty() { + streams::Result::Done + } else if pending_buf.len() >= buffered.len() { + pending_buf[..buffered.len()].copy_from_slice(&buffered); + streams::Result::IntoArrayAndDone(streams::IntoArray { + value: self.pending_value.get().get().unwrap_or_default(), + len: buffered.len() as u64, + }) + } else { + streams::Result::OwnedAndDone(buffered) }; - - self.pending_value - .with_mut(|p| p.clear_without_deallocation()); - self.pending_view.set(&mut []); - // Pin across `p.run()`: a re-entrant cancel() reaches - // on_reader_done, which drops the across-read ref and lets a GC - // free this box while the io caller still holds `&mut` into it. - let parent = self.parent(); - // SAFETY: see `parent()`. - unsafe { (*parent).increment_count() }; - self.pending.with_mut(|p| p.run()); - close_if_needed!(); - // Re-entrant cancel (sets `done`) or a nested on_pull that read to - // EOF (sets IS_DONE via on_reader_done but not `self.done`) closed - // the reader; tell the io caller to stop so it does not re-read the - // captured fd. - let ret = if self.done.get() || self.reader().is_done() { - false + self.pending.with_mut(|p| p.result = result); + false + } else { + let result = if pending_buf.len() >= chunk.len() { + pending_buf[..chunk.len()].copy_from_slice(&chunk); + let into = streams::IntoArray { + value: self.pending_value.get().get().unwrap_or_default(), + len: chunk.len() as u64, + }; + if was_done { + streams::Result::IntoArrayAndDone(into) + } else { + streams::Result::IntoArray(into) + } + } else if chunk.is_owned() || !has_more { + let owned = chunk.take(); + if was_done { + streams::Result::OwnedAndDone(owned) + } else { + streams::Result::Owned(owned) + } } else { - ret + // Copied into a fresh Uint8Array by `run()` below, before this returns. + streams::Result::Temporary(bun_ptr::RawSlice::new(&chunk)) }; - // SAFETY: see `parent()`; the pin keeps the count >= 1, so this - // never frees. `self` is not accessed after. - let _ = unsafe { Source::decrement_count(parent) }; - return ret; - } else if !is_slice_in_vec_capacity(buf, self.buffered.get()) { - self.buffered.with_mut(|b| b.extend_from_slice(buf)); - // SAFETY: see `reader_buffer` decl. - if is_slice_in_vec_capacity(buf, unsafe { &*reader_buffer }) { - // SAFETY: see `reader_buffer` decl. - unsafe { (*reader_buffer).clear() }; - } - } - - // No JS read is waiting; stop at the highwater mark. onPull restarts. - // - // `started` gates the backstop: a `from_pipe()` reader for non-lazy - // `Bun.spawn` is already reading when it arrives here, and throttling - // before any consumer has attached deadlocks a child that alternates - // stdout/stderr writes while the caller only awaits one of them. - // SAFETY: see `reader_buffer` decl. - let reader_buffer_len = unsafe { (*reader_buffer).len() }; - let ret = !matches!( - self.read_inside_on_pull.get(), - ReadDuringJSOnPullResult::Temporary(_) - ) && (!self.started.get() - || (self.flowing.get() - && self.buffered.get().len() + reader_buffer_len < self.highwater_mark)); - close_if_needed!(); + self.pending.with_mut(|p| p.result = result); + !was_done + }; + self.pending_value + .with_mut(|p| p.clear_without_deallocation()); + self.pending_view.set(&mut []); + // Pin across `run()`: a re-entrant cancel() reaches on_reader_done, which drops the across-read ref and lets a GC free this box while the io caller still holds `&mut` into it. + let parent = self.parent(); + // SAFETY: see `parent()`. + unsafe { (*parent).increment_count() }; + self.pending.with_mut(|p| p.run()); + // Re-entrant cancel or a nested pull that read to EOF closed the reader; tell the io caller to stop so it does not re-read the captured fd. + let ret = ret && !self.done.get() && !self.reader().is_done(); + // SAFETY: see `parent()`; the pin keeps the count >= 1, so this never frees. `self` is not accessed after. + let _ = unsafe { Source::decrement_count(parent) }; ret } - fn is_pulling(&self) -> bool { - !self.read_inside_on_pull.get().is_none() - } - pub(crate) fn on_pull(&self, buffer: &'static mut [u8], array: JSValue) -> streams::Result { // `buffer` borrows a JS typed array kept alive by `array`. array.ensure_still_alive(); @@ -1001,95 +827,33 @@ impl FileReader { return streams::Result::Done; } - if !self.reader().has_pending_read() { - // If not flowing (paused), don't initiate new reads - if !self.flowing.get() { - bun_core::scoped_log!( - FileReader, - "onPull({}) = pending (not flowing)", - buffer.len() - ); - let global = self.parent_global(); - self.pending_value.with_mut(|p| p.set(&global, array)); - self.pending_view.set(buffer); - return streams::Result::Pending(self.pending.as_ptr()); + if !self.reader().has_pending_read() && self.flowing.get() { + // SAFETY: the reader cell is live for `self`'s lifetime; `read_into` is the raw re-entrancy-safe entry (EOF/error dispatch runs user JS). + let (amount_read, state) = unsafe { IOReader::read_into(self.reader.get(), buffer) }; + bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer.len(), amount_read); + let done = state == ReadState::Eof || self.reader().is_done(); + if amount_read > 0 { + let into = streams::IntoArray { + value: array, + len: amount_read as u64, + }; + return if done { + streams::Result::IntoArrayAndDone(into) + } else { + streams::Result::IntoArray(into) + }; } - - let buffer_len = buffer.len(); - self.read_inside_on_pull - .set(ReadDuringJSOnPullResult::Js(buffer)); - // SAFETY: the reader cell is live for `self`'s lifetime; `read` is - // the raw re-entrancy-safe entry (its dispatch runs user JS). - unsafe { IOReader::read(self.reader.get()) }; - - // `replace` resets the field before matching, covering all return paths. - let pulled = self - .read_inside_on_pull - .replace(ReadDuringJSOnPullResult::None); - match pulled { - ReadDuringJSOnPullResult::Js(remaining_buf) => { - let amount_read = buffer_len - remaining_buf.len(); - - bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer_len, amount_read); - - if amount_read > 0 { - if self.reader().is_done() { - return streams::Result::IntoArrayAndDone(streams::IntoArray { - value: array, - len: amount_read as u64, // @truncate - }); - } - - return streams::Result::IntoArray(streams::IntoArray { - value: array, - len: amount_read as u64, // @truncate - }); - } - - if self.reader().is_done() { - return streams::Result::Done; - } - // fallthrough — but `buffer` was moved into read_inside_on_pull. - // Recover it from `remaining_buf` (amount_read == 0 ⇒ same slice). - let global = self.parent_global(); - self.pending_value.with_mut(|p| p.set(&global, array)); - self.pending_view.set(remaining_buf); - bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len); - return streams::Result::Pending(self.pending.as_ptr()); - } - ReadDuringJSOnPullResult::Temporary(buf) => { - bun_core::scoped_log!(FileReader, "onPull({}) = {}", buffer_len, buf.len()); - if self.reader().is_done() { - return streams::Result::TemporaryAndDone(buf); - } - - return streams::Result::Temporary(buf); - } - ReadDuringJSOnPullResult::UseBuffered(_) => { - bun_core::scoped_log!( - FileReader, - "onPull({}) = {}", - buffer_len, - self.buffered.get().len() - ); - let buffered = self.buffered.replace(Vec::new()); - if self.reader().is_done() { - return streams::Result::OwnedAndDone(Vec::::move_from_list(buffered)); - } - return streams::Result::Owned(Vec::::move_from_list(buffered)); - } - _ => { - // Falls through to set - // `pending_view = buffer`. The only variants reaching this arm - // are `None` (impossible — we just stored `Js(buffer)` above and - // `on_read_chunk` never sets `None`) and `AmountRead` (never - // produced by `on_read_chunk`). Unreachable in the current state - // machine; if that invariant ever changes, the buffer slice must - // be recovered from a captured raw ptr+len before the move. - unreachable!( - "on_read_chunk never yields None/AmountRead while read_inside_on_pull == Js" - ); - } + // A completion may have landed in `buffered` while `read_into` ran user JS. + let drained = self.drain(); + if !drained.is_empty() { + return if done { + streams::Result::OwnedAndDone(drained) + } else { + streams::Result::Owned(drained) + }; + } + if done { + return streams::Result::Done; } } @@ -1097,6 +861,10 @@ impl FileReader { let global = self.parent_global(); self.pending_value.with_mut(|p| p.set(&global, array)); self.pending_view.set(buffer); + #[cfg(windows)] + if self.flowing.get() { + self.reader().unpause(); + } bun_core::scoped_log!(FileReader, "onPull({}) = pending", buffer_len); @@ -1149,7 +917,7 @@ impl FileReader { } sink.end(None); } - } else if !self.is_pulling() { + } else { self.consume_reader_buffer(); if self.pending.get().state == streams::PendingState::Pending { if !self.buffered.get().is_empty() { @@ -1336,16 +1104,3 @@ fn read_state_tag(state: ReadState) -> &'static str { ReadState::Drained => "drained", } } - -/// Checks whether `slice` lies within `vec`'s allocation (including spare -/// capacity). Replaces the previous `AllocatedSlice` trait, which materialised -/// a `&[u8]` over `[len, capacity)` — uninitialised memory — purely to feed -/// `bun_core::is_slice_in_buffer`. That was UB-adjacent (a `&[u8]` asserts its -/// bytes are initialised); this helper does the same containment check with -/// pure address arithmetic and never forms a reference over uninit bytes. -#[inline] -fn is_slice_in_vec_capacity(slice: &[u8], vec: &Vec) -> bool { - let slice_start = slice.as_ptr() as usize; - let buf_start = vec.as_ptr() as usize; - buf_start <= slice_start && (slice_start + slice.len()) <= (buf_start + vec.capacity()) -} diff --git a/src/sql_jsc/mysql/MySQLValue.rs b/src/sql_jsc/mysql/MySQLValue.rs index 4a4ce32ad0f6..21e959fe69bf 100644 --- a/src/sql_jsc/mysql/MySQLValue.rs +++ b/src/sql_jsc/mysql/MySQLValue.rs @@ -361,15 +361,15 @@ impl Value { // collect it (and free the backing store despite // the pin) if user JS drops the last reference from // a later parameter. - 2 => { + kind @ (2 | 3) => { roots.append(value); Ok(Value::Bytes(Bytes { - // SAFETY: backing ArrayBuffer is pinned (non-detachable) and + // SAFETY: backing storage is pinned or held (bufferless view) and // rooted via `roots`; slice stays valid until Bytes::drop unpins. slice: ZigStringSlice::from_utf8_never_free(unsafe { core::slice::from_raw_parts(ptr, len) }), - pinned: value, + pinned: if kind == 2 { value } else { JSValue::ZERO }, })) } _ => unreachable!(), diff --git a/test/js/bun/shell/shell-pipe-read-fault.test.ts b/test/js/bun/shell/shell-pipe-read-fault.test.ts index d7feee18bb71..44dbdcde6d7d 100644 --- a/test/js/bun/shell/shell-pipe-read-fault.test.ts +++ b/test/js/bun/shell/shell-pipe-read-fault.test.ts @@ -407,6 +407,10 @@ test.concurrent.skipIf(!isLinux || !cc || !isASAN)( const mkfifo = Bun.which("mkfifo"); const cat = Bun.which("cat"); +// The bulk recv (256 KB of 'A') is delivered, then the real recv picks up the +// child's `printf AAAA` before the EAGAIN whose re-registration fails: the read +// loop hands over everything it read before attempting the (failing, possibly +// parent-freeing) re-arm, so all 256 KB + 4 bytes reach stdout. test.concurrent.skipIf(!isLinux || !cc || !mkfifo || !cat)( "shell delivers the already-read output and the reader error when the read fails while that output is still queued for stdout", async () => { @@ -457,7 +461,7 @@ test.concurrent.skipIf(!isLinux || !cc || !mkfifo || !cat)( readerExitCode, }).toEqual({ teedIsAllA: true, - teedLength: 256 * 1024, + teedLength: 256 * 1024 + "AAAA".length, parsed: { exitCode: ENOMEM }, stderr: "", readerStderr: "", diff --git a/test/js/node/child_process/child_process.test.ts b/test/js/node/child_process/child_process.test.ts index a7bf59ba32ac..acac21c9b10b 100644 --- a/test/js/node/child_process/child_process.test.ts +++ b/test/js/node/child_process/child_process.test.ts @@ -1,7 +1,18 @@ import { semver, write } from "bun"; import { afterAll, beforeEach, describe, expect, it } from "bun:test"; import fs from "fs"; -import { bunEnv, bunExe, isLinux, isPosix, isWindows, nodeExe, runBunInstall, shellExe, tmpdirSync } from "harness"; +import { + bunEnv, + bunExe, + isLinux, + isPosix, + isWindows, + nodeExe, + runBunInstall, + shellExe, + tempDir, + tmpdirSync, +} from "harness"; import { ChildProcess, exec, execFile, execFileSync, execSync, fork, spawn, spawnSync } from "node:child_process"; import { getEventListeners, once, setMaxListeners } from "node:events"; import { promisify } from "node:util"; @@ -1270,3 +1281,108 @@ describe("spawn/execFile({signal}) does not leak abort listeners on spawn failur expect(errors.map(e => e.code)).toEqual(["ENOENT"]); }); }); + +// A 'data' handler runs inside the native read loop that delivered its chunk, +// and node streams pull again from there, so that pull reads synchronously, +// nested in the outer loop, and can run all the way to EOF. The reader used to +// collect a nested read in a heap buffer it freed as soon as the chunk was +// handed over, while FileReader kept pointing at a tail that did not fit the +// pull buffer (heap-use-after-free under ASAN, corrupt or short output +// otherwise). +// +// Pinned down with two markers: the head is bigger than half of the reader's +// 256 KiB scratch, so it is flushed to JS from the middle of the outer loop, +// and the 'data' handler does not return until the tail and EOF are in the +// socket. The tail is bigger than the 64 KiB pull buffer. +describe.skipIf(!isPosix)("child.stdout pull nested in a 'data' event", () => { + it("delivers a tail read to EOF that does not fit the pull buffer", async () => { + const HEAD = 136 * 1024; + const TAIL = 96 * 1024; + using dir = tempDir("child-stdout-nested-pull", { + "producer.js": ` + const fs = require("node:fs"); + const [headMarker, headDone, tailMarker, tailDone] = process.argv.slice(2); + const deadline = Date.now() + 15_000; + function waitFor(file) { + while (!fs.existsSync(file)) { + if (Date.now() > deadline) throw new Error("producer timed out waiting for " + file); + Bun.sleepSync(1); + } + } + function writeAll(buf) { + for (let off = 0; off < buf.length; ) off += fs.writeSync(1, buf, off); + } + waitFor(headMarker); + writeAll(Buffer.alloc(${HEAD}, "h")); + fs.writeFileSync(headDone, ""); + waitFor(tailMarker); + writeAll(Buffer.alloc(${TAIL}, "t")); + fs.closeSync(1); + fs.writeFileSync(tailDone, ""); + `, + "reader.js": ` + const { spawn } = require("node:child_process"); + const fs = require("node:fs"); + const path = require("node:path"); + const file = name => path.join(__dirname, name); + const deadline = Date.now() + 15_000; + function waitFor(name) { + while (!fs.existsSync(file(name))) { + if (Date.now() > deadline) throw new Error("reader timed out waiting for " + name); + Bun.sleepSync(1); + } + } + const child = spawn( + process.execPath, + [file("producer.js"), file("head"), file("head-done"), file("tail"), file("tail-done")], + { stdio: ["ignore", "pipe", "inherit"] }, + ); + const chunks = []; + // 'resume' is emitted after the stream's first read() has been issued + // and found the socket empty, so the head written now is picked up by + // the poll it armed and arrives in one wake. + child.stdout.once("resume", () => { + fs.writeFileSync(file("head"), ""); + waitFor("head-done"); + }); + child.stdout.on("data", chunk => { + chunks.push(chunk); + if (chunks.length === 1) { + fs.writeFileSync(file("tail"), ""); + waitFor("tail-done"); + } + }); + child.on("close", exitCode => { + const out = Buffer.concat(chunks); + console.log( + JSON.stringify({ + exitCode, + firstChunkOverHalfScratch: chunks[0].length > 128 * 1024, + length: out.length, + head: out.subarray(0, ${HEAD}).equals(Buffer.alloc(${HEAD}, "h")), + tail: out.subarray(${HEAD}).equals(Buffer.alloc(${TAIL}, "t")), + }), + ); + }); + `, + }); + await using proc = Bun.spawn({ + cmd: [bunExe(), "reader.js"], + cwd: String(dir), + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode }).toEqual({ + stdout: + JSON.stringify({ exitCode: 0, firstChunkOverHalfScratch: true, length: HEAD + TAIL, head: true, tail: true }) + + "\n", + stderr: "", + exitCode: 0, + }); + // The budget covers the failure modes: a symbolized ASAN report takes + // several seconds, and the fixtures give up on their markers after 15 s so + // their own error, not a test timeout, is what gets reported. + }, 30_000); +}); diff --git a/test/js/node/http/node-http-pinned-write.test.ts b/test/js/node/http/node-http-pinned-write.test.ts index a7367a3d2069..ae4cba85e46c 100644 --- a/test/js/node/http/node-http-pinned-write.test.ts +++ b/test/js/node/http/node-http-pinned-write.test.ts @@ -18,6 +18,12 @@ function makePayload(size: number): Buffer { return Buffer.alloc(size, PATTERN_256); } +// A Buffer that already has an ArrayBuffer behind it: the pending write pins +// that ArrayBuffer, so transfer() copies rather than detaches while it is held. +function makeArrayBufferBackedPayload(size: number): Buffer { + return Buffer.from(new ArrayBuffer(size)).fill(PATTERN_256); +} + function sha1(buf: Uint8Array): string { return createHash("sha1").update(buf).digest("hex"); } @@ -28,9 +34,9 @@ describe("node:http large Buffer writes are sent zero-copy", () => { // reached on Windows (the bytes go straight to the kernel instead). The // correctness tests below still cover the write path there. test.skipIf(isWindows)( - "the buffer backing store is pinned while the write is pending, then released on drain", + "an ArrayBuffer-backed buffer is pinned while the write is pending, then released on drain", async () => { - const payload = makePayload(CHUNK_SIZE); + const payload = makeArrayBufferBackedPayload(CHUNK_SIZE); const expectedHash = sha1(payload); let detachedWhilePending: boolean | undefined; @@ -102,6 +108,56 @@ describe("node:http large Buffer writes are sent zero-copy", () => { }, ); + // A plain Buffer has no ArrayBuffer until `.buffer` is touched, and the + // pending write holds it without materializing one (doing so registers the + // bytes with the GC a second time). transfer() mid-write therefore detaches, + // as in Node, but it moves the storage rather than freeing it, so the bytes + // still to be written reach the client intact. + test.skipIf(isWindows)("a plain Buffer transferred while its write is pending still arrives intact", async () => { + const payload = makePayload(CHUNK_SIZE); + const expectedHash = sha1(payload); + let detachedWhilePending: boolean | undefined; + let moved: ArrayBuffer | undefined; + let handlerError: unknown; + const serverReady = Promise.withResolvers(); + + await using server = http.createServer(async (req, res) => { + try { + res.writeHead(200, { "Content-Type": "application/octet-stream", "Content-Length": String(CHUNK_SIZE) }); + res.write(payload); + moved = payload.buffer.transfer(); + detachedWhilePending = payload.buffer.detached; + serverReady.resolve(); + await once(res, "drain"); + res.end(); + } catch (e) { + handlerError = e; + serverReady.resolve(); + res.destroy(); + } + }); + await once(server.listen(0), "listening"); + const port = (server.address() as AddressInfo).port; + const socket = net.connect(port, "127.0.0.1"); + await once(socket, "connect"); + socket.pause(); + socket.write(`GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n`); + await serverReady.promise; + const chunks: Buffer[] = []; + socket.on("data", chunk => chunks.push(chunk)); + const closed = once(socket, "close"); + socket.resume(); + await closed; + const received = Buffer.concat(chunks); + + expect(handlerError).toBeUndefined(); + const body = received.subarray(received.indexOf("\r\n\r\n") + 4); + expect(body.length).toBe(CHUNK_SIZE); + expect(sha1(body)).toBe(expectedHash); + expect(detachedWhilePending).toBe(true); + expect(moved?.byteLength).toBe(CHUNK_SIZE); + }); + test("Content-Length (non-chunked) path delivers the exact bytes", async () => { const payload = makePayload(CHUNK_SIZE); const expectedHash = sha1(payload); diff --git a/test/js/web/streams/streams-leak.test.ts b/test/js/web/streams/streams-leak.test.ts index 0a2862ade9c1..2f9f2d81391f 100644 --- a/test/js/web/streams/streams-leak.test.ts +++ b/test/js/web/streams/streams-leak.test.ts @@ -60,18 +60,13 @@ test("native ReadableStream reuses the pull buffer across small reads", async () // through the native pull path. expect(chunks.length).toBeGreaterThanOrEqual(CHUNKS_TO_WRITE); - // Consecutive small reads should land in the same backing buffer (the - // tail subarray is reused until a read fills it). 128 bytes of 2-byte - // chunks fits well inside one 256KB buffer, so the whole stream should - // share a handful at most. Pre-fix every chunk had its own 256KB - // buffer, so this was ~chunks.length. + // A small read is copied out right-sized and the pull slab is reused for + // the next read, so each chunk's backing store is its own few bytes rather + // than a 256KB slab per chunk (~chunks.length * 256KB ≈ 16 MB before). const distinctBuffers = new Set(chunks.map(c => c.buffer)); - expect(distinctBuffers.size).toBeLessThan(8); - let backingBytes = 0; for (const buf of distinctBuffers) backingBytes += buf.byteLength; - // Pre-fix this was ~chunks.length * 256KB ≈ 16 MB. - expect(backingBytes).toBeLessThan(4 * 1024 * 1024); + expect(backingBytes).toBeLessThan(64 * 1024); }); // Abandoning a Bun.file().stream() reader mid-file (no cancel(), no EOF) must diff --git a/test/js/workerd/html-rewriter.test.js b/test/js/workerd/html-rewriter.test.js index 73e494b0b28f..22a3d7f282a7 100644 --- a/test/js/workerd/html-rewriter.test.js +++ b/test/js/workerd/html-rewriter.test.js @@ -1895,8 +1895,18 @@ describe("streamed input pacing", () => { const count = 2500; // ~2.4 MB of input: many upstream chunks const input = Buffer.alloc(piece.length * count, piece).toString(); const rewritten = Buffer.alloc((piece.length + 6) * count, `

${text}

`).toString(); - const dir = tempDirWithFiles("hr-pacing", { "in.html": input }); + // A second document made of different bytes, for the tests below that read + // it while another document is being parsed: if that read lands in the + // other document's buffer, these bytes show up in its output. + const otherText = Buffer.alloc(1000, "b").toString(); + const otherPiece = `

${otherText}

`; + const otherRewritten = Buffer.alloc((otherPiece.length + 6) * count, `

${otherText}

`).toString(); + const dir = tempDirWithFiles("hr-pacing", { + "in.html": input, + "other.html": Buffer.alloc(otherPiece.length * count, otherPiece).toString(), + }); const file = path.join(dir, "in.html"); + const otherFile = path.join(dir, "other.html"); function transformInput(body = Bun.file(file)) { let seen = 0; @@ -1992,16 +2002,25 @@ describe("streamed input pacing", () => { expect(seen()).toBe(count); }); - it("a locked but idle reader holds the input", async () => { + // Progress is driven by the reader's pulls: one read yields one chunk and + // leaves the rest of the document unread until the next. + it("a locked reader paces the input by its reads", async () => { const { res, seen } = transformInput(); const reader = res.body.getReader(); - for (let i = 0; i < 5; i++) await setImmediatePromise(); + const first = await reader.read(); const held = seen(); expect(held).toBeLessThan(count); - for (let i = 0; i < 5; i++) await setImmediatePromise(); - expect(seen()).toBe(held); + // Locked but not reading: give the loop real work to turn on and check the input did not advance meanwhile. + // (Windows reads are completions: the one already in flight when the sink pushed back still lands, nothing after it.) + expect((await Bun.file(otherFile).bytes()).length).toBe(otherPiece.length * count); + expect(seen() - held).toBeLessThanOrEqual(isWindows ? (256 * 1024) / piece.length : 0); + const second = await reader.read(); + expect(seen()).toBeGreaterThan(held); + expect(seen()).toBeLessThan(count); reader.releaseLock(); - expect(await readAll(res.body)).toBe(rewritten); + const rest = await readAll(res.body); + expect(Buffer.concat([first.value, second.value]).toString() + rest).toBe(rewritten); + expect(seen()).toBe(count); }); // With nothing reading the output the rewrite is a side effect the caller is @@ -2077,6 +2096,120 @@ describe("streamed input pacing", () => { expect(await inner).toBe(rewritten); }); + // Starting a transform inside the handler and reading it are two reads + // nested in the outer one, which still has its document in the read buffer: + // the second nested read must be kept out of it just like the first. + it("a handler may transform and read another file", async () => { + let inner; + const outer = new HTMLRewriter() + .on("p", { + element(e) { + e.setAttribute("x", "1"); + inner ??= transformInput(Bun.file(otherFile)).res.text(); + }, + }) + .transform(new Response(Bun.file(file))); + expect(await outer.text()).toBe(rewritten); + expect(await inner).toBe(otherRewritten); + }); + + // Same, with the outer document arriving on a pipe (the child's stdin): a + // pipe is read by a different loop than a regular file, out of the same + // buffer. + it("a handler may transform and read another file while its document arrives on stdin", async () => { + const pieces = 64; + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `const attr = { element: e => void e.setAttribute("x", "1") }; + let inner; + const outer = new HTMLRewriter() + .on("p", { + element(e) { + attr.element(e); + inner ??= new HTMLRewriter().on("p", attr).transform(new Response(Bun.file(process.argv[1]))).text(); + }, + }) + .transform(new Response(Bun.stdin)); + const out = await outer.text(); + const innerText = await inner; + const expectedInner = await new HTMLRewriter().on("p", attr).transform(new Response(await Bun.file(process.argv[1]).bytes())).text(); + console.log(JSON.stringify({ out, innerMatches: innerText === expectedInner, innerLength: innerText.length }));`, + otherFile, + ], + env: bunEnv, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + proc.stdin.write(Buffer.alloc(piece.length * pieces, piece)); + proc.stdin.end(); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(JSON.parse(stdout)).toEqual({ + out: Buffer.alloc((piece.length + 6) * pieces, `

${text}

`).toString(), + innerMatches: true, + innerLength: otherRewritten.length, + }); + expect(exitCode).toBe(0); + }); + + // Same hazard with the other user of that read buffer: readFileSync reads + // the file into it before deciding whether it needs to stat. Covers both a + // regular-file input and a pipe (stdin of a child). + describe("a handler may call readFileSync", () => { + const otherContent = Buffer.alloc(4096, "Z").toString(); + const otherTxt = path.join(dir, "other.txt"); + beforeAll(() => fs.writeFileSync(otherTxt, otherContent)); + + it("while the input is a file", async () => { + let intactInnerReads = 0; + const res = new HTMLRewriter() + .on("p", { + element(e) { + e.setAttribute("x", "1"); + if (fs.readFileSync(otherTxt, "utf8") === otherContent) intactInnerReads++; + }, + }) + .transform(new Response(Bun.file(file))); + expect(await res.text()).toBe(rewritten); + expect(intactInnerReads).toBe(count); + }); + + it("while the input is a pipe", async () => { + const pieces = 64; + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + `import { readFileSync } from "fs"; + const otherContent = Buffer.alloc(4096, "Z").toString(); + const res = new HTMLRewriter() + .on("p", { + element(e) { + e.setAttribute("x", "1"); + if (readFileSync(process.argv[1], "utf8") !== otherContent) throw new Error("inner read corrupted"); + }, + }) + .transform(new Response(Bun.stdin)); + process.stdout.write(await res.text());`, + otherTxt, + ], + env: bunEnv, + stdin: "pipe", + stdout: "pipe", + stderr: "pipe", + }); + proc.stdin.write(Buffer.alloc(piece.length * pieces, piece)); + proc.stdin.end(); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe(Buffer.alloc((piece.length + 6) * pieces, `

${text}

`).toString()); + expect(exitCode).toBe(0); + }); + }); + // Regular-file reads are synchronous on POSIX, so reading ahead of the // consumer would show up as `transform()` itself reading (and buffering the // rewrite of) the entire file. Sparse files keep these cheap.