Skip to content

Commit d2ae71e

Browse files
priyeshkarathapriyeshkaratha
authored
HDDS-15065. Replace Ratis Snapshot Trigger with DB Flush for Periodic Flush Operations (#10100)
1 parent 386c567 commit d2ae71e

13 files changed

Lines changed: 382 additions & 77 deletions

File tree

‎hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -584,15 +584,6 @@ public final class ScmConfigKeys {
584584
public static final long OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD_DEFAULT =
585585
1000L;
586586

587-
/**
588-
* the config will transfer value to ratis config
589-
* raft.server.snapshot.creation.gap, used by ratis to take snapshot
590-
* when manual trigger using api.
591-
*/
592-
public static final String OZONE_SCM_HA_RATIS_SNAPSHOT_GAP
593-
= "ozone.scm.ha.ratis.server.snapshot.creation.gap";
594-
public static final long OZONE_SCM_HA_RATIS_SNAPSHOT_GAP_DEFAULT =
595-
1024L;
596587
public static final String OZONE_SCM_HA_RATIS_SNAPSHOT_DIR =
597588
"ozone.scm.ha.ratis.snapshot.dir";
598589

‎hadoop-hdds/common/src/main/resources/ozone-default.xml‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4201,12 +4201,6 @@
42014201
topology cluster tree from SCM.
42024202
</description>
42034203
</property>
4204-
<property>
4205-
<name>ozone.scm.ha.ratis.server.snapshot.creation.gap</name>
4206-
<value>1024</value>
4207-
<tag>SCM, OZONE</tag>
4208-
<description>Raft snapshot gap index after which snapshot can be taken.</description>
4209-
</property>
42104204
<property>
42114205
<name>ozone.scm.ha.dbtransactionbuffer.flush.interval</name>
42124206
<value>60s</value>

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/RatisUtil.java‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -231,9 +231,6 @@ private static void setRaftSnapshotProperties(
231231
Snapshot.setAutoTriggerThreshold(properties,
232232
ozoneConf.getLong(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD,
233233
ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_THRESHOLD_DEFAULT));
234-
Snapshot.setCreationGap(properties,
235-
ozoneConf.getLong(ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP,
236-
ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP_DEFAULT));
237234
}
238235

239236
public static void checkRatisException(IOException e, String port,

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBuffer.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,14 @@ public interface SCMHADBTransactionBuffer
4444

4545
void flush() throws RocksDatabaseException, CodecException;
4646

47+
void flushIfNeeded(long snapshotWaitTime)
48+
throws RocksDatabaseException, CodecException;
49+
4750
boolean shouldFlush(long snapshotWaitTime);
4851

4952
void init() throws RocksDatabaseException, CodecException;
53+
54+
void beginApplyingTransaction();
55+
56+
void endApplyingTransaction();
5057
}

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferImpl.java‎

Lines changed: 54 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
2121

2222
import com.google.common.base.Preconditions;
23+
import java.util.concurrent.atomic.AtomicInteger;
2324
import java.util.concurrent.atomic.AtomicLong;
2425
import java.util.concurrent.atomic.AtomicReference;
2526
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -53,6 +54,7 @@ public class SCMHADBTransactionBufferImpl implements SCMHADBTransactionBuffer {
5354
private final AtomicReference<SnapshotInfo> latestSnapshot
5455
= new AtomicReference<>();
5556
private final AtomicLong txFlushPending = new AtomicLong(0);
57+
private final AtomicInteger applyingTransactions = new AtomicInteger(0);
5658
private long lastSnapshotTimeMs = 0;
5759
private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
5860

@@ -124,30 +126,64 @@ public AtomicReference<SnapshotInfo> getLatestSnapshotRef() {
124126
public void flush() throws RocksDatabaseException, CodecException {
125127
rwLock.writeLock().lock();
126128
try {
127-
// write latest trx info into trx table in the same batch
128-
Table<String, TransactionInfo> transactionInfoTable
129-
= metadataStore.getTransactionInfoTable();
130-
transactionInfoTable.putWithBatch(currentBatchOperation,
131-
TRANSACTION_INFO_KEY, latestTrxInfo);
129+
flushUnderWriteLock();
130+
} finally {
131+
rwLock.writeLock().unlock();
132+
}
133+
}
132134

133-
metadataStore.getStore().commitBatchOperation(currentBatchOperation);
134-
currentBatchOperation.close();
135-
this.latestSnapshot.set(latestTrxInfo.toSnapshotInfo());
136-
// reset batch operation
137-
currentBatchOperation = metadataStore.getStore().initBatchOperation();
138-
139-
DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
140-
.getDeletedBlockLog();
141-
Preconditions.checkArgument(
142-
deletedBlockLog instanceof DeletedBlockLogImpl);
143-
((DeletedBlockLogImpl) deletedBlockLog).onFlush();
135+
@Override
136+
public void flushIfNeeded(long snapshotWaitTime)
137+
throws RocksDatabaseException, CodecException {
138+
rwLock.writeLock().lock();
139+
try {
140+
if (applyingTransactions.get() > 0) {
141+
return;
142+
}
143+
long timeDiff = scm.getSystemClock().millis() - lastSnapshotTimeMs;
144+
if (txFlushPending.get() > 0 && timeDiff > snapshotWaitTime) {
145+
LOG.debug("Running TransactionFlushTask");
146+
flushUnderWriteLock();
147+
}
144148
} finally {
145-
txFlushPending.set(0);
146-
lastSnapshotTimeMs = scm.getSystemClock().millis();
147149
rwLock.writeLock().unlock();
148150
}
149151
}
150152

153+
private void flushUnderWriteLock()
154+
throws RocksDatabaseException, CodecException {
155+
// write latest trx info into trx table in the same batch
156+
Table<String, TransactionInfo> transactionInfoTable
157+
= metadataStore.getTransactionInfoTable();
158+
transactionInfoTable.putWithBatch(currentBatchOperation,
159+
TRANSACTION_INFO_KEY, latestTrxInfo);
160+
161+
metadataStore.getStore().commitBatchOperation(currentBatchOperation);
162+
currentBatchOperation.close();
163+
this.latestSnapshot.set(latestTrxInfo.toSnapshotInfo());
164+
// reset batch operation
165+
currentBatchOperation = metadataStore.getStore().initBatchOperation();
166+
167+
DeletedBlockLog deletedBlockLog = scm.getScmBlockManager()
168+
.getDeletedBlockLog();
169+
Preconditions.checkArgument(
170+
deletedBlockLog instanceof DeletedBlockLogImpl);
171+
((DeletedBlockLogImpl) deletedBlockLog).onFlush();
172+
173+
txFlushPending.set(0);
174+
lastSnapshotTimeMs = scm.getSystemClock().millis();
175+
}
176+
177+
@Override
178+
public void beginApplyingTransaction() {
179+
applyingTransactions.incrementAndGet();
180+
}
181+
182+
@Override
183+
public void endApplyingTransaction() {
184+
applyingTransactions.decrementAndGet();
185+
}
186+
151187
@Override
152188
public void init() throws RocksDatabaseException, CodecException {
153189
metadataStore = scm.getScmMetadataStore();

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHADBTransactionBufferStub.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,19 @@ public AtomicReference<SnapshotInfo> getLatestSnapshotRef() {
101101
return null;
102102
}
103103

104+
@Override
105+
public void flushIfNeeded(long snapshotWaitTime) throws RocksDatabaseException {
106+
flush();
107+
}
108+
109+
@Override
110+
public void beginApplyingTransaction() {
111+
}
112+
113+
@Override
114+
public void endApplyingTransaction() {
115+
}
116+
104117
@Override
105118
public void flush() throws RocksDatabaseException {
106119
rwLock.writeLock().lock();

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -142,8 +142,7 @@ private void createStartTransactionBufferMonitor() {
142142
OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL_DEFAULT,
143143
TimeUnit.MILLISECONDS);
144144
SCMHATransactionBufferMonitorTask monitorTask
145-
= new SCMHATransactionBufferMonitorTask(
146-
transactionBuffer, ratisServer, interval);
145+
= new SCMHATransactionBufferMonitorTask(transactionBuffer, interval);
147146
trxBufferMonitorService =
148147
new BackgroundSCMService.Builder().setClock(scm.getSystemClock())
149148
.setScmContext(scm.getScmContext())

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHATransactionBufferMonitorTask.java‎

Lines changed: 5 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
package org.apache.hadoop.hdds.scm.ha;
1919

2020
import java.io.IOException;
21-
import org.apache.ratis.statemachine.SnapshotInfo;
2221
import org.slf4j.Logger;
2322
import org.slf4j.LoggerFactory;
2423

@@ -29,39 +28,24 @@
2928
public class SCMHATransactionBufferMonitorTask implements Runnable {
3029
private static final Logger LOG =
3130
LoggerFactory.getLogger(SCMHATransactionBufferMonitorTask.class);
32-
private final SCMRatisServer server;
3331
private final SCMHADBTransactionBuffer transactionBuffer;
3432
private final long flushInterval;
3533

3634
/**
3735
* SCMService related variables.
3836
*/
3937
public SCMHATransactionBufferMonitorTask(
40-
SCMHADBTransactionBuffer transactionBuffer,
41-
SCMRatisServer server, long flushInterval) {
38+
SCMHADBTransactionBuffer transactionBuffer, long flushInterval) {
4239
this.flushInterval = flushInterval;
4340
this.transactionBuffer = transactionBuffer;
44-
this.server = server;
4541
}
4642

4743
@Override
4844
public void run() {
49-
if (transactionBuffer.shouldFlush(flushInterval)) {
50-
LOG.debug("Running TransactionFlushTask");
51-
// set latest snapshot to null for force snapshot
52-
// the value will be reset again when snapshot is taken
53-
final SnapshotInfo lastSnapshot = transactionBuffer
54-
.getLatestSnapshotRef().getAndSet(null);
55-
try {
56-
server.triggerSnapshot();
57-
} catch (IOException e) {
58-
LOG.error("Snapshot request is failed", e);
59-
} finally {
60-
// under failure case, if unable to take snapshot, its value
61-
// is reset to previous known value
62-
transactionBuffer.getLatestSnapshotRef().compareAndSet(
63-
null, lastSnapshot);
64-
}
45+
try {
46+
transactionBuffer.flushIfNeeded(flushInterval);
47+
} catch (IOException e) {
48+
LOG.error("TransactionFlushTask is failed", e);
6549
}
6650
}
6751
}

‎hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java‎

Lines changed: 27 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,7 @@ public CompletableFuture<Message> applyTransaction(
160160
final TransactionContext trx) {
161161
final CompletableFuture<Message> applyTransactionFuture =
162162
new CompletableFuture<>();
163+
transactionBuffer.beginApplyingTransaction();
163164
try {
164165
final SCMRatisRequest request = SCMRatisRequest.decode(
165166
Message.valueOf(trx.getStateMachineLogEntry().getLogData()));
@@ -192,6 +193,8 @@ public CompletableFuture<Message> applyTransaction(
192193
} catch (Exception ex) {
193194
applyTransactionFuture.completeExceptionally(ex);
194195
ExitUtils.terminate(1, ex.getMessage(), ex, StateMachine.LOG);
196+
} finally {
197+
transactionBuffer.endApplyingTransaction();
195198
}
196199
return applyTransactionFuture;
197200
}
@@ -363,23 +366,28 @@ public long takeSnapshot() throws IOException {
363366
return lastAppliedIndex;
364367
}
365368

366-
long startTime = Time.monotonicNow();
369+
transactionBuffer.beginApplyingTransaction();
370+
try {
371+
long startTime = Time.monotonicNow();
367372

368-
TransactionInfo latestTrxInfo = transactionBuffer.getLatestTrxInfo();
369-
final TransactionInfo lastAppliedTrxInfo = TransactionInfo.valueOf(lastTermIndex);
373+
TransactionInfo latestTrxInfo = transactionBuffer.getLatestTrxInfo();
374+
final TransactionInfo lastAppliedTrxInfo = TransactionInfo.valueOf(lastTermIndex);
370375

371-
if (latestTrxInfo.compareTo(lastAppliedTrxInfo) < 0) {
372-
transactionBuffer.updateLatestTrxInfo(lastAppliedTrxInfo);
373-
transactionBuffer.setLatestSnapshot(lastAppliedTrxInfo.toSnapshotInfo());
374-
} else {
375-
lastAppliedIndex = latestTrxInfo.getTransactionIndex();
376-
}
376+
if (latestTrxInfo.compareTo(lastAppliedTrxInfo) < 0) {
377+
transactionBuffer.updateLatestTrxInfo(lastAppliedTrxInfo);
378+
transactionBuffer.setLatestSnapshot(lastAppliedTrxInfo.toSnapshotInfo());
379+
} else {
380+
lastAppliedIndex = latestTrxInfo.getTransactionIndex();
381+
}
377382

378-
transactionBuffer.flush();
383+
transactionBuffer.flush();
379384

380-
LOG.info("Current Snapshot Index {}, takeSnapshot took {} ms",
381-
lastAppliedIndex, Time.monotonicNow() - startTime);
382-
return lastAppliedIndex;
385+
LOG.info("Current Snapshot Index {}, takeSnapshot took {} ms",
386+
lastAppliedIndex, Time.monotonicNow() - startTime);
387+
return lastAppliedIndex;
388+
} finally {
389+
transactionBuffer.endApplyingTransaction();
390+
}
383391
}
384392

385393
@Override
@@ -399,7 +407,12 @@ public void notifyTermIndexUpdated(long term, long index) {
399407
}
400408

401409
if (transactionBuffer != null) {
402-
transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term, index));
410+
transactionBuffer.beginApplyingTransaction();
411+
try {
412+
transactionBuffer.updateLatestTrxInfo(TransactionInfo.valueOf(term, index));
413+
} finally {
414+
transactionBuffer.endApplyingTransaction();
415+
}
403416
}
404417

405418
if (currentLeaderTerm.get() == term) {

0 commit comments

Comments
 (0)