[jira] [Commented] (SPARK-13122) Race condition in MemoryStore.unrollSafely() causes memory leak

["Apache Spark (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-13122?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15127381#comment-15127381 ] 

Apache Spark commented on SPARK-13122:
--------------------------------------

User 'budde' has created a pull request for this issue:
https://github.com/apache/spark/pull/11012

> Race condition in MemoryStore.unrollSafely() causes memory leak
> ---------------------------------------------------------------
>
>                 Key: SPARK-13122
>                 URL: https://issues.apache.org/jira/browse/SPARK-13122
>             Project: Spark
>          Issue Type: Bug
>          Components: Spark Core, Streaming
>    Affects Versions: 1.6.0
>            Reporter: Adam Budde
>
> The [unrollSafely()|https://github.com/apache/spark/blob/v1.6.0/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala#L249] method in MemoryStore will progressively unroll the contents of a block iterator into memory. It works by reserving an initial chunk of unroll memory and periodically checking if more memory must be reserved as it unrolls the iterator. The memory reserved for performing the unroll is considered "pending" memory and is tracked on a per-task attempt ID bases in a map object named pendingUnrollMemoryMap. When the unrolled block is committed to storage memory in the [tryToPut()|https://github.com/apache/spark/blob/v1.6.0/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala#L362] method, a method named [releasePendingUnrollMemoryForThisTask()|https://github.com/apache/spark/blob/v1.6.0/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala#L521] is invoked and this pending memory is released. tryToPut() then proceeds to allocate the storage memory required for the block.
> The unrollSafely() method computes the amount of pending memory used for the unroll operation by saving the amount of unroll memory reserved for the particular task attempt ID at the start of the method in a variable named previousMemoryReserved and subtracting this value from the unroll memory dedicated to the task at the end of the method. This value is stored as the variable amountToTransferToPending. This amount is then subtracted from the per-task unrollMemoryMap and added to pendingUnrollMemoryMap.
> The amount of unroll memory consumed for the task is obtained from unrollMemoryMap via the currentUnrollMemoryForThisTask method. In order for the semantics of unrollSafely() to work, the value of unrollMemoryMap for the task returned by [currentTaskAttemptId()|https://github.com/apache/spark/blob/v1.6.0/core/src/main/scala/org/apache/spark/storage/MemoryStore.scala#L475] must not be mutated between the computation of previousMemoryReserved and amountToTransferToPending. However, since there is no synchronization in place to ensure that computing both variables and updating the memory maps happens atomically, a race condition can occur when multiple threads for which currentTaskAttemptId() returns the same value are both trying to store blocks. This can lead to a negative value being computed for amountToTransferToPending, corrupting the unrollMemoryMap and pendingUnrollMemoryMap memory maps which in turn can lead to the memory manager leaking unroll memory.
> For example, lets consider how the state of the unrollMemoryMap and pendingUnrollMemoryMap variables might be affected if two threads returning the same value for currentTaskAttemptId() both execute unrollSafely() concurrently:
> ||Thread 1||Thread 2||unrollMemoryMap||pendingUnrollMemoryMap||
> |Enter unrollSafely()|-|0|0|
> |perviousMemoryReserved = 0|-|0|0|
> |(perform unroll)|-|2097152 (2 MiB)|0|
> |-|Enter unrollSafely()|2097152 (2 MiB)|0| 
> |-|perviousMemoryReserved = 2097152|2097152 (2 MiB)|0|
> |-|(performUnroll)|3145728 (3 MiB)|0|
> |Enter finally { }|-|3145728 (3 MiB)|0| 
> |amtToTransfer =  3145728|-|3145728 (3 MiB)|0|
> |Update memory maps|-|0|3145728 (3 MiB)|
> |Return|Enter finally { }|0|3145728 (3 MiB)|
> |-|amtToTrasnfer = -3145728|0|3145728 (3 MiB)|
> |-|Update memory maps|-3145728 (3 MiB)|0|
> In this example, we end up leaking 3 MiB of unroll memory since both Thread 1 and Thread 2 think that the task has no pending unroll memory allocated to it. The negative value stored in unrollMemoryMap will also propagate to future invocations of unrollSafely().
> In our particular case, this behavior manifests since the currentTaskAttemptId() method is returning -1 for each Spark receiver task. This in and of itself could be a bug and is something I'm going to look into. We noticed that blocks would start to spill over to disk when more than enough storage memory was available, so we inserted log statements into MemoryManager's acquireUnrollMemory() and releaseUnrollMemory() in order to collect the number of unroll bytes acquired and released. When we plot the output, it is apparent that unroll memory is being leaked:
> !https://raw.githubusercontent.com/budde/spark_debug/master/plots/UnifiedMemoryManager/leaked_unroll_bytes.png!
> Running difference between acquire/release bytes:
> !https://raw.githubusercontent.com/budde/spark_debug/master/plots/UnifiedMemoryManager/leaked_unroll_bytes_running_diff.png!
> However, if we change the implementation of unrollSafely() so that the entire method is within a synchronized block, we no longer see this leak:
> !https://raw.githubusercontent.com/budde/spark_debug/master/plots/UnifiedMemoryManager/synchronized_unroll_bytes.png!
> Running difference between acquire/release bytes:
> !https://raw.githubusercontent.com/budde/spark_debug/master/plots/UnifiedMemoryManager/synchronized_unroll_bytes_running_diff.png!



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org