smengcl commented on code in PR #11190:
URL: https://github.com/apache/ozone/pull/11190#discussion_r4216689349
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java:
##########
@@ -536,18 +564,34 @@ public CompletableFuture<Message> query(Message request) {
}
@Override
- public synchronized void pause() {
- LOG.info("OzoneManagerStateMachine is pausing");
- statePausedCount.incrementAndGet();
- final LifeCycle.State state = getLifeCycleState();
- if (state == LifeCycle.State.PAUSED) {
- return;
- }
- if (state != LifeCycle.State.NEW) {
- getLifeCycle().transition(LifeCycle.State.PAUSING);
- getLifeCycle().transition(LifeCycle.State.PAUSED);
+ public void pause() {
+ synchronized (this) {
+ LOG.info("OzoneManagerStateMachine is pausing");
+ statePausedCount.incrementAndGet();
+ final LifeCycle.State state = getLifeCycleState();
+ if (state == LifeCycle.State.PAUSED) {
+ return;
+ }
+ if (state != LifeCycle.State.NEW) {
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
+ }
}
+ pauseApplyTransaction();
+ boolean interrupted = false;
+ while (true) {
+ try {
+ ozoneManagerDoubleBuffer.awaitFlush();
+ break;
+ } catch (InterruptedException ex) {
+ interrupted = true;
+ LOG.warn("Interrupted while waiting for OM double buffer to flush
during pause.", ex);
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
ozoneManagerDoubleBuffer.stop();
Review Comment:
Moving the drain outside `synchronized (this)` fixes the prepare deadlock,
but we still need to synchronize with an in-progress `takeSnapshotImpl()`
before returning.
A snapshot can acquire `this` during the flush wait. `pause()` then returns
while that snapshot is still writing, allowing checkpoint installation to
close/replace the DB underneath it. I reproduced this with the snapshot blocked
in the transaction-info table write.
Reacquiring the monitor after the drain restores that ordering without
holding it while waiting for prepare:
```diff
- ozoneManagerDoubleBuffer.stop();
+ synchronized (this) {
+ ozoneManagerDoubleBuffer.stop();
+ }
```
Please cover a snapshot entering during the flush wait. This change passed
that interleaving and the prepare deadlock check.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java:
##########
@@ -536,18 +564,34 @@ public CompletableFuture<Message> query(Message request) {
}
@Override
- public synchronized void pause() {
- LOG.info("OzoneManagerStateMachine is pausing");
- statePausedCount.incrementAndGet();
- final LifeCycle.State state = getLifeCycleState();
- if (state == LifeCycle.State.PAUSED) {
- return;
- }
- if (state != LifeCycle.State.NEW) {
- getLifeCycle().transition(LifeCycle.State.PAUSING);
- getLifeCycle().transition(LifeCycle.State.PAUSED);
+ public void pause() {
+ synchronized (this) {
+ LOG.info("OzoneManagerStateMachine is pausing");
+ statePausedCount.incrementAndGet();
+ final LifeCycle.State state = getLifeCycleState();
+ if (state == LifeCycle.State.PAUSED) {
+ return;
+ }
+ if (state != LifeCycle.State.NEW) {
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
+ }
}
+ pauseApplyTransaction();
+ boolean interrupted = false;
+ while (true) {
+ try {
+ ozoneManagerDoubleBuffer.awaitFlush();
Review Comment:
This also requires updating `testDeleteWithMultiLevelsBlockDoubleBuffer()`
in `TestDirectoryDeletingServiceWithFSO`.
That test calls `stopDaemon()` before `omStateMachine.pause()`, so this wait
has no running flush thread to complete it. Its manual resume happens only
after pause returns. The OM CI job was cancelled while running this test class,
and the stopped-daemon sequence reproduces the hang.
The test can use the double buffer's existing pause/unpause helpers and keep
its original daemon:
```diff
-import java.util.concurrent.CompletableFuture;
```
```diff
- // stop daemon to simulate blocked double buffer
+ // Pause flushing to simulate a blocked double buffer
OzoneManagerStateMachine omStateMachine =
cluster.getOzoneManager().getOmRatisServer().getOmStateMachine();
OzoneManagerDoubleBuffer omDoubleBuffer =
omStateMachine.getOzoneManagerDoubleBuffer();
omDoubleBuffer.awaitFlush();
- omDoubleBuffer.stopDaemon();
+ omDoubleBuffer.pause();
```
```diff
- // re-init double buffer in state machine to resume further processing
- omStateMachine.pause();
-
omStateMachine.unpause(omStateMachine.getLastAppliedTermIndex().getIndex(),
- omStateMachine.getLatestSnapshot().getIndex());
- // flush previous pending transaction manually
- omDoubleBuffer.resume();
- CompletableFuture.supplyAsync(() -> {
- omDoubleBuffer.flushTransactions();
- return null;
- });
+ // Resume flushing the pending transactions.
+ omDoubleBuffer.unpause();
omDoubleBuffer.awaitFlush();
- omDoubleBuffer.stop();
```
Please rerun that integration test with this adjustment.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java:
##########
@@ -796,4 +833,53 @@ public void awaitDoubleBufferFlush() throws
InterruptedException {
public OzoneManagerDoubleBuffer getOzoneManagerDoubleBuffer() {
return ozoneManagerDoubleBuffer;
}
+
+ private void pauseApplyTransaction() {
+ synchronized (applyTransactionMonitor) {
+ if (applyTransactionPaused) {
+ return;
+ }
+ applyTransactionPaused = true;
+ boolean interrupted = false;
+ while (inFlightApplyTransactions > 0) {
+ try {
+ applyTransactionMonitor.wait();
+ } catch (InterruptedException ex) {
+ LOG.warn("Interrupted while waiting for in-flight apply transactions
to complete.");
+ interrupted = true;
+ }
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private void resumeApplyTransaction() {
+ synchronized (applyTransactionMonitor) {
+ if (!applyTransactionPaused) {
+ return;
+ }
+ applyTransactionPaused = false;
+ applyTransactionMonitor.notifyAll();
+ }
+ }
+
+ private void enterApplyTransaction() throws InterruptedException {
+ synchronized (applyTransactionMonitor) {
+ while (applyTransactionPaused) {
+ applyTransactionMonitor.wait();
Review Comment:
Returning an exceptional future removes the deadlock, but Ratis does not
retry this committed entry. `StateMachineUpdater.applyLog()` advances its
cursor regardless of whether the future succeeds.
If `installCheckpoint()` restores the old DB and returns `null`, Ratis does
not reload or rewind that cursor. Entries rejected during the pause are then
skipped. A controlled reproduction executed `[1, 2]` before this change, but
only `[2]` afterward.
Please preserve/replay these entries when installation fails. The new test
manually calls `applyTransaction(trx)` again after unpause, which Ratis does
not do on this path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
