Skip to content

Commit fa194d3

Browse files
CaideyipiCaideyipi
authored
[Pipe] Improve pipe control-flow lock responsiveness (#18076)
* Improve pipe control flow lock responsiveness * Fix pipe meta helper test initialization
1 parent 9554e5d commit fa194d3

5 files changed

Lines changed: 529 additions & 48 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java‎

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,6 @@
5353
import java.util.Collections;
5454
import java.util.List;
5555
import java.util.Map;
56-
import java.util.concurrent.TimeUnit;
5756
import java.util.concurrent.atomic.AtomicBoolean;
5857
import java.util.concurrent.atomic.AtomicReference;
5958

@@ -100,6 +99,10 @@ public abstract class AbstractOperatePipeProcedureV2
10099
// putting it here is just for convenience
101100
protected AtomicReference<PipeTaskInfo> pipeTaskInfo;
102101

102+
// Only used to release global locks before retrying the same state. Do not serialize it because a
103+
// recovered procedure is already re-scheduled by the procedure framework.
104+
private transient boolean shouldYieldAfterExecution;
105+
103106
private static final String SKIP_PIPE_PROCEDURE_MESSAGE =
104107
"Try to start a RUNNING pipe or stop a STOPPED pipe, do nothing.";
105108

@@ -177,15 +180,17 @@ protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
177180
} else {
178181
LOGGER.debug(
179182
ProcedureMessages.PROCEDUREID_RELEASE_LOCK_PIPE_LOCK_WILL_BE_RELEASED, getProcId());
180-
if (this instanceof PipeMetaSyncProcedure) {
183+
if (isSuccess() && this instanceof PipeMetaSyncProcedure) {
181184
configNodeProcedureEnv
182185
.getConfigManager()
183186
.getPipeManager()
184187
.getPipeTaskCoordinator()
185188
.updateLastSyncedVersion();
186189
}
187-
PipeProcedureMetrics.getInstance()
188-
.updateTimer(this.getOperation().getName(), this.elapsedTime());
190+
if (isFinished()) {
191+
PipeProcedureMetrics.getInstance()
192+
.updateTimer(this.getOperation().getName(), this.elapsedTime());
193+
}
189194
releasePipeTaskCoordinatorLock(configNodeProcedureEnv);
190195
}
191196
}
@@ -211,7 +216,7 @@ private void releasePipeTaskCoordinatorLock(ConfigNodeProcedureEnv configNodePro
211216
public abstract void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env);
212217

