Fixed GitLab concurrency issue.

Author: AlexeyKuznetsov-DD

build-logic/smoke-test/src/main/kotlin/datadog/buildlogic/smoketest/GradleDistributionCache.kt

@@ -8,8 +8,8 @@ import java.io.RandomAccessFile
 import java.net.HttpURLConnection
 import java.net.URI
 import java.net.URLConnection
+import java.nio.channels.OverlappingFileLockException
 import java.security.MessageDigest
-import java.util.concurrent.ConcurrentHashMap
 import java.util.zip.ZipInputStream
 
 /** Nested distributions, under `wrapper/` so CI restores them from its existing cache. */
@@ -23,9 +23,7 @@ private val LAUNCHER_JAR = Regex("gradle-launcher-.*\\.jar")
 
 private const val CONNECT_TIMEOUT_MS = 30_000
 private const val READ_TIMEOUT_MS = 120_000
-
-/** Serializes same-JVM callers before they acquire the cross-process file lock. */
-private val installMonitors = ConcurrentHashMap<String, Any>()
+private const val LOCK_RETRY_DELAY_MS = 100L
 
 /** Provisions [gradleVersion] once in [cacheDir], trying [distributionUris] in order. */
 internal fun provisionGradleDistribution(
@@ -42,22 +40,35 @@ internal fun provisionGradleDistribution(
 
   installedDistributionRoot(installDir, gradleVersion, marker)?.let { return it }
 
-  val monitor = installMonitors.computeIfAbsent(installDir.absolutePath) { Any() }
-  synchronized(monitor) {
-    installedDistributionRoot(installDir, gradleVersion, marker)?.let { return it }
-    if (!cacheDir.isDirectory && !cacheDir.mkdirs()) {
-      throw GradleException("Could not create Gradle distribution cache: ${cacheDir.absolutePath}")
-    }
-    RandomAccessFile(File(cacheDir, "gradle-$gradleVersion-bin.lock"), "rw").use { lockFile ->
-      lockFile.channel.lock().use {
-        // Recheck after waiting for the file lock.
-        installedDistributionRoot(installDir, gradleVersion, marker)?.let { return it }
-        install(installDir, marker, gradleVersion, distributionUris, logger)
-        return installedDistributionRoot(installDir, gradleVersion, marker)
-          ?: throw GradleException(
-            "Gradle $gradleVersion was unpacked into ${installDir.absolutePath} but no " +
-              "distribution root could be found in it",
-          )
+  if (!cacheDir.isDirectory && !cacheDir.mkdirs() && !cacheDir.isDirectory) {
+    throw GradleException("Could not create Gradle distribution cache: ${cacheDir.absolutePath}")
+  }
+  return withDistributionFileLock(File(cacheDir, "gradle-$gradleVersion-bin.lock")) {
+    // Recheck after waiting for the file lock.
+    installedDistributionRoot(installDir, gradleVersion, marker)
+      ?.let { return@withDistributionFileLock it }
+    install(installDir, marker, gradleVersion, distributionUris, logger)
+    installedDistributionRoot(installDir, gradleVersion, marker)
+      ?: throw GradleException(
+        "Gradle $gradleVersion was unpacked into ${installDir.absolutePath} but no " +
+          "distribution root could be found in it",
+      )
+  }
+}
+
+private fun <T> withDistributionFileLock(lockPath: File, action: () -> T): T {
+  RandomAccessFile(lockPath, "rw").use { lockFile ->
+    while (true) {
+      val lock =
+        try {
+          lockFile.channel.lock()
+        } catch (_: OverlappingFileLockException) {
+          // Another caller in this JVM holds the lock.
+          Thread.sleep(LOCK_RETRY_DELAY_MS)
+          continue
+        }
+      lock.use {
+        return action()
       }
     }
   }

build-logic/smoke-test/src/test/kotlin/datadog/buildlogic/smoketest/GradleDistributionCacheTest.kt

@@ -8,11 +8,14 @@ import org.gradle.api.logging.Logging
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.io.TempDir
 import java.io.File
+import java.io.RandomAccessFile
 import java.net.InetSocketAddress
 import java.net.URI
 import java.nio.file.Path
 import java.security.MessageDigest
+import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.CyclicBarrier
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
@@ -192,6 +195,42 @@ class GradleDistributionCacheTest {
     }
   }
 
+  @Test
+  fun `waits when the distribution lock is already held in this JVM`() {
+    val distribution = writeDistributionArchive("overlapping-lock.zip", VERSION)
+    assertThat(cacheDir.mkdirs()).isTrue()
+    val started = CountDownLatch(1)
+    val result = CompletableFuture<File>()
+    val worker = Thread {
+      started.countDown()
+      try {
+        result.complete(
+          provisionGradleDistribution(cacheDir, VERSION, listOf(distribution), logger),
+        )
+      } catch (e: Throwable) {
+        result.completeExceptionally(e)
+      }
+    }
+
+    val root =
+      try {
+        RandomAccessFile(File(cacheDir, "gradle-$VERSION-bin.lock"), "rw").use { lockFile ->
+          lockFile.channel.lock().use {
+            worker.start()
+            assertThat(started.await(10, TimeUnit.SECONDS)).isTrue()
+            Thread.sleep(200)
+            assertThat(result.isDone).isFalse()
+          }
+        }
+        result.get(30, TimeUnit.SECONDS)
+      } finally {
+        worker.join(TimeUnit.SECONDS.toMillis(30))
+      }
+
+    assertThat(root).isDirectory()
+    assertThat(File(root, "lib/gradle-launcher-$VERSION.jar")).exists()
+  }
+
   @Test
   fun `requires at least one source`() {
     assertThatThrownBy {