linzebing commented on a change in pull request #27223:
[SPARK-30511][SPARK-28403][CORE] Don't treat failed/killed speculative tasks as
pending in ExecutorAllocationManager
URL: https://github.com/apache/spark/pull/27223#discussion_r372187103
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -614,18 +614,24 @@ private[spark] class ExecutorAllocationManager(
stageAttemptToNumRunningTask -= stageAttempt
}
}
- // If the task failed, we expect it to be resubmitted later. To ensure
we have
- // enough resources to run the resubmitted task, we need to mark the
scheduler
- // as backlogged again if it's not already marked as such (SPARK-8366)
- if (taskEnd.reason != Success) {
- if (totalPendingTasks() == 0) {
- allocationManager.onSchedulerBacklogged()
- }
- if (taskEnd.taskInfo.speculative) {
- stageAttemptToSpeculativeTaskIndices.get(stageAttempt).foreach
{_.remove(taskIndex)}
- } else {
- stageAttemptToTaskIndices.get(stageAttempt).foreach
{_.remove(taskIndex)}
- }
+
+ if (taskEnd.taskInfo.speculative) {
+ stageAttemptToSpeculativeTaskIndices.get(stageAttempt).foreach
{_.remove{taskIndex}}
+ stageAttemptToNumSpeculativeTasks(stageAttempt) -= 1
Review comment:
> sorry I don't follow the logic. According to `TSM.dequeueTask()`:
>
> > ```
> > // Tries to schedule a regular task first; if it returns None, then
schedules
> > // a speculative task
> > ```
>
> It would first try to schedule a regular task (which would return None in
this case), then it would try to schedule a speculative task. Since we have
only the original task running, it shall schedule one speculative task in this
case. Am I missing something? I also searched the test case but didn't locate
the place describing the behavior.
@jiangxb1987 Let's make sure we are on the same page. I assume you are
talking about the case on "speculation task failure". As I have explained
above, when a speculative task fails, `handleFailedTask` will be invoked, and
we will have `addPendingTask(index)` at L894. This will add a pending task for
the failed task index (with `speculatable` default to false).
Then from `TSM.dequeueTask()`, we have a call chain of `dequeueTaskHelper`
-> `dequeueTaskFromList`, see
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L289-L293:
as we already have one copy running (the original task), the enqueued pending
task will not be resubmitted here. Instead, in a future speculation cycle, we
will see the original task running too slow and will try to speculate again on
the original task from there.
From my understanding,
> // Tries to schedule a regular task first; if it returns None, then
schedules
> // a speculative task
This is how we are prioritizing speculative vs. non-speculative tasks on
receiving an executor. This is orthogonal to what we are discussing here.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
