From a7c56bd88869c0bedd0f68970db140a015b590ad Mon Sep 17 00:00:00 2001 From: Vadim Petrochenkov Date: Thu, 16 Jul 2026 17:45:14 +0300 Subject: [PATCH 1/3] rustc_codegen_ssa: Do not create jobserver helper threads unless necessary --- compiler/rustc_codegen_ssa/src/back/write.rs | 43 ++++++++++++-------- 1 file changed, 27 insertions(+), 16 deletions(-) diff --git a/compiler/rustc_codegen_ssa/src/back/write.rs b/compiler/rustc_codegen_ssa/src/back/write.rs index 1419fcaf733f3..baacefc3f2d03 100644 --- a/compiler/rustc_codegen_ssa/src/back/write.rs +++ b/compiler/rustc_codegen_ssa/src/back/write.rs @@ -1011,12 +1011,17 @@ fn do_thin_lto( // After we've requested tokens then we'll, when we can, // get tokens on `coordinator_receive` which will // get managed in the main loop below. - let coordinator_send2 = coordinator_send.clone(); - let helper = jobserver::client() - .into_helper_thread(move |token| { - drop(coordinator_send2.send(ThinLtoMessage::Token(token))); - }) - .expect("failed to spawn helper thread"); + // Note that using `jobserver::Proxy` is not necessary here, the code below always acquires + // tokens before releasing them, so we can never accidentally release the last token + // permanently held by rustc process. + let jobserver_helper = cgcx.parallel.then(|| { + let coordinator_send2 = coordinator_send.clone(); + jobserver::client() + .into_helper_thread(move |token| { + drop(coordinator_send2.send(ThinLtoMessage::Token(token))); + }) + .expect("failed to spawn helper thread") + }); let mut work_items = vec![]; @@ -1036,7 +1041,7 @@ fn do_thin_lto( let insertion_index = work_items.binary_search_by_key(&cost, |&(_, cost)| cost).unwrap_or_else(|e| e); work_items.insert(insertion_index, (work, cost)); - if cgcx.parallel { + if let Some(helper) = &jobserver_helper { helper.request_token(); } } @@ -1240,12 +1245,18 @@ fn start_executing_work( // After we've requested tokens then we'll, when we can, // get tokens on `coordinator_receive` which will // get managed in the main loop below. - let coordinator_send2 = coordinator_send.clone(); - let helper = jobserver::client() - .into_helper_thread(move |token| { - drop(coordinator_send2.send(Message::Token::(token))); - }) - .expect("failed to spawn helper thread"); + // Note that using `jobserver::Proxy` is not necessary here, the code below always acquires + // tokens before releasing them, so we can never accidentally release the last token + // permanently held by rustc process. + let parallel = !sess.opts.unstable_opts.no_parallel_backend && backend.supports_parallel(); + let jobserver_helper = parallel.then(|| { + let coordinator_send2 = coordinator_send.clone(); + jobserver::client() + .into_helper_thread(move |token| { + drop(coordinator_send2.send(Message::Token::(token))); + }) + .expect("failed to spawn helper thread") + }); let opt_level = tcx.backend_optimization_level(()); let backend_features = tcx.global_backend_features(()).clone(); @@ -1286,7 +1297,7 @@ fn start_executing_work( target_is_like_gpu: tcx.sess.target.is_like_gpu, split_debuginfo: tcx.sess.split_debuginfo(), split_dwarf_kind: tcx.sess.opts.unstable_opts.split_dwarf_kind, - parallel: backend.supports_parallel() && !sess.opts.unstable_opts.no_parallel_backend, + parallel, pointer_size: tcx.data_layout.pointer_size(), }; @@ -1633,7 +1644,7 @@ fn start_executing_work( }; work_items.insert(insertion_index, (llvm_work_item, cost)); - if cgcx.parallel { + if let Some(helper) = &jobserver_helper { helper.request_token(); } assert_eq!(main_thread_state, MainThreadState::Codegenning); @@ -1716,7 +1727,7 @@ fn start_executing_work( drop(codegen_state); drop(tokens); - drop(helper); + drop(jobserver_helper); assert!(work_items.is_empty()); if !needs_fat_lto.is_empty() { From cc4af8e81be9263b58e8d78192bbe49dd0bd873d Mon Sep 17 00:00:00 2001 From: Vadim Petrochenkov Date: Thu, 16 Jul 2026 23:11:32 +0300 Subject: [PATCH 2/3] rustc_interface: Minor cleanup to jobserver initialization --- compiler/rustc_interface/src/interface.rs | 19 +++++++------------ compiler/rustc_interface/src/tests.rs | 4 +--- 2 files changed, 8 insertions(+), 15 deletions(-) diff --git a/compiler/rustc_interface/src/interface.rs b/compiler/rustc_interface/src/interface.rs index 95aeb1f7a234f..c7ea358f08170 100644 --- a/compiler/rustc_interface/src/interface.rs +++ b/compiler/rustc_interface/src/interface.rs @@ -369,16 +369,6 @@ pub struct Config { pub using_internal_features: &'static std::sync::atomic::AtomicBool, } -/// Initialize jobserver before getting `jobserver::client` and `build_session`. -pub(crate) fn initialize_checked_jobserver(early_dcx: &EarlyDiagCtxt) { - jobserver::initialize_checked(|err| { - early_dcx - .early_struct_warn(err) - .with_note("the build environment is likely misconfigured") - .emit() - }); -} - // JUSTIFICATION: before session exists, only config #[allow(rustc::bad_opt_access)] pub fn run_compiler(config: Config, f: impl FnOnce(&Compiler) -> R + Send) -> R { @@ -389,9 +379,14 @@ pub fn run_compiler(config: Config, f: impl FnOnce(&Compiler) -> R + Se config.opts.unstable_opts.threads.is_some(), ); - // Check jobserver before run_in_thread_pool_with_globals, which call jobserver::acquire_thread + // Initialize jobserver as early as possible. let early_dcx = EarlyDiagCtxt::new(config.opts.error_format); - initialize_checked_jobserver(&early_dcx); + jobserver::initialize_checked(|err| { + early_dcx + .early_struct_warn(err) + .with_note("the build environment is likely misconfigured") + .emit() + }); crate::callbacks::setup_callbacks(); diff --git a/compiler/rustc_interface/src/tests.rs b/compiler/rustc_interface/src/tests.rs index 71d1829f76584..22b643e74e582 100644 --- a/compiler/rustc_interface/src/tests.rs +++ b/compiler/rustc_interface/src/tests.rs @@ -31,15 +31,13 @@ use rustc_target::spec::{ RelocModel, RelroLevel, SanitizerSet, SplitDebuginfo, StackProtector, TlsModel, }; -use crate::interface::{initialize_checked_jobserver, parse_cfg}; +use crate::interface::parse_cfg; fn sess_and_cfg(args: &[&'static str], f: F) where F: FnOnce(Session, Cfg), { let mut early_dcx = EarlyDiagCtxt::new(ErrorOutputType::default()); - initialize_checked_jobserver(&early_dcx); - let matches = optgroups().parse(args).unwrap(); let sessopts = build_session_options(&mut early_dcx, &matches); let target = rustc_session::config::build_target_config( From eb9dd038d72ec3efdcb7aca256bc62663c28eff4 Mon Sep 17 00:00:00 2001 From: Vadim Petrochenkov Date: Thu, 16 Jul 2026 23:12:56 +0300 Subject: [PATCH 3/3] rustc_data_structures: Expand documentation for rustc jobserver APIs --- .../rustc_data_structures/src/jobserver.rs | 106 ++++++++++++------ 1 file changed, 69 insertions(+), 37 deletions(-) diff --git a/compiler/rustc_data_structures/src/jobserver.rs b/compiler/rustc_data_structures/src/jobserver.rs index 3ed1ea7543f40..6f275d106d640 100644 --- a/compiler/rustc_data_structures/src/jobserver.rs +++ b/compiler/rustc_data_structures/src/jobserver.rs @@ -1,19 +1,16 @@ use std::sync::{Arc, LazyLock, OnceLock}; -pub use jobserver_crate::{Acquired, Client, HelperThread}; -use jobserver_crate::{FromEnv, FromEnvErrorKind}; +pub use jobserver_crate::Acquired; +use jobserver_crate::{Client, FromEnv, FromEnvErrorKind, HelperThread}; use parking_lot::{Condvar, Mutex}; -// We can only call `from_env_ext` once per process - -// We stick this in a global because there could be multiple rustc instances -// in this process, and the jobserver is per-process. +// We stick the jobserver client into a global and initialize it once, because there could be +// multiple compiler instances in this process, and the jobserver is per-process. static GLOBAL_CLIENT: LazyLock> = LazyLock::new(|| { - // Note that this is unsafe because it may misinterpret file descriptors - // on Unix as jobserver file descriptors. We hopefully execute this near - // the beginning of the process though to ensure we don't get false - // positives, or in other words we try to execute this before we open - // any file descriptors ourselves. + // Safety: the checked client construction ensures that the jobserver file descriptors + // (if any) are open and valid. We also try to initialize the jobserver as early as possible + // to avoid unrelated file descriptors with matching values becoming open and valid between + // the process start and the jobserver initialization. let FromEnv { client, var } = unsafe { Client::from_env_ext(true) }; let error = match client { @@ -40,14 +37,17 @@ static GLOBAL_CLIENT: LazyLock> = LazyLock::new(|| { )) }); -// Create a new jobserver if there's no inherited one. +// Creates a new jobserver if there's no inherited one. fn default_client() -> Client { // Pick a "reasonable maximum" capping out at 32 // so we don't take everything down by hogging the process run queue. // The fixed number is used to have deterministic compilation across machines. let client = Client::new(32).expect("failed to create jobserver"); - // Acquire a token for the main thread which we can release later + // Acquire the single token that is always held by the rustc process. + // This is an equivalent of the single token held by a higher level build tool while running + // this instance of rustc. This token is never released - if we are here, then rustc owns the + // jobserver, it is teared down when rustc exits, and there's no one to return the token to. client.acquire_raw().ok(); client @@ -55,47 +55,75 @@ fn default_client() -> Client { static GLOBAL_CLIENT_CHECKED: OnceLock = OnceLock::new(); -pub fn initialize_checked(report_warning: impl FnOnce(&'static str)) { +/// Initializes a jobserver client for the current rustc process. +/// If inheriting jobserver from the environment fails for some reason, an new jobserver owned by +/// the current rustc process will be created. If the inheritance failure reason is non-benign, +/// the passed callback will be used to report the error. +pub fn initialize_checked(report: impl FnOnce(&'static str)) { let client_checked = match &*GLOBAL_CLIENT { Ok(client) => client.clone(), Err(e) => { - report_warning(e); + report(e); default_client() } }; GLOBAL_CLIENT_CHECKED.set(client_checked).ok(); } -const ACCESS_ERROR: &str = "jobserver check should have been called earlier"; - +/// Returns the jobserver client previously initialized by `initialize_checked`. +/// +/// # Assumptions about holding jobserver tokens +/// +/// Rustc process must always hold a single token to avoid being permanently starved and blocked. +/// - If the jobserver is inherited from a higher level build tool, the assumption is that the tool +/// will hold the token and not release it until the rustc process exits. +/// - If the jobserver is owned by the current rustc, the token is acquired by `default_client`. +/// +/// To avoid releasing the last token, users of the client returned by this function must ensure +/// that they never release more tokens than was previously explicitly acquired. +/// Example of a sequence that can accidentally release the last token: +/// `release_raw` -> `wait` -> `acquire_raw`. +/// To avoid situations like this use the `jobserver::Proxy` wrapper instead, +/// it will ensure that the last token is never released. pub fn client() -> Client { - GLOBAL_CLIENT_CHECKED.get().expect(ACCESS_ERROR).clone() + GLOBAL_CLIENT_CHECKED.get().expect("uninitialized jobserver client").clone() } struct ProxyData { - /// The number of tokens assigned to threads. - /// If this is 0, a single token is still assigned to this process, but is unused. + /// The number of tokens assigned to actively working threads, + /// possibly including the single permanently held token. + /// If this number is 0, the single token is still held by the process, + /// but is not currently used for active CPU work. + /// This can happen, for example, if the main thread is waiting for something, + /// in that case some other thread can start using this token to do work. used: u16, - - /// The number of threads requesting a token + /// The number of threads currently requesting a token and waiting. + /// If the proxy releases a token it can immediately give it to one of these threads + /// without going through the real jobserver. pending: u16, } -/// This is a jobserver proxy used to ensure that we hold on to at least one token. +/// A wrapper around jobserver client used for two purposes: +/// - Ensuring that the single token that must be permanently held by the rustc process +/// cannot be accidentally released. +/// - "Token buffering", immediately acquiring freshly released tokens if necessary, +/// without going through the real jobserver. pub struct Proxy { + /// The wrapped jobserver client. client: Client, + /// Helper thread associated with the wrapped client. + helper: OnceLock, + /// The proxy's own data. data: Mutex, - - /// Threads which are waiting on a token will wait on this. + /// Threads which are currently waiting for a token will wait on this condvar. wake_pending: Condvar, - - helper: OnceLock, } impl Proxy { pub fn new() -> Arc { let proxy = Arc::new(Proxy { client: client(), + // Assume that the main thread is actively doing work when it creates the proxy. data: Mutex::new(ProxyData { used: 1, pending: 0 }), wake_pending: Condvar::new(), helper: OnceLock::new(), @@ -105,56 +133,60 @@ impl Proxy { .client .clone() .into_helper_thread(move |token| { + // Reminder: this callback runs when the helper acquires a token. if let Ok(token) = token { let mut data = proxy_.data.lock(); if data.pending > 0 { - // Give the token to a waiting thread + // The token is still needed, give it to one of the waiting threads. token.drop_without_releasing(); assert!(data.used > 0); data.used += 1; data.pending -= 1; proxy_.wake_pending.notify_one(); } else { - // The token is no longer needed, drop it. + // The token is no longer needed, release it by dropping. drop(data); drop(token); } } }) - .expect("failed to create helper thread"); + .expect("failed to spawn helper thread"); proxy.helper.set(helper).unwrap(); proxy } + /// Acquires a token, possibly using some buffered tokens as an optimization. + /// May block and wait until the token is available. pub fn acquire_thread(&self) { let mut data = self.data.lock(); if data.used == 0 { - // There was a free token around. This can - // happen when all threads release their token. + // No threads are doing any active work, but we are still holding the last token. + // Give that token to the current thread. assert_eq!(data.pending, 0); data.used += 1; } else { - // Request a token from the helper thread. We can't directly use `acquire_raw` - // as we also need to be able to wait for the final token in the process which - // does not get a corresponding `release_raw` call. + // Request a token from the helper thread, this is a non-blocking operation. + // Then wait until this or some other request succeeds. self.helper.get().unwrap().request_token(); data.pending += 1; self.wake_pending.wait(&mut data); } } + /// Releases a token, possibly immediately giving it to some other thread as an optimization. + /// Makes sure that the last token is never actually released to the wrapped jobserver. pub fn release_thread(&self) { let mut data = self.data.lock(); if data.pending > 0 { - // Give the token to a waiting thread + // Immediately give the released token to one of the waiting threads. data.pending -= 1; self.wake_pending.notify_one(); } else { data.used -= 1; - // Release the token unless it's the last one in the process + // Release the token to the wrapped jobserver, unless it's the last one in the process. if data.used > 0 { drop(data); self.client.release_raw().ok();