You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2020/08/27 07:46:52 UTC
[incubator-iotdb] branch cluster_receiver_wait created (now 9f342f2)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch cluster_receiver_wait
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at 9f342f2 wait for prevlog on receiver side
This branch includes the following new commits:
new 9f342f2 wait for prevlog on receiver side
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: wait for prevlog on receiver side
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch cluster_receiver_wait
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 9f342f21e15f02129185b480c4f4f920f30518b9
Author: jt2594838 <jt...@163.com>
AuthorDate: Thu Aug 27 15:46:15 2020 +0800
wait for prevlog on receiver side
---
.../iotdb/cluster/log/manage/RaftLogManager.java | 16 ++++++
.../iotdb/cluster/server/member/RaftMember.java | 67 +++++++++++++++-------
2 files changed, 63 insertions(+), 20 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
index 5d8632c..29de9f0 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/manage/RaftLogManager.java
@@ -75,6 +75,12 @@ public abstract class RaftLogManager {
private int maxNumOfLogsInMem = ClusterDescriptor.getInstance().getConfig()
.getMaxNumOfLogsInMem();
+ /**
+ * Each time new logs are appended, this condition will be notified so logs that have larger
+ * indices but arrived earlier can proceed.
+ */
+ private final Object logUpdateCondition = new Object();
+
public RaftLogManager(StableEntryManager stableEntryManager, LogApplier applier, String name) {
this.logApplier = applier;
@@ -317,6 +323,9 @@ public abstract class RaftLogManager {
return -1;
}
getUnCommittedEntryManager().truncateAndAppend(entries);
+ synchronized (logUpdateCondition) {
+ logUpdateCondition.notifyAll();
+ }
return getLastLogIndex();
}
@@ -334,6 +343,9 @@ public abstract class RaftLogManager {
return -1;
}
getUnCommittedEntryManager().truncateAndAppend(entry);
+ synchronized (logUpdateCondition) {
+ logUpdateCondition.notifyAll();
+ }
return getLastLogIndex();
}
@@ -651,4 +663,8 @@ public abstract class RaftLogManager {
logger.error("{}: regular compact log entries failed, error={}", name, e.getMessage());
}
}
+
+ public Object getLogUpdateCondition() {
+ return logUpdateCondition;
+ }
}
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 dad3a0b..69b2a53 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
@@ -430,6 +430,14 @@ public abstract class RaftMember {
*/
private long appendEntry(long prevLogIndex, long prevLogTerm, long leaderCommit, Log log) {
long resp;
+
+ long lastLogIndex = logManager.getLastLogIndex();
+ if (lastLogIndex < prevLogIndex) {
+ if (!waitForPrevLog(prevLogIndex)) {
+ return Response.RESPONSE_LOG_MISMATCH;
+ }
+ }
+
synchronized (logManager) {
long success = logManager.maybeAppend(prevLogIndex, prevLogTerm, leaderCommit, log);
if (success != -1) {
@@ -443,6 +451,25 @@ public abstract class RaftMember {
return resp;
}
+ private boolean waitForPrevLog(long prevLogIndex) {
+ long waitStart = System.currentTimeMillis();
+ long alreadyWait = 0;
+ Object logUpdateCondition = logManager.getLogUpdateCondition();
+ while (logManager.getLastLogIndex() < prevLogIndex &&
+ alreadyWait <= RaftServer.getWriteOperationTimeoutMS()) {
+ synchronized (logUpdateCondition) {
+ try {
+ logUpdateCondition.wait(100);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ alreadyWait = System.currentTimeMillis() - waitStart;
+ }
+ return alreadyWait <= RaftServer.getWriteOperationTimeoutMS();
+ }
+
/**
* Process an AppendEntryRequest. First check the term of the leader, then parse the log and
* finally see if we can find a position to append the log.
@@ -663,26 +690,26 @@ public abstract class RaftMember {
return;
}
Peer peer = peerMap.computeIfAbsent(node, k -> new Peer(logManager.getLastLogIndex()));
- long waitStart = System.currentTimeMillis();
- long alreadyWait = 0;
- while (peer.getMatchIndex() < log.getCurrLogIndex() - 1
- && character == NodeCharacter.LEADER
- && alreadyWait <= RaftServer.getWriteOperationTimeoutMS()) {
- synchronized (peer) {
- try {
- peer.wait(RaftServer.getWriteOperationTimeoutMS());
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- logger.warn("Waiting for peer to catch up interrupted");
- return;
- }
- }
- alreadyWait = System.currentTimeMillis() - waitStart;
- }
- if (alreadyWait > RaftServer.getWriteOperationTimeoutMS()) {
- logger.warn("{}: node {} timed out when appending {}", name, node, log);
- return;
- }
+// long waitStart = System.currentTimeMillis();
+// long alreadyWait = 0;
+// while (peer.getMatchIndex() < log.getCurrLogIndex() - 1
+// && character == NodeCharacter.LEADER
+// && alreadyWait <= RaftServer.getWriteOperationTimeoutMS()) {
+// synchronized (peer) {
+// try {
+// peer.wait(RaftServer.getWriteOperationTimeoutMS());
+// } catch (InterruptedException e) {
+// Thread.currentThread().interrupt();
+// logger.warn("Waiting for peer to catch up interrupted");
+// return;
+// }
+// }
+// alreadyWait = System.currentTimeMillis() - waitStart;
+// }
+// if (alreadyWait > RaftServer.getWriteOperationTimeoutMS()) {
+// logger.warn("{}: node {} timed out when appending {}", name, node, log);
+// return;
+// }
if (character != NodeCharacter.LEADER) {
return;