ES-10826

Author: tlrx

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

@@ -2347,7 +2347,7 @@ public record FlushResult(boolean flushPerformed, long generation) {
      * in-progress operations and listeners (e.g., primary term and generation listeners).
      * At the moment, this is implemented in serverless for a special case that ensures the engine is prepared for reset.
      */
-    public void prepareForEngineReset() throws IOException {
+    public void beforeReset() throws IOException {
         throw new UnsupportedOperationException("does not support engine reset");
     }
 

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

@@ -2227,8 +2227,10 @@ protected void flushHoldingLock(boolean force, boolean waitIfOngoing, ActionList
                     preCommitSegmentGeneration.set(lastCommittedSegmentInfos.getGeneration() + 1);
                     commitIndexWriter(indexWriter, translog);
                     logger.trace("finished commit for flush");
-                    // we need to refresh in order to clear older version values
-                    refresh("version_table_flush", SearcherScope.INTERNAL, true);
+                    if (isClosing() == false) {
+                        // we need to refresh in order to clear older version values
+                        refresh("version_table_flush", SearcherScope.INTERNAL, true);
+                    }
                     translog.trimUnreferencedReaders();
                     // Update the translog location for flushListener if (1) the writeLocation has changed during the flush and
                     // (2) indexWriter has committed all the changes (checks must be done in this order).

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -45,6 +45,8 @@
 import org.elasticsearch.cluster.routing.RecoverySource.SnapshotRecoverySource;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDecider;
+import org.elasticsearch.cluster.service.ClusterApplierService;
+import org.elasticsearch.cluster.service.MasterService;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.lucene.Lucene;
@@ -152,6 +154,7 @@
 import org.elasticsearch.search.internal.FieldUsageTrackingDirectoryReader;
 import org.elasticsearch.search.suggest.completion.CompletionStats;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.Transports;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -178,6 +181,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -240,8 +244,9 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
     // ensure happens-before relation between addRefreshListener() and postRecovery()
     private volatile SubscribableListener<Void> postRecoveryComplete;
     private volatile long pendingPrimaryTerm; // see JavaDocs for getPendingPrimaryTerm
-    private final Object engineMutex = new Object(); // lock ordering: engineMutex -> mutex
-    private final AtomicReference<Engine> currentEngineReference = new AtomicReference<>();
+
+    private final ReentrantReadWriteLock engineLock = new ReentrantReadWriteLock();  // lock ordering: engineLock.writeLock -> mutex
+    private Engine currentEngine = null; // must be accessed while holding engineLock
     final EngineFactory engineFactory;
 
     private final IndexingOperationListener indexingOperationListeners;
@@ -1613,7 +1618,8 @@ public Store.MetadataSnapshot snapshotStoreMetadata() throws IOException {
         Engine.IndexCommitRef indexCommit = null;
         store.incRef();
         try {
-            synchronized (engineMutex) {
+            engineLock.readLock().lock();
+            try {
                 // if the engine is not running, we can access the store directly, but we need to make sure no one starts
                 // the engine on us. If the engine is running, we can get a snapshot via the deletion policy of the engine.
                 final Engine engine = getEngineOrNull();
@@ -1623,6 +1629,8 @@ public Store.MetadataSnapshot snapshotStoreMetadata() throws IOException {
                 if (indexCommit == null) {
                     return store.getMetadata(null, true);
                 }
+            } finally {
+                engineLock.readLock().unlock();
             }
             return store.getMetadata(indexCommit.getIndexCommit());
         } finally {
@@ -1776,14 +1784,15 @@ public CacheHelper getReaderCacheHelper() {
     }
 
     public void close(String reason, boolean flushEngine, Executor closeExecutor, ActionListener<Void> closeListener) throws IOException {
-        synchronized (engineMutex) {
+        engineLock.writeLock().lock();
+        try {
             try {
                 synchronized (mutex) {
                     changeState(IndexShardState.CLOSED, reason);
                 }
                 checkAndCallWaitForEngineOrClosedShardListeners();
             } finally {
-                final Engine engine = this.currentEngineReference.getAndSet(null);
+                final Engine engine = getAndSetCurrentEngine(null);
                 closeExecutor.execute(ActionRunnable.run(closeListener, new CheckedRunnable<>() {
                     @Override
                     public void run() throws Exception {
@@ -1810,6 +1819,8 @@ public String toString() {
                     }
                 }));
             }
+        } finally {
+            engineLock.writeLock().unlock();
         }
     }
 
@@ -1857,7 +1868,7 @@ public void prepareForIndexRecovery() {
             throw new IndexShardNotRecoveringException(shardId, state);
         }
         recoveryState.setStage(RecoveryState.Stage.INDEX);
-        assert currentEngineReference.get() == null;
+        assert this.currentEngine == null;
     }
 
     /**
@@ -1936,8 +1947,11 @@ private void doLocalRecovery(
             // First, start a temporary engine, recover the local translog up to the given checkpoint, and then close the engine again.
             .<Void>newForked(l -> ActionListener.runWithResource(ActionListener.assertOnce(l), () -> () -> {
                 assert Thread.holdsLock(mutex) == false : "must not hold the mutex here";
-                synchronized (engineMutex) {
-                    IOUtils.close(currentEngineReference.getAndSet(null));
+                engineLock.writeLock().lock();
+                try {
+                    IOUtils.close(getAndSetCurrentEngine(null));
+                } finally {
+                    engineLock.writeLock().unlock();
                 }
             }, (recoveryCompleteListener, ignoredRef) -> {
                 assert Thread.holdsLock(mutex) == false : "must not hold the mutex here";
@@ -2167,16 +2181,19 @@ private void innerOpenEngineAndTranslog(LongSupplier globalCheckpointSupplier) t
                 + recoveryState.getRecoverySource()
                 + "] but got "
                 + getRetentionLeases();
-        synchronized (engineMutex) {
-            assert currentEngineReference.get() == null : "engine is running";
+        engineLock.writeLock().lock();
+        try {
+            assert currentEngine == null : "engine is running";
             verifyNotClosed();
             // we must create a new engine under mutex (see IndexShard#snapshotStoreMetadata).
             final Engine newEngine = createEngine(config);
             onNewEngine(newEngine);
-            currentEngineReference.set(newEngine);
+            getAndSetCurrentEngine(newEngine);
             // We set active because we are now writing operations to the engine; this way,
             // we can flush if we go idle after some time and become inactive.
             active.set(true);
+        } finally {
+            engineLock.writeLock().unlock();
         }
         // time elapses after the engine is created above (pulling the config settings) until we set the engine reference, during
         // which settings changes could possibly have happened, so here we forcefully push any config changes to the new engine.
@@ -2241,7 +2258,7 @@ private boolean assertLastestCommitUserData() throws IOException {
     }
 
     private void onNewEngine(Engine newEngine) {
-        assert Thread.holdsLock(engineMutex);
+        assert engineLock.isWriteLockedByCurrentThread();
         refreshListeners.setCurrentRefreshLocationSupplier(newEngine::getTranslogLastWriteLocation);
         refreshListeners.setCurrentProcessedCheckpointSupplier(newEngine::getProcessedLocalCheckpoint);
         refreshListeners.setMaxIssuedSeqNoSupplier(newEngine::getMaxSeqNo);
@@ -2252,10 +2269,13 @@ private void onNewEngine(Engine newEngine) {
      */
     public void performRecoveryRestart() throws IOException {
         assert Thread.holdsLock(mutex) == false : "restart recovery under mutex";
-        synchronized (engineMutex) {
+        engineLock.writeLock().lock();
+        try {
             assert refreshListeners.pendingCount() == 0 : "we can't restart with pending listeners";
-            IOUtils.close(currentEngineReference.getAndSet(null));
+            IOUtils.close(getAndSetCurrentEngine(null));
             resetRecoveryStage();
+        } finally {
+            engineLock.writeLock().unlock();
         }
     }
 
@@ -2264,7 +2284,7 @@ public void performRecoveryRestart() throws IOException {
      */
     public void resetRecoveryStage() {
         assert routingEntry().recoverySource().getType() == RecoverySource.Type.PEER : "not a peer recovery [" + routingEntry() + "]";
-        assert currentEngineReference.get() == null;
+        assert this.currentEngine == null;
         if (state != IndexShardState.RECOVERING) {
             throw new IndexShardNotRecoveringException(shardId, state);
         }
@@ -2593,9 +2613,20 @@ boolean shouldRollTranslogGeneration() {
     }
 
     public void onSettingsChanged() {
-        Engine engineOrNull = getEngineOrNull();
-        if (engineOrNull != null) {
-            engineOrNull.onSettingsChanged();
+        // This method can be called within the cluster state applier thread
+        if (engineLock.readLock().tryLock() == false) {
+            // Attempt to acquire a read lock failed:
+            // - the engine is closing, in which case we don't need to apply the updated index settings
+            // - otherwise the onSettingsChanged() should be called again after the new engine is created and the write lock is released
+            return;
+        }
+        try {
+            var engineOrNull = getCurrentEngine(true);
+            if (engineOrNull != null) {
+                engineOrNull.onSettingsChanged();
+            }
+        } finally {
+            engineLock.readLock().unlock();
         }
     }
 
@@ -3286,19 +3317,66 @@ private void doCheckIndex() throws IOException {
     }
 
     Engine getEngine() {
-        Engine engine = getEngineOrNull();
-        if (engine == null) {
-            throw new AlreadyClosedException("engine is closed");
+        engineLock.readLock().lock();
+        try {
+            return getCurrentEngine(false);
+        } finally {
+            engineLock.readLock().unlock();
         }
-        return engine;
     }
 
     /**
      * NOTE: returns null if engine is not yet started (e.g. recovery phase 1, copying over index files, is still running), or if engine is
      * closed.
      */
     public Engine getEngineOrNull() {
-        return this.currentEngineReference.get();
+        engineLock.readLock().lock();
+        try {
+            return getCurrentEngine(true);
+        } finally {
+            engineLock.readLock().unlock();
+        }
+    }
+
+    private Engine getCurrentEngine(boolean allowNoEngine) {
+        assert engineLock.getReadHoldCount() > 0 || engineLock.isWriteLockedByCurrentThread();
+        var engine = this.currentEngine;
+        if (engine == null && allowNoEngine == false) {
+            throw new AlreadyClosedException("engine is closed");
+        }
+        return engine;
+    }
+
+    private Engine getAndSetCurrentEngine(Engine newEngine) {
+        assert engineLock.isWriteLockedByCurrentThread();
+        var previousEngine = this.currentEngine;
+        this.currentEngine = newEngine;
+        return previousEngine;
+    }
+
+    /**
+     * Executes an operation while preventing the shard's engine instance to be changed or closed during the execution. The parameter
+     * {@code allowNoEngine} is used to allow the operation to be executed with a null engine instance. When {@code allowNoEngine} is set
+     * to {@code `false`} the method will throw an {@link AlreadyClosedException} if the current engine is null.
+     *
+     * @param operation
+     * @param allowNoEngine
+     * @return
+     * @param <R>
+     */
+    private <R> R withEngine(Function<Engine, R> operation, boolean allowNoEngine) {
+        assert ClusterApplierService.assertNotClusterStateUpdateThread("IndexShard.withEngine() can block");
+        assert MasterService.assertNotMasterUpdateThread("IndexShard.withEngine() can block");
+        assert Transports.assertNotTransportThread("IndexShard.withEngine() can block");
+        assert operation != null;
+
+        engineLock.readLock().lock();
+        try {
+            var engine = getCurrentEngine(allowNoEngine);
+            return operation.apply(engine);
+        } finally {
+            engineLock.readLock().unlock();
+        }
     }
 
     public void startRecovery(
@@ -4302,7 +4380,7 @@ public void afterRefresh(boolean didRefresh) {
     /**
      * Reset the current engine to a new one.
      *
-     * Calls {@link Engine#prepareForEngineReset()} on the current engine, then closes it, and loads a new engine without
+     * Calls {@link Engine#beforeReset()} on the current engine, then closes it, and loads a new engine without
      * doing any translog recovery.
      *
      * In general, resetting the engine should be done with care, to consider any in-progress operations and listeners.
@@ -4312,12 +4390,15 @@ public void resetEngine() {
         assert Thread.holdsLock(mutex) == false : "resetting engine under mutex";
         assert waitForEngineOrClosedShardListeners.isDone();
         try {
-            synchronized (engineMutex) {
+            engineLock.writeLock().lock(); // might already be held
+            try {
                 verifyNotClosed();
-                getEngine().prepareForEngineReset();
+                currentEngine.beforeReset();
                 var newEngine = createEngine(newEngineConfig(replicationTracker));
-                IOUtils.close(currentEngineReference.getAndSet(newEngine));
+                IOUtils.close(getAndSetCurrentEngine(newEngine));
                 onNewEngine(newEngine);
+            } finally {
+                engineLock.writeLock().unlock();
             }
             onSettingsChanged();
         } catch (Exception e) {
@@ -4342,7 +4423,8 @@ assert getActiveOperationsCount() == OPERATIONS_BLOCKED
         SetOnce<Engine> newEngineReference = new SetOnce<>();
         final long globalCheckpoint = getLastKnownGlobalCheckpoint();
         assert globalCheckpoint == getLastSyncedGlobalCheckpoint();
-        synchronized (engineMutex) {
+        engineLock.writeLock().lock();
+        try {
             verifyNotClosed();
             // we must create both new read-only engine and new read-write engine under engineMutex to ensure snapshotStoreMetadata,
             // acquireXXXCommit and close works.
@@ -4357,41 +4439,52 @@ assert getActiveOperationsCount() == OPERATIONS_BLOCKED
             ) {
                 @Override
                 public IndexCommitRef acquireLastIndexCommit(boolean flushFirst) {
-                    synchronized (engineMutex) {
+                    engineLock.readLock().lock();
+                    try {
                         if (newEngineReference.get() == null) {
                             throw new AlreadyClosedException("engine was closed");
                         }
                         // ignore flushFirst since we flushed above and we do not want to interfere with ongoing translog replay
                         return newEngineReference.get().acquireLastIndexCommit(false);
+                    } finally {
+                        engineLock.readLock().unlock();
                     }
                 }
 
                 @Override
                 public IndexCommitRef acquireSafeIndexCommit() {
-                    synchronized (engineMutex) {
+                    engineLock.readLock().lock();
+                    try {
                         if (newEngineReference.get() == null) {
                             throw new AlreadyClosedException("engine was closed");
                         }
                         return newEngineReference.get().acquireSafeIndexCommit();
+                    } finally {
+                        engineLock.readLock().unlock();
                     }
                 }
 
                 @Override
                 public void close() throws IOException {
                     Engine newEngine;
-                    synchronized (engineMutex) {
+                    engineLock.readLock().lock();
+                    try {
                         newEngine = newEngineReference.get();
-                        if (newEngine == currentEngineReference.get()) {
+                        if (newEngine == getCurrentEngine(true)) {
                             // we successfully installed the new engine so do not close it.
                             newEngine = null;
                         }
+                    } finally {
+                        engineLock.readLock().unlock();
                     }
                     IOUtils.close(super::close, newEngine);
                 }
             };
-            IOUtils.close(currentEngineReference.getAndSet(readOnlyEngine));
+            IOUtils.close(getAndSetCurrentEngine(readOnlyEngine));
             newEngineReference.set(engineFactory.newReadWriteEngine(newEngineConfig(replicationTracker)));
             onNewEngine(newEngineReference.get());
+        } finally {
+            engineLock.writeLock().unlock();
         }
         final Engine.TranslogRecoveryRunner translogRunner = (engine, snapshot) -> runTranslogRecovery(
             engine,
@@ -4403,12 +4496,15 @@ public void close() throws IOException {
         );
         newEngineReference.get().recoverFromTranslog(translogRunner, globalCheckpoint);
         newEngineReference.get().refresh("reset_engine");
-        synchronized (engineMutex) {
+        engineLock.writeLock().lock();
+        try {
             verifyNotClosed();
-            IOUtils.close(currentEngineReference.getAndSet(newEngineReference.get()));
+            IOUtils.close(getAndSetCurrentEngine(newEngineReference.get()));
             // We set active because we are now writing operations to the engine; this way,
             // if we go idle after some time and become inactive, we still give sync'd flush a chance to run.
             active.set(true);
+        } finally {
+            engineLock.writeLock().unlock();
         }
         // time elapses after the engine is created above (pulling the config settings) until we set the engine reference, during
         // which settings changes could possibly have happened, so here we forcefully push any config changes to the new engine.