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 2021/12/09 01:17:41 UTC
[iotdb] branch expr updated: add configs to file
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch expr
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/expr by this push:
new 30a7c52 add configs to file
30a7c52 is described below
commit 30a7c52365dc984fb80eb89959db5e4c21d709d9
Author: jt <jt...@163.com>
AuthorDate: Thu Dec 9 09:16:52 2021 +0800
add configs to file
---
.../assembly/resources/conf/iotdb-cluster.properties | 4 +++-
.../java/org/apache/iotdb/cluster/ClusterIoTDB.java | 6 +++---
.../org/apache/iotdb/cluster/config/ClusterConfig.java | 10 ++++++++++
.../apache/iotdb/cluster/config/ClusterDescriptor.java | 17 +++++++++++++++++
.../org/apache/iotdb/cluster/log/LogDispatcher.java | 6 +++---
.../iotdb/cluster/log/appender/BlockingLogAppender.java | 5 +++++
.../apache/iotdb/cluster/log/appender/LogAppender.java | 2 ++
.../cluster/log/appender/SlidingWindowLogAppender.java | 12 ++++++++++--
.../iotdb/cluster/log/snapshot/PartitionedSnapshot.java | 1 +
.../iotdb/cluster/server/member/DataGroupMember.java | 3 ---
.../apache/iotdb/cluster/server/member/RaftMember.java | 4 ++--
.../cluster/server/member/MetaGroupMemberTest.java | 4 ----
12 files changed, 56 insertions(+), 18 deletions(-)
diff --git a/cluster/src/assembly/resources/conf/iotdb-cluster.properties b/cluster/src/assembly/resources/conf/iotdb-cluster.properties
index 48254e7..65cbccd 100644
--- a/cluster/src/assembly/resources/conf/iotdb-cluster.properties
+++ b/cluster/src/assembly/resources/conf/iotdb-cluster.properties
@@ -197,4 +197,6 @@ multi_raft_factor=1
# appended or committed.
# Up to max_num_of_logs_in_mem * 2 requests may be lost when the client and the leader fail
# simultaneously, so be extremely careful before turning it on.
-# enable_weak_acceptance=true
\ No newline at end of file
+# enable_weak_acceptance=true
+
+# dispatcher_binding_thread_num=16
\ No newline at end of file
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java
index c5a6713..cd4e8d8 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/ClusterIoTDB.java
@@ -206,12 +206,12 @@ public class ClusterIoTDB implements ClusterIoTDBMBean {
* node. This will help to see if the node is in a consistent and right state during debugging.
*/
private void generateNodeReport() {
- if (logger.isDebugEnabled() && allowReport) {
+ if (logger.isInfoEnabled() && allowReport) {
try {
NodeReport report = new NodeReport(thisNode);
report.setMetaMemberReport(metaGroupMember.genMemberReport());
report.setDataMemberReportList(dataGroupEngine.genMemberReports());
- logger.debug(report.toString());
+ logger.info(report.toString());
} catch (Exception e) {
logger.error("exception occurred when generating node report", e);
}
@@ -336,7 +336,7 @@ public class ClusterIoTDB implements ClusterIoTDBMBean {
}
private void preInitCluster() throws StartupException {
- stopRaftInfoReport();
+ // stopRaftInfoReport();
JMXService.registerMBean(this, mbeanName);
// register MetaGroupMember. MetaGroupMember has the same position with "StorageEngine" in the
// cluster module.
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
index e746b3a..d7c8ae7 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConfig.java
@@ -188,6 +188,8 @@ public class ClusterConfig {
private boolean enableWeakAcceptance = true;
+ private int dispatcherBindingThreadNum = 16;
+
/**
* create a clusterConfig class. The internalIP will be set according to the server's hostname. If
* there is something error for getting the ip of the hostname, then set the internalIp as
@@ -570,4 +572,12 @@ public class ClusterConfig {
public void setEnableWeakAcceptance(boolean enableWeakAcceptance) {
this.enableWeakAcceptance = enableWeakAcceptance;
}
+
+ public int getDispatcherBindingThreadNum() {
+ return dispatcherBindingThreadNum;
+ }
+
+ public void setDispatcherBindingThreadNum(int dispatcherBindingThreadNum) {
+ this.dispatcherBindingThreadNum = dispatcherBindingThreadNum;
+ }
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
index 0ebac8c..9f76e15 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterDescriptor.java
@@ -318,6 +318,23 @@ public class ClusterDescriptor {
properties.getProperty(
"wait_client_timeout_ms", String.valueOf(config.getWaitClientTimeoutMS()))));
+ config.setUseFollowerSlidingWindow(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "use_follower_sliding_window",
+ String.valueOf(config.isUseFollowerSlidingWindow()))));
+
+ config.setEnableWeakAcceptance(
+ Boolean.parseBoolean(
+ properties.getProperty(
+ "enable_weak_acceptance", String.valueOf(config.isEnableWeakAcceptance()))));
+
+ config.setDispatcherBindingThreadNum(
+ Integer.parseInt(
+ properties.getProperty(
+ "dispatcher_binding_thread_num",
+ String.valueOf(config.getDispatcherBindingThreadNum()))));
+
String consistencyLevel = properties.getProperty("consistency_level");
if (consistencyLevel != null) {
config.setConsistencyLevel(ConsistencyLevel.getConsistencyLevel(consistencyLevel));
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/LogDispatcher.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/LogDispatcher.java
index bfb2295..ba8f55f 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/LogDispatcher.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/LogDispatcher.java
@@ -67,7 +67,7 @@ public class LogDispatcher {
private static final Logger logger = LoggerFactory.getLogger(LogDispatcher.class);
RaftMember member;
- private ClusterConfig clusterConfig = ClusterDescriptor.getInstance().getConfig();
+ private static final ClusterConfig clusterConfig = ClusterDescriptor.getInstance().getConfig();
private boolean useBatchInLogCatchUp = clusterConfig.isUseBatchInLogCatchUp();
List<BlockingQueue<SendLogRequest>> nodesLogQueues = new ArrayList<>();
ExecutorService executorService;
@@ -75,8 +75,8 @@ public class LogDispatcher {
IoTDBThreadPoolFactory.newFixedThreadPoolWithDaemonThread(
Runtime.getRuntime().availableProcessors(), "DispatcherEncoder");
- public static int bindingThreadNum = 1;
- public static int maxBatchSize = 1;
+ public static int bindingThreadNum = clusterConfig.getDispatcherBindingThreadNum();
+ public static int maxBatchSize = 10;
public LogDispatcher(RaftMember member) {
this.member = member;
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
index 7cb614a..d2d9056 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/BlockingLogAppender.java
@@ -172,6 +172,11 @@ public class BlockingLogAppender implements LogAppender {
return result;
}
+ @Override
+ public void reset() {
+ // no states maintained by this implementation
+ }
+
public static class Factory implements LogAppenderFactory {
@Override
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
index b5bd35f..01f16da 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/LogAppender.java
@@ -34,4 +34,6 @@ public interface LogAppender {
long prevLogIndex, long prevLogTerm, long leaderCommit, List<Log> logs);
AppendEntryResult appendEntry(long prevLogIndex, long prevLogTerm, long leaderCommit, Log log);
+
+ void reset();
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
index d2ecb1c..29a2689 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/appender/SlidingWindowLogAppender.java
@@ -46,8 +46,7 @@ public class SlidingWindowLogAppender implements LogAppender {
public SlidingWindowLogAppender(RaftMember member) {
this.member = member;
this.logManager = member.getLogManager();
- this.firstPosPrevIndex = logManager.getLastLogIndex();
- this.prevTerms[0] = logManager.getLastLogTerm();
+ reset();
}
/**
@@ -202,6 +201,15 @@ public class SlidingWindowLogAppender implements LogAppender {
return result;
}
+ @Override
+ public void reset() {
+ this.firstPosPrevIndex = logManager.getLastLogIndex();
+ this.prevTerms[0] = logManager.getLastLogTerm();
+ logWindow = new Log[windowCapacity];
+ prevTerms = new long[windowCapacity];
+ windowLength = 0;
+ }
+
public static class Factory implements LogAppenderFactory {
@Override
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshot.java b/cluster/src/main/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshot.java
index e346d9c..f9f045b 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshot.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/log/snapshot/PartitionedSnapshot.java
@@ -179,6 +179,7 @@ public class PartitionedSnapshot<T extends Snapshot> extends Snapshot {
}
synchronized (dataGroupMember.getLogManager()) {
dataGroupMember.getLogManager().applySnapshot(snapshot);
+ dataGroupMember.getLogAppender().reset();
}
}
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
index 104adfa..8ef416b 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/DataGroupMember.java
@@ -24,20 +24,17 @@ import org.apache.iotdb.cluster.client.ClientManager;
import org.apache.iotdb.cluster.config.ClusterConstant;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
import org.apache.iotdb.cluster.exception.CheckConsistencyException;
-import org.apache.iotdb.cluster.exception.LogExecutionException;
import org.apache.iotdb.cluster.exception.SnapshotInstallationException;
import org.apache.iotdb.cluster.exception.UnknownLogTypeException;
import org.apache.iotdb.cluster.log.Log;
import org.apache.iotdb.cluster.log.LogApplier;
import org.apache.iotdb.cluster.log.LogParser;
import org.apache.iotdb.cluster.log.Snapshot;
-import org.apache.iotdb.cluster.log.VotingLog;
import org.apache.iotdb.cluster.log.appender.BlockingLogAppender;
import org.apache.iotdb.cluster.log.appender.SlidingWindowLogAppender;
import org.apache.iotdb.cluster.log.applier.AsyncDataLogApplier;
import org.apache.iotdb.cluster.log.applier.DataLogApplier;
import org.apache.iotdb.cluster.log.logtypes.AddNodeLog;
-import org.apache.iotdb.cluster.log.logtypes.CloseFileLog;
import org.apache.iotdb.cluster.log.logtypes.RemoveNodeLog;
import org.apache.iotdb.cluster.log.manage.FilePartitionedSnapshotLogManager;
import org.apache.iotdb.cluster.log.manage.PartitionedSnapshotLogManager;
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 0dcc4ce..94bf84a 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
@@ -136,7 +136,7 @@ import static org.apache.iotdb.cluster.config.ClusterConstant.THREAD_POLL_WAIT_T
public abstract class RaftMember implements RaftMemberMBean {
private static final Logger logger = LoggerFactory.getLogger(RaftMember.class);
- public static boolean USE_LOG_DISPATCHER = false;
+ public static boolean USE_LOG_DISPATCHER = true;
private static final boolean USE_INDIRECT_LOG_DISPATCHER =
ClusterDescriptor.getInstance().getConfig().isUseIndirectBroadcasting();
private static final boolean ENABLE_WEAK_ACCEPTANCE =
@@ -1666,7 +1666,7 @@ public abstract class RaftMember implements RaftMemberMBean {
long alreadyWait = 0;
while (stronglyAcceptedNodeNum < quorumSize
&& (!(ENABLE_WEAK_ACCEPTANCE && canBeWeaklyAccepted(log.getLog()))
- || (totalAccepted < allNodes.size() - 1)
+ || (totalAccepted < quorumSize)
|| votingLogList.size() > config.getMaxNumOfLogsInMem())
&& alreadyWait < ClusterConstant.getWriteOperationTimeoutMS()
&& !log.getStronglyAcceptedNodeIds().contains(Integer.MAX_VALUE)) {
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
index b9d1efa..eaab695 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/server/member/MetaGroupMemberTest.java
@@ -85,7 +85,6 @@ import org.apache.iotdb.db.auth.authorizer.LocalFileAuthorizer;
import org.apache.iotdb.db.auth.entity.Role;
import org.apache.iotdb.db.auth.entity.User;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -96,7 +95,6 @@ import org.apache.iotdb.db.metadata.path.MeasurementPath;
import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.metadata.template.Template;
import org.apache.iotdb.db.metadata.template.TemplateManager;
-import org.apache.iotdb.db.qp.executor.PlanExecutor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.sys.CreateTemplatePlan;
@@ -136,8 +134,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;