[SPARK-22083][CORE] Release locks in MemoryStore.evictBlocksToFreeSpace #19311

squito · 2017-09-21T19:16:24Z

What changes were proposed in this pull request?

MemoryStore.evictBlocksToFreeSpace acquires write locks for all the
blocks it intends to evict up front. If there is a failure to evict
blocks (eg., some failure dropping a block to disk), then we have to
release the lock. Otherwise the lock is never released and an executor
trying to get the lock will wait forever.

How was this patch tested?

Added unit test.

SparkQA · 2017-09-21T22:18:14Z

Test build #82046 has finished for PR 19311 at commit 1ff270a.

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

vanzin · 2017-09-21T22:57:29Z

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

+          if (exceptionWasThrown) {
+            selectedBlocks.foreach { id =>
+              // some of the blocks may have already been unlocked, or completely removed
+              blockInfoManager.get(id).foreach { info =>


This feels racy. Let's say you're dropping 10 blocks here.

You try to drop the first one, but newEffectiveStorageLevel.isValid is true, so you just unlock the block. Then you get to this code some time later, but some other thread has locked that first block. Aren't you going to drop that lock which you don't really own?

I think you'd need to keep track of which blocks have successfully been processed by dropBlock instead of doing this.

good point, thanks, I've handled this now

MemoryStore.evictBlocksToFreeSpace acquires write locks for all the blocks it intends to evict up front. If there is a failure to evict blocks (eg., some failure dropping a block to disk), then we have to release the lock. Otherwise the lock is never released and an executor trying to get the lock will wait forever.

SparkQA · 2017-09-22T22:33:51Z

Test build #82091 has finished for PR 19311 at commit 7ce162d.

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

vanzin · 2017-09-22T22:57:16Z

core/src/test/scala/org/apache/spark/storage/MemoryStoreSuite.scala

+      memManager.setMemoryStore(memoryStore)
+
+      // Put in some small blocks to fill up the memory store
+      val initialBlocks = (1 to 10).map { id =>


The logic looks fine, but I kinda dislike the magic number ("10") being used everywhere.

A constant would make this much better (val initialBlocks = 10).

To piggy back on @vanzin's comment, sizePerBlock also please (so that 100 goes away) ? Thx

mridulm

Thanks @squito for the change, this looks like a nasty bugfix !

mridulm · 2017-09-24T09:24:38Z

core/src/test/scala/org/apache/spark/storage/MemoryStoreSuite.scala

+    // blocks getting evicted.  We'll make the eviction throw an exception, and make sure that
+    // all locks are released.
+    val ct = implicitly[ClassTag[Array[Byte]]]
+    def testFailureOnNthDrop(failAfterDroppingNBlocks: Int, readLockAfterDrop: Boolean): Unit = {


Nit: failAfterDroppingNBlocks -> numValidBlocks, readLockAfterDrop -> validBlock ?

I don't think validBlock captures the intent here -- I don't see anything valid or invalid about it either way. The part of the behavior which changes is whether or not another thread grabs a reader lock on the thread after it gets dropped to disk.

(To go along with that, we drop the block to disk, rather than just evicting it completely, as otherwise there is nothing to grab a lock of. I could always drop the block to disk, instead of having that depend on this, it just seemed like another useful thing to check, whether the number of blocks was successfully updated in blockInfoManager, when the block was dropped completely.)

mridulm · 2017-09-24T09:28:33Z

core/src/test/scala/org/apache/spark/storage/MemoryStoreSuite.scala

+      memManager.setMemoryStore(memoryStore)
+
+      // Put in some small blocks to fill up the memory store
+      val initialBlocks = (1 to 10).map { id =>


To piggy back on @vanzin's comment, sizePerBlock also please (so that 100 goes away) ? Thx

mridulm · 2017-09-24T09:34:35Z

core/src/test/scala/org/apache/spark/storage/MemoryStoreSuite.scala

+      val blocksStillInMemory = blockInfoManager.entries.filter { case (id, info) =>
+        assert(info.writerTask === BlockInfo.NO_WRITER, id)
+        // in this test, all the blocks in memory have no reader, but everything dropped to disk
+        // had another thread read the block.  We shouldn't lose the other thread's reader lock.


I am curious about this part of the test.
Why are we checking for this (and so, why afterDropAction in the test case) ? Even without this, the change and testcase looks fine to me.

Am I missing something ?

Are we testing for write lock release resulting in read unlock for other task's as well ?
Or something else ?

(To nitpick, the write lock release and read lock acquire can be interspersed by another read or write acquire (ofcourse not in this test) )

In an earlier version of this, I was always unconditionally releasing all locks that were held by anything in the finally. I've changed it to only release locks that this thread holds, and this part of the test is to verify that. We simulate another thread grabbing a lock on the blocks which get successfully dropped (just a read lock in this case, though doesn't really matter). The test makes sure that even though we drop some of the remaining locks owned by this thread in the finally, the other thread still keeps its read lock.

Yes, there are many other possible interleavings of locks possible with other threads, but thats not the point of this test case. Its to make sure that the finally block releases only the correct set of locks.

Thanks for clarifying @squito ... I was assuming the test was for something along those line, but good to know I did not misunderstand !
This patch looks great.

vanzin · 2017-09-25T18:08:38Z

LGTM. Merging to master / 2.2 / 2.1.

vanzin · 2017-09-25T18:08:50Z

(After tests are done, I mean.)

SparkQA · 2017-09-25T19:01:35Z

Test build #82150 has finished for PR 19311 at commit dcb7f73.

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

## What changes were proposed in this pull request? MemoryStore.evictBlocksToFreeSpace acquires write locks for all the blocks it intends to evict up front. If there is a failure to evict blocks (eg., some failure dropping a block to disk), then we have to release the lock. Otherwise the lock is never released and an executor trying to get the lock will wait forever. ## How was this patch tested? Added unit test. Author: Imran Rashid <[email protected]> Closes #19311 from squito/SPARK-22083. (cherry picked from commit 2c5b9b1) Signed-off-by: Marcelo Vanzin <[email protected]>

## What changes were proposed in this pull request? MemoryStore.evictBlocksToFreeSpace acquires write locks for all the blocks it intends to evict up front. If there is a failure to evict blocks (eg., some failure dropping a block to disk), then we have to release the lock. Otherwise the lock is never released and an executor trying to get the lock will wait forever. ## How was this patch tested? Added unit test. Author: Imran Rashid <[email protected]> Closes apache#19311 from squito/SPARK-22083. (cherry picked from commit 2c5b9b1) Signed-off-by: Marcelo Vanzin <[email protected]>

MemoryStore.evictBlocksToFreeSpace acquires write locks for all the blocks it intends to evict up front. If there is a failure to evict blocks (eg., some failure dropping a block to disk), then we have to release the lock. Otherwise the lock is never released and an executor trying to get the lock will wait forever. Added unit test. Author: Imran Rashid <[email protected]> Closes apache#19311 from squito/SPARK-22083.

vanzin reviewed Sep 21, 2017

View reviewed changes

squito added 2 commits September 21, 2017 18:54

handle race if another thread grabs lock after block is dropped

dfd1258

squito force-pushed the SPARK-22083 branch from 1ff270a to dfd1258 Compare September 22, 2017 18:03

cleanup

7ce162d

vanzin reviewed Sep 22, 2017

View reviewed changes

mridulm reviewed Sep 24, 2017

View reviewed changes

review feedback

dcb7f73

asfgit closed this in 2c5b9b1 Sep 25, 2017

squito deleted the SPARK-22083 branch September 25, 2017 20:57

[SPARK-22083][CORE] Release locks in MemoryStore.evictBlocksToFreeSpace #19311

[SPARK-22083][CORE] Release locks in MemoryStore.evictBlocksToFreeSpace #19311

Uh oh!

Conversation

squito commented Sep 21, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Sep 21, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 22, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mridulm 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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mridulm Sep 24, 2017 • 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!

vanzin commented Sep 25, 2017

Uh oh!

vanzin commented Sep 25, 2017

Uh oh!

SparkQA commented Sep 25, 2017

Uh oh!

Uh oh!

mridulm Sep 24, 2017 •

edited

Loading