Fix infinite spin-wait and missing cancellation propagation in TaskDeduplicator. (#28938)

The test TaskDeduplicatorTest.executeIfNeeded_executeAndCancelLoop_noErrors sporadically hung during the ExecutorService.close() call. This was due to two issues:

1. In executeIfNew, if a thread encountered a RefcountedFuture that was already canceled (refcount = 0), it would call Thread.yield() and continue the while(true) loop. It relied on a listener attached to the canceled future to eventually remove it from the map. However, when using virtual threads, many threads could enter this spin loop simultaneously, saturating the underlying carrier threads. This prevented the listener from being scheduled, creating a deadlock where the spinning threads never saw the entry removed.

2. RefcountedFuture was not propagating the cancel() call to its delegate future. This meant that even if all callers canceled their interest in a task, the task would continue to execute in the background, wasting resources and increasing contention.

This PR makes the following changes:

1. executeIfNew and maybeJoinExecution are modified to explicitly call inFlightTasks.remove(key, future) if retain() fails. This ensures the spin loop is broken immediately by the next thread to encounter the canceled future, rather than waiting for an asynchronous listener.

2. RefcountedFuture.cancel now calls delegate.cancel(mayInterruptIfRunning) when the internal reference count drops to zero.

3. Thread.yield() and the associated @SuppressWarnings("ThreadPriorityCheck") are removed, as they are no longer necessary.

Fixes #28302.

Closes #28938.

PiperOrigin-RevId: 883147973
Change-Id: I28c1db252573a4c39b1a9e53d32e218327340054

Author: tjgq

src/main/java/com/google/devtools/build/lib/concurrent/TaskDeduplicator.java

@@ -45,7 +45,6 @@ public final class TaskDeduplicator<K, V> {
    * effects.
    */
   @CheckReturnValue
-  @SuppressWarnings("ThreadPriorityCheck") // for Thread.yield()
   public ListenableFuture<V> executeIfNew(K key, Supplier<ListenableFuture<V>> taskSupplier) {
     while (true) {
       var isNewHolder = new boolean[1];
@@ -59,12 +58,9 @@ public ListenableFuture<V> executeIfNew(K key, Supplier<ListenableFuture<V>> tas
       if (isNewHolder[0]) {
         future.addListener(() -> inFlightTasks.remove(key, future), directExecutor());
       } else {
-        // The shared future may have been canceled between the lookup and the call to retain(). In
-        // that unlikely case, just look it up again - the listener above will remove it.
+        // The shared future may have been canceled between the lookup and the call to retain().
         if (!future.retain()) {
-          // Avoid spinning to increase the chance that the listener gets to run and removes the
-          // canceled future.
-          Thread.yield();
+          inFlightTasks.remove(key, future);
           continue;
         }
       }
@@ -100,7 +96,11 @@ public ListenableFuture<V> executeUnconditionally(
   @Nullable
   public ListenableFuture<V> maybeJoinExecution(K key) {
     var future = inFlightTasks.get(key);
-    if (future == null || !future.retain()) {
+    if (future == null) {
+      return null;
+    }
+    if (!future.retain()) {
+      inFlightTasks.remove(key, future);
       return null;
     }
     return IndividuallyCancelableFuture.wrap(future);
@@ -125,20 +125,21 @@ static <V> RefcountedFuture<V> wrap(ListenableFuture<V> delegate) {
 
     RefcountedFuture(ListenableFuture<V> delegate) {
       this.delegate = delegate;
+      setFuture(delegate);
     }
 
     @Override
-    public void run() {
-      setFuture(delegate);
-    }
+    public void run() {}
 
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
       if (!mayInterruptIfRunning) {
         this.mayInterruptIfRunning = false;
       }
-      return refcount.updateAndGet(oldCount -> oldCount >= 1 ? oldCount - 1 : 0) == 0
-          && super.cancel(this.mayInterruptIfRunning);
+      if (refcount.updateAndGet(oldCount -> oldCount >= 1 ? oldCount - 1 : 0) == 0) {
+        return super.cancel(this.mayInterruptIfRunning);
+      }
+      return false;
     }
 
     @Nullable