Cleanup some AtomicLong usage (#91132)

Cleaning up a couple spots where we were using atomics inefficiently:
* removed our home-brew get and update in the countdown
* removed needlessly capturing lambdas in update and gets

Author: original-brownbear

server/src/main/java/org/elasticsearch/cluster/coordination/LagDetector.java

@@ -142,7 +142,7 @@ private class NodeAppliedStateTracker {
         }
 
         void increaseAppliedVersion(long appliedVersion) {
-            long maxAppliedVersion = this.appliedVersion.updateAndGet(v -> Math.max(v, appliedVersion));
+            long maxAppliedVersion = this.appliedVersion.accumulateAndGet(appliedVersion, Math::max);
             logger.trace("{} applied version {}, max now {}", this, appliedVersion, maxAppliedVersion);
         }
 

server/src/main/java/org/elasticsearch/common/util/concurrent/CountDown.java

@@ -35,16 +35,10 @@ public CountDown(int count) {
      */
     public boolean countDown() {
         assert originalCount > 0;
-        for (;;) {
-            final int current = countDown.get();
+        return countDown.getAndUpdate(current -> {
             assert current >= 0;
-            if (current == 0) {
-                return false;
-            }
-            if (countDown.compareAndSet(current, current - 1)) {
-                return current == 1;
-            }
-        }
+            return current == 0 ? 0 : current - 1;
+        }) == 1;
     }
 
     /**

server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedThrottledTaskRunner.java

@@ -77,7 +77,7 @@ protected void pollAndSpawn() {
 
     // Each worker thread that runs a task, first needs to get a "free slot" in order to respect maxRunningTasks.
     private boolean incrementRunningTasks() {
-        int preUpdateValue = runningTasks.getAndUpdate(v -> v < maxRunningTasks ? v + 1 : v);
+        int preUpdateValue = runningTasks.getAndAccumulate(maxRunningTasks, (v, maxRunning) -> v < maxRunning ? v + 1 : v);
         assert preUpdateValue <= maxRunningTasks;
         return preUpdateValue < maxRunningTasks;
     }

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -2914,7 +2914,7 @@ public void afterRefresh(boolean didRefresh) {
         }
 
         void updateRefreshedCheckpoint(long checkpoint) {
-            refreshedCheckpoint.updateAndGet(curr -> Math.max(curr, checkpoint));
+            refreshedCheckpoint.accumulateAndGet(checkpoint, Math::max);
             assert refreshedCheckpoint.get() >= checkpoint : refreshedCheckpoint.get() + " < " + checkpoint;
         }
     }
@@ -2931,9 +2931,9 @@ public final void updateMaxUnsafeAutoIdTimestamp(long newTimestamp) {
 
     private void updateAutoIdTimestamp(long newTimestamp, boolean unsafe) {
         assert newTimestamp >= -1 : "invalid timestamp [" + newTimestamp + "]";
-        maxSeenAutoIdTimestamp.updateAndGet(curr -> Math.max(curr, newTimestamp));
+        maxSeenAutoIdTimestamp.accumulateAndGet(newTimestamp, Math::max);
         if (unsafe) {
-            maxUnsafeAutoIdTimestamp.updateAndGet(curr -> Math.max(curr, newTimestamp));
+            maxUnsafeAutoIdTimestamp.accumulateAndGet(newTimestamp, Math::max);
         }
         assert maxUnsafeAutoIdTimestamp.get() <= maxSeenAutoIdTimestamp.get();
     }
@@ -2949,7 +2949,7 @@ public void advanceMaxSeqNoOfUpdatesOrDeletes(long maxSeqNoOfUpdatesOnPrimary) {
             assert false : "max_seq_no_of_updates on primary is unassigned";
             throw new IllegalArgumentException("max_seq_no_of_updates on primary is unassigned");
         }
-        this.maxSeqNoOfUpdatesOrDeletes.updateAndGet(curr -> Math.max(curr, maxSeqNoOfUpdatesOnPrimary));
+        this.maxSeqNoOfUpdatesOrDeletes.accumulateAndGet(maxSeqNoOfUpdatesOnPrimary, Math::max);
     }
 
     private boolean assertMaxSeqNoOfUpdatesIsAdvanced(Term id, long seqNo, boolean allowDeleted, boolean relaxIfGapInSeqNo) {

server/src/main/java/org/elasticsearch/index/engine/LiveVersionMap.java

@@ -85,8 +85,7 @@ public VersionValue remove(BytesRef uid) {
         }
 
         public void updateMinDeletedTimestamp(DeleteVersionValue delete) {
-            long time = delete.time;
-            minDeleteTimestamp.updateAndGet(prev -> Math.min(time, prev));
+            minDeleteTimestamp.accumulateAndGet(delete.time, Math::min);
         }
 
     }

server/src/main/java/org/elasticsearch/indices/recovery/RecoverySourceHandler.java

@@ -1206,7 +1206,7 @@ protected void executeChunkRequest(OperationChunkRequest request, ActionListener
                 retentionLeases,
                 mappingVersion,
                 listener.delegateFailure((l, newCheckpoint) -> {
-                    targetLocalCheckpoint.updateAndGet(curr -> SequenceNumbers.max(curr, newCheckpoint));
+                    targetLocalCheckpoint.accumulateAndGet(newCheckpoint, SequenceNumbers::max);
                     l.onResponse(null);
                 })
             );

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -676,7 +676,7 @@ public void updateState(ClusterState state) {
                 );
             }
             final long finalBestGen = Math.max(bestGenerationFromCS, metadata.generation());
-            latestKnownRepoGen.updateAndGet(known -> Math.max(known, finalBestGen));
+            latestKnownRepoGen.accumulateAndGet(finalBestGen, Math::max);
         } else {
             final long previousBest = latestKnownRepoGen.getAndSet(metadata.generation());
             if (previousBest != metadata.generation()) {
@@ -861,7 +861,7 @@ private RepositoryData safeRepositoryData(long repositoryStateId, Map<String, Bl
         final long genToLoad;
         final RepositoryData cached;
         if (bestEffortConsistency) {
-            genToLoad = latestKnownRepoGen.updateAndGet(known -> Math.max(known, repositoryStateId));
+            genToLoad = latestKnownRepoGen.accumulateAndGet(repositoryStateId, Math::max);
             cached = null;
         } else {
             genToLoad = latestKnownRepoGen.get();
@@ -1906,7 +1906,7 @@ private void doGetRepositoryData(ActionListener<RepositoryData> listener) {
                     );
                     return;
                 }
-                genToLoad = latestKnownRepoGen.updateAndGet(known -> Math.max(known, generation));
+                genToLoad = latestKnownRepoGen.accumulateAndGet(generation, Math::max);
                 if (genToLoad > generation) {
                     logger.info(
                         "Determined repository generation [{}] from repository contents but correct generation must be at " + "least [{}]",

server/src/main/java/org/elasticsearch/search/internal/ReaderContext.java

@@ -111,7 +111,7 @@ public Engine.Searcher acquireSearcher(String source) {
     }
 
     private void tryUpdateKeepAlive(long keepAlive) {
-        this.keepAlive.updateAndGet(curr -> Math.max(curr, keepAlive));
+        this.keepAlive.accumulateAndGet(keepAlive, Math::max);
     }
 
     /**

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/IndexInputStats.java

@@ -250,8 +250,8 @@ public static class Counter {
         void add(final long value) {
             count.increment();
             total.add(value);
-            min.updateAndGet(prev -> Math.min(prev, value));
-            max.updateAndGet(prev -> Math.max(prev, value));
+            min.accumulateAndGet(value, Math::min);
+            max.accumulateAndGet(value, Math::max);
         }
 
         public long count() {