This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch cluster_new
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/cluster_new by this push:
new 3c4de8d use binary search when check the matched index when snapshot
new 8b75116 Merge pull request #1836 from
neuyilan/apache_cluster_new_1020_binary_search_matchterm
3c4de8d is described below
commit 3c4de8d7fbe532141f2fc4319dfd6eaa09995211
Author: HouliangQi <[email protected]>
AuthorDate: Tue Oct 20 10:51:31 2020 +0800
use binary search when check the matched index when snapshot
---
.../iotdb/cluster/log/catchup/CatchUpTask.java | 65 +++++++++++++--------
.../iotdb/cluster/server/member/RaftMember.java | 2 +-
.../iotdb/cluster/log/catchup/CatchUpTaskTest.java | 67 +++++++++++++++++++++-
.../java/org/apache/iotdb/db/metadata/MTree.java | 1 -
4 files changed, 106 insertions(+), 29 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/log/catchup/CatchUpTask.java
b/cluster/src/main/java/org/apache/iotdb/cluster/log/catchup/CatchUpTask.java
index 7758187..90d493f 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/log/catchup/CatchUpTask.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/log/catchup/CatchUpTask.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.cluster.server.NodeCharacter;
import org.apache.iotdb.cluster.server.Peer;
import org.apache.iotdb.cluster.server.member.RaftMember;
import org.apache.iotdb.cluster.utils.ClientUtils;
+import org.apache.iotdb.db.utils.TestOnly;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,18 +92,40 @@ public class CatchUpTask implements Runnable {
logger.error("Unexpected error in logManager's getEntries during
matchIndexCheck", e);
}
- int index = logs.size() - 1;
- // if index < 0 then send Snapshot and all the logs in logManager
- // if index >= 0 but there is no matched log, still send Snapshot and all
the logs in logManager
- while (index >= 0) {
- if (checkMatchIndex(index)) {
- logger.info("{}: Find a match index {} of {}", raftMember.getName(),
index, node);
- return true;
+ int index = findLastMatchIndex(logs);
+ if (index == -1) {
+ logger.info("Cannot find matched of {} within [{}, {}]", node, lo, hi);
+ return false;
+ }
+
+ // if follower return RESPONSE.AGREE with this empty log, then start
sending real logs from index.
+ logs.subList(0, index).clear();
+ if (logger.isDebugEnabled()) {
+ if (logs.isEmpty()) {
+ logger.debug("{}: {} has caught up by previous catch up",
raftMember.getName(), node);
+ } else {
+ logger.debug("{}: makes {} catch up with {} and other {} logs",
raftMember.getName(),
+ node, logs.get(0), logs.size());
+ }
+ }
+ return true;
+ }
+
+ public int findLastMatchIndex(List<Log> logs)
+ throws LeaderUnknownException, TException, InterruptedException {
+ int start = 0;
+ int end = logs.size() - 1;
+ int matchedIndex = -1;
+ while (start <= end) {
+ int mid = start + (end - start) / 2;
+ if (checkMatchIndex(mid)) {
+ start = mid + 1;
+ matchedIndex = mid;
+ } else {
+ end = mid - 1;
}
- index--;
}
- logger.info("Cannot find matched of {} within [{}, {}]", node, lo, hi);
- return false;
+ return matchedIndex;
}
/**
@@ -115,7 +138,6 @@ public class CatchUpTask implements Runnable {
*/
private boolean checkMatchIndex(int index)
throws LeaderUnknownException, TException, InterruptedException {
- boolean isLogDebug = logger.isDebugEnabled();
Log log = logs.get(index);
synchronized (raftMember.getTerm()) {
// make sure this node is still a leader
@@ -136,20 +158,7 @@ public class CatchUpTask implements Runnable {
raftMember.getLastCatchUpResponseTime().put(node,
System.currentTimeMillis());
logger.debug("{} check {}'s matchIndex {} with log [{}]",
raftMember.getName(), node,
matched ? "succeed" : "failed", log);
- if (!matched) {
- return false;
- }
- // if follower return RESPONSE.AGREE with this empty log, then start
sending real logs from index.
- logs.subList(0, index).clear();
- if (isLogDebug) {
- if (logs.isEmpty()) {
- logger.debug("{}: {} has caught up by previous catch up",
raftMember.getName(), node);
- } else {
- logger.debug("{}: makes {} catch up with {} and other {} logs",
raftMember.getName(),
- node, logs.get(0), logs.size());
- }
- }
- return true;
+ return matched;
}
/**
@@ -211,6 +220,7 @@ public class CatchUpTask implements Runnable {
}
}
+ @Override
public void run() {
try {
boolean findMatchedIndex = checkMatchIndex();
@@ -250,4 +260,9 @@ public class CatchUpTask implements Runnable {
// the next catch up is enabled
raftMember.getLastCatchUpResponseTime().remove(node);
}
+
+ @TestOnly
+ public void setLogs(List<Log> logs) {
+ this.logs = logs;
+ }
}
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
index 14b1610..c56c1f6 100644
---
a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
+++
b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java
@@ -232,7 +232,7 @@ public abstract class RaftMember {
*/
private LogDispatcher logDispatcher;
- public RaftMember() {
+ protected RaftMember() {
}
protected RaftMember(String name, AsyncClientPool asyncPool, SyncClientPool
syncPool,
diff --git
a/cluster/src/test/java/org/apache/iotdb/cluster/log/catchup/CatchUpTaskTest.java
b/cluster/src/test/java/org/apache/iotdb/cluster/log/catchup/CatchUpTaskTest.java
index 0df45ae..6620fb0 100644
---
a/cluster/src/test/java/org/apache/iotdb/cluster/log/catchup/CatchUpTaskTest.java
+++
b/cluster/src/test/java/org/apache/iotdb/cluster/log/catchup/CatchUpTaskTest.java
@@ -31,10 +31,10 @@ import org.apache.iotdb.cluster.common.TestMetaGroupMember;
import org.apache.iotdb.cluster.common.TestSyncClient;
import org.apache.iotdb.cluster.common.TestUtils;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
+import org.apache.iotdb.cluster.exception.LeaderUnknownException;
import org.apache.iotdb.cluster.exception.LogExecutionException;
import org.apache.iotdb.cluster.log.Log;
import org.apache.iotdb.cluster.log.LogParser;
-import org.apache.iotdb.cluster.log.Snapshot;
import org.apache.iotdb.cluster.log.logtypes.EmptyContentLog;
import org.apache.iotdb.cluster.partition.PartitionTable;
import org.apache.iotdb.cluster.partition.slot.SlotPartitionTable;
@@ -49,8 +49,10 @@ import org.apache.iotdb.cluster.server.Peer;
import org.apache.iotdb.cluster.server.Response;
import org.apache.iotdb.cluster.server.member.RaftMember;
import org.apache.iotdb.db.service.IoTDB;
+import org.apache.thrift.TException;
import org.apache.thrift.async.AsyncMethodCallback;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -265,7 +267,8 @@ public class CatchUpTaskTest {
}
sender.getLogManager().append(logList);
sender.getLogManager().commitTo(9);
-
sender.getLogManager().setMaxHaveAppliedCommitIndex(sender.getLogManager().getCommitLogIndex());
+ sender.getLogManager()
+
.setMaxHaveAppliedCommitIndex(sender.getLogManager().getCommitLogIndex());
Node receiver = new Node();
sender.setCharacter(NodeCharacter.LEADER);
Peer peer = new Peer(10);
@@ -329,4 +332,64 @@ public class CatchUpTaskTest {
assertEquals(logList, receivedLogs.subList(1, receivedLogs.size()));
assertEquals(9, leaderCommit);
}
+
+ @Test
+ public void testFindLastMatchIndex() throws LogExecutionException {
+ List<Log> logList = new ArrayList<>();
+ int lastMatchedIndex = 6;
+ for (int i = 0; i < 10; i++) {
+ Log log = new EmptyContentLog();
+ log.setCurrLogIndex(i);
+ log.setCurrLogTerm(i);
+ logList.add(log);
+
+ if (i < lastMatchedIndex) {
+ receivedLogs.add(log);
+ }
+ }
+ sender.getLogManager().append(logList);
+ sender.getLogManager().commitTo(9);
+ sender.getLogManager()
+
.setMaxHaveAppliedCommitIndex(sender.getLogManager().getCommitLogIndex());
+ Node receiver = new Node();
+ sender.setCharacter(NodeCharacter.LEADER);
+ Peer peer = new Peer(10);
+ peer.setCatchUp(false);
+ peer.setMatchIndex(0);
+ peer.setNextIndex(0);
+
+ CatchUpTask task = new CatchUpTask(receiver, peer, sender);
+ task.setLogs(logList);
+ try {
+ // 1. case 1: the matched index is in the middle of the logs interval
+ int resultMatchIndex = task.findLastMatchIndex(logList);
+ assertEquals(lastMatchedIndex, resultMatchIndex);
+
+ // 2. case 2: no matched index case
+ lastMatchedIndex = -1;
+ receivedLogs.subList(1, receivedLogs.size()).clear();
+ resultMatchIndex = task.findLastMatchIndex(logList);
+ assertEquals(lastMatchedIndex, resultMatchIndex);
+
+ // 3. case 3: the matched index is at the last index of the logs
+ logList.clear();
+ receivedLogs.subList(1, receivedLogs.size()).clear();
+ lastMatchedIndex = 9;
+ for (int i = 0; i < 10; i++) {
+ Log log = new EmptyContentLog();
+ log.setCurrLogIndex(i);
+ log.setCurrLogTerm(i);
+ logList.add(log);
+
+ if (i < lastMatchedIndex) {
+ receivedLogs.add(log);
+ }
+ }
+ resultMatchIndex = task.findLastMatchIndex(logList);
+ assertEquals(lastMatchedIndex, resultMatchIndex);
+
+ } catch (LeaderUnknownException | TException | InterruptedException e) {
+ Assert.fail(e.getMessage());
+ }
+ }
}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
index 5eed13b..8330f79 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
@@ -36,7 +36,6 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayDeque;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.Deque;
