Make registry locking more coarse

This commit updates the locking strategy in Cargo handle the recent
addition of creating a cache of the on-disk index also on disk. The goal
here is reduce the overhead of locking both cognitively when reading but
also performance wise by requiring fewer locks. Previously Cargo had a
bunch of fine-grained locks throughout the index and git repositories,
but after this commit there's just one global "package cache" lock.

This global lock now serves to basically synchronize the entire crate
graph resolution step. This shouldn't really take that long unless it's
downloading, in which case there's not a ton of benefit to running in
parallel anyway. The other intention of this single global lock is to
make it much easier on the sources to not worry so much about lock
ordering or when to acquire locks, but rather they just assert in their
various operations that they're locked.

Cargo now has a few coarse-grained locations where locks are held (for
example during resolution and during package downloading). These locks
are a bit sprinkled about but they have in-code asserts which assert
that they're held, so we'll find bugs quickly if any lock isn't held
(before a race condition is hit that is)

Author: alexcrichton

src/cargo/core/package.rs

@@ -25,6 +25,7 @@ use crate::ops;
 use crate::util::errors::{CargoResult, CargoResultExt, HttpNot200};
 use crate::util::network::Retry;
 use crate::util::{self, internal, lev_distance, Config, Progress, ProgressStyle};
+use crate::util::config::PackageCacheLock;
 
 /// Information about a package that is available somewhere in the file system.
 ///
@@ -339,6 +340,9 @@ pub struct Downloads<'a, 'cfg: 'a> {
     /// trigger a timeout; reset `next_speed_check` and set this back to the
     /// configured threshold.
     next_speed_check_bytes_threshold: Cell<u64>,
+    /// Global filesystem lock to ensure only one Cargo is downloading at a
+    /// time.
+    _lock: PackageCacheLock<'cfg>,
 }
 
 struct Download<'cfg> {
@@ -437,6 +441,7 @@ 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()?,
         })
     }
 

src/cargo/ops/cargo_generate_lockfile.rs

@@ -40,6 +40,10 @@ pub fn update_lockfile(ws: &Workspace<'_>, opts: &UpdateOptions<'_>) -> CargoRes
         failure::bail!("you can't update in the offline mode");
     }
 
+    // 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 previous_resolve = match ops::load_pkg_lockfile(ws)? {
         Some(resolve) => resolve,
         None => return generate_lockfile(ws),
@@ -73,6 +77,7 @@ pub fn update_lockfile(ws: &Workspace<'_>, opts: &UpdateOptions<'_>) -> CargoRes
                 });
             }
         }
+
         registry.add_sources(sources)?;
     }
 

src/cargo/ops/cargo_install.rs

