You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2018/11/28 15:22:50 UTC
spark git commit: [SPARK-25998][CORE] Change TorrentBroadcast to hold
weak reference of broadcast object
Repository: spark
Updated Branches:
refs/heads/master ce61bac1d -> 87bd9c75d
[SPARK-25998][CORE] Change TorrentBroadcast to hold weak reference of broadcast object
## What changes were proposed in this pull request?
This PR changes the broadcast object in TorrentBroadcast from a strong reference to a weak reference. This allows it to be garbage collected even if the Dataset is held in memory. This is ok, because the broadcast object can always be re-read.
## How was this patch tested?
Tested in Spark shell by taking a heap dump, full repro steps listed in https://issues.apache.org/jira/browse/SPARK-25998.
Closes #22995 from bkrieger/bk/torrent-broadcast-weak.
Authored-by: Brandon Krieger <bk...@palantir.com>
Signed-off-by: Sean Owen <se...@databricks.com>
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/87bd9c75
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/87bd9c75
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/87bd9c75
Branch: refs/heads/master
Commit: 87bd9c75df6b67bef903751269a4fd381f9140d9
Parents: ce61bac
Author: Brandon Krieger <bk...@palantir.com>
Authored: Wed Nov 28 07:22:48 2018 -0800
Committer: Sean Owen <se...@databricks.com>
Committed: Wed Nov 28 07:22:48 2018 -0800
----------------------------------------------------------------------
.../spark/broadcast/TorrentBroadcast.scala | 22 ++++++++++++++------
1 file changed, 16 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/87bd9c75/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala b/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
index cbd49e0..26ead57 100644
--- a/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
+++ b/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
@@ -18,6 +18,7 @@
package org.apache.spark.broadcast
import java.io._
+import java.lang.ref.SoftReference
import java.nio.ByteBuffer
import java.util.zip.Adler32
@@ -61,9 +62,11 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
* Value of the broadcast object on executors. This is reconstructed by [[readBroadcastBlock]],
* which builds this value by reading blocks from the driver and/or other executors.
*
- * On the driver, if the value is required, it is read lazily from the block manager.
+ * On the driver, if the value is required, it is read lazily from the block manager. We hold
+ * a soft reference so that it can be garbage collected if required, as we can always reconstruct
+ * in the future.
*/
- @transient private lazy val _value: T = readBroadcastBlock()
+ @transient private var _value: SoftReference[T] = _
/** The compression codec to use, or None if compression is disabled */
@transient private var compressionCodec: Option[CompressionCodec] = _
@@ -92,8 +95,15 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
/** The checksum for all the blocks. */
private var checksums: Array[Int] = _
- override protected def getValue() = {
- _value
+ override protected def getValue() = synchronized {
+ val memoized: T = if (_value == null) null.asInstanceOf[T] else _value.get
+ if (memoized != null) {
+ memoized
+ } else {
+ val newlyRead = readBroadcastBlock()
+ _value = new SoftReference[T](newlyRead)
+ newlyRead
+ }
}
private def calcChecksum(block: ByteBuffer): Int = {
@@ -205,8 +215,8 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
}
private def readBroadcastBlock(): T = Utils.tryOrIOException {
- TorrentBroadcast.synchronized {
- val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
+ val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
+ broadcastCache.synchronized {
Option(broadcastCache.get(broadcastId)).map(_.asInstanceOf[T]).getOrElse {
setConf(SparkEnv.get.conf)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org