[SPARK-24677][Core]Avoid NoSuchElementException from MedianHeap #21656

cxzl25 · 2018-06-28T07:47:03Z

What changes were proposed in this pull request?

When speculation is enabled,
TaskSetManager#markPartitionCompleted should write successful task duration to MedianHeap,
not just increase tasksSuccessful.

Otherwise when TaskSetManager#checkSpeculatableTasks,tasksSuccessful non-zero, but MedianHeap is empty.
Then throw an exception successfulTaskDurations.median java.util.NoSuchElementException: MedianHeap is empty.
Finally led to stopping SparkContext.

How was this patch tested?

TaskSetManagerSuite.scala
unit test:[SPARK-24677] MedianHeap should not be empty when speculation is enabled

…text to stop.

maropu · 2018-06-28T08:11:42Z

Can you add a test to check if no exception thrown in that condition with this patch?

…culation is enabled

cxzl25 · 2018-07-02T06:24:54Z

@maropu
I have added a unit test.
Can you trigger a test for this?

cxzl25 · 2018-07-04T13:49:59Z

@maropu @cloud-fan @squito
Can you trigger a test for this?
This is the exception stack in the log:

ERROR Utils: uncaught error in thread task-scheduler-speculation, stopping SparkContext
java.util.NoSuchElementException: MedianHeap is empty.
at org.apache.spark.util.collection.MedianHeap.median(MedianHeap.scala:83)
at org.apache.spark.scheduler.TaskSetManager.checkSpeculatableTasks(TaskSetManager.scala:968)
at org.apache.spark.scheduler.Pool$$anonfun$checkSpeculatableTasks$1.apply(Pool.scala:94)
at org.apache.spark.scheduler.Pool$$anonfun$checkSpeculatableTasks$1.apply(Pool.scala:93)
at scala.collection.Iterator$class.foreach(Iterator.scala:742)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at org.apache.spark.scheduler.Pool.checkSpeculatableTasks(Pool.scala:93)
at org.apache.spark.scheduler.Pool$$anonfun$checkSpeculatableTasks$1.apply(Pool.scala:94)
at org.apache.spark.scheduler.Pool$$anonfun$checkSpeculatableTasks$1.apply(Pool.scala:93)

squito · 2018-07-05T01:59:07Z

ok to test

cloud-fan · 2018-07-05T04:37:30Z

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

