Use RunOnce when appropriate (#35553)

This pull request replaces some blocks of code that must be run once
and that are currently based on AtomicBoolean by the convient RunOnce
class added in #35489.

Author: tlrx

server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
 import org.elasticsearch.core.internal.io.IOUtils;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -128,12 +129,12 @@ public void asyncBlockOperations(final ActionListener<Releasable> onAcquired, fi
         delayOperations();
         threadPool.executor(ThreadPool.Names.GENERIC).execute(new AbstractRunnable() {
 
-            final AtomicBoolean released = new AtomicBoolean(false);
+            final RunOnce released = new RunOnce(() -> releaseDelayedOperations());
 
             @Override
             public void onFailure(final Exception e) {
                 try {
-                    releaseDelayedOperationsIfNeeded(); // resume delayed operations as soon as possible
+                    released.run(); // resume delayed operations as soon as possible
                 } finally {
                     onAcquired.onFailure(e);
                 }
@@ -146,16 +147,10 @@ protected void doRun() throws Exception {
                     try {
                         releasable.close();
                     } finally {
-                        releaseDelayedOperationsIfNeeded();
+                        released.run();
                     }
                 });
             }
-
-            private void releaseDelayedOperationsIfNeeded() {
-                if (released.compareAndSet(false, true)) {
-                    releaseDelayedOperations();
-                }
-            }
         });
     }
 
@@ -177,13 +172,11 @@ private Releasable acquireAll(final long timeout, final TimeUnit timeUnit) throw
             }
         }
         if (semaphore.tryAcquire(TOTAL_PERMITS, timeout, timeUnit)) {
-            final AtomicBoolean closed = new AtomicBoolean();
-            return () -> {
-                if (closed.compareAndSet(false, true)) {
-                    assert semaphore.availablePermits() == 0;
-                    semaphore.release(TOTAL_PERMITS);
-                }
-            };
+            final RunOnce release = new RunOnce(() -> {
+                assert semaphore.availablePermits() == 0;
+                semaphore.release(TOTAL_PERMITS);
+            });
+            return release::run;
         } else {
             throw new TimeoutException("timeout while blocking operations");
         }

test/framework/src/main/java/org/elasticsearch/test/transport/MockTransportService.java

@@ -37,6 +37,7 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.plugins.Plugin;
@@ -66,7 +67,6 @@
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.LinkedBlockingDeque;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -348,21 +348,17 @@ public void sendRequest(Transport.Connection connection, long requestId, String
                 request.writeTo(bStream);
                 final TransportRequest clonedRequest = reg.newRequest(bStream.bytes().streamInput());
 
-                Runnable runnable = new AbstractRunnable() {
-                    AtomicBoolean requestSent = new AtomicBoolean();
-
+                final RunOnce runnable = new RunOnce(new AbstractRunnable() {
                     @Override
                     public void onFailure(Exception e) {
                         logger.debug("failed to send delayed request", e);
                     }
 
                     @Override
                     protected void doRun() throws IOException {
-                        if (requestSent.compareAndSet(false, true)) {
-                            connection.sendRequest(requestId, action, clonedRequest, options);
-                        }
+                        connection.sendRequest(requestId, action, clonedRequest, options);
                     }
-                };
+                });
 
                 // store the request to send it once the rule is cleared.
                 synchronized (this) {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/FlushListener.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.ml.job.process.autodetect.output;
 
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.xpack.core.ml.job.process.autodetect.output.FlushAcknowledgement;
 
 import java.time.Duration;
@@ -14,16 +15,21 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 class FlushListener {
 
     final ConcurrentMap<String, FlushAcknowledgementHolder> awaitingFlushed = new ConcurrentHashMap<>();
-    final AtomicBoolean cleared = new AtomicBoolean(false);
+    final RunOnce onClear = new RunOnce(() -> {
+        Iterator<ConcurrentMap.Entry<String, FlushAcknowledgementHolder>> latches = awaitingFlushed.entrySet().iterator();
+        while (latches.hasNext()) {
+            latches.next().getValue().latch.countDown();
+            latches.remove();
+        }
+    });
 
     @Nullable
     FlushAcknowledgement waitForFlush(String flushId, Duration timeout) throws InterruptedException {
-        if (cleared.get()) {
+        if (onClear.hasRun()) {
             return null;
         }
 
@@ -49,13 +55,7 @@ void clear(String flushId) {
     }
 
     void clear() {
-        if (cleared.compareAndSet(false, true)) {
-            Iterator<ConcurrentMap.Entry<String, FlushAcknowledgementHolder>> latches = awaitingFlushed.entrySet().iterator();
-            while (latches.hasNext()) {
-                latches.next().getValue().latch.countDown();
-                latches.remove();
-            }
-        }
+        onClear.run();
     }
 
     private static class FlushAcknowledgementHolder {

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/FlushListenerTests.java

@@ -60,14 +60,14 @@ public void testClear() throws Exception {
         }
         assertBusy(() -> assertEquals(numWaits, listener.awaitingFlushed.size()));
         assertThat(flushAcknowledgementHolders.stream().map(f -> f.get()).filter(f -> f != null).findAny().isPresent(), is(false));
-        assertFalse(listener.cleared.get());
+        assertFalse(listener.onClear.hasRun());
 
         listener.clear();
 
         for (AtomicReference<FlushAcknowledgement> f : flushAcknowledgementHolders) {
             assertBusy(() -> assertNotNull(f.get()));
         }
         assertTrue(listener.awaitingFlushed.isEmpty());
-        assertTrue(listener.cleared.get());
+        assertTrue(listener.onClear.hasRun());
     }
 }