rustc: Support --jobs options for limiting parallelism in various parts of the compiler

compiler/rustc_codegen_ssa/src/back/link.rs

@@ -16,6 +16,7 @@ use regex::Regex;
 use rustc_arena::TypedArena;
 use rustc_attr_parsing::eval_config_entry;
 use rustc_data_structures::fx::{FxHashSet, FxIndexSet};
+use rustc_data_structures::jobserver;
 use rustc_data_structures::memmap::Mmap;
 use rustc_data_structures::temp_dir::MaybeTempDir;
 use rustc_errors::DiagCtxtHandle;
@@ -36,8 +37,8 @@ use rustc_middle::middle::debugger_visualizer::DebuggerVisualizerFile;
 use rustc_middle::middle::dependency_format::Linkage;
 use rustc_middle::middle::exported_symbols::SymbolExportKind;
 use rustc_session::config::{
-    self, CFGuard, CrateType, DebugInfo, InstrumentMcount, LinkerFeaturesCli, OutFileName,
-    OutputFilenames, OutputType, PrintKind, SplitDwarfKind, Strip,
+    self, CFGuard, CrateType, DebugInfo, InstrumentMcount, LinkerFeaturesCli, LinkerJobs,
+    OutFileName, OutputFilenames, OutputType, PrintKind, SplitDwarfKind, Strip,
 };
 use rustc_session::lint::builtin::LINKER_MESSAGES;
 use rustc_session::output::{check_file_is_writeable, invalid_output_for_target, out_filename};
