[SPARK-3137][Core]Replace the global TorrentBroadcast lock with fine grained KeyLock by zsxwing · Pull Request #25612 · apache/spark

zsxwing · 2019-08-28T18:00:07Z

What changes were proposed in this pull request?

This PR provides a new lock mechanism KeyLock to lock with a given key. Also use this new lock in TorrentBroadcast to avoid blocking tasks from fetching different broadcast values.

Why are the changes needed?

TorrentBroadcast.readObject uses a global lock so only one task can be fetching the blocks at the same time. This is not optimal if we are running multiple stages concurrently because they should be able to independently fetch their own blocks.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests.

zsxwing · 2019-08-28T18:01:44Z

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

-    new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)
+    Collections.synchronizedMap(
+      new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)
+        .asInstanceOf[java.util.Map[Any, Any]]


This one now needs protection as the global lock is removed. Casting it because Scala compiler cannot figure it out.

nit: would it make compiler happy and be nicer if the val was typed (val cachedValues: JMap[Any, Any]) instead of doing asInstanceOf?

No. The compiler doesn't like Java classes :P

zsxwing · 2019-08-28T18:02:37Z

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

   * Otherwise, if not running in a task then immediately release the lock.
   */
-  private def releaseLock(blockId: BlockId): Unit = {
+  private def releaseBlockManagerLock(blockId: BlockId): Unit = {


Rename this method to avoid confusing.

gatorsmile · 2019-08-28T18:46:20Z

cc @Ngone51

SparkQA · 2019-08-28T20:29:44Z

Test build #109880 has finished for PR 25612 at commit 290a3e5.

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

srowen · 2019-08-28T22:03:05Z

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

+  private def acquireTorrentBroadcastLock(broadcastId: BroadcastBlockId): Unit = {
+    while (true) {
+      val lock = torrentBroadcastLock.putIfAbsent(broadcastId, new Object)
+      if (lock == null) return


Can this happen?

Yep when no such key in ConcurrentHashMap.

Right OK, but then it isn't locking on anything the first time, how does this work?

The caller adding an object to ConcurrentHashMap successfully wins the lock and acquireTorrentBroadcastLock will return. Others should wait until they add an object to ConcurrentHashMap successfully.

In releaseTorrentBroadcastLock, it will remove the object from ConcurrentHashMap and wake up threads waiting for this object to retry.

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?

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?

The unlock code path is pretty tricky. Needs to unlock and remove from the map (the removal will allow a thread to put a new lock). I can try to make these two actions atomic by adding a global lock. But that's more complicated than the current way.

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.

The number of broadcast IDs is significant enough for a long running application. Even if the user doesn't use broadcast, we will still use it internally, for example, task binary is a broadcast object, a Spark SQL query may use broadcast join...

OK, makes sense. Well, could still make sense to use a Condition but it wouldn't change much.

srowen · 2019-08-28T22:03:46Z

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

+      if (lock == null) return
+      lock.synchronized {
+        while (torrentBroadcastLock.get(broadcastId) eq lock) {
+          lock.wait()


How about just using a Java Lock implementation rather than re-roll the signalling?

Using a <ReentrantLock, Condition> pair? The codes would be more complicated.

No, just a ReentrantLock. I think you don't need to deal with spurious wakeups then either

I just tried ReentrantLock but didn't find a simple way to implement the same function.

Why not just stripe over a fixed number of Reentrant locks? This should reduce contention enough (and potential collissions can be controlled with numLocks) is easy to understand and doesn't reimplement its own lock queues. It's also fully interruptible, while in the current implementation the threads that are queueing for the lock are not interruptible.

private val numLocks = 100 private val locks = (1 to numLocks).map(_ => new java.util.concurrent.locks.ReentrantLock) private def withTorrentBroadcastLock[T](broadcastId: BroadcastBlockId)(func: => T): T = { val lock = locks(broadcastId.broadcastId % numLocks) lock.lockInterruptibly() try { func } finally { lock.unlock() } }

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

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

core/src/main/scala/org/apache/spark/util/KeyLock.scala

SparkQA · 2019-08-29T20:49:05Z

Test build #109916 has finished for PR 25612 at commit 2cf1826.

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

SparkQA · 2019-08-29T20:57:00Z

Test build #109917 has finished for PR 25612 at commit 844be5c.

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

Ngone51 · 2019-08-30T13:59:15Z

core/src/main/scala/org/apache/spark/util/KeyLock.scala

+      val lock = lockMap.putIfAbsent(key, new Object)
+      if (lock == null) return
+      lock.synchronized {
+        while (lockMap.get(key) eq lock) {


After releasing keylock, if a new thread for the same broadcastId enters to put a new object before another queueing thread to check lockMap.get(key) eq lock, both threads could get the keylock finally ?

The thread inside this inner while loop will end, and will retry the outer while loop. The line if (lock == null) return is the only exit. When acquireLock returns, this thread has putIfAbsent a new object to lockMap successfully.

zsxwing · 2019-08-30T17:23:11Z

@juliuszsompolski @srowen I added tests. Please take another look.

srowen · 2019-08-30T17:35:33Z

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

@@ -56,7 +57,10 @@ private[spark] class BroadcastManager(
  private val nextBroadcastId = new AtomicLong(0)

  private[broadcast] val cachedValues = {


You don't need the braces here, but I wouldn't bother unless you're changing this file again

SparkQA · 2019-08-30T19:37:30Z

Test build #109955 has finished for PR 25612 at commit 2e9639c.

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

juliuszsompolski

LGTM!
Made two tiny nits reading through it.

juliuszsompolski · 2019-09-03T18:01:58Z

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

-    new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)
+    Collections.synchronizedMap(
+      new ReferenceMap(AbstractReferenceMap.HARD, AbstractReferenceMap.WEAK)
+        .asInstanceOf[java.util.Map[Any, Any]]


nit: would it make compiler happy and be nicer if the val was typed (val cachedValues: JMap[Any, Any]) instead of doing asInstanceOf?

juliuszsompolski · 2019-09-03T18:03:11Z

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

-    val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
-    broadcastCache.synchronized {
+    TorrentBroadcast.torrentBroadcastLock.withLock(broadcastId) {
+      // As we just lock based on `broadcastId`, whenever using `broadcastCache`, we should just


nit: just -> only

SparkQA · 2019-09-03T20:49:59Z

Test build #110051 has finished for PR 25612 at commit 9b59ca2.

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

zsxwing · 2019-09-03T21:07:38Z

Thanks! Merging to master.

jiangxb1987 · 2019-09-03T21:12:30Z

A late LGTM :)

### What changes were proposed in this pull request? Use `KeyLock` added in #25612 to simplify `MapOutputTracker.getStatuses`. It also has some improvement after the refactoring: - `InterruptedException` is no longer sallowed. - When a shuffle block is fetched, we don't need to wake up unrelated sleeping threads. ### Why are the changes needed? `MapOutputTracker.getStatuses` is pretty hard to maintain right now because it has a special lock mechanism which we needs to pay attention to whenever updating this method. As we can use `KeyLock` to hide the complexity of locking behind a dedicated lock class, it's better to refactor it to make it easy to understand and maintain. ### Does this PR introduce any user-facing change? No ### How was this patch tested? Existing tests. Closes #25680 from zsxwing/getStatuses. Authored-by: Shixiong Zhu <zsxwing@gmail.com> Signed-off-by: Shixiong Zhu <zsxwing@gmail.com>

### What changes were proposed in this pull request? Use `KeyLock` added in apache#25612 to simplify `MapOutputTracker.getStatuses`. It also has some improvement after the refactoring: - `InterruptedException` is no longer sallowed. - When a shuffle block is fetched, we don't need to wake up unrelated sleeping threads. ### Why are the changes needed? `MapOutputTracker.getStatuses` is pretty hard to maintain right now because it has a special lock mechanism which we needs to pay attention to whenever updating this method. As we can use `KeyLock` to hide the complexity of locking behind a dedicated lock class, it's better to refactor it to make it easy to understand and maintain. ### Does this PR introduce any user-facing change? No ### How was this patch tested? Existing tests. Closes apache#25680 from zsxwing/getStatuses. Authored-by: Shixiong Zhu <zsxwing@gmail.com> Signed-off-by: Shixiong Zhu <zsxwing@gmail.com>

Replace the global TorrentBroadcast lock with fine grained locks

290a3e5

zsxwing requested a review from jiangxb1987 August 28, 2019 18:00

zsxwing commented Aug 28, 2019

View reviewed changes

srowen reviewed Aug 28, 2019

View reviewed changes

zsxwing added 2 commits August 29, 2019 11:33

add KeyLock so we can reuse this pattern later

2cf1826

update comment

844be5c

zsxwing changed the title ~~[SPARK-3137][Core]Replace the global TorrentBroadcast lock with fine grained locks~~ [SPARK-3137][Core]Replace the global TorrentBroadcast lock with fine grained KeyLock Aug 29, 2019

juliuszsompolski reviewed Aug 29, 2019

View reviewed changes

core/src/main/scala/org/apache/spark/util/KeyLock.scala Outdated Show resolved Hide resolved

dongjoon-hyun added the SPARK CORE label Aug 29, 2019

Ngone51 reviewed Aug 30, 2019

View reviewed changes

test

2e9639c

srowen approved these changes Aug 30, 2019

View reviewed changes

juliuszsompolski approved these changes Sep 3, 2019

View reviewed changes

address nits

9b59ca2

asfgit closed this in 8980093 Sep 3, 2019

zsxwing deleted the SPARK-3137 branch September 3, 2019 21:09

zsxwing mentioned this pull request Sep 4, 2019

[SPARK-28976][CORE]Use KeyLock to simplify MapOutputTracker.getStatuses #25680

Closed

		@@ -56,7 +57,10 @@ private[spark] class BroadcastManager(
		private val nextBroadcastId = new AtomicLong(0)

		private[broadcast] val cachedValues = {

Comments

Conversation

zsxwing commented Aug 28, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

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!

gatorsmile commented Aug 28, 2019

Uh oh!

SparkQA commented Aug 28, 2019

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!

zsxwing Aug 28, 2019 • 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!

zsxwing Aug 29, 2019 • 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!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Aug 29, 2019

Uh oh!

SparkQA commented Aug 29, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing Aug 30, 2019 • 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!

zsxwing commented Aug 30, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 30, 2019

Uh oh!

juliuszsompolski left a comment

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

zsxwing commented Aug 28, 2019 •

edited

Loading

zsxwing Aug 28, 2019 •

edited

Loading

zsxwing Aug 29, 2019 •

edited

Loading

zsxwing Aug 30, 2019 •

edited

Loading