[SPARK-22986][Core] Use a cache to avoid instantiating multiple instances of broadcast variable values #20183

ho3rexqj · 2018-01-08T03:11:48Z

When resources happen to be constrained on an executor the first time a broadcast variable is instantiated it is persisted to disk by the BlockManager. Consequently, every subsequent call to TorrentBroadcast::readBroadcastBlock from other instances of that broadcast variable spawns another instance of the underlying value. That is, broadcast variables are spawned once per executor unless memory is constrained, in which case every instance of a broadcast variable is provided with a unique copy of the underlying value.

This patch fixes the above by explicitly caching the underlying values using weak references in a ReferenceMap.

…variables.

…multiple times on executors when memory is constrained.

jiangxb1987

This should be useful, cc @cloud-fan to also take a look.

jiangxb1987 · 2018-01-11T07:34:33Z

core/src/test/scala/org/apache/spark/broadcast/BroadcastSuite.scala


+  test("One broadcast value instance per executor") {
+    val conf = new SparkConf()
+      .setMaster("local[10]")


nit: normally local[4] is sufficient.

jiangxb1987 · 2018-01-11T07:36:29Z

Please also update the PR title and description. Thanks!

ho3rexqj · 2018-01-11T14:43:34Z

Updated the above, thanks!

cloud-fan · 2018-01-12T02:23:18Z

core/src/main/scala/org/apache/spark/broadcast/BroadcastManager.scala

  private val nextBroadcastId = new AtomicLong(0)

+  private[broadcast] val cachedValues = {
+    new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)


why is the key not a weak reference?

Suppose the first thread to request the broadcast variable's value destroyed it's instance of the broadcast variable (which, I believe, is what will happen when that thread finishes processing it's partition) - if the key were a weak reference in the above cache it would become eligible for GC at that point. I'm reasonably certain at that point the associated key/value pair would be removed from the cache; in other words, if the key were a weak reference the key/value pair would be removed as soon as the key or value was garbage collected.

Note that I haven't used ReferenceMap extensively, so I could be wrong about the above - feel free to correct me if that's the case.

what's the difference between this and "weak key, hard value"?

This is the state of an executor at some point in time:

Cache: IdInstance1 => ValueInstance1
Thread1: TorrentBroadcastInstance1(broadcastId = IdInstance1, value = ValueInstance1)
Thread2: TorrentBroadcastInstance2(broadcastId = IdInstance2, value = ValueInstance1)
Thread3: TorrentBroadcastInstance3(broadcastId = IdInstance3, value = ValueInstance1)
Thread4: TorrentBroadcastInstance4(broadcastId = IdInstance4, value = ValueInstance1)

After some time Thread1 finishes process the partition it's working on and starts on the next - the state becomes:

Cache: IdInstance1 => ValueInstance1
Thread1: TorrentBroadcastInstance5(broadcastId = IdInstance5, value = ValueInstance1)
Thread2: TorrentBroadcastInstance2(broadcastId = IdInstance2, value = ValueInstance1)
Thread3: TorrentBroadcastInstance3(broadcastId = IdInstance3, value = ValueInstance1)
Thread4: TorrentBroadcastInstance4(broadcastId = IdInstance4, value = ValueInstance1)

At some point the GC destroys TorrentBroadcastInstance1. Now, if the key is a weak reference, the state becomes:

Cache: Empty
Thread1: TorrentBroadcastInstance5(broadcastId = IdInstance5, value = ValueInstance1)
Thread2: TorrentBroadcastInstance2(broadcastId = IdInstance2, value = ValueInstance1)
Thread3: TorrentBroadcastInstance3(broadcastId = IdInstance3, value = ValueInstance1)
Thread4: TorrentBroadcastInstance4(broadcastId = IdInstance4, value = ValueInstance1)

The next thread to finish processing a partition then creates a new instance of the broadcast value:

Cache: IdInstance6 => ValueInstance2
Thread1: TorrentBroadcastInstance5(broadcastId = IdInstance5, value = ValueInstance1)
Thread2: TorrentBroadcastInstance2(broadcastId = IdInstance2, value = ValueInstance1)
Thread3: TorrentBroadcastInstance6(broadcastId = IdInstance6, value = ValueInstance2)
Thread4: TorrentBroadcastInstance4(broadcastId = IdInstance4, value = ValueInstance1)

On the other hand, if the key is a strong reference the the value is weak, the cached value isn't eligible for GC above. As such, when Thread3 finishes processing it's partition and starts the next, the state becomes:

Cache: IdInstance1 => ValueInstance1
Thread1: TorrentBroadcastInstance5(broadcastId = IdInstance5, value = ValueInstance1)
Thread2: TorrentBroadcastInstance2(broadcastId = IdInstance2, value = ValueInstance1)
Thread3: TorrentBroadcastInstance6(broadcastId = IdInstance6, value = ValueInstance1)
Thread4: TorrentBroadcastInstance4(broadcastId = IdInstance4, value = ValueInstance1)

ah i see, because we never get the key reference outside of the map, makes sense.

This is what I found from the doc:

Hash-based Map implementation that allows mappings to be removed by the garbage collector.
When you construct a ReferenceMap, you can specify what kind of references are used to store the map's keys and values. If non-hard references are used, then the garbage collector can remove mappings if a key or value becomes unreachable, or if the JVM's memory is running low. For information on how the different reference types behave, see Reference.

It only mentions that non-hard references can be removed by GC, please correct me if I'm wrong.

That should be fine even if that hard references not removed, since the memory consumption should be quite minor.

BTW - welcome back @jerryshao ! long time no see!

can remove mappings... That's why I say entries can be removed automatically.

ohh, I see. Thanks!

cloud-fan · 2018-01-12T02:24:29Z

core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

-              throw new SparkException(s"Failed to store $broadcastId in BlockManager")
+      val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
+
+      Option(broadcastCache.get(broadcastId)).map(_.asInstanceOf[T]).getOrElse({


nit: code style

...getOrElse { xxx }

Fixed, thanks!

cloud-fan · 2018-01-12T02:25:26Z

core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

+      val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
+
+      Option(broadcastCache.get(broadcastId)).map(_.asInstanceOf[T]).getOrElse({
+        setConf(SparkEnv.get.conf)


just to confirm, the following code is exactly same as before, right?

Yes - everything within the getOrElse block is unchanged.

No, sorry - the cache update takes place within that block. With the exception of those blocks (lines 220-222 and lines 244-246), yes.

cloud-fan · 2018-01-12T02:28:31Z

ok to test

jerryshao · 2018-01-12T06:22:24Z

core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala

-              throw new SparkException(s"Failed to store $broadcastId in BlockManager")
+      val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
+
+      Option(broadcastCache.get(broadcastId)).map(_.asInstanceOf[T]).getOrElse {


Is this ReferenceMap thread safe?

ReferenceMap is not thread safe, no - however, all operations on broadcastCache occur within the context of a synchronized block; TorrentBroadcast.scala lines 208-254.

I see, thanks for pointing out.

SparkQA · 2018-01-12T06:38:09Z

Test build #86010 has finished for PR 20183 at commit 8e13585.

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

SparkQA · 2018-01-12T07:16:44Z

Test build #86016 has finished for PR 20183 at commit a6e5e81.

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

cloud-fan · 2018-01-12T07:27:14Z

thanks, merging to master/2.3!

…nces of broadcast variable values When resources happen to be constrained on an executor the first time a broadcast variable is instantiated it is persisted to disk by the BlockManager. Consequently, every subsequent call to TorrentBroadcast::readBroadcastBlock from other instances of that broadcast variable spawns another instance of the underlying value. That is, broadcast variables are spawned once per executor **unless** memory is constrained, in which case every instance of a broadcast variable is provided with a unique copy of the underlying value. This patch fixes the above by explicitly caching the underlying values using weak references in a ReferenceMap. Author: ho3rexqj <[email protected]> Closes #20183 from ho3rexqj/fix/cache-broadcast-values. (cherry picked from commit cbe7c6f) Signed-off-by: Wenchen Fan <[email protected]>

ho3rexqj added 2 commits January 7, 2018 20:44

Adding tests to illustrate the problem we were having with broadcast …

45d5024

…variables.

Adding broadcast value cache to avoid instantiating broadcast values …

c3e2f42

…multiple times on executors when memory is constrained.

jiangxb1987 reviewed Jan 11, 2018

View reviewed changes

Reduced the number of local threads in tests.

8e13585

ho3rexqj changed the title ~~[SPARK-22986][Core] Fix/cache broadcast values~~ [SPARK-22986][Core] Use a local cache to avoid instantiating multiple instances of broadcast variable values Jan 11, 2018

ho3rexqj changed the title ~~[SPARK-22986][Core] Use a local cache to avoid instantiating multiple instances of broadcast variable values~~ [SPARK-22986][Core] Use a cache to avoid instantiating multiple instances of broadcast variable values Jan 11, 2018

cloud-fan reviewed Jan 12, 2018

View reviewed changes

Code style fix.

a6e5e81

jerryshao reviewed Jan 12, 2018

View reviewed changes

asfgit closed this in cbe7c6f Jan 12, 2018

ho3rexqj deleted the fix/cache-broadcast-values branch January 12, 2018 14:01

JoshRosen mentioned this pull request Aug 9, 2022

[SPARK-39983][CORE][SQL] Do not cache unserialized broadcast relations on the driver #37413

Closed

[SPARK-22986][Core] Use a cache to avoid instantiating multiple instances of broadcast variable values #20183

[SPARK-22986][Core] Use a cache to avoid instantiating multiple instances of broadcast variable values #20183

Uh oh!

Conversation

ho3rexqj commented Jan 8, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jiangxb1987 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jiangxb1987 commented Jan 11, 2018

Uh oh!

ho3rexqj commented Jan 11, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jan 12, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jan 12, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 12, 2018

Uh oh!

SparkQA commented Jan 12, 2018

Uh oh!

cloud-fan commented Jan 12, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

ho3rexqj commented Jan 8, 2018 •

edited

Loading

cloud-fan Jan 12, 2018 •

edited

Loading