Make GlobalContext Sync (#15967)

### What does this PR try to resolve?

By making `GlobalContext` `Sync`, it will become much easier to
parallelize parts of cargo. A concrete example is #15676.

It was my understanding from
#15934 (comment)
that this would be a welcome change by the cargo team.

#### Overview
In `GlobalContext` and structs used in its fields
- `RefCell`-s were replaced by either an `std::sync::Mutex` or an
`std::sync::RwLock`, depending on API needs
- `LazyCell`-s were replaced by a new `OnceLock` implementation backed
by `std::sync::OnceLock`, emulating unstable features needed by cargo
- Removed `LazyCell`/`OnceLock` from fields where the initialization
function is just a `Mutex<HashMap>::default()`
- added `util::context::tests::sync_context` test that does not compile
if `GlobalContext` is not `Sync`

### How to test and review this PR?

This PR should add no user-facing changes. Tests must pass and
benchmarks must report no changes.

Author: epage

src/cargo/core/registry.rs

@@ -800,11 +800,9 @@ impl<'gctx> Registry for PackageRegistry<'gctx> {
 
     #[tracing::instrument(skip_all)]
     fn block_until_ready(&mut self) -> CargoResult<()> {
-        if cfg!(debug_assertions) {
-            // Force borrow to catch invalid borrows, regardless of which source is used and how it
-            // happens to behave this time
-            self.gctx.shell().verbosity();
-        }
+        // Ensure `shell` is not already in use,
+        // regardless of which source is used and how it happens to behave this time
+        self.gctx.debug_assert_shell_not_borrowed();
         for (source_id, source) in self.sources.sources_mut() {
             source
                 .block_until_ready()

src/cargo/core/shell.rs

@@ -65,7 +65,7 @@ impl Shell {
     }
 
     /// Creates a shell from a plain writable object, with no color, and max verbosity.
-    pub fn from_write(out: Box<dyn Write>) -> Shell {
+    pub fn from_write(out: Box<dyn Write + Send + Sync>) -> Shell {
         Shell {
             output: ShellOut::Write(AutoStream::never(out)), // strip all formatting on write
             verbosity: Verbosity::Verbose,
@@ -432,7 +432,7 @@ impl Default for Shell {
 /// A `Write`able object, either with or without color support
 enum ShellOut {
     /// A plain write object without color support
-    Write(AutoStream<Box<dyn Write>>),
+    Write(AutoStream<Box<dyn Write + Send + Sync>>),
     /// Color-enabled stdio, with information on whether color should be used
     Stream {
         stdout: AutoStream<std::io::Stdout>,

src/cargo/core/workspace.rs

@@ -2037,7 +2037,7 @@ fn find_workspace_root_with_loader(
 ) -> CargoResult<Option<PathBuf>> {
     // Check if there are any workspace roots that have already been found that would work
     {
-        let roots = gctx.ws_roots.borrow();
+        let roots = gctx.ws_roots();
         // Iterate through the manifests parent directories until we find a workspace
         // root. Note we skip the first item since that is just the path itself
         for current in manifest_path.ancestors().skip(1) {

src/cargo/sources/git/utils.rs

@@ -1528,7 +1528,7 @@ fn github_fast_path(
         "https://api.github.com/repos/{}/{}/commits/{}",
         username, repository, github_branch_name,
     );
-    let mut handle = gctx.http()?.borrow_mut();
+    let mut handle = gctx.http()?.lock().unwrap();
     debug!("attempting GitHub fast path for {}", url);
     handle.get(true)?;
     handle.url(&url)?;

src/cargo/util/cache_lock.rs

@@ -91,8 +91,8 @@ use super::FileLock;
 use crate::CargoResult;
 use crate::GlobalContext;
 use anyhow::Context as _;
-use std::cell::RefCell;
 use std::io;
+use std::sync::Mutex;
 
 /// The style of lock to acquire.
 #[derive(Copy, Clone, Debug, PartialEq)]
@@ -435,7 +435,11 @@ pub struct CacheLock<'lock> {
 impl Drop for CacheLock<'_> {
     fn drop(&mut self) {
         use CacheLockMode::*;
-        let mut state = self.locker.state.borrow_mut();
+        let mut state = match self.locker.state.lock() {
+            Ok(result) => result,
+            // we should release the cache even if a thread panicked while holding a lock
+            Err(poison) => poison.into_inner(),
+        };
         match self.mode {
             Shared => {
                 state.mutate_lock.decrement();
@@ -472,24 +476,25 @@ pub struct CacheLocker {
     ///
     /// [`CacheLocker`] uses interior mutability because it is stuffed inside
     /// [`GlobalContext`], which does not allow mutation.
-    state: RefCell<CacheState>,
+    state: Mutex<CacheState>,
 }
 
 impl CacheLocker {
     /// Creates a new `CacheLocker`.
     pub fn new() -> CacheLocker {
         CacheLocker {
-            state: RefCell::new(CacheState {
+            state: CacheState {
                 cache_lock: RecursiveLock::new(CACHE_LOCK_NAME),
                 mutate_lock: RecursiveLock::new(MUTATE_NAME),
-            }),
+            }
+            .into(),
         }
     }
 
     /// Acquires a lock with the given mode, possibly blocking if another
     /// cargo is holding the lock.
     pub fn lock(&self, gctx: &GlobalContext, mode: CacheLockMode) -> CargoResult<CacheLock<'_>> {
-        let mut state = self.state.borrow_mut();
+        let mut state = self.state.lock().unwrap();
         let _ = state.lock(gctx, mode, Blocking)?;
         Ok(CacheLock { mode, locker: self })
     }
@@ -501,7 +506,7 @@ impl CacheLocker {
         gctx: &GlobalContext,
         mode: CacheLockMode,
     ) -> CargoResult<Option<CacheLock<'_>>> {
-        let mut state = self.state.borrow_mut();
+        let mut state = self.state.lock().unwrap();
         if state.lock(gctx, mode, NonBlocking)? == LockAcquired {
             Ok(Some(CacheLock { mode, locker: self }))
         } else {
@@ -519,7 +524,7 @@ impl CacheLocker {
     /// `DownloadExclusive` will return true if a `MutateExclusive` lock is
     /// held since they overlap.
     pub fn is_locked(&self, mode: CacheLockMode) -> bool {
-        let state = self.state.borrow();
+        let state = self.state.lock().unwrap();
         match (
             mode,
             state.cache_lock.count,