[SPARK-3885] Provide mechanism to remove accumulators once they are no longer used #4021

ilganeli · 2015-01-13T17:53:48Z

Instead of storing a strong reference to accumulators, I've replaced this with a weak reference and updated any code that uses these accumulators to check whether the reference resolves before using the accumulator. A weak reference will be cleared when there is no longer an existing copy of the variable versus using a soft reference in which case accumulators would only be cleared when the GC explicitly ran out of memory.

… references to allow garbage collection of old accumulators

… WeakRef Accumulator storage

SparkQA · 2015-01-13T17:57:42Z

Test build #25475 has started for PR 4021 at commit 28f705c.

This patch merges cleanly.

SparkQA · 2015-01-13T19:00:23Z

Test build #25475 has finished for PR 4021 at commit 28f705c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- val classServer = new HttpServer(conf, outputDir, new SecurityManager(conf), classServerPort, "HTTP class server")

AmplabJenkins · 2015-01-13T19:00:27Z

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

JoshRosen · 2015-01-13T19:48:57Z

It looks like a number of file permissions changes got mixed into this PR; mind reverting those changes?

JoshRosen · 2015-01-13T19:53:07Z

core/src/main/scala/org/apache/spark/Accumulators.scala

Guava MapMaper supports weakValues; not sure if we want to use that here, since it's not super Scala-friendly (e.g. returns nulls, etc): http://docs.guava-libraries.googlecode.com/git-history/release/javadoc/com/google/common/collect/MapMaker.html

Hi Josh - are you suggesting to replace this snippet with a MapMaker just to simplify the initialization code? I believe the usage of either object would be the same - do you see a specific advantage to trying to use the MapMaker?

Let's just leave this as-is; I don't think MapMaker will buy us much now that I think about it.

SparkQA · 2015-01-15T15:42:43Z

Test build #25606 has started for PR 4021 at commit 18d62ec.

This patch merges cleanly.

ilganeli · 2015-01-15T15:45:37Z

I've updated the code to throw an exception in the error case you mentioned and I've reverted the file permission change. Thanks!

SparkQA · 2015-01-15T16:51:01Z

Test build #25606 has finished for PR 4021 at commit 18d62ec.

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

AmplabJenkins · 2015-01-15T16:51:05Z

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

JoshRosen · 2015-01-15T20:09:02Z

This looks pretty nice. I wonder if there's a good way to add a unit test for this behavior (maybe in AccumulatorSuite or something) to ensure that we don't accidentally break the garbage-collection. This could be tricky, though, since we don't want to introduce a test that's flaky due to nondeterminism. Maybe a good test would be something like creating a job that uses an accumulator, storing its id, letting all references to it fall out of scope, then calling System.gc() and asserting that dereferencing the WeakReference returns null. This would help to avoid memory leaks if we accidentally write some new code in the future that stores strong references to accumulators.

ilganeli · 2015-01-15T21:51:36Z

Thanks for the suggestion Josh - that sounds quite reasonable. I was having trouble coming up with an effective test for this. I'll add a test based on this idea to the accumulator suite.

ilganeli · 2015-02-07T18:21:53Z

Hi @JoshRosen, I've added a test for this patch to the Accumulator Suite. Please let me know if you think it's sufficient. Thanks!

SparkQA · 2015-02-07T18:22:56Z

Test build #27005 has started for PR 4021 at commit c8e0f2b.

This patch merges cleanly.

SparkQA · 2015-02-07T18:59:02Z

Test build #27005 has finished for PR 4021 at commit c8e0f2b.

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

AmplabJenkins · 2015-02-07T18:59:05Z

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

AmplabJenkins · 2015-02-21T01:31:16Z

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

SparkQA · 2015-02-21T01:33:00Z

Test build #27801 has started for PR 4021 at commit 8510943.

This patch merges cleanly.

SparkQA · 2015-02-21T02:12:12Z

Test build #27801 has finished for PR 4021 at commit 8510943.

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

AmplabJenkins · 2015-02-21T02:12:16Z

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

SparkQA · 2015-02-21T03:22:38Z

Test build #27803 has started for PR 4021 at commit 4ba9575.

This patch merges cleanly.

SparkQA · 2015-02-21T04:45:50Z

Test build #27803 has finished for PR 4021 at commit 4ba9575.

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

AmplabJenkins · 2015-02-21T04:45:55Z

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

JoshRosen · 2015-02-23T06:40:22Z

