gianm commented on code in PR #18851:
URL: https://github.com/apache/druid/pull/18851#discussion_r2968031890
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java:
##########
@@ -1190,197 +1153,197 @@ private void startPendingTaskHandling()
.emit();
}
finally {
- log.info("PendingTaskExecution loop exited.");
+ log.info("PendingTaskExecution loop exited");
}
}
);
}
}
- private void pendingTasksExecutionLoop()
+ @VisibleForTesting
+ void pendingTasksExecutionLoop()
{
while (!Thread.interrupted() && lifecycleLock.awaitStarted(1,
TimeUnit.MILLISECONDS)) {
try {
- // Find one pending task to run and a worker to run on
- HttpRemoteTaskRunnerWorkItem taskItem = null;
- ImmutableWorkerInfo immutableWorker = null;
-
- synchronized (statusLock) {
- Iterator<String> iter = pendingTaskIds.iterator();
- while (iter.hasNext()) {
- String taskId = iter.next();
- HttpRemoteTaskRunnerWorkItem ti = tasks.get(taskId);
-
- if (ti == null || !ti.getState().isPending()) {
- // happens if the task was shutdown, failed or observed running
by a worker
- iter.remove();
- continue;
- }
-
- if (ti.getState() ==
HttpRemoteTaskRunnerWorkItem.State.PENDING_WORKER_ASSIGN) {
- // picked up by another pending task executor thread which is in
the process of trying to
- // run it on a worker, skip to next.
- continue;
- }
-
- if (ti.getTask() == null) {
- // this is not supposed to happen except for a bug, we want to
mark this task failed but
- // taskComplete(..) can not be called while holding statusLock.
See the javadoc on that
- // method.
- // so this will get marked failed afterwards outside of current
synchronized block.
- taskItem = ti;
- break;
- }
-
- immutableWorker = findWorkerToRunTask(ti.getTask());
- if (immutableWorker == null) {
- continue;
- }
-
- String prevUnackedTaskId =
workersWithUnacknowledgedTask.putIfAbsent(
- immutableWorker.getWorker().getHost(),
- taskId
- );
- if (prevUnackedTaskId != null) {
- log.makeAlert(
- "Found worker[%s] with unacked task[%s] but still was
identified to run task[%s].",
- immutableWorker.getWorker().getHost(),
- prevUnackedTaskId,
- taskId
- ).emit();
- }
-
- // set state to PENDING_WORKER_ASSIGN before releasing the lock so
that this task item is not picked
- // up by another task execution thread.
- // note that we can't simply delete this task item from
pendingTaskIds or else we would have to add it
- // back if this thread couldn't run this task for any reason,
which we will know at some later time
- // and also we will need to add it back to its old position in the
list. that becomes complex quickly.
- // Instead we keep the PENDING_WORKER_ASSIGN to notify other task
execution threads not to pick this one up.
- // And, it is automatically removed by any of the task execution
threads when they notice that
- // ti.getState().isPending() is false (at the beginning of this
loop)
-
ti.setState(HttpRemoteTaskRunnerWorkItem.State.PENDING_WORKER_ASSIGN);
- taskItem = ti;
- break;
- }
-
- if (taskItem == null) {
- // Either no pending task is found or no suitable worker is found
for any of the pending tasks.
- // statusLock.notifyAll() is called whenever a new task shows up
or if there is a possibility for a task
- // to successfully get worker to run, for example when a new
worker shows up, a task slot opens up
- // because some task completed etc.
- statusLock.wait(TimeUnit.MINUTES.toMillis(1));
+ final PendingTaskQueueItem taskItem = pendingTasks.poll(1,
TimeUnit.MINUTES);
+ if (taskItem == null) {
+ log.info("Found no available tasks. Waiting for tasks to assign");
+ continue;
+ }
+ final String taskId = taskItem.getTask().getId();
+
+ ImmutableWorkerInfo workerToAssign;
+ // Set to false inside tasks.compute() if the task is no longer
PENDING when we attempt
+ // to transition it to PENDING_WORKER_ASSIGN. A task can complete
concurrently (e.g. via
+ // cancellation) without holding workerStateLock, so we cannot rely on
a pre-check.
+ final AtomicBoolean taskWasPending = new AtomicBoolean(true);
+
+ synchronized (workerStateLock) {
+ workerToAssign = findWorkerToRunTask(taskItem.getTask());
+
+ if (workerToAssign == null) {
Review Comment:
What do you mean by "rescheduling this task"?
The thing I'm worried about is that we need line-skipping behavior.
Especially with strong worker affinity, it's important for tasks to be able to
skip the line. For example: a typical config would have a set of affinity
workers for batch tasks and a set for realtime tasks. When the batch affinity
workers are full we want to continue to assign realtime tasks to the realtime
affinity workers.
So, if the solution does give tasks the ability to skip the line, it should
be OK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
