Streams: one PipeReader loop with owned chunks, hold-not-adopt buffer pins, right-sized native pulls

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<AnyTaskWithExtraContext>;
 
@@ -82,7 +79,7 @@ pub struct MiniEventLoop {
     // Opaque ctx assigned externally; only read/cleared here.
     pub(crate) after_event_loop_callback_ctx: Option<NonNull<c_void>>,
     pub(crate) after_event_loop_callback: Option<unsafe extern "C" fn(*mut c_void)>,
-    pub pipe_read_buffer: Option<Box<PipeReadBuffer>>,
+    pub pipe_read_scratch: Box<bun_io::PipeReadScratch>,
 }
 
 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::<PipeReadBuffer>)[..]
-    }
-
     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,
     }
 }
 

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();

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_();