avoiding concurrent modification of package metadata (#6524)

* avoiding concurrent modification of package metadata

* move checksums closer

Co-authored-by: James <james@conan.io>

conans/client/cmd/uploader.py

@@ -334,7 +334,8 @@ def _upload_package(self, pref, retry=None, retry_wait=None, integrity_check=Fal
         assert (pref.revision is not None), "Cannot upload a package without PREV"
         assert (pref.ref.revision is not None), "Cannot upload a package without RREV"
 
-        conanfile_path = self._cache.package_layout(pref.ref).conanfile()
+        pkg_layout = self._cache.package_layout(pref.ref)
+        conanfile_path = pkg_layout.conanfile()
         self._hook_manager.execute("pre_upload_package", conanfile_path=conanfile_path,
                                    reference=pref.ref,
                                    package_id=pref.id,
@@ -343,9 +344,6 @@ def _upload_package(self, pref, retry=None, retry_wait=None, integrity_check=Fal
         t1 = time.time()
         the_files = self._compress_package_files(pref, integrity_check)
 
-        with self._cache.package_layout(pref.ref).update_metadata() as metadata:
-            metadata.packages[pref.id].checksums = calc_files_checksum(the_files)
-
         if policy == UPLOAD_POLICY_SKIP:
             return None
         files_to_upload, deleted = self._package_files_to_upload(pref, policy, the_files, p_remote)
@@ -364,11 +362,13 @@ def _upload_package(self, pref, retry=None, retry_wait=None, integrity_check=Fal
 
         logger.debug("UPLOAD: Time uploader upload_package: %f" % (time.time() - t1))
 
-        metadata = self._cache.package_layout(pref.ref).load_metadata()
-        cur_package_remote = metadata.packages[pref.id].remote
-        if not cur_package_remote and policy != UPLOAD_POLICY_SKIP:
-            with self._cache.package_layout(pref.ref).update_metadata() as metadata:
+        # Update the package metadata
+        checksums = calc_files_checksum(the_files)
+        with pkg_layout.update_metadata() as metadata:
+            cur_package_remote = metadata.packages[pref.id].remote
+            if not cur_package_remote:
                 metadata.packages[pref.id].remote = p_remote.name
+            metadata.packages[pref.id].checksums = checksums
 
         return pref
 

conans/paths/package_layouts/package_cache_layout.py

@@ -2,6 +2,7 @@
 
 import os
 import platform
+import threading
 from contextlib import contextmanager
 
 
@@ -170,16 +171,25 @@ def load_metadata(self):
             raise RecipeNotFoundException(self._ref)
         return PackageMetadata.loads(text)
 
+    _metadata_locks = {}  # Needs to be shared among all instances
+
     @contextmanager
     def update_metadata(self):
-        lockfile = self.package_metadata() + ".lock"
+        metadata_path = self.package_metadata()
+        lockfile = metadata_path + ".lock"
         with fasteners.InterProcessLock(lockfile, logger=logger):
+            lock_name = self.package_metadata()  # The path is the thing that defines mutex
+            thread_lock = PackageCacheLayout._metadata_locks.setdefault(lock_name, threading.Lock())
+            thread_lock.acquire()
             try:
-                metadata = self.load_metadata()
-            except RecipeNotFoundException:
-                metadata = PackageMetadata()
-            yield metadata
-            save(self.package_metadata(), metadata.dumps())
+                try:
+                    metadata = self.load_metadata()
+                except RecipeNotFoundException:
+                    metadata = PackageMetadata()
+                yield metadata
+                save(metadata_path, metadata.dumps())
+            finally:
+                thread_lock.release()
 
     # Locks
     def conanfile_read_lock(self, output):