+        if (speculationEnabled) {
+          taskAttempts(index).headOption.map { info =>
+            info.markFinished(TaskState.FINISHED, clock.getTimeMillis())
+            successfulTaskDurations.insert(info.duration)


what's the normal code path to update task durations?

TaskSetManager#handleSuccessfulTask update successful task durations, and write to successfulTaskDurations.

When there are multiple tasksets for this stage, markPartitionCompletedInAllTaskSets is
accumulate the value of tasksSuccessful.

In this case, when checkSpeculatableTasks is called, the value of tasksSuccessful matches the condition, but successfulTaskDurations is empty.

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L723

def handleSuccessfulTask(tid: Long, result: DirectTaskResult[_]): Unit = { val info = taskInfos(tid) val index = info.index info.markFinished(TaskState.FINISHED, clock.getTimeMillis()) if (speculationEnabled) { successfulTaskDurations.insert(info.duration) } // ... // There may be multiple tasksets for this stage -- we let all of them know that the partition // was completed. This may result in some of the tasksets getting completed. sched.markPartitionCompletedInAllTaskSets(stageId, tasks(index).partitionId)

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L987

override def checkSpeculatableTasks(minTimeToSpeculation: Int): Boolean = { //... if (tasksSuccessful >= minFinishedForSpeculation && tasksSuccessful > 0) { val time = clock.getTimeMillis() val medianDuration = successfulTaskDurations.median

cloud-fan · 2018-07-05T04:37:43Z

cc @jiangxb1987

SparkQA · 2018-07-05T05:41:14Z

Test build #92631 has finished for PR 21656 at commit 55ddbeb.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

squito · 2018-07-05T14:08:40Z

Thanks for finding this and suggesting a fix @cxzl25. But, I'm not sure it makes sense to use this duration. its not how long the task actually took to complete. I think it might make more sense to just ignore this task for speculation. I will think about it some more.

cc @markhamstra @tgravescs

jiangxb1987 · 2018-07-05T14:11:14Z

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

  private[scheduler] def markPartitionCompleted(partitionId: Int): Unit = {
    partitionToIndex.get(partitionId).foreach { index =>
      if (!successful(index)) {
+        if (speculationEnabled) {


IIUC in this case no task in this taskSet actually successfully finishes, it's another task attempt from another taskSet for the same stage that succeeded. In stead of changing this code path, I'd suggest we have another flag to show whether any task succeeded in current taskSet, and if no task have succeeded, skip L987.

WDYT @squito ?

yeah that is sort of what I was suggesting -- but I was thinking rather than just a flag, maybe we separate out tasksSuccessful into tasksCompletedSuccessfully (from this taskset) and tasksNoLongerNecessary (from any taskset), perhaps with better names. If you just had a flag, you would avoid the exception from the empty heap, but you still might decide to enable speculation prematurely as you really haven't finished enough for SPECULATION_QUANTILE:

spark/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

Line 987 in a381bce

if (tasksSuccessful >= minFinishedForSpeculation && tasksSuccessful > 0) {

tgravescs · 2018-07-06T20:56:48Z

I assume this is really that it isn't updating successfulTaskDurations? MedianHeap is a collection, can you please update description and title to be more explicit

tgravescs · 2018-07-06T21:39:23Z

In this case one of the older stage attempts (that is a zombie) marked the task as successful but then the newest stage attempt checked to see if it needed to speculate. Is that correct?

Ideally I think for speculation we want to look at the task time for all stage attempts. But that is probably a bigger change then this. If we aren't doing that then I think ignoring it for speculation is ok. Otherwise how hard is it to send the actual task info into here so it could use the real time the successful task took?

squito · 2018-07-06T21:45:10Z

Ideally I think for speculation we want to look at the task time for all stage attempts. But that is probably a bigger change then this

yeah I agree, on both points. One thing which is a little tricky is that you probably want to make sure you're only counting times from different partitions -- you might times from the same partition from multiple attempts, but that shouldn't count. (or maybe we don't really care that much as its just a heuristic anyway ...)

mridulm · 2018-07-07T07:00:34Z

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

  private[scheduler] def markPartitionCompleted(partitionId: Int): Unit = {
    partitionToIndex.get(partitionId).foreach { index =>
      if (!successful(index)) {
+        if (speculationEnabled) {


speculationEnabled && ! isZombie

tgravescs · 2018-07-10T13:17:10Z

Ok what did we decide on the time then? I would say for now either ignore or send down the real time.

@cxzl25 How hard is it to send the actual task info into here so it could use the real time the successful task took? At a glance it doesn't look to hard to add in the additional information into the function calls to pass it into markPartitionCompleted

cxzl25 · 2018-07-10T13:48:09Z

@tgravescs
This is really not difficult.
I'm just not sure if we want to ignore or send down the real time.
Now I have submitted a change, use actual time of successful task.

jiangxb1987 · 2018-07-10T14:42:52Z

The changes LGTM

SparkQA · 2018-07-10T16:58:27Z

Test build #92817 has finished for PR 21656 at commit d8fdceb.

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

SparkQA · 2018-07-10T18:08:03Z

Test build #92820 has finished for PR 21656 at commit 1c1df5c.

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

tgravescs · 2018-07-12T19:59:12Z

@squito are you ok with this approach?

tgravescs · 2018-07-16T15:57:53Z

+1

squito · 2018-07-17T17:05:00Z

lgtm

## What changes were proposed in this pull request? When speculation is enabled, TaskSetManager#markPartitionCompleted should write successful task duration to MedianHeap, not just increase tasksSuccessful. Otherwise when TaskSetManager#checkSpeculatableTasks,tasksSuccessful non-zero, but MedianHeap is empty. Then throw an exception successfulTaskDurations.median java.util.NoSuchElementException: MedianHeap is empty. Finally led to stopping SparkContext. ## How was this patch tested? TaskSetManagerSuite.scala unit test:[SPARK-24677] MedianHeap should not be empty when speculation is enabled Author: sychen <[email protected]> Closes #21656 from cxzl25/fix_MedianHeap_empty. (cherry picked from commit c8bee93) Signed-off-by: Thomas Graves <[email protected]>

tgravescs · 2018-07-18T19:03:37Z

merged thanks @cxzl25

## What changes were proposed in this pull request? When speculation is enabled, TaskSetManager#markPartitionCompleted should write successful task duration to MedianHeap, not just increase tasksSuccessful. Otherwise when TaskSetManager#checkSpeculatableTasks,tasksSuccessful non-zero, but MedianHeap is empty. Then throw an exception successfulTaskDurations.median java.util.NoSuchElementException: MedianHeap is empty. Finally led to stopping SparkContext. ## How was this patch tested? TaskSetManagerSuite.scala unit test:[SPARK-24677] MedianHeap should not be empty when speculation is enabled Author: sychen <[email protected]> Closes apache#21656 from cxzl25/fix_MedianHeap_empty. (cherry picked from commit c8bee93) Signed-off-by: Thomas Graves <[email protected]>

Ref: LIHADOOP-52383 When speculation is enabled, TaskSetManager#markPartitionCompleted should write successful task duration to MedianHeap, not just increase tasksSuccessful. Otherwise when TaskSetManager#checkSpeculatableTasks,tasksSuccessful non-zero, but MedianHeap is empty. Then throw an exception successfulTaskDurations.median java.util.NoSuchElementException: MedianHeap is empty. Finally led to stopping SparkContext. TaskSetManagerSuite.scala unit test:[SPARK-24677] MedianHeap should not be empty when speculation is enabled Author: sychen <[email protected]> Closes apache#21656 from cxzl25/fix_MedianHeap_empty.

MedianHeap is empty when speculation is enabled, causing the SparkCon…

467f0bc

…text to stop.

Add a unit test:[SPARK-24677] MedianHeap should not be empty when spe…

55ddbeb

…culation is enabled

cloud-fan reviewed Jul 5, 2018

View reviewed changes

jiangxb1987 reviewed Jul 5, 2018

View reviewed changes

mridulm reviewed Jul 7, 2018

View reviewed changes

speculationEnabled && !isZombie

3d6682f

cxzl25 changed the title ~~[SPARK-24677][Core]MedianHeap is empty when speculation is enabled, causing the SparkContext to stop~~ [SPARK-24677][Core]Avoid NoSuchElementException from MedianHeap Jul 10, 2018

change test name

d8fdceb

use actual time of successful task

1c1df5c

asfgit closed this in c8bee93 Jul 18, 2018

squito mentioned this pull request Mar 5, 2019

[SPARK-23433][SPARK-25250] [CORE] Later created TaskSet should learn about the finished partitions #23871

Closed

[SPARK-24677][Core]Avoid NoSuchElementException from MedianHeap #21656

[SPARK-24677][Core]Avoid NoSuchElementException from MedianHeap #21656

Uh oh!

Conversation

cxzl25 commented Jun 28, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

maropu commented Jun 28, 2018

Uh oh!

cxzl25 commented Jul 2, 2018

Uh oh!

cxzl25 commented Jul 4, 2018

Uh oh!

squito commented Jul 5, 2018

Uh oh!

cloud-fan Jul 5, 2018

Choose a reason for hiding this comment

Uh oh!

cxzl25 Jul 5, 2018

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jul 5, 2018

Uh oh!

SparkQA commented Jul 5, 2018

Uh oh!

squito commented Jul 5, 2018

Uh oh!

jiangxb1987 Jul 5, 2018

Choose a reason for hiding this comment

Uh oh!

squito Jul 6, 2018

Choose a reason for hiding this comment

Uh oh!

tgravescs commented Jul 6, 2018

Uh oh!

tgravescs commented Jul 6, 2018

Uh oh!

squito commented Jul 6, 2018

Uh oh!

mridulm Jul 7, 2018

Choose a reason for hiding this comment

Uh oh!

tgravescs commented Jul 10, 2018

Uh oh!

cxzl25 commented Jul 10, 2018

Uh oh!

jiangxb1987 commented Jul 10, 2018

Uh oh!

SparkQA commented Jul 10, 2018

Uh oh!

SparkQA commented Jul 10, 2018

Uh oh!

tgravescs commented Jul 12, 2018

Uh oh!

tgravescs commented Jul 16, 2018

Uh oh!

squito commented Jul 17, 2018

Uh oh!

tgravescs commented Jul 18, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

cxzl25 commented Jun 28, 2018 •

edited

Loading