FrankChen021 commented on code in PR #20299:
URL: https://github.com/apache/druid/pull/20299#discussion_r3968301294
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java:
##########
@@ -1097,6 +1105,14 @@ private void pendingTasksExecutionLoop()
ImmutableWorkerInfo immutableWorker = null;
synchronized (statusLock) {
+ // Compute the eligible-worker snapshot ONCE per pass instead of
rebuilding it for every
+ // pending task. Within a single synchronized(statusLock) pass no
worker reservation is made
+ // until the break below, so getWorkersEligibleToRunTasks() is
invariant across the inner
+ // loop. Rebuilding it per pending task made this loop
O(pendingTasks x workers x
+ // tasksPerWorker) and held statusLock for long periods under a
large pending backlog,
+ // which stalled TaskQueue.add/manage and task submission
cluster-wide.
+ final ImmutableMap<String, ImmutableWorkerInfo> eligibleWorkers =
Review Comment:
[P2] Avoid snapshotting workers when the queue is empty
With an empty `pendingTaskIds` (the normal idle state), this new code still
materializes `getWorkersEligibleToRunTasks()` while holding `statusLock`, even
though the loop will immediately wait at line 1178. That operation walks every
initialized worker and rebuilds its task-derived `ImmutableWorkerInfo`;
previously it was only reached after a pending task was found. On a large
worker/task set, every wake (including worker-sync notifications) now
needlessly holds `statusLock`, delaying `TaskQueue.add/manage` and status
callbacks. Check for pending work before materializing the snapshot.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java:
##########
@@ -1097,6 +1105,14 @@ private void pendingTasksExecutionLoop()
ImmutableWorkerInfo immutableWorker = null;
synchronized (statusLock) {
+ // Compute the eligible-worker snapshot ONCE per pass instead of
rebuilding it for every
+ // pending task. Within a single synchronized(statusLock) pass no
worker reservation is made
+ // until the break below, so getWorkersEligibleToRunTasks() is
invariant across the inner
+ // loop. Rebuilding it per pending task made this loop
O(pendingTasks x workers x
+ // tasksPerWorker) and held statusLock for long periods under a
large pending backlog,
+ // which stalled TaskQueue.add/manage and task submission
cluster-wide.
+ final ImmutableMap<String, ImmutableWorkerInfo> eligibleWorkers =
+ ImmutableMap.copyOf(getWorkersEligibleToRunTasks());
Review Comment:
[P2] Do not reuse a stale capacity snapshot
`WorkerHolder` publishes task announcements into `tasksSnapshotRef` from its
sync executor before `taskAddedOrUpdated` can acquire `statusLock`, so a
worker's capacity can change while this pass scans the pending queue. Reusing
this pass-start snapshot lets a later task be selected for a worker that has
since become full (for example, a recovered or other-runner task not
represented in `workersWithUnacknowledgedTask`). `runTaskOnWorker` only
rechecks removal/lazy/blacklist and the worker endpoint accepts assignments
without a capacity check, so this can oversubscribe a worker. Refresh or
validate the worker snapshot immediately before reserving/assigning, or
otherwise make the capacity check and reservation atomic.
--
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]
