Do not wrap CacheFile reentrant r/w locks with ReleasableLock (#58244)

Today the read/write locks used internally by CacheFile object are 
wrapped into a ReleasableLock. This is not strictly required and also 
prevents usage of the tryLock() methods which we would like to use 
for early releasing of read operations (#58164).

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/cache/CacheFile.java

@@ -11,8 +11,8 @@ import org.elasticsearch.common.CheckedBiConsumer;
 import org.elasticsearch.common.CheckedBiFunction;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.common.util.concurrent.AbstractRefCounted;
-import org.elasticsearch.common.util.concurrent.ReleasableLock;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
@@ -48,8 +48,8 @@ public class CacheFile {
         }
     };
 
-    private final ReleasableLock evictionLock;
-    private final ReleasableLock readLock;
+    private final ReentrantReadWriteLock.WriteLock evictionLock;
+    private final ReentrantReadWriteLock.ReadLock readLock;
 
     private final SparseFileTracker tracker;
     private final String description;
@@ -69,8 +69,8 @@ public class CacheFile {
         this.evicted = false;
 
         final ReentrantReadWriteLock cacheLock = new ReentrantReadWriteLock();
-        this.evictionLock = new ReleasableLock(cacheLock.writeLock());
-        this.readLock = new ReleasableLock(cacheLock.readLock());
+        this.evictionLock = cacheLock.writeLock();
+        this.readLock = cacheLock.readLock();
 
         assert invariant();
     }
@@ -83,9 +83,9 @@ public class CacheFile {
         return file;
     }
 
-    ReleasableLock fileLock() {
+    Releasable fileLock() {
         boolean success = false;
-        final ReleasableLock fileLock = readLock.acquire();
+        readLock.lock();
         try {
             ensureOpen();
             // check if we have a channel while holding the read lock
@@ -93,10 +93,10 @@ public class CacheFile {
                 throw new AlreadyClosedException("Cache file channel has been released and closed");
             }
             success = true;
-            return fileLock;
+            return readLock::unlock;
         } finally {
             if (success == false) {
-                fileLock.close();
+                readLock.unlock();
             }
         }
     }
@@ -112,19 +112,22 @@ public class CacheFile {
         ensureOpen();
         boolean success = false;
         if (refCounter.tryIncRef()) {
-            try (ReleasableLock ignored = evictionLock.acquire()) {
+            evictionLock.lock();
+            try {
+                ensureOpen();
+                final Set<EvictionListener> newListeners = new HashSet<>(listeners);
+                final boolean added = newListeners.add(listener);
+                assert added : "listener already exists " + listener;
+                maybeOpenFileChannel(newListeners);
+                listeners = Collections.unmodifiableSet(newListeners);
+                success = true;
+            } finally {
                 try {
-                    ensureOpen();
-                    final Set<EvictionListener> newListeners = new HashSet<>(listeners);
-                    final boolean added = newListeners.add(listener);
-                    assert added : "listener already exists " + listener;
-                    maybeOpenFileChannel(newListeners);
-                    listeners = Collections.unmodifiableSet(newListeners);
-                    success = true;
-                } finally {
                     if (success == false) {
                         refCounter.decRef();
                     }
+                } finally {
+                    evictionLock.unlock();
                 }
             }
         }
@@ -136,7 +139,8 @@ public class CacheFile {
         assert listener != null;
 
         boolean success = false;
-        try (ReleasableLock ignored = evictionLock.acquire()) {
+        evictionLock.lock();
+        try {
             try {
                 final Set<EvictionListener> newListeners = new HashSet<>(listeners);
                 final boolean removed = newListeners.remove(Objects.requireNonNull(listener));
@@ -152,6 +156,8 @@ public class CacheFile {
                     refCounter.decRef();
                 }
             }
+        } finally {
+            evictionLock.unlock();
         }
         assert invariant();
         return success;
@@ -171,12 +177,15 @@ public class CacheFile {
     public void startEviction() {
         if (evicted == false) {
             final Set<EvictionListener> evictionListeners = new HashSet<>();
-            try (ReleasableLock ignored = evictionLock.acquire()) {
+            evictionLock.lock();
+            try {
                 if (evicted == false) {
                     evicted = true;
                     evictionListeners.addAll(listeners);
                     refCounter.decRef();
                 }
+            } finally {
+                evictionLock.unlock();
             }
             evictionListeners.forEach(listener -> listener.onEviction(this));
         }
@@ -206,7 +215,8 @@ public class CacheFile {
     }
 
     private boolean invariant() {
-        try (ReleasableLock ignored = readLock.acquire()) {
+        readLock.lock();
+        try {
             assert listeners != null;
             if (listeners.isEmpty()) {
                 assert channel == null;
@@ -217,6 +227,8 @@ public class CacheFile {
                 assert channel.isOpen();
                 assert Files.exists(file);
             }
+        } finally {
+            readLock.unlock();
         }
         return true;
     }

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/index/store/cache/CachedBlobContainerIndexInput.java

@@ -16,7 +16,7 @@ import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.Channels;
-import org.elasticsearch.common.util.concurrent.ReleasableLock;
+import org.elasticsearch.common.lease.Releasable;
 import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
 import org.elasticsearch.index.store.BaseSearchableSnapshotIndexInput;
 import org.elasticsearch.index.store.IndexInputStats;
@@ -140,7 +140,7 @@ public class CachedBlobContainerIndexInput extends BaseSearchableSnapshotIndexIn
             int bytesRead = 0;
             try {
                 final CacheFile cacheFile = getCacheFileSafe();
-                try (ReleasableLock ignored = cacheFile.fileLock()) {
+                try (Releasable ignored = cacheFile.fileLock()) {
                     final Tuple<Long, Long> range = computeRange(pos);
                     bytesRead = cacheFile.fetchRange(
                         range.v1(),
@@ -184,7 +184,7 @@ public class CachedBlobContainerIndexInput extends BaseSearchableSnapshotIndexIn
 
         try {
             final CacheFile cacheFile = getCacheFileSafe();
-            try (ReleasableLock ignored = cacheFile.fileLock()) {
+            try (Releasable ignored = cacheFile.fileLock()) {
 
                 final Tuple<Long, Long> range = cacheFile.getAbsentRangeWithin(partRange.v1(), partRange.v2());
                 if (range == null) {