Fix deadlock between Cache.put and invalidateAll (#99480) (#99579)

The invalidateAll method is taking out the lru lock and segment locks in a different order to the put method, when the put is replacing an existing value. This results in a deadlock between the two methods as they try to swap locks. This fixes it by making sure invalidateAll takes out locks in the same order as put.

This is difficult to test because the put needs to be replacing an existing value, and invalidateAll clears the cache, resulting in subsequent puts not hitting the deadlock condition. A test that overrides some internal implementations to expose this particular deadlock will be coming later.

Author: thecoop

docs/changelog/99480.yaml

@@ -0,0 +1,6 @@
+pr: 99480
+summary: Fix deadlock between Cache.put and Cache.invalidateAll
+area: Infra/Core
+type: bug
+issues:
+ - 99326

server/src/main/java/org/elasticsearch/common/cache/Cache.java

@@ -521,12 +521,12 @@ public void invalidateAll() {
         Entry<K, V> h;
 
         boolean[] haveSegmentLock = new boolean[NUMBER_OF_SEGMENTS];
-        try {
-            for (int i = 0; i < NUMBER_OF_SEGMENTS; i++) {
-                segments[i].segmentLock.writeLock().lock();
-                haveSegmentLock[i] = true;
-            }
-            try (ReleasableLock ignored = lruLock.acquire()) {
+        try (ReleasableLock ignored = lruLock.acquire()) {
+            try {
+                for (int i = 0; i < NUMBER_OF_SEGMENTS; i++) {
+                    segments[i].segmentLock.writeLock().lock();
+                    haveSegmentLock[i] = true;
+                }
                 h = head;
                 for (CacheSegment segment : segments) {
                     segment.map = null;
@@ -539,11 +539,11 @@ public void invalidateAll() {
                 head = tail = null;
                 count = 0;
                 weight = 0;
-            }
-        } finally {
-            for (int i = NUMBER_OF_SEGMENTS - 1; i >= 0; i--) {
-                if (haveSegmentLock[i]) {
-                    segments[i].segmentLock.writeLock().unlock();
+            } finally {
+                for (int i = NUMBER_OF_SEGMENTS - 1; i >= 0; i--) {
+                    if (haveSegmentLock[i]) {
+                        segments[i].segmentLock.writeLock().unlock();
+                    }
                 }
             }
         }