[SPARK-1777] Prevent OOMs from single partitions

[andrewor14]

Problem. When caching, we currently unroll the entire RDD partition before making sure we have enough free memory. This is a common cause for OOMs especially when (1) the BlockManager has little free space left in memory, and (2) the partition is large.

Solution. We maintain a global memory pool of M bytes shared across all threads, similar to the way we currently manage memory for shuffle aggregation. Then, while we unroll each partition, periodically check if there is enough space to continue. If not, drop enough RDD blocks to ensure we have at least M bytes to work with, then try again. If we still don't have enough space to unroll the partition, give up and drop the block to disk directly if applicable.

New configurations.

• spark.storage.unrollFraction - the value of M as a fraction of the storage memory. (default: 0.2)
• spark.storage.safetyFraction - a margin of safety in case size estimation is slightly off. This is the equivalent of the existing spark.shuffle.safetyFraction. (default 0.9)

For more detail, see the design document. Tests pending for performance and memory usage patterns.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15979/

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15981/

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build triggered.

[AmplabJenkins]

Merged build started.

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15987/

[AmplabJenkins]

Merged build finished.

[AmplabJenkins]

Refer to this link for build results: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/15988/

[AmplabJenkins]

Merged build triggered.

[andrewor14]

I have addressed your latest comments and rebased to master. Anything else?

[SparkQA]

QA tests have started for PR 1165. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17245/consoleFull

[SparkQA]

QA results for PR 1165:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds the following public classes (experimental):
case class Sample(size: Long, numUpdates: Long)

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/17245/consoleFull

[mateiz]

Thanks Andrew! The changes look good to me -- I've merged this in.

[andrewor14]

Great. Thanks for reviewing @pwendell @mateiz @mridulm.

[liyezhang556520]

@andrewor14
Hi Andrew, I have two questions here:
1, If we return an array, I think we should not release the memory for this thread here, instead should release the memory immediately after the block has been put into the memory. Because the memory is reserved by this thread for caching the block in memory in future, if we release it here, the released memory might be regarded as free memory by other threads. For example here is a case: there are 2 threads executing this function (unrollSafely) in serial (for some reason not in parallel), the second thread can gets more actualFreeMemory because the first thread has released it's memory for unrolling in future. And when the two threads try to call putArray in parallel to cache the block in memory, they may find that there is no space for them to cache the block at the same time. So will this cause the wrong estimation?

2, if we return an iterator, why shall we release the memory later on when the task finishes? Returning iterator means we ran out of memory and would drop the block to disk when using MEMORY_AND_DISK or just skip caching in memory when using MEMORY_ONLY. So the memory stored in unrollMemoryMap for this thread will no longer be used. why not release it here for other threads to use?

Maybe I have some misunderstanding of the code, can you explain it?

[andrewor14]

@liyezhang556520 Thanks for your detailed comments.

1. Your analysis is correct. If we return an array, we still hold onto the space occupied for unrolling, it's just that we are using this space to additionally store the return values. In retrospect it makes little sense to release memory here.

2. If we return an iterator, we cannot actually release the occupied memory here. Note that the iterator that we return depends on the buffer we used for unrolling, so we can't just release it. Unfortunately I don't think there is an alternative here because the original iterator passed in as an argument is already partially exhausted, so the only way to retrieve the traversed items is through the buffer we temporarily stored them in.

Could you submit a PR for (1)? This hasn't come to bite us yet, but it would be good to have it fixed.

[liyezhang556520]

@andrewor14
Oh I see, we can not release the memory when returning an iterator (the original iterator can not roll back or cloned).

I'm making a patch for SPARK-3000, in which I will resolve (1), and I'll make the PR soon, you can help me to review the code, thanks.

[cloud-fan]

I understand we need release the memory later when the task finishes, but where? I checked CacheManager#putInBlockManager but didn't find it. Sorry if I missed something.

[andrewor14]

This is done in Executor.scala (see TaskRunner)

[cloud-fan]

Hi @andrewor14 , I think about your fix and focus on the timing to release unrollMemory. If we unroll a partition successfully, currently we release unrollMemory immediately(this may be dangerous as we still hold the unrollMemory) and return the vector. If we unroll a part of partition, we return the iterator of vector and the rest partition's iterator. Then we release unrollMemory after the whole task is finished. I think this is a little late as we can release unrollMemory once the vector has been iterated over. Maybe we can create a special iterator for the vector that automatically release the unrollMemory when reach the end. An example:

class SpecialIterator(private var data: Seq[Any], amountToRelease: Long) {
  private var cur = 0
  def hasNext = cur < data.size

  def next = {
    if (hasNext) {
      val re = data(cur)
      cur += 1
      release()
      re
    } else {
      throw new NoSuchElementException
    }
  }

  private def release(): Unit = {
    if (cur == data.size) {
      data = null
      ... //do the release job
    }
  }
}

We can also create this iterator for vector if unroll successfully in CacheManager#putInBlockManager and avoid release unrollMemory immediately in unrollSafely.

[cloud-fan]

why not use super.trim.array? I think array copy is more efficient when there are a lot of elements.