Skip to content

[SPARK-13931] Resolve stage hanging up problem in a particular case #11760

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
GavinGavinNo1 wants to merge 2 commits into from
Closed

[SPARK-13931] Resolve stage hanging up problem in a particular case #11760

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e527d67
Resolve stage hanging up problem in a particular case
GavinGavinNo1 Mar 16, 2016
a1eb0f9
make it more reasonable to judge isZombie and add unit test
GavinGavinNo1 Mar 24, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -776,6 +776,7 @@ private[spark] class TaskSetManager(

/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason) {
if (isZombie) return

@mccheah mccheah Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd add a logging line here at least.

@kayousterhout kayousterhout Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it safe to just return here if the task is a zombie? IIRC we need to mark all of the tasks in the TaskSetManager as either completed or failed at some point (otherwise I think this TaskSetManager will never get cleaned up).

@kayousterhout kayousterhout Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I looked at this a little more and I think the right solution is to add "!zombie" to the if-condition below, on line 784 (the reasoning being that, if a task set is a zombie and some shuffle map output was lost, this will be handled when the reduce tasks try to fetch the output, so it's too complicated to bring the task set back from the dead here). That way, the loop on line 799 will still run, so the DAGScheduler will still get told that any speculated tasks running on the lost executor have failed (so the web UI can be updated correctly etc.). Does this seem reasonable?

@mccheah mccheah Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking a bit more closely, I concur with this idea. We basically don't want to resubmit the task, but we also want to mark the task as failed and update metrics. FWIW the if switch in handleFailedTask() should also prevent this task from erroneously counting against the failed task attempt count that would otherwise possibly cause the stage to fail.

@mccheah mccheah Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm... but It's slightly tricky though because I just noticed handleFailedTask has calls like sched.dagScheduler.taskEnded, addPendingTask... https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L721 - are the effects of these calls what was causing the problem in the first place?

@kayousterhout kayousterhout Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My understanding is that the problem is from this line: https://github.com/GavinGavinNo1/spark/blob/resolve-stage-blocked/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L794 which makes the DAGScheduler think the task is still outstanding. handleFailedTask does seem to erroneously call addPendingTask, but then it calls maybeFinishTaskSet, which will see that there are no more running tasks (and that it's a zombie) and then mark it as really finished.

@mccheah mccheah Mar 24, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SGTM - and I suppose the call to dagScheduler.taskEnded is fine in handleFailedTask, as opposed to the call on L794 which marks the task as Resubmitted.

@GavinGavinNo1 GavinGavinNo1 Mar 24, 2016

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kayousterhout I agree with you, it's reasonable to add "!zombie" to the if-condition below, on line 784.
@mccheah Sorry, I haven't kept up your mind. Will it be fine to just add "!zombie" to the if-condition below, on line 784, or there's a better way? Thanks.

// Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
// and we are not using an external shuffle server which could serve the shuffle outputs.
// The reason is the next stage wouldn't be able to fetch the data from this dead executor
Expand Down