Ensure BuildOperationQueue will progress without extra leases (#37629)

blindpirate · web-flow · commit d5cfd079acd2 · 2026-04-23T03:31:12.000Z
diff --git a/subprojects/core/src/main/java/org/gradle/internal/operations/DefaultBuildOperationQueue.java b/subprojects/core/src/main/java/org/gradle/internal/operations/DefaultBuildOperationQueue.java
@@ -242,9 +242,8 @@ public void run() {
         private void runOperations() {
             CurrentBuildOperationRef.instance().with(parent, () -> {
                 try {
-                    T operation;
-                    while ((operation = waitForNextOperation()) != null) {
-                        runBatch(operation);
+                    while (waitForNextOperation()) {
+                        runBatch();
                     }
                 } catch (Throwable t) {
                     addFailure(t);
@@ -254,39 +253,38 @@ private void runOperations() {
             });
         }
 
-        @Nullable
-        private T waitForNextOperation() {
+        private boolean waitForNextOperation() {
             lock.lock();
             try {
                 // If the token was already invalidated (e.g. in runBatch), exit immediately
                 // to avoid becoming a zombie thread stuck in await().
                 if (token != null && !token.isValid()) {
-                    return null;
+                    return false;
                 }
                 while (queueState == QueueState.Working && helper.isQueueEmpty()) {
                     if (helper.isExtraWorker()) {
                         // We should exit, immediately invalidate our token to ensure the count goes down now.
                         invalidateIfNeeded();
-                        return null;
+                        return false;
                     }
                     try {
                         workAvailable.await();
                     } catch (InterruptedException e) {
                         throw UncheckedException.throwAsUncheckedException(e);
                     }
                 }
-                return helper.pollWork();
+                return !helper.isQueueEmpty();
             } finally {
                 lock.unlock();
             }
         }
 
-        private void runBatch(final T firstOperation) {
+        private void runBatch() {
             int operationsExecuted;
             if (context.requiresWorkerLease()) {
-                operationsExecuted = workerLeases.runAsWorkerThread(() -> executePendingWork(firstOperation));
+                operationsExecuted = workerLeases.runAsWorkerThread(this::executePendingWork);
             } else {
-                operationsExecuted = executePendingWork(firstOperation);
+                operationsExecuted = executePendingWork();
             }
 
             // We need to update pending count outside of withLocks() so that we don't have a race
@@ -295,9 +293,9 @@ private void runBatch(final T firstOperation) {
             completeOperations(operationsExecuted);
         }
 
-        private int executePendingWork(T firstOperation) {
+        private int executePendingWork() {
             if (allowAccessToProjectState) {
-                return doRunBatch(firstOperation);
+                return doRunBatch();
             } else {
                 // Disallow this thread from making any changes to the project locks while it is running the work. This implies that this thread will not
                 // block waiting for access to some other project, which means it can proceed even if some other thread is waiting for a project lock it
@@ -309,26 +307,22 @@ private int executePendingWork(T firstOperation) {
                 // constraint and then gradually roll this out to other worker threads, such as task action workers.
                 //
                 // See {@link ProjectLeaseRegistry#whileDisallowingProjectLockChanges} for more details
-                return workerLeases.whileDisallowingProjectLockChanges(() -> doRunBatch(firstOperation));
+                return workerLeases.whileDisallowingProjectLockChanges(this::doRunBatch);
             }
         }
 
         /**
          * Run as much work as possible until the queue is empty or the queue is cancelled.
          * Then, we return and release the worker lease while we wait for more work to be added to the queue.
          */
-        private int doRunBatch(T firstOperation) {
+        private int doRunBatch() {
             int operationCount = 0;
-            T operation = firstOperation;
-            while (operation != null) {
+            while (true) {
                 if (queueState == QueueState.Cancelled) {
-                    // If an operation was pulled from the queue, but the queue was cancelled before this operation could start
-                    // (for instance, because this worker was waiting on a worker lease) discard it without running.
-                    return ++operationCount;
+                    break;
                 }
-                runOperation(operation);
-                operationCount++;
 
+                T operation;
                 lock.lock();
                 try {
                     if (helper.isExtraWorker()) {
@@ -340,6 +334,14 @@ private int doRunBatch(T firstOperation) {
                 } finally {
                     lock.unlock();
                 }
+
+                if (operation == null) {
+                    break;
+                }
+
+                runOperation(operation);
+                operationCount++;
+
             }
             return operationCount;
         }
diff --git a/subprojects/core/src/test/groovy/org/gradle/internal/operations/DefaultBuildOperationQueueTest.groovy b/subprojects/core/src/test/groovy/org/gradle/internal/operations/DefaultBuildOperationQueueTest.groovy
@@ -17,6 +17,7 @@
 package org.gradle.internal.operations
 
 import org.gradle.api.GradleException
+import org.gradle.internal.Factory
 import org.gradle.internal.concurrent.ExecutorPolicy
 import org.gradle.internal.concurrent.ManagedExecutor
 import org.gradle.internal.concurrent.ManagedExecutorImpl
@@ -27,7 +28,9 @@ import org.gradle.internal.work.DefaultWorkerLimits
 import org.gradle.internal.work.ResourceLockStatistics
 import org.gradle.internal.work.WorkerLeaseRegistry
 import org.gradle.internal.work.WorkerLeaseService
+import spock.lang.Issue
 import spock.lang.Specification
+import spock.lang.Timeout
 
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Executors
@@ -275,6 +278,60 @@ class DefaultBuildOperationQueueTest extends Specification {
         5    | 10
     }
 
+    @Issue("https://github.com/gradle/gradle/issues/37613")
+    @Timeout(value = 30, unit = TimeUnit.SECONDS)
+    def "workers do not pull operations without a lease, and main thread can progress the queue"() {
+        given:
+        // Slightly modified from setupQueue to allow certain injection points.
+        def mainThread = Thread.currentThread()
+        def workerAboutToBlockForLease = new CountDownLatch(1)
+        def executedByMain = new AtomicInteger()
+        def executedByOther = new AtomicInteger()
+        def recordingWorker = { TestBuildOperation op ->
+            if (Thread.currentThread() === mainThread) {
+                executedByMain.incrementAndGet()
+            } else {
+                executedByOther.incrementAndGet()
+            }
+            op.run(null)
+        } as BuildOperationQueue.QueueWorker<TestBuildOperation>
+
+        coordinationService = new DefaultResourceLockCoordinationService()
+        workerRegistry = new DefaultWorkerLeaseService(coordinationService, new DefaultWorkerLimits(1), ResourceLockStatistics.NO_OP) {
+            @Override
+            <T> T runAsWorkerThread(Factory<T> action) {
+                if (Thread.currentThread() !== mainThread) {
+                    workerAboutToBlockForLease.countDown()
+                }
+                return super.runAsWorkerThread(action)
+            }
+        }
+        workerRegistry.startProjectExecution(true)
+        // Keep the lease on the main thread so the spawned worker starves on runAsWorkerThread.
+        lease = workerRegistry.startWorker()
+        def executionContext = new BuildOperationExecutionContext(
+            new ManagedExecutorImpl(Executors.newFixedThreadPool(1), new ExecutorPolicy.CatchAndRecordFailures()),
+            1,
+            true
+        )
+        operationQueue = new DefaultBuildOperationQueue(false, workerRegistry, executionContext, recordingWorker, null)
+
+        when:
+        operationQueue.add(new Success())
+
+        and:
+        // Wait until the worker is about to block on the lease.
+        // This ensures that the main thread will be needed to progress the queue.
+        assert workerAboutToBlockForLease.await(10, TimeUnit.SECONDS)
+
+        and:
+        operationQueue.waitForCompletion()
+
+        then:
+        executedByMain.get() + executedByOther.get() == 1
+        executedByMain.get() == 1
+    }
+
     static class SynchronizedBuildOperation extends TestBuildOperation {
         final Runnable operationAction
         final CountDownLatch startedLatch
