You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2023/04/20 06:21:50 UTC
[iotdb] branch QueryStateMachineBug created (now 273f7623ff)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a change to branch QueryStateMachineBug
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 273f7623ff Fix concurrent state change bug in QueryStateMachine
This branch includes the following new commits:
new 273f7623ff Fix concurrent state change bug in QueryStateMachine
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.
[iotdb] 01/01: Fix concurrent state change bug in QueryStateMachine
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch QueryStateMachineBug
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 273f7623ffa67384a3e533587beec0f183766637
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Thu Apr 20 14:21:37 2023 +0800
Fix concurrent state change bug in QueryStateMachine
---
.../iotdb/db/mpp/execution/QueryStateMachine.java | 75 +++++++++-------------
1 file changed, 32 insertions(+), 43 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryStateMachine.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryStateMachine.java
index 53a52bd940..d4c2574187 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/QueryStateMachine.java
@@ -27,12 +27,23 @@ import com.google.common.util.concurrent.ListenableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.Objects.requireNonNull;
+import static org.apache.iotdb.db.mpp.execution.QueryState.ABORTED;
+import static org.apache.iotdb.db.mpp.execution.QueryState.CANCELED;
+import static org.apache.iotdb.db.mpp.execution.QueryState.DISPATCHING;
+import static org.apache.iotdb.db.mpp.execution.QueryState.FAILED;
+import static org.apache.iotdb.db.mpp.execution.QueryState.FINISHED;
+import static org.apache.iotdb.db.mpp.execution.QueryState.PENDING_RETRY;
+import static org.apache.iotdb.db.mpp.execution.QueryState.PLANNED;
+import static org.apache.iotdb.db.mpp.execution.QueryState.QUEUED;
+import static org.apache.iotdb.db.mpp.execution.QueryState.RUNNING;
+
/**
* State machine for a QueryExecution. It stores the states for the QueryExecution. Others can
* register listeners when the state changes of the QueryExecution.
*/
public class QueryStateMachine {
- private final String name;
private final StateMachine<QueryState> queryState;
// The executor will be used in all the state machines belonged to this query.
@@ -41,13 +52,12 @@ public class QueryStateMachine {
private TSStatus failureStatus;
public QueryStateMachine(QueryId queryId, ExecutorService executor) {
- this.name = String.format("QueryStateMachine[%s]", queryId);
this.stateMachineExecutor = executor;
this.queryState =
new StateMachine<>(
queryId.toString(),
this.stateMachineExecutor,
- QueryState.QUEUED,
+ QUEUED,
QueryState.TERMINAL_INSTANCE_STATES);
}
@@ -60,89 +70,68 @@ public class QueryStateMachine {
return queryState.getStateChange(currentState);
}
- private String getName() {
- return name;
- }
-
public QueryState getState() {
return queryState.get();
}
public void transitionToQueued() {
- queryState.set(QueryState.QUEUED);
+ queryState.set(QUEUED);
}
public void transitionToPlanned() {
- queryState.set(QueryState.PLANNED);
+ queryState.setIf(PLANNED, currentState -> currentState == QUEUED);
}
public void transitionToDispatching() {
- queryState.set(QueryState.DISPATCHING);
+ queryState.setIf(DISPATCHING, currentState -> currentState == PLANNED);
}
public void transitionToPendingRetry(TSStatus failureStatus) {
- if (queryState.get().isDone()) {
- return;
- }
this.failureStatus = failureStatus;
- queryState.set(QueryState.PENDING_RETRY);
+ queryState.setIf(PENDING_RETRY, currentState -> currentState == DISPATCHING);
}
public void transitionToRunning() {
- queryState.set(QueryState.RUNNING);
+ queryState.setIf(RUNNING, currentState -> currentState == DISPATCHING);
}
public void transitionToFinished() {
- if (queryState.get().isDone()) {
- return;
- }
- queryState.set(QueryState.FINISHED);
+ transitionToDoneState(FINISHED);
}
public void transitionToCanceled() {
- if (queryState.get().isDone()) {
- return;
- }
- queryState.set(QueryState.CANCELED);
+ transitionToDoneState(CANCELED);
}
public void transitionToCanceled(Throwable throwable, TSStatus failureStatus) {
- if (queryState.get().isDone()) {
- return;
- }
this.failureException = throwable;
this.failureStatus = failureStatus;
- queryState.set(QueryState.CANCELED);
+ transitionToDoneState(CANCELED);
}
public void transitionToAborted() {
- if (queryState.get().isDone()) {
- return;
- }
- queryState.set(QueryState.ABORTED);
+ transitionToDoneState(ABORTED);
}
public void transitionToFailed() {
- if (queryState.get().isDone()) {
- return;
- }
- queryState.set(QueryState.FAILED);
+ transitionToDoneState(FAILED);
}
public void transitionToFailed(Throwable throwable) {
- if (queryState.get().isDone()) {
- return;
- }
this.failureException = throwable;
- queryState.set(QueryState.FAILED);
+ transitionToDoneState(FAILED);
}
public void transitionToFailed(TSStatus failureStatus) {
- if (queryState.get().isDone()) {
- return;
- }
this.failureStatus = failureStatus;
- queryState.set(QueryState.FAILED);
+ transitionToDoneState(FAILED);
+ }
+
+ private void transitionToDoneState(QueryState doneState) {
+ requireNonNull(doneState, "doneState is null");
+ checkArgument(doneState.isDone(), "doneState %s is not a done state", doneState);
+
+ queryState.setIf(doneState, currentState -> !currentState.isDone());
}
public String getFailureMessage() {