[SPARK-6737] Fix memory leak in OutputCommitCoordinator #5397

JoshRosen · 2015-04-07T17:48:25Z

This patch fixes a memory leak in the DAGScheduler, which caused us to leak a map entry per submitted stage. The problem is that the OutputCommitCoordinator needs to be informed when stages end in order to remove entries from its authorizedCommitters map, but the DAGScheduler only called it in one of the four code paths that are used to mark stages as completed.

This patch fixes this issue by consolidating the processing of stage completion into a new markStageAsFinished method and updates DAGSchedulerSuite's assertDataStructuresEmpty assertion to also check the OutputCommitCoordinator data structures. I've also added a comment at the top of DAGScheduler so that we remember to update this test when adding new data structures.

…events.

SparkQA · 2015-04-07T17:53:19Z

Test build #29803 has started for PR 5397 at commit 3052aea.

vanzin · 2015-04-07T18:01:14Z

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

@@ -1264,6 +1270,7 @@ class DAGScheduler(
            try { // cancelTasks will fail if a SchedulerBackend does not implement killTask
              taskScheduler.cancelTasks(stageId, shouldInterruptThread)
              stage.latestInfo.stageFailed(failureReason)
+              outputCommitCoordinator.stageEnd(stage.id)


Wouldn't it be better to put these three lines (at least) in a separate method:

private def endStage(stage: Stage): Unit = { stage.latestInfo.stageFailed(failureReason) outputCommitCoordinator.stageEnd(stage.id) listenerBus.post(SparkListenerStageCompleted(stage.latestInfo)) }

That would make it harder to miss things like this.

It actually looks like handleTaskCompletion has a nested markStageAsFinished method that looks like it should do this. That method also removes the stage from runningStages, which doesn't appear to happen in all of the paths where we post SparkListenerStageCompleted. Let me take a look at this and see whether there's a safe way to refactor things to use this method.

JoshRosen · 2015-04-07T18:21:26Z

@vanzin, I've refactored this code to extract the stage completion code into a new markStageAsFinished method. There's one small change in behavior that I'll comment on inline.

JoshRosen · 2015-04-07T18:22:25Z

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

@@ -1098,7 +1085,6 @@ class DAGScheduler(
          logInfo(s"Marking $failedStage (${failedStage.name}) as failed " +
            s"due to a fetch failure from $mapStage (${mapStage.name})")
          markStageAsFinished(failedStage, Some(failureMessage))
-          runningStages -= failedStage


This was redundant, since markStagesAsFinished removes the stage from runningStages.

SparkQA · 2015-04-07T18:23:19Z

Test build #29806 has started for PR 5397 at commit af3b02f.

JoshRosen · 2015-04-07T18:23:45Z

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

@@ -1263,8 +1269,7 @@ class DAGScheduler(
          if (runningStages.contains(stage)) {
            try { // cancelTasks will fail if a SchedulerBackend does not implement killTask
              taskScheduler.cancelTasks(stageId, shouldInterruptThread)
-              stage.latestInfo.stageFailed(failureReason)
-              listenerBus.post(SparkListenerStageCompleted(stage.latestInfo))
+              markStageAsFinished(stage, Some(failureReason))


There's a slight change in behavior here: in the old code, we never removed the stage from runningStages here even though we posted a StageCompletion listener event. I think this was probably a bug. In this updated code, markStagesAsFinished will remove the stage from runningStages.

vanzin · 2015-04-07T18:33:06Z

LGTM.

SparkQA · 2015-04-07T19:20:21Z

Test build #29803 has finished for PR 5397 at commit 3052aea.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

AmplabJenkins · 2015-04-07T19:20:25Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29803/
Test PASSed.

SparkQA · 2015-04-07T19:48:11Z

Test build #29806 has finished for PR 5397 at commit af3b02f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

AmplabJenkins · 2015-04-07T19:48:15Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29806/
Test PASSed.

aarondav · 2015-04-07T22:31:33Z

LGTM too.

JoshRosen · 2015-04-07T23:17:56Z

Thanks for reviewing. I'm going to merge this into master and open a separate PR to backport to branch-1.3.

JoshRosen · 2015-04-07T23:25:29Z

Actually, it appears that the merge conflicts with branch-1.3 are trivial to resolve, so I'll perform a cherry pick, run the tests locally, the monitor Jenkins for the backport commit.

This patch fixes a memory leak in the DAGScheduler, which caused us to leak a map entry per submitted stage. The problem is that the OutputCommitCoordinator needs to be informed when stages end in order to remove entries from its `authorizedCommitters` map, but the DAGScheduler only called it in one of the four code paths that are used to mark stages as completed. This patch fixes this issue by consolidating the processing of stage completion into a new `markStageAsFinished` method and updates DAGSchedulerSuite's `assertDataStructuresEmpty` assertion to also check the OutputCommitCoordinator data structures. I've also added a comment at the top of DAGScheduler so that we remember to update this test when adding new data structures. Author: Josh Rosen <joshrosen@databricks.com> Closes #5397 from JoshRosen/SPARK-6737 and squashes the following commits: af3b02f [Josh Rosen] Consolidate stage completion handling code in a single method. e96ce3a [Josh Rosen] Consolidate stage completion handling code in a single method. 3052aea [Josh Rosen] Comment update 7896899 [Josh Rosen] Fix SPARK-6737 by informing OutputCommitCoordinator of all stage end events. 4ead1dc [Josh Rosen] Add regression tests for SPARK-6737 (cherry picked from commit c83e039) Signed-off-by: Josh Rosen <joshrosen@databricks.com> Conflicts: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

JoshRosen added 3 commits April 7, 2015 10:42

Add regression tests for SPARK-6737

4ead1dc

Fix SPARK-6737 by informing OutputCommitCoordinator of all stage end …

7896899

…events.

Comment update

3052aea

vanzin reviewed Apr 7, 2015
View reviewed changes

JoshRosen added 2 commits April 7, 2015 11:18

Consolidate stage completion handling code in a single method.

e96ce3a

Consolidate stage completion handling code in a single method.

af3b02f

JoshRosen reviewed Apr 7, 2015
View reviewed changes

asfgit closed this in c83e039 Apr 7, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[SPARK-6737] Fix memory leak in OutputCommitCoordinator #5397

[SPARK-6737] Fix memory leak in OutputCommitCoordinator #5397

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

vanzin Apr 7, 2015

Uh oh!

JoshRosen Apr 7, 2015

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

JoshRosen Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

JoshRosen Apr 7, 2015

Uh oh!

vanzin commented Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

AmplabJenkins commented Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

AmplabJenkins commented Apr 7, 2015

Uh oh!

aarondav commented Apr 7, 2015

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

Uh oh!

[SPARK-6737] Fix memory leak in OutputCommitCoordinator #5397

[SPARK-6737] Fix memory leak in OutputCommitCoordinator #5397

Uh oh!

Conversation

JoshRosen commented Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

vanzin Apr 7, 2015

Choose a reason for hiding this comment

Uh oh!

JoshRosen Apr 7, 2015

Choose a reason for hiding this comment

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

JoshRosen Apr 7, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

JoshRosen Apr 7, 2015

Choose a reason for hiding this comment

Uh oh!

vanzin commented Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

AmplabJenkins commented Apr 7, 2015

Uh oh!

SparkQA commented Apr 7, 2015

Uh oh!

AmplabJenkins commented Apr 7, 2015

Uh oh!

aarondav commented Apr 7, 2015

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

JoshRosen commented Apr 7, 2015

Uh oh!

Uh oh!