From 2274c28d94e80b144680dec2bdce80050093c8e2 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Sat, 13 Aug 2022 23:31:38 -0700 Subject: [PATCH 1/3] Fix race in TaskQueue.notifyStatus. It was possible for manageInternal to relaunch a task while it was being cleaned up, due to a race that happens when notifyStatus is called to clean up a successful task: 1) In a critical section, notifyStatus removes the task from "tasks". 2) Outside a critical section, notifyStatus calls taskRunner.shutdown to let the task runner know it can clear out its data structures. 3) In a critical section, syncFromStorage adds the task back to "tasks", because it is still present in metadata storage. 4) In a critical section, manageInternalCritical notices that the task is in "tasks" and is not running in the taskRunner, so it launches it again. 5) In a (different) critical section, notifyStatus updates the metadata store to set the task status to SUCCESS. 6) The task continues running even though it should not be. The possibility for this race was introduced in #12099, which shrunk the critical section in notifyStatus. Prior to that patch, a single critical section encompassed (1), (2), and (5), so the ordering above was not possible. This patch does the following: 1) Fixes the race by adding a recentlyCompletedTasks set that prevents the main management loop from doing anything with tasks that are currently being cleaned up. 2) Switches the order of the critical sections in notifyStatus, so metadata store updates happen first. This is useful in case of server failures: it ensures that if the Overlord fails in the midst of notifyStatus, then completed-task statuses are still available in ZK or on MMs for the next Overlord. (Those are cleaned up by taskRunner.shutdown, which formerly ran first.) This isn't related to the race described above, but is fixed opportunistically as part of the same patch. 3) Changes the "tasks" list to a map. Many operations require retrieval or removal of individual tasks; those are now O(1) instead of O(N) in the number of running tasks. 4) Changes various log messages to use task ID instead of full task payload, to make the logs more readable. --- .../druid/indexing/overlord/TaskQueue.java | 172 ++++++++++-------- 1 file changed, 99 insertions(+), 73 deletions(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index 50d5d8b4fc7a..b7c36990df4b 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -42,6 +42,7 @@ import org.apache.druid.indexing.overlord.config.DefaultTaskConfig; import org.apache.druid.indexing.overlord.config.TaskLockConfig; import org.apache.druid.indexing.overlord.config.TaskQueueConfig; +import org.apache.druid.java.util.common.ISE; import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.concurrent.ScheduledExecutors; import org.apache.druid.java.util.common.lifecycle.LifecycleStart; @@ -56,6 +57,7 @@ import java.util.Collection; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -82,14 +84,21 @@ */ public class TaskQueue { - private final long MANAGEMENT_WAIT_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(60); - private final long MIN_WAIT_TIME_MS = 100; + private static final long MANAGEMENT_WAIT_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(60); + private static final long MIN_WAIT_TIME_MS = 100; + // Task ID -> Task that is currently running @GuardedBy("giant") - private final List tasks = new ArrayList<>(); + private final LinkedHashMap tasks = new LinkedHashMap<>(); + + // Task ID -> Future from the TaskRunner @GuardedBy("giant") private final Map> taskFutures = new HashMap<>(); + // Tasks that are in the process of being cleaned up by notifyStatus. Prevents manageInternal from re-launching them. + @GuardedBy("giant") + private final Set recentlyCompletedTasks = new HashSet<>(); + private final TaskLockConfig lockConfig; private final TaskQueueConfig config; private final DefaultTaskConfig defaultTaskConfig; @@ -349,11 +358,19 @@ private void manageInternalCritical( { // Task futures available from the taskRunner for (final TaskRunnerWorkItem workItem : taskRunner.getKnownTasks()) { - runnerTaskFutures.put(workItem.getTaskId(), workItem.getResult()); + if (!recentlyCompletedTasks.contains(workItem.getTaskId())) { + // Don't do anything with tasks that have recently finished; notifyStatus will handle it. + runnerTaskFutures.put(workItem.getTaskId(), workItem.getResult()); + } } // Attain futures for all active tasks (assuming they are ready to run). // Copy tasks list, as notifyStatus may modify it. - for (final Task task : ImmutableList.copyOf(tasks)) { + for (final Task task : ImmutableList.copyOf(tasks.values())) { + if (recentlyCompletedTasks.contains(task.getId())) { + // Don't do anything with tasks that have recently finished; notifyStatus will handle it. + continue; + } + knownTaskIds.add(task.getId()); if (!taskFutures.containsKey(task.getId())) { @@ -478,20 +495,32 @@ public boolean add(final Task task) throws EntryExistsException } } - // Should always be called after taking giantLock @GuardedBy("giant") private void addTaskInternal(final Task task) { - tasks.add(task); - taskLockbox.add(task); + final Task existingTask = tasks.putIfAbsent(task.getId(), task); + + if (existingTask == null) { + taskLockbox.add(task); + } else if (!existingTask.equals(task)) { + throw new ISE("Cannot add task ID [%s] with same ID as task that has already been added"); + } } - // Should always be called after taking giantLock + /** + * Removes a task from {@link #tasks} and {@link #taskLockbox}, if it exists. Returns whether the task was + * removed or not. + */ @GuardedBy("giant") - private void removeTaskInternal(final Task task) + private boolean removeTaskInternal(final String taskId) { - taskLockbox.remove(task); - tasks.remove(task); + final Task task = tasks.remove(taskId); + if (task != null) { + taskLockbox.remove(task); + return true; + } else { + return false; + } } /** @@ -506,12 +535,9 @@ public void shutdown(final String taskId, String reasonFormat, Object... args) giant.lock(); try { - Preconditions.checkNotNull(taskId, "taskId"); - for (final Task task : tasks) { - if (task.getId().equals(taskId)) { - notifyStatus(task, TaskStatus.failure(taskId, StringUtils.format(reasonFormat, args)), reasonFormat, args); - break; - } + final Task task = tasks.get(Preconditions.checkNotNull(taskId, "taskId")); + if (task != null) { + notifyStatus(task, TaskStatus.failure(taskId, StringUtils.format(reasonFormat, args)), reasonFormat, args); } } finally { @@ -531,12 +557,9 @@ public void shutdownWithSuccess(final String taskId, String reasonFormat, Object giant.lock(); try { - Preconditions.checkNotNull(taskId, "taskId"); - for (final Task task : tasks) { - if (task.getId().equals(taskId)) { - notifyStatus(task, TaskStatus.success(taskId), reasonFormat, args); - break; - } + final Task task = tasks.get(Preconditions.checkNotNull(taskId, "taskId")); + if (task != null) { + notifyStatus(task, TaskStatus.success(taskId), reasonFormat, args); } } finally { @@ -568,62 +591,65 @@ private void notifyStatus(final Task task, final TaskStatus taskStatus, String r taskStatus.getId() ); - // Inform taskRunner that this task can be shut down - TaskLocation taskLocation = TaskLocation.unknown(); - try { - taskLocation = taskRunner.getTaskLocation(task.getId()); - taskRunner.shutdown(task.getId(), reasonFormat, args); + if (!taskStatus.isComplete()) { + // Nothing to do for incomplete statuses. + return; } - catch (Exception e) { - log.warn(e, "TaskRunner failed to cleanup task after completion: %s", task.getId()); - } - - int removed = 0; - - ///////// critical section + // Critical section: add this task to recentlyCompletedTasks, so it isn't managed while being cleaned up. giant.lock(); try { - // Remove from running tasks - for (int i = tasks.size() - 1; i >= 0; i--) { - if (tasks.get(i).getId().equals(task.getId())) { - removed++; - removeTaskInternal(tasks.get(i)); - break; - } - } - - // Remove from futures list - taskFutures.remove(task.getId()); + recentlyCompletedTasks.add(task.getId()); } finally { giant.unlock(); } - ///////// end critical + final TaskLocation taskLocation = taskRunner.getTaskLocation(task.getId()); - if (removed == 0) { - log.warn("Unknown task completed: %s", task.getId()); + // Save status to metadata store first, so if we crash while doing the rest of the shutdown, our successor + // remembers that this task has completed. + try { + final Optional previousStatus = taskStorage.getStatus(task.getId()); + if (!previousStatus.isPresent() || !previousStatus.get().isRunnable()) { + log.makeAlert("Ignoring notification for already-complete task").addData("task", task.getId()).emit(); + } else { + taskStorage.setStatus(taskStatus.withLocation(taskLocation)); + } + } + catch (Throwable e) { + // If persist fails, even after the retries performed in taskStorage, then metadata store and actual cluster + // state have diverged. Send out an alert and continue with the task shutdown routine. + log.makeAlert(e, "Failed to persist status for task") + .addData("task", task.getId()) + .addData("statusCode", taskStatus.getStatusCode()) + .emit(); } - if (removed > 0) { - // If we thought this task should be running, save status to DB - try { - final Optional previousStatus = taskStorage.getStatus(task.getId()); - if (!previousStatus.isPresent() || !previousStatus.get().isRunnable()) { - log.makeAlert("Ignoring notification for already-complete task").addData("task", task.getId()).emit(); - } else { - taskStorage.setStatus(taskStatus.withLocation(taskLocation)); - log.info("Task done: %s", task); - requestManagement(); - } - } - catch (Exception e) { - log.makeAlert(e, "Failed to persist status for task") - .addData("task", task.getId()) - .addData("statusCode", taskStatus.getStatusCode()) - .emit(); + // Inform taskRunner that this task can be shut down. + try { + taskRunner.shutdown(task.getId(), reasonFormat, args); + } + catch (Throwable e) { + // If task runner shutdown fails, continue with the task shutdown routine. We'll come back and try to + // shut it down again later in manageInternalPostCritical, once it's removed from the "tasks" map. + log.warn(e, "TaskRunner failed to cleanup task after completion: %s", task.getId()); + } + + // Critical section: remove this task from all of our tracking data structures. + giant.lock(); + try { + if (removeTaskInternal(task.getId())) { + taskFutures.remove(task.getId()); + } else { + log.warn("Unknown task completed: %s", task.getId()); } + + recentlyCompletedTasks.remove(task.getId()); + requestManagement(); + } + finally { + giant.unlock(); } } @@ -684,7 +710,7 @@ private void handleStatus(final TaskStatus status) log.info( "Task %s: %s (%d run duration)", status.getStatusCode(), - task, + task.getId(), status.getDuration() ); @@ -719,7 +745,7 @@ private void syncFromStorage() if (active) { final Map newTasks = toTaskIDMap(taskStorage.getActiveTasks()); final int tasksSynced = newTasks.size(); - final Map oldTasks = toTaskIDMap(tasks); + final Map oldTasks = new HashMap<>(tasks); // Calculate differences on IDs instead of Task Objects. Set commonIds = Sets.newHashSet(Sets.intersection(newTasks.keySet(), oldTasks.keySet())); @@ -732,7 +758,7 @@ private void syncFromStorage() // Clean up removed Tasks for (Task task : removedTasks) { - removeTaskInternal(task); + removeTaskInternal(task.getId()); } // Add newly Added tasks to the queue @@ -800,7 +826,7 @@ Map getCurrentTaskDatasources() { giant.lock(); try { - return tasks.stream().collect(Collectors.toMap(Task::getId, Task::getDataSource)); + return tasks.values().stream().collect(Collectors.toMap(Task::getId, Task::getDataSource)); } finally { giant.unlock(); @@ -840,7 +866,7 @@ public Map getWaitingTaskCount() giant.lock(); try { - return tasks.stream().filter(task -> !runnerKnownTaskIds.contains(task.getId())) + return tasks.values().stream().filter(task -> !runnerKnownTaskIds.contains(task.getId())) .collect(Collectors.toMap(Task::getDataSource, task -> 1L, Long::sum)); } finally { @@ -853,7 +879,7 @@ List getTasks() { giant.lock(); try { - return new ArrayList(tasks); + return new ArrayList<>(tasks.values()); } finally { giant.unlock(); From 1416accbebb08b06cd900b5de626c25cd1c7dea6 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Sun, 14 Aug 2022 01:50:20 -0700 Subject: [PATCH 2/3] Fix format string. --- .../main/java/org/apache/druid/indexing/overlord/TaskQueue.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index b7c36990df4b..fbbaa4d8bc21 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -503,7 +503,7 @@ private void addTaskInternal(final Task task) if (existingTask == null) { taskLockbox.add(task); } else if (!existingTask.equals(task)) { - throw new ISE("Cannot add task ID [%s] with same ID as task that has already been added"); + throw new ISE("Cannot add task ID [%s] with same ID as task that has already been added", task.getId()); } } From 53b4e64c247d12839d378a0c291e775715a14723 Mon Sep 17 00:00:00 2001 From: Gian Merlino Date: Sun, 14 Aug 2022 15:43:43 -0700 Subject: [PATCH 3/3] Update comment. --- .../main/java/org/apache/druid/indexing/overlord/TaskQueue.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java index fbbaa4d8bc21..d3196f350671 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java @@ -87,7 +87,7 @@ public class TaskQueue private static final long MANAGEMENT_WAIT_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(60); private static final long MIN_WAIT_TIME_MS = 100; - // Task ID -> Task that is currently running + // Task ID -> Task, for tasks that are active in some way (submitted, running, or finished and to-be-cleaned-up). @GuardedBy("giant") private final LinkedHashMap tasks = new LinkedHashMap<>();