From c1931861f3617a93be7153ef80e137e547c65298 Mon Sep 17 00:00:00 2001 From: Daria Sukhonina Date: Wed, 29 Apr 2026 15:18:21 +0300 Subject: [PATCH 1/6] Use thread index instead of condvar for query waiters --- Cargo.lock | 1 + compiler/rustc_interface/src/interface.rs | 8 +--- compiler/rustc_interface/src/passes.rs | 1 - compiler/rustc_interface/src/util.rs | 22 +++++------ compiler/rustc_middle/src/query/job.rs | 44 ++++++++-------------- compiler/rustc_middle/src/ty/context.rs | 6 --- compiler/rustc_query_impl/src/execution.rs | 2 +- compiler/rustc_query_impl/src/job.rs | 7 +--- compiler/rustc_thread_pool/Cargo.toml | 1 + compiler/rustc_thread_pool/src/lib.rs | 2 +- compiler/rustc_thread_pool/src/registry.rs | 20 +++++++--- 11 files changed, 47 insertions(+), 67 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9c6d6c22de4dd..54185740ebf46 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4688,6 +4688,7 @@ dependencies = [ "crossbeam-deque", "crossbeam-utils", "libc", + "parking_lot", "rand 0.9.2", "rand_xorshift", "scoped-tls", diff --git a/compiler/rustc_interface/src/interface.rs b/compiler/rustc_interface/src/interface.rs index 875ed4ae5d307..826d59e5ca228 100644 --- a/compiler/rustc_interface/src/interface.rs +++ b/compiler/rustc_interface/src/interface.rs @@ -5,7 +5,7 @@ use std::sync::Arc; use rustc_ast::{LitKind, MetaItemKind, token}; use rustc_codegen_ssa::traits::CodegenBackend; use rustc_data_structures::fx::{FxHashMap, FxHashSet}; -use rustc_data_structures::jobserver::{self, Proxy}; +use rustc_data_structures::jobserver; use rustc_errors::{DiagCtxtHandle, ErrorGuaranteed}; use rustc_lint::LintStore; use rustc_middle::ty; @@ -41,9 +41,6 @@ pub struct Compiler { /// A reference to the current `GlobalCtxt` which we pass on to `GlobalCtxt`. pub(crate) current_gcx: CurrentGcx, - - /// A jobserver reference which we pass on to `GlobalCtxt`. - pub(crate) jobserver_proxy: Arc, } /// Converts strings provided as `--cfg [cfgspec]` into a `Cfg`. @@ -412,7 +409,7 @@ pub fn run_compiler(config: Config, f: impl FnOnce(&Compiler) -> R + Se config.opts.unstable_opts.threads.unwrap_or(1), &config.extra_symbols, SourceMapInputs { file_loader, path_mapping, hash_kind, checksum_hash_kind }, - |current_gcx, jobserver_proxy| { + |current_gcx| { // The previous `early_dcx` can't be reused here because it doesn't // impl `Send`. Creating a new one is fine. let early_dcx = EarlyDiagCtxt::new(config.opts.error_format); @@ -484,7 +481,6 @@ pub fn run_compiler(config: Config, f: impl FnOnce(&Compiler) -> R + Se codegen_backend, override_queries: config.override_queries, current_gcx, - jobserver_proxy, }; // There are two paths out of `f`. diff --git a/compiler/rustc_interface/src/passes.rs b/compiler/rustc_interface/src/passes.rs index bcd1a52ce9dcd..0798518fb1610 100644 --- a/compiler/rustc_interface/src/passes.rs +++ b/compiler/rustc_interface/src/passes.rs @@ -1001,7 +1001,6 @@ pub fn create_and_enter_global_ctxt FnOnce(TyCtxt<'tcx>) -> T>( ), providers.hooks, compiler.current_gcx.clone(), - Arc::clone(&compiler.jobserver_proxy), |tcx| { let feed = tcx.create_crate_num(stable_crate_id).unwrap(); assert_eq!(feed.key(), LOCAL_CRATE); diff --git a/compiler/rustc_interface/src/util.rs b/compiler/rustc_interface/src/util.rs index d7d306918fd0d..157ba89abbfee 100644 --- a/compiler/rustc_interface/src/util.rs +++ b/compiler/rustc_interface/src/util.rs @@ -132,7 +132,7 @@ fn init_stack_size(early_dcx: &EarlyDiagCtxt) -> usize { }) } -fn run_in_thread_with_globals) -> R + Send, R: Send>( +fn run_in_thread_with_globals R + Send, R: Send>( thread_stack_size: usize, edition: Edition, sm_inputs: SourceMapInputs, @@ -158,7 +158,7 @@ fn run_in_thread_with_globals) -> R + Send, R: edition, extra_symbols, Some(sm_inputs), - || f(CurrentGcx::new(), Proxy::new()), + || f(CurrentGcx::new()), ) }) .unwrap() @@ -171,10 +171,7 @@ fn run_in_thread_with_globals) -> R + Send, R: }) } -pub(crate) fn run_in_thread_pool_with_globals< - F: FnOnce(CurrentGcx, Arc) -> R + Send, - R: Send, ->( +pub(crate) fn run_in_thread_pool_with_globals R + Send, R: Send>( thread_builder_diag: &EarlyDiagCtxt, edition: Edition, threads: usize, @@ -198,11 +195,11 @@ pub(crate) fn run_in_thread_pool_with_globals< edition, sm_inputs, extra_symbols, - |current_gcx, jobserver_proxy| { + |current_gcx| { // Register the thread for use with the `WorkerLocal` type. registry.register(); - f(current_gcx, jobserver_proxy) + f(current_gcx) }, ); }; @@ -211,13 +208,12 @@ pub(crate) fn run_in_thread_pool_with_globals< let current_gcx2 = current_gcx.clone(); let proxy = Proxy::new(); - let proxy_ = Arc::clone(&proxy); - let proxy__ = Arc::clone(&proxy); + let builder = rustc_thread_pool::ThreadPoolBuilder::new() .thread_name(|_| "rustc".to_string()) - .acquire_thread_handler(move || proxy_.acquire_thread()) - .release_thread_handler(move || proxy__.release_thread()) + .acquire_thread_handler(move || proxy.acquire_thread()) + .release_thread_handler(move || proxy_.release_thread()) .num_threads(threads) .deadlock_handler(move || { // On deadlock, creates a new thread and forwards information in thread @@ -293,7 +289,7 @@ internal compiler error: query cycle handler thread panicked, aborting process"; }, // Run `f` on the first thread in the thread pool. move |pool: &rustc_thread_pool::ThreadPool| { - pool.install(|| f(current_gcx.into_inner(), proxy)) + pool.install(|| f(current_gcx.into_inner())) }, ) .unwrap_or_else(|err| { diff --git a/compiler/rustc_middle/src/query/job.rs b/compiler/rustc_middle/src/query/job.rs index 8c78bf24287e0..33e04134837b5 100644 --- a/compiler/rustc_middle/src/query/job.rs +++ b/compiler/rustc_middle/src/query/job.rs @@ -3,11 +3,10 @@ use std::hash::Hash; use std::num::NonZero; use std::sync::Arc; -use parking_lot::{Condvar, Mutex}; +use parking_lot::Mutex; use rustc_span::Span; use crate::query::Cycle; -use crate::ty::TyCtxt; /// A value uniquely identifying an active query job. #[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)] @@ -54,15 +53,16 @@ impl<'tcx> QueryJob<'tcx> { #[derive(Debug)] pub struct QueryWaiter<'tcx> { pub parent: Option, - pub condvar: Condvar, + // FIXME: could be made u16 due to rustc_thread_pool limiting number of threads + pub thread_index: usize, pub span: Span, - pub cycle: Mutex>>, + pub cycle: Arc>>>, } #[derive(Clone, Debug)] pub struct QueryLatch<'tcx> { /// The `Option` is `Some(..)` when the job is active, and `None` once completed. - pub waiters: Arc>>>>>, + pub waiters: Arc>>>>, } impl<'tcx> QueryLatch<'tcx> { @@ -71,46 +71,33 @@ impl<'tcx> QueryLatch<'tcx> { } /// Awaits for the query job to complete. - pub fn wait_on( - &self, - tcx: TyCtxt<'tcx>, - query: Option, - span: Span, - ) -> Result<(), Cycle<'tcx>> { + pub fn wait_on(&self, query: Option, span: Span) -> Result<(), Cycle<'tcx>> { + let thread_index = rustc_thread_pool::current_thread_index().unwrap(); let mut waiters_guard = self.waiters.lock(); let Some(waiters) = &mut *waiters_guard else { return Ok(()); // already complete }; - let waiter = Arc::new(QueryWaiter { - parent: query, - span, - cycle: Mutex::new(None), - condvar: Condvar::new(), - }); + let cycle = Arc::new(Mutex::new(None)); + let waiter = QueryWaiter { parent: query, span, cycle: Arc::clone(&cycle), thread_index }; // We push the waiter on to the `waiters` list. It can be accessed inside // the `wait` call below, by 1) the `set` method or 2) by deadlock detection. // Both of these will remove it from the `waiters` list before resuming // this thread. - waiters.push(Arc::clone(&waiter)); + waiters.push(waiter); // Awaits the caller on this latch by blocking the current thread. // If this detects a deadlock and the deadlock handler wants to resume this thread // we have to be in the `wait` call. This is ensured by the deadlock handler // getting the self.info lock. - rustc_thread_pool::mark_blocked(); - tcx.jobserver_proxy.release_thread(); - waiter.condvar.wait(&mut waiters_guard); - // Release the lock before we potentially block in `acquire_thread` - drop(waiters_guard); - tcx.jobserver_proxy.acquire_thread(); + rustc_thread_pool::park(waiters_guard); // FIXME: Get rid of this lock. We have ownership of the QueryWaiter // although another thread may still have a Arc reference so we cannot // use Arc::get_mut - let mut cycle = waiter.cycle.lock(); - match cycle.take() { + let mut cycle_lock = cycle.lock(); + match cycle_lock.take() { None => Ok(()), Some(cycle) => Err(cycle), } @@ -122,14 +109,13 @@ impl<'tcx> QueryLatch<'tcx> { let waiters = waiters_guard.take().unwrap(); // mark the latch as complete let registry = rustc_thread_pool::Registry::current(); for waiter in waiters { - rustc_thread_pool::mark_unblocked(®istry); - waiter.condvar.notify_one(); + rustc_thread_pool::unpark(®istry, waiter.thread_index); } } /// Removes a single waiter from the list of waiters. /// This is used to break query cycles. - pub fn extract_waiter(&self, waiter: usize) -> Arc> { + pub fn extract_waiter(&self, waiter: usize) -> QueryWaiter<'tcx> { let mut waiters_guard = self.waiters.lock(); let waiters = waiters_guard.as_mut().expect("non-empty waiters vec"); // Remove the waiter from the list of waiters diff --git a/compiler/rustc_middle/src/ty/context.rs b/compiler/rustc_middle/src/ty/context.rs index 1c7bba82d3a7b..ecb2770799346 100644 --- a/compiler/rustc_middle/src/ty/context.rs +++ b/compiler/rustc_middle/src/ty/context.rs @@ -20,7 +20,6 @@ use rustc_ast as ast; use rustc_data_structures::defer; use rustc_data_structures::fx::FxHashMap; use rustc_data_structures::intern::Interned; -use rustc_data_structures::jobserver::Proxy; use rustc_data_structures::profiling::SelfProfilerRef; use rustc_data_structures::sharded::{IntoPointer, ShardedHashMap}; use rustc_data_structures::stable_hasher::StableHash; @@ -759,9 +758,6 @@ pub struct GlobalCtxt<'tcx> { pub(crate) alloc_map: interpret::AllocMap<'tcx>, current_gcx: CurrentGcx, - - /// A jobserver reference used to release then acquire a token while waiting on a query. - pub jobserver_proxy: Arc, } impl<'tcx> GlobalCtxt<'tcx> { @@ -937,7 +933,6 @@ impl<'tcx> TyCtxt<'tcx> { query_system: QuerySystem<'tcx>, hooks: crate::hooks::Providers, current_gcx: CurrentGcx, - jobserver_proxy: Arc, f: impl FnOnce(TyCtxt<'tcx>) -> T, ) -> T { let data_layout = sess.target.parse_data_layout().unwrap_or_else(|err| { @@ -975,7 +970,6 @@ impl<'tcx> TyCtxt<'tcx> { data_layout, alloc_map: interpret::AllocMap::new(), current_gcx, - jobserver_proxy, }); // This is a separate function to work around a crash with parallel rustc (#135870) diff --git a/compiler/rustc_query_impl/src/execution.rs b/compiler/rustc_query_impl/src/execution.rs index b614bc14b4539..793677f28cba6 100644 --- a/compiler/rustc_query_impl/src/execution.rs +++ b/compiler/rustc_query_impl/src/execution.rs @@ -247,7 +247,7 @@ fn wait_for_query<'tcx, C: QueryCache>( let query_blocked_prof_timer = tcx.prof.query_blocked(); // With parallel queries we might just have to wait on some other thread. - let result = latch.wait_on(tcx, current, span); + let result = latch.wait_on(current, span); match result { Ok(()) => { diff --git a/compiler/rustc_query_impl/src/job.rs b/compiler/rustc_query_impl/src/job.rs index bf0493b29fd1e..a47b3fc62440c 100644 --- a/compiler/rustc_query_impl/src/job.rs +++ b/compiler/rustc_query_impl/src/job.rs @@ -1,6 +1,5 @@ use std::io::Write; use std::ops::ControlFlow; -use std::sync::Arc; use std::{iter, mem}; use rustc_data_structures::fx::{FxHashMap, FxHashSet}; @@ -308,7 +307,7 @@ fn process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId fn find_and_process_cycle<'tcx>( job_map: &QueryJobMap<'tcx>, query: QueryJobId, -) -> Option>> { +) -> Option> { let mut visited = FxHashSet::default(); let mut stack = Vec::new(); if let ControlFlow::Break(resumable) = @@ -350,9 +349,7 @@ pub fn break_query_cycle<'tcx>(job_map: QueryJobMap<'tcx>, registry: &rustc_thre .expect("unable to find a query cycle"); // Mark the thread we're about to wake up as unblocked. - rustc_thread_pool::mark_unblocked(registry); - - assert!(waiter.condvar.notify_one(), "unable to wake the waiter"); + assert!(rustc_thread_pool::unpark(registry, waiter.thread_index), "unable to wake the waiter"); } pub fn print_query_stack<'tcx>( diff --git a/compiler/rustc_thread_pool/Cargo.toml b/compiler/rustc_thread_pool/Cargo.toml index c92984470b7ae..dd3d4acfb0978 100644 --- a/compiler/rustc_thread_pool/Cargo.toml +++ b/compiler/rustc_thread_pool/Cargo.toml @@ -15,6 +15,7 @@ categories = ["concurrency"] [dependencies] crossbeam-deque = "0.8" crossbeam-utils = "0.8" +parking_lot = "0.12" smallvec = "1.8.1" [dev-dependencies] diff --git a/compiler/rustc_thread_pool/src/lib.rs b/compiler/rustc_thread_pool/src/lib.rs index 7ce7fbc27eabe..3ac3f37a866b6 100644 --- a/compiler/rustc_thread_pool/src/lib.rs +++ b/compiler/rustc_thread_pool/src/lib.rs @@ -95,7 +95,7 @@ pub use worker_local::WorkerLocal; pub use self::broadcast::{BroadcastContext, broadcast, spawn_broadcast}; pub use self::join::{join, join_context}; use self::registry::{CustomSpawn, DefaultSpawn, ThreadSpawn}; -pub use self::registry::{Registry, ThreadBuilder, mark_blocked, mark_unblocked}; +pub use self::registry::{Registry, ThreadBuilder, park, unpark}; pub use self::scope::{Scope, ScopeFifo, in_place_scope, in_place_scope_fifo, scope, scope_fifo}; pub use self::spawn::{spawn, spawn_fifo}; pub use self::thread_pool::{ diff --git a/compiler/rustc_thread_pool/src/registry.rs b/compiler/rustc_thread_pool/src/registry.rs index 9510c1842f86a..dc5be471914f3 100644 --- a/compiler/rustc_thread_pool/src/registry.rs +++ b/compiler/rustc_thread_pool/src/registry.rs @@ -617,19 +617,26 @@ impl Registry { /// Mark a Rayon worker thread as blocked. This triggers the deadlock handler /// if no other worker thread is active #[inline] -pub fn mark_blocked() { +pub fn park(mut mutex_guard: parking_lot::MutexGuard<'_, T>) { let worker_thread = WorkerThread::current(); assert!(!worker_thread.is_null()); unsafe { - let registry = &(*worker_thread).registry; - registry.sleep.mark_blocked(®istry.deadlock_handler) + let worker_thread = &*worker_thread; + let registry = &worker_thread.registry; + registry.sleep.mark_blocked(®istry.deadlock_handler); + registry.release_thread(); + registry.thread_infos[worker_thread.index].condvar.wait(&mut mutex_guard); + // Release the lock before we potentially block in `acquire_thread` + drop(mutex_guard); + registry.acquire_thread(); } } /// Mark a previously blocked Rayon worker thread as unblocked #[inline] -pub fn mark_unblocked(registry: &Registry) { - registry.sleep.mark_unblocked() +pub fn unpark(registry: &Registry, thread_index: usize) -> bool { + registry.sleep.mark_unblocked(); + registry.thread_infos[thread_index].condvar.notify_one() } #[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] @@ -655,6 +662,8 @@ struct ThreadInfo { /// the "stealer" half of the worker's deque stealer: Stealer, + + condvar: parking_lot::Condvar, } impl ThreadInfo { @@ -663,6 +672,7 @@ impl ThreadInfo { primed: LockLatch::new(), stopped: LockLatch::new(), terminate: OnceLatch::new(), + condvar: parking_lot::Condvar::new(), stealer, } } From 71078493a1d3a38410287686be8d9945304d5e02 Mon Sep 17 00:00:00 2001 From: Daria Sukhonina Date: Thu, 30 Apr 2026 13:50:04 +0300 Subject: [PATCH 2/6] Use Arc::get_mut --- compiler/rustc_middle/src/query/job.rs | 16 +++++++++------- compiler/rustc_query_impl/src/job.rs | 22 ++++++++++------------ 2 files changed, 19 insertions(+), 19 deletions(-) diff --git a/compiler/rustc_middle/src/query/job.rs b/compiler/rustc_middle/src/query/job.rs index 33e04134837b5..14540d1c88fda 100644 --- a/compiler/rustc_middle/src/query/job.rs +++ b/compiler/rustc_middle/src/query/job.rs @@ -78,7 +78,7 @@ impl<'tcx> QueryLatch<'tcx> { return Ok(()); // already complete }; - let cycle = Arc::new(Mutex::new(None)); + let mut cycle = Arc::new(Mutex::new(None)); let waiter = QueryWaiter { parent: query, span, cycle: Arc::clone(&cycle), thread_index }; // We push the waiter on to the `waiters` list. It can be accessed inside @@ -93,11 +93,9 @@ impl<'tcx> QueryLatch<'tcx> { // getting the self.info lock. rustc_thread_pool::park(waiters_guard); - // FIXME: Get rid of this lock. We have ownership of the QueryWaiter - // although another thread may still have a Arc reference so we cannot - // use Arc::get_mut - let mut cycle_lock = cycle.lock(); - match cycle_lock.take() { + // We make sure to drop waiter before unparking a worker thread + let cycle = Arc::get_mut(&mut cycle).unwrap().get_mut(); + match cycle.take() { None => Ok(()), Some(cycle) => Err(cycle), } @@ -109,7 +107,11 @@ impl<'tcx> QueryLatch<'tcx> { let waiters = waiters_guard.take().unwrap(); // mark the latch as complete let registry = rustc_thread_pool::Registry::current(); for waiter in waiters { - rustc_thread_pool::unpark(®istry, waiter.thread_index); + // Return waiter thread's index to resume and drop `waiter` for resumed thread + // to use `Arc::get_mut` on its cycle arc pointer. + let waiter_thread = waiter.thread_index; + drop(waiter); + rustc_thread_pool::unpark(®istry, waiter_thread); } } diff --git a/compiler/rustc_query_impl/src/job.rs b/compiler/rustc_query_impl/src/job.rs index a47b3fc62440c..cc8e16aece54d 100644 --- a/compiler/rustc_query_impl/src/job.rs +++ b/compiler/rustc_query_impl/src/job.rs @@ -6,7 +6,7 @@ use rustc_data_structures::fx::{FxHashMap, FxHashSet}; use rustc_errors::{Diag, DiagCtxtHandle}; use rustc_hir::def::DefKind; use rustc_middle::queries::TaggedQueryKey; -use rustc_middle::query::{Cycle, QueryJob, QueryJobId, QueryLatch, QueryStackFrame, QueryWaiter}; +use rustc_middle::query::{Cycle, QueryJob, QueryJobId, QueryLatch, QueryStackFrame}; use rustc_middle::ty::TyCtxt; use rustc_span::{DUMMY_SP, Span}; @@ -303,11 +303,8 @@ fn process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId } /// Looks for a query cycle starting at `query`. -/// Returns a waiter to resume if a cycle is found. -fn find_and_process_cycle<'tcx>( - job_map: &QueryJobMap<'tcx>, - query: QueryJobId, -) -> Option> { +/// Returns a waiter thread's index to resume if a cycle is found. +fn find_and_process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, query: QueryJobId) -> Option { let mut visited = FxHashSet::default(); let mut stack = Vec::new(); if let ControlFlow::Break(resumable) = @@ -326,8 +323,9 @@ fn find_and_process_cycle<'tcx>( // Set the cycle error so it will be picked up when resumed *waiter.cycle.lock() = Some(error); - // Put the waiter on the list of things to resume - Some(waiter) + // Return waiter thread's index to resume and drop `QueryWaiter::cycle` for resumed thread + // to use `Arc::get_mut`. + Some(waiter.thread_index) } else { None } @@ -341,15 +339,15 @@ fn find_and_process_cycle<'tcx>( /// there will be multiple rounds through the deadlock handler if multiple cycles are present. #[allow(rustc::potential_query_instability)] pub fn break_query_cycle<'tcx>(job_map: QueryJobMap<'tcx>, registry: &rustc_thread_pool::Registry) { - // Look for a cycle starting at each query job - let waiter = job_map + // Look for a cycle starting at each query job, + let waiter_thread = job_map .map .keys() .find_map(|query| find_and_process_cycle(&job_map, *query)) .expect("unable to find a query cycle"); - // Mark the thread we're about to wake up as unblocked. - assert!(rustc_thread_pool::unpark(registry, waiter.thread_index), "unable to wake the waiter"); + // Unpark one waiter thread. + assert!(rustc_thread_pool::unpark(registry, waiter_thread), "unable to wake the waiter"); } pub fn print_query_stack<'tcx>( From 4c10e0afb7eb08a493173c9235a53b43260f2cd6 Mon Sep 17 00:00:00 2001 From: Daria Sukhonina Date: Fri, 8 May 2026 14:57:13 +0300 Subject: [PATCH 3/6] Index waiters per worker thread --- compiler/rustc_middle/src/query/job.rs | 16 +++++++++++----- compiler/rustc_query_impl/src/job.rs | 3 +++ 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/compiler/rustc_middle/src/query/job.rs b/compiler/rustc_middle/src/query/job.rs index 14540d1c88fda..2cf107cf47cc0 100644 --- a/compiler/rustc_middle/src/query/job.rs +++ b/compiler/rustc_middle/src/query/job.rs @@ -5,6 +5,7 @@ use std::sync::Arc; use parking_lot::Mutex; use rustc_span::Span; +use rustc_thread_pool::current_num_threads; use crate::query::Cycle; @@ -50,7 +51,7 @@ impl<'tcx> QueryJob<'tcx> { } } -#[derive(Debug)] +#[derive(Clone, Debug)] pub struct QueryWaiter<'tcx> { pub parent: Option, // FIXME: could be made u16 due to rustc_thread_pool limiting number of threads @@ -62,12 +63,12 @@ pub struct QueryWaiter<'tcx> { #[derive(Clone, Debug)] pub struct QueryLatch<'tcx> { /// The `Option` is `Some(..)` when the job is active, and `None` once completed. - pub waiters: Arc>>>>, + pub waiters: Arc>]>>>>, } impl<'tcx> QueryLatch<'tcx> { fn new() -> Self { - QueryLatch { waiters: Arc::new(Mutex::new(Some(Vec::new()))) } + QueryLatch { waiters: Arc::new(Mutex::new(Some(vec![None; current_num_threads()].into_boxed_slice()))) } } /// Awaits for the query job to complete. @@ -85,7 +86,9 @@ impl<'tcx> QueryLatch<'tcx> { // the `wait` call below, by 1) the `set` method or 2) by deadlock detection. // Both of these will remove it from the `waiters` list before resuming // this thread. - waiters.push(waiter); + if waiters[thread_index].replace(waiter).is_some() { + panic!("tried to place a waiter twice for a worker thread") + } // Awaits the caller on this latch by blocking the current thread. // If this detects a deadlock and the deadlock handler wants to resume this thread @@ -107,6 +110,9 @@ impl<'tcx> QueryLatch<'tcx> { let waiters = waiters_guard.take().unwrap(); // mark the latch as complete let registry = rustc_thread_pool::Registry::current(); for waiter in waiters { + let Some(waiter) = waiter else { + continue + }; // Return waiter thread's index to resume and drop `waiter` for resumed thread // to use `Arc::get_mut` on its cycle arc pointer. let waiter_thread = waiter.thread_index; @@ -121,6 +127,6 @@ impl<'tcx> QueryLatch<'tcx> { let mut waiters_guard = self.waiters.lock(); let waiters = waiters_guard.as_mut().expect("non-empty waiters vec"); // Remove the waiter from the list of waiters - waiters.remove(waiter) + waiters[waiter].take().unwrap() } } diff --git a/compiler/rustc_query_impl/src/job.rs b/compiler/rustc_query_impl/src/job.rs index cc8e16aece54d..d2ce083b4b901 100644 --- a/compiler/rustc_query_impl/src/job.rs +++ b/compiler/rustc_query_impl/src/job.rs @@ -140,6 +140,9 @@ fn abstracted_waiters_of(job_map: &QueryJobMap<'_>, query: QueryJobId) -> Vec Date: Fri, 8 May 2026 15:19:46 +0300 Subject: [PATCH 4/6] remove thread_index field from QueryWaiter --- compiler/rustc_middle/src/query/job.rs | 7 ++----- compiler/rustc_query_impl/src/job.rs | 2 +- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/compiler/rustc_middle/src/query/job.rs b/compiler/rustc_middle/src/query/job.rs index 2cf107cf47cc0..4ebc77bdc2725 100644 --- a/compiler/rustc_middle/src/query/job.rs +++ b/compiler/rustc_middle/src/query/job.rs @@ -54,8 +54,6 @@ impl<'tcx> QueryJob<'tcx> { #[derive(Clone, Debug)] pub struct QueryWaiter<'tcx> { pub parent: Option, - // FIXME: could be made u16 due to rustc_thread_pool limiting number of threads - pub thread_index: usize, pub span: Span, pub cycle: Arc>>>, } @@ -80,7 +78,7 @@ impl<'tcx> QueryLatch<'tcx> { }; let mut cycle = Arc::new(Mutex::new(None)); - let waiter = QueryWaiter { parent: query, span, cycle: Arc::clone(&cycle), thread_index }; + let waiter = QueryWaiter { parent: query, span, cycle: Arc::clone(&cycle) }; // We push the waiter on to the `waiters` list. It can be accessed inside // the `wait` call below, by 1) the `set` method or 2) by deadlock detection. @@ -109,13 +107,12 @@ impl<'tcx> QueryLatch<'tcx> { let mut waiters_guard = self.waiters.lock(); let waiters = waiters_guard.take().unwrap(); // mark the latch as complete let registry = rustc_thread_pool::Registry::current(); - for waiter in waiters { + for (waiter_thread, waiter) in waiters.into_iter().enumerate() { let Some(waiter) = waiter else { continue }; // Return waiter thread's index to resume and drop `waiter` for resumed thread // to use `Arc::get_mut` on its cycle arc pointer. - let waiter_thread = waiter.thread_index; drop(waiter); rustc_thread_pool::unpark(®istry, waiter_thread); } diff --git a/compiler/rustc_query_impl/src/job.rs b/compiler/rustc_query_impl/src/job.rs index d2ce083b4b901..d36c139eee3b2 100644 --- a/compiler/rustc_query_impl/src/job.rs +++ b/compiler/rustc_query_impl/src/job.rs @@ -328,7 +328,7 @@ fn find_and_process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, query: QueryJobId) // Return waiter thread's index to resume and drop `QueryWaiter::cycle` for resumed thread // to use `Arc::get_mut`. - Some(waiter.thread_index) + Some(waiter_idx) } else { None } From a843b43fb7ae00c63982bf0b0007426d6e2a2bff Mon Sep 17 00:00:00 2001 From: Daria Sukhonina Date: Fri, 8 May 2026 16:46:05 +0300 Subject: [PATCH 5/6] Second attempt --- .../src/sync/worker_local.rs | 12 ++++ compiler/rustc_interface/src/util.rs | 2 +- compiler/rustc_middle/src/query/job.rs | 67 +++++++++---------- compiler/rustc_middle/src/ty/context.rs | 5 +- compiler/rustc_query_impl/src/execution.rs | 2 +- compiler/rustc_query_impl/src/job.rs | 56 ++++++++++------ compiler/rustc_thread_pool/src/registry.rs | 5 +- .../rustc_thread_pool/src/sleep/counters.rs | 6 +- 8 files changed, 87 insertions(+), 68 deletions(-) diff --git a/compiler/rustc_data_structures/src/sync/worker_local.rs b/compiler/rustc_data_structures/src/sync/worker_local.rs index d75af00985047..479dabdbf5d8a 100644 --- a/compiler/rustc_data_structures/src/sync/worker_local.rs +++ b/compiler/rustc_data_structures/src/sync/worker_local.rs @@ -129,6 +129,18 @@ impl WorkerLocal { pub fn into_inner(self) -> impl Iterator { self.locals.into_vec().into_iter().map(|local| local.0) } + + #[inline] + pub unsafe fn as_slice_unchecked(this: &Self) -> &[CacheAligned] { + &this.locals + } +} + +impl WorkerLocal { + #[inline] + pub fn as_slice(this: &Self) -> &[CacheAligned] { + &this.locals + } } impl Deref for WorkerLocal { diff --git a/compiler/rustc_interface/src/util.rs b/compiler/rustc_interface/src/util.rs index 157ba89abbfee..737b5d647903b 100644 --- a/compiler/rustc_interface/src/util.rs +++ b/compiler/rustc_interface/src/util.rs @@ -258,7 +258,7 @@ internal compiler error: query cycle handler thread panicked, aborting process"; ) }, ); - break_query_cycle(job_map, ®istry); + break_query_cycle(tcx, job_map, ®istry); }) }) }); diff --git a/compiler/rustc_middle/src/query/job.rs b/compiler/rustc_middle/src/query/job.rs index 4ebc77bdc2725..3fc07f3b70e39 100644 --- a/compiler/rustc_middle/src/query/job.rs +++ b/compiler/rustc_middle/src/query/job.rs @@ -1,13 +1,15 @@ use std::fmt::Debug; use std::hash::Hash; +use std::marker::PhantomData; +use std::mem; use std::num::NonZero; use std::sync::Arc; use parking_lot::Mutex; use rustc_span::Span; -use rustc_thread_pool::current_num_threads; use crate::query::Cycle; +use crate::ty::TyCtxt; /// A value uniquely identifying an active query job. #[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)] @@ -51,79 +53,70 @@ impl<'tcx> QueryJob<'tcx> { } } -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct QueryWaiter<'tcx> { pub parent: Option, pub span: Span, - pub cycle: Arc>>>, + pub cycle: Option>, } #[derive(Clone, Debug)] pub struct QueryLatch<'tcx> { - /// The `Option` is `Some(..)` when the job is active, and `None` once completed. - pub waiters: Arc>]>>>>, + /// The `waiters` is not `usize::MAX` when the job is active, and `usize::MAX` once completed. + pub waiters: Arc>, + pub _marker: PhantomData<&'tcx ()>, } impl<'tcx> QueryLatch<'tcx> { fn new() -> Self { - QueryLatch { waiters: Arc::new(Mutex::new(Some(vec![None; current_num_threads()].into_boxed_slice()))) } + QueryLatch { waiters: Arc::new(Mutex::new(0)), _marker: PhantomData } } /// Awaits for the query job to complete. - pub fn wait_on(&self, query: Option, span: Span) -> Result<(), Cycle<'tcx>> { + pub fn wait_on(&self, tcx: TyCtxt<'tcx>, query: Option, span: Span) -> Result<(), Cycle<'tcx>> { let thread_index = rustc_thread_pool::current_thread_index().unwrap(); let mut waiters_guard = self.waiters.lock(); - let Some(waiters) = &mut *waiters_guard else { + if *waiters_guard == usize::MAX { return Ok(()); // already complete }; + debug_assert!(*waiters_guard & (1 << thread_index) == 0); - let mut cycle = Arc::new(Mutex::new(None)); - let waiter = QueryWaiter { parent: query, span, cycle: Arc::clone(&cycle) }; + let waiter = QueryWaiter { parent: query, span, cycle: None }; // We push the waiter on to the `waiters` list. It can be accessed inside // the `wait` call below, by 1) the `set` method or 2) by deadlock detection. // Both of these will remove it from the `waiters` list before resuming // this thread. - if waiters[thread_index].replace(waiter).is_some() { + if tcx.waiters.replace(Some(waiter)).is_some() { panic!("tried to place a waiter twice for a worker thread") } + *waiters_guard |= 1 << thread_index; + // Awaits the caller on this latch by blocking the current thread. // If this detects a deadlock and the deadlock handler wants to resume this thread // we have to be in the `wait` call. This is ensured by the deadlock handler // getting the self.info lock. - rustc_thread_pool::park(waiters_guard); - - // We make sure to drop waiter before unparking a worker thread - let cycle = Arc::get_mut(&mut cycle).unwrap().get_mut(); - match cycle.take() { - None => Ok(()), - Some(cycle) => Err(cycle), - } + rustc_thread_pool::park(waiters_guard, |_| { + // Reset our QueryWaiter to None + let mut waiter = tcx.waiters.take().unwrap(); + match waiter.cycle.take() { + None => Ok(()), + Some(cycle) => Err(cycle), + } + }) } /// Sets the latch and resumes all waiters on it fn set(&self) { let mut waiters_guard = self.waiters.lock(); - let waiters = waiters_guard.take().unwrap(); // mark the latch as complete + let waiters = mem::replace(&mut *waiters_guard, usize::MAX); // mark the latch as complete + debug_assert!(waiters != usize::MAX); let registry = rustc_thread_pool::Registry::current(); - for (waiter_thread, waiter) in waiters.into_iter().enumerate() { - let Some(waiter) = waiter else { - continue - }; - // Return waiter thread's index to resume and drop `waiter` for resumed thread - // to use `Arc::get_mut` on its cycle arc pointer. - drop(waiter); - rustc_thread_pool::unpark(®istry, waiter_thread); + for waiter_thread in 0..usize::BITS - 1 { + if waiters & (1 << waiter_thread) != 0 { + rustc_thread_pool::unpark(®istry, waiter_thread as usize); + } } } - - /// Removes a single waiter from the list of waiters. - /// This is used to break query cycles. - pub fn extract_waiter(&self, waiter: usize) -> QueryWaiter<'tcx> { - let mut waiters_guard = self.waiters.lock(); - let waiters = waiters_guard.as_mut().expect("non-empty waiters vec"); - // Remove the waiter from the list of waiters - waiters[waiter].take().unwrap() - } } diff --git a/compiler/rustc_middle/src/ty/context.rs b/compiler/rustc_middle/src/ty/context.rs index ecb2770799346..cf8d1d736dcd2 100644 --- a/compiler/rustc_middle/src/ty/context.rs +++ b/compiler/rustc_middle/src/ty/context.rs @@ -6,6 +6,7 @@ mod impl_interner; pub mod tls; use std::borrow::{Borrow, Cow}; +use std::cell::RefCell; use std::cmp::Ordering; use std::env::VarError; use std::ffi::OsStr; @@ -59,7 +60,7 @@ use crate::middle::codegen_fn_attrs::{CodegenFnAttrs, TargetFeature}; use crate::middle::resolve_bound_vars; use crate::mir::interpret::{self, Allocation, ConstAllocation}; use crate::mir::{Body, Local, Place, PlaceElem, ProjectionKind, Promoted}; -use crate::query::{IntoQueryKey, LocalCrate, Providers, QuerySystem, TyCtxtAt}; +use crate::query::{IntoQueryKey, LocalCrate, Providers, QuerySystem, QueryWaiter, TyCtxtAt}; use crate::thir::Thir; use crate::traits; use crate::traits::solve::{ExternalConstraints, ExternalConstraintsData, PredefinedOpaques}; @@ -692,6 +693,7 @@ impl<'tcx> Deref for TyCtxt<'tcx> { pub struct GlobalCtxt<'tcx> { pub arena: &'tcx WorkerLocal>, pub hir_arena: &'tcx WorkerLocal>, + pub waiters: WorkerLocal>>>, interners: CtxtInterners<'tcx>, @@ -949,6 +951,7 @@ impl<'tcx> TyCtxt<'tcx> { stable_crate_id, arena, hir_arena, + waiters: Default::default(), interners, dep_graph, hooks, diff --git a/compiler/rustc_query_impl/src/execution.rs b/compiler/rustc_query_impl/src/execution.rs index 793677f28cba6..b614bc14b4539 100644 --- a/compiler/rustc_query_impl/src/execution.rs +++ b/compiler/rustc_query_impl/src/execution.rs @@ -247,7 +247,7 @@ fn wait_for_query<'tcx, C: QueryCache>( let query_blocked_prof_timer = tcx.prof.query_blocked(); // With parallel queries we might just have to wait on some other thread. - let result = latch.wait_on(current, span); + let result = latch.wait_on(tcx, current, span); match result { Ok(()) => { diff --git a/compiler/rustc_query_impl/src/job.rs b/compiler/rustc_query_impl/src/job.rs index d36c139eee3b2..2fbf49d008a3a 100644 --- a/compiler/rustc_query_impl/src/job.rs +++ b/compiler/rustc_query_impl/src/job.rs @@ -3,6 +3,7 @@ use std::ops::ControlFlow; use std::{iter, mem}; use rustc_data_structures::fx::{FxHashMap, FxHashSet}; +use rustc_data_structures::sync::WorkerLocal; use rustc_errors::{Diag, DiagCtxtHandle}; use rustc_hir::def::DefKind; use rustc_middle::queries::TaggedQueryKey; @@ -112,7 +113,7 @@ pub(crate) fn find_dep_kind_root<'tcx>( } /// The locaton of a resumable waiter. The usize is the index into waiters in the query's latch. -/// We'll use this to remove the waiter using `QueryLatch::extract_waiter` if we're waking it up. +/// We'll use this to remove the waiter if we're waking it up. type ResumableWaiterLocation = (QueryJobId, usize); /// This abstracts over non-resumable waiters which are found in `QueryJob`'s `parent` field @@ -127,7 +128,7 @@ struct AbstractedWaiter { /// Returns all the non-resumable and resumable waiters of a query. /// This is used so we can uniformly loop over both non-resumable and resumable waiters. -fn abstracted_waiters_of(job_map: &QueryJobMap<'_>, query: QueryJobId) -> Vec { +fn abstracted_waiters_of<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, query: QueryJobId) -> Vec { let mut result = Vec::new(); // Add the parent which is a non-resumable waiter since it's on the same stack @@ -139,14 +140,18 @@ fn abstracted_waiters_of(job_map: &QueryJobMap<'_>, query: QueryJobId) -> Vec, query: QueryJobId) -> Vec( + tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, query: QueryJobId, span: Span, @@ -183,13 +189,13 @@ fn find_cycle<'tcx>( stack.push((span, query)); // Visit all the waiters - for abstracted_waiter in abstracted_waiters_of(job_map, query) { + for abstracted_waiter in abstracted_waiters_of(tcx, job_map, query) { let Some(parent) = abstracted_waiter.parent else { // Skip waiters which are not queries continue; }; if let ControlFlow::Break(maybe_resumable) = - find_cycle(job_map, parent, abstracted_waiter.span, stack, visited) + find_cycle(tcx, job_map, parent, abstracted_waiter.span, stack, visited) { // Return the resumable waiter in `waiter.resumable` if present return ControlFlow::Break(abstracted_waiter.resumable.or(maybe_resumable)); @@ -206,6 +212,7 @@ fn find_cycle<'tcx>( /// from `query` without going through any of the queries in `visited`. /// This is achieved with a depth first search. fn connected_to_root<'tcx>( + tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, query: QueryJobId, visited: &mut FxHashSet, @@ -216,12 +223,12 @@ fn connected_to_root<'tcx>( } // Visit all the waiters - for abstracted_waiter in abstracted_waiters_of(job_map, query) { + for abstracted_waiter in abstracted_waiters_of(tcx, job_map, query) { match abstracted_waiter.parent { // This query is connected to the root None => return true, Some(parent) => { - if connected_to_root(job_map, parent, visited) { + if connected_to_root(tcx, job_map, parent, visited) { return true; } } @@ -232,7 +239,7 @@ fn connected_to_root<'tcx>( } /// Processes a found query cycle into a `Cycle` -fn process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId)>) -> Cycle<'tcx> { +fn process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId)>) -> Cycle<'tcx> { // The stack is a vector of pairs of spans and queries; reverse it so that // the earlier entries require later entries let (mut spans, queries): (Vec<_>, Vec<_>) = stack.into_iter().rev().unzip(); @@ -257,7 +264,7 @@ fn process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId let mut query_waiting_on_cycle = None; // Find a direct waiter who leads to the root - for abstracted_waiter in abstracted_waiters_of(job_map, query_in_cycle) { + for abstracted_waiter in abstracted_waiters_of(tcx, job_map, query_in_cycle) { let Some(parent) = abstracted_waiter.parent else { // The query in the cycle is directly connected to root. entrypoint = true; @@ -268,7 +275,7 @@ fn process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId // so paths to the root through the cycle itself won't count. let mut visited = FxHashSet::from_iter(stack.iter().map(|q| q.1)); - if connected_to_root(job_map, parent, &mut visited) { + if connected_to_root(tcx, job_map, parent, &mut visited) { query_waiting_on_cycle = Some((abstracted_waiter.span, parent)); entrypoint = true; break; @@ -307,24 +314,29 @@ fn process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId /// Looks for a query cycle starting at `query`. /// Returns a waiter thread's index to resume if a cycle is found. -fn find_and_process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, query: QueryJobId) -> Option { +fn find_and_process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, query: QueryJobId) -> Option { let mut visited = FxHashSet::default(); let mut stack = Vec::new(); if let ControlFlow::Break(resumable) = - find_cycle(job_map, query, DUMMY_SP, &mut stack, &mut visited) + find_cycle(tcx, job_map, query, DUMMY_SP, &mut stack, &mut visited) { // Create the cycle error - let error = process_cycle(job_map, stack); + let error = process_cycle(tcx, job_map, stack); // We unwrap `resumable` here since there must always be one // edge which is resumable / waited using a query latch let (waitee_query, waiter_idx) = resumable.unwrap(); - // Extract the waiter we want to resume - let waiter = job_map.latch_of(waitee_query).unwrap().extract_waiter(waiter_idx); + // Disable the waiter we want to resume + let mut waiters_guard = job_map.latch_of(waitee_query).unwrap().waiters.lock(); + *waiters_guard &= !(1 << waiter_idx); // Set the cycle error so it will be picked up when resumed - *waiter.cycle.lock() = Some(error); + // + // SAFETY: We are in a deadlock, so we are synced with whatever tcx.waiters contains. + // Also it's later synced though `waiters_guard` unlocking the mutex and park call reclaiming + // its lock through a condvar after being unparked. + unsafe { WorkerLocal::as_slice_unchecked(&tcx.waiters)[waiter_idx].0.borrow_mut().as_mut().unwrap().cycle = Some(error) }; // Return waiter thread's index to resume and drop `QueryWaiter::cycle` for resumed thread // to use `Arc::get_mut`. @@ -341,12 +353,12 @@ fn find_and_process_cycle<'tcx>(job_map: &QueryJobMap<'tcx>, query: QueryJobId) /// There may be multiple cycles involved in a deadlock, but this only breaks one at a time so /// there will be multiple rounds through the deadlock handler if multiple cycles are present. #[allow(rustc::potential_query_instability)] -pub fn break_query_cycle<'tcx>(job_map: QueryJobMap<'tcx>, registry: &rustc_thread_pool::Registry) { - // Look for a cycle starting at each query job, +pub fn break_query_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: QueryJobMap<'tcx>, registry: &rustc_thread_pool::Registry) { + // Look for a cycle starting at each query job let waiter_thread = job_map .map .keys() - .find_map(|query| find_and_process_cycle(&job_map, *query)) + .find_map(|query| find_and_process_cycle(tcx, &job_map, *query)) .expect("unable to find a query cycle"); // Unpark one waiter thread. diff --git a/compiler/rustc_thread_pool/src/registry.rs b/compiler/rustc_thread_pool/src/registry.rs index dc5be471914f3..bf7ef37a7816f 100644 --- a/compiler/rustc_thread_pool/src/registry.rs +++ b/compiler/rustc_thread_pool/src/registry.rs @@ -617,7 +617,7 @@ impl Registry { /// Mark a Rayon worker thread as blocked. This triggers the deadlock handler /// if no other worker thread is active #[inline] -pub fn park(mut mutex_guard: parking_lot::MutexGuard<'_, T>) { +pub fn park(mut mutex_guard: parking_lot::MutexGuard<'_, T>, f: impl FnOnce(&mut T) -> R) -> R { let worker_thread = WorkerThread::current(); assert!(!worker_thread.is_null()); unsafe { @@ -626,9 +626,12 @@ pub fn park(mut mutex_guard: parking_lot::MutexGuard<'_, T>) { registry.sleep.mark_blocked(®istry.deadlock_handler); registry.release_thread(); registry.thread_infos[worker_thread.index].condvar.wait(&mut mutex_guard); + let res = f(&mut mutex_guard); // Release the lock before we potentially block in `acquire_thread` + // NOTE: This mutex is used for synchronization drop(mutex_guard); registry.acquire_thread(); + res } } diff --git a/compiler/rustc_thread_pool/src/sleep/counters.rs b/compiler/rustc_thread_pool/src/sleep/counters.rs index f2682028b96a6..9a4e153567e58 100644 --- a/compiler/rustc_thread_pool/src/sleep/counters.rs +++ b/compiler/rustc_thread_pool/src/sleep/counters.rs @@ -53,11 +53,7 @@ impl JobsEventCounter { } /// Number of bits used for the thread counters. -#[cfg(target_pointer_width = "64")] -const THREADS_BITS: usize = 16; - -#[cfg(target_pointer_width = "32")] -const THREADS_BITS: usize = 8; +const THREADS_BITS: usize = usize::BITS.ilog2() as usize; /// Bits to shift to select the sleeping threads /// (used with `select_bits`). From 29c7837a06cf3a9bf025da2bb7aa6796d1c91fdd Mon Sep 17 00:00:00 2001 From: Daria Sukhonina Date: Fri, 8 May 2026 17:29:56 +0300 Subject: [PATCH 6/6] Max precautions --- compiler/rustc_middle/src/query/job.rs | 16 ++++++--- compiler/rustc_middle/src/ty/context.rs | 4 +-- compiler/rustc_query_impl/src/job.rs | 41 ++++++++++++++++------ compiler/rustc_thread_pool/src/registry.rs | 5 ++- 4 files changed, 47 insertions(+), 19 deletions(-) diff --git a/compiler/rustc_middle/src/query/job.rs b/compiler/rustc_middle/src/query/job.rs index 3fc07f3b70e39..05c4ce925b0cd 100644 --- a/compiler/rustc_middle/src/query/job.rs +++ b/compiler/rustc_middle/src/query/job.rs @@ -73,7 +73,12 @@ impl<'tcx> QueryLatch<'tcx> { } /// Awaits for the query job to complete. - pub fn wait_on(&self, tcx: TyCtxt<'tcx>, query: Option, span: Span) -> Result<(), Cycle<'tcx>> { + pub fn wait_on( + &self, + tcx: TyCtxt<'tcx>, + query: Option, + span: Span, + ) -> Result<(), Cycle<'tcx>> { let thread_index = rustc_thread_pool::current_thread_index().unwrap(); let mut waiters_guard = self.waiters.lock(); if *waiters_guard == usize::MAX { @@ -87,11 +92,12 @@ impl<'tcx> QueryLatch<'tcx> { // the `wait` call below, by 1) the `set` method or 2) by deadlock detection. // Both of these will remove it from the `waiters` list before resuming // this thread. - if tcx.waiters.replace(Some(waiter)).is_some() { + let mut waiters_state = tcx.waiters.lock(); + if mem::replace(&mut *waiters_state, Some(waiter)).is_some() { panic!("tried to place a waiter twice for a worker thread") } - *waiters_guard |= 1 << thread_index; + drop(waiters_state); // Awaits the caller on this latch by blocking the current thread. // If this detects a deadlock and the deadlock handler wants to resume this thread @@ -99,8 +105,8 @@ impl<'tcx> QueryLatch<'tcx> { // getting the self.info lock. rustc_thread_pool::park(waiters_guard, |_| { // Reset our QueryWaiter to None - let mut waiter = tcx.waiters.take().unwrap(); - match waiter.cycle.take() { + let waiter = tcx.waiters.lock().take().unwrap(); + match waiter.cycle { None => Ok(()), Some(cycle) => Err(cycle), } diff --git a/compiler/rustc_middle/src/ty/context.rs b/compiler/rustc_middle/src/ty/context.rs index cf8d1d736dcd2..4c15201a42ef9 100644 --- a/compiler/rustc_middle/src/ty/context.rs +++ b/compiler/rustc_middle/src/ty/context.rs @@ -6,7 +6,6 @@ mod impl_interner; pub mod tls; use std::borrow::{Borrow, Cow}; -use std::cell::RefCell; use std::cmp::Ordering; use std::env::VarError; use std::ffi::OsStr; @@ -16,6 +15,7 @@ use std::ops::Deref; use std::sync::{Arc, OnceLock}; use std::{fmt, iter, mem}; +use parking_lot::Mutex; use rustc_abi::{ExternAbi, FieldIdx, Layout, LayoutData, TargetDataLayout, VariantIdx}; use rustc_ast as ast; use rustc_data_structures::defer; @@ -693,7 +693,7 @@ impl<'tcx> Deref for TyCtxt<'tcx> { pub struct GlobalCtxt<'tcx> { pub arena: &'tcx WorkerLocal>, pub hir_arena: &'tcx WorkerLocal>, - pub waiters: WorkerLocal>>>, + pub waiters: WorkerLocal>>>, interners: CtxtInterners<'tcx>, diff --git a/compiler/rustc_query_impl/src/job.rs b/compiler/rustc_query_impl/src/job.rs index 2fbf49d008a3a..a803ced823838 100644 --- a/compiler/rustc_query_impl/src/job.rs +++ b/compiler/rustc_query_impl/src/job.rs @@ -128,7 +128,11 @@ struct AbstractedWaiter { /// Returns all the non-resumable and resumable waiters of a query. /// This is used so we can uniformly loop over both non-resumable and resumable waiters. -fn abstracted_waiters_of<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, query: QueryJobId) -> Vec { +fn abstracted_waiters_of<'tcx>( + tcx: TyCtxt<'tcx>, + job_map: &QueryJobMap<'tcx>, + query: QueryJobId, +) -> Vec { let mut result = Vec::new(); // Add the parent which is a non-resumable waiter since it's on the same stack @@ -144,9 +148,9 @@ fn abstracted_waiters_of<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, q debug_assert!(*waiters_guard != usize::MAX); for i in 0..usize::BITS - 1 { if *waiters_guard & (1 << i) == 0 { - continue + continue; } - let waiter = unsafe { WorkerLocal::as_slice_unchecked(&tcx.waiters)[i as usize].0.borrow() }; + let waiter = WorkerLocal::as_slice(&tcx.waiters)[i as usize].0.lock(); let waiter = waiter.as_ref().unwrap(); result.push(AbstractedWaiter { span: waiter.span, @@ -239,7 +243,11 @@ fn connected_to_root<'tcx>( } /// Processes a found query cycle into a `Cycle` -fn process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, stack: Vec<(Span, QueryJobId)>) -> Cycle<'tcx> { +fn process_cycle<'tcx>( + tcx: TyCtxt<'tcx>, + job_map: &QueryJobMap<'tcx>, + stack: Vec<(Span, QueryJobId)>, +) -> Cycle<'tcx> { // The stack is a vector of pairs of spans and queries; reverse it so that // the earlier entries require later entries let (mut spans, queries): (Vec<_>, Vec<_>) = stack.into_iter().rev().unzip(); @@ -314,7 +322,11 @@ fn process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, stack: Ve /// Looks for a query cycle starting at `query`. /// Returns a waiter thread's index to resume if a cycle is found. -fn find_and_process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, query: QueryJobId) -> Option { +fn find_and_process_cycle<'tcx>( + tcx: TyCtxt<'tcx>, + job_map: &QueryJobMap<'tcx>, + query: QueryJobId, +) -> Option<(usize, QueryJobId)> { let mut visited = FxHashSet::default(); let mut stack = Vec::new(); if let ControlFlow::Break(resumable) = @@ -328,19 +340,19 @@ fn find_and_process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, let (waitee_query, waiter_idx) = resumable.unwrap(); // Disable the waiter we want to resume - let mut waiters_guard = job_map.latch_of(waitee_query).unwrap().waiters.lock(); - *waiters_guard &= !(1 << waiter_idx); + let _g = job_map.latch_of(waitee_query).unwrap().waiters.lock(); // Set the cycle error so it will be picked up when resumed // // SAFETY: We are in a deadlock, so we are synced with whatever tcx.waiters contains. // Also it's later synced though `waiters_guard` unlocking the mutex and park call reclaiming // its lock through a condvar after being unparked. - unsafe { WorkerLocal::as_slice_unchecked(&tcx.waiters)[waiter_idx].0.borrow_mut().as_mut().unwrap().cycle = Some(error) }; + WorkerLocal::as_slice(&tcx.waiters)[waiter_idx].0.lock().as_mut().unwrap().cycle = + Some(error); // Return waiter thread's index to resume and drop `QueryWaiter::cycle` for resumed thread // to use `Arc::get_mut`. - Some(waiter_idx) + Some((waiter_idx, waitee_query)) } else { None } @@ -353,14 +365,21 @@ fn find_and_process_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: &QueryJobMap<'tcx>, /// There may be multiple cycles involved in a deadlock, but this only breaks one at a time so /// there will be multiple rounds through the deadlock handler if multiple cycles are present. #[allow(rustc::potential_query_instability)] -pub fn break_query_cycle<'tcx>(tcx: TyCtxt<'tcx>, job_map: QueryJobMap<'tcx>, registry: &rustc_thread_pool::Registry) { +pub fn break_query_cycle<'tcx>( + tcx: TyCtxt<'tcx>, + job_map: QueryJobMap<'tcx>, + registry: &rustc_thread_pool::Registry, +) { // Look for a cycle starting at each query job - let waiter_thread = job_map + let (waiter_thread, waitee_query) = job_map .map .keys() .find_map(|query| find_and_process_cycle(tcx, &job_map, *query)) .expect("unable to find a query cycle"); + let mut waiters_guard = job_map.latch_of(waitee_query).unwrap().waiters.lock(); + debug_assert!(*waiters_guard & (1 << waiter_thread) != 0); + *waiters_guard &= !(1 << waiter_thread); // Unpark one waiter thread. assert!(rustc_thread_pool::unpark(registry, waiter_thread), "unable to wake the waiter"); } diff --git a/compiler/rustc_thread_pool/src/registry.rs b/compiler/rustc_thread_pool/src/registry.rs index bf7ef37a7816f..f9f433b5f9341 100644 --- a/compiler/rustc_thread_pool/src/registry.rs +++ b/compiler/rustc_thread_pool/src/registry.rs @@ -617,7 +617,10 @@ impl Registry { /// Mark a Rayon worker thread as blocked. This triggers the deadlock handler /// if no other worker thread is active #[inline] -pub fn park(mut mutex_guard: parking_lot::MutexGuard<'_, T>, f: impl FnOnce(&mut T) -> R) -> R { +pub fn park( + mut mutex_guard: parking_lot::MutexGuard<'_, T>, + f: impl FnOnce(&mut T) -> R, +) -> R { let worker_thread = WorkerThread::current(); assert!(!worker_thread.is_null()); unsafe {