From e886d9d7c14023aeab87672cdd6cea289c0050f5 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Sun, 23 Aug 2026 22:21:42 +0000 Subject: [PATCH 1/2] shell: remove unsafe from the rm/cp/ls/mkdir/yes builtins and EnvStr rm: the recursive DirTask tree is an owned tree. Each directory is a Box; a directory that spawned child tasks parks its box in an Arc (its own slot plus one per unfinished child) and whoever takes the count to zero owns the box and deletes the directory. Per-root shared state is Arc; the root DirTask carries the ShellRmTask and posts it back when the tree is done; verbose output hops post the DirTask box itself. On non-Linux, a directory found non-empty after its children finished (and re-spawned) now releases its slot again instead of never completing. rm numbers its per-root error chunks like its verbose chunks, so `rm nope1 nope2 2>&1 | true` no longer waits for callbacks the closed pipe collapsed into one. cp: the fs.cp task owns the ShellCpTask box and hands it back through cp_on_finish. yes: the bounce payload is a Box taken from the builtin, queued through EventLoopHandle::enqueue_boxed_after_yield and put back when it runs. mkdir collects verbose output through a RefCell; ls -l writes the mode/time bytes directly. ParsedShellScript drops a write-only JsRef. EnvStr is an enum (Empty / Rc<[u8]> / BackRef<[u8]> for literals, script text and the dotenv loader) instead of a packed tagged pointer with manual ref/deref; EnvMap looks up by &[u8] and owns what it stores. PWD/OLDPWD hold copies rather than slices of the cwd buffers. Instruction counts vs the base branch (release, perf stat, pinned, interleaved): 1000x `bun run --shell=bun` true -0.02%, 200x rm -rf of a 100-file dir -0.03%, 100x rm -rf of a 131-dir tree +0.09% (-v: +0.04%), 3000x JS $`true` +0.05%. --- src/event_loop/AnyEventLoop.rs | 26 + src/event_loop/MiniEventLoop.rs | 7 + src/runtime/dispatch.rs | 13 +- src/runtime/node/node_fs.rs | 36 +- src/runtime/shell/Builtin.rs | 6 +- src/runtime/shell/EnvMap.rs | 84 +- src/runtime/shell/EnvStr.rs | 198 +--- src/runtime/shell/ParsedShellScript.rs | 23 +- src/runtime/shell/RefCountedStr.rs | 64 -- src/runtime/shell/builtin/cd.rs | 4 +- src/runtime/shell/builtin/cp.rs | 113 +- src/runtime/shell/builtin/export.rs | 30 +- src/runtime/shell/builtin/ls.rs | 12 +- src/runtime/shell/builtin/mkdir.rs | 18 +- src/runtime/shell/builtin/rm.rs | 1442 ++++++++---------------- src/runtime/shell/builtin/which.rs | 10 +- src/runtime/shell/builtin/yes.rs | 100 +- src/runtime/shell/dispatch_tasks.rs | 7 +- src/runtime/shell/interpreter.rs | 18 +- src/runtime/shell/mod.rs | 2 - src/runtime/shell/states/Assigns.rs | 3 +- src/runtime/shell/states/Expansion.rs | 15 +- test/js/bun/shell/commands/rm.test.ts | 12 +- 23 files changed, 750 insertions(+), 1493 deletions(-) delete mode 100644 src/runtime/shell/RefCountedStr.rs diff --git a/src/event_loop/AnyEventLoop.rs b/src/event_loop/AnyEventLoop.rs index a5786fbb7311..b8d5f5753ef2 100644 --- a/src/event_loop/AnyEventLoop.rs +++ b/src/event_loop/AnyEventLoop.rs @@ -462,6 +462,32 @@ impl EventLoopHandle { EnteredEventLoop(self) } + /// Owning thread: queue `owner` to run once pending I/O has had a turn — + /// the JS loop's after-yield queue (delivered by `T::TAG`), or for a mini + /// loop one tick of its platform loop, then its task queue (delivered to + /// `R`). For work that re-queues itself and would otherwise starve I/O. + pub fn enqueue_boxed_after_yield( + self, + owner: Box, + node: fn(&mut T) -> &mut AnyTaskWithExtraContext, + ) where + T: crate::Taskable, + R: crate::AnyTaskWithExtraContext::BoxedMiniTaskRunner, + { + match self { + EventLoopHandle::Js { owner: js } => { + js.enqueue_task_after_yield(crate::Task::from_boxed(owner)); + } + EventLoopHandle::Mini(mut mini) => { + let mini = mini_mut(&mut mini); + mini.tick_platform_loop(); + mini.enqueue_task_concurrent(AnyTaskWithExtraContext::arm_boxed::( + owner, node, + )); + } + } + } + /// Owning thread: the weak poster other threads use to deliver JS-loop /// tasks to this handle's VM; `None` for a mini loop (post to it directly — /// it is owned by, and outlives the work of, its thread). diff --git a/src/event_loop/MiniEventLoop.rs b/src/event_loop/MiniEventLoop.rs index 40f26a45deda..10e4de1dbb73 100644 --- a/src/event_loop/MiniEventLoop.rs +++ b/src/event_loop/MiniEventLoop.rs @@ -351,6 +351,13 @@ impl MiniEventLoop { } } + /// One tick of the platform (uSockets) loop, so pending I/O is serviced + /// before the caller queues more work. + pub fn tick_platform_loop(&mut self) { + // SAFETY: see `loop_ptr()` invariant. + unsafe { (*self.loop_ptr()).tick() }; + } + /// `task` must outlive the queued work item; ownership of the intrusive /// node stays with the caller until the callback runs. pub fn enqueue_task_concurrent(&mut self, task: NonNull) { diff --git a/src/runtime/dispatch.rs b/src/runtime/dispatch.rs index 7dfb8d203424..47999162cdb8 100644 --- a/src/runtime/dispatch.rs +++ b/src/runtime/dispatch.rs @@ -539,16 +539,13 @@ fn run_task_cold(task: Task) { task_tag::ShellMvBatchedTask => shell_dispatch!(ShellMvBatchedTask), task_tag::ShellMvCheckTargetTask => shell_dispatch!(ShellMvCheckTargetTask), task_tag::ShellRmTask => shell_dispatch!(ShellRmTask), - task_tag::ShellRmDirTask => { - let t = cast_ptr!(ShellRmDirTask); - ShellRmDirTask::run_from_main_thread(t); - } + task_tag::ShellRmDirTask => shell_dispatch!(ShellRmDirTask), task_tag::ShellGlobTask => shell_dispatch!(ShellGlobTask), task_tag::ShellYesTask => { - // SAFETY: §Dispatch — tag identifies pointee; enqueued by - // `YesTask::enqueue`, storage lives inside `Box` in the - // interpreter arena and is stable until the builtin deinits. - ShellYesTask::run_from_main_thread(unsafe { &*cast_ptr!(ShellYesTask) }); + // SAFETY: §Dispatch — tag identifies pointee: the box + // `YesTask::enqueue` leaked when posting. + let t = unsafe { bun_core::heap::take(cast_ptr!(ShellYesTask)) }; + t.run(); } // ── bake dev-server ────────────────────────────────────────────── diff --git a/src/runtime/node/node_fs.rs b/src/runtime/node/node_fs.rs index 36f8b2081040..b215544919af 100644 --- a/src/runtime/node/node_fs.rs +++ b/src/runtime/node/node_fs.rs @@ -1371,10 +1371,9 @@ mod _async_tasks { /// enqueued once the count reaches zero, so subtasks still running on the /// thread pool never dereference a freed parent. pub(crate) subtask_count: AtomicUsize, - /// BACKREF — `Some` iff `IS_SHELL`. The shell `ShellCpTask` owns and - /// outlives this task; `ParentRef` gives a safe `&ShellCpTask` projection - /// for `cp_on_copy` and round-trips the `*mut` for `cp_on_finish`. - pub(crate) shelltask: Option>, + /// `Some` iff `IS_SHELL`: the shell `cp` task this copy reports to + /// (`cp_on_copy` per file; handed back through `cp_on_finish`). + pub(crate) shelltask: Option>, } bun_threading::intrusive_work_task!([const IS_SHELL: bool] NewAsyncCpTask, task); @@ -1507,10 +1506,10 @@ mod _async_tasks { if !IS_SHELL { return; } - // When IS_SHELL, `shelltask` is `Some` (ParentRef invariant: owner - // outlives this task). Shared borrow only — concurrent subtasks may - // call this in parallel; `cp_on_copy` serialises via its internal mutex. + // Shared borrow only — concurrent subtasks may call this in + // parallel; `cp_on_copy` serialises via its internal mutex. self.shelltask + .as_deref() .expect("IS_SHELL ⇒ shelltask") .cp_on_copy(src.as_ref(), dest.as_ref()); } @@ -1531,7 +1530,7 @@ mod _async_tasks { EventLoopHandle::init(vm.event_loop.cast()), bun_jsc::ConcurrentPoster::Js(vm.ticket()), tracker, - core::ptr::null_mut(), + None, ); // SAFETY: `schedule_new` returns a Box::leak'd pointer; valid until destroy() unsafe { &*task }.promise.value() @@ -1544,16 +1543,16 @@ mod _async_tasks { cp_args: ThreadIsolated>, evtloop: EventLoopHandle, poster: bun_jsc::ConcurrentPoster, - shelltask: *mut ShellCpTask, - ) -> *mut Self { + shelltask: Box, + ) { Self::schedule_new( JSPromiseStrong::default(), cp_args, evtloop, poster, AsyncTaskTracker { id: 0 }, - shelltask, - ) + Some(shelltask), + ); } fn schedule_new( @@ -1562,7 +1561,7 @@ mod _async_tasks { evtloop: EventLoopHandle, poster: bun_jsc::ConcurrentPoster, tracker: AsyncTaskTracker, - shelltask: *mut ShellCpTask, + shelltask: Option>, ) -> *mut Self { let mut task = Box::new(Self { promise, @@ -1577,9 +1576,7 @@ mod _async_tasks { r#ref: KeepAlive::default(), tracker, subtask_count: AtomicUsize::new(1), - // SAFETY: `shelltask` (when non-null) is the live heap-alloc'd `ShellCpTask` - // that owns and outlives this task; pointer carries write provenance. - shelltask: unsafe { bun_ptr::ParentRef::from_nullable_mut(shelltask) }, + shelltask, }); if !IS_SHELL { task.r#ref.ref_(event_loop_handle_to_ctx(task.evtloop)); @@ -1678,16 +1675,13 @@ mod _async_tasks { pub(crate) fn run_from_js_thread(&mut self) -> JsResult<()> { if IS_SHELL { - // SAFETY: shelltask is set by create_for_shell and outlives this task // Move the result out — `Maybe` (= `Maybe<()>`) has a cheap // `Ok(())` placeholder. let result = core::mem::replace(self.result.get_mut(), Ok(())); let src = core::mem::take(&mut self.args.src); let dest = core::mem::take(&mut self.args.dest); - let shelltask = self.shelltask.expect("IS_SHELL ⇒ shelltask").as_mut_ptr(); - // SAFETY: shelltask is non-null in the IS_SHELL specialization and - // outlives this task; `cp_on_finish` reclaims it. - unsafe { ShellCpTask::cp_on_finish(shelltask, src, dest, result) }; + let shelltask = self.shelltask.take().expect("IS_SHELL ⇒ shelltask"); + shelltask.cp_on_finish(src, dest, result); // SAFETY: self was Box::leak'd in create*(); destroyed exactly once here unsafe { Self::destroy(std::ptr::from_mut::(self)) }; return Ok(()); diff --git a/src/runtime/shell/Builtin.rs b/src/runtime/shell/Builtin.rs index e632913e6766..52a64eee9dcb 100644 --- a/src/runtime/shell/Builtin.rs +++ b/src/runtime/shell/Builtin.rs @@ -158,7 +158,7 @@ macro_rules! shell_builtins { /// Hoisted dispatch for the `onIOWriterChunk` callback. `seq != 0` /// names a parked [`OutputTask`](crate::shell::interpreter::OutputTask) - /// chunk of an ls/mkdir/touch/cp. + /// chunk of an ls/mkdir/touch/cp, or one of rm's numbered chunks. pub fn on_io_writer_chunk( interp: &Interpreter, cmd: NodeId, @@ -174,8 +174,8 @@ macro_rules! shell_builtins { Kind::Mkdir => OutputTask::::on_chunk(interp, cmd, seq, written, err), Kind::Touch => OutputTask::::on_chunk(interp, cmd, seq, written, err), Kind::Cp => OutputTask::::on_chunk(interp, cmd, seq, written, err), - // rm numbers its verbose chunks only so that each is - // called back. + // rm numbers its verbose and error chunks only so that + // each is called back. Kind::Rm => crate::shell::builtins::rm::Rm::on_io_writer_chunk(interp, cmd, written, err), other => unreachable!("{} queues no numbered chunks", other.as_str()), }; diff --git a/src/runtime/shell/EnvMap.rs b/src/runtime/shell/EnvMap.rs index 7ac7347397f2..412cf725c1cf 100644 --- a/src/runtime/shell/EnvMap.rs +++ b/src/runtime/shell/EnvMap.rs @@ -1,4 +1,6 @@ -use bun_collections::array_hash_map::{self, ArrayHashContext, ArrayHashMap, Iter}; +use bun_collections::array_hash_map::{ + self, ArrayHashAdapter, ArrayHashContext, ArrayHashMap, Iter, +}; use crate::shell::EnvStr; @@ -15,37 +17,56 @@ type EnvMapInner = ArrayHashMap; #[derive(Default)] struct EnvMapContext; -impl ArrayHashContext for EnvMapContext { - fn hash(&self, s: &EnvStr) -> u32 { +impl EnvMapContext { + #[inline] + fn hash_bytes(s: &[u8]) -> u32 { #[cfg(windows)] { return >::hash( &array_hash_map::CaseInsensitiveAsciiStringContext::default(), - s.slice(), + s, ); } #[cfg(not(windows))] { - array_hash_map::hash_string(s.slice()) + array_hash_map::hash_string(s) } } - fn eql(&self, a: &EnvStr, b: &EnvStr, _b_index: usize) -> bool { + #[inline] + fn eql_bytes(a: &[u8], b: &[u8]) -> bool { #[cfg(windows)] { // Must be length-checked: "PATH" must NOT match "PATHEXT". - return bun_core::strings::eql_case_insensitive_asciii_check_length( - a.slice(), - b.slice(), - ); + return bun_core::strings::eql_case_insensitive_asciii_check_length(a, b); } #[cfg(not(windows))] { - a.slice() == b.slice() + a == b } } } +impl ArrayHashContext for EnvMapContext { + fn hash(&self, s: &EnvStr) -> u32 { + Self::hash_bytes(s.slice()) + } + + fn eql(&self, a: &EnvStr, b: &EnvStr, _b_index: usize) -> bool { + Self::eql_bytes(a.slice(), b.slice()) + } +} + +impl ArrayHashAdapter<[u8], EnvStr> for EnvMapContext { + fn hash(&self, key: &[u8]) -> u32 { + Self::hash_bytes(key) + } + + fn eql(&self, a: &[u8], b: &EnvStr, _b_index: usize) -> bool { + Self::eql_bytes(a, b.slice()) + } +} + impl EnvMap { pub(crate) fn init() -> EnvMap { EnvMap { @@ -71,16 +92,8 @@ impl EnvMap { } } - /// NOTE: This will `.ref()` value, so you should `defer value.deref()` it - /// before handing it to this function!!! pub(crate) fn insert(&mut self, key: EnvStr, val: EnvStr) { let result = self.map.get_or_put(key).expect("OOM"); - if !result.found_existing { - key.ref_(); - } else { - result.value_ptr.deref(); - } - val.ref_(); *result.value_ptr = val; } @@ -96,39 +109,14 @@ impl EnvMap { self.map.ensure_total_capacity(new_capacity).expect("OOM"); } - /// NOTE: Make sure you deref the string when done! - pub fn get(&self, key: EnvStr) -> Option { - let val = *self.map.get(&key)?; - val.ref_(); - Some(val) + pub fn get(&self, key: &[u8]) -> Option<&EnvStr> { + let i = self.map.get_index_adapted(key, &EnvMapContext)?; + Some(&self.map.values()[i]) } pub(crate) fn clone(&self) -> EnvMap { - let new = EnvMap { + EnvMap { map: self.map.clone().expect("OOM"), - }; - new.ref_strings(); - new - } - - fn ref_strings(&self) { - for (key, value) in self.map.keys().iter().zip(self.map.values()) { - key.ref_(); - value.ref_(); } } - - fn deref_strings(&self) { - for (key, value) in self.map.keys().iter().zip(self.map.values()) { - key.deref(); - value.deref(); - } - } -} - -impl Drop for EnvMap { - fn drop(&mut self) { - self.deref_strings(); - // map storage freed by its own Drop - } } diff --git a/src/runtime/shell/EnvStr.rs b/src/runtime/shell/EnvStr.rs index 9aa0975f9277..a1cedb5d441c 100644 --- a/src/runtime/shell/EnvStr.rs +++ b/src/runtime/shell/EnvStr.rs @@ -7,201 +7,65 @@ //! //! So environment strings can be ref counted or borrowed slices -use core::ffi::c_void; +use std::rc::Rc; -use super::ref_counted_str::RefCountedStr; - -/// Packed `u128` layout (LSB-first): -/// - bits 0..48 : `ptr` (u48) -/// - bits 48..64 : `tag` (u16) -/// - bits 64..128 : `len` (usize) -#[repr(transparent)] -#[derive(Copy, Clone, Eq, PartialEq)] -pub struct EnvStr(u128); - -#[repr(u16)] -#[derive(Copy, Clone, Eq, PartialEq)] -pub enum Tag { +#[derive(Clone, Default)] +pub enum EnvStr { /// no value - Empty = 0, + #[default] + Empty, /// Dealloced by reference counting - Refcounted = 1, + Refcounted(Rc<[u8]>), - /// Memory is managed elsewhere so don't dealloc it - Slice = 2, + /// Memory is managed elsewhere: a literal, text of the parsed script, or + /// the mini event loop's dotenv loader — each outlives every env map of + /// the interpreter holding it (the [`bun_ptr::BackRef`] invariant). + Slice(bun_ptr::BackRef<[u8]>), } -const PTR_MASK: u128 = (1u128 << 48) - 1; -const TAG_SHIFT: u32 = 48; -const TAG_MASK: u128 = 0xFFFF; -const LEN_SHIFT: u32 = 64; - impl EnvStr { - #[inline] - const fn pack(ptr: u64, tag: Tag, len: usize) -> EnvStr { - EnvStr( - (ptr as u128 & PTR_MASK) - | ((tag as u16 as u128) << TAG_SHIFT) - | ((len as u64 as u128) << LEN_SHIFT), - ) - } - - #[inline] - fn ptr(self) -> u64 { - (self.0 & PTR_MASK) as u64 - } - - #[inline] - fn tag(self) -> Tag { - // Only constructed via `pack` with a valid `Tag` discriminant (0..=2); - // any other value is corruption — trap rather than silently folding - // to `Empty`. - match ((self.0 >> TAG_SHIFT) & TAG_MASK) as u16 { - 0 => Tag::Empty, - 1 => Tag::Refcounted, - 2 => Tag::Slice, - n => unreachable!("invalid EnvStr tag {n}"), - } - } - - #[inline] - fn len(self) -> usize { - (self.0 >> LEN_SHIFT) as u64 as usize - } - + /// `str` must outlive every env map this value is inserted into; see + /// [`EnvStr::Slice`]. Use [`Self::dupe_ref_counted`] for anything shorter + /// lived (argv, cwd buffers, JS strings). #[inline] pub(crate) fn init_slice(str: &[u8]) -> EnvStr { if str.is_empty() { - // Zero length strings may have invalid pointers, leading to a bad integer cast. - return Self::pack(0, Tag::Empty, 0); + return EnvStr::Empty; } - - Self::pack(to_ptr(str.as_ptr().cast::()), Tag::Slice, str.len()) + EnvStr::Slice(bun_ptr::BackRef::new(str)) } /// Same thing as `init_ref_counted` except it duplicates the passed string pub(crate) fn dupe_ref_counted(old_str: &[u8]) -> EnvStr { if old_str.is_empty() { - return Self::pack(0, Tag::Empty, 0); + return EnvStr::Empty; } - - // Global mimalloc aborts on OOM; ownership of the duplicated bytes - // transfers to RefCountedStr. - let str: Box<[u8]> = Box::<[u8]>::from(old_str); - let len = str.len(); - Self::pack( - to_ptr(RefCountedStr::init(str) as *const c_void), - Tag::Refcounted, - len, - ) + EnvStr::Refcounted(Rc::from(old_str)) } - /// Takes ownership of the backing allocation (hands the slice to - /// `RefCountedStr` without copying). Use [`Self::dupe_ref_counted`] to - /// copy a borrowed slice instead. - pub(crate) fn init_ref_counted(str: Box<[u8]>) -> EnvStr { + /// Takes ownership of the bytes. Use [`Self::dupe_ref_counted`] to copy a + /// borrowed slice instead. + pub(crate) fn init_ref_counted(str: Vec) -> EnvStr { if str.is_empty() { - return Self::pack(0, Tag::Empty, 0); + return EnvStr::Empty; } - - // NOTE: `len` is left 0 here (only `ptr` + `tag` set); the slice - // length is recovered via RefCountedStr::byte_slice(). - Self::pack( - to_ptr(RefCountedStr::init(str) as *const c_void), - Tag::Refcounted, - 0, - ) + EnvStr::Refcounted(Rc::from(str)) } pub(crate) fn slice(&self) -> &[u8] { - // NOTE: the returned slice borrows either external memory (Tag::Slice) or the - // RefCountedStr buffer. Tying the return lifetime to `&self` prevents the caller from - // conjuring `&'static [u8]` (PORTING.md §Forbidden: lifetime-extension via raw-pointer - // deref). `EnvStr` is still `Copy`, so this is a best-effort bound — the caller is - // responsible for keeping the backing storage alive. - match self.tag() { - Tag::Empty => b"", - Tag::Slice => self.cast_slice(), - Tag::Refcounted => match self.as_ref_counted() { - Some(r) => r.byte_slice(), - // Unreachable: tag check above guarantees `Some`. - None => b"", - }, + match self { + EnvStr::Empty => b"", + EnvStr::Slice(s) => s.get(), + EnvStr::Refcounted(r) => r, } } - pub(crate) fn memory_cost(self) -> usize { - let divisor: usize = 'brk: { - if let Some(refc) = self.as_ref_counted() { - break 'brk refc.refcount.get() as usize; - } - break 'brk 1; - }; - if divisor == 0 { - bun_core::hint::cold(); - return 0; + pub(crate) fn memory_cost(&self) -> usize { + match self { + EnvStr::Empty => 0, + EnvStr::Slice(s) => s.len(), + EnvStr::Refcounted(r) => r.len() / Rc::strong_count(r), } - - self.len() / divisor } - - pub fn ref_(self) { - if let Some(refc) = self.as_ref_counted() { - refc.ref_(); - } - } - - pub fn deref(self) { - if self.tag() == Tag::Refcounted { - // SAFETY: tag == Refcounted guarantees a live *mut RefCountedStr; - // `deref` may free it, so this stays raw-ptr (not `as_ref_counted`). - unsafe { RefCountedStr::deref(self.cast_ref_counted()) }; - } - } - - /// Shared-borrow accessor for the ref-counted backing — centralises the - /// `unsafe { &*self.cast_ref_counted() }` back-ref deref under the - /// `Tag::Refcounted ⇒ live heap RefCountedStr` invariant. Returns `None` - /// for `Slice`/`Empty`. The borrow is tied to `&self` (best-effort: `EnvStr` - /// is `Copy`, so the caller is still responsible for keeping the +1 alive). - #[inline] - fn as_ref_counted(&self) -> Option<&RefCountedStr> { - if self.tag() == Tag::Refcounted { - // SAFETY: tag == Refcounted guarantees `ptr` is a live - // *mut RefCountedStr (set by init_ref_counted/dupe_ref_counted) - // with refcount >= 1; read-only borrow here. - return Some(unsafe { &*self.cast_ref_counted() }); - } - None - } - - #[inline] - fn cast_slice(&self) -> &[u8] { - // SAFETY: tag == Slice guarantees `ptr` was derived from a valid `[*]const u8` of - // length `len` whose lifetime is managed elsewhere (caller contract of init_slice). - // The returned borrow is tied to `&self` so callers cannot pick `'static`. - // Provenance: the packed-tagged-value design round-trips the pointer - // through an integer by construction (exposed provenance), so this is - // not strict-provenance clean and cannot be without unpacking EnvStr. - unsafe { core::slice::from_raw_parts(self.ptr() as usize as *const u8, self.len()) } - } - - #[inline] - fn cast_ref_counted(self) -> *mut RefCountedStr { - // SAFETY: tag == Refcounted guarantees `ptr` was derived from RefCountedStr::init. - self.ptr() as usize as *mut RefCountedStr - } -} - -impl Default for EnvStr { - fn default() -> Self { - Self::pack(0, Tag::Empty, 0) - } -} - -#[inline] -fn to_ptr(ptr_val: *const c_void) -> u64 { - // Masks the low 48 bits of the address. - (ptr_val as usize as u64) & ((1u64 << 48) - 1) } diff --git a/src/runtime/shell/ParsedShellScript.rs b/src/runtime/shell/ParsedShellScript.rs index 1bb01b72d947..cbf18af1d2f8 100644 --- a/src/runtime/shell/ParsedShellScript.rs +++ b/src/runtime/shell/ParsedShellScript.rs @@ -5,7 +5,7 @@ use core::sync::atomic::Ordering; use bun_core::String as BunString; use bun_jsc::{ CallFrame, JSGlobalObject, JSPropertyIterator, JSPropertyIteratorOptions, JSValue, JsCell, - JsRef, JsResult, MarkedArgumentBuffer, StringJsc as _, + JsResult, MarkedArgumentBuffer, StringJsc as _, }; use super::EnvStr; @@ -29,10 +29,6 @@ pub struct ParsedShellScript { pub(crate) export_env: JsCell>, pub(crate) quiet: Cell, pub(crate) cwd: JsCell>, - /// Self-wrapper backref. `.classes.ts` has `finalize: true`, so the weak arm is - /// sound: the codegen finalizer drops this Box (and the `JsRef`) at sweep. - /// Read-only after construction. - pub(crate) this_jsvalue: JsRef, /// Read-only after construction (set once before the JS wrapper exists). pub(crate) estimated_size_for_gc: usize, } @@ -45,7 +41,6 @@ impl Default for ParsedShellScript { export_env: JsCell::new(None), quiet: Cell::new(false), cwd: JsCell::new(None), - this_jsvalue: JsRef::empty(), estimated_size_for_gc: 0, } } @@ -158,12 +153,10 @@ impl ParsedShellScript { let keyslice = key.to_owned_slice(); // errdefer free(keyslice) — Drop on early-return handles this. let slice = value.to_bun_string(global)?.to_owned_slice(); - let keyref = EnvStr::init_ref_counted(keyslice.into_boxed_slice()); - let valueref = EnvStr::init_ref_counted(slice.into_boxed_slice()); + let keyref = EnvStr::init_ref_counted(keyslice); + let valueref = EnvStr::init_ref_counted(slice); env.insert(keyref, valueref); - keyref.deref(); - valueref.deref(); } // Dropping the previous Option deinits it. self.export_env.set(Some(env)); @@ -254,17 +247,13 @@ fn create_parsed_shell_script_impl( ..Default::default() }); parsed_shell_script.estimated_size_for_gc = parsed_shell_script.compute_estimated_size_for_gc(); - let parsed_shell_script_ptr = bun_core::heap::into_raw(parsed_shell_script); - // `parsed_shell_script_ptr` is a fresh `heap::alloc`; ownership transfers - // to the C++ wrapper (stored opaquely, freed via the generated finalize - // callback). + // Ownership transfers to the C++ wrapper (stored opaquely, freed via the + // generated finalize callback). let this_jsvalue = ParsedShellScript__createWithValues( global, - parsed_shell_script_ptr.cast::(), + bun_core::heap::into_raw(parsed_shell_script).cast::(), marked_argument_buffer, ); - // SAFETY: pointer just created above; wrapper now owns it but we need one more field write. - unsafe { (*parsed_shell_script_ptr).this_jsvalue = JsRef::init_weak(this_jsvalue) }; bun_analytics::features::shell.fetch_add(1, Ordering::Relaxed); Ok(this_jsvalue) diff --git a/src/runtime/shell/RefCountedStr.rs b/src/runtime/shell/RefCountedStr.rs deleted file mode 100644 index b6a67c6fb1da..000000000000 --- a/src/runtime/shell/RefCountedStr.rs +++ /dev/null @@ -1,64 +0,0 @@ -use core::cell::Cell; - -bun_core::declare_scope!(RefCountedEnvStr, hidden); - -pub struct RefCountedStr { - pub(super) refcount: Cell, - // Owning `Box<[u8]>`, so `byte_slice`/`free_str` need no raw-parts rebuild. - data: Box<[u8]>, -} - -impl RefCountedStr { - // Takes ownership of `slice` (global mimalloc) and stores it directly. - pub(crate) fn init(slice: Box<[u8]>) -> *mut RefCountedStr { - bun_core::scoped_log!(RefCountedEnvStr, "init: {}", bstr::BStr::new(&*slice)); - // bun.handleOom(bun.default_allocator.create(...)) → Box::new (aborts on OOM) - bun_core::heap::into_raw(Box::new(RefCountedStr { - refcount: Cell::new(1), - data: slice, - })) - } - - pub(crate) fn byte_slice(&self) -> &[u8] { - &self.data - } - - pub fn ref_(&self) { - self.refcount.set(self.refcount.get() + 1); - } - - // Takes `*mut Self` because reaching refcount==0 deallocates the - // `Box` that backs `this`; a `&self` borrow would dangle across that drop. - pub unsafe fn deref(this: *mut RefCountedStr) { - // SAFETY: caller guarantees `this` was produced by `init` and is still - // live; on hitting 0, `this` is uniquely owned and Box-allocated. - unsafe { - let rc = &(*this).refcount; - rc.set(rc.get() - 1); - if rc.get() == 0 { - Self::deinit(this); - } - } - } - - // Not `impl Drop` — this is the intrusive-rc self-destroy path - // (`bun.default_allocator.destroy(this)`), which must deallocate the Box backing `self`. - unsafe fn deinit(this: *mut RefCountedStr) { - // SAFETY: refcount just reached 0; `this` is uniquely owned and was Box-allocated in `init`. - unsafe { - bun_core::scoped_log!( - RefCountedEnvStr, - "deinit: {}", - bstr::BStr::new((*this).byte_slice()) - ); - (*this).free_str(); - drop(bun_core::heap::take(this)); - } - } - - fn free_str(&mut self) { - // Dropping the old `Box<[u8]>` returns its allocation; an empty box - // owns no heap storage so the `len == 0` early-out is preserved. - self.data = Box::default(); - } -} diff --git a/src/runtime/shell/builtin/cd.rs b/src/runtime/shell/builtin/cd.rs index b5907e213906..328c7fa69a97 100644 --- a/src/runtime/shell/builtin/cd.rs +++ b/src/runtime/shell/builtin/cd.rs @@ -33,9 +33,7 @@ impl Cd { let shell = Builtin::shell(interp, cmd); if argc == 0 { - let home_str = shell.borrow().get_homedir(); - let home = home_str.slice().to_vec(); - home_str.deref(); + let home = shell.borrow().get_homedir().slice().to_vec(); if home.is_empty() { return Self::write_stderr_non_blocking( interp, diff --git a/src/runtime/shell/builtin/cp.rs b/src/runtime/shell/builtin/cp.rs index f3b407430cba..620994e68c12 100644 --- a/src/runtime/shell/builtin/cp.rs +++ b/src/runtime/shell/builtin/cp.rs @@ -419,60 +419,50 @@ impl ShellCpTask { } } - /// Called on the JS thread when the node:fs async cp completes (success or - /// first error). Records the error (if any) and finishes this `ShellCpTask` - /// in place so the interpreter can drain `verbose_output` / surface it. - /// - /// # Safety - /// `this` is the live task `run_on_pool` released to the - /// `ShellAsyncCpTask`; reclaimed here, not touched by the caller after. - pub(crate) unsafe fn cp_on_finish( - this: *mut ShellCpTask, + /// Called on the JS thread by the node:fs async cp that owns this task, + /// when it completes (success or first error). Takes back the absolute + /// paths lent to it, records the error (if any) and finishes in place so + /// the interpreter can drain `verbose_output` / surface it. + pub(crate) fn cp_on_finish( + mut self: Box, src: PathLike<'static>, dest: PathLike<'static>, result: bun_sys::Maybe<()>, ) { - // SAFETY: caller contract — JS thread, from the `ShellAsyncCpTask`'s - // completion; `this` is live and ours (the box `run_on_pool` released - // to it). The pool side finished (and dropped its poster) when it - // handed the copy to that task, so continue in place rather than - // bouncing through the concurrent queue again. - let mut this = unsafe { bun_core::heap::take(this) }; - this.src_absolute = Some(src.into_vec()); - this.tgt_absolute = Some(dest.into_vec()); + self.src_absolute = Some(src.into_vec()); + self.tgt_absolute = Some(dest.into_vec()); if let Err(e) = result { - this.err = Some(ShellErr::new_sys(&e)); + self.err = Some(ShellErr::new_sys(&e)); } - ShellTask::run_from_main_thread::(this); + ShellTask::run_from_main_thread::(self); } - /// Pool-side body: run the impl, and on error post back immediately; - /// the success path hands the allocation to a `ShellAsyncCpTask`, whose - /// completion (`cp_on_finish`) reclaims it. - fn run_on_pool(this: Box) { - let this = bun_core::heap::into_raw(this); - // SAFETY: `this` is the box just released; the worker thread has - // exclusive access until it is handed off. Raw because on success the - // `ShellAsyncCpTask` it now belongs to may free it from another thread - // at once. - unsafe { - // Moved out first: see above. - let poster = (*this) - .task - .poster - .take() - .expect("shell cp task on the pool is armed"); - if let Some(e) = (*this).run_from_thread_pool_impl(&poster) { - (*this).err = Some(e); - (*this).task.poster = Some(poster); - ShellTask::on_finish::(bun_core::heap::take(this)); - } else { - // The copy now belongs to a `ShellAsyncCpTask` (holding its - // own poster, completing on the JS thread via `cp_on_finish`); - // this task's pool part is over. - drop(poster); - } + /// Pool-side body: resolve the operands, and on error post back + /// immediately; the success path hands this task to a `ShellAsyncCpTask`, + /// whose completion is [`cp_on_finish`](Self::cp_on_finish). + fn run_on_pool(mut this: Box) { + if let Some(e) = this.resolve_operands() { + this.err = Some(e); + return ShellTask::on_finish::(this); } + let args = crate::node::fs::args::Cp::owned( + this.src_absolute.take().expect("resolve_operands set it"), + this.tgt_absolute.take().expect("resolve_operands set it"), + crate::node::fs::args::CpFlags { + recursive: this.opts.recursive, + force: true, + error_on_exist: false, + }, + ); + // Pool thread: hand the copy, and with it the loop and poster this + // shell task captured on its own thread, to an fs.cp task. + let poster = this + .task + .poster + .take() + .expect("shell cp task on the pool is armed"); + let event_loop = this.task.event_loop; + crate::node::fs::ShellAsyncCpTask::create_for_shell(args, event_loop, poster, this); } fn has_trailing_sep(path: &[u8]) -> bool { @@ -498,14 +488,10 @@ impl ShellCpTask { } } - /// Resolves src/tgt to absolute paths, classifies them per the three - /// POSIX `cp` synopses - /// (), then hands off to - /// the node:fs async cp implementation. - fn run_from_thread_pool_impl( - &mut self, - poster: &bun_jsc::ConcurrentPoster, - ) -> Option { + /// Resolves src/tgt to absolute paths (`src_absolute`/`tgt_absolute`), + /// classifying them per the three POSIX `cp` synopses + /// (). Returns the error. + fn resolve_operands(&mut self) -> Option { use resolve_path::{Platform, platform}; let mut buf2 = bun_paths::path_buffer_pool::get(); @@ -617,25 +603,8 @@ impl ShellCpTask { _copying_many = true; } - let args = crate::node::fs::args::Cp::owned( - src.as_bytes().to_vec(), - tgt.as_bytes().to_vec(), - crate::node::fs::args::CpFlags { - recursive: self.opts.recursive, - force: true, - error_on_exist: false, - }, - ); - - // Pool thread: hand the copy to an fs.cp task bound to the loop and - // poster this shell task captured on its own thread. - let _ = crate::node::fs::ShellAsyncCpTask::create_for_shell( - args, - self.task.event_loop, - poster.clone(), - std::ptr::from_mut::(self), - ); - + self.src_absolute = Some(src.as_bytes().to_vec()); + self.tgt_absolute = Some(tgt.as_bytes().to_vec()); None } } diff --git a/src/runtime/shell/builtin/export.rs b/src/runtime/shell/builtin/export.rs index 78ca23836cc4..60eec6bdac9f 100644 --- a/src/runtime/shell/builtin/export.rs +++ b/src/runtime/shell/builtin/export.rs @@ -43,27 +43,27 @@ impl Export { let val = EnvStr::dupe_ref_counted(value); drop(bltn); shell.borrow_mut().export_env.insert(label, val); - label.deref(); - val.deref(); } Builtin::done(interp, cmd, 0) } fn print_all(interp: &Interpreter, cmd: NodeId) -> Yield { - let mut entries: Vec<(EnvStr, EnvStr)> = Builtin::shell(interp, cmd) - .borrow() - .export_env - .iter() - .map(|(k, v)| (*k, *v)) - .collect(); - index_sort::sort_slice_by(&mut entries, |a, b| a.0.slice().cmp(b.0.slice())); - let mut buf = Vec::new(); - for (k, v) in &entries { - buf.extend_from_slice(k.slice()); - buf.push(b'='); - buf.extend_from_slice(v.slice()); - buf.push(b'\n'); + { + let shell = Builtin::shell(interp, cmd); + let shell = shell.borrow(); + let mut entries: Vec<(&[u8], &[u8])> = shell + .export_env + .iter() + .map(|(k, v)| (k.slice(), v.slice())) + .collect(); + index_sort::sort_slice_by(&mut entries, |a, b| a.0.cmp(b.0)); + for (k, v) in &entries { + buf.extend_from_slice(k); + buf.push(b'='); + buf.extend_from_slice(v); + buf.push(b'\n'); + } } let stdout_needs_io = Builtin::of(interp, cmd).stdout.needs_io(); diff --git a/src/runtime/shell/builtin/ls.rs b/src/runtime/shell/builtin/ls.rs index ceef155e75d3..d43864fbfe88 100644 --- a/src/runtime/shell/builtin/ls.rs +++ b/src/runtime/shell/builtin/ls.rs @@ -512,15 +512,15 @@ impl ShellLsTask { let mtime = bun_sys::stat_mtime(stat); let time_str = format_time(mtime.sec, self.now_secs); - // SAFETY: `format_permissions` only writes ASCII bytes (`r`/`w`/`x`/`s`/`S`/`t`/`T`/`-`). - let perms_str = unsafe { core::str::from_utf8_unchecked(&perms) }; - // SAFETY: `format_time` only writes ASCII bytes (month abbrevs, digits, spaces, `:`). - let time_s = unsafe { core::str::from_utf8_unchecked(&time_str) }; + self.output.push(file_type); + self.output.extend_from_slice(&perms); let _ = write!( self.output, - "{}{} {:>3} {:>5} {:>5} {:>8} {} ", - file_type as char, perms_str, nlink, uid, gid, size, time_s, + " {:>3} {:>5} {:>5} {:>8} ", + nlink, uid, gid, size ); + self.output.extend_from_slice(&time_str); + self.output.push(b' '); self.output.extend_from_slice(name); self.output.push(b'\n'); } diff --git a/src/runtime/shell/builtin/mkdir.rs b/src/runtime/shell/builtin/mkdir.rs index 02b201393a3f..a122581dea49 100644 --- a/src/runtime/shell/builtin/mkdir.rs +++ b/src/runtime/shell/builtin/mkdir.rs @@ -250,7 +250,7 @@ impl ShellMkdirTask { if this.opts.parents { let vtable = MkdirVerboseVTable { - inner: &raw mut this.created_directories, + inner: core::cell::RefCell::new(&mut this.created_directories), active: this.opts.verbose, }; if let Err(e) = node_fs.mkdir_recursive_impl(&args, &vtable) { @@ -282,23 +282,19 @@ impl ShellMkdirTask { // the keep-alive and the box. /// Collects each created directory into -/// `created_directories` (newline-separated) when `-v` is set. Passed by value -/// to `NodeFS::mkdir_recursive_impl`; `on_create_dir` writes through the raw -/// back-ref because the trait method takes `&self`. -struct MkdirVerboseVTable { - inner: *mut Vec, +/// `created_directories` (newline-separated) when `-v` is set +/// (`MkdirCtx::on_create_dir` takes `&self`). +struct MkdirVerboseVTable<'a> { + inner: core::cell::RefCell<&'a mut Vec>, active: bool, } -impl MkdirCtx for MkdirVerboseVTable { +impl MkdirCtx for MkdirVerboseVTable<'_> { fn on_create_dir(&self, dirpath: &bun_paths::OSPathSliceZ) { if !self.active { return; } - // SAFETY: `inner` points at `ShellMkdirTask::created_directories`; the - // worker thread is the sole accessor for the duration of - // `run_from_thread_pool`, and `mkdir_recursive_impl` does not alias it. - let out = unsafe { &mut *self.inner }; + let mut out = self.inner.borrow_mut(); #[cfg(windows)] { let mut buf = bun_paths::path_buffer_pool::get(); diff --git a/src/runtime/shell/builtin/rm.rs b/src/runtime/shell/builtin/rm.rs index 36b8a0db8823..95d47c7a8873 100644 --- a/src/runtime/shell/builtin/rm.rs +++ b/src/runtime/shell/builtin/rm.rs @@ -1,4 +1,5 @@ -use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use bun_core::{ZBox, ZStr, strings}; use bun_paths::resolve_path::{self, Platform, platform}; @@ -6,9 +7,7 @@ use bun_sys::{E, FdExt, dir_iterator}; use crate::shell::ExitCode; use crate::shell::builtin::{Builtin, IoKind, Kind}; -use crate::shell::interpreter::{ - EventLoopHandle, Interpreter, NodeId, ShellTask, WorkPoolTask, shell_openat, -}; +use crate::shell::interpreter::{Interpreter, NodeId, ShellTask, ShellTaskCtx, shell_openat}; use crate::shell::io_writer::{ChildPtr, WriterTag}; use crate::shell::yield_::Yield; @@ -41,7 +40,8 @@ pub struct ExecState { pub(crate) error_signal: AtomicBool, pub(crate) output_done: AtomicUsize, pub(crate) output_count: AtomicUsize, - pub(crate) verbose_seq: u32, + /// Numbers the chunks queued on a writer so each is called back. + pub(crate) chunk_seq: u32, pub(crate) tasks_done: usize, pub(crate) started: bool, } @@ -51,6 +51,11 @@ impl ExecState { fn output_drained(&self) -> bool { self.output_done.load(Ordering::SeqCst) >= self.output_count.load(Ordering::SeqCst) } + + fn next_chunk(&mut self, cmd: NodeId) -> ChildPtr { + self.chunk_seq += 1; + ChildPtr::builtin_task(cmd, self.chunk_seq) + } } #[derive(Clone, Copy)] @@ -251,7 +256,7 @@ impl Rm { error_signal: AtomicBool::new(false), output_done: AtomicUsize::new(0), output_count: AtomicUsize::new(0), - verbose_seq: 0, + chunk_seq: 0, tasks_done: 0, started: false, }); @@ -323,10 +328,7 @@ impl Rm { }; for i in args_start..argc { let root = Builtin::of(interp, cmd).arg_bytes(i).to_vec(); - let task = - ShellRmTask::create(cmd, opts, &root, cwd, sig, out_count, interp); - // SAFETY: freshly heap-allocated. - unsafe { ShellRmTask::schedule(task) }; + ShellRmTask::start(cmd, opts, &root, cwd, sig, out_count, interp); } } return Yield::suspended(); @@ -383,17 +385,10 @@ impl Rm { } } - /// # Safety - /// `task` must be a live `heap::alloc`'d [`ShellRmTask`]; main thread. - fn on_shell_rm_task_done(interp: &Interpreter, cmd: NodeId, task: *mut ShellRmTask) { - // In verbose mode the root DirTask may also be queued for write_verbose; - // both callbacks hold a pending count and the last one to run frees the - // ShellRmTask. - // SAFETY: `task` is a live heap-allocated allocation; main thread. - scopeguard::defer! { unsafe { ShellRmTask::decr_pending_and_maybe_deinit(task) }; } - - // SAFETY: `task` is live; exclusive on main thread until decr above runs. - let task_err = unsafe { (*task).err.get_mut().take() }; + /// The tree under one root argument is done (main thread). + fn on_shell_rm_task_done(interp: &Interpreter, cmd: NodeId, task: Box) { + let task_err = task.tree.err.lock().take(); + drop(task); // Format the error string before // stashing the error on `exec` (formatting needs &mut interp). let errstr: Option> = task_err @@ -405,10 +400,7 @@ impl Rm { }; exec.tasks_done += 1; if let Some(e) = task_err { - // Ownership of err.path stayed with the task (freed in - // ShellRmTask::deinit); exec.err is only used as a - // did-anything-fail flag from here, so drop the - // soon-to-be-dangling path slice from our copy. + // Only used as a did-anything-fail flag from here. exec.err = Some(e.without_path()); } (exec.tasks_done, exec.total_tasks) @@ -417,15 +409,16 @@ impl Rm { if let Some(s) = errstr { let stderr_needs_io = Builtin::of(interp, cmd).stderr.needs_io(); if let Some(safeguard) = stderr_needs_io { - let seq = match &mut Self::state_mut(interp, cmd).state { - RmState::Exec(exec) => { - exec.output_count.fetch_add(1, Ordering::SeqCst); - exec.verbose_seq += 1; - exec.verbose_seq - } - _ => 1, + // One error chunk per root can be in flight; number them so + // each is called back (a failed writer calls equal children + // back once). + let child = { + let RmState::Exec(exec) = &mut Self::state_mut(interp, cmd).state else { + unreachable!() + }; + exec.output_count.fetch_add(1, Ordering::SeqCst); + exec.next_chunk(cmd) }; - let child = ChildPtr::builtin_task(cmd, seq); Builtin::write_out(interp, cmd, IoKind::Stderr, child, &s, safeguard).run(interp); return; } @@ -452,44 +445,21 @@ impl Rm { } } - /// Flushes a `DirTask`'s buffered list of - /// deleted paths to stdout, then frees the DirTask (non-root) and releases - /// the pending-main-callback count taken in `DirTask::post_run`. - fn write_verbose(interp: &Interpreter, cmd: NodeId, verbose: *mut DirTask) -> Yield { - // SAFETY: `verbose` is a live DirTask posted via queue_for_write; main - // thread, exclusive. Take the buffer up-front so the cleanup guard can - // own the raw pointers without overlapping a borrow. - let (tm, has_parent, buf) = unsafe { - ( - (*verbose).task_manager, - !(*verbose).parent_task.is_null(), - core::mem::take(&mut (*verbose).deleted_entries), - ) - }; - let _guard = scopeguard::guard((tm, verbose, has_parent), |(tm, v, hp)| { - // SAFETY: non-root DirTask is its own Box (reclaim); pending count - // was bumped in `post_run` before `queue_for_write`. - unsafe { - if hp { - DirTask::deinit(v); - } - ShellRmTask::decr_pending_and_maybe_deinit(tm); - } - }); + /// Flushes a `DirTask`'s buffered list of deleted paths to stdout (main + /// thread); the task is done with. + fn write_verbose(interp: &Interpreter, cmd: NodeId, mut dir: Box) -> Yield { + let buf = core::mem::take(&mut dir.deleted_entries); + drop(dir); let stdout_needs_io = Builtin::of(interp, cmd).stdout.needs_io(); if let Some(safeguard) = stdout_needs_io { // One chunk per DirTask can be in flight; number them so each is // called back (a failed writer calls equal children back once). - let seq = match &mut Self::state_mut(interp, cmd).state { - RmState::Exec(exec) => { - exec.verbose_seq += 1; - exec.verbose_seq - } - _ => 1, + let child = match &mut Self::state_mut(interp, cmd).state { + RmState::Exec(exec) => exec.next_chunk(cmd), + _ => ChildPtr::builtin_task(cmd, 1), }; - let child = ChildPtr::builtin_task(cmd, seq); return Builtin::write_out(interp, cmd, IoKind::Stdout, child, &buf, safeguard); } let _ = Builtin::write_no_io(interp, cmd, IoKind::Stdout, &buf); @@ -611,70 +581,100 @@ pub enum EntryKindHint { Dir, } -/// One per filepath argument; owns the root -/// [`DirTask`] and tracks the cross-thread error state. +/// One per filepath argument: rides inside the root [`DirTask`] while the +/// tree under it is deleted on the pool, then is posted back to the owning +/// thread ([`Rm::on_shell_rm_task_done`]). pub struct ShellRmTask { - pub(crate) cmd: NodeId, - pub(crate) opts: Opts, - pub(crate) cwd: bun_sys::Fd, - /// The root DirTask lives in - /// its own `heap::alloc`'d allocation so that `&ShellRmTask` (held as - /// the `&self` receiver throughout `remove_entry*`) never overlaps the - /// `&mut DirTask` borrows those methods take on the root — embedding it - /// would make every `verbose_deleted` call on the root UB under Stacked - /// Borrows. Freed in `Drop`. - pub(crate) root_task: *mut DirTask, - /// Backref into `Rm::ExecState.error_signal`; the boxed `Rm` ExecState - /// outlives every in-flight `ShellRmTask`. - pub(crate) error_signal: bun_ptr::BackRef, - /// Backref into `Rm::ExecState.output_count` so [`verbose_deleted`] can - /// bump it from worker threads. - output_count: bun_ptr::BackRef, - /// Main-thread callbacks that must complete before this task can be freed: - /// always one for `on_shell_rm_task_done` (via `finish_concurrently`), plus - /// one per DirTask whose verbose output was queued. Decremented by - /// [`decr_pending_and_maybe_deinit`]. - pub(crate) pending_main_callbacks: AtomicU32, - /// First error hit by any worker thread. Mutex-wrapped so [`handle_err`] - /// can take `&self` without an interior `&mut` cast. - pub(crate) err: bun_threading::Guarded>, - pub(crate) join_style: JoinStyle, - pub(crate) event_loop: EventLoopHandle, + pub(crate) tree: Arc, pub task: ShellTask, } -/// One per directory in the recursive -/// walk; root and children alike are heap-allocated (see the comment on -/// [`ShellRmTask::root_task`]). +// `runtime::dispatch::run_task`'s `task_tag::ShellRmTask` arm reboxes the +// pointer `ShellTask::on_finish` posted; a completion that will not run drops +// the keep-alive and the box. +bun_event_loop::boxed_taskable!(ShellRmTask, ShellRmTask, |this| this.task.unref_unrun()); + +/// Shared by every [`DirTask`] under one root argument, from any pool thread. +pub struct RmTree { + cmd: NodeId, + opts: Opts, + cwd: bun_sys::Fd, + /// Into `Rm`'s `ExecState`, which stays in place until every task it + /// started has reported back. + error_signal: bun_ptr::BackRef, + /// As `error_signal`; bumped once per `DirTask` that has verbose output. + output_count: bun_ptr::BackRef, + /// First error hit by any worker thread. + err: bun_threading::Guarded>, + join_style: JoinStyle, +} + +/// One per directory (and per root path) in the recursive walk. Whoever holds +/// the box owns the non-atomic fields: the pool thread running it, then — once +/// it has handed off to its children — the last child to finish (through +/// [`Children`]), then the main thread for the verbose write. pub struct DirTask { - pub(crate) task_manager: *mut ShellRmTask, - pub(crate) parent_task: *mut DirTask, - pub path: ZBox, - pub(crate) is_absolute: bool, - pub(crate) subtask_count: AtomicUsize, - pub(crate) need_to_wait: AtomicBool, - pub(crate) deleting_after_waiting_for_children: AtomicBool, - pub(crate) kind_hint: EntryKindHint, - pub(crate) deleted_entries: Vec, - /// Intrusive node for the verbose-write bounce-back to the main thread. - pub(crate) concurrent_task: bun_event_loop::EventLoopTask, - /// Intrusive node for the thread-pool dispatch. - pub(crate) work_task: WorkPoolTask, + info: DirInfo, + deleted_entries: Vec, + /// Set once this directory has spawned a child task. + children: Option>, + pub task: ShellTask, +} + +struct DirInfo { + tree: Arc, + path: ZBox, + kind_hint: EntryKindHint, + is_absolute: bool, + /// The parent directory's [`Children`]; `None` for a root. + parent: Option>, + /// Root only: the rm task to post back once the whole tree is done. + rm: Option>, } -// SAFETY: raw-pointer fields are only dereferenced on the threads that own -// them (worker pool / main thread); the surrounding atomics + `err` mutex -// provide the necessary synchronisation. -unsafe impl Send for ShellRmTask {} -// SAFETY: `task_manager` / `parent_task` point at heap allocations kept alive -// by the `subtask_count` / `pending_main_callbacks` atomic protocol; non-atomic -// fields are single-owner per the `need_to_wait` handoff (see -// `verbose_deleted`). -unsafe impl Send for DirTask {} +// Pool entry is `run_owned`; `runtime::dispatch::run_task`'s +// `task_tag::ShellRmDirTask` arm reboxes the verbose-output hop (a hop that +// will not run just drops the box: it took no keep-alive). +bun_threading::owned_task!(DirTask, task.task, run = DirTask::run_owned); +bun_event_loop::boxed_taskable!(DirTask, ShellRmDirTask); + +/// A directory's own slot plus one per unfinished child task. Before giving +/// up its slot the directory parks its box here; whoever takes `pending` to +/// zero owns the box (and gets the slot back). +struct Children { + pending: AtomicUsize, + parked: bun_threading::Guarded>>, +} + +impl Children { + fn new() -> Children { + Children { + pending: AtomicUsize::new(1), + parked: bun_threading::Guarded::new(None), + } + } + + /// The holder of `dir` gives up its slot. `Some` if every child had + /// already finished: the caller keeps the directory. + fn release(&self, dir: Box) -> Option> { + *self.parked.lock() = Some(dir); + self.done() + } + + /// One slot is done; the last one out takes the parked directory. + fn done(&self) -> Option> { + if self.pending.fetch_sub(1, Ordering::SeqCst) != 1 { + return None; + } + self.pending.store(1, Ordering::SeqCst); + let dir = self.parked.lock().take(); + debug_assert!(dir.is_some(), "rm: last child out but no parked directory"); + dir + } +} impl ShellRmTask { - #[allow(clippy::too_many_arguments)] - pub(crate) fn create( + pub(crate) fn start( cmd: NodeId, opts: Opts, root_path: &[u8], @@ -682,214 +682,72 @@ impl ShellRmTask { error_signal: bun_ptr::BackRef, output_count: bun_ptr::BackRef, interp: &Interpreter, - ) -> *mut ShellRmTask { - let evtloop = interp.event_loop; - let join_style = JoinStyle::from_path(root_path); - // Separate allocation — see the comment on `root_task`. - let root_task = bun_core::heap::into_raw(Box::new(DirTask { - // task_manager is fixed up below once we have the ShellRmTask address. - task_manager: core::ptr::null_mut(), - parent_task: core::ptr::null_mut(), - path: ZBox::from_bytes(root_path), - is_absolute: false, - subtask_count: AtomicUsize::new(1), - need_to_wait: AtomicBool::new(false), - deleting_after_waiting_for_children: AtomicBool::new(false), - kind_hint: EntryKindHint::Idk, - deleted_entries: Vec::new(), - concurrent_task: bun_event_loop::EventLoopTask::from_event_loop(evtloop), - work_task: WorkPoolTask { - node: Default::default(), - callback: DirTask::work_pool_callback, - }, - })); - let boxed = Box::new(ShellRmTask { + ) { + let tree = Arc::new(RmTree { cmd, opts, cwd, - root_task, error_signal, output_count, - pending_main_callbacks: AtomicU32::new(1), err: bun_threading::Guarded::new(None), - join_style, - event_loop: evtloop, + join_style: JoinStyle::from_path(root_path), + }); + let mut rm = Box::new(ShellRmTask { + tree: Arc::clone(&tree), + task: ShellTask::new(interp), + }); + // Posted back from whichever pool thread finishes the tree, so take + // the keep-alive and poster here on the owning thread. + rm.task + .keep_alive + .ref_(rm.task.event_loop.as_event_loop_ctx()); + rm.task.arm(); + let root = Box::new(DirTask { + info: DirInfo { + tree, + path: ZBox::from_bytes(root_path), + kind_hint: EntryKindHint::Idk, + is_absolute: false, + parent: None, + rm: Some(rm), + }, + deleted_entries: Vec::new(), + children: None, task: ShellTask::new(interp), }); - let raw = bun_core::heap::into_raw(boxed); - // SAFETY: both freshly leaked; exclusive. - unsafe { (*root_task).task_manager = raw }; - raw + ShellTask::schedule_no_ref(root); } +} - /// Unlike most shell builtins this does NOT use the generic - /// [`ShellTask::schedule`] trampoline (which auto-enqueues back to main - /// on return): the recursive - /// DirTask tree owns the bounce-back via [`finish_concurrently`]. - /// - /// # Safety - /// `this` must be a fresh `heap::alloc`'d task (see [`create`]). - pub(crate) unsafe fn schedule(this: *mut ShellRmTask) { - use bun_threading::work_pool::WorkPool; - // SAFETY: `this` is live; `task` is the embedded `ShellTask`. Stay on - // raw pointers — once `WorkPool::schedule` returns the worker thread - // may already be running. - unsafe { - let st = &raw mut (*this).task; - (*st).task.callback = Self::work_pool_callback; - (*st).keep_alive.ref_((*st).event_loop.as_event_loop_ctx()); - (*st).arm(); - WorkPool::schedule(&raw mut (*st).task); - } +impl ShellTaskCtx for ShellRmTask { + fn shell_task(&self) -> &ShellTask { + &self.task } - - /// Recover `*ShellRmTask` from the intrusive `*WorkPoolTask` and run the - /// root DirTask. - unsafe fn work_pool_callback(task: *mut WorkPoolTask) { - // SAFETY: `task` is `self.task.task`, embedded in a live heap-allocated - // `ShellRmTask`; the worker thread has exclusive access to - // `root_task` until it spawns subtasks. - unsafe { - let this = - bun_ptr::container_of::(task, core::mem::offset_of!(Self, task.task)); - DirTask::run_from_thread_pool_impl((*this).root_task); - } + fn shell_task_mut(&mut self) -> &mut ShellTask { + &mut self.task } - - /// Post this task to the main-thread - /// concurrent queue; routed by `dispatch.rs` → [`run_from_main_thread`]. - /// - /// # Safety - /// `this` is the live `heap::alloc`'d task; not touched again on this - /// thread after return (unless a verbose pending-count keeps it alive). - unsafe fn finish_concurrently(this: *mut ShellRmTask) { - use bun_event_loop::{ArmedLoopTask, ConcurrentTask::AutoDeinit, EventLoopTask}; - use core::ptr::NonNull; - // SAFETY: caller contract. Raw place access only: queued verbose hops - // may read `*this` on the main thread concurrently, so no `Box`/`&mut` - // may cover it here; `poster` is moved out before the post, after - // which the main thread may free `*this`. - unsafe { - let poster = (*this) - .task - .poster - .take() - .expect("shell task on the pool is armed"); - let armed = match &mut (*this).task.concurrent_task { - EventLoopTask::Js(ct) => { - ArmedLoopTask::Js(NonNull::from(ct.from(this, AutoDeinit::ManualDeinit))) - } - EventLoopTask::Mini(at) => ArmedLoopTask::Mini( - NonNull::new(at.from(this, shell_rm_task_run_from_main_thread_mini)) - .expect("intrusive task"), - ), - }; - poster.post(armed); - drop(poster); - } + fn run_from_thread_pool(&mut self) { + // Not scheduled itself: it rides inside the root DirTask. + debug_assert!(false, "ShellRmTask is not a pool task"); } - - /// # Safety - /// `this` is a live `heap::alloc`'d task; main thread. - pub(crate) unsafe fn decr_pending_and_maybe_deinit(this: *mut ShellRmTask) { - // SAFETY: caller contract; paired with `heap::alloc` in `create`. - unsafe { - if (*this) - .pending_main_callbacks - .fetch_sub(1, Ordering::SeqCst) - == 1 - { - drop(bun_core::heap::take(this)); - } - } + fn run_from_main_thread(self: Box, interp: &Interpreter) { + let cmd = self.tree.cmd; + Rm::on_shell_rm_task_done(interp, cmd, self); } +} +impl RmTree { #[inline] - fn error_signal(&self) -> &AtomicBool { - // `error_signal` is a `BackRef` into the boxed `Rm` ExecState which - // outlives every in-flight ShellRmTask. - self.error_signal.get() + fn errored(&self) -> bool { + self.error_signal.load(Ordering::SeqCst) } - /// Joins `path` onto `parent_dir.path` and spawns - /// a child DirTask. - fn enqueue( - &self, - parent_dir: *mut DirTask, - path: &[u8], - is_absolute: bool, - kind_hint: EntryKindHint, - ) { - if self.error_signal().load(Ordering::SeqCst) { - return; - } - // SAFETY: `parent_dir` is live for the duration of its run_from_thread_pool_impl. - let parent_path = unsafe { (*parent_dir).path.as_bytes() }; - let new_path = self.join(&[parent_path, path], is_absolute); - self.enqueue_no_join(parent_dir, new_path, kind_hint); - } - - /// Takes ownership of `path`. - fn enqueue_no_join(&self, parent: *mut DirTask, path: ZBox, kind_hint: EntryKindHint) { - if self.error_signal().load(Ordering::SeqCst) { - return; - } - // SAFETY: `parent` is live; reuse its `task_manager` (preserves the - // original `*mut` provenance from `heap::alloc` rather than deriving - // a writeable pointer from `&self`). - let task_manager = unsafe { (*parent).task_manager }; - let subtask = bun_core::heap::into_raw(Box::new(DirTask { - task_manager, - parent_task: parent, - path, - is_absolute: false, - subtask_count: AtomicUsize::new(1), - need_to_wait: AtomicBool::new(false), - deleting_after_waiting_for_children: AtomicBool::new(false), - kind_hint, - deleted_entries: Vec::new(), - concurrent_task: bun_event_loop::EventLoopTask::from_event_loop(self.event_loop), - work_task: WorkPoolTask { - node: Default::default(), - callback: DirTask::work_pool_callback, - }, - })); - // SAFETY: `parent` is live until its `subtask_count` drains to 0; - // `subtask` is freshly heap-allocated. - unsafe { - let count = (*parent).subtask_count.fetch_add(1, Ordering::Relaxed); - debug_assert!(count > 0); - bun_threading::work_pool::WorkPool::schedule(&raw mut (*subtask).work_task); - } - } - - /// Takes `dir_task` as a raw pointer (not `&mut DirTask`) so callers in - /// `remove_entry*` — which already hold `&self: &ShellRmTask` and a - /// `&ZStr` borrowed from `dir_task.path` — never materialise an aliasing - /// `&mut DirTask`. Only the disjoint `deleted_entries` field is reborrowed - /// mutably here. - fn verbose_deleted(&self, dir_task: *mut DirTask, path: &[u8]) -> bun_sys::Maybe<()> { - if !self.opts.verbose { - return Ok(()); - } - // SAFETY: a DirTask's non-atomic fields are single-owner-at-a-time. - // Ownership starts on the worker that runs `remove_entry*` and is - // handed off via `subtask_count`: the parent's final append here is - // sequenced-before its `subtask_count.fetch_sub(SeqCst)` in - // `remove_entry_dir`, and the child taking the counter to 0 in - // `post_run` acquire-reads from that release sequence before - // re-entering via `delete_after_waiting_for_children`. That pair - // makes prior `deleted_entries` writes visible to the new owner, so - // this `&mut` is exclusive and race-free. The reborrow is also - // disjoint from every other live borrow (`&self`, `path`). - let entries = unsafe { &mut (*dir_task).deleted_entries }; - if entries.is_empty() { - // `output_count` is a `BackRef` into the boxed `Rm` ExecState. - self.output_count.fetch_add(1, Ordering::SeqCst); + fn handle_err(&self, err: bun_sys::Error) { + let mut slot = self.err.lock(); + if slot.is_none() { + *slot = Some(err); + self.error_signal.store(true, Ordering::SeqCst); } - entries.extend_from_slice(path); - entries.push(b'\n'); - Ok(()) } /// Join into `buf` honoring [`join_style`]. @@ -940,61 +798,227 @@ impl ShellRmTask { let joined = resolve_path::join_spill::(&mut spill, parts); ZBox::from_bytes(joined) } +} - #[inline] - fn error_with_path(&self, e: &bun_sys::Error, path: &[u8]) -> bun_sys::Error { - e.with_path(path) +// ── DirTask ──────────────────────────────────────────────────────────────── + +impl DirTask { + fn walk(&mut self) -> Walk<'_> { + Walk { + info: &self.info, + entries: &mut self.deleted_entries, + children: &mut self.children, + task: &self.task, + } } - /// Returns `Ok(true)` when [`remove_entry_dir`] handed `dir_task` off to - /// a child (its `subtask_count` slot was released without reaching 0). - /// Once that happens, a child finishing on another thread may run - /// [`DirTask::delete_after_waiting_for_children`] → `post_run` → `deinit` - /// (or, for the root, the main thread may free the whole - /// [`ShellRmTask`]). Callers must therefore treat `dir_task` as - /// potentially-freed when this returns `Ok(true)`. The bool is threaded - /// out locally instead of re-reading the atomic so the caller never - /// dereferences `dir_task` to find out. - fn remove_entry(&self, dir_task: *mut DirTask, is_absolute: bool) -> bun_sys::Maybe { - let mut waiting = false; - let mut buf = bun_paths::path_buffer_pool::get(); - // SAFETY: `dir_task` is live; this thread owns it. `kind_hint` / - // `path` are read-only after construction. - let (kind_hint, path) = unsafe { ((*dir_task).kind_hint, (*dir_task).path.as_zstr()) }; - match kind_hint { - EntryKindHint::Idk => { - let mut vtable = RemoveFileVTable { - task: self, - on_dir: OnDir::Recurse { - need_to_wait_out: &mut waiting, - }, - }; - self.remove_entry_file(dir_task, path, is_absolute, &mut buf, &mut vtable)?; + /// Pool thread: delete this entry (recursing into subdirectories as their + /// own tasks), then either finish it or leave that to the last child. + pub fn run_owned(mut this: Box) { + this.info.is_absolute = Platform::AUTO.is_absolute(this.info.path.as_bytes()); + let mut handed_off = false; + if let Err(err) = this.walk().remove_entry(&mut handed_off) { + this.info.tree.handle_err(err); + } + if handed_off { + let children = Arc::clone(this.children.as_ref().expect("handed off to a child")); + this = match children.release(this) { + // A child still runs; the last one finishes this directory. + None => return, + // Every child already finished: ours again. + Some(dir) => dir, + }; + if let Err(err) = this.walk().remove_dir_itself() { + this.info.tree.handle_err(err); + } + } + Self::post_run(this); + } + + /// `this` and everything under it is deleted (or failed): queue its + /// verbose output, then report to the parent directory or, for a root, + /// post the rm task back. + fn post_run(mut this: Box) { + debug_assert!( + this.children + .as_ref() + .is_none_or(|c| c.pending.load(Ordering::SeqCst) == 1), + "rm: directory finished with a child task outstanding" + ); + let parent = this.info.parent.take(); + let rm = this.info.rm.take(); + if this.info.tree.opts.verbose && !this.deleted_entries.is_empty() { + // To `Rm::write_verbose`. Posted before the parent (or the rm + // task) can complete, so it is written first. + ShellTask::on_finish(this); + } else { + drop(this); + } + + if let Some(parent) = parent { + if let Some(dir) = parent.done() { + Self::delete_after_waiting_for_children(dir); + } + return; + } + ShellTask::on_finish(rm.expect("root DirTask carries the rm task")); + } + + /// Pool thread, as the last child of `dir` to finish: delete the (now + /// empty) directory itself. + fn delete_after_waiting_for_children(mut dir: Box) { + loop { + let mut spawned = false; + if !dir.info.tree.errored() { + match dir.walk().remove_entry_dir_after_children() { + Err(e) => dir.info.tree.handle_err(e), + Ok(deleted) => spawned = !deleted, + } } - EntryKindHint::Dir => { - self.remove_entry_dir(dir_task, is_absolute, &mut buf, &mut waiting)?; + if !spawned { + return Self::post_run(dir); } + // Not empty after all and a child was spawned for it: wait again. + let children = Arc::clone(dir.children.as_ref().expect("spawned a child")); + dir = match children.release(dir) { + None => return, + Some(d) => d, + }; + } + } +} + +impl ShellTaskCtx for DirTask { + fn shell_task(&self) -> &ShellTask { + &self.task + } + fn shell_task_mut(&mut self) -> &mut ShellTask { + &mut self.task + } + fn run_from_thread_pool(&mut self) { + // `owned_task!(DirTask, .., run = DirTask::run_owned)`. + debug_assert!(false, "DirTask runs through run_owned"); + } + /// The verbose-output hop. + fn run_from_main_thread(self: Box, interp: &Interpreter) { + let cmd = self.info.tree.cmd; + Rm::write_verbose(interp, cmd, self).run(interp); + } +} + +/// A [`DirTask`] being worked on by the thread that holds it. +struct Walk<'a> { + info: &'a DirInfo, + entries: &'a mut Vec, + children: &'a mut Option>, + task: &'a ShellTask, +} + +/// What to do when the entry being unlinked turns out to be a directory. +enum OnDir<'a> { + /// Nothing (it was reported as not-a-directory a moment ago). + Ignore, + /// A child found while iterating a directory: spawn a [`DirTask`] for it. + Enqueue, + /// The task's own path ([`Walk::remove_entry`]): recurse into + /// [`Walk::remove_entry_dir`]. + Recurse { handed_off: &'a mut bool }, + /// The task's own path once its children are done. + Parent(&'a mut RemoveFileParent), +} + +struct RemoveFileParent { + treat_as_dir: bool, + #[cfg(not(any(target_os = "linux", target_os = "android")))] + allow_enqueue: bool, + enqueued: bool, +} + +impl Walk<'_> { + #[inline] + fn tree(&self) -> &RmTree { + &self.info.tree + } + + /// Spawns a child task for `path` under this directory. + fn enqueue(&mut self, path: &[u8], is_absolute: bool, kind_hint: EntryKindHint) { + if self.tree().errored() { + return; + } + let new_path = self + .tree() + .join(&[self.info.path.as_bytes(), path], is_absolute); + self.enqueue_no_join(new_path, kind_hint); + } + + /// Takes ownership of `path`. + fn enqueue_no_join(&mut self, path: ZBox, kind_hint: EntryKindHint) { + if self.tree().errored() { + return; + } + let children = self + .children + .get_or_insert_with(|| Arc::new(Children::new())); + let subtask = Box::new(DirTask { + info: DirInfo { + tree: Arc::clone(&self.info.tree), + path, + kind_hint, + is_absolute: false, + parent: Some(Arc::clone(children)), + rm: None, + }, + deleted_entries: Vec::new(), + children: None, + // Pool thread: the subtask reports to our loop through a clone of + // our poster; no keep-alive ref (no VM thread-local here). + task: ShellTask::new_child(self.task), + }); + let count = children.pending.fetch_add(1, Ordering::Relaxed); + debug_assert!(count > 0); + ShellTask::schedule_no_ref(subtask); + } + + fn verbose_deleted(&mut self, path: &[u8]) -> bun_sys::Maybe<()> { + if !self.tree().opts.verbose { + return Ok(()); + } + if self.entries.is_empty() { + self.tree().output_count.fetch_add(1, Ordering::SeqCst); + } + self.entries.extend_from_slice(path); + self.entries.push(b'\n'); + Ok(()) + } + + /// `handed_off` is set when [`remove_entry_dir`](Self::remove_entry_dir) + /// spawned children: the directory itself is then deleted by whoever + /// finishes last (see [`DirTask::run_owned`]). + fn remove_entry(&mut self, handed_off: &mut bool) -> bun_sys::Maybe<()> { + let mut buf = bun_paths::path_buffer_pool::get(); + let (path, is_absolute) = (self.info.path.as_zstr(), self.info.is_absolute); + match self.info.kind_hint { + EntryKindHint::Idk => self.remove_entry_file( + path, + is_absolute, + &mut buf, + &mut OnDir::Recurse { handed_off }, + ), + EntryKindHint::Dir => self.remove_entry_dir(is_absolute, &mut buf, handed_off), } - Ok(waiting) } - /// `need_to_wait_out` is left `true` only when the hand-off in this - /// function gave ownership of `dir_task` to a child; see - /// [`remove_entry`] for why the caller needs this on its own stack - /// rather than reading it back from `dir_task`. fn remove_entry_dir( - &self, - dir_task: *mut DirTask, + &mut self, is_absolute: bool, buf: &mut bun_paths::PathBuffer, - need_to_wait_out: &mut bool, + handed_off: &mut bool, ) -> bun_sys::Maybe<()> { - // SAFETY: `dir_task` is live; this thread owns it. - let path = unsafe { (*dir_task).path.as_zstr() }; - let dirfd = self.cwd; + let path = self.info.path.as_zstr(); + let dirfd = self.tree().cwd; // If `-d` is specified without `-r` then we can just use `rmdirat`. - if self.opts.remove_empty_dirs && !self.opts.recursive { + if self.tree().opts.remove_empty_dirs && !self.tree().opts.recursive { let mut state = RemoveFileParent { treat_as_dir: true, #[cfg(not(any(target_os = "linux", target_os = "android")))] @@ -1006,25 +1030,30 @@ impl ShellRmTask { Ok(()) => return Ok(()), Err(e) => match e.get_errno() { E::ENOENT => { - if self.opts.force { - return self.verbose_deleted(dir_task, path.as_bytes()); + if self.tree().opts.force { + return self.verbose_deleted(path.as_bytes()); } - return Err(self.error_with_path(&e, path.as_bytes())); + return Err(e.with_path(path.as_bytes())); } E::ENOTDIR => { state.treat_as_dir = false; - self.remove_entry_file(dir_task, path, is_absolute, buf, &mut state)?; + self.remove_entry_file( + path, + is_absolute, + buf, + &mut OnDir::Parent(&mut state), + )?; if !state.treat_as_dir { return Ok(()); } } - _ => return Err(self.error_with_path(&e, path.as_bytes())), + _ => return Err(e.with_path(path.as_bytes())), }, } } } - if !self.opts.recursive { + if !self.tree().opts.recursive { return Err( bun_sys::Error::from_code(E::EISDIR, bun_sys::Tag::TODO).with_path(path.as_bytes()) ); @@ -1039,16 +1068,15 @@ impl ShellRmTask { Ok(fd) => fd, Err(e) => match e.get_errno() { E::ENOENT => { - if self.opts.force { - return self.verbose_deleted(dir_task, path.as_bytes()); + if self.tree().opts.force { + return self.verbose_deleted(path.as_bytes()); } - return Err(self.error_with_path(&e, path.as_bytes())); + return Err(e.with_path(path.as_bytes())); } E::ENOTDIR => { - let mut dummy = DummyRemoveFile; - return self.remove_entry_file(dir_task, path, is_absolute, buf, &mut dummy); + return self.remove_entry_file(path, is_absolute, buf, &mut OnDir::Ignore); } - _ => return Err(self.error_with_path(&e, path.as_bytes())), + _ => return Err(e.with_path(path.as_bytes())), }, }; @@ -1060,59 +1088,48 @@ impl ShellRmTask { } }); - if self.error_signal().load(Ordering::SeqCst) { + if self.tree().errored() { return Ok(()); } let mut iterator = dir_iterator::iterate(fd); - let mut child_vtable = RemoveFileVTable { - task: self, - on_dir: OnDir::Enqueue, - }; - // The loop may have already enqueued child DirTasks (bumping - // `dir_task.subtask_count`) when a readdir/unlink error or an - // `error_signal` from another worker aborts it. Returning from - // inside the loop would skip the `need_to_wait` hand-off below, - // so the last child's `post_run` would never drive - // `delete_after_waiting_for_children` on this dir and the owning - // `ShellRmTask` would never be freed. + // An error (readdir/unlink, or `error_signal` from another worker) + // may abort the loop after children were already spawned; the + // hand-off below must still happen so the last child finishes this + // directory. let mut i: usize = 0; let mut join_spill = Vec::new(); let loop_result: bun_sys::Maybe<()> = loop { let current = match iterator.next() { - Err(e) => break Err(self.error_with_path(&e, path.as_bytes())), + Err(e) => break Err(e.with_path(path.as_bytes())), Ok(None) => break Ok(()), Ok(Some(ent)) => ent, }; // TODO this seems bad maybe better to listen to kqueue/epoll event - if (i & 3) == 0 && self.error_signal().load(Ordering::SeqCst) { + if (i & 3) == 0 && self.tree().errored() { break Ok(()); } i += 1; match current.kind { bun_sys::EntryKind::Directory => { - self.enqueue( - dir_task, - current.name.slice_u8(), - is_absolute, - EntryKindHint::Dir, - ); + self.enqueue(current.name.slice_u8(), is_absolute, EntryKindHint::Dir); } _ => { let name = current.name.slice_u8(); // Copy the join into an owned ZBox so `buf` is free to - // be re-borrowed by the vtable callback. + // be re-borrowed while removing the file. let file_path = { - let joined = self.buf_join(buf, &mut join_spill, &[path.as_bytes(), name]); + let joined = + self.tree() + .buf_join(buf, &mut join_spill, &[path.as_bytes(), name]); ZBox::from_bytes(joined.as_bytes()) }; if let Err(e) = self.remove_entry_file( - dir_task, file_path.as_zstr(), is_absolute, buf, - &mut child_vtable, + &mut OnDir::Enqueue, ) { break Err(e); } @@ -1120,47 +1137,15 @@ impl ShellRmTask { } }; - // Stash any loop error before the hand-off — once `need_to_wait` - // is published `self` may be freed, and `handle_err` takes a - // mutex so it must not sit inside the hand-off below. The - // `error_signal` check afterwards covers the no-children early-out. if let Err(e) = loop_result { - self.handle_err(e); - } - - // Hand off to children. Publish `need_to_wait` first, then release - // this task's own slot on `subtask_count` with the same `fetch_sub` - // the children use: whoever takes the counter to 0 owns the rmdir. - // A load-then-store here had a lost-wakeup window where the last - // child could decrement and read `need_to_wait == false` between - // the two operations, stranding `dir_task` forever. - // - // SAFETY: `dir_task` is live until `subtask_count` drains to 0, and - // we still hold one count. Atomics through a short-lived `&DirTask` - // — interior mutability. - unsafe { - let dt = &*dir_task; - *need_to_wait_out = true; - dt.need_to_wait.store(true, Ordering::SeqCst); - if dt.subtask_count.fetch_sub(1, Ordering::SeqCst) != 1 { - // A child still holds a count. It (or a later child) will - // take the counter to 0 and drive - // `delete_after_waiting_for_children`; nothing after this - // may dereference `dir_task`. The directory fd is closed by - // the `close_fd` scopeguard on return — that touches only a - // stack local, not `dir_task`. - return Ok(()); - } - // Every child already released its count (each saw the counter - // > 1 and so did not take ownership). `dir_task` is exclusively - // ours again: restore the invariants `post_run` expects and - // fall through to delete inline. - dt.subtask_count.store(1, Ordering::SeqCst); - dt.need_to_wait.store(false, Ordering::SeqCst); - *need_to_wait_out = false; + self.tree().handle_err(e); } - if self.error_signal().load(Ordering::SeqCst) { + if self.children.is_some() { + // Children were spawned: whoever finishes last (them, or us in + // `run_owned`) deletes the directory itself. The fd is closed by + // `_close_fd` on return. + *handed_off = true; return Ok(()); } @@ -1172,14 +1157,24 @@ impl ShellRmTask { } } - match bun_sys::unlinkat_with_flags(self.cwd, path, bun_sys::AT_REMOVEDIR) { - Ok(()) => self.verbose_deleted(dir_task, path.as_bytes()), + self.remove_dir_itself() + } + + /// The directory's contents are gone and no child is outstanding: delete + /// the directory. + fn remove_dir_itself(&mut self) -> bun_sys::Maybe<()> { + if self.tree().errored() { + return Ok(()); + } + let path = self.info.path.as_zstr(); + match bun_sys::unlinkat_with_flags(self.tree().cwd, path, bun_sys::AT_REMOVEDIR) { + Ok(()) => self.verbose_deleted(path.as_bytes()), Err(e) => match e.get_errno() { E::ENOENT => { - if self.opts.force { - return self.verbose_deleted(dir_task, path.as_bytes()); + if self.tree().opts.force { + return self.verbose_deleted(path.as_bytes()); } - Err(self.error_with_path(&e, path.as_bytes())) + Err(e.with_path(path.as_bytes())) } _ => Err(e), }, @@ -1188,11 +1183,10 @@ impl ShellRmTask { /// Returns `Ok(true)` if the /// directory was deleted (or force-ignored), `Ok(false)` if a subtask was - /// enqueued and the caller should not run `post_run` yet. - fn remove_entry_dir_after_children(&self, dir_task: *mut DirTask) -> bun_sys::Maybe { - let dirfd = self.cwd; - // SAFETY: `dir_task` is live; this thread owns it. - let (path, is_abs) = unsafe { ((*dir_task).path.as_zstr(), (*dir_task).is_absolute) }; + /// spawned and this directory has to wait for it. + fn remove_entry_dir_after_children(&mut self) -> bun_sys::Maybe { + let dirfd = self.tree().cwd; + let (path, is_abs) = (self.info.path.as_zstr(), self.info.is_absolute); let mut state = RemoveFileParent { treat_as_dir: true, #[cfg(not(any(target_os = "linux", target_os = "android")))] @@ -1203,27 +1197,27 @@ impl ShellRmTask { if state.treat_as_dir { match bun_sys::rmdirat(dirfd, path) { Ok(()) => { - let _ = self.verbose_deleted(dir_task, path.as_bytes()); + let _ = self.verbose_deleted(path.as_bytes()); return Ok(true); } Err(e) => match e.get_errno() { E::ENOENT => { - if self.opts.force { - let _ = self.verbose_deleted(dir_task, path.as_bytes()); + if self.tree().opts.force { + let _ = self.verbose_deleted(path.as_bytes()); return Ok(true); } - return Err(self.error_with_path(&e, path.as_bytes())); + return Err(e.with_path(path.as_bytes())); } E::ENOTDIR => { state.treat_as_dir = false; continue; } - _ => return Err(self.error_with_path(&e, path.as_bytes())), + _ => return Err(e.with_path(path.as_bytes())), }, } } else { let mut buf = bun_paths::path_buffer_pool::get(); - self.remove_entry_file(dir_task, path, is_abs, &mut buf, &mut state)?; + self.remove_entry_file(path, is_abs, &mut buf, &mut OnDir::Parent(&mut state))?; if state.enqueued { return Ok(false); } @@ -1235,25 +1229,66 @@ impl ShellRmTask { } } - fn remove_entry_file( - &self, - parent_dir_task: *mut DirTask, + fn on_is_dir( + &mut self, + handler: &mut OnDir<'_>, + path: &ZStr, + is_absolute: bool, + buf: &mut bun_paths::PathBuffer, + ) -> bun_sys::Maybe<()> { + match handler { + OnDir::Ignore => Ok(()), + OnDir::Enqueue => { + self.enqueue_no_join(ZBox::from_bytes(path.as_bytes()), EntryKindHint::Dir); + Ok(()) + } + OnDir::Recurse { handed_off } => self.remove_entry_dir(is_absolute, buf, handed_off), + OnDir::Parent(state) => { + state.treat_as_dir = true; + Ok(()) + } + } + } + + #[cfg(not(any(target_os = "linux", target_os = "android")))] + fn on_dir_not_empty( + &mut self, + handler: &mut OnDir<'_>, path: &ZStr, is_absolute: bool, buf: &mut bun_paths::PathBuffer, - vtable: &mut V, ) -> bun_sys::Maybe<()> { - let dirfd = self.cwd; + match handler { + OnDir::Parent(state) => { + state.treat_as_dir = true; + if state.allow_enqueue { + self.enqueue_no_join(ZBox::from_bytes(path.as_bytes()), EntryKindHint::Dir); + state.enqueued = true; + } + Ok(()) + } + _ => self.on_is_dir(handler, path, is_absolute, buf), + } + } + + fn remove_entry_file( + &mut self, + path: &ZStr, + is_absolute: bool, + buf: &mut bun_paths::PathBuffer, + handler: &mut OnDir<'_>, + ) -> bun_sys::Maybe<()> { + let dirfd = self.tree().cwd; match bun_sys::unlinkat_with_flags(dirfd, path, 0) { - Ok(()) => self.verbose_deleted(parent_dir_task, path.as_bytes()), + Ok(()) => self.verbose_deleted(path.as_bytes()), Err(e) => match e.get_errno() { E::ENOENT => { - if self.opts.force { - return self.verbose_deleted(parent_dir_task, path.as_bytes()); + if self.tree().opts.force { + return self.verbose_deleted(path.as_bytes()); } - Err(self.error_with_path(&e, path.as_bytes())) + Err(e.with_path(path.as_bytes())) } - E::EISDIR => vtable.on_is_dir(parent_dir_task, path, is_absolute, buf), + E::EISDIR => self.on_is_dir(handler, path, is_absolute, buf), // This might happen if the file is actually a directory. E::EPERM => { // Non-Linux POSIX systems and Windows return EPERM when @@ -1276,32 +1311,29 @@ impl ShellRmTask { // it is deleted (if empty) or we handle it as a // directory. If it's actually a file, we get an error // so we don't need to call `stat` to check that. - if self.opts.recursive || self.opts.remove_empty_dirs { + if self.tree().opts.recursive || self.tree().opts.remove_empty_dirs { return match bun_sys::unlinkat_with_flags( - self.cwd, + self.tree().cwd, path, bun_sys::AT_REMOVEDIR, ) { // it was empty, we saved a syscall - Ok(()) => self.verbose_deleted(parent_dir_task, path.as_bytes()), + Ok(()) => self.verbose_deleted(path.as_bytes()), Err(e2) => match e2.get_errno() { // not empty, process directory as we would normally - E::ENOTEMPTY => vtable.on_dir_not_empty( - parent_dir_task, - path, - is_absolute, - buf, - ), + E::ENOTEMPTY => { + self.on_dir_not_empty(handler, path, is_absolute, buf) + } // actually a file, the error is a permissions error - E::ENOTDIR => Err(self.error_with_path(&e, path.as_bytes())), - _ => Err(self.error_with_path(&e2, path.as_bytes())), + E::ENOTDIR => Err(e.with_path(path.as_bytes())), + _ => Err(e2.with_path(path.as_bytes())), }, }; } // We don't know if it was an actual permissions error // or it was a directory so we need to try to delete it // as a directory. - return vtable.on_is_dir(parent_dir_task, path, is_absolute, buf); + return self.on_is_dir(handler, path, is_absolute, buf); } #[cfg(not(any( target_os = "macos", @@ -1314,494 +1346,10 @@ impl ShellRmTask { target_os = "illumos", windows, )))] - Err(self.error_with_path(&e, path.as_bytes())) + Err(e.with_path(path.as_bytes())) } - _ => Err(self.error_with_path(&e, path.as_bytes())), + _ => Err(e.with_path(path.as_bytes())), }, } } - - fn handle_err(&self, err: bun_sys::Error) { - let mut slot = self.err.lock(); - if slot.is_none() { - *slot = Some(err); - self.error_signal().store(true, Ordering::SeqCst); - } - } -} - -impl Drop for ShellRmTask { - fn drop(&mut self) { - // SAFETY: `root_task` was `heap::alloc`'d in `create` and is never - // freed by `DirTask::deinit` (root has `parent_task == null`). - drop(unsafe { bun_core::heap::take(self.root_task) }); - } -} - -// ── DirTask ──────────────────────────────────────────────────────────────── - -impl DirTask { - /// Recover `*mut DirTask` from the intrusive `*WorkPoolTask`. - unsafe fn work_pool_callback(task: *mut WorkPoolTask) { - // SAFETY: `work_task` is at a fixed offset within DirTask; `this` is a - // live DirTask owned by this worker thread. - unsafe { - Self::run_from_thread_pool_impl(bun_core::from_field_ptr!(DirTask, work_task, task)) - }; - } - - /// # Safety - /// `this` is a live DirTask (root or heap child); the calling worker - /// thread has exclusive access to its non-atomic fields. - unsafe fn run_from_thread_pool_impl(this: *mut DirTask) { - // Stay on the raw pointer throughout: `remove_entry` re-derives - // `&mut *this` internally (via `verbose_deleted` / `remove_entry_dir`) - // and schedules child DirTasks that concurrently touch our atomics, so - // holding a long-lived `&mut *this` across that call would alias under - // Stacked Borrows. - // - // SAFETY: caller contract — `this` is a live DirTask; this worker - // thread has exclusive access to its non-atomic fields (`task_manager` - // / `parent_task` / `path` / `is_absolute`). `tm_ptr` is live until - // `pending_main_callbacks` hits 0. - let (tm_ptr, is_absolute): (*mut ShellRmTask, bool) = unsafe { - let tm_ptr = (*this).task_manager; - let abs = Platform::AUTO.is_absolute((*this).path.as_bytes()); - (*this).is_absolute = abs; - (tm_ptr, abs) - }; - - // SAFETY: `task_manager` is live until pending_main_callbacks hits 0. - // `root_task` lives in a separate allocation, so this borrow does not - // overlap any DirTask and the field-level `&mut` taken inside - // `verbose_deleted` cannot pop its tag under Stacked Borrows. - let tm = unsafe { &*tm_ptr }; - let waiting = match tm.remove_entry(this, is_absolute) { - Ok(waiting) => waiting, - Err(err) => { - tm.handle_err(err); - // `need_to_wait` is only stored on the `Ok(())` return path - // of `remove_entry_dir`, so an `Err` return guarantees - // ownership of `this` was not handed off. - false - } - }; - - // Re-reading `this` here to decide whether to skip `post_run` would - // be a use-after-free race: once `remove_entry_dir` publishes - // `need_to_wait = true`, the last child may run - // `delete_after_waiting_for_children(this)` → `post_run` → `deinit` - // (or, for the root, the main thread may drop the whole - // `ShellRmTask`) before we get here. So the hand-off is threaded out - // as a stack-local bool via `remove_entry`'s return value and `this` - // is never touched again when set. When `waiting` is false the - // hand-off never happened, so `deleting_after_waiting_for_children` - // (only ever stored by a child that observed `need_to_wait == true`) - // is necessarily false and `post_run` is both safe and required. - if !waiting { - // SAFETY: `waiting == false` ⇒ `need_to_wait` was never published, - // so no other thread can have driven `delete_after_waiting_for_children` - // / `deinit` on `this`; it is still live and owned by this worker. - unsafe { Self::post_run(this) }; - } - } - - /// # Safety - /// `this` is a live DirTask; called from a worker thread that just - /// finished its body. - unsafe fn post_run(this: *mut DirTask) { - // SAFETY: caller contract — `this` is a live DirTask owned by the - // calling worker thread. `me` is held only over atomic / read-only - // field access; `task_manager` is a separate allocation live until - // `pending_main_callbacks` hits 0; `parent_task` is live until its - // `subtask_count` drains. `delete_after_waiting_for_children` and - // `queue_for_write` re-derive from raw `*mut` (no overlap with `me`, - // which is a `&` over atomics + Copy fields). Non-root `deinit` - // reclaims `this`'s Box; `finish_concurrently` may free the root. - unsafe { - let me = &*this; - // This is true if the directory has subdirectories that need to be deleted. - if me.need_to_wait.load(Ordering::SeqCst) { - return; - } - // We have executed all the children of this task. - if me.subtask_count.fetch_sub(1, Ordering::SeqCst) == 1 { - let tm = &*me.task_manager; - // If a verbose write will be queued, take a pending count on the - // ShellRmTask now — before decrementing the parent (children) or - // calling finish_concurrently (root) — so the main thread can't - // free it out from under write_verbose. - let will_queue_verbose = tm.opts.verbose && !me.deleted_entries.is_empty(); - if will_queue_verbose { - tm.pending_main_callbacks.fetch_add(1, Ordering::SeqCst); - } - - // The verbose hop is posted while the rm task is still counted - // work of its VM — i.e. before anything below can let the root - // `finish_concurrently` (which releases that count) run: our - // parent decrement can cascade into it, and for the root it is - // the next statement. `this` may be freed by the JS thread once - // posted, so what we still need is captured first. - let (parent_task, task_manager) = (me.parent_task, me.task_manager); - if will_queue_verbose { - Self::queue_for_write(this); - } else if !parent_task.is_null() { - Self::deinit(this); - } - - // If we have a parent and we are the last child, now we can delete the parent. - if !parent_task.is_null() { - let p = &*parent_task; - // The parent releases its own slot on this counter in - // `remove_entry_dir`; whoever takes it to 0 owns the - // parent's rmdir. The parent's `fetch_sub` is sequenced - // after its final `deleted_entries` write and its - // `need_to_wait.store(true)`, and every decrement is a - // SeqCst RMW, so reading 1 here synchronizes-with the - // parent's release and makes those writes visible to - // `delete_after_waiting_for_children`. - if p.subtask_count.fetch_sub(1, Ordering::SeqCst) == 1 { - Self::delete_after_waiting_for_children(parent_task); - } - return; - } - - // Root task: hand it back. It may be freed at any time after - // this unless the verbose hop's pending count keeps it. - ShellRmTask::finish_concurrently(task_manager); - } - } - // Otherwise need to wait. - } - - /// # Safety - /// `this` is a live DirTask; called from a worker thread. - unsafe fn delete_after_waiting_for_children(this: *mut DirTask) { - // SAFETY: caller contract. Stay on raw `this` — - // `remove_entry_dir_after_children` reborrows `(*this).deleted_entries` - // mutably (via `verbose_deleted`), so we never materialise a - // long-lived `&DirTask` covering that field. The two atomic stores - // and the `task_manager` read use raw place projection only. - // `task_manager` lives in a separate allocation, so `tm: &ShellRmTask` - // does not overlap any DirTask borrow. - unsafe { - // `run_from_thread_pool_impl` has a `defer post_run` so set this to skip that. - (*this) - .deleting_after_waiting_for_children - .store(true, Ordering::SeqCst); - (*this).need_to_wait.store(false, Ordering::SeqCst); - // The caller just took `subtask_count` to 0; restore the slot - // `post_run`'s own decrement expects. - (*this).subtask_count.store(1, Ordering::SeqCst); - let tm = &*(*this).task_manager; - let mut do_post_run = true; - if !tm.error_signal().load(Ordering::SeqCst) { - match tm.remove_entry_dir_after_children(this) { - Err(e) => tm.handle_err(e), - Ok(deleted) => { - if !deleted { - do_post_run = false; - } - } - } - } - if do_post_run { - Self::post_run(this); - } - } - } - - /// Post this DirTask to the main - /// thread for [`Rm::write_verbose`]. - /// - /// # Safety - /// `this` is a live DirTask; the pending-main-callback count on the - /// owning ShellRmTask was bumped before calling. - unsafe fn queue_for_write(this: *mut DirTask) { - use bun_event_loop::{ConcurrentTask::AutoDeinit, EventLoopTask}; - // SAFETY: caller contract — `this` is live; `task_manager` is live - // (pending count > 0). On the early-return path `deinit` reclaims a - // non-root Box and `decr_pending_and_maybe_deinit` releases the - // pending count taken in `post_run`. - let (me, poster) = unsafe { - let me = &mut *this; - if me.deleted_entries.is_empty() { - // Deinit non-root and bail. The pending count was already - // taken so release it again. Capture before the decrement — - // dropping the ShellRmTask drops the root DirTask, so for the - // root `me` may dangle immediately after. - let (tm, has_parent) = (me.task_manager, !me.parent_task.is_null()); - if has_parent { - Self::deinit(this); - } - ShellRmTask::decr_pending_and_maybe_deinit(tm); - return; - } - // The root rm task is still out (pending > 0), so its poster is set. - let poster = (*me.task_manager) - .task - .poster - .as_ref() - .expect("rm root task on the pool is armed") - .clone(); - (me, poster) - }; - match &mut me.concurrent_task { - EventLoopTask::Js(ct) => { - ct.from(this, AutoDeinit::ManualDeinit); - poster.post_js(core::ptr::NonNull::from(ct)); - } - EventLoopTask::Mini(at) => { - let at = at.from(this, dir_task_run_from_main_thread_mini); - poster.post_mini(core::ptr::NonNull::new(at).expect("intrusive task")); - } - } - } - - /// Flush verbose output. - /// - /// Reached only via `runtime::dispatch::run_task` for - /// `task_tag::ShellRmDirTask` (or the mini-loop trampoline below), which - /// always passes a live DirTask posted via [`queue_for_write`]. - /// - /// # Safety - /// `this` must be a live [`DirTask`] posted via [`queue_for_write`]; the - /// pending count keeps `task_manager` alive; main thread. - pub(crate) fn run_from_main_thread(this: *mut DirTask) { - // SAFETY: caller contract — `interp` set at create. - let (interp, cmd) = unsafe { - let tm = (*this).task_manager; - ((*tm).task.interp, (*tm).cmd) - }; - let interp = interp.get(); - Rm::write_verbose(interp, cmd, this).run(interp); - } - - /// # Safety - /// `this` is a live heap-allocated (non-root) DirTask; reclaimed once. - unsafe fn deinit(this: *mut DirTask) { - // The root task is owned by `ShellRmTask` (freed in its `Drop`); only - // non-root children are reclaimed here. - // SAFETY: caller contract. - unsafe { - debug_assert!(!(*this).parent_task.is_null()); - drop(bun_core::heap::take(this)); - } - } -} - -fn dir_task_run_from_main_thread_mini(this: *mut DirTask, _: *mut ()) { - // SAFETY: mini-loop trampoline for a DirTask posted via `queue_for_write`. - DirTask::run_from_main_thread(this); -} - -// ── RemoveFileHandler ─────────────────────────────────────────────────────── - -trait RemoveFileHandler { - fn on_is_dir( - &mut self, - parent_dir_task: *mut DirTask, - path: &ZStr, - is_absolute: bool, - buf: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()>; - #[cfg(not(any(target_os = "linux", target_os = "android")))] - fn on_dir_not_empty( - &mut self, - parent_dir_task: *mut DirTask, - path: &ZStr, - is_absolute: bool, - buf: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()>; -} - -struct DummyRemoveFile; -impl RemoveFileHandler for DummyRemoveFile { - fn on_is_dir( - &mut self, - _: *mut DirTask, - _: &ZStr, - _: bool, - _: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - Ok(()) - } - #[cfg(not(any(target_os = "linux", target_os = "android")))] - fn on_dir_not_empty( - &mut self, - _: *mut DirTask, - _: &ZStr, - _: bool, - _: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - Ok(()) - } -} - -struct RemoveFileVTable<'a> { - task: &'a ShellRmTask, - on_dir: OnDir<'a>, -} - -/// What [`RemoveFileVTable`] does when the entry it was asked to unlink turns -/// out to be a directory. -enum OnDir<'a> { - /// The entry is a child found while iterating a directory - /// ([`ShellRmTask::remove_entry_dir`]): enqueue it as its own [`DirTask`]. - Enqueue, - /// The entry is the [`DirTask`] itself ([`ShellRmTask::remove_entry`]): - /// recurse into [`ShellRmTask::remove_entry_dir`] here. The out-param lets - /// `remove_entry` learn that `need_to_wait` was published without - /// re-reading the (possibly already-freed) DirTask. - Recurse { need_to_wait_out: &'a mut bool }, -} - -impl RemoveFileVTable<'_> { - fn on_dir( - &mut self, - parent: *mut DirTask, - path: &ZStr, - is_absolute: bool, - buf: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - match &mut self.on_dir { - OnDir::Enqueue => { - self.task.enqueue_no_join( - parent, - ZBox::from_bytes(path.as_bytes()), - EntryKindHint::Dir, - ); - Ok(()) - } - OnDir::Recurse { need_to_wait_out } => { - self.task - .remove_entry_dir(parent, is_absolute, buf, need_to_wait_out) - } - } - } -} - -impl RemoveFileHandler for RemoveFileVTable<'_> { - fn on_is_dir( - &mut self, - parent: *mut DirTask, - path: &ZStr, - is_absolute: bool, - buf: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - self.on_dir(parent, path, is_absolute, buf) - } - #[cfg(not(any(target_os = "linux", target_os = "android")))] - fn on_dir_not_empty( - &mut self, - parent: *mut DirTask, - path: &ZStr, - is_absolute: bool, - buf: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - self.on_dir(parent, path, is_absolute, buf) - } -} - -struct RemoveFileParent { - treat_as_dir: bool, - #[cfg(not(any(target_os = "linux", target_os = "android")))] - allow_enqueue: bool, - enqueued: bool, -} -impl RemoveFileHandler for RemoveFileParent { - fn on_is_dir( - &mut self, - _: *mut DirTask, - _: &ZStr, - _: bool, - _: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - self.treat_as_dir = true; - Ok(()) - } - #[cfg(not(any(target_os = "linux", target_os = "android")))] - fn on_dir_not_empty( - &mut self, - parent: *mut DirTask, - path: &ZStr, - _: bool, - _: &mut bun_paths::PathBuffer, - ) -> bun_sys::Maybe<()> { - self.treat_as_dir = true; - if self.allow_enqueue { - // SAFETY: `parent` is a live DirTask; `task_manager` is live until - // `pending_main_callbacks` hits 0. - let task = unsafe { &*(*parent).task_manager }; - task.enqueue_no_join( - parent, - ZBox::from_bytes(path.as_bytes()), - EntryKindHint::Dir, - ); - self.enqueued = true; - } - Ok(()) - } -} - -// ── Taskable / ShellTaskCtx glue ────────────────────────────────────────── - -impl bun_event_loop::Taskable for ShellRmTask { - const TAG: bun_event_loop::TaskTag = bun_event_loop::task_tag::ShellRmTask; - /// The rm's own completion hop: drop the keep-alive and this pending - /// callback's count (frees the task when it was the last). - unsafe fn release_unrun(this: *mut Self) { - // SAFETY: fn contract. - unsafe { - (*this).task.unref_unrun(); - ShellRmTask::decr_pending_and_maybe_deinit(this); - } - } -} -impl bun_event_loop::Taskable for DirTask { - const TAG: bun_event_loop::TaskTag = bun_event_loop::task_tag::ShellRmDirTask; - /// A verbose-output hop: free the (non-root) dir task and give back the - /// pending-callback unit it holds on its rm — as `run_from_main_thread` - /// does when there is nothing to write. - unsafe fn release_unrun(this: *mut Self) { - // SAFETY: fn contract; capture before the decrement (it may free the root). - unsafe { - let (tm, has_parent) = ((*this).task_manager, !(*this).parent_task.is_null()); - if has_parent { - Self::deinit(this); - } - ShellRmTask::decr_pending_and_maybe_deinit(tm); - } - } -} - -/// Mini-loop trampoline for [`ShellRmTask::finish_concurrently`]. -fn shell_rm_task_run_from_main_thread_mini(this: *mut ShellRmTask, _: *mut ()) { - // SAFETY: `this` is the live heap task `finish_concurrently` posted; the - // mini loop fires it once on the main thread. - ShellTask::run_from_main_thread::(unsafe { bun_core::heap::take(this) }); -} - -impl crate::shell::interpreter::ShellTaskCtx for ShellRmTask { - fn shell_task(&self) -> &ShellTask { - &self.task - } - fn shell_task_mut(&mut self) -> &mut ShellTask { - &mut self.task - } - fn run_from_thread_pool(&mut self) { - // Not reached: `ShellRmTask::schedule` installs `work_pool_callback` - // directly (the generic trampoline auto-posts back, which would race - // the recursive DirTask tree's own `finish_concurrently`). - debug_assert!( - false, - "ShellRmTask scheduled via ShellTask::schedule; use ShellRmTask::schedule" - ); - } - /// Released back to the raw pending-callback protocol - /// (`decr_pending_and_maybe_deinit` frees it). - fn run_from_main_thread(self: Box, interp: &Interpreter) { - let cmd = self.cmd; - Rm::on_shell_rm_task_done(interp, cmd, bun_core::heap::into_raw(self)); - } } diff --git a/src/runtime/shell/builtin/which.rs b/src/runtime/shell/builtin/which.rs index 2bd050eb6b2d..b8a7d2343d96 100644 --- a/src/runtime/shell/builtin/which.rs +++ b/src/runtime/shell/builtin/which.rs @@ -5,7 +5,6 @@ //! are processed. use crate::shell::builtin::{Builtin, BuiltinState, IoKind, Kind}; -use crate::shell::env_str::EnvStr; use crate::shell::interpreter::{Interpreter, NodeId}; use crate::shell::io_writer::{ChildPtr, WriterTag}; use crate::shell::yield_::Yield; @@ -207,15 +206,10 @@ impl SearchEnv { fn load(interp: &Interpreter, cmd: NodeId) -> Self { let shell = Builtin::shell(interp, cmd); let shell = shell.borrow(); - // `EnvMap::get` refs the returned string; balance it. let path_env = shell .export_env - .get(EnvStr::init_slice(b"PATH")) - .map(|s| { - let v = s.slice().to_vec(); - s.deref(); - v - }) + .get(b"PATH") + .map(|s| s.slice().to_vec()) .unwrap_or_default(); Self { path_env, diff --git a/src/runtime/shell/builtin/yes.rs b/src/runtime/shell/builtin/yes.rs index 651c0d885aab..4c5cc5ab6fb6 100644 --- a/src/runtime/shell/builtin/yes.rs +++ b/src/runtime/shell/builtin/yes.rs @@ -5,7 +5,7 @@ use crate::shell::io_writer::{ChildPtr, WriterTag}; use crate::shell::states::cmd::Exec; use crate::shell::yield_::Yield; -use bun_event_loop::{EventLoopTask, TaskTag, Taskable, task_tag}; +use bun_event_loop::AnyTaskWithExtraContext::{AnyTaskWithExtraContext, BoxedMiniTaskRunner}; #[derive(Clone, Copy, PartialEq, Eq, Default)] pub enum State { @@ -23,8 +23,9 @@ pub struct Yes { /// out to ~BUFSIZ. pub(crate) buffer: Vec, pub(crate) buffer_used: usize, - /// Populated in `start()`. - pub task: Option, + /// The bounce payload: created in `start()`, out on the event loop between + /// `write_no_io_loop` and `YesTask::run`. + pub task: Option>, } impl Yes { @@ -68,12 +69,12 @@ impl Yes { let mut me = Self::state_mut(interp, cmd); me.buffer = buf; me.buffer_used = filled; - me.task = Some(YesTask { + me.task = Some(Box::new(YesTask { interp: bun_ptr::ParentRef::new(interp), cmd, evtloop, - concurrent_task: EventLoopTask::from_event_loop(evtloop), - }); + concurrent_task: AnyTaskWithExtraContext::default(), + })); } let stdout_needs_io = Builtin::of(interp, cmd).stdout.needs_io(); @@ -118,16 +119,13 @@ impl Yes { return Self::write_failing_error(interp, cmd, &buf, 1); } // Bounce back via the event loop so we don't block the main thread. - let task: *mut YesTask = Self::state_mut(interp, cmd) + // `enqueue` ticks the event loop and may re-enter shell dispatch — no + // borrow of `interp` state is held across it. + let task = Self::state_mut(interp, cmd) .task - .as_mut() + .take() .expect("YesTask set in start()"); - // SAFETY: `task` was set in `start()`; `Yes` lives in a `Box` inside - // the interpreter arena, so the address is stable across the enqueue - // and the later main-thread callback. `enqueue` ticks the event loop - // and may re-enter shell dispatch — we hold no `&mut` derived from - // `interp` across the call. - unsafe { YesTask::enqueue(task) }; + YesTask::enqueue(task); Yield::suspended() } @@ -170,8 +168,8 @@ impl Yes { Self::enqueue_chunk(interp, cmd, OutputNeedsIOSafeGuard::OutputNeedsIo) } - /// Split-borrow `&mut Builtin` into `(&mut stdout, &mut Yes)`; the fields - /// are disjoint so this is a sound reborrow without `unsafe`. + /// Split-borrow `&mut Builtin` into `(&mut stdout, &mut Yes)` (disjoint + /// fields). #[inline] fn split_stdout_state(me: &mut Builtin) -> (&mut BuiltinIO, &mut Yes) { let Impl::Yes(yes) = &mut me.impl_ else { @@ -186,68 +184,36 @@ impl Yes { /// Re-queues `yes` onto the event loop after a burst of no-IO writes so we /// don't block the main thread forever. -#[repr(C)] pub struct YesTask { /// Back-ref to the owning [`Interpreter`]. pub(crate) interp: bun_ptr::ParentRef, pub(crate) cmd: NodeId, pub(crate) evtloop: EventLoopHandle, - pub(crate) concurrent_task: EventLoopTask, + /// Intrusive node for the mini-loop post (the JS loop queues a `Task`). + pub(crate) concurrent_task: AnyTaskWithExtraContext, } -impl Taskable for YesTask { - const TAG: TaskTag = task_tag::ShellYesTask; - /// Lives inside the builtin's `Box` (freed with the interpreter) and - /// took nothing for the bounce; nothing to do. - unsafe fn release_unrun(_: *mut Self) {} -} +// `runtime::dispatch::run_task`'s `task_tag::ShellYesTask` arm reboxes the +// enqueued pointer as `YesTask`; both sides MUST agree. +bun_event_loop::boxed_taskable!(YesTask, ShellYesTask); -impl YesTask { - /// # Safety - /// `this` must point to a live `YesTask` whose storage is stable until the - /// enqueued task fires (it lives inside `Box` in the interpreter - /// arena). - unsafe fn enqueue(this: *mut Self) { - // SAFETY: caller contract — `this` is live and stable; `evtloop` / - // `concurrent_task` were initialised together by `Yes::start` so the - // Js/Mini discriminants agree. `owner`/`mini` are live event-loop - // backrefs (single-threaded shell). - unsafe { - match (*this).evtloop { - // Next loop iteration, after I/O has had a turn. - EventLoopHandle::Js { owner } => { - owner.enqueue_task_after_yield(bun_jsc::Task::init(this)); - } - EventLoopHandle::Mini(mut mini) => { - (*mini.loop_).tick(); - let at = - core::ptr::NonNull::new_unchecked(match &mut (*this).concurrent_task { - EventLoopTask::Mini(at) => { - at.from(this, Self::run_from_main_thread_mini) - } - EventLoopTask::Js(_) => unreachable!(), - }); - mini.get_mut().enqueue_task_concurrent(at); - } - } - } +impl BoxedMiniTaskRunner for YesTask { + fn run_from_loop_thread(owner: Box) { + owner.run(); } +} - /// `this` must be a live `YesTask` whose storage is stable inside - /// `Box` in the interpreter arena, with `interp` initialised by - /// [`Yes::start`]. Reached only via the concurrent-task dispatch installed - /// in [`enqueue`](Self::enqueue). - pub(crate) fn run_from_main_thread(this: &Self) { - let (interp, cmd) = (&*this.interp, this.cmd); - Yes::write_no_io_loop(interp, cmd).run(interp); +impl YesTask { + /// Next loop iteration, after I/O has had a turn. + fn enqueue(self: Box) { + let evtloop = self.evtloop; + evtloop.enqueue_boxed_after_yield::(self, |t| &mut t.concurrent_task); } - /// Signature matches - /// [`AnyTaskWithExtraContext::from`](bun_event_loop::AnyTaskWithExtraContext::AnyTaskWithExtraContext::from)'s - /// callback shape (`fn(*mut T, *mut ())`). - fn run_from_main_thread_mini(this: *mut Self, _: *mut ()) { - // SAFETY: dispatch contract — `this` is the live task previously passed - // to `enqueue`; see `run_from_main_thread`. - Self::run_from_main_thread(unsafe { &*this }) + /// Main thread: park the payload back on its `Yes` and write the next burst. + pub(crate) fn run(self: Box) { + let (interp, cmd) = (self.interp, self.cmd); + Yes::state_mut(&interp, cmd).task = Some(self); + Yes::write_no_io_loop(&interp, cmd).run(&interp); } } diff --git a/src/runtime/shell/dispatch_tasks.rs b/src/runtime/shell/dispatch_tasks.rs index a7eae6f35d96..11757125b279 100644 --- a/src/runtime/shell/dispatch_tasks.rs +++ b/src/runtime/shell/dispatch_tasks.rs @@ -154,9 +154,6 @@ impl ShellGlobTask { } } -/// A child node in the -/// recursive rm tree-walk; posts back to main when its subtree is empty. -/// Re-export: the real DirTask lives in `builtins::rm` (full recursive -/// tree-walk node). `dispatch.rs` calls `ShellRmDirTask::run_from_main_thread` -/// for the verbose-write bounce-back. +/// A node in the recursive rm tree-walk (`builtins::rm`); `dispatch.rs` +/// reboxes it for the verbose-write bounce-back. pub(crate) use crate::shell::builtins::rm::DirTask as ShellRmDirTask; diff --git a/src/runtime/shell/interpreter.rs b/src/runtime/shell/interpreter.rs index 7e3b1661fd26..6dd83902f016 100644 --- a/src/runtime/shell/interpreter.rs +++ b/src/runtime/shell/interpreter.rs @@ -1848,16 +1848,15 @@ impl ShellExecEnv { // Only `OLDPWD` is gated on `!in_init`; // `PWD` is written unconditionally so the very first env (built during // `init()` with `in_init = true`) still exports the resolved cwd. - // Note: reshaped for borrowck — materialize the EnvStr (which - // erases the slice lifetime into a packed ptr) before taking - // `&mut self.export_env`. + // The cwd buffers are rewritten by the next `cd`, so the env holds + // copies. use crate::shell::env_str::EnvStr; if !in_init { - let oldpwd = EnvStr::init_slice(self.prev_cwd()); + let oldpwd = EnvStr::dupe_ref_counted(self.prev_cwd()); self.export_env .insert(EnvStr::init_slice(b"OLDPWD"), oldpwd); } - let pwd = EnvStr::init_slice(self.cwd()); + let pwd = EnvStr::dupe_ref_counted(self.cwd()); self.export_env.insert(EnvStr::init_slice(b"PWD"), pwd); Ok(()) @@ -1865,8 +1864,6 @@ impl ShellExecEnv { /// Routes `label = value` into an env map depending on where the /// assignment appeared (`FOO=1 cmd` → cmd-local, bare `FOO=1` → shell). - /// NOTE: `EnvMap::insert` `.ref()`s the value, so callers should deref - /// `value` after the call. pub(crate) fn assign_var( &mut self, label: crate::shell::env_str::EnvStr, @@ -1884,14 +1881,15 @@ impl ShellExecEnv { /// `cd` with no args / `~` expansion never sees a null. pub(crate) fn get_homedir(&self) -> crate::shell::env_str::EnvStr { use crate::shell::env_str::EnvStr; - let key = if cfg!(windows) { - EnvStr::init_slice(b"USERPROFILE") + let key: &[u8] = if cfg!(windows) { + b"USERPROFILE" } else { - EnvStr::init_slice(b"HOME") + b"HOME" }; self.shell_env .get(key) .or_else(|| self.export_env.get(key)) + .cloned() .unwrap_or_else(|| { EnvStr::init_slice(if bun_core::env::IS_ANDROID { b"/data/local/tmp" diff --git a/src/runtime/shell/mod.rs b/src/runtime/shell/mod.rs index 32fc0eff8991..d5c299e4526b 100644 --- a/src/runtime/shell/mod.rs +++ b/src/runtime/shell/mod.rs @@ -25,8 +25,6 @@ pub use shell_body as shell; pub mod env_map; #[path = "EnvStr.rs"] pub mod env_str; -#[path = "RefCountedStr.rs"] -pub mod ref_counted_str; #[path = "util.rs"] pub mod util; diff --git a/src/runtime/shell/states/Assigns.rs b/src/runtime/shell/states/Assigns.rs index 040e8659953b..92f3b9e13cab 100644 --- a/src/runtime/shell/states/Assigns.rs +++ b/src/runtime/shell/states/Assigns.rs @@ -132,13 +132,12 @@ impl Assigns { merged }; - let value_ref = EnvStr::init_ref_counted(value.into_boxed_slice()); + let value_ref = EnvStr::init_ref_counted(value); interp.as_assigns(this).base.shell_mut().assign_var( EnvStr::init_slice(label), value_ref, ctx, ); - value_ref.deref(); Yield::Next(this) } diff --git a/src/runtime/shell/states/Expansion.rs b/src/runtime/shell/states/Expansion.rs index d0199a2f7ef4..a3aa07384b3e 100644 --- a/src/runtime/shell/states/Expansion.rs +++ b/src/runtime/shell/states/Expansion.rs @@ -236,7 +236,6 @@ impl Expansion { *off += prepended; } } - home.deref(); } // Brace expansion // first, then glob, else flush as a single word. @@ -459,7 +458,6 @@ impl Expansion { has_quoted_empty: &mut bool, expand_tilde: bool, ) -> bool { - use crate::shell::env_str::EnvStr; match atom { ast::SimpleAtom::Text(txt) => out.extend_from_slice(txt), ast::SimpleAtom::QuotedEmpty => { @@ -471,13 +469,13 @@ impl Expansion { } ast::SimpleAtom::Var(label) => { // Spec `expandVar`: shell_env first, then export_env, else "". - let key = EnvStr::init_slice(label); - if let Some(v) = shell.shell_env.get(key) { + let label: &[u8] = label; + if let Some(v) = shell + .shell_env + .get(label) + .or_else(|| shell.export_env.get(label)) + { out.extend_from_slice(v.slice()); - v.deref(); - } else if let Some(v) = shell.export_env.get(EnvStr::init_slice(label)) { - out.extend_from_slice(v.slice()); - v.deref(); } } ast::SimpleAtom::VarArgv(int) => interp.append_var_argv(out, *int), @@ -508,7 +506,6 @@ impl Expansion { if expand_tilde { let home = shell.get_homedir(); out.extend_from_slice(home.slice()); - home.deref(); } else { out.push(b'~'); } diff --git a/test/js/bun/shell/commands/rm.test.ts b/test/js/bun/shell/commands/rm.test.ts index c588b0d414a9..1cc5c5703e09 100644 --- a/test/js/bun/shell/commands/rm.test.ts +++ b/test/js/bun/shell/commands/rm.test.ts @@ -520,13 +520,9 @@ test("rm errors for several operands into a closed pipe finish", async () => { env: bunEnv, cwd: String(dir), stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect({ stdout, stderr, exitCode, signalCode: proc.signalCode }).toEqual({ - stdout: "0\n", - stderr: "", - exitCode: 0, - signalCode: null, + stderr: "inherit", }); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + expect(stdout).toBe("0\n"); + expect(exitCode).toBe(0); }); From 93a4ba046087a3a929fc34b3bf29d723c2956496 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Thu, 27 Aug 2026 09:48:33 +0000 Subject: [PATCH 2/2] test(rm): pipe stderr and assert the closed-pipe result as one object --- test/js/bun/shell/commands/rm.test.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/test/js/bun/shell/commands/rm.test.ts b/test/js/bun/shell/commands/rm.test.ts index 1cc5c5703e09..c588b0d414a9 100644 --- a/test/js/bun/shell/commands/rm.test.ts +++ b/test/js/bun/shell/commands/rm.test.ts @@ -520,9 +520,13 @@ test("rm errors for several operands into a closed pipe finish", async () => { env: bunEnv, cwd: String(dir), stdout: "pipe", - stderr: "inherit", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout, stderr, exitCode, signalCode: proc.signalCode }).toEqual({ + stdout: "0\n", + stderr: "", + exitCode: 0, + signalCode: null, }); - const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); - expect(stdout).toBe("0\n"); - expect(exitCode).toBe(0); });