You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by lt...@apache.org on 2019/04/15 09:36:01 UTC
[incubator-iotdb] branch cluster_concurrent updated: fix bugs
This is an automated email from the ASF dual-hosted git repository.
lta pushed a commit to branch cluster_concurrent
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/cluster_concurrent by this push:
new 62d5766 fix bugs
62d5766 is described below
commit 62d5766f6c868ad8546ad2ff3a1f818bf82b6aa1
Author: lta <li...@163.com>
AuthorDate: Mon Apr 15 17:35:41 2019 +0800
fix bugs
---
.../cluster/concurrent/pool/RaftTaskManager.java | 66 ----------------
.../apache/iotdb/cluster/config/ClusterConfig.java | 16 ----
.../iotdb/cluster/config/ClusterConstant.java | 7 --
.../iotdb/cluster/config/ClusterDescriptor.java | 7 --
.../org/apache/iotdb/cluster/entity/Server.java | 2 -
.../cluster/entity/raft/DataStateMachine.java | 23 +++---
.../cluster/entity/raft/MetadataStateManchine.java | 90 ++++++++++++----------
.../cluster/qp/executor/NonQueryExecutor.java | 48 ++++++++----
.../org/apache/iotdb/cluster/utils/RaftUtils.java | 7 ++
iotdb/iotdb/conf/iotdb-cluster.properties | 9 +--
10 files changed, 101 insertions(+), 174 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/concurrent/pool/RaftTaskManager.java b/cluster/src/main/java/org/apache/iotdb/cluster/concurrent/pool/RaftTaskManager.java
deleted file mode 100644
index 62259cd..0000000
--- a/cluster/src/main/java/org/apache/iotdb/cluster/concurrent/pool/RaftTaskManager.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.cluster.concurrent.pool;
-
-import org.apache.iotdb.cluster.concurrent.ThreadName;
-import org.apache.iotdb.cluster.config.ClusterConfig;
-import org.apache.iotdb.cluster.config.ClusterDescriptor;
-import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
-
-/**
- * Manage all raft tasks which are applied to data state machine in thread.
- */
-public class RaftTaskManager extends ThreadPoolManager{
-
- private static final String MANAGER_NAME = "raft task manager";
-
- private RaftTaskManager() {
- init();
- }
-
- public static RaftTaskManager getInstance() {
- return RaftTaskManager.InstanceHolder.instance;
- }
-
- /**
- * Name of Pool Manager
- */
- @Override
- public String getManagerName() {
- return MANAGER_NAME;
- }
-
- @Override
- public String getThreadName() {
- return ThreadName.RAFT_TASK.getName();
- }
-
- @Override
- public int getThreadPoolSize() {
- return ClusterDescriptor.getInstance().getConfig().getConcurrentQPSubTaskThread();
- }
-
- private static class InstanceHolder {
-
- private InstanceHolder() {
- }
-
- private static RaftTaskManager instance = new RaftTaskManager();
- }
-}
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 5250742..b860835 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
@@ -137,14 +137,6 @@ public class ClusterConfig {
*/
private int concurrentQPSubTaskThread = Runtime.getRuntime().availableProcessors() * 10;
- /**
- * For each raft log, the state machine will start a thread to apply this log . This value
- * indicates maximum number of threads for a state machine to create. Currently, this param only
- * works for state machine of leader in data partition holder. Staten machine in metadata holder
- * will apply raft log serially. When this value <= 0, use CPU core number.
- */
- private int concurrentRaftTaskThread = Runtime.getRuntime().availableProcessors();
-
public ClusterConfig() {
// empty constructor
}
@@ -331,12 +323,4 @@ public class ClusterConfig {
public void setConcurrentQPSubTaskThread(int concurrentQPSubTaskThread) {
this.concurrentQPSubTaskThread = concurrentQPSubTaskThread;
}
-
- public int getConcurrentRaftTaskThread() {
- return concurrentRaftTaskThread;
- }
-
- public void setConcurrentRaftTaskThread(int concurrentRaftTaskThread) {
- this.concurrentRaftTaskThread = concurrentRaftTaskThread;
- }
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConstant.java b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConstant.java
index 280b955..5aca9b0 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConstant.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/config/ClusterConstant.java
@@ -32,13 +32,6 @@ public class ClusterConstant {
/**
* Maximum time of blocking main thread for waiting for all running task threads and tasks in the
- * queue until end. Raft Task refers to tasks that need to be performed in the state machine.The
- * unit is milliseconds. Due to guarantee data consistency, it must ensure that tasks finish.
- */
- public static final int CLOSE_RAFT_TASK_BLOCK_TIMEOUT = Integer.MAX_VALUE;
-
- /**
- * Maximum time of blocking main thread for waiting for all running task threads and tasks in the
* queue until end. Each client request corresponds to a QP Task. A QP task may be divided into
* several sub-tasks.The unit is milliseconds.
*/
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 13df817..8a49d7e 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
@@ -159,13 +159,6 @@ public class ClusterDescriptor {
conf.setConcurrentQPSubTaskThread(Runtime.getRuntime().availableProcessors() * 10);
}
- conf.setConcurrentRaftTaskThread(Integer
- .parseInt(properties.getProperty("concurrent_raft_task_thread",
- Integer.toString(conf.getConcurrentQPSubTaskThread()))));
- if (conf.getConcurrentRaftTaskThread() <= 0) {
- conf.setConcurrentRaftTaskThread(Runtime.getRuntime().availableProcessors());
- }
-
} catch (IOException e) {
LOGGER.warn("Cannot load config file because, use default configuration", e);
} catch (Exception e) {
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/entity/Server.java b/cluster/src/main/java/org/apache/iotdb/cluster/entity/Server.java
index b001073..ebda973 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/entity/Server.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/entity/Server.java
@@ -24,7 +24,6 @@ import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
import java.util.HashMap;
import java.util.Map;
import org.apache.iotdb.cluster.concurrent.pool.QPTaskManager;
-import org.apache.iotdb.cluster.concurrent.pool.RaftTaskManager;
import org.apache.iotdb.cluster.config.ClusterConfig;
import org.apache.iotdb.cluster.config.ClusterConstant;
import org.apache.iotdb.cluster.config.ClusterDescriptor;
@@ -123,7 +122,6 @@ public class Server {
}
public void stop() throws ProcessorException {
- RaftTaskManager.getInstance().close(true, ClusterConstant.CLOSE_RAFT_TASK_BLOCK_TIMEOUT);
QPTaskManager.getInstance().close(true, ClusterConstant.CLOSE_QP_SUB_TASK_BLOCK_TIMEOUT);
iotdb.deactivate();
CLIENT_MANAGER.shutdown();
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/DataStateMachine.java b/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/DataStateMachine.java
index 872046b..ebac074 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/DataStateMachine.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/DataStateMachine.java
@@ -78,13 +78,7 @@ public class DataStateMachine extends StateMachineAdapter {
while (iterator.hasNext()) {
final Closure closure = iterator.done();
final ByteBuffer data = iterator.getData();
-
-// /** It's leader to apply task **/
-// if (closure != null) {
-// RaftTaskManager.getInstance().execute(() -> applySingleTask(closure, data));
-// } else {
- applySingleTask(closure, data);
-// }
+ applySingleTask(closure, data);
iterator.next();
}
}
@@ -97,28 +91,31 @@ public class DataStateMachine extends StateMachineAdapter {
* @param data Request data
*/
private void applySingleTask(Closure closure, ByteBuffer data) {
+ /** If closure is not null, the node is leader **/
BasicResponse response = (closure == null) ? null : ((ResponseClosure) closure).getResponse();
- DataGroupNonQueryRequest request = null;
+ DataGroupNonQueryRequest request;
try {
request = SerializerManager.getSerializer(SerializerManager.Hessian2)
.deserialize(data.array(), DataGroupNonQueryRequest.class.getName());
} catch (final CodecException e) {
- LOGGER.error("Fail to decode IncrementAndGetRequest", e);
+ LOGGER.error("Fail to deserialize DataGroupNonQueryRequest", e);
+ if (closure != null) {
+ closure.run(RaftUtils.createErrorStatus(e.getMessage()));
+ }
+ return;
}
Status status = Status.OK();
- assert request != null;
-
List<byte[]> planBytes = request.getPhysicalPlanBytes();
- LOGGER.debug(String.format("State machine batch size(): %d", planBytes.size()));
+ LOGGER.debug("State machine batch size(): {}", planBytes.size());
/** handle batch plans(planBytes.size() > 0) or single plan(planBytes.size()==1) **/
for (byte[] planByte : planBytes) {
try {
PhysicalPlan plan = PhysicalPlanLogTransfer.logToOperator(planByte);
- LOGGER.debug(String.format("OperatorType :%s", plan.getOperatorType()));
+ LOGGER.debug("OperatorType :{}", plan.getOperatorType());
/** If the request is to set path and sg of the path doesn't exist, it needs to run null-read in meta group to avoid out of data sync **/
if (plan.getOperatorType() == OperatorType.CREATE_TIMESERIES && !checkPathExistence(
((MetadataPlan) plan).getPath().getFullPath())) {
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/MetadataStateManchine.java b/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/MetadataStateManchine.java
index 623b80d..9592718 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/MetadataStateManchine.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/entity/raft/MetadataStateManchine.java
@@ -47,6 +47,9 @@ import org.apache.iotdb.db.writelog.transfer.PhysicalPlanLogTransfer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * StateMachine of metadata group node.
+ */
public class MetadataStateManchine extends StateMachineAdapter {
private static final Logger LOGGER = LoggerFactory.getLogger(MetadataStateManchine.class);
@@ -78,50 +81,57 @@ public class MetadataStateManchine extends StateMachineAdapter {
@Override
public void onApply(Iterator iterator) {
while (iterator.hasNext()) {
-
- /** If closure is not null, the node is leader **/
- Closure closure = iterator.done();
- BasicResponse response = (closure==null) ? null: ((ResponseClosure)closure).getResponse();
- MetaGroupNonQueryRequest request = null;
final ByteBuffer data = iterator.getData();
- try {
- request = SerializerManager.getSerializer(SerializerManager.Hessian2)
- .deserialize(data.array(), MetaGroupNonQueryRequest.class.getName());
- } catch (final CodecException e) {
- LOGGER.error("Fail to decode IncrementAndGetRequest", e);
+ final Closure closure = iterator.done();
+ applySingleTask(closure, data);
+ iterator.next();
+ }
+ }
+
+ private void applySingleTask(Closure closure, ByteBuffer data){
+ /** If closure is not null, the node is leader **/
+ BasicResponse response = (closure==null) ? null: ((ResponseClosure)closure).getResponse();
+ MetaGroupNonQueryRequest request;
+
+ try {
+ request = SerializerManager.getSerializer(SerializerManager.Hessian2)
+ .deserialize(data.array(), MetaGroupNonQueryRequest.class.getName());
+ } catch (final CodecException e) {
+ LOGGER.error("Fail to deserialize MetadataGroupNonQueryRequest", e);
+ if(closure != null){
+ closure.run(RaftUtils.createErrorStatus(e.getMessage()));
}
- Status status = new Status();
- assert request != null;
-
- List<byte[]> planBytes = request.getPhysicalPlanBytes();
-
- /** handle batch plans(planBytes.size() > 0) or single plan(planBytes.size()==1) **/
- for (byte[] planByte : planBytes) {
- try {
- PhysicalPlan physicalPlan = PhysicalPlanLogTransfer
- .logToOperator(planByte);
- if (physicalPlan.getOperatorType() == OperatorType.SET_STORAGE_GROUP) {
- MetadataPlan plan = (MetadataPlan) physicalPlan;
- addStorageGroup(plan.getPath().getFullPath());
- } else {
- AuthorPlan plan = (AuthorPlan) physicalPlan;
- qpExecutor.processNonQuery(plan);
- }
- addResult(response, true);
- } catch (IOException | PathErrorException e) {
- LOGGER.error("Execute metadata plan error", e);
- status = new Status(-1, e.getMessage());
- addResult(response, false);
- } catch (ProcessorException e) {
- LOGGER.error("Execute author plan error", e);
- status = new Status(-1, e.getMessage());
- addResult(response, false);
+ return;
+ }
+ Status status = new Status();
+
+ List<byte[]> planBytes = request.getPhysicalPlanBytes();
+
+ /** handle batch plans(planBytes.size() > 0) or single plan(planBytes.size()==1) **/
+ for (byte[] planByte : planBytes) {
+ try {
+ PhysicalPlan physicalPlan = PhysicalPlanLogTransfer
+ .logToOperator(planByte);
+ if (physicalPlan.getOperatorType() == OperatorType.SET_STORAGE_GROUP) {
+ MetadataPlan plan = (MetadataPlan) physicalPlan;
+ addStorageGroup(plan.getPath().getFullPath());
+ } else {
+ AuthorPlan plan = (AuthorPlan) physicalPlan;
+ qpExecutor.processNonQuery(plan);
}
+ addResult(response, true);
+ } catch (IOException | PathErrorException e) {
+ LOGGER.error("Execute metadata plan error", e);
+ status = new Status(-1, e.getMessage());
+ addResult(response, false);
+ } catch (ProcessorException e) {
+ LOGGER.error("Execute author plan error", e);
+ status = new Status(-1, e.getMessage());
+ addResult(response, false);
}
- if (closure != null) {
- closure.run(status);
- }
- iterator.next();
+ }
+ if (closure != null) {
+ closure.run(status);
}
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/qp/executor/NonQueryExecutor.java b/cluster/src/main/java/org/apache/iotdb/cluster/qp/executor/NonQueryExecutor.java
index cbd1b65..d2c986c 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/qp/executor/NonQueryExecutor.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/qp/executor/NonQueryExecutor.java
@@ -29,15 +29,15 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-import org.apache.iotdb.cluster.qp.callback.BatchQPTask;
-import org.apache.iotdb.cluster.qp.callback.QPTask;
-import org.apache.iotdb.cluster.qp.callback.SingleQPTask;
import org.apache.iotdb.cluster.config.ClusterConfig;
import org.apache.iotdb.cluster.entity.raft.DataPartitionRaftHolder;
import org.apache.iotdb.cluster.entity.raft.MetadataRaftHolder;
import org.apache.iotdb.cluster.entity.raft.RaftService;
import org.apache.iotdb.cluster.exception.RaftConnectionException;
import org.apache.iotdb.cluster.qp.ClusterQPExecutor;
+import org.apache.iotdb.cluster.qp.callback.BatchQPTask;
+import org.apache.iotdb.cluster.qp.callback.QPTask;
+import org.apache.iotdb.cluster.qp.callback.SingleQPTask;
import org.apache.iotdb.cluster.rpc.raft.request.BasicRequest;
import org.apache.iotdb.cluster.rpc.raft.request.DataGroupNonQueryRequest;
import org.apache.iotdb.cluster.rpc.raft.request.MetaGroupNonQueryRequest;
@@ -110,6 +110,27 @@ public class NonQueryExecutor extends ClusterQPExecutor {
/** 1. Classify physical plans by group id **/
Map<String, List<PhysicalPlan>> physicalPlansMap = new HashMap<>();
Map<String, List<Integer>> planIndexMap = new HashMap<>();
+ classifyPhysicalPlanByGroupId(physicalPlans, batchResult, physicalPlansMap, planIndexMap);
+
+ /** 2. Construct Multiple Data Group Requests **/
+ Map<String, QPTask> subTaskMap = new HashMap<>();
+ constructMultipleRequests(physicalPlansMap, planIndexMap, subTaskMap, batchResult);
+
+ /** 3. Execute Multiple Sub Tasks **/
+ BatchQPTask task = new BatchQPTask(subTaskMap.size(), batchResult, subTaskMap, planIndexMap);
+ currentTask = task;
+ task.execute(this);
+ task.await();
+ batchResult.setAllSuccessful(task.isAllSuccessful());
+ batchResult.setBatchErrorMessage(task.getBatchErrorMessage());
+ }
+
+ /**
+ * Classify batch physical plan by groupId
+ */
+ private void classifyPhysicalPlanByGroupId(PhysicalPlan[] physicalPlans, BatchResult batchResult,
+ Map<String, List<PhysicalPlan>> physicalPlansMap, Map<String, List<Integer>> planIndexMap) {
+ int[] result = batchResult.getResult();
for (int i = 0; i < result.length; i++) {
/** Check if the request has failed. If it has failed, ignore it. **/
if (result[i] != Statement.EXECUTE_FAILED) {
@@ -139,16 +160,21 @@ public class NonQueryExecutor extends ClusterQPExecutor {
}
}
}
+ }
- /** 2. Construct Multiple Data Group Requests **/
- Map<String, QPTask> subTaskMap = new HashMap<>();
+ /**
+ * Construct multiple data group requests
+ */
+ private void constructMultipleRequests(Map<String, List<PhysicalPlan>> physicalPlansMap,
+ Map<String, List<Integer>> planIndexMap, Map<String, QPTask> subTaskMap,
+ BatchResult batchResult) {
+ int[] result = batchResult.getResult();
for (Entry<String, List<PhysicalPlan>> entry : physicalPlansMap.entrySet()) {
String groupId = entry.getKey();
SingleQPTask singleQPTask;
BasicRequest request;
try {
- LOGGER.debug(
- String.format("DATA_GROUP_ID Send batch size() : %d", entry.getValue().size()));
+ LOGGER.debug("DATA_GROUP_ID Send batch size() : {}", entry.getValue().size());
request = new DataGroupNonQueryRequest(groupId, entry.getValue());
singleQPTask = new SingleQPTask(false, request);
subTaskMap.put(groupId, singleQPTask);
@@ -160,14 +186,6 @@ public class NonQueryExecutor extends ClusterQPExecutor {
}
}
}
-
- /** 3. Execute Multiple Sub Tasks **/
- BatchQPTask task = new BatchQPTask(subTaskMap.size(), batchResult, subTaskMap, planIndexMap);
- currentTask = task;
- task.execute(this);
- task.await();
- batchResult.setAllSuccessful(task.isAllSuccessful());
- batchResult.setBatchErrorMessage(task.getBatchErrorMessage());
}
/**
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/utils/RaftUtils.java b/cluster/src/main/java/org/apache/iotdb/cluster/utils/RaftUtils.java
index 7259090..bb5296b 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/utils/RaftUtils.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/utils/RaftUtils.java
@@ -278,4 +278,11 @@ public class RaftUtils {
status.setErrorMsg(e.getMessage());
}
}
+
+ public static Status createErrorStatus(String errorMsg){
+ Status status = new Status();
+ status.setErrorMsg(errorMsg);
+ status.setCode(-1);
+ return status;
+ }
}
diff --git a/iotdb/iotdb/conf/iotdb-cluster.properties b/iotdb/iotdb/conf/iotdb-cluster.properties
index 488cc03..86a46c2 100644
--- a/iotdb/iotdb/conf/iotdb-cluster.properties
+++ b/iotdb/iotdb/conf/iotdb-cluster.properties
@@ -80,11 +80,4 @@ read_data_consistency_level = 1
# Each client request corresponds to a QP Task. A QP task may be divided into several sub-tasks.
# So this value is the sum of all sub-tasks.
# When this value <= 0, use CPU core number * 10
-concurrent_qp_sub_task_thread = 0
-
-# For each raft log, its state machine will start a thread to apply this log.
-# This value indicates maximum number of threads for a state machine to create.
-# Currently, this param only works for state machine of leader in data partition holder.
-# Staten machine in metadata holder will apply raft log serially.
-# When this value <= 0, use CPU core number.
-concurrent_raft_task_thread = 0
\ No newline at end of file
+concurrent_qp_sub_task_thread = 0
\ No newline at end of file