Use lock for resets only

Author: tlrx

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -676,7 +676,7 @@ public void testInvalidateIndicesRequestCacheWhenRollbackEngine() throws Excepti
         final CountDownLatch engineResetLatch = new CountDownLatch(1);
         shard.acquireAllPrimaryOperationsPermits(ActionListener.wrap(r -> {
             try {
-                shard.resetEngineToGlobalCheckpoint();
+                shard.rollbackEngineToGlobalCheckpoint();
             } finally {
                 r.close();
                 engineResetLatch.countDown();

server/src/internalClusterTest/java/org/elasticsearch/indices/IndexingMemoryControllerIT.java

@@ -85,7 +85,7 @@ EngineConfig engineConfigWithLargerIndexingMemory(EngineConfig config) {
                 config.getIndexCommitListener(),
                 config.isPromotableToPrimary(),
                 config.getMapperService(),
-                config.getEngineLock()
+                config.getEngineResetLock()
             );
         }
 

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

@@ -2386,9 +2386,12 @@ public long getLastUnsafeSegmentGenerationForGets() {
         throw new UnsupportedOperationException("Doesn't support getting the latest segment generation");
     }
 
-    protected static <R extends ReferenceManager<ElasticsearchDirectoryReader>> R wrapForAssertions(R referenceManager, EngineConfig engineConfig) {
+    protected static <R extends ReferenceManager<ElasticsearchDirectoryReader>> R wrapForAssertions(
+        R referenceManager,
+        EngineConfig engineConfig
+    ) {
         if (Assertions.ENABLED) {
-            referenceManager.addListener(new AssertRefreshListenerHoldsEngineReadLock(engineConfig.getEngineLock()));
+            referenceManager.addListener(new AssertRefreshListenerHoldsEngineReadLock(engineConfig.getEngineResetLock()));
         }
         return referenceManager;
     }
@@ -2398,9 +2401,9 @@ protected static <R extends ReferenceManager<ElasticsearchDirectoryReader>> R wr
      */
     private static class AssertRefreshListenerHoldsEngineReadLock implements ReferenceManager.RefreshListener {
 
-        private final EngineReadWriteLock engineLock;
+        private final EngineResetLock engineLock;
 
-        private AssertRefreshListenerHoldsEngineReadLock(EngineReadWriteLock engineLock) {
+        private AssertRefreshListenerHoldsEngineReadLock(EngineResetLock engineLock) {
             this.engineLock = Objects.requireNonNull(engineLock);
         }
 

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

@@ -146,7 +146,7 @@ public Supplier<RetentionLeases> retentionLeasesSupplier() {
 
     private final boolean promotableToPrimary;
 
-    private final EngineReadWriteLock engineLock;
+    private final EngineResetLock engineResetLock;
 
     /**
      * Creates a new {@link org.elasticsearch.index.engine.EngineConfig}
@@ -180,7 +180,7 @@ public EngineConfig(
         Engine.IndexCommitListener indexCommitListener,
         boolean promotableToPrimary,
         MapperService mapperService,
-        EngineReadWriteLock engineLock
+        EngineResetLock engineResetLock
     ) {
         this.shardId = shardId;
         this.indexSettings = indexSettings;
@@ -227,7 +227,7 @@ public EngineConfig(
         this.promotableToPrimary = promotableToPrimary;
         // always use compound on flush - reduces # of file-handles on refresh
         this.useCompoundFile = indexSettings.getSettings().getAsBoolean(USE_COMPOUND_FILE, true);
-        this.engineLock = engineLock;
+        this.engineResetLock = engineResetLock;
     }
 
     /**
@@ -473,7 +473,7 @@ public MapperService getMapperService() {
         return mapperService;
     }
 
-    public EngineReadWriteLock getEngineLock() {
-        return engineLock;
+    public EngineResetLock getEngineResetLock() {
+        return engineResetLock;
     }
 }

server/src/main/java/org/elasticsearch/index/engine/EngineReadWriteLock.java renamed to server/src/main/java/org/elasticsearch/index/engine/EngineResetLock.java

@@ -9,21 +9,25 @@
 
 package org.elasticsearch.index.engine;
 
+import org.elasticsearch.core.Assertions;
+
+import java.util.Collection;
+import java.util.List;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
- * Reentrant read/write lock used to guard engine changes in a shard.
+ * Reentrant read/write lock used to control accesses to a shard's engine that can be reset.
  *
- * Implemented as a simple wrapper around a {@link ReentrantReadWriteLock} to make it easier to add/override methods in the future.
+ * Implemented as a simple wrapper around a {@link ReentrantReadWriteLock} to make it easier to add/override methods.
  */
-public final class EngineReadWriteLock implements ReadWriteLock {
+public final class EngineResetLock implements ReadWriteLock {
 
     private final ReentrantReadWriteLock lock;
 
-    public EngineReadWriteLock() {
-        this.lock = new ReentrantReadWriteLock();
+    public EngineResetLock() {
+        this.lock = Assertions.ENABLED ? new QueuedWriterThreadsReentrantReadWriteLock() : new ReentrantReadWriteLock();
     }
 
     @Override
@@ -69,4 +73,24 @@ public boolean isReadLocked() {
     public boolean isReadLockedByCurrentThread() {
         return lock.getReadHoldCount() > 0;
     }
+
+    /**
+     * See {@link ReentrantReadWriteLock#getQueuedWriterThreads()}
+     */
+    public Collection<Thread> getQueuedWriterThreads() {
+        if (lock instanceof QueuedWriterThreadsReentrantReadWriteLock queuedLock) {
+            return queuedLock.queuedWriterThreads();
+        } else {
+            return List.of();
+        }
+    }
+
+    /**
+     * Extends ReentrantReadWriteLock to expose the protected {@link ReentrantReadWriteLock#getQueuedWriterThreads()} method
+     */
+    private static class QueuedWriterThreadsReentrantReadWriteLock extends ReentrantReadWriteLock {
+        Collection<Thread> queuedWriterThreads() {
+            return super.getQueuedWriterThreads();
+        }
+    }
 }

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

@@ -2056,7 +2056,7 @@ protected final RefreshResult refresh(String source, SearcherScope scope, boolea
                     // waiting for the write lock over the threads trying to acquire a (non-reentrant) read lock. Because refresh listeners
                     // are accessing the engine read lock, we need to ensure that they won't block if another thread is waiting for the
                     // engine write lock, so we acquire the read lock upfront before the refresh lock.
-                    final var engineReadLock = engineConfig.getEngineLock().readLock();
+                    final var engineReadLock = engineConfig.getEngineResetLock().readLock();
 
                     // it is intentional that we never refresh both internal / external together
                     if (block) {