213218
/**
214-
* Execute at state {@link OperatePipeTaskState#WRITE_CONFIG_NODE_CONSENSUS}.‘
219+
* Execute at state {@link OperatePipeTaskState#WRITE_CONFIG_NODE_CONSENSUS}.
215220
*
216221
* @throws PipeException if configNode consensus write failed
217222
*/
@@ -230,6 +235,7 @@ public abstract void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env)
230235
@Override
231236
protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState state)
232237
throws InterruptedException {
238+
shouldYieldAfterExecution = false;
233239
if (pipeTaskInfo == null) {
234240
LOGGER.warn(
235241
ProcedureMessages.PROCEDUREID_PIPE_LOCK_IS_NOT_ACQUIRED_EXECUTEFROMSTATE_S_EXECUTION_WILL,
@@ -278,8 +284,7 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState
278284
RETRY_THRESHOLD,
279285
e);
280286
setNextState(getCurrentState());
281-
// Wait 3s for next retry
282-
TimeUnit.MILLISECONDS.sleep(3000L);
287+
shouldYieldAfterExecution = true;
283288
} else {
284289
LOGGER.warn(
285290
ProcedureMessages.PROCEDUREID_ALL_RETRIES_FAILED_WHEN_TRYING_TO_AT_STATE_WILL,
@@ -301,6 +306,11 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, OperatePipeTaskState
301306
return Flow.HAS_MORE_STATE;
302307
}
303308

309+
@Override
310+
protected boolean isYieldAfterExecution(final ConfigNodeProcedureEnv env) {
311+
return shouldYieldAfterExecution;
312+
}
313+
304314
@Override
305315
protected boolean isRollbackSupported(OperatePipeTaskState state) {
306316
return true;

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2Test.java‎

Lines changed: 168 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,168 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.confignode.procedure.impl.pipe;
21+
22+
import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo;
23+
import org.apache.iotdb.confignode.procedure.Procedure;
24+
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
25+
import org.apache.iotdb.confignode.procedure.state.pipe.task.OperatePipeTaskState;
26+
import org.apache.iotdb.pipe.api.exception.PipeException;
27+
28+
import org.junit.Assert;
29+
import org.junit.Test;
30+
31+
import java.io.IOException;
32+
import java.util.concurrent.atomic.AtomicReference;
33+
34+
public class AbstractOperatePipeProcedureV2Test {
35+
36+
@Test
37+
public void testSuccessfulStateDoesNotYield() throws Exception {
38+
final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure();
39+
40+
Assert.assertEquals(
41+
StateMachineProcedure.Flow.HAS_MORE_STATE,
42+
procedure.executeFromState(null, OperatePipeTaskState.VALIDATE_TASK));
43+
44+
Assert.assertFalse(procedure.isYieldAfterExecution(null));
45+
Assert.assertEquals(1, procedure.validateExecutionCount);
46+
}
47+
48+
@Test
49+
public void testRetryStateYieldsAndResetsAfterNextExecution() throws Exception {
50+
final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure();
51+
procedure.failValidation = true;
52+
53+
Assert.assertEquals(
54+
StateMachineProcedure.Flow.HAS_MORE_STATE,
55+
procedure.executeFromState(null, OperatePipeTaskState.VALIDATE_TASK));
56+
57+
Assert.assertTrue(procedure.isYieldAfterExecution(null));
58+
Assert.assertEquals(1, procedure.validateExecutionCount);
59+
60+
procedure.failValidation = false;
61+
Assert.assertEquals(
62+
StateMachineProcedure.Flow.HAS_MORE_STATE,
63+
procedure.executeFromState(null, OperatePipeTaskState.VALIDATE_TASK));
64+
65+
Assert.assertFalse(procedure.isYieldAfterExecution(null));
66+
Assert.assertEquals(2, procedure.validateExecutionCount);
67+
}
68+
69+
@Test
70+
public void testRetryStateYieldsOnlyBeforeRetryThreshold() throws Exception {
71+
final TestOperatePipeProcedure procedure = new TestOperatePipeProcedure();
72+
73+
final Procedure<?>[] validateSubProcedures = procedure.runOnce();
74+
Assert.assertEquals(1, validateSubProcedures.length);
75+
Assert.assertSame(procedure, validateSubProcedures[0]);
76+
Assert.assertFalse(procedure.isYieldAfterExecution(null));
77+
78+
procedure.failCalculation = true;
79+
final Procedure<?>[] calculateSubProcedures = procedure.runOnce();
80+
Assert.assertEquals(1, calculateSubProcedures.length);
81+
Assert.assertSame(procedure, calculateSubProcedures[0]);
82+
Assert.assertTrue(procedure.isYieldAfterExecution(null));
83+
Assert.assertEquals(1, procedure.calculateExecutionCount);
84+
85+
Assert.assertNull(procedure.runOnce());
86+
Assert.assertTrue(procedure.hasException());
87+
Assert.assertFalse(procedure.isYieldAfterExecution(null));
88+
Assert.assertEquals(2, procedure.calculateExecutionCount);
89+
}
90+
91+
private static class TestOperatePipeProcedure extends AbstractOperatePipeProcedureV2 {
92+
93+
private int validateExecutionCount;
94+
private int calculateExecutionCount;
95+
private boolean failValidation;
96+
private boolean failCalculation;
97+
98+
private TestOperatePipeProcedure() {
99+
pipeTaskInfo = new AtomicReference<>(new PipeTaskInfo());
100+
}
101+
102+
private Procedure<?>[] runOnce() throws InterruptedException {
103+
return execute(null);
104+
}
105+
106+
@Override
107+
protected PipeTaskOperation getOperation() {
108+
return PipeTaskOperation.START_PIPE;
109+
}
110+
111+
@Override
112+
public boolean executeFromValidateTask(
113+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env)
114+
throws PipeException {
115+
validateExecutionCount++;
116+
if (failValidation) {
117+
throw new PipeException("retry");
118+
}
119+
return true;
120+
}
121+
122+
@Override
123+
public void executeFromCalculateInfoForTask(
124+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env) {
125+
calculateExecutionCount++;
126+
if (failCalculation) {
127+
throw new RuntimeException("retry");
128+
}
129+
}
130+
131+
@Override
132+
public void executeFromWriteConfigNodeConsensus(
133+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env) {
134+
// Do nothing
135+
}
136+
137+
@Override
138+
public void executeFromOperateOnDataNodes(
139+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env) {
140+
// Do nothing
141+
}
142+
143+
@Override
144+
public void rollbackFromValidateTask(
145+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env) {
146+
// Do nothing
147+
}
148+
149+
@Override
150+
public void rollbackFromCalculateInfoForTask(
151+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env) {
152+
// Do nothing
153+
}
154+
155+
@Override
156+
public void rollbackFromWriteConfigNodeConsensus(
157+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env) {
158+
// Do nothing
159+
}
160+
161+
@Override
162+
public void rollbackFromOperateOnDataNodes(
163+
final org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv env)
164+
throws IOException {
165+
// Do nothing
166+
}
167+
}
168+
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java‎

Lines changed: 12 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -1392,49 +1392,20 @@ public TPushPipeMetaResp pushSinglePipeMeta(final TPushSinglePipeMetaReq req) {
13921392

13931393
@Override
13941394
public TPushPipeMetaResp pushMultiPipeMeta(TPushMultiPipeMetaReq req) {
1395-
boolean hasException = false;
1396-
// If there is any exception, we use the size of exceptionMessages to record the fail index
1397-
List<TPushPipeMetaRespExceptionMessage> exceptionMessages = new ArrayList<>();
1398-
try {
1399-
if (req.isSetPipeNamesToDrop()) {
1400-
for (String pipeNameToDrop : req.getPipeNamesToDrop()) {
1401-
TPushPipeMetaRespExceptionMessage message =
1402-
PipeDataNodeAgent.task().handleDropPipe(pipeNameToDrop);
1403-
exceptionMessages.add(message);
1404-
if (message != null) {
1405-
// If there is any exception, skip the remaining pipes
1406-
hasException = true;
1407-
break;
1408-
}
1409-
}
1410-
} else if (req.isSetPipeMetas()) {
1411-
for (ByteBuffer byteBuffer : req.getPipeMetas()) {
1412-
final PipeMeta pipeMeta = PipeMeta.deserialize4TaskAgent(byteBuffer);
1413-
TPushPipeMetaRespExceptionMessage message =
1414-
PipeDataNodeAgent.task().handleSinglePipeMetaChanges(pipeMeta);
1415-
exceptionMessages.add(message);
1416-
if (message != null) {
1417-
// If there is any exception, skip the remaining pipes
1418-
hasException = true;
1419-
break;
1395+
return PushMultiPipeMetaHelper.pushMultiPipeMeta(
1396+
req,
1397+
new PushMultiPipeMetaHelper.Handler() {
1398+
@Override
1399+
public TPushPipeMetaRespExceptionMessage handleDropPipe(final String pipeName) {
1400+
return PipeDataNodeAgent.task().handleDropPipe(pipeName);
14201401
}
1421-
}
1422-
} else {
1423-
throw new Exception(DataNodeMiscMessages.INVALID_PUSH_MULTI_PIPE_META_REQ);
1424-
}
14251402

1426-
return hasException
1427-
? new TPushPipeMetaResp()
1428-
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
1429-
.setExceptionMessages(exceptionMessages)
1430-
: new TPushPipeMetaResp()
1431-
.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
1432-
} catch (Exception e) {
1433-
LOGGER.warn(DataNodeMiscMessages.ERROR_PUSHING_MULTI_PIPE_META, e);
1434-
return new TPushPipeMetaResp()
1435-
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
1436-
.setExceptionMessages(exceptionMessages);
1437-
}
1403+
@Override
1404+
public TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(final ByteBuffer pipeMeta) {
1405+
return PipeDataNodeAgent.task()
1406+
.handleSinglePipeMetaChanges(PipeMeta.deserialize4TaskAgent(pipeMeta));
1407+
}
1408+
});
14381409
}
14391410

14401411
@Override

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/PushMultiPipeMetaHelper.java‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.db.protocol.thrift.impl;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
24+
import org.apache.iotdb.mpp.rpc.thrift.TPushMultiPipeMetaReq;
25+
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaResp;
26+
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
27+
import org.apache.iotdb.rpc.TSStatusCode;
28+
29+
import org.slf4j.Logger;
30+
import org.slf4j.LoggerFactory;
31+
32+
import java.nio.ByteBuffer;
33+
import java.util.ArrayList;
34+
import java.util.List;
35+
36+
final class PushMultiPipeMetaHelper {
37+
38+
private static final Logger LOGGER = LoggerFactory.getLogger(PushMultiPipeMetaHelper.class);
39+
40+
private PushMultiPipeMetaHelper() {
41+
// Utility class
42+
}
43+
44+
interface Handler {
45+
46+
TPushPipeMetaRespExceptionMessage handleDropPipe(String pipeName) throws Exception;
47+
48+
TPushPipeMetaRespExceptionMessage handleSinglePipeMeta(ByteBuffer pipeMeta) throws Exception;
49+
}
50+
51+
static TPushPipeMetaResp pushMultiPipeMeta(
52+
final TPushMultiPipeMetaReq req, final Handler handler) {
53+
final List<TPushPipeMetaRespExceptionMessage> exceptionMessages = new ArrayList<>();
54+
try {
55+
if (req.isSetPipeNamesToDrop()) {
56+
for (final String pipeNameToDrop : req.getPipeNamesToDrop()) {
57+
final TPushPipeMetaRespExceptionMessage message = handler.handleDropPipe(pipeNameToDrop);
58+
if (message != null) {
59+
exceptionMessages.add(message);
60+
}
61+
}
62+
} else if (req.isSetPipeMetas()) {
63+
for (final ByteBuffer pipeMeta : req.getPipeMetas()) {
64+
final TPushPipeMetaRespExceptionMessage message = handler.handleSinglePipeMeta(pipeMeta);
65+
if (message != null) {
66+
exceptionMessages.add(message);
67+
}
68+
}
69+
} else {
70+
throw new Exception(DataNodeMiscMessages.INVALID_PUSH_MULTI_PIPE_META_REQ);
71+
}
72+
73+
return exceptionMessages.isEmpty()
74+
? new TPushPipeMetaResp()
75+
.setStatus(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()))
76+
: new TPushPipeMetaResp()
77+
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
78+
.setExceptionMessages(exceptionMessages);
79+
} catch (final Exception e) {
80+
LOGGER.warn(DataNodeMiscMessages.ERROR_PUSHING_MULTI_PIPE_META, e);
81+
return new TPushPipeMetaResp()
82+
.setStatus(new TSStatus(TSStatusCode.PIPE_PUSH_META_ERROR.getStatusCode()))
83+
.setExceptionMessages(exceptionMessages);
84+
}
85+
}
86+
}

0 commit comments

Comments
 (0)