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_r318838642
##########
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:
Isn't this all there is to it, more or less?
```
val lock = torrentBroadcastLock.computeIfAbsent(broadcastId, new
ReentrantLock())
lock.lock()
try {
func
} finally {
lock.unlock()
}
```
The downside I suppose is that you have a map of locks that keeps growing,
but how many broadcast IDs would there be? vs a relatively small lock object
for each.
----------------------------------------------------------------
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]
