This is an automated email from the ASF dual-hosted git repository. tanxinyu pushed a commit to branch jira5840 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d1472bf47a7b750040b81ab981858b15e7f3e945 Author: OneSizeFitQuorum <[email protected]> AuthorDate: Fri May 5 18:51:17 2023 +0800 combine rpc Signed-off-by: OneSizeFitQuorum <[email protected]> --- .../db/mpp/plan/scheduler/AsyncPlanNodeSender.java | 54 ++++++++++++++-------- .../plan/scheduler/AsyncSendPlanNodeHandler.java | 27 ++++++----- .../scheduler/FragmentInstanceDispatcherImpl.java | 20 +++++--- .../impl/DataNodeInternalRPCServiceImpl.java | 36 +++++++++------ .../DataNodeInternalRPCServiceImplTest.java | 31 ++++++++----- thrift/src/main/thrift/datanode.thrift | 16 +++++-- 6 files changed, 119 insertions(+), 65 deletions(-) diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java index 29a5f80f628..c18d5fa4ffb 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncPlanNodeSender.java @@ -25,15 +25,18 @@ import org.apache.iotdb.commons.client.IClientManager; import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient; import org.apache.iotdb.db.mpp.plan.planner.plan.FragmentInstance; import org.apache.iotdb.mpp.rpc.thrift.TPlanNode; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeReq; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -43,38 +46,53 @@ import java.util.concurrent.atomic.AtomicLong; import static com.google.common.util.concurrent.Futures.immediateFuture; public class AsyncPlanNodeSender { + private static final Logger logger = LoggerFactory.getLogger(AsyncPlanNodeSender.class); private final IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient> asyncInternalServiceClientManager; private final List<FragmentInstance> instances; - private final Map<Integer, TSendPlanNodeResp> instanceId2RespMap; + + private final Map<TEndPoint, Pair<List<Integer>, TSendBatchPlanNodeReq>> batchRequests; + private final Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap; private final AtomicLong pendingNumber; + private final long startSendTime; public AsyncPlanNodeSender( IClientManager<TEndPoint, AsyncDataNodeInternalServiceClient> asyncInternalServiceClientManager, List<FragmentInstance> instances) { + this.startSendTime = System.nanoTime(); this.asyncInternalServiceClientManager = asyncInternalServiceClientManager; this.instances = instances; - this.instanceId2RespMap = new ConcurrentHashMap<>(); - this.pendingNumber = new AtomicLong(instances.size()); + this.batchRequests = new HashMap<>(); + for (int i = 0; i < instances.size(); i++) { + Pair<List<Integer>, TSendBatchPlanNodeReq> value = + this.batchRequests.computeIfAbsent( + instances.get(i).getHostDataNode().getInternalEndPoint(), + x -> new Pair<>(new ArrayList<>(), new TSendBatchPlanNodeReq())); + value.getLeft().add(i); + value + .getRight() + .addToRequests( + new TSendSinglePlanNodeReq( + new TPlanNode( + instances.get(i).getFragment().getPlanNodeTree().serializeToByteBuffer()), + instances.get(i).getRegionReplicaSet().getRegionId())); + } + this.instanceId2RespMap = new ConcurrentHashMap<>(instances.size() + 1, 1); + this.pendingNumber = new AtomicLong(batchRequests.keySet().size()); } public void sendAll() { - long startSendTime = System.nanoTime(); - for (int i = 0; i < instances.size(); ++i) { - FragmentInstance instance = instances.get(i); + for (Map.Entry<TEndPoint, Pair<List<Integer>, TSendBatchPlanNodeReq>> entry : + batchRequests.entrySet()) { AsyncSendPlanNodeHandler handler = - new AsyncSendPlanNodeHandler(i, pendingNumber, instanceId2RespMap, startSendTime); + new AsyncSendPlanNodeHandler( + entry.getValue().getLeft(), pendingNumber, instanceId2RespMap, startSendTime); try { - TSendPlanNodeReq sendPlanNodeReq = - new TSendPlanNodeReq( - new TPlanNode(instance.getFragment().getPlanNodeTree().serializeToByteBuffer()), - instance.getRegionReplicaSet().getRegionId()); AsyncDataNodeInternalServiceClient client = - asyncInternalServiceClientManager.borrowClient( - instance.getHostDataNode().getInternalEndPoint()); - client.sendPlanNode(sendPlanNodeReq, handler); + asyncInternalServiceClientManager.borrowClient(entry.getKey()); + client.sendBatchPlanNode(entry.getValue().getRight(), handler); } catch (Exception e) { handler.onError(e); } @@ -92,7 +110,7 @@ public class AsyncPlanNodeSender { public List<TSStatus> getFailureStatusList() { List<TSStatus> failureStatusList = new ArrayList<>(); TSStatus status; - for (Map.Entry<Integer, TSendPlanNodeResp> entry : instanceId2RespMap.entrySet()) { + for (Map.Entry<Integer, TSendSinglePlanNodeResp> entry : instanceId2RespMap.entrySet()) { status = entry.getValue().getStatus(); if (!entry.getValue().accepted) { if (status == null) { @@ -122,7 +140,7 @@ public class AsyncPlanNodeSender { } public Future<FragInstanceDispatchResult> getResult() { - for (Map.Entry<Integer, TSendPlanNodeResp> entry : instanceId2RespMap.entrySet()) { + for (Map.Entry<Integer, TSendSinglePlanNodeResp> entry : instanceId2RespMap.entrySet()) { if (!entry.getValue().accepted) { logger.warn( "dispatch write failed. status: {}, code: {}, message: {}, node {}", diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java index 2bd50a6c7ed..61c517ac73e 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/AsyncSendPlanNodeHandler.java @@ -19,37 +19,42 @@ package org.apache.iotdb.db.mpp.plan.scheduler; import org.apache.iotdb.commons.service.metric.enums.PerformanceOverviewMetrics; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.thrift.async.AsyncMethodCallback; +import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicLong; -public class AsyncSendPlanNodeHandler implements AsyncMethodCallback<TSendPlanNodeResp> { - private final int instanceId; +public class AsyncSendPlanNodeHandler implements AsyncMethodCallback<TSendBatchPlanNodeResp> { + + private final List<Integer> instanceIds; private final AtomicLong pendingNumber; - private final Map<Integer, TSendPlanNodeResp> instanceId2RespMap; + private final Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap; private final long sendTime; private static final PerformanceOverviewMetrics PERFORMANCE_OVERVIEW_METRICS = PerformanceOverviewMetrics.getInstance(); public AsyncSendPlanNodeHandler( - int instanceId, + List<Integer> instanceIds, AtomicLong pendingNumber, - Map<Integer, TSendPlanNodeResp> instanceId2RespMap, + Map<Integer, TSendSinglePlanNodeResp> instanceId2RespMap, long sendTime) { - this.instanceId = instanceId; + this.instanceIds = instanceIds; this.pendingNumber = pendingNumber; this.instanceId2RespMap = instanceId2RespMap; this.sendTime = sendTime; } @Override - public void onComplete(TSendPlanNodeResp tSendPlanNodeResp) { - instanceId2RespMap.put(instanceId, tSendPlanNodeResp); + public void onComplete(TSendBatchPlanNodeResp tSendPlanNodeResp) { + for (int i = 0; i < tSendPlanNodeResp.getResponses().size(); i++) { + instanceId2RespMap.put(instanceIds.get(i), tSendPlanNodeResp.getResponses().get(i)); + } if (pendingNumber.decrementAndGet() == 0) { PERFORMANCE_OVERVIEW_METRICS.recordScheduleRemoteCost(System.nanoTime() - sendTime); synchronized (pendingNumber) { @@ -60,13 +65,13 @@ public class AsyncSendPlanNodeHandler implements AsyncMethodCallback<TSendPlanNo @Override public void onError(Exception e) { - TSendPlanNodeResp resp = new TSendPlanNodeResp(); + TSendSinglePlanNodeResp resp = new TSendSinglePlanNodeResp(); String errorMsg = String.format("Fail to send plan node, exception message: %s", e); resp.setAccepted(false); resp.setMessage(errorMsg); resp.setStatus( RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), errorMsg)); - instanceId2RespMap.put(instanceId, resp); + instanceIds.forEach(instanceId -> instanceId2RespMap.put(instanceId, resp)); if (pendingNumber.decrementAndGet() == 0) { PERFORMANCE_OVERVIEW_METRICS.recordScheduleRemoteCost(System.nanoTime() - sendTime); synchronized (pendingNumber) { diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java index a471e9256aa..cb64acaf6bd 100644 --- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/scheduler/FragmentInstanceDispatcherImpl.java @@ -40,10 +40,11 @@ import org.apache.iotdb.db.mpp.plan.planner.plan.node.PlanNode; import org.apache.iotdb.db.utils.SetThreadName; import org.apache.iotdb.mpp.rpc.thrift.TFragmentInstance; import org.apache.iotdb.mpp.rpc.thrift.TPlanNode; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq; import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeReq; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -52,6 +53,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; @@ -268,11 +270,15 @@ public class FragmentInstanceDispatcherImpl implements IFragInstanceDispatcher { } break; case WRITE: - TSendPlanNodeReq sendPlanNodeReq = - new TSendPlanNodeReq( - new TPlanNode(instance.getFragment().getPlanNodeTree().serializeToByteBuffer()), - instance.getRegionReplicaSet().getRegionId()); - TSendPlanNodeResp sendPlanNodeResp = client.sendPlanNode(sendPlanNodeReq); + TSendBatchPlanNodeReq sendPlanNodeReq = + new TSendBatchPlanNodeReq( + Collections.singletonList( + new TSendSinglePlanNodeReq( + new TPlanNode( + instance.getFragment().getPlanNodeTree().serializeToByteBuffer()), + instance.getRegionReplicaSet().getRegionId()))); + TSendSinglePlanNodeResp sendPlanNodeResp = + client.sendBatchPlanNode(sendPlanNodeReq).getResponses().get(0); if (!sendPlanNodeResp.accepted) { logger.warn( "dispatch write failed. status: {}, code: {}, message: {}, node {}", diff --git a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java index 7f45d9b5e93..13690778a09 100644 --- a/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/server/src/main/java/org/apache/iotdb/db/service/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -173,10 +173,11 @@ import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListReq; import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListWithTemplateReq; import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchRequest; import org.apache.iotdb.mpp.rpc.thrift.TSchemaFetchResponse; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp; import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceReq; import org.apache.iotdb.mpp.rpc.thrift.TSendFragmentInstanceResp; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeResp; import org.apache.iotdb.mpp.rpc.thrift.TTsFilePieceReq; import org.apache.iotdb.mpp.rpc.thrift.TUpdateConfigNodeGroupReq; import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq; @@ -289,18 +290,25 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface } @Override - public TSendPlanNodeResp sendPlanNode(TSendPlanNodeReq req) { - LOGGER.debug("receive PlanNode to group[{}]", req.getConsensusGroupId()); - ConsensusGroupId groupId = - ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId()); - PlanNode planNode = PlanNodeType.deserialize(req.planNode.body); - RegionWriteExecutor executor = new RegionWriteExecutor(); - TSendPlanNodeResp resp = new TSendPlanNodeResp(); - RegionExecutionResult executionResult = executor.execute(groupId, planNode); - resp.setAccepted(executionResult.isAccepted()); - resp.setMessage(executionResult.getMessage()); - resp.setStatus(executionResult.getStatus()); - return resp; + public TSendBatchPlanNodeResp sendBatchPlanNode(TSendBatchPlanNodeReq req) { + List<TSendSinglePlanNodeResp> responses = + req.getRequests().stream() + .map( + request -> { + ConsensusGroupId groupId = + ConsensusGroupId.Factory.createFromTConsensusGroupId( + request.getConsensusGroupId()); + PlanNode planNode = PlanNodeType.deserialize(request.planNode.body); + RegionWriteExecutor executor = new RegionWriteExecutor(); + TSendSinglePlanNodeResp resp = new TSendSinglePlanNodeResp(); + RegionExecutionResult executionResult = executor.execute(groupId, planNode); + resp.setAccepted(executionResult.isAccepted()); + resp.setMessage(executionResult.getMessage()); + resp.setStatus(executionResult.getStatus()); + return resp; + }) + .collect(Collectors.toList()); + return new TSendBatchPlanNodeResp(responses); } @Override diff --git a/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java b/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java index c06fcf60078..6e183699289 100644 --- a/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java +++ b/server/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java @@ -43,8 +43,9 @@ import org.apache.iotdb.db.service.thrift.impl.DataNodeInternalRPCServiceImpl; import org.apache.iotdb.db.service.thrift.impl.DataNodeRegionManager; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.mpp.rpc.thrift.TPlanNode; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeReq; -import org.apache.iotdb.mpp.rpc.thrift.TSendPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq; +import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp; +import org.apache.iotdb.mpp.rpc.thrift.TSendSinglePlanNodeReq; import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType; import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType; import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding; @@ -61,11 +62,13 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; public class DataNodeInternalRPCServiceImplTest { + private static final IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig(); DataNodeInternalRPCServiceImpl dataNodeInternalRPCServiceImpl; private static final int dataNodeId = 0; @@ -145,16 +148,18 @@ public class DataNodeInternalRPCServiceImplTest { ByteBuffer byteBuffer = createTimeSeriesNode.serializeToByteBuffer(); // put serialized planNode to TSendPlanNodeReq - TSendPlanNodeReq request = new TSendPlanNodeReq(); + TSendSinglePlanNodeReq request = new TSendSinglePlanNodeReq(); TPlanNode tPlanNode = new TPlanNode(); tPlanNode.setBody(byteBuffer); request.setPlanNode(tPlanNode); request.setConsensusGroupId(regionReplicaSet.getRegionId()); // Use consensus layer to execute request - TSendPlanNodeResp response = dataNodeInternalRPCServiceImpl.sendPlanNode(request); + TSendBatchPlanNodeResp response = + dataNodeInternalRPCServiceImpl.sendBatchPlanNode( + new TSendBatchPlanNodeReq(Collections.singletonList(request))); - Assert.assertTrue(response.accepted); + Assert.assertTrue(response.getResponses().get(0).accepted); } @Test @@ -221,16 +226,18 @@ public class DataNodeInternalRPCServiceImplTest { ByteBuffer byteBuffer = createAlignedTimeSeriesNode.serializeToByteBuffer(); // put serialized planNode to TSendPlanNodeReq - TSendPlanNodeReq request = new TSendPlanNodeReq(); + TSendSinglePlanNodeReq request = new TSendSinglePlanNodeReq(); TPlanNode tPlanNode = new TPlanNode(); tPlanNode.setBody(byteBuffer); request.setPlanNode(tPlanNode); request.setConsensusGroupId(regionReplicaSet.getRegionId()); // Use consensus layer to execute request - TSendPlanNodeResp response = dataNodeInternalRPCServiceImpl.sendPlanNode(request); + TSendBatchPlanNodeResp response = + dataNodeInternalRPCServiceImpl.sendBatchPlanNode( + new TSendBatchPlanNodeReq(Collections.singletonList(request))); - Assert.assertTrue(response.accepted); + Assert.assertTrue(response.getResponses().get(0).accepted); } @Test @@ -308,16 +315,18 @@ public class DataNodeInternalRPCServiceImplTest { ByteBuffer byteBuffer = createMultiTimeSeriesNode.serializeToByteBuffer(); // put serialized planNode to TSendPlanNodeReq - TSendPlanNodeReq request = new TSendPlanNodeReq(); + TSendSinglePlanNodeReq request = new TSendSinglePlanNodeReq(); TPlanNode tPlanNode = new TPlanNode(); tPlanNode.setBody(byteBuffer); request.setPlanNode(tPlanNode); request.setConsensusGroupId(regionReplicaSet.getRegionId()); // Use consensus layer to execute request - TSendPlanNodeResp response = dataNodeInternalRPCServiceImpl.sendPlanNode(request); + TSendBatchPlanNodeResp response = + dataNodeInternalRPCServiceImpl.sendBatchPlanNode( + new TSendBatchPlanNodeReq(Collections.singletonList(request))); - Assert.assertTrue(response.accepted); + Assert.assertTrue(response.getResponses().get(0).accepted); } private TRegionReplicaSet genRegionReplicaSet() { diff --git a/thrift/src/main/thrift/datanode.thrift b/thrift/src/main/thrift/datanode.thrift index 73b631487b7..a9017c2c3ce 100644 --- a/thrift/src/main/thrift/datanode.thrift +++ b/thrift/src/main/thrift/datanode.thrift @@ -118,17 +118,25 @@ struct TSendFragmentInstanceResp { 2: optional string message } -struct TSendPlanNodeReq { +struct TSendSinglePlanNodeReq { 1: required TPlanNode planNode 2: required common.TConsensusGroupId consensusGroupId } -struct TSendPlanNodeResp { +struct TSendSinglePlanNodeResp { 1: required bool accepted 2: optional string message 3: optional common.TSStatus status } +struct TSendBatchPlanNodeReq { + 1: required list<TSendSinglePlanNodeReq> requests; +} + +struct TSendBatchPlanNodeResp { + 1: required list<TSendSinglePlanNodeResp> responses; +} + struct TFetchFragmentInstanceInfoReq { 1: required TFragmentInstanceId fragmentInstanceId } @@ -470,9 +478,9 @@ service IDataNodeRPCService { TSendFragmentInstanceResp sendFragmentInstance(TSendFragmentInstanceReq req); /** - * dispatch PlanNode to remote node for write request in order to save resource + * dispatch PlanNodes in batches to remote node for write request in order to save resource */ - TSendPlanNodeResp sendPlanNode(TSendPlanNodeReq req); + TSendBatchPlanNodeResp sendBatchPlanNode(TSendBatchPlanNodeReq req); TFragmentInstanceInfoResp fetchFragmentInstanceInfo(TFetchFragmentInstanceInfoReq req);