@@ -1085,7 +1086,7 @@ fn link_natively(
         should_archive.then(|| tmpdir.join(out_filename.file_name().unwrap()).with_extension("so"));
     let temp_filename = archive_member.as_deref().unwrap_or(out_filename);
 
-    let mut cmd = linker_with_args(
+    let (mut cmd, jobserver_tokens) = linker_with_args(
         &linker_path,
         flavor,
         sess,
@@ -1242,6 +1243,9 @@ fn link_natively(
         break;
     }
 
+    // Finished running linker, release the tokens.
+    drop(jobserver_tokens);
+
     match prog {
         Ok(prog) => {
             if !prog.status.success() {
@@ -2736,7 +2740,7 @@ fn linker_with_args(
     metadata: &EncodedMetadata,
     self_contained_components: LinkSelfContainedComponents,
     codegen_backend: &'static str,
-) -> Command {
+) -> (Command, Vec<jobserver::Acquired>) {
     let self_contained_crt_objects = self_contained_components.is_crt_objects_enabled();
     let cmd = &mut *super::linker::get_linker(
         sess,
@@ -3012,7 +3016,38 @@ fn linker_with_args(
     // to it and remove the option. Currently the last holdout is wasm32-unknown-emscripten.
     add_post_link_args(cmd, sess, flavor);
 
-    cmd.take_cmd()
+    // Only LLD supports controlling parallelism at the moment.
+    let mut tokens = Vec::new();
+    if let LinkerJobs::Explicit(limit) = sess.opts.jobs.linker
+        && flavor.uses_lld()
+    {
+        // Try obtaining as many jobserver tokens as possible (within the limit) to run parallel
+        // linking. One token is available implicitly since we are running on the main thread.
+        let client = jobserver::client();
+
+        let mut unsupported = false;
+        for _ in 0..limit.get() - 1 {
+            match client.try_acquire() {
+                Ok(Some(token)) => tokens.push(token),
+                Ok(None) => {}
+                Err(e) if e.kind() == io::ErrorKind::Unsupported => {
+                    assert!(tokens.is_empty());
+                    unsupported = true;
+                    break;
+                }
+                Err(e) => bug!("IO error when acquiring jobserver token: {e}"),
+            }
+        }
+
+        let prefix = if sess.target.is_like_windows { "/threads:" } else { "--threads=" };
+        // Error on the side of oversubscription if non-blocking token acquiring is unsupported.
+        // Linking is typically the last step in a multi-crate project build,
+        // so the resources should usually be free.
+        let threads = if unsupported { limit.get() } else { 1 + tokens.len() };
+        cmd.link_arg(format!("{prefix}{threads}"));
+    }
+
+    (cmd.take_cmd(), tokens)
 }
 
 fn add_order_independent_options(

compiler/rustc_codegen_ssa/src/back/write.rs

@@ -354,7 +354,7 @@ pub struct CodegenContext {
     pub incr_comp_session_dir: Option<PathBuf>,
     /// `true` if the codegen should be run in parallel.
     ///
-    /// Depends on [`WriteBackendMethods::supports_parallel()`] and `-Zno_parallel_backend`.
+    /// Depends on [`WriteBackendMethods::supports_parallel()`] and `--jobs-backend`.
     pub parallel: bool,
 }
 
@@ -1251,7 +1251,11 @@ fn start_executing_work<B: WriteBackendMethods>(
     // 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();
+    // FIXME: the backend parallelism is currently limited solely by the jobserver,
+    // so if `--jobs-backend` is smaller than `--jobs(-frontend)`, or than the number of tokens
+    // that the external jobserver can give, then it won't be respected.
+    // Below we'll need to add some additional work limiting for `--jobs-backend` to be respected.
+    let parallel = sess.opts.jobs.backend.is_some() && backend.supports_parallel();
     let jobserver_helper = parallel.then(|| {
         let coordinator_send2 = coordinator_send.clone();
         jobserver::client()

compiler/rustc_codegen_ssa/src/base.rs

@@ -807,14 +807,14 @@ pub fn codegen_crate<
     // This likely is a temporary measure. Once we don't have to support the
     // non-parallel compiler anymore, we can compile CGUs end-to-end in
     // parallel and get rid of the complicated scheduling logic.
-    let mut pre_compiled_cgus = if let Some(threads) = tcx.sess.threads() {
+    let mut pre_compiled_cgus = if let Some(threads) = tcx.sess.opts.jobs.frontend {
         tcx.sess.time("compile_first_CGU_batch", || {
             // Try to find one CGU to compile per thread.
             let cgus: Vec<_> = cgu_reuse
                 .iter()
                 .enumerate()
                 .filter(|&(_, reuse)| reuse == &CguReuse::No)
-                .take(threads)
+                .take(threads.get())
                 .collect();
 
             // Compile the found CGUs in parallel.

compiler/rustc_data_structures/src/jobserver.rs

@@ -1,12 +1,10 @@
-use std::sync::{Arc, LazyLock, OnceLock};
+use std::sync::{Arc, OnceLock};
 
 pub use jobserver_crate::Acquired;
 use jobserver_crate::{Client, FromEnv, FromEnvErrorKind, HelperThread};
 use parking_lot::{Condvar, Mutex};
 
-// 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<Result<Client, String>> = LazyLock::new(|| {
+fn create_client(limit: usize) -> Result<Client, String> {
     // 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
@@ -25,7 +23,7 @@ static GLOBAL_CLIENT: LazyLock<Result<Client, String>> = LazyLock::new(|| {
             | FromEnvErrorKind::NegativeFd
             | FromEnvErrorKind::Unsupported
     ) {
-        return Ok(default_client());
+        return Ok(default_client(limit));
     }
 
     // Environment specifies jobserver, but it looks incorrect.
@@ -35,14 +33,11 @@ static GLOBAL_CLIENT: LazyLock<Result<Client, String>> = LazyLock::new(|| {
         "failed to connect to jobserver from environment variable `{name}={:?}`: {error}",
         value
     ))
-});
+}
 
 // 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");
+fn default_client(limit: usize) -> Client {
+    let client = Client::new(limit).expect("failed to create jobserver");
 
     // 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
@@ -53,21 +48,22 @@ fn default_client() -> Client {
     client
 }
 
+// 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_CHECKED: OnceLock<Client> = OnceLock::new();
 
 /// 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(),
+pub fn initialize_checked(limit: usize, report: impl FnOnce(String)) {
+    GLOBAL_CLIENT_CHECKED.get_or_init(|| match create_client(limit) {
+        Ok(client) => client,
         Err(e) => {
             report(e);
-            default_client()
+            default_client(limit)
         }
-    };
-    GLOBAL_CLIENT_CHECKED.set(client_checked).ok();
+    });
 }
 
 /// Returns the jobserver client previously initialized by `initialize_checked`.

compiler/rustc_interface/src/interface.rs

@@ -15,7 +15,7 @@ use rustc_parse::lexer::StripTokens;
 use rustc_parse::new_parser_from_source_str;
 use rustc_parse::parser::Recovery;
 use rustc_query_impl::print_query_stack;
-use rustc_session::config::{self, Cfg, CheckCfg, ExpectedValues, Input, OutFileName};
+use rustc_session::config::{self, BackendJobs, Cfg, CheckCfg, ExpectedValues, Input, OutFileName};
 use rustc_session::parse::ParseSess;
 use rustc_session::{CompilerIO, EarlyDiagCtxt, Session, lint};
 use rustc_span::source_map::{FileLoader, RealFileLoader, SourceMapInputs};
@@ -371,18 +371,20 @@ pub fn run_compiler<R: Send>(config: Config, f: impl FnOnce(&Compiler) -> R + Se
     trace!("run_compiler");
 
     // Set parallel mode before thread pool creation, which will create `Lock`s.
-    rustc_data_structures::sync::set_dyn_thread_safe_mode(
-        config.opts.unstable_opts.threads.is_some(),
-    );
+    rustc_data_structures::sync::set_dyn_thread_safe_mode(config.opts.jobs.frontend.is_some());
 
     // Initialize jobserver as early as possible.
     let early_dcx = EarlyDiagCtxt::new(config.opts.error_format);
-    jobserver::initialize_checked(|err| {
-        early_dcx
-            .early_struct_warn(err)
-            .with_note("the build environment is likely misconfigured")
-            .emit()
-    });
+    if let Some(limit) =
+        config.opts.jobs.frontend.max(config.opts.jobs.backend.map(BackendJobs::value))
+    {
+        jobserver::initialize_checked(limit.get(), |err| {
+            early_dcx
+                .early_struct_warn(err)
+                .with_note("the build environment is likely misconfigured")
+                .emit()
+        });
+    }
 
     crate::callbacks::setup_callbacks();
 
@@ -400,7 +402,7 @@ pub fn run_compiler<R: Send>(config: Config, f: impl FnOnce(&Compiler) -> R + Se
     util::run_in_thread_pool_with_globals(
         &early_dcx,
         config.opts.edition,
-        config.opts.unstable_opts.threads.unwrap_or(1),
+        config.opts.jobs,
         &config.extra_symbols,
         SourceMapInputs { file_loader, path_mapping, hash_kind, checksum_hash_kind },
         |current_gcx| {

compiler/rustc_interface/src/tests.rs

@@ -756,7 +756,7 @@ fn test_unstable_options_tracking_hash() {
     untracked!(span_debug, true);
     untracked!(span_free_formats, true);
     untracked!(temps_dir, Some(String::from("abc")));
-    untracked!(threads, Some(99));
+    untracked!(threads, Some(String::from("99")));
     untracked!(time_llvm_passes, true);
     untracked!(time_passes, true);
     untracked!(time_passes_format, TimePassesFormat::Json);

compiler/rustc_interface/src/util.rs

@@ -1,5 +1,6 @@
 use std::any::Any;
 use std::env::consts::{DLL_PREFIX, DLL_SUFFIX};
+use std::num::NonZero;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, OnceLock};
@@ -21,7 +22,7 @@ use rustc_middle::dep_graph::WorkProductMap;
 use rustc_middle::ty::{CurrentGcx, TyCtxt};
 use rustc_query_impl::{CollectActiveJobsKind, collect_active_query_jobs};
 use rustc_session::config::{
-    Cfg, CrateType, OutFileName, OutputFilenames, OutputTypes, Sysroot, host_tuple,
+    Cfg, CrateType, Jobs, OutFileName, OutputFilenames, OutputTypes, Sysroot, host_tuple,
 };
 use rustc_session::{EarlyDiagCtxt, Session, filesearch};
 use rustc_span::edition::Edition;
@@ -175,7 +176,7 @@ fn run_in_thread_with_globals<F: FnOnce(CurrentGcx) -> R + Send, R: Send>(
 pub(crate) fn run_in_thread_pool_with_globals<F: FnOnce(CurrentGcx) -> R + Send, R: Send>(
     thread_builder_diag: &EarlyDiagCtxt,
     edition: Edition,
-    threads: usize,
+    jobs: Jobs,
     extra_symbols: &[&'static str],
     sm_inputs: SourceMapInputs,
     f: F,
@@ -188,9 +189,11 @@ pub(crate) fn run_in_thread_pool_with_globals<F: FnOnce(CurrentGcx) -> R + Send,
 
     let thread_stack_size = init_stack_size(thread_builder_diag);
 
-    let registry = sync::Registry::new(std::num::NonZero::new(threads).unwrap());
+    let jobs_frontend = jobs.frontend.or(NonZero::new(1)).unwrap();
+    let registry = sync::Registry::new(jobs_frontend);
 
     let Some(proof) = sync::check_dyn_thread_safe() else {
+        assert_eq!(jobs_frontend.get(), 1);
         return run_in_thread_with_globals(
             thread_stack_size,
             edition,
@@ -215,7 +218,7 @@ pub(crate) fn run_in_thread_pool_with_globals<F: FnOnce(CurrentGcx) -> R + Send,
         .thread_name(|_| "rustc".to_string())
         .acquire_thread_handler(move || proxy.acquire_thread())
         .release_thread_handler(move || proxy_.release_thread())
-        .num_threads(threads)
+        .num_threads(jobs_frontend.get())
         .deadlock_handler(move || {
             // On deadlock, creates a new thread and forwards information in thread
             // locals to it. The new thread runs the deadlock handler.
@@ -295,7 +298,7 @@ internal compiler error: query cycle handler thread panicked, aborting process";
                 )
                 .unwrap_or_else(|err| {
                     let mut diag = thread_builder_diag.early_struct_fatal(format!(
-                        "failed to spawn compiler thread pool: could not create {threads} threads ({err})",
+                        "failed to spawn compiler thread pool: could not create {jobs_frontend} threads ({err})",
                     ));
                     diag.help(
                         "try lowering `-Z threads` or checking the operating system's resource limits",

compiler/rustc_metadata/src/rmeta/encoder.rs

@@ -2487,7 +2487,7 @@ pub fn encode_metadata(tcx: TyCtxt<'_>, path: &Path, ref_path: Option<&Path>) {
         return;
     };
 
-    if tcx.sess.threads().is_some() {
+    if tcx.sess.opts.jobs.frontend.is_some() {
         // Prefetch some queries used by metadata encoding.
         // This is not necessary for correctness, but is only done for performance reasons.
         // It can be removed if it turns out to cause trouble or be detrimental to performance.

compiler/rustc_middle/src/dep_graph/graph.rs

@@ -676,7 +676,7 @@ impl DepGraphData {
             let ok = match color {
                 DepNodeColor::Unknown => true,
                 DepNodeColor::Red => false,
-                DepNodeColor::Green(..) => sess.threads().is_some(), // Other threads may mark this green
+                DepNodeColor::Green(..) => sess.opts.jobs.frontend.is_some(), // Other threads may mark this green
             };
             if !ok {
                 panic!("{}", msg())

compiler/rustc_query_impl/src/execution.rs

@@ -300,7 +300,7 @@ fn try_execute_query<'tcx, C: QueryCache, const INCR: bool>(
     // re-executing the query since `try_start` only checks that the query is not currently
     // executing, but another thread may have already completed the query and stores it result
     // in the query cache.
-    if tcx.sess.threads().is_some() {
+    if tcx.sess.opts.jobs.frontend.is_some() {
         if let Some((value, index)) = query.cache.lookup(&key) {
             tcx.prof.query_cache_hit(index.into());
             return (value, Some(index));