diff --git a/src/install/isolated_install.rs b/src/install/isolated_install.rs index 98772c911a9..ac8561c5d19 100644 --- a/src/install/isolated_install.rs +++ b/src/install/isolated_install.rs @@ -37,6 +37,7 @@ use bun_paths::path_options::AssumeOk as _; use bun_paths::{self as paths, AutoAbsPath as AbsPath, AutoRelPath, PathBuffer}; use bun_semver as semver; use bun_sys::{self as sys, Fd}; +use bun_threading::WaitGroup; use bun_wyhash::{Wyhash, Wyhash11}; use crate::analytics; @@ -2008,6 +2009,8 @@ pub(crate) fn install_isolated_packages( // TODO: delete let mut seen_workspace_ids: HashMap = HashMap::default(); + let tasks_in_flight = WaitGroup::init(); + // `installer::Task` carries `result: Result` (Drop via `TaskError` // payloads) and a non-nullable fn-ptr in `thread_pool::Task`, so // `assume_init()` on uninit memory is instant UB and a subsequent @@ -2026,10 +2029,10 @@ pub(crate) fn install_isolated_packages( installer: bun_ptr::BackRef::from(core::ptr::NonNull::dangling()), result: installer::Result::None, relink: installer::Relink::Off, - task: bun_threading::thread_pool::Task { - callback: installer::Task::callback, - node: Default::default(), - }, + task: bun_threading::thread_pool::CountedTask::new( + installer::Task::callback, + &tasks_in_flight, + ), next: bun_threading::Link::new(), }); } @@ -2039,6 +2042,7 @@ pub(crate) fn install_isolated_packages( let show_progress = manager.options.log_level.show_progress(); let installed = DynamicBitSet::init_empty(lockfile.packages.len())?; + let released = DynamicBitSet::init_empty(store.entries.len())?; let trusted_dependencies_from_update_requests = manager.find_trusted_dependencies_from_update_requests(); // `Installer.manager` is a BACKREF raw pointer; copying `manager_ptr` @@ -2057,6 +2061,8 @@ pub(crate) fn install_isolated_packages( }, store: &store, tasks, + tasks_in_flight: &tasks_in_flight, + released, waiters_head: vec![store::entry::Id::INVALID; store.entries.len()].into_boxed_slice(), next_waiter: vec![store::entry::Id::INVALID; store.entries.len()].into_boxed_slice(), trusted_dependencies_mutex: Default::default(), diff --git a/src/install/isolated_install/Installer.rs b/src/install/isolated_install/Installer.rs index ce346a116b3..1f661382673 100644 --- a/src/install/isolated_install/Installer.rs +++ b/src/install/isolated_install/Installer.rs @@ -7,7 +7,7 @@ use bun_core::{Environment, Global, Output}; use bun_core::{ZStr, strings}; use bun_paths::{self as paths, AbsPath, AutoAbsPath, AutoRelPath}; use bun_sys::{self as sys, Fd}; -use bun_threading::{Mutex, UnboundedQueue, thread_pool}; +use bun_threading::{Mutex, UnboundedQueue, WaitGroup, thread_pool}; use bun_semver::String as SemverString; use bun_sys::{FdDirExt as _, FdExt as _}; @@ -91,6 +91,13 @@ pub struct Installer<'a> { pub(crate) task_queue: UnboundedQueue, // intrusive via .next pub(crate) tasks: Box<[Task]>, + /// Tasks currently on the thread pool: `start_task` adds one, the + /// `CountedTask` in `Task::task` finishes it, and `Drop` waits for it. + pub(crate) tasks_in_flight: &'a WaitGroup, + + /// Main thread only: entries whose pending-task slot has been released. + pub(crate) released: Bitset, + /// Stable Rust has no /// generic atomic-enum, so store the `#[repr(u8)]` discriminant and /// round-trip via `Method::from_u8` at the load sites below. @@ -119,6 +126,14 @@ pub struct Installer<'a> { pub(crate) next_waiter: Box<[StoreEntryId]>, } +impl Drop for Installer<'_> { + /// Tasks on the pool read `self` and the store through `Task::installer`, so + /// neither may be freed until they have returned. + fn drop(&mut self) { + self.tasks_in_flight.wait(); + } +} + impl<'a> Installer<'a> { // BACKREF accessors — `manager` points outside `Self`; see field doc. #[inline] @@ -166,10 +181,36 @@ impl<'a> Installer<'a> { | Result::RunScripts(_) )); + if Environment::CI_ASSERT { + assert!( + !self.released.is_set(entry_id.get() as usize), + "isolated install: started entry {} after its pending task was released", + entry_id.get() + ); + } + task.result = Result::None; + self.tasks_in_flight.add_one(); + let task: *mut Task = task; + // SAFETY: raw projection through the whole `tasks[entry_id]` slot, so the + // pointer `Task::callback` gets back keeps provenance over the whole `Task`. + let pool_task = unsafe { &raw mut (*task).task.task }; manager .thread_pool - .schedule(thread_pool::Batch::from(&raw mut task.task)); + .schedule(thread_pool::Batch::from(pool_task)); + } + + /// Main thread only. Every entry releases its pending-task slot exactly once. + fn release_pending_task(&mut self, entry_id: StoreEntryId) { + if Environment::CI_ASSERT { + assert!( + !self.released.is_set(entry_id.get() as usize), + "isolated install: released the pending task of entry {} twice", + entry_id.get() + ); + } + self.released.set(entry_id.get() as usize); + self.manager_mut().decrement_pending_tasks(); } /// Called from main thread, for an existing project-local store entry. @@ -434,14 +475,10 @@ impl<'a> Installer<'a> { self.summary.fail += 1; - self.decrement_pending_tasks(); + self.release_pending_task(entry_id); self.resume_unblocked_tasks(entry_id); } - pub(crate) fn decrement_pending_tasks(&mut self) { - self.manager_mut().decrement_pending_tasks(); - } - /// Called from main thread pub(crate) fn on_task_blocked(&mut self, entry_id: StoreEntryId) { // race condition (fixed now): task decides it is blocked because one of its dependencies @@ -516,7 +553,7 @@ impl<'a> Installer<'a> { ); } - self.decrement_pending_tasks(); + self.release_pending_task(entry_id); self.resume_unblocked_tasks(entry_id); if let Some(node) = self.install_node.as_mut() { @@ -651,7 +688,8 @@ pub struct Task { /// any `start_task` call — see `isolated_install.rs`. pub(crate) installer: bun_ptr::BackRef>, - pub(crate) task: thread_pool::Task, + /// Counted on `Installer::tasks_in_flight`; runs `Task::callback`. + pub(crate) task: thread_pool::CountedTask, pub(crate) next: bun_threading::Link, // INTRUSIVE: bun.UnboundedQueue(Task, .next) link pub(crate) result: Result, @@ -1947,15 +1985,17 @@ impl Task { Err(_oom) => bun_core::out_of_memory(), }; - // SAFETY: installer outlives all tasks (BACKREF). `callback` runs on the - // thread pool concurrently across many `Task`s sharing the same - // `*mut Installer` — never materialize `&mut Installer`. `task_queue.push` - // takes `&self` (lock-free); `store.entries` columns are atomic / per-entry; - // both are reached through a shared `&Installer`. `manager.wake()` is the - // cross-thread wakeup: route through `PackageManager::wake_raw` which never - // forms `&mut PackageManager`, so two threads finishing simultaneously do - // not hold aliased exclusive borrows ("deref it fresh per call" alone - // would not prevent the `&mut` lifetimes from overlapping). + // SAFETY: the installer is alive until this fn returns: `Task::task` counts + // on `Installer::tasks_in_flight` until then, and `Installer::drop` waits + // for that count. `callback` runs on the thread pool concurrently across + // many `Task`s sharing the same `*mut Installer` — never materialize + // `&mut Installer`. `task_queue.push` takes `&self` (lock-free); + // `store.entries` columns are atomic / per-entry; both are reached through + // a shared `&Installer`. `manager.wake()` is the cross-thread wakeup: route + // through `PackageManager::wake_raw` which never forms + // `&mut PackageManager`, so two threads finishing simultaneously do not + // hold aliased exclusive borrows ("deref it fresh per call" alone would + // not prevent the `&mut` lifetimes from overlapping). let installer_ptr = this.installer; let installer = installer_ptr.get(); let manager_ptr: *mut PackageManager = installer.manager; diff --git a/src/install/isolated_install/Store.rs b/src/install/isolated_install/Store.rs index b1a3bef316a..3f39db1fced 100644 --- a/src/install/isolated_install/Store.rs +++ b/src/install/isolated_install/Store.rs @@ -99,7 +99,7 @@ impl Drop for Store { use entry::EntryColumns as _; for slot in self.entries.items_scripts() { if let Some(list) = slot.take() { - // SAFETY: the installer returned only after every task finished, so nothing else references the list boxed in Installer's RunPreinstall step. + // SAFETY: `Installer::drop` waited for every task on the pool, so nothing else references the list boxed in Installer's RunPreinstall step. unsafe { bun_core::heap::destroy(list) }; } }