[SPARK-6076][Block Manager] Fix a potential OOM issue when StorageLevel is MEMORY_AND_DISK_SER

[zsxwing]

In

spark/core/src/main/scala/org/apache/spark/storage/BlockManager.scala

Line 538 in dcd1e42

val copyForMemory = ByteBuffer.allocate(bytes.limit)

, when StorageLevel is MEMORY_AND_DISK_SER, it will copy the content from file into memory, then put it into MemoryStore.

              val copyForMemory = ByteBuffer.allocate(bytes.limit)
              copyForMemory.put(bytes)
              memoryStore.putBytes(blockId, copyForMemory, level)
              bytes.rewind()

However, if the file is bigger than the free memory, OOM will happen. A better approach is testing if there is enough memory. If not, copyForMemory should not be created, since this is an optional operation.

[SparkQA]

Test build #28118 has started for PR 4827 at commit 0cc0257.

• This patch merges cleanly.

[SparkQA]

Test build #28118 has finished for PR 4827 at commit 0cc0257.

• 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/28118/
Test PASSed.

[srowen]

CC @andrewor14

[rxin]

CC @andrewor14 again in case it was missed.

[andrewor14]

Sorry for the delay. I was away for a little more than a week and I will look at this later today.

[andrewor14]

could you add a // comment here that links to this JIRA? Otherwise it's not clear why we need to do this lazily.

[andrewor14]

@zsxwing the approach looks reasonable. I'm wondering if there's a more readable way to do the same thing. In particular, currently this uses call-by-name (tryToPut, dropFromMemory...) and chained parameters (your new MemoryStore#putBytes), which we actually discourage in the Databricks style guide: https://github.com/databricks/scala-style-guide. How much of a change will this patch be if we rewrite this patch according to the style guide instead?

[zsxwing]

@andrewor14 is it acceptable that changing call-by-name to () => ByteBuffer?

[zsxwing]

@andrewor14 I updated to use () => T.

[SparkQA]

Test build #28976 has started for PR 4827 at commit 1100a54.

• This patch merges cleanly.

[SparkQA]

Test build #28976 has finished for PR 4827 at commit 1100a54.

• 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/28976/
Test PASSed.

[andrewor14]

it's a little odd for this to be () => bytes everywhere. Can you create an alias that looks something like:

private def tryToPut(blockId: BlockId, value: Any, ...): ResultWithDroppedBlocks = {
  tryToPut(blockId, () => value, ...)
}

such that we only use the lazy version if we have to. Same for dropFromMemory

[andrewor14]

Thanks @zsxwing I left one more code style comment and I think it's ready after that.

[SparkQA]

Test build #29137 has started for PR 4827 at commit 7d25545.

• This patch merges cleanly.

[andrewor14]

Instead of duplicating the javadocs here I would just refer to the other method. I can fix this when I merge don't worry.

[andrewor14]

LGTM. I will merge this once tests pass thanks.

[SparkQA]

Test build #29137 has finished for PR 4827 at commit 7d25545.

• 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/29137/
Test PASSed.

[andrewor14]

Merged this into master.