Skip to content

[SPARK-25998] [CORE] Change TorrentBroadcast to hold weak reference of broadcast object #22995

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
bkrieger wants to merge 8 commits into from
Closed

[SPARK-25998] [CORE] Change TorrentBroadcast to hold weak reference of broadcast object #22995

Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a2683b6
use weak reference for torrent broadcast
Nov 8, 2018
99fbeec
fix compile
Nov 9, 2018
5e0a179
fix
Nov 9, 2018
1908b5b
no npe
Nov 9, 2018
24183e5
no option
Nov 9, 2018
f212da3
typo
Nov 9, 2018
09ae762
code review
Nov 26, 2018
4a771cf
change sync
Nov 27, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.broadcast

import java.io._
import java.lang.ref.WeakReference
import java.nio.ByteBuffer
import java.util.zip.Adler32

Expand Down Expand Up @@ -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 weak 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: WeakReference[T] = _

/** The compression codec to use, or None if compression is disabled */
@transient private var compressionCodec: Option[CompressionCodec] = _
Expand Down Expand Up @@ -93,7 +96,14 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
private var checksums: Array[Int] = _

override protected def getValue() = {
_value
val memoized: T = if (_value == null) null.asInstanceOf[T] else _value.get

@srowen srowen Nov 17, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suppose there is a race condition here, in that several threads could end up simultaneously setting the reference. It won't be incorrect as the data ought to be the same. I am not sure of the access pattern for this object; maybe it's always single-threaded. But if lots are reading, you can imagine them all causing a call to readBroadcastBlock() simultaneously.

Introducing another object to lock on is safe and not too much extra legwork. Might be worth it.

Isn't WeakReference cleared on any GC? would SoftReference be better to hold out until memory is exhausted? to avoid re-reading. There's a tradeoff there.

Good idea, just surprisingly full of possible gotchas.

Nit: isn't val memoized = if (_value == null) null else _value.get sufficient?

@bkrieger bkrieger Nov 26, 2018

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. I'll make it synchronized, so it only loads one at a time.

Re: WeakReference, sure, I can change it to SoftReference. That'll be closer to the original behavior, and should still give the improvement we want.

When I try with .asInstanceOf[T] it fails to compile with:

[error] /Users/bkrieger/Documents/git/spark/core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala:98: type mismatch;
[error]  found   : Null(null)
[error]  required: T
[error]     val memoized: T = if (_value == null) null else _value.get
[error]                                           ^
[info] Null(null) <: T?
[info] false
[error] one error found

@srowen srowen Nov 26, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, weird, I thought it would work based on a little local example, but yeah leave the cast in of course.

if (memoized != null) {
memoized
} else {
val newlyRead = readBroadcastBlock()
_value = new WeakReference[T](newlyRead)
newlyRead
}
}

private def calcChecksum(block: ByteBuffer): Int = {
Expand Down