diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java index c2143a66..8dfa41ec 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java @@ -269,11 +269,11 @@ public void doWork() { if (DLedgerUtils.elapsed(lastPrintWatermarkTimeMs) > 3000) { if (DLedgerEntryPusher.this.fsmCaller.isPresent()) { final long lastAppliedIndex = DLedgerEntryPusher.this.fsmCaller.get().getLastAppliedIndex(); - logger.info("[{}][{}] term={} ledgerBegin={} ledgerEnd={} committed={} watermarks={} appliedIndex={}", - memberState.getSelfId(), memberState.getRole(), memberState.currTerm(), dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex(), dLedgerStore.getCommittedIndex(), JSON.toJSONString(peerWaterMarksByTerm), lastAppliedIndex); + logger.info("[{}][{}] term={} ledgerBeforeBegin={} ledgerEnd={} committed={} watermarks={} appliedIndex={}", + memberState.getSelfId(), memberState.getRole(), memberState.currTerm(), dLedgerStore.getLedgerBeforeBeginIndex(), dLedgerStore.getLedgerEndIndex(), dLedgerStore.getCommittedIndex(), JSON.toJSONString(peerWaterMarksByTerm), lastAppliedIndex); } else { - logger.info("[{}][{}] term={} ledgerBegin={} ledgerEnd={} committed={} watermarks={}", - memberState.getSelfId(), memberState.getRole(), memberState.currTerm(), dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex(), dLedgerStore.getCommittedIndex(), JSON.toJSONString(peerWaterMarksByTerm)); + logger.info("[{}][{}] term={} ledgerBeforeBegin={} ledgerEnd={} committed={} watermarks={}", + memberState.getSelfId(), memberState.getRole(), memberState.currTerm(), dLedgerStore.getLedgerBeforeBeginIndex(), dLedgerStore.getLedgerEndIndex(), dLedgerStore.getCommittedIndex(), JSON.toJSONString(peerWaterMarksByTerm)); } lastPrintWatermarkTimeMs = System.currentTimeMillis(); } @@ -320,10 +320,11 @@ public void doWork() { final Optional fsmCaller = DLedgerEntryPusher.this.fsmCaller; if (fsmCaller.isPresent()) { // If there exist statemachine - DLedgerEntryPusher.this.dLedgerStore.updateCommittedIndex(currTerm, quorumIndex); final StateMachineCaller caller = fsmCaller.get(); - caller.onCommitted(quorumIndex); - + if (quorumIndex > this.lastQuorumIndex) { + DLedgerEntryPusher.this.dLedgerStore.updateCommittedIndex(currTerm, quorumIndex); + caller.onCommitted(quorumIndex); + } // Check elapsed if (DLedgerUtils.elapsed(lastCheckLeakTimeMs) > 1000) { updatePeerWaterMark(currTerm, memberState.getSelfId(), dLedgerStore.getLedgerEndIndex()); @@ -335,7 +336,9 @@ public void doWork() { waitForRunning(1); } } else { - dLedgerStore.updateCommittedIndex(currTerm, quorumIndex); + if (quorumIndex > this.lastQuorumIndex) { + dLedgerStore.updateCommittedIndex(currTerm, quorumIndex); + } ConcurrentMap> responses = pendingAppendResponsesByTerm.get(currTerm); boolean needCheck = false; int ackNum = 0; @@ -733,8 +736,8 @@ private void doCompare() throws Exception { if (compareIndex == -1) { compareIndex = dLedgerStore.getLedgerEndIndex(); logger.info("[Push-{}][DoCompare] compareIndex=-1 means start to compare", peerId); - } else if (compareIndex > dLedgerStore.getLedgerEndIndex() || compareIndex < dLedgerStore.getLedgerBeginIndex()) { - logger.info("[Push-{}][DoCompare] compareIndex={} out of range {}-{}", peerId, compareIndex, dLedgerStore.getLedgerBeginIndex(), dLedgerStore.getLedgerEndIndex()); + } else if (compareIndex > dLedgerStore.getLedgerEndIndex() || compareIndex <= dLedgerStore.getLedgerBeforeBeginIndex()) { + logger.info("[Push-{}][DoCompare] compareIndex={} out of range ({}-{}]", peerId, compareIndex, dLedgerStore.getLedgerBeforeBeginIndex(), dLedgerStore.getLedgerEndIndex()); compareIndex = dLedgerStore.getLedgerEndIndex(); } @@ -761,21 +764,21 @@ private void doCompare() throws Exception { } else { truncateIndex = compareIndex; } - } else if (response.getEndIndex() < dLedgerStore.getLedgerBeginIndex() + } else if (response.getEndIndex() <= dLedgerStore.getLedgerBeforeBeginIndex() || response.getBeginIndex() > dLedgerStore.getLedgerEndIndex()) { /* The follower's entries does not intersect with the leader. This usually happened when the follower has crashed for a long time while the leader has deleted the expired entries. Just truncate the follower. */ - truncateIndex = dLedgerStore.getLedgerBeginIndex(); + truncateIndex = dLedgerStore.getLedgerBeforeBeginIndex() + 1; } else if (compareIndex < response.getBeginIndex()) { /* The compared index is smaller than the follower's begin index. This happened rarely, usually means some disk damage. Just truncate the follower. */ - truncateIndex = dLedgerStore.getLedgerBeginIndex(); + truncateIndex = dLedgerStore.getLedgerBeforeBeginIndex() + 1; } else if (compareIndex > response.getEndIndex()) { /* The compared index is bigger than the follower's end index. @@ -791,8 +794,8 @@ private void doCompare() throws Exception { /* The compared index is smaller than the leader's begin index, truncate the follower. */ - if (compareIndex < dLedgerStore.getLedgerBeginIndex()) { - truncateIndex = dLedgerStore.getLedgerBeginIndex(); + if (compareIndex <= dLedgerStore.getLedgerBeforeBeginIndex()) { + truncateIndex = dLedgerStore.getLedgerBeforeBeginIndex() + 1; } /* If get value for truncateIndex, do it right now. @@ -890,7 +893,7 @@ private PushEntryResponse buildResponse(PushEntryRequest request, int code) { response.setIndex(request.getFirstEntryIndex()); response.setCount(request.getCount()); } - response.setBeginIndex(dLedgerStore.getLedgerBeginIndex()); + response.setBeginIndex(dLedgerStore.getLedgerBeforeBeginIndex() + 1); response.setEndIndex(dLedgerStore.getLedgerEndIndex()); return response; } diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/SnapshotManager.java b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/SnapshotManager.java index a537287f..8be5dd0d 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/SnapshotManager.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/SnapshotManager.java @@ -43,8 +43,8 @@ public class SnapshotManager { public static final String SNAPSHOT_TEMP_DIR = "tmp"; private DLedgerServer dLedgerServer; - private long lastSnapshotIndex; - private long lastSnapshotTerm; + private long lastSnapshotIndex = -1; + private long lastSnapshotTerm = -1; private final SnapshotStore snapshotStore; private volatile boolean savingSnapshot; private volatile boolean loadingSnapshot; @@ -126,7 +126,7 @@ public void saveSnapshot(DLedgerEntry dLedgerEntry) { return; } // Check if applied index reaching the snapshot threshold - if (dLedgerEntry.getIndex() - this.lastSnapshotIndex <= this.dLedgerServer.getDLedgerConfig().getSnapshotThreshold()) { + if (dLedgerEntry.getIndex() - this.lastSnapshotIndex < this.dLedgerServer.getDLedgerConfig().getSnapshotThreshold()) { return; } // Create snapshot writer @@ -172,7 +172,7 @@ private void saveSnapshotAfter(SnapshotWriter writer, SnapshotMeta snapshotMeta, private void truncatePrefix(DLedgerEntry entry) { deleteExpiredSnapshot(); - this.dLedgerServer.getFsmCaller().getdLedgerStore().resetOffsetAfterSnapshot(entry); + this.dLedgerServer.getDLedgerStore().resetOffsetAfterSnapshot(entry); } private void deleteExpiredSnapshot() { @@ -221,6 +221,7 @@ private void loadSnapshotAfter(SnapshotReader reader, SnapshotMeta snapshotMeta, this.lastSnapshotIndex = snapshotMeta.getLastIncludedIndex(); this.lastSnapshotTerm = snapshotMeta.getLastIncludedTerm(); this.loadingSnapshot = false; + this.dLedgerServer.getDLedgerStore().updateIndexAfterLoadingSnapshot(this.lastSnapshotIndex, this.lastSnapshotTerm); logger.info("Snapshot {} loaded successfully", snapshotMeta); } else { // Stop the loading process if the snapshot is expired @@ -244,7 +245,7 @@ private void loadSnapshotAfter(SnapshotReader reader, SnapshotMeta snapshotMeta, } if (failed) { // Still able to recover from files if the beginning index of file store is 0 - if (this.dLedgerServer.getFsmCaller().getdLedgerStore().getLedgerBeginIndex() == 0) { + if (this.dLedgerServer.getFsmCaller().getdLedgerStore().getLedgerBeforeBeginIndex() == -1) { this.loadingSnapshot = false; return; } diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java index 11ef354d..5009cdd3 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/snapshot/file/FileSnapshotWriter.java @@ -99,6 +99,6 @@ public void setSnapshotMeta(SnapshotMeta snapshotMeta) { } public long getSnapshotIndex() { - return this.snapshotMeta != null ? this.snapshotMeta.getLastIncludedIndex() : 0; + return this.snapshotMeta != null ? this.snapshotMeta.getLastIncludedIndex() : -1; } } diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java b/dledger/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java index 67779a9b..e25f33a1 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java @@ -170,11 +170,11 @@ private void doCommitted(final long committedIndex) { if (this.error != null) { return; } - if (this.snapshotManager.isLoadingSnapshot()) { + if (this.snapshotManager.isLoadingSnapshot() || this.snapshotManager.isSavingSnapshot()) { this.scheduledExecutorService.schedule(() -> { try { onCommitted(committedIndex); - logger.info("Still loading snapshot, retry the commit task later"); + logger.info("Still loading or saving snapshot, retry the commit task later"); } catch (Throwable e) { e.printStackTrace(); } diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerMemoryStore.java b/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerMemoryStore.java index 542c3c2c..398ed4a4 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerMemoryStore.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerMemoryStore.java @@ -32,6 +32,7 @@ public class DLedgerMemoryStore extends DLedgerStore { private static final Logger LOGGER = LoggerFactory.getLogger(DLedgerMemoryStore.class); + private long ledgerBeforeBeginIndex = -1; private long ledgerBeginIndex = -1; private long ledgerEndIndex = -1; private long committedIndex = -1; @@ -61,9 +62,6 @@ public DLedgerEntry appendAsLeader(DLedgerEntry entry) { LOGGER.debug("[{}] Append as Leader {} {}", memberState.getSelfId(), entry.getIndex(), entry.getBody().length); } cachedEntries.put(entry.getIndex(), entry); - if (ledgerBeginIndex == -1) { - ledgerBeginIndex = ledgerEndIndex; - } updateLedgerEndIndexAndTerm(); return entry; } @@ -74,6 +72,28 @@ public long truncate(DLedgerEntry entry, long leaderTerm, String leaderId) { return appendAsFollower(entry, leaderTerm, leaderId).getIndex(); } + @Override + public void resetOffsetAfterSnapshot(DLedgerEntry entry) { + + } + + @Override + public void updateIndexAfterLoadingSnapshot(long lastIncludedIndex, long lastIncludedTerm) { + this.ledgerBeforeBeginIndex = lastIncludedIndex; + this.ledgerEndIndex = lastIncludedIndex; + this.ledgerEndTerm = lastIncludedTerm; + } + + @Override + public void startup() { + + } + + @Override + public void shutdown() { + + } + @Override public DLedgerEntry appendAsFollower(DLedgerEntry entry, long leaderTerm, String leaderId) { PreConditions.check(memberState.isFollower(), DLedgerResponseCode.NOT_FOLLOWER); @@ -88,9 +108,6 @@ public DLedgerEntry appendAsFollower(DLedgerEntry entry, long leaderTerm, String ledgerEndIndex = entry.getIndex(); committedIndex = entry.getIndex(); cachedEntries.put(entry.getIndex(), entry); - if (ledgerBeginIndex == -1) { - ledgerBeginIndex = ledgerEndIndex; - } updateLedgerEndIndexAndTerm(); return entry; } @@ -107,11 +124,17 @@ public long getLedgerEndIndex() { return ledgerEndIndex; } + @Deprecated @Override public long getLedgerBeginIndex() { return ledgerBeginIndex; } + @Override + public long getLedgerBeforeBeginIndex() { + return ledgerBeforeBeginIndex; + } + @Override public long getCommittedIndex() { return committedIndex; diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerStore.java b/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerStore.java index 6f1f981b..9b0b0bcb 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerStore.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/store/DLedgerStore.java @@ -43,6 +43,8 @@ public void updateCommittedIndex(long term, long committedIndex) { public abstract long getLedgerBeginIndex(); + public abstract long getLedgerBeforeBeginIndex(); + protected void updateLedgerEndIndexAndTerm() { if (getMemberState() != null) { getMemberState().updateLedgerIndexAndTerm(getLedgerEndIndex(), getLedgerEndTerm()); @@ -57,15 +59,12 @@ public long truncate(DLedgerEntry entry, long leaderTerm, String leaderId) { return -1; } - public void resetOffsetAfterSnapshot(DLedgerEntry entry) { + public abstract void resetOffsetAfterSnapshot(DLedgerEntry entry); - } + public abstract void updateIndexAfterLoadingSnapshot(long lastIncludedIndex, long lastIncludedTerm); - public void startup() { + public abstract void startup(); - } + public abstract void shutdown(); - public void shutdown() { - - } } diff --git a/dledger/src/main/java/io/openmessaging/storage/dledger/store/file/DLedgerMmapFileStore.java b/dledger/src/main/java/io/openmessaging/storage/dledger/store/file/DLedgerMmapFileStore.java index 08fb677e..c9a08366 100644 --- a/dledger/src/main/java/io/openmessaging/storage/dledger/store/file/DLedgerMmapFileStore.java +++ b/dledger/src/main/java/io/openmessaging/storage/dledger/store/file/DLedgerMmapFileStore.java @@ -50,6 +50,8 @@ public class DLedgerMmapFileStore extends DLedgerStore { private static final Logger LOGGER = LoggerFactory.getLogger(DLedgerMmapFileStore.class); public List appendHooks = new ArrayList<>(); + + private volatile long ledgerBeforeBeginIndex = -1; private long ledgerBeginIndex = -1; private long ledgerEndIndex = -1; private long committedIndex = -1; @@ -301,7 +303,7 @@ public void recover() { DLedgerEntry entry = get(lastEntryIndex); PreConditions.check(entry != null, DLedgerResponseCode.DISK_ERROR, "recheck get null entry"); PreConditions.check(entry.getIndex() == lastEntryIndex, DLedgerResponseCode.DISK_ERROR, "recheck index %d != %d", entry.getIndex(), lastEntryIndex); - reviseLedgerBeginIndex(); + reviseLedgerBeforeBeginIndex(); } this.dataFileList.updateWherePosition(processOffset); this.dataFileList.truncateOffset(processOffset); @@ -325,6 +327,7 @@ public void recover() { } + @Deprecated private void reviseLedgerBeginIndex() { //get ledger begin index MmapFile firstFile = dataFileList.getFirstMappedFile(); @@ -341,6 +344,28 @@ private void reviseLedgerBeginIndex() { } } + private void reviseLedgerBeforeBeginIndex() { + // get ledger begin index + MmapFile firstFile = dataFileList.getFirstMappedFile(); + SelectMmapBufferResult sbr = firstFile.selectMappedBuffer(0); + try { + ByteBuffer tmpBuffer = sbr.getByteBuffer(); + tmpBuffer.position(firstFile.getStartPosition()); + tmpBuffer.getInt(); //magic + int size = tmpBuffer.getInt();//size + if (size == 0) { + // means that now empty entry + return; + } + // begin index + long beginIndex = tmpBuffer.getLong(); + this.ledgerBeforeBeginIndex = beginIndex - 1; + indexFileList.resetOffset(beginIndex * INDEX_UNIT_SIZE); + } finally { + SelectMmapBufferResult.release(sbr); + } + } + @Override public DLedgerEntry appendAsLeader(DLedgerEntry entry) { PreConditions.check(memberState.isLeader(), DLedgerResponseCode.NOT_LEADER); @@ -375,9 +400,9 @@ public DLedgerEntry appendAsLeader(DLedgerEntry entry) { } ledgerEndIndex++; ledgerEndTerm = memberState.currTerm(); - if (ledgerBeginIndex == -1) { - ledgerBeginIndex = ledgerEndIndex; - } +// if (ledgerBeginIndex == -1) { +// ledgerBeginIndex = ledgerEndIndex; +// } updateLedgerEndIndexAndTerm(); return entry; } @@ -428,7 +453,7 @@ public long truncate(DLedgerEntry entry, long leaderTerm, String leaderId) { PreConditions.check(indexPos == entry.getIndex() * INDEX_UNIT_SIZE, DLedgerResponseCode.DISK_ERROR, null); ledgerEndTerm = entry.getTerm(); ledgerEndIndex = entry.getIndex(); - reviseLedgerBeginIndex(); + reviseLedgerBeforeBeginIndex(); updateLedgerEndIndexAndTerm(); return entry.getIndex(); } @@ -470,11 +495,25 @@ private long calculateWherePosition(final MmapFileList mappedFileList, long cont @Override public void resetOffsetAfterSnapshot(DLedgerEntry entry) { - long resetPos = entry.getPos() + entry.getSize(); - dataFileList.resetOffset(resetPos); - long resetIndexOffset = entry.getIndex() * INDEX_UNIT_SIZE; - indexFileList.resetOffset(resetIndexOffset); - reviseLedgerBeginIndex(); + // judge expired + if (entry.getIndex() <= this.ledgerBeforeBeginIndex) { + return; + } + synchronized (this.memberState) { + long resetPos = entry.getPos() + entry.getSize(); + dataFileList.resetOffset(resetPos); + long resetIndexOffset = entry.getIndex() * INDEX_UNIT_SIZE; + indexFileList.resetOffset(resetIndexOffset); + // reset ledgerBeforeBeginIndex + this.ledgerBeforeBeginIndex = entry.getIndex(); + } + } + + @Override + public void updateIndexAfterLoadingSnapshot(long lastIncludedIndex, long lastIncludedTerm) { + this.ledgerBeforeBeginIndex = lastIncludedIndex; + this.ledgerEndIndex = lastIncludedIndex; + this.ledgerEndTerm = lastIncludedTerm; } @Override @@ -498,9 +537,9 @@ public DLedgerEntry appendAsFollower(DLedgerEntry entry, long leaderTerm, String PreConditions.check(indexPos == entry.getIndex() * INDEX_UNIT_SIZE, DLedgerResponseCode.DISK_ERROR, null); ledgerEndTerm = entry.getTerm(); ledgerEndIndex = entry.getIndex(); - if (ledgerBeginIndex == -1) { - ledgerBeginIndex = ledgerEndIndex; - } +// if (ledgerBeginIndex == -1) { +// ledgerBeginIndex = ledgerEndIndex; +// } updateLedgerEndIndexAndTerm(); return entry; } @@ -535,11 +574,17 @@ public long getLedgerEndIndex() { return ledgerEndIndex; } + @Deprecated @Override public long getLedgerBeginIndex() { return ledgerBeginIndex; } + @Override + public long getLedgerBeforeBeginIndex() { + return ledgerBeforeBeginIndex; + } + @Override public DLedgerEntry get(Long index) { indexCheck(index); @@ -582,8 +627,8 @@ public Pair getEntryPosAndSize(Long index) { public void indexCheck(Long index) { PreConditions.check(index >= 0, DLedgerResponseCode.INDEX_OUT_OF_RANGE, "%d should gt 0", index); - PreConditions.check(index >= ledgerBeginIndex, DLedgerResponseCode.INDEX_LESS_THAN_LOCAL_BEGIN, "%d should be gt %d, ledgerBeginIndex may be revised", index, ledgerBeginIndex); - PreConditions.check(index <= ledgerEndIndex, DLedgerResponseCode.INDEX_OUT_OF_RANGE, "%d should between %d-%d", index, ledgerBeginIndex, ledgerEndIndex); + PreConditions.check(index > ledgerBeforeBeginIndex, DLedgerResponseCode.INDEX_LESS_THAN_LOCAL_BEGIN, "%d should be gt %d, beforeBeginIndex may be revised", index, ledgerBeforeBeginIndex); + PreConditions.check(index <= ledgerEndIndex, DLedgerResponseCode.INDEX_OUT_OF_RANGE, "%d should between (%d-%d]", index, ledgerBeforeBeginIndex, ledgerEndIndex); } @Override @@ -600,8 +645,8 @@ public void updateCommittedIndex(long term, long newCommittedIndex) { return; } if (newCommittedIndex < this.committedIndex - || newCommittedIndex < this.ledgerBeginIndex) { - LOGGER.warn("[MONITOR]Skip update committed index for new={} < old={} or new={} < beginIndex={}", newCommittedIndex, this.committedIndex, newCommittedIndex, this.ledgerBeginIndex); + || newCommittedIndex <= this.ledgerBeforeBeginIndex) { + LOGGER.warn("[MONITOR]Skip update committed index for new={} < old={} or new={} <= beforeBeginIndex={}", newCommittedIndex, this.committedIndex, newCommittedIndex, this.ledgerBeforeBeginIndex); return; } long endIndex = ledgerEndIndex; @@ -727,7 +772,7 @@ public CleanSpaceService(String name, Logger logger) { count, timeUp, checkExpired, forceClean, enableForceClean, isDiskFull, storeBaseRatio, dataRatio); } if (count > 0) { - DLedgerMmapFileStore.this.reviseLedgerBeginIndex(); + DLedgerMmapFileStore.this.reviseLedgerBeforeBeginIndex(); } } getDataFileList().retryDeleteFirstFile(intervalForcibly); diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/AppendAndPushTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/AppendAndPushTest.java index 09c4e693..09c4be25 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/AppendAndPushTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/AppendAndPushTest.java @@ -144,9 +144,9 @@ public void testPushNetworkNotStable() throws Exception { Thread.sleep(1500); Assertions.assertTrue(sendSuccess.get()); - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(1, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(1, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); } @@ -176,10 +176,10 @@ public void testPushMissed() throws Exception { Assertions.assertEquals(appendEntryResponse.getCode(), DLedgerResponseCode.SUCCESS.getCode()); Assertions.assertEquals(i, appendEntryResponse.getIndex()); } - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); } @@ -194,7 +194,7 @@ public void testPushTruncate() throws Exception { DLedgerEntry resEntry = dLedgerServer0.getDLedgerStore().appendAsLeader(entry); Assertions.assertEquals(i, resEntry.getIndex()); } - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); List entries = new ArrayList<>(); for (long i = 0; i < 10; i++) { @@ -213,9 +213,9 @@ public void testPushTruncate() throws Exception { dLedgerServer1 = launchServer(group, peers, "n1", "n1", DLedgerConfig.FILE); dLedgerServer0 = launchServer(group, peers, "n0", "n1", DLedgerConfig.FILE); Thread.sleep(1000); - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(4, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(4, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); for (int i = 0; i < 10; i++) { AppendEntryRequest request = new AppendEntryRequest(); @@ -261,10 +261,10 @@ public void testBatchAppend() throws Exception { Assertions.assertEquals(appendEntryResponse.getCode(), DLedgerResponseCode.SUCCESS.getCode()); Assertions.assertEquals(count - 1, appendEntryResponse.getIndex()); - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(count - 1, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(count - 1, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); Thread.sleep(1000); } diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/BatchPushTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/BatchPushTest.java index 9434d41d..040ccea9 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/BatchPushTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/BatchPushTest.java @@ -178,9 +178,9 @@ public void testBatchPushNetworkNotStable() throws Exception { Thread.sleep(1500); Assertions.assertTrue(sendSuccess.get()); - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(1, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(1, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); } @@ -210,10 +210,10 @@ public void testBatchPushMissed() throws Exception { Assertions.assertEquals(appendEntryResponse.getCode(), DLedgerResponseCode.SUCCESS.getCode()); Assertions.assertEquals(i, appendEntryResponse.getIndex()); } - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); } @@ -228,7 +228,7 @@ public void testBatchPushTruncate() throws Exception { DLedgerEntry resEntry = dLedgerServer0.getDLedgerStore().appendAsLeader(entry); Assertions.assertEquals(i, resEntry.getIndex()); } - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); List entries = new ArrayList<>(); for (long i = 0; i < 10; i++) { @@ -247,9 +247,9 @@ public void testBatchPushTruncate() throws Exception { dLedgerServer1 = launchServerEnableBatchPush(group, peers, "n1", "n1", DLedgerConfig.FILE); dLedgerServer0 = launchServerEnableBatchPush(group, peers, "n0", "n1", DLedgerConfig.FILE); Thread.sleep(1000); - Assertions.assertEquals(0, dLedgerServer0.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer0.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(4, dLedgerServer0.getDLedgerStore().getLedgerEndIndex()); - Assertions.assertEquals(0, dLedgerServer1.getDLedgerStore().getLedgerBeginIndex()); + Assertions.assertEquals(-1, dLedgerServer1.getDLedgerStore().getLedgerBeforeBeginIndex()); Assertions.assertEquals(4, dLedgerServer1.getDLedgerStore().getLedgerEndIndex()); for (int i = 0; i < 10; i++) { AppendEntryRequest request = new AppendEntryRequest(); diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java index a71334b7..bd9b11c0 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/snapshot/SnapshotManagerTest.java @@ -41,21 +41,35 @@ public void testSaveAndLoadSnapshot() throws InterruptedException { }; // Launch client DLedgerClient dLedgerClient = launchClient(group, peers.split(";")[0]); - for (int i = 0; i < 100; i++) { + // append 99 entries, each 10 entries will trigger one snapshotting + for (int i = 0; i < 99; i++) { AppendEntryResponse appendEntryResponse = dLedgerClient.append(new byte[512]); assertEquals(DLedgerResponseCode.SUCCESS.getCode(), appendEntryResponse.getCode()); assertEquals(i, appendEntryResponse.getIndex()); } - Thread.sleep(1200); + Thread.sleep(2000); for (DLedgerServer server : serverList) { - assertEquals(99, server.getdLedgerStore().getLedgerEndIndex()); + assertEquals(98, server.getDLedgerStore().getLedgerEndIndex()); + assertEquals(89, server.getDLedgerStore().getLedgerBeforeBeginIndex()); + // check statemachine + final MockStateMachine fsm = (MockStateMachine) server.getStateMachine(); + assertEquals(99, fsm.getTotalEntries()); } - // Check state machine + + // now we append an entry will trigger the snapshotting + // this time will delete entries on a scale of 90 to 99 + AppendEntryResponse appendEntryResponse = dLedgerClient.append(new byte[512]); + assertEquals(DLedgerResponseCode.SUCCESS.getCode(), appendEntryResponse.getCode()); + assertEquals(99, appendEntryResponse.getIndex()); + Thread.sleep(2000); for (DLedgerServer server : serverList) { + assertEquals(99, server.getDLedgerStore().getLedgerEndIndex()); + assertEquals(99, server.getDLedgerStore().getLedgerBeforeBeginIndex()); + // check statemachine final MockStateMachine fsm = (MockStateMachine) server.getStateMachine(); - assertEquals(99, fsm.getAppliedIndex()); assertEquals(100, fsm.getTotalEntries()); } + Thread.sleep(100); // Shutdown server dLedgerServer0.shutdown(); @@ -69,11 +83,13 @@ public void testSaveAndLoadSnapshot() throws InterruptedException { serverList.add(newDLedgerServer0); serverList.add(newDLedgerServer1); serverList.add(newDLedgerServer2); - Thread.sleep(1000); + Thread.sleep(2000); // State machine could only be recovered from snapshot due to the entry has been removed after saving snapshot for (DLedgerServer server : serverList) { + assertEquals(99, server.getDLedgerStore().getLedgerEndIndex()); + assertEquals(99, server.getDLedgerStore().getLedgerBeforeBeginIndex()); + // check statemachine final MockStateMachine fsm = (MockStateMachine) server.getStateMachine(); - assertEquals(99, server.getFsmCaller().getLastAppliedIndex()); assertEquals(100, fsm.getTotalEntries()); } } diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/MockStateMachine.java b/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/MockStateMachine.java index dcddd9de..2fa33bd8 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/MockStateMachine.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/statemachine/MockStateMachine.java @@ -31,7 +31,6 @@ public class MockStateMachine implements StateMachine { private static Logger logger = LoggerFactory.getLogger(MockStateMachine.class); private volatile long appliedIndex = -1L; private final AtomicLong totalEntries = new AtomicLong(0); - private final AtomicLong lastAppliedIndex = new AtomicLong(-1); @Override public void onApply(final CommittedEntryIterator iter) { @@ -43,6 +42,8 @@ public void onApply(final CommittedEntryIterator iter) { } this.totalEntries.addAndGet(1); this.appliedIndex = next.getIndex(); + System.out.println("apply index: " + next.getIndex()); + System.out.println("total entries: " + this.totalEntries.get()); } } } @@ -51,6 +52,7 @@ public void onApply(final CommittedEntryIterator iter) { public boolean onSnapshotSave(final SnapshotWriter writer) { long curEntryCnt = this.totalEntries.get(); MockSnapshotFile snapshotFile = new MockSnapshotFile(writer.getSnapshotStorePath() + File.separator + SnapshotManager.SNAPSHOT_DATA_FILE); + System.out.println("save snapshot, total entries: " + curEntryCnt); return snapshotFile.save(curEntryCnt); } diff --git a/dledger/src/test/java/io/openmessaging/storage/dledger/store/DLedgerMappedFileStoreTest.java b/dledger/src/test/java/io/openmessaging/storage/dledger/store/DLedgerMappedFileStoreTest.java index 0a09ed32..45bf12a6 100644 --- a/dledger/src/test/java/io/openmessaging/storage/dledger/store/DLedgerMappedFileStoreTest.java +++ b/dledger/src/test/java/io/openmessaging/storage/dledger/store/DLedgerMappedFileStoreTest.java @@ -108,7 +108,7 @@ public void testCommittedIndex() throws Exception { } fileStore.shutdown(); fileStore = createFileStore(group, peers, "n0", "n0"); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(99, fileStore.getLedgerEndIndex()); Assertions.assertEquals(90, fileStore.getCommittedIndex()); } @@ -185,7 +185,7 @@ public void testNormalRecovery() { } fileStore.shutdown(); fileStore = createFileStore(group, peers, "n0", "n0"); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, fileStore.getLedgerEndIndex()); for (long i = 0; i < 10; i++) { DLedgerEntry entry = fileStore.get(i); @@ -208,7 +208,7 @@ public void testAbnormalRecovery() { } Assertions.assertEquals(12, fileStore.getDataFileList().getMappedFiles().size()); Assertions.assertEquals(99, fileStore.getLedgerEndIndex()); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); while (fileStore.getFlushPos() != fileStore.getWritePos()) { fileStore.flush(); } @@ -217,7 +217,7 @@ public void testAbnormalRecovery() { { DLedgerMmapFileStore fileStore = createFileStore(group, peers, "n0", "n0", 10 * 1024 + MIN_BLANK_LEN, 10 * DLedgerMmapFileStore.INDEX_UNIT_SIZE, 2); Assertions.assertEquals(10, fileStore.getDataFileList().getMappedFiles().size()); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(89, fileStore.getLedgerEndIndex()); for (long i = 0; i < 89; i++) { DLedgerEntry entry = fileStore.get(i); @@ -228,7 +228,7 @@ public void testAbnormalRecovery() { { DLedgerMmapFileStore fileStore = createFileStore(group, peers, "n0", "n0", 10 * 1024 + MIN_BLANK_LEN, 10 * DLedgerMmapFileStore.INDEX_UNIT_SIZE, 10); Assertions.assertEquals(0, fileStore.getDataFileList().getMappedFiles().size()); - Assertions.assertEquals(-1, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(-1, fileStore.getLedgerEndIndex()); fileStore.shutdown(); } @@ -246,7 +246,7 @@ public void testTruncate() { Assertions.assertEquals(i, resEntry.getIndex()); } Assertions.assertEquals(2, fileStore.getDataFileList().getMappedFiles().size()); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, fileStore.getLedgerEndIndex()); fileStore.getMemberState().changeToFollower(fileStore.getLedgerEndTerm(), "n0"); @@ -258,7 +258,7 @@ public void testTruncate() { Assertions.assertNotNull(midEntry); long midIndex = fileStore.truncate(midEntry, fileStore.getLedgerEndTerm(), "n0"); Assertions.assertEquals(5, midIndex); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(5, fileStore.getLedgerEndIndex()); Assertions.assertEquals(midEntry.getPos() + midEntry.getSize(), fileStore.getDataFileList().getMaxWrotePosition()); Assertions.assertEquals((midIndex + 1) * DLedgerMmapFileStore.INDEX_UNIT_SIZE, fileStore.getIndexFileList().getMaxWrotePosition()); @@ -269,7 +269,7 @@ public void testTruncate() { Assertions.assertNotNull(afterEntry); long afterIndex = fileStore.truncate(afterEntry, fileStore.getLedgerEndTerm(), "n0"); Assertions.assertEquals(6, afterIndex); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(6, fileStore.getLedgerEndIndex()); Assertions.assertEquals(afterEntry.getPos() + afterEntry.getSize(), fileStore.getDataFileList().getMaxWrotePosition()); Assertions.assertEquals((afterIndex + 1) * DLedgerMmapFileStore.INDEX_UNIT_SIZE, fileStore.getIndexFileList().getMaxWrotePosition()); @@ -282,7 +282,7 @@ public void testTruncate() { long endIndex = fileStore.truncate(endEntry, fileStore.getLedgerEndTerm(), "n0"); Assertions.assertEquals(9, endIndex); Assertions.assertEquals(9, fileStore.getLedgerEndIndex()); - Assertions.assertEquals(9, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(8, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(endEntry.getPos() + endEntry.getSize(), fileStore.getDataFileList().getMaxWrotePosition()); Assertions.assertEquals((endIndex + 1) * DLedgerMmapFileStore.INDEX_UNIT_SIZE, fileStore.getIndexFileList().getMaxWrotePosition()); } @@ -302,7 +302,7 @@ public void testResetOffsetAndRecover() { Assertions.assertEquals(i, resEntry.getIndex()); } Assertions.assertEquals(10, fileStore.getDataFileList().getMappedFiles().size()); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, fileStore.getLedgerEndIndex()); // reset offset, discard the first 9 entries @@ -324,7 +324,7 @@ public void testResetOffsetAndRecover() { fileStore.shutdown(); fileStore = createFileStore(group, peers, "n0", "n0", 1024, 1024, 0); Assertions.assertEquals(1, fileStore.getDataFileList().getMappedFiles().size()); - Assertions.assertEquals(9, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(8, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(9, fileStore.getLedgerEndIndex()); } @@ -364,7 +364,7 @@ public void testReviseWherePosition() throws Exception { fileStore.getDataFileList().flush(0); Assertions.assertEquals(3, fileStore.getDataFileList().getMappedFiles().size()); - Assertions.assertEquals(0, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(-1, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(19, fileStore.getLedgerEndIndex()); fileStore.getMemberState().changeToFollower(fileStore.getLedgerEndTerm(), "n0"); @@ -379,7 +379,7 @@ public void testReviseWherePosition() throws Exception { Assertions.assertNotNull(entry); long index = fileStore.truncate(entry, fileStore.getLedgerEndTerm(), "n0"); Assertions.assertEquals(15, index); - Assertions.assertEquals(14, fileStore.getLedgerBeginIndex()); + Assertions.assertEquals(13, fileStore.getLedgerBeforeBeginIndex()); Assertions.assertEquals(15, fileStore.getLedgerEndIndex()); Assertions.assertEquals(entry.getPos() + entry.getSize(), fileStore.getDataFileList().getMaxWrotePosition()); Assertions.assertEquals((index + 1) * DLedgerMmapFileStore.INDEX_UNIT_SIZE, fileStore.getIndexFileList().getMaxWrotePosition());