I just realized that ContextCleaner won't manage cleanup of accumulators in the thread-local accumulator registries in executors, but I don't think this is a big deal: we're more concerned about the driver OOMing in a long-running app than the executors, and it would be prohibitively expensive to add a bunch of driver -> executor RPCs just to reclaim one map entry.

Therefore, this looks good to me. There's one minor import-ordering nit, but I'll just fix that up myself on merge. Thanks for working on this, esp. since this code is a bit tricky to test.

JoshRosen · 2015-02-23T06:45:59Z

Actually, we won't leak accumulators on executors because we clear the thread-local between tasks. So this is great!

JoshRosen · 2015-02-23T07:00:23Z

I've merged this into master (1.4.0). Thanks!

@ilganeli, it looks like your text editor might be leaving trailing whitespace on lines, which makes the diffs look kind of messy:

I fixed this myself when committing, so not a huge deal, but you might consider changing your editor settings to prevent htis in the future (I have a special vim setting that complains about trailing whitespace, and I think there's a way to configure IntelliJ to do the same).

ilganeli · 2015-02-23T07:27:50Z

Hi Josh - thanks for the review and suggestions. With regards to the local thread cleanup I didn't add anything since as you pointed out it already gets cleaned up. Cheers!

tedyu · 2015-02-23T22:32:54Z

core/src/main/scala/org/apache/spark/Accumulators.scala

If Accumulator is garbage collected, should we log and continue ?

The exception thrown here is caught at higher levels of the stack. For example, DAGScheduler wraps calls to accumulator methods in a try block and logs any uncaught exceptions. Have you run into a case where the current behavior causes a problem?

tedyu · 2015-02-23T23:59:27Z

Not yet.

JoshRosen · 2015-02-28T05:24:52Z

I think that this patch may have introduced a bug that may cause accumulator updates to be lost:
https://issues.apache.org/jira/browse/SPARK-6075

I'm still trying to see if I can spot the problem, but my hunch is that maybe the localAccums thread-local maps should not hold weak references. When deserializing an accumulator in an executor and registering it with localAccums, is there ever a moment in which the accumulator has no strong references pointing to it? Does someone object hold a strong reference to an accumulator while it's being deserialized? If not, this could lead to it being dropped from the localAccums map, causing that task's accumulator updates to be lost.

JoshRosen · 2015-02-28T05:50:13Z

Another thought: if register() is somehow called twice for the same accumulator, then it looks like we'll silently overwrite the existing value in localAccums. We should probably throw an exception instead, since that scenario could lead to lost updates.

JoshRosen · 2015-02-28T19:53:57Z

I think I've figured it out: consider the lifecycle of an accumulator in a task, say ShuffleMapTask: on the executor, each task deserializes its own copy of the RDD inside of its runTask method, so the strong reference to the RDD disappears at the end of runTask. In Executor.run(), we call Accumulators.values after runTask has exited, so there's a small window in which the tasks's RDD can be GC'd, causing accumulators to be GC'd as well because there are no longer any strong references to them.

The fix is to keep strong references in localAccums, since we clear this at the end of each task anyways. I'm glad that I was able to figure out precisely why this was necessary and sorry that I missed this during review; I'll submit a fix shortly. In terms of preventative measures, it might be a good idea to write up the lifetime / lifecycle of objects' strong references whenever we're using WeakReferences, since the process of explicitly writing that out would prevent these sorts of mistakes in the future.

JoshRosen · 2015-02-28T20:03:06Z

Actually, we won't leak accumulators on executors because we clear the thread-local between tasks.

At this point, I should have spotted that it doesn't make sense to store WeakReferences if those references are going to be destroyed at the end of tasks anyways.

JoshRosen · 2015-02-28T20:27:30Z

core/src/main/scala/org/apache/spark/Accumulators.scala

This was dumb for me to overlook, too: if the data structure is invalid, then this would just silently ignore it; even if this was a rare error-condition, there should have been a warning here.

JoshRosen · 2015-02-28T20:44:05Z

I have a fix at #4835.

ilganeli · 2015-03-01T00:36:28Z

Thanks for the detailed write up and the fix.

…store WeakReferences in localAccums map This fixes a non-deterministic bug introduced in #4021 that could cause tasks' accumulator updates to be lost. The problem is that `localAccums` should not hold weak references: after the task finishes running there won't be any strong references to these local accumulators, so they can get garbage-collected before the executor reads the `localAccums` map. We don't need weak references here anyways, since this map is cleared at the end of each task. Author: Josh Rosen <[email protected]> Closes #4835 from JoshRosen/SPARK-6075 and squashes the following commits: 4f4b5b2 [Josh Rosen] Remove defensive assertions that caused test failures in code unrelated to this change 120c7b0 [Josh Rosen] [SPARK-6075] Do not store WeakReferences in localAccums map

Ilya Ganelin added 8 commits January 7, 2015 14:57

Updated Accumulators class to store weak references instead of strong…

a77d11b

… references to allow garbage collection of old accumulators

Merge remote-tracking branch 'upstream/master' into SPARK-3885

cbb9023

Merge remote-tracking branch 'upstream/master' into SPARK-3885

c49066a

Updated DAGScheduler and Suite to correctly use new implementation of…

3350852

… WeakRef Accumulator storage

Updated DAGSchedulerSUite to fix bug

0746e61

Removed obsolete comment

d78f4bf

reset

b820ab4

Merge remote-tracking branch 'upstream/master' into SPARK-3885

28f705c

JoshRosen reviewed Jan 13, 2015
View reviewed changes

Ilya Ganelin added 2 commits January 15, 2015 10:36

Reverting permissions changes

9a81928

Updated to throw Exception when accessing a GCd accumulator

18d62ec

Ilya Ganelin added 4 commits January 15, 2015 17:02

Added test for accumulator garbage collection

7414a9c

Fixing gc test

8722b63

Still not being properly garbage collected

94ce754

Added working test for accumulator garbage collection

c8e0f2b

Fixed error in test suite

4ba9575

asfgit closed this in 95cd643 Feb 23, 2015

tedyu reviewed Feb 23, 2015
View reviewed changes

JoshRosen mentioned this pull request Feb 28, 2015

[SPARK-6075] Fix bug in that caused lost accumulator updates: do not store WeakReferences in localAccums map #4835

Closed

JoshRosen reviewed Feb 28, 2015
View reviewed changes

[SPARK-3885] Provide mechanism to remove accumulators once they are no longer used #4021

[SPARK-3885] Provide mechanism to remove accumulators once they are no longer used #4021

Uh oh!

Conversation

ilganeli commented Jan 13, 2015

Uh oh!

SparkQA commented Jan 13, 2015

Uh oh!

SparkQA commented Jan 13, 2015

Uh oh!

AmplabJenkins commented Jan 13, 2015

Uh oh!

JoshRosen commented Jan 13, 2015

Uh oh!

JoshRosen Jan 13, 2015

Choose a reason for hiding this comment

Uh oh!

ilganeli Jan 15, 2015

Choose a reason for hiding this comment

Uh oh!

JoshRosen Jan 15, 2015

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 15, 2015

Uh oh!

ilganeli commented Jan 15, 2015

Uh oh!

SparkQA commented Jan 15, 2015

Uh oh!

AmplabJenkins commented Jan 15, 2015

Uh oh!

JoshRosen commented Jan 15, 2015

Uh oh!

ilganeli commented Jan 15, 2015

Uh oh!

ilganeli commented Feb 7, 2015

Uh oh!

SparkQA commented Feb 7, 2015

Uh oh!

SparkQA commented Feb 7, 2015

Uh oh!

AmplabJenkins commented Feb 7, 2015

Uh oh!

AmplabJenkins commented Feb 21, 2015

Uh oh!

SparkQA commented Feb 21, 2015

Uh oh!

SparkQA commented Feb 21, 2015

Uh oh!

AmplabJenkins commented Feb 21, 2015

Uh oh!

SparkQA commented Feb 21, 2015

Uh oh!

SparkQA commented Feb 21, 2015

Uh oh!

AmplabJenkins commented Feb 21, 2015

Uh oh!

JoshRosen commented Feb 23, 2015

Uh oh!

JoshRosen commented Feb 23, 2015

Uh oh!

JoshRosen commented Feb 23, 2015

Uh oh!

ilganeli commented Feb 23, 2015

Uh oh!

tedyu Feb 23, 2015

Choose a reason for hiding this comment

Uh oh!

JoshRosen Feb 23, 2015

Choose a reason for hiding this comment

Uh oh!

tedyu commented Feb 23, 2015

Uh oh!

JoshRosen commented Feb 28, 2015

Uh oh!

JoshRosen commented Feb 28, 2015

Uh oh!

JoshRosen commented Feb 28, 2015

Uh oh!

JoshRosen commented Feb 28, 2015