@@ -488,6 +488,11 @@ fn check_yanked_install(ws: &Workspace<'_>) -> CargoResult<()> {
     // wouldn't be available for `compile_ws`.
     let (pkg_set, resolve) = ops::resolve_ws_with_method(ws, Method::Everything, &specs)?;
     let mut sources = pkg_set.sources_mut();
+
+    // Checking the yanked status invovles taking a look at the registry and
+    // maybe updating files, so be sure to lock it here.
+    let _lock = ws.config().acquire_package_cache_lock()?;
+
     for pkg_id in resolve.iter() {
         if let Some(source) = sources.get_mut(pkg_id.source_id()) {
             if source.is_yanked(pkg_id)? {

src/cargo/ops/cargo_package.rs

@@ -548,6 +548,10 @@ fn compare_resolve(
 }
 
 fn check_yanked(config: &Config, pkg_set: &PackageSet<'_>, resolve: &Resolve) -> CargoResult<()> {
+    // Checking the yanked status invovles 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 mut sources = pkg_set.sources_mut();
     for pkg_id in resolve.iter() {
         if let Some(source) = sources.get_mut(pkg_id.source_id()) {

src/cargo/ops/common_for_install_and_uninstall.rs

@@ -564,6 +564,11 @@ pub fn select_pkg<'a, T>(
 where
     T: Source + 'a,
 {
+    // This operation may involve updating some sources or making a few queries
+    // which may involve frobbing caches, as a result make sure we synchronize
+    // with other global Cargos
+    let _lock = config.acquire_package_cache_lock()?;
+
     if needs_update {
         source.update()?;
     }

src/cargo/ops/registry.rs

@@ -353,6 +353,7 @@ fn registry(
     let token = token.or(token_config);
     let sid = get_source_id(config, index_config.or(index), registry)?;
     let api_host = {
+        let _lock = config.acquire_package_cache_lock()?;
         let mut src = RegistrySource::remote(sid, &HashSet::new(), config);
         // Only update the index if the config is not available or `force` is set.
         let cfg = src.config();

src/cargo/ops/resolve.rs

@@ -146,6 +146,10 @@ pub fn resolve_with_previous<'cfg>(
     specs: &[PackageIdSpec],
     register_patches: bool,
 ) -> CargoResult<Resolve> {
+    // We only want one Cargo at a time resolving a crate graph since this can
+    // involve a lot of frobbing of the global caches.
+    let _lock = ws.config().acquire_package_cache_lock()?;
+
     // Here we place an artificial limitation that all non-registry sources
     // cannot be locked at more than one revision. This means that if a Git
     // repository provides more than one package, they must all be updated in

src/cargo/sources/git/source.rs

@@ -154,12 +154,9 @@ impl<'cfg> Source for GitSource<'cfg> {
     }
 
     fn update(&mut self) -> CargoResult<()> {
-        let lock =
-            self.config
-                .git_path()
-                .open_rw(".cargo-lock-git", self.config, "the git checkouts")?;
-
-        let db_path = lock.parent().join("db").join(&self.ident);
+        let git_path = self.config.git_path();
+        let git_path = self.config.assert_package_cache_locked(&git_path);
+        let db_path = git_path.join("db").join(&self.ident);
 
         if self.config.cli_unstable().offline && !db_path.exists() {
             failure::bail!(
@@ -189,21 +186,17 @@ impl<'cfg> Source for GitSource<'cfg> {
             (self.remote.db_at(&db_path)?, actual_rev.unwrap())
         };
 
-        // Don’t use the full hash, in order to contribute less to reaching the path length limit
-        // on Windows. See <https://github.com/servo/servo/pull/14397>.
+        // Don’t use the full hash, in order to contribute less to reaching the
+        // path length limit on Windows. See
+        // <https://github.com/servo/servo/pull/14397>.
         let short_id = db.to_short_id(&actual_rev).unwrap();
 
-        let checkout_path = lock
-            .parent()
+        let checkout_path = git_path
             .join("checkouts")
             .join(&self.ident)
             .join(short_id.as_str());
 
-        // Copy the database to the checkout location. After this we could drop
-        // the lock on the database as we no longer needed it, but we leave it
-        // in scope so the destructors here won't tamper with too much.
-        // Checkout is immutable, so we don't need to protect it with a lock once
-        // it is created.
+        // Copy the database to the checkout location.
         db.copy_to(actual_rev.clone(), &checkout_path, self.config)?;
 
         let source_id = self.source_id.with_precise(Some(actual_rev.to_string()));

src/cargo/sources/registry/index.rs

@@ -78,8 +78,7 @@ use semver::{Version, VersionReq};
 
 use crate::core::dependency::Dependency;
 use crate::core::{InternedString, PackageId, SourceId, Summary};
-use crate::sources::registry::RegistryData;
-use crate::sources::registry::{RegistryPackage, INDEX_LOCK};
+use crate::sources::registry::{RegistryData, RegistryPackage};
 use crate::util::{internal, CargoResult, Config, Filesystem, ToSemver};
 
 /// Crates.io treats hyphen and underscores as interchangeable, but the index and old Cargo do not.
@@ -172,7 +171,6 @@ pub struct RegistryIndex<'cfg> {
     path: Filesystem,
     summaries_cache: HashMap<InternedString, Summaries>,
     config: &'cfg Config,
-    locked: bool,
 }
 
 /// An internal cache of summaries for a particular package.
@@ -237,14 +235,12 @@ impl<'cfg> RegistryIndex<'cfg> {
         source_id: SourceId,
         path: &Filesystem,
         config: &'cfg Config,
-        locked: bool,
     ) -> RegistryIndex<'cfg> {
         RegistryIndex {
             source_id,
             path: path.clone(),
             summaries_cache: HashMap::new(),
             config,
-            locked,
         }
     }
 
@@ -314,35 +310,19 @@ impl<'cfg> RegistryIndex<'cfg> {
         }
 
         // Prepare the `RegistryData` which will lazily initialize internal data
-        // structures. Note that this is also importantly needed to initialize
-        // to avoid deadlocks where we acquire a lock below but the `load`
-        // function inside *also* wants to acquire a lock. See an instance of
-        // this on #5551.
+        // structures.
         load.prepare()?;
 
-        // Synchronize access to the index. For remote indices we want to make
-        // sure that while we're reading the index no one is trying to update
-        // it.
-        let (root, lock) = if self.locked {
-            let lock = self
-                .path
-                .open_ro(Path::new(INDEX_LOCK), self.config, "the registry index");
-            match lock {
-                Ok(lock) => (lock.path().parent().unwrap().to_path_buf(), Some(lock)),
-                Err(_) => {
-                    self.summaries_cache.insert(name, Summaries::default());
-                    return Ok(self.summaries_cache.get_mut(&name).unwrap());
-                }
-            }
-        } else {
-            (self.path.clone().into_path_unlocked(), None)
-        };
+        // let root = self.config.assert_package_cache_locked(&self.path);
+        let root = load.assert_index_locked(&self.path);
         let cache_root = root.join(".cache");
+
         // TODO: comment
-        let lock_mtime = lock
-            .as_ref()
-            .and_then(|l| l.file().metadata().ok())
-            .map(|t| FileTime::from_last_modification_time(&t));
+        let lock_mtime = None;
+        // let lock_mtime = lock
+        //     .as_ref()
+        //     .and_then(|l| l.file().metadata().ok())
+        //     .map(|t| FileTime::from_last_modification_time(&t));
 
 
         // See module comment in `registry/mod.rs` for why this is structured
@@ -371,6 +351,7 @@ impl<'cfg> RegistryIndex<'cfg> {
                 path.as_ref(),
                 self.source_id,
                 load,
+                self.config,
             )?;
             if let Some(summaries) = summaries {
                 self.summaries_cache.insert(name, summaries);
@@ -503,6 +484,7 @@ impl Summaries {
         relative: &Path,
         source_id: SourceId,
         load: &mut dyn RegistryData,
+        config: &Config,
     ) -> CargoResult<Option<Summaries>> {
         // First up, attempt to load the cache. This could fail for all manner
         // of reasons, but consider all of them non-fatal and just log their
@@ -579,7 +561,8 @@ impl Summaries {
         // This is opportunistic so we ignore failure here but are sure to log
         // something in case of error.
         if fs::create_dir_all(cache_path.parent().unwrap()).is_ok() {
-            // TODO: somehow need to globally synchronize this
+            let path = Filesystem::new(cache_path.clone());
+            config.assert_package_cache_locked(&path);
             if let Err(e) = fs::write(cache_path, cache_bytes) {
                 log::info!("failed to write cache: {}", e);
             }

src/cargo/sources/registry/local.rs

@@ -1,12 +1,13 @@
-use std::io::prelude::*;
+use std::fs::File;
 use std::io::SeekFrom;
+use std::io::prelude::*;
 use std::path::Path;
 
 use crate::core::PackageId;
 use crate::sources::registry::{MaybeLock, RegistryConfig, RegistryData};
 use crate::util::errors::{CargoResult, CargoResultExt};
 use crate::util::paths;
-use crate::util::{Config, FileLock, Filesystem, Sha256};
+use crate::util::{Config, Filesystem, Sha256};
 use hex;
 
 pub struct LocalRegistry<'cfg> {
@@ -36,6 +37,12 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
         &self.index_path
     }
 
+    fn assert_index_locked<'a>(&self, path: &'a Filesystem) -> &'a Path {
+        // Note that the `*_unlocked` variant is used here since we're not
+        // modifying the index and it's required to be externally synchronized.
+        path.as_path_unlocked()
+    }
+
     fn load(
         &self,
         root: &Path,
@@ -71,7 +78,12 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
 
     fn download(&mut self, pkg: PackageId, checksum: &str) -> CargoResult<MaybeLock> {
         let crate_file = format!("{}-{}.crate", pkg.name(), pkg.version());
-        let mut crate_file = self.root.open_ro(&crate_file, self.config, "crate file")?;
+
+        // Note that the usage of `into_path_unlocked` here is because the local
+        // crate files here never change in that we're not the one writing them,
+        // so it's not our responsibility to synchronize access to them.
+        let path = self.root.join(&crate_file).into_path_unlocked();
+        let mut crate_file = File::open(&path)?;
 
         // If we've already got an unpacked version of this crate, then skip the
         // checksum below as it is in theory already verified.
@@ -89,7 +101,7 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
         loop {
             let n = crate_file
                 .read(&mut buf)
-                .chain_err(|| format!("failed to read `{}`", crate_file.path().display()))?;
+                .chain_err(|| format!("failed to read `{}`", path.display()))?;
             if n == 0 {
                 break;
             }
@@ -109,7 +121,7 @@ impl<'cfg> RegistryData for LocalRegistry<'cfg> {
         _pkg: PackageId,
         _checksum: &str,
         _data: &[u8],
-    ) -> CargoResult<FileLock> {
+    ) -> CargoResult<File> {
         panic!("this source doesn't download")
     }
 }