Fix lock accounting in releasable lock

Releasble locks hold accounting on who holds the lock when assertions
are enabled. However, the underlying lock can be re-entrant yet we mark
the lock as not held by the current thread as soon as the releasable is
closed. For a re-entrant lock this is not right because the thread could
have entered the lock multiple times. Instead, we have to count how many
times the thread has entered the lock and only mark the lock as not held
by the current thread when the counter reaches zero.

Relates #28202

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

@@ -31,8 +31,9 @@ import java.util.concurrent.locks.Lock;
 public class ReleasableLock implements Releasable {
     private final Lock lock;
 
-    /* a per thread boolean indicating the lock is held by it. only works when assertions are enabled */
-    private final ThreadLocal<Boolean> holdingThreads;
+
+    // a per-thread count indicating how many times the thread has entered the lock; only works if assertions are enabled
+    private final ThreadLocal<Integer> holdingThreads;
 
     public ReleasableLock(Lock lock) {
         this.lock = lock;
@@ -57,12 +58,19 @@ public class ReleasableLock implements Releasable {
     }
 
     private boolean addCurrentThread() {
-        holdingThreads.set(true);
+        final Integer current = holdingThreads.get();
+        holdingThreads.set(current == null ? 1 : current + 1);
         return true;
     }
 
     private boolean removeCurrentThread() {
-        holdingThreads.remove();
+        final Integer count = holdingThreads.get();
+        assert count != null && count > 0;
+        if (count == 1) {
+            holdingThreads.remove();
+        } else {
+            holdingThreads.set(count - 1);
+        }
         return true;
     }
 
@@ -70,7 +78,7 @@ public class ReleasableLock implements Releasable {
         if (holdingThreads == null) {
             throw new UnsupportedOperationException("asserts must be enabled");
         }
-        Boolean b = holdingThreads.get();
-        return b != null && b.booleanValue();
+        final Integer count = holdingThreads.get();
+        return count != null && count > 0;
     }
 }

server/src/test/java/org/elasticsearch/common/util/concurrent/ReleasableLockTests.java

@@ -0,0 +1,97 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.common.util.concurrent;
+
+import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public class ReleasableLockTests extends ESTestCase {
+
+    /**
+     * Test that accounting on whether or not a thread holds a releasable lock is correct. Previously we had a bug where on a re-entrant
+     * lock that if a thread entered the lock twice we would declare that it does not hold the lock after it exits its first entrance but
+     * not its second entrance.
+     *
+     * @throws BrokenBarrierException if awaiting on the synchronization barrier breaks
+     * @throws InterruptedException   if awaiting on the synchronization barrier is interrupted
+     */
+    public void testIsHeldByCurrentThread() throws BrokenBarrierException, InterruptedException {
+        final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+        final ReleasableLock readLock = new ReleasableLock(readWriteLock.readLock());
+        final ReleasableLock writeLock = new ReleasableLock(readWriteLock.writeLock());
+
+        final int numberOfThreads = scaledRandomIntBetween(1, 32);
+        final int iterations = scaledRandomIntBetween(1, 32);
+        final CyclicBarrier barrier = new CyclicBarrier(1 + numberOfThreads);
+        final List<Thread> threads = new ArrayList<>();
+        for (int i = 0; i < numberOfThreads; i++) {
+            final Thread thread = new Thread(() -> {
+                try {
+                    barrier.await();
+                } catch (final BrokenBarrierException | InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+                for (int j = 0; j < iterations; j++) {
+                    if (randomBoolean()) {
+                        acquire(readLock, writeLock);
+                    } else {
+                        acquire(writeLock, readLock);
+                    }
+                }
+                try {
+                    barrier.await();
+                } catch (final BrokenBarrierException | InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            });
+            threads.add(thread);
+            thread.start();
+        }
+
+        barrier.await();
+        barrier.await();
+        for (final Thread thread : threads) {
+            thread.join();
+        }
+    }
+
+    private void acquire(final ReleasableLock lockToAcquire, final ReleasableLock otherLock) {
+        try (@SuppressWarnings("unused") Releasable outer = lockToAcquire.acquire()) {
+            assertTrue(lockToAcquire.isHeldByCurrentThread());
+            assertFalse(otherLock.isHeldByCurrentThread());
+            try (@SuppressWarnings("unused") Releasable inner = lockToAcquire.acquire()) {
+                assertTrue(lockToAcquire.isHeldByCurrentThread());
+                assertFalse(otherLock.isHeldByCurrentThread());
+            }
+            // previously there was a bug here and this would return false
+            assertTrue(lockToAcquire.isHeldByCurrentThread());
+            assertFalse(otherLock.isHeldByCurrentThread());
+        }
+        assertFalse(lockToAcquire.isHeldByCurrentThread());
+        assertFalse(otherLock.isHeldByCurrentThread());
+    }
+
+}