[SPARK-11334] numRunningTasks can't be less than 0, or it will affect executor allocation

[XuTingjun]

With Dynamic Allocation function, a task failed over maxFailure time, all the dependent jobs, stages, tasks will be killed or aborted. In this process, SparkListenerTaskEnd event will be behind in SparkListenerStageCompleted and SparkListenerJobEnd. Like the Event Log below:

{"Event":"SparkListenerStageCompleted","Stage Info":{"Stage ID":20,"Stage Attempt ID":0,"Stage Name":"run at AccessController.java:-2","Number of Tasks":200}
{"Event":"SparkListenerJobEnd","Job ID":9,"Completion Time":1444914699829}
{"Event":"SparkListenerTaskEnd","Stage ID":20,"Stage Attempt ID":0,"Task Type":"ResultTask","Task End Reason":{"Reason":"TaskKilled"},"Task Info":{"Task ID":1955,"Index":88,"Attempt":2,"Launch Time":1444914699763,"Executor ID":"5","Host":"linux-223","Locality":"PROCESS_LOCAL","Speculative":false,"Getting Result Time":0,"Finish Time":1444914699864,"Failed":true,"Accumulables":[]}}

Because that, the numRunningTasks in ExecutorAllocationManager class will be less than 0, and it will affect executor allocation.

[SparkQA]

Test build #44397 has finished for PR 9288 at commit e32e684.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[jerryshao]

IMHO, would it be better to fix this unexpected ordering of events, from my understanding, SparkListenerTaskEnd should be triggered before SparkListenerStageCompleted, right? I think that's root cause should be fixed, rather than adding some guard codes in dynamic allocation.

[srowen]

Hm, this seems like a band-aid though. Is this because the task-end event happens in the wrong order with the job end event? can we catch and guard for that case more directly?

[XuTingjun]

Yeah, I know the root cause is the wrong ordering of events.
The code of these event order are: kill Task, StageCompleted, JobEnd.
Because the TaskEnd is not serial with StageComplete and JobEnd, so I think maybe we can't control the ordering.
Can you read the code, and give me some good idea? Thanks.

[srowen]

Hm, is it better to check the status of the task, and only decrement if it wasn't dead already? (I may not know what I'm talking about there)

[jerryshao]

Can we do this by adding pending to kill tasks into list, only when all the tasks marked as finished, then call markStageAsFinished, also post SparkListenerJobEnd. AFAIK, this is the way to manage executor killing in CoarseGrainedSchedulerBacked, from the code level (not sure the complexity) this can also be achieved here in DAGScheduler.

[andrewor14]

I would say

... but numRunningTasks ($numRunningTasks) != 0

[andrewor14]

It's unclear how the wrong ordering resulted because DAGScheduler supposably posts events from a single thread. I think investigating this will take much longer since the scheduler is quite complicated. I'm OK with doing a fix on the ExecutorAllocationManager side.

[XuTingjun]

@jerryshao, I try to realize the code of your suggestion, but many unit tests are failed. I think it's difficult for me. If you can help me, I would be very grateful.

[XuTingjun]

@andrewor14, yeah the DAGScheduler post events from a single thread, but the root cause is that DAGScheduler receive the SparkListenerTaskEnd behind SparkListenerStageCompleted.
When DAGScheduler ask executor to cancel running tasks, it will immediately mark stage finished, not wait for tasks to complete. So it cause post SparkListenerStageCompleted before SparkListenerTaskEnd.

I try to resolve the root cause, as you said scheduler is quite complicated, my realization caused many unit tests failed. So if you OK with doing a fix on the ExecutorAllocationManager side, I will update the ExecutorAllocationManager code.

[jerryshao]

@XuTingjun , I'm OK with the current way if finding the root cause is a little bit complicated. Just need to add more comments why we have to change to that way.

[jerryshao]

@andrewor14 , from my understanding this wrong ordering is happened when driver explicitly abort the running stages because of exceeding max failures. So SparkListenerStageCompleted will be received firstly, then SparkListenerJobEnd, SparkListenerTaskEnd will be received at last because it has to be killed and return back the result remotely.

[andrewor14]

I see, thanks for the explanation.

@XuTingjun A more robust way to do this may be to keep track of

private val stageIdToRunningTaskIds = new mutable.HashMap[Int, mutable.HashSet[Int]]

then you can just remove all the tasks associated with the stage when the stage is completed.

def totalRunningTasks(): Int = {
  stageIdToRunningTaskIds.mapValues(_.size).sum
}

Can you address the comments?

[tgravescs]

this seems very similar to what I'm seeing in https://issues.apache.org/jira/browse/SPARK-11701. that is dealing with speculation but also causes the numbers to be off. This is only fixing the executor allocation though. Is the executor page also showing a negative number of active tasks? I found that for speculation it affects multiple things that are doing accounting based on the taskEnd event that is coming after the stage is finished. if we can't fix the root cause of out of order we should check the other places it might affect.

[tgravescs]

ping @XuTingjun are you still working on this?

[andrewor14]

Let's close this PR for now since it's been inactive for many months.
@tgravescs @jerryshao would one of you have time to pick up this patch so we can make progress on it?

[jerryshao]

Let me take a crack at this issue :).