srowen commented on a change in pull request #25612: [SPARK-3137][Core]Replace
the global TorrentBroadcast lock with fine grained locks
URL: https://github.com/apache/spark/pull/25612#discussion_r318824982
##########
File path:
core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
##########
@@ -290,6 +293,42 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj:
T, id: Long)
private object TorrentBroadcast extends Logging {
+ /** Locks to ensure there is only one thread fetching the same
[[TorrentBroadcast]] block. */
+ private val torrentBroadcastLock = new ConcurrentHashMap[BroadcastBlockId,
AnyRef]()
+
+ /** Acquire a lock for fetching a [[TorrentBroadcast]] block. */
+ private def acquireTorrentBroadcastLock(broadcastId: BroadcastBlockId): Unit
= {
+ while (true) {
+ val lock = torrentBroadcastLock.putIfAbsent(broadcastId, new Object)
+ if (lock == null) return
Review comment:
Sorry I get it now. The Object isn't really the lock. It's the presence of
absence of the Object. The Object is used to queue up waiting callers.
OK but this seems simpler if you `computeIfAbsent` a `ReentrantLock` and
just use the lock as a lock directly. It manages all the details of allowing
one caller, waking up waiters, etc. I'd imagine it's more straightforward to
understand?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
