rustc_data_structures: Expand documentation for rustc jobserver APIs

compiler/rustc_codegen_ssa/src/back/write.rs

@@ -1011,12 +1011,17 @@ fn do_thin_lto<B: WriteBackendMethods>(
     // 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<B: WriteBackendMethods>(
         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<B: WriteBackendMethods>(
     // 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::<B>(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::<B>(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<B: WriteBackendMethods>(
         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<B: WriteBackendMethods>(
                     };
                     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<B: WriteBackendMethods>(
 
         drop(codegen_state);
         drop(tokens);
-        drop(helper);
+        drop(jobserver_helper);
         assert!(work_items.is_empty());
 
         if !needs_fat_lto.is_empty() {

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<Result<Client, String>> = 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,62 +37,93 @@ static GLOBAL_CLIENT: LazyLock<Result<Client, String>> = 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
 }
 
 static GLOBAL_CLIENT_CHECKED: OnceLock<Client> = 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<HelperThread>,
+    /// The proxy's own data.
     data: Mutex<ProxyData>,
-
-    /// 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<HelperThread>,
 }
 
 impl Proxy {
     pub fn new() -> Arc<Self> {
         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();

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<R: Send>(config: Config, f: impl FnOnce(&Compiler) -> R + Send) -> R {
@@ -389,9 +379,14 @@ pub fn run_compiler<R: Send>(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();
 

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<F>(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(