[SPARK-4714][CORE]: Check block have removed or not after got info lock in dropFromMemory

[suyanNone]

[SPARK-4714][CORE]: Check block have removed or not after got info lock in dropFromMemory

After got info lock in one of them: removeBlock/dropOldBlocks/dropFromMemory method in BlockManager, the block that info represented may have already removed.

The Three method have the same logic to got info lock:

   info = blockInfo.get(id)
   if (info != null) {
     info.synchronized {
       do sth
     } 
   }

So it is chanced that when one thread got info.synchronized but it already removed from other threads who got info.synchronized before him.

In removeBlock and dropOldBlocks is idempotent if they not have blockinfo.get(blockId) != null
But in dropFromMemory it may be problematic for it may drop block data(which already removed) into diskstore, this method calls data store operations that should not handle missing blocks.

[AmplabJenkins]

Can one of the admins verify this patch?

[JoshRosen]

Unless I'm missing something, this seems like the opposite of the condition we want here: if blockInfo doesn't contain an entry for blockId, then why would we want to remove that entry?

[suyanNone]

1. ThreadA into RemoveBlock(), and got info for blockId1
2. ThreadB into DropFromMemory(), and got info for blockId1

now Thread A, B all want got info.sychronized

B got, and drop block(blockId) from memory, and this block is use memory only, so it remove from blockinfo. and then release the lock.

then A got, but info is already not in blockinfo.

Did I have miss sth or misunderstand sth? may dropForMemory and removeBlock can't happen at the same time?

[JoshRosen]

I don't think that this extra check is necessary; check out my comment on the main pull request and see if you agree.

Even if we did want to add a check here, I think we want to check for if(blockInfo.get(blockId).nonEmpty), since this branch handles the case where blocks have not been removed already.

[JoshRosen]

Can you elaborate on the problem that this PR addresses? Have you observed an actual error that this fixes?

[suyanNone]

@JoshRosen
ThreadA into RemoveBlock(), and got info for blockId1
ThreadB into DropFromMemory(), and got info for blockId1
now Thread A, B all want got info.sychronized

B got, and drop block(blockId) from memory, and this block is use memory only, so it remove from blockinfo. and then release the lock.

then A got, but info is already not in blockinfo.

Did I have miss sth or misunderstand sth? may dropForMemory and removeBlock can't happen at the same time?

[JoshRosen]

Ah, I think I see your concern: let's say that we a block and there are two threads that are racing to perform operations on it: to use your example, thread A wants to call removeBlock() and thread B wants to call dropFromMemory(). For this code to work correctly, we want it to operate correctly for all possible interleavings of those threads

If we consider the case where all of thread A's steps execute before any of thread B's, then things work okay: thread A will have removed the entry from blockInfo before thread B runs, so B will see that info == null and log a warning that the block has already been removed. The same is true for B before A.

In another execution, though, both threads could find the BlockInfo instance in the blockInfo map but race on acquiring its lock (info.synchronized), so info != null will be true for both threads. I agree that this could be a problem, but it might not be if the operations performed in those threads are idempotent. Let's take a look and see if that's the case:

• removeBlock: all of the operations here are idempotent, so at worst we get a warning if we run this on a block that's removed by another racing thread.
• dropOldBlocks: similarly, this just consists of calls to *Store.remove(), which are idempotent.
• dropFromMemory: this case might actually be problematic, since I think that this method calls data store operations that don't handle missing blocks. I'm going to look at this case in a little more detail, but I think that your fix for this might be a good idea.

[JoshRosen]

Similarly, I don't think that we strictly need a check here since the remove(id) operations are idempotent. It might be nice to log a warning if the block has already been removed, but that might not be necessary since this is a background cleanup call.

[suyanNone]

@JoshRosen
yes, it will not cause any error if there not have a check in removeBlock and dropOldBlocks
I pull this request, because I think it will be more reasonable in semantic.
and I already remove the check in removeBlock and dropOldBlocks

[JoshRosen]

Minor style nit: this needs a space after the if and before the open paren: if (!info....

[JoshRosen]

Jenkins, this is ok to test.

[SparkQA]

Test build #24225 has started for PR 3574 at commit 55fa4ba.

• This patch merges cleanly.

[SparkQA]

Test build #24225 has finished for PR 3574 at commit 55fa4ba.

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

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24225/
Test PASSed.

[JoshRosen]

This comment actually refers to the !info.waitForReady() case, so I'd like to either move the comment or swap the order of these checks so that we check for blockInfo.get(blockId).isEmpty in the else if clause instead.

[JoshRosen]

Left one minor code organization comment; aside from that, this looks good to me and should be ready to merge after you fix that up (I can do it if you don't have time, though; just let me know).

There are a couple of edits that I'd like to make to the commit title / description before merging this, but I can do it myself on merge.

Thanks for the careful analysis and for catching this issue!

[suyanNone]

@JoshRosen I refine the code according your comments.
If still have problem, it OK for you to fix up including the title and comments, and thanks for you to check my code and give suggestion.

[SparkQA]

Test build #24234 has started for PR 3574 at commit edb989d.

• This patch merges cleanly.

[SparkQA]

Test build #24234 has finished for PR 3574 at commit edb989d.

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

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24234/
Test PASSed.

[JoshRosen]

LGTM. I like how this ended up evolving into a very small, focused fix.

I'm going to fix up the title / commit message and merge this into master. Even though I don't think we've observed failures from this in the wild, I'm in favor of backporting this into the maintenance branches, too, since it probably won't cause any harm to add more defensive checks here. I'll merge this into branch-1.0 and branch-1.1 and hold off on branch-1.2 for now (I'll add a backport-needed JIRA label and pull this in after 1.2.0 is released).

Thanks!

[JoshRosen]

I've merged this into branch-1.2.