Auto merge of #12706 - ehuss:cache-lock-mode, r=epage

Add new package cache lock modes

The way locking worked before this PR is that only one cargo could write to the package cache at once (otherwise it could cause corruption). However, it allowed cargo's to read from the package cache while running a build under the assumption that writers are append-only and won't affect reading. This allows multiple builds to run concurrently, only blocking on the part where it is not possible to run concurrently (downloading to the cache).

This introduces a new package cache locking strategy to support the ability to safely modify existing cache entries while other cargos are potentially reading from the cache. It has different locking modes:

- `MutateExclusive` (new) — Held when cargo wants to modify existing cache entries (such as being introduced for garbage collection in #12634), and ensures only one cargo has access to the cache during that time.
- `DownloadExclusive`  (renamed) — This is a more specialized name for the lock that was before this PR.   A caller should acquire this when downloading into the cache and doing resolution.  It ensures that only one cargo can append to the cache, but allows other cargos to concurrently read from the cache.
- `Shared` (new) — This is to preserve the old concurrent build behavior by allowing multiple concurrent cargos to hold this while a build is running when it is reading from the cache

**Reviewing suggestions:**
There are a few commits needed to help with testing which are first. The main commit has the following:
- `src/cargo/util/cache_lock.rs` is an abstraction around package cache locks, and is the heart of the change. It should have comments and notes which should guide what it is doing. The `CacheLocker` is stored in `Config` along with all our other global stuff.
- Every call to `config.acquire_package_cache_lock()` has been changed to explicitly state which lock mode it wants to lock the package cache in.
- `Context::compile` is the key point where the `Shared` lock is acquired, ensuring that no mutation is done while the cache is being read.
- `MutateExclusive` is not used in this PR, but is being added in preparation for #12634.
- The non-blocking `try_acquire_package_cache_lock` API is not used in this PR, but is being added in preparation for #12634 to allow automatic gc to skip running if another cargo is already running (to avoid unnecessary blocking).
- `src/cargo/util/flock.rs` has been updated with some code cleanup (removing unused stuff), adds support for non-blocking locks, and renames some functions to make their operation clearer.
- `tests/testsuite/cache_lock.rs` contains tests for all the different permutations of ways of acquiring locks.

Author: bors

crates/cargo-test-support/src/compare.rs

@@ -236,6 +236,7 @@ fn substitute_macros(input: &str) -> String {
         ("[SKIPPING]", "    Skipping"),
         ("[WAITING]", "     Waiting"),
         ("[PUBLISHED]", "   Published"),
+        ("[BLOCKING]", "    Blocking"),
     ];
     let mut result = input.to_owned();
     for &(pat, subst) in &macros {

crates/cargo-test-support/src/lib.rs

@@ -13,6 +13,7 @@ use std::path::{Path, PathBuf};
 use std::process::{Command, Output};
 use std::str;
 use std::sync::OnceLock;
+use std::thread::JoinHandle;
 use std::time::{self, Duration};
 
 use anyhow::{bail, Result};
@@ -1470,3 +1471,50 @@ pub fn symlink_supported() -> bool {
 pub fn no_such_file_err_msg() -> String {
     std::io::Error::from_raw_os_error(2).to_string()
 }
+
+/// Helper to retry a function `n` times.
+///
+/// The function should return `Some` when it is ready.
+pub fn retry<F, R>(n: u32, mut f: F) -> R
+where
+    F: FnMut() -> Option<R>,
+{
+    let mut count = 0;
+    let start = std::time::Instant::now();
+    loop {
+        if let Some(r) = f() {
+            return r;
+        }
+        count += 1;
+        if count > n {
+            panic!(
+                "test did not finish within {n} attempts ({:?} total)",
+                start.elapsed()
+            );
+        }
+        sleep_ms(100);
+    }
+}
+
+#[test]
+#[should_panic(expected = "test did not finish")]
+fn retry_fails() {
+    retry(2, || None::<()>);
+}
+
+/// Helper that waits for a thread to finish, up to `n` tenths of a second.
+pub fn thread_wait_timeout<T>(n: u32, thread: JoinHandle<T>) -> T {
+    retry(n, || thread.is_finished().then_some(()));
+    thread.join().unwrap()
+}
+
+/// Helper that runs some function, and waits up to `n` tenths of a second for
+/// it to finish.
+pub fn threaded_timeout<F, R>(n: u32, f: F) -> R
+where
+    F: FnOnce() -> R + Send + 'static,
+    R: Send + 'static,
+{
+    let thread = std::thread::spawn(|| f());
+    thread_wait_timeout(n, thread)
+}

crates/xtask-bump-check/src/xtask.rs

@@ -22,6 +22,7 @@ use cargo::core::Registry;
 use cargo::core::SourceId;
 use cargo::core::Workspace;
 use cargo::sources::source::QueryKind;
+use cargo::util::cache_lock::CacheLockMode;
 use cargo::util::command_prelude::*;
 use cargo::util::ToSemver;
 use cargo::CargoResult;
@@ -347,7 +348,7 @@ fn check_crates_io<'a>(
 ) -> CargoResult<()> {
     let source_id = SourceId::crates_io(config)?;
     let mut registry = PackageRegistry::new(config)?;
-    let _lock = config.acquire_package_cache_lock()?;
+    let _lock = config.acquire_package_cache_lock(CacheLockMode::DownloadExclusive)?;
     registry.lock_patches();
     config.shell().status(
         STATUS,

src/cargo/core/compiler/context/mod.rs

@@ -7,6 +7,7 @@ use std::sync::{Arc, Mutex};
 use crate::core::compiler::compilation::{self, UnitOutput};
 use crate::core::compiler::{self, artifact, Unit};
 use crate::core::PackageId;
+use crate::util::cache_lock::CacheLockMode;
 use crate::util::errors::CargoResult;
 use crate::util::profile;
 use anyhow::{bail, Context as _};
@@ -132,6 +133,13 @@ impl<'a, 'cfg> Context<'a, 'cfg> {
     ///
     /// [`ops::cargo_compile`]: ../../../ops/cargo_compile/index.html
     pub fn compile(mut self, exec: &Arc<dyn Executor>) -> CargoResult<Compilation<'cfg>> {
+        // A shared lock is held during the duration of the build since rustc
+        // needs to read from the `src` cache, and we don't want other
+        // commands modifying the `src` cache while it is running.
+        let _lock = self
+            .bcx
+            .config
+            .acquire_package_cache_lock(CacheLockMode::Shared)?;
         let mut queue = JobQueue::new(self.bcx);
         let mut plan = BuildPlan::new();
         let build_plan = self.bcx.build_config.build_plan;

src/cargo/core/compiler/future_incompat.rs

@@ -37,6 +37,7 @@ use crate::core::compiler::BuildContext;
 use crate::core::{Dependency, PackageId, Workspace};
 use crate::sources::source::QueryKind;
 use crate::sources::SourceConfigMap;
+use crate::util::cache_lock::CacheLockMode;
 use crate::util::{iter_join, CargoResult};
 use anyhow::{bail, format_err, Context};
 use serde::{Deserialize, Serialize};
@@ -166,7 +167,7 @@ impl OnDiskReports {
         let on_disk = serde_json::to_vec(&self).unwrap();
         if let Err(e) = ws
             .target_dir()
-            .open_rw(
+            .open_rw_exclusive_create(
                 FUTURE_INCOMPAT_FILE,
                 ws.config(),
                 "Future incompatibility report",
@@ -190,7 +191,7 @@ impl OnDiskReports {
 
     /// Loads the on-disk reports.
     pub fn load(ws: &Workspace<'_>) -> CargoResult<OnDiskReports> {
-        let report_file = match ws.target_dir().open_ro(
+        let report_file = match ws.target_dir().open_ro_shared(
             FUTURE_INCOMPAT_FILE,
             ws.config(),
             "Future incompatible report",
@@ -297,7 +298,10 @@ fn render_report(per_package_reports: &[FutureIncompatReportPackage]) -> BTreeMa
 /// This is best-effort - if an error occurs, `None` will be returned.
 fn get_updates(ws: &Workspace<'_>, package_ids: &BTreeSet<PackageId>) -> Option<String> {
     // This in general ignores all errors since this is opportunistic.
-    let _lock = ws.config().acquire_package_cache_lock().ok()?;
+    let _lock = ws
+        .config()
+        .acquire_package_cache_lock(CacheLockMode::DownloadExclusive)
+        .ok()?;
     // Create a set of updated registry sources.
     let map = SourceConfigMap::new(ws.config()).ok()?;
     let mut package_ids: BTreeSet<_> = package_ids

src/cargo/core/compiler/layout.rs

@@ -166,7 +166,7 @@ impl Layout {
         // For now we don't do any more finer-grained locking on the artifact
         // directory, so just lock the entire thing for the duration of this
         // compile.
-        let lock = dest.open_rw(".cargo-lock", ws.config(), "build directory")?;
+        let lock = dest.open_rw_exclusive_create(".cargo-lock", ws.config(), "build directory")?;
         let root = root.into_path_unlocked();
         let dest = dest.into_path_unlocked();
         let deps = dest.join("deps");

src/cargo/core/package.rs

@@ -24,7 +24,7 @@ use crate::core::resolver::{HasDevUnits, Resolve};
 use crate::core::{Dependency, Manifest, PackageId, SourceId, Target};
 use crate::core::{Summary, Workspace};
 use crate::sources::source::{MaybePackage, SourceMap};
-use crate::util::config::PackageCacheLock;
+use crate::util::cache_lock::{CacheLock, CacheLockMode};
 use crate::util::errors::{CargoResult, HttpNotSuccessful};
 use crate::util::interning::InternedString;
 use crate::util::network::http::http_handle_and_timeout;
@@ -367,7 +367,7 @@ pub struct Downloads<'a, 'cfg> {
     next_speed_check_bytes_threshold: Cell<u64>,
     /// Global filesystem lock to ensure only one Cargo is downloading at a
     /// time.
-    _lock: PackageCacheLock<'cfg>,
+    _lock: CacheLock<'cfg>,
 }
 
 struct Download<'cfg> {
@@ -465,7 +465,9 @@ impl<'cfg> PackageSet<'cfg> {
             timeout,
             next_speed_check: Cell::new(Instant::now()),
             next_speed_check_bytes_threshold: Cell::new(0),
-            _lock: self.config.acquire_package_cache_lock()?,
+            _lock: self
+                .config
+                .acquire_package_cache_lock(CacheLockMode::DownloadExclusive)?,
         })
     }
 
@@ -478,6 +480,9 @@ impl<'cfg> PackageSet<'cfg> {
 
     pub fn get_many(&self, ids: impl IntoIterator<Item = PackageId>) -> CargoResult<Vec<&Package>> {
         let mut pkgs = Vec::new();
+        let _lock = self
+            .config
+            .acquire_package_cache_lock(CacheLockMode::DownloadExclusive)?;
         let mut downloads = self.enable_download()?;
         for id in ids {
             pkgs.extend(downloads.start(id)?);

src/cargo/ops/cargo_add/mod.rs

@@ -23,6 +23,7 @@ use crate::core::Shell;
 use crate::core::Summary;
 use crate::core::Workspace;
 use crate::sources::source::QueryKind;
+use crate::util::cache_lock::CacheLockMode;
 use crate::util::style;
 use crate::util::toml_mut::dependency::Dependency;
 use crate::util::toml_mut::dependency::GitSource;
@@ -77,7 +78,9 @@ pub fn add(workspace: &Workspace<'_>, options: &AddOptions<'_>) -> CargoResult<(
     let mut registry = PackageRegistry::new(options.config)?;
 
     let deps = {
-        let _lock = options.config.acquire_package_cache_lock()?;
+        let _lock = options
+            .config
+            .acquire_package_cache_lock(CacheLockMode::DownloadExclusive)?;
         registry.lock_patches();
         options
             .dependencies

src/cargo/ops/cargo_generate_lockfile.rs

@@ -3,6 +3,7 @@ use crate::core::resolver::features::{CliFeatures, HasDevUnits};
 use crate::core::{PackageId, PackageIdSpec};
 use crate::core::{Resolve, SourceId, Workspace};
 use crate::ops;
+use crate::util::cache_lock::CacheLockMode;
 use crate::util::config::Config;
 use crate::util::style;
 use crate::util::CargoResult;
@@ -48,7 +49,9 @@ pub fn update_lockfile(ws: &Workspace<'_>, opts: &UpdateOptions<'_>) -> CargoRes
 
     // Updates often require a lot of modifications to the registry, so ensure
     // that we're synchronized against other Cargos.
-    let _lock = ws.config().acquire_package_cache_lock()?;
+    let _lock = ws
+        .config()
+        .acquire_package_cache_lock(CacheLockMode::DownloadExclusive)?;
 
     let max_rust_version = ws.rust_version();
 

src/cargo/ops/cargo_package.rs

@@ -13,6 +13,7 @@ use crate::core::{registry::PackageRegistry, resolver::HasDevUnits};
 use crate::core::{Feature, Shell, Verbosity, Workspace};
 use crate::core::{Package, PackageId, PackageSet, Resolve, SourceId};
 use crate::sources::PathSource;
+use crate::util::cache_lock::CacheLockMode;
 use crate::util::config::JobsConfig;
 use crate::util::errors::CargoResult;
 use crate::util::toml::TomlManifest;
@@ -132,7 +133,7 @@ pub fn package_one(
     let dir = ws.target_dir().join("package");
     let mut dst = {
         let tmp = format!(".{}", filename);
-        dir.open_rw(&tmp, config, "package scratch space")?
+        dir.open_rw_exclusive_create(&tmp, config, "package scratch space")?
     };
 
     // Package up and test a temporary tarball and only move it to the final
@@ -806,7 +807,7 @@ pub fn check_yanked(
 ) -> CargoResult<()> {
     // Checking the yanked status involves taking a look at the registry and
     // maybe updating files, so be sure to lock it here.
-    let _lock = config.acquire_package_cache_lock()?;
+    let _lock = config.acquire_package_cache_lock(CacheLockMode::DownloadExclusive)?;
 
     let mut sources = pkg_set.sources_mut();
     let mut pending: Vec<PackageId> = resolve.iter().collect();