This is an automated email from the ASF dual-hosted git repository.
smengcl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 31b4e332b3e HDDS-16155. Fix OM checkpoint installation deadlock
between state machine pause and double-buffer flush (#10995)
31b4e332b3e is described below
commit 31b4e332b3e9c36bc08ec196117888d227fe97bd
Author: Siyao Meng <[email protected]>
AuthorDate: Mon Aug 24 18:05:42 2026 -0700
HDDS-16155. Fix OM checkpoint installation deadlock between state machine
pause and double-buffer flush (#10995)
---
.../ozone/om/ratis/OzoneManagerStateMachine.java | 97 ++++++++++++++--------
.../om/ratis/TestOzoneManagerStateMachine.java | 89 ++++++++++++++++++++
2 files changed, 150 insertions(+), 36 deletions(-)
diff --git
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
index feeda4ca72b..9bb39d19727 100644
---
a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
+++
b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java
@@ -105,6 +105,12 @@ public class OzoneManagerStateMachine extends
BaseStateMachine {
private final AtomicInteger statePausedCount = new AtomicInteger(0);
private final String threadPrefix;
+ /**
+ * Guards updates to notified, skipped and applied term-index state.
+ * When both locks are needed, acquire the state-machine monitor first.
+ */
+ private final Object termIndexLock = new Object();
+
/** The last {@link TermIndex} received from {@link
#notifyTermIndexUpdated(long, long)}. */
private volatile TermIndex lastNotifiedTermIndex = TermIndex.valueOf(0,
RaftLog.INVALID_LOG_INDEX);
/** The last index skipped by {@link #notifyTermIndexUpdated(long, long)}. */
@@ -239,18 +245,20 @@ public void notifyLeaderChanged(RaftGroupMemberId
groupMemberId,
/** Notified by Ratis for non-StateMachine term-index update. */
@Override
- public synchronized void notifyTermIndexUpdated(long currentTerm, long
newIndex) {
- // lastSkippedIndex is start of sequence (one less) of continuous
notification from ratis
- // if there is any applyTransaction (double buffer index), then this gap
is handled during double buffer
- // notification and lastSkippedIndex will be the start of last continuous
sequence.
- final long oldIndex = lastNotifiedTermIndex.getIndex();
- if (newIndex - oldIndex > 1) {
- lastSkippedIndex = newIndex - 1;
- }
- final TermIndex newTermIndex = TermIndex.valueOf(currentTerm, newIndex);
- lastNotifiedTermIndex = assertUpdateIncreasingly("lastNotified",
lastNotifiedTermIndex, newTermIndex);
- if (lastNotifiedTermIndex.getIndex() -
getLastAppliedTermIndex().getIndex() == 1) {
- updateLastAppliedTermIndex(lastNotifiedTermIndex);
+ public void notifyTermIndexUpdated(long currentTerm, long newIndex) {
+ synchronized (termIndexLock) {
+ // lastSkippedIndex is start of sequence (one less) of continuous
notification from ratis
+ // if there is any applyTransaction (double buffer index), then this gap
is handled during double buffer
+ // notification and lastSkippedIndex will be the start of last
continuous sequence.
+ final long oldIndex = lastNotifiedTermIndex.getIndex();
+ if (newIndex - oldIndex > 1) {
+ lastSkippedIndex = newIndex - 1;
+ }
+ final TermIndex newTermIndex = TermIndex.valueOf(currentTerm, newIndex);
+ lastNotifiedTermIndex = assertUpdateIncreasingly("lastNotified",
lastNotifiedTermIndex, newTermIndex);
+ if (lastNotifiedTermIndex.getIndex() -
getLastAppliedTermIndex().getIndex() == 1) {
+ updateLastAppliedTermIndex(lastNotifiedTermIndex);
+ }
}
}
@@ -259,17 +267,26 @@ public TermIndex getLastNotifiedTermIndex() {
}
@Override
- protected synchronized boolean updateLastAppliedTermIndex(TermIndex
newTermIndex) {
- TermIndex lastApplied = getLastAppliedTermIndex();
- assertUpdateIncreasingly("lastApplied", lastApplied, newTermIndex);
- // if newTermIndex getting updated is within sequence of notifiedTermIndex
(i.e. from lastSkippedIndex and
- // notifiedTermIndex), then can update directly to lastNotifiedTermIndex
as it ensure previous double buffer's
- // Index is notified or getting notified matching lastSkippedIndex
- if (newTermIndex.getIndex() < getLastNotifiedTermIndex().getIndex()
- && newTermIndex.getIndex() >= lastSkippedIndex) {
- newTermIndex = getLastNotifiedTermIndex();
+ protected boolean updateLastAppliedTermIndex(TermIndex newTermIndex) {
+ synchronized (termIndexLock) {
+ TermIndex lastApplied = getLastAppliedTermIndex();
+ assertUpdateIncreasingly("lastApplied", lastApplied, newTermIndex);
+ // if newTermIndex getting updated is within sequence of
notifiedTermIndex (i.e. from lastSkippedIndex and
+ // notifiedTermIndex), then can update directly to lastNotifiedTermIndex
as it ensure previous double buffer's
+ // Index is notified or getting notified matching lastSkippedIndex
+ if (newTermIndex.getIndex() < getLastNotifiedTermIndex().getIndex()
+ && newTermIndex.getIndex() >= lastSkippedIndex) {
+ newTermIndex = getLastNotifiedTermIndex();
+ }
+ return super.updateLastAppliedTermIndex(newTermIndex);
+ }
+ }
+
+ /** Restore the applied term index when loading persisted state. */
+ private void restoreLastAppliedTermIndex(TermIndex termIndex) {
+ synchronized (termIndexLock) {
+ setLastAppliedTermIndex(termIndex);
}
- return super.updateLastAppliedTermIndex(newTermIndex);
}
/** Assert if the given {@link TermIndex} is updated increasingly. */
@@ -544,7 +561,7 @@ public synchronized void unpause(long
newLastAppliedSnaphsotIndex,
if (statePausedCount.decrementAndGet() == 0) {
getLifeCycle().startAndTransition(() -> {
this.ozoneManagerDoubleBuffer = buildDoubleBufferForRatis();
- this.setLastAppliedTermIndex(TermIndex.valueOf(
+ restoreLastAppliedTermIndex(TermIndex.valueOf(
newLastAppliedSnapShotTermIndex, newLastAppliedSnaphsotIndex));
LOG.info("{}: OzoneManagerStateMachine un-pause completed. " +
"newLastAppliedSnapshotIndex: {}, newLastAppliedSnapShotTermIndex:
{}",
@@ -593,20 +610,28 @@ public long takeSnapshot() throws IOException {
return takeSnapshotImpl();
}
+ /**
+ * Keep this synchronized on the state-machine monitor so an in-progress
+ * snapshot finishes before checkpoint pause returns and installation can
+ * replace the metadata store. The nested term-index lock provides a
+ * consistent index to persist.
+ */
private synchronized long takeSnapshotImpl() throws IOException {
- final TermIndex applied = getLastAppliedTermIndex();
- final TermIndex notified = getLastNotifiedTermIndex();
- final TermIndex snapshot = applied.compareTo(notified) > 0 ? applied :
notified;
+ synchronized (termIndexLock) {
+ final TermIndex applied = getLastAppliedTermIndex();
+ final TermIndex notified = getLastNotifiedTermIndex();
+ final TermIndex snapshot = applied.compareTo(notified) > 0 ? applied :
notified;
- long startTime = Time.monotonicNow();
- final TransactionInfo transactionInfo = TransactionInfo.valueOf(snapshot);
- ozoneManager.setTransactionInfo(transactionInfo);
-
ozoneManager.getMetadataManager().getTransactionInfoTable().put(TRANSACTION_INFO_KEY,
transactionInfo);
- ozoneManager.getMetadataManager().getStore().flushDB();
- LOG.info("{}: taking snapshot. applied = {}, skipped = {}, " +
- "notified = {}, current snapshot index = {}, took {} ms",
- getId(), applied, lastSkippedIndex, notified, snapshot,
Time.monotonicNow() - startTime);
- return snapshot.getIndex();
+ long startTime = Time.monotonicNow();
+ final TransactionInfo transactionInfo =
TransactionInfo.valueOf(snapshot);
+ ozoneManager.setTransactionInfo(transactionInfo);
+
ozoneManager.getMetadataManager().getTransactionInfoTable().put(TRANSACTION_INFO_KEY,
transactionInfo);
+ ozoneManager.getMetadataManager().getStore().flushDB();
+ LOG.info("{}: taking snapshot. applied = {}, skipped = {}, " +
+ "notified = {}, current snapshot index = {}, took {} ms",
+ getId(), applied, lastSkippedIndex, notified, snapshot,
Time.monotonicNow() - startTime);
+ return snapshot.getIndex();
+ }
}
/**
@@ -715,7 +740,7 @@ public void loadSnapshotInfoFromDB() throws IOException {
ozoneManager.getMetadataManager());
if (transactionInfo != null) {
final TermIndex ti = transactionInfo.getTermIndex();
- setLastAppliedTermIndex(ti);
+ restoreLastAppliedTermIndex(ti);
ozoneManager.setTransactionInfo(transactionInfo);
LOG.info("LastAppliedIndex is set from TransactionInfo from OM DB as
{}", ti);
} else {
diff --git
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
index b9fb82786e1..f1ca677cbfb 100644
---
a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
+++
b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerStateMachine.java
@@ -17,6 +17,7 @@
package org.apache.hadoop.ozone.om.ratis;
+import static org.apache.ozone.test.GenericTestUtils.waitFor;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -41,9 +42,13 @@
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.hdds.utils.db.DBStore;
@@ -65,6 +70,7 @@
import org.apache.hadoop.ozone.om.helpers.OMRatisHelper;
import org.apache.hadoop.ozone.om.lock.OMLockDetails;
import org.apache.hadoop.ozone.om.ratis_snapshot.OmRatisSnapshotProvider;
+import org.apache.hadoop.ozone.om.response.DummyOMClientResponse;
import org.apache.hadoop.ozone.om.response.OMClientResponse;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
import
org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateKeyRequest;
@@ -918,6 +924,74 @@ public void testPauseAlreadyPaused() {
verify(doubleBuffer, times(2)).stop();
}
+ /**
+ * Reproduces the checkpoint-pause deadlock by holding the double-buffer
+ * flush callback immediately before it updates the last-applied index, then
+ * pausing the state machine while it waits for the flush thread to exit.
+ * Releasing the callback must allow both the index update and pause to
finish.
+ */
+ @Test
+ public void testPauseWhileDoubleBufferUpdatesLastAppliedIndex(@TempDir Path
tmpDir) throws Exception {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(OMConfigKeys.OZONE_OM_DB_DIRS,
tmpDir.toAbsolutePath().toString());
+ OzoneManager testOm = mock(OzoneManager.class);
+ when(testOm.getConfiguration()).thenReturn(conf);
+ when(testOm.getConfig()).thenReturn(conf.getObject(OmConfig.class));
+ OmMetadataManagerImpl metadataManager = new OmMetadataManagerImpl(conf,
testOm);
+ when(testOm.getMetadataManager()).thenReturn(metadataManager);
+
+ CountDownLatch flushCallbackReached = new CountDownLatch(1);
+ CountDownLatch releaseFlushCallback = new CountDownLatch(1);
+ AtomicReference<OzoneManagerStateMachine> stateMachineRef = new
AtomicReference<>();
+ OzoneManagerDoubleBuffer testDoubleBuffer =
OzoneManagerDoubleBuffer.newBuilder()
+ .setOmMetadataManager(metadataManager)
+ .setUpdateLastAppliedIndex(termIndex -> {
+ flushCallbackReached.countDown();
+ awaitUninterruptibly(releaseFlushCallback);
+ stateMachineRef.get().updateLastAppliedTermIndex(termIndex);
+ })
+ .setMaxUnFlushedTransactionCount(10)
+ .build()
+ .start();
+ ExecutorService stateMachineExecutor = Executors.newSingleThreadExecutor();
+ OzoneManagerStateMachine testStateMachine = new OzoneManagerStateMachine(
+ testOm, testDoubleBuffer, mock(RequestHandler.class),
stateMachineExecutor, null);
+ stateMachineRef.set(testStateMachine);
+
+ AtomicReference<Thread> pauseThread = new AtomicReference<>();
+ ExecutorService pauseExecutor = Executors.newSingleThreadExecutor(runnable
-> {
+ Thread thread = new Thread(runnable, "test-state-machine-pause");
+ pauseThread.set(thread);
+ return thread;
+ });
+ Future<?> pauseFuture = null;
+ try {
+ OMResponse response = OMResponse.newBuilder()
+ .setCmdType(Type.CreateKey)
+ .setStatus(Status.OK)
+ .setSuccess(true)
+ .build();
+ testDoubleBuffer.add(new DummyOMClientResponse(response),
TermIndex.valueOf(1, 1));
+ assertTrue(flushCallbackReached.await(5, TimeUnit.SECONDS));
+
+ pauseFuture = pauseExecutor.submit(testStateMachine::pause);
+ waitFor(() -> pauseThread.get() != null && pauseThread.get().getState()
== Thread.State.WAITING, 10, 5000);
+
+ releaseFlushCallback.countDown();
+ pauseFuture.get(5, TimeUnit.SECONDS);
+ assertEquals(1, testStateMachine.getLastAppliedTermIndex().getIndex());
+ } finally {
+ releaseFlushCallback.countDown();
+ if (pauseFuture != null && !pauseFuture.isDone() && pauseThread.get() !=
null) {
+ pauseThread.get().interrupt();
+ }
+ pauseExecutor.shutdownNow();
+ pauseExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ testStateMachine.stop();
+ metadataManager.stop();
+ }
+ }
+
@Test
public void testStopShutdownsResources() {
sm.stop();
@@ -976,6 +1050,21 @@ private static void assertTermIndex(long expectedTerm,
long expectedIndex, TermI
assertEquals(expectedIndex, computed.getIndex());
}
+ private static void awaitUninterruptibly(CountDownLatch latch) {
+ boolean interrupted = false;
+ while (true) {
+ try {
+ latch.await();
+ break;
+ } catch (InterruptedException ex) {
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
private OMRequest sampleWriteRequest() {
return OMRequest.newBuilder()
.setCmdType(Type.CreateKey)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
