Apache
/
druid
mirror of https://github.com/apache/druid.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MSQ: Launch initial tasks faster. (#13393) 

Notify the mainLoop thread to skip a sleep when the desired task
count changes.
This commit is contained in:
Gian Merlino 2022-11-21 05:41:18 -08:00 committed by GitHub
parent b8ca03d283
commit f037776fd8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncher.java
View File

@ -93,9 +93,14 @@ public class MSQWorkerTaskLauncher
private final AtomicReference<State> state = new AtomicReference<>(State.NEW); private final AtomicReference<State> state = new AtomicReference<>(State.NEW);
private final AtomicBoolean cancelTasksOnStop = new AtomicBoolean(); private final AtomicBoolean cancelTasksOnStop = new AtomicBoolean();
// Set by launchTasksIfNeeded.
@GuardedBy("taskIds") @GuardedBy("taskIds")
private int desiredTaskCount = 0; private int desiredTaskCount = 0;
// Set by the main loop when it acknowledges a new desiredTaskCount.
@GuardedBy("taskIds")
private int acknowledgedDesiredTaskCount = 0;
// Worker number -> task ID. // Worker number -> task ID.
@GuardedBy("taskIds") @GuardedBy("taskIds")
private final List<String> taskIds = new ArrayList<>(); private final List<String> taskIds = new ArrayList<>();
@ -208,6 +213,7 @@ public class MSQWorkerTaskLauncher
synchronized (taskIds) { synchronized (taskIds) {
if (taskCount > desiredTaskCount) { if (taskCount > desiredTaskCount) {
desiredTaskCount = taskCount; desiredTaskCount = taskCount;
taskIds.notifyAll();
} }
while (taskIds.size() < taskCount || !IntStream.range(0, taskCount).allMatch(fullyStartedTasks::contains)) { while (taskIds.size() < taskCount || !IntStream.range(0, taskCount).allMatch(fullyStartedTasks::contains)) {
@ -325,6 +331,7 @@ public class MSQWorkerTaskLauncher
synchronized (taskIds) { synchronized (taskIds) {
firstTask = taskIds.size(); firstTask = taskIds.size();
taskCount = desiredTaskCount; taskCount = desiredTaskCount;
acknowledgedDesiredTaskCount = desiredTaskCount;
} }
for (int i = firstTask; i < taskCount; i++) { for (int i = firstTask; i < taskCount; i++) {
@ -460,7 +467,11 @@ public class MSQWorkerTaskLauncher
} else { } else {
// wait on taskIds so we can wake up early if needed. // wait on taskIds so we can wake up early if needed.
synchronized (taskIds) { synchronized (taskIds) {
taskIds.wait(sleepMillis); // desiredTaskCount is set by launchTasksIfNeeded, and acknowledgedDesiredTaskCount is set by mainLoop when
// it acknowledges a new target. If these are not equal, do another run immediately and launch more tasks.
if (acknowledgedDesiredTaskCount == desiredTaskCount) {
taskIds.wait(sleepMillis);
}
} }
} }
} else { } else {