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/19 07:52:48 UTC
[iotdb] branch master updated: [IOTDB-5786] Fix potential deadlock in DriverScheduler
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b9fa37c912 [IOTDB-5786] Fix potential deadlock in DriverScheduler
b9fa37c912 is described below
commit b9fa37c912975340533ea8927afad4b99886df58
Author: Liao Lanyu <14...@qq.com>
AuthorDate: Wed Apr 19 15:52:41 2023 +0800
[IOTDB-5786] Fix potential deadlock in DriverScheduler
---
.../db/mpp/execution/schedule/DriverScheduler.java | 116 ++++++++++-----------
1 file changed, 56 insertions(+), 60 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
index bddc8ea5a1..9ed599642f 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/schedule/DriverScheduler.java
@@ -299,13 +299,8 @@ public class DriverScheduler implements IDriverScheduler, IService {
for (Set<DriverTask> fragmentRelatedTasks : queryRelatedTasks.values()) {
if (fragmentRelatedTasks != null) {
for (DriverTask task : fragmentRelatedTasks) {
- task.lock();
- try {
- task.setAbortCause(DriverTaskAbortedException.BY_QUERY_CASCADING_ABORTED);
- clearDriverTask(task);
- } finally {
- task.unlock();
- }
+ task.setAbortCause(DriverTaskAbortedException.BY_QUERY_CASCADING_ABORTED);
+ clearDriverTask(task);
}
}
}
@@ -324,13 +319,8 @@ public class DriverScheduler implements IDriverScheduler, IService {
if (task == null) {
return;
}
- task.lock();
- try {
- task.setAbortCause(DriverTaskAbortedException.BY_FRAGMENT_ABORT_CALLED);
- clearDriverTask(task);
- } finally {
- task.unlock();
- }
+ task.setAbortCause(DriverTaskAbortedException.BY_FRAGMENT_ABORT_CALLED);
+ clearDriverTask(task);
}
}
}
@@ -340,26 +330,31 @@ public class DriverScheduler implements IDriverScheduler, IService {
private void clearDriverTask(DriverTask task) {
try (SetThreadName driverTaskName =
new SetThreadName(task.getDriver().getDriverTaskId().getFullId())) {
- DriverTaskStatus status = task.getStatus();
- switch (status) {
- // If it has been aborted, return directly
- case ABORTED:
- return;
- case READY:
- task.setStatus(DriverTaskStatus.ABORTED);
- readyQueue.remove(task.getDriverTaskId());
- break;
- case BLOCKED:
- task.setStatus(DriverTaskStatus.ABORTED);
- blockedTasks.remove(task);
- readyQueue.decreaseReservedSize();
- break;
- case RUNNING:
- task.setStatus(DriverTaskStatus.ABORTED);
- readyQueue.decreaseReservedSize();
- break;
- case FINISHED:
- break;
+ try {
+ task.lock();
+ DriverTaskStatus status = task.getStatus();
+ switch (status) {
+ // If it has been aborted, return directly
+ case ABORTED:
+ return;
+ case READY:
+ task.setStatus(DriverTaskStatus.ABORTED);
+ readyQueue.remove(task.getDriverTaskId());
+ break;
+ case BLOCKED:
+ task.setStatus(DriverTaskStatus.ABORTED);
+ blockedTasks.remove(task);
+ readyQueue.decreaseReservedSize();
+ break;
+ case RUNNING:
+ task.setStatus(DriverTaskStatus.ABORTED);
+ readyQueue.decreaseReservedSize();
+ break;
+ case FINISHED:
+ break;
+ }
+ } finally {
+ task.unlock();
}
timeoutQueue.remove(task.getDriverTaskId());
@@ -378,26 +373,31 @@ public class DriverScheduler implements IDriverScheduler, IService {
queryMap.remove(task.getDriverTaskId().getQueryId());
}
}
- if (task.getAbortCause() != null) {
- try {
- task.getDriver()
- .failed(
- new DriverTaskAbortedException(
- task.getDriver().getDriverTaskId().getFullId(), task.getAbortCause()));
- } catch (Exception e) {
- logger.error("Clear DriverTask failed", e);
+ try {
+ task.lock();
+ if (task.getAbortCause() != null) {
+ try {
+ task.getDriver()
+ .failed(
+ new DriverTaskAbortedException(
+ task.getDriver().getDriverTaskId().getFullId(), task.getAbortCause()));
+ } catch (Exception e) {
+ logger.error("Clear DriverTask failed", e);
+ }
}
- }
- if (task.getStatus() == DriverTaskStatus.ABORTED) {
- try {
- blockManager.forceDeregisterFragmentInstance(
- new TFragmentInstanceId(
- task.getDriverTaskId().getQueryId().getId(),
- task.getDriverTaskId().getFragmentId().getId(),
- task.getDriverTaskId().getFragmentInstanceId().getInstanceId()));
- } catch (Exception e) {
- logger.error("Clear DriverTask failed", e);
+ if (task.getStatus() == DriverTaskStatus.ABORTED) {
+ try {
+ blockManager.forceDeregisterFragmentInstance(
+ new TFragmentInstanceId(
+ task.getDriverTaskId().getQueryId().getId(),
+ task.getDriverTaskId().getFragmentId().getId(),
+ task.getDriverTaskId().getFragmentInstanceId().getInstanceId()));
+ } catch (Exception e) {
+ logger.error("Clear DriverTask failed", e);
+ }
}
+ } finally {
+ task.unlock();
}
}
}
@@ -534,6 +534,7 @@ public class DriverScheduler implements IDriverScheduler, IService {
} finally {
task.unlock();
}
+ // wrap this clearDriverTask to avoid that status is changed to Aborted during clearDriverTask
task.lock();
try {
clearDriverTask(task);
@@ -556,10 +557,10 @@ public class DriverScheduler implements IDriverScheduler, IService {
logger.warn(
"The task {} is aborted. All other tasks in the same query will be cancelled",
task.getDriverTaskId());
- clearDriverTask(task);
} finally {
task.unlock();
}
+ clearDriverTask(task);
QueryId queryId = task.getDriverTaskId().getQueryId();
Map<FragmentInstanceId, Set<DriverTask>> queryRelatedTasks = queryMap.remove(queryId);
if (queryRelatedTasks != null) {
@@ -570,13 +571,8 @@ public class DriverScheduler implements IDriverScheduler, IService {
if (task.equals(otherTask)) {
continue;
}
- otherTask.lock();
- try {
- otherTask.setAbortCause(DriverTaskAbortedException.BY_QUERY_CASCADING_ABORTED);
- clearDriverTask(otherTask);
- } finally {
- otherTask.unlock();
- }
+ otherTask.setAbortCause(DriverTaskAbortedException.BY_QUERY_CASCADING_ABORTED);
+ clearDriverTask(otherTask);
}
}
}