You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by JoshRosen <gi...@git.apache.org> on 2016/03/09 22:42:47 UTC
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
GitHub user JoshRosen opened a pull request:
https://github.com/apache/spark/pull/11613
[SPARK-10907][SPARK-6157][WIP] Remove pendingUnrollMemory from MemoryStore
This patch refactors the MemoryStore to remove the concept of `pendingUnrollMemory`. It also fixes fixes SPARK-6157: "Unrolling with MEMORY_AND_DISK should always release memory".
Key changes:
- Inline `MemoryStore.tryToPut` at its three call sites in the `MemoryStore`.
- Inline `Memory.unrollSafely` at its only call site (in `MemoryStore.putIterator`).
- Inline `MemoryManager.acquireStorageMemory` at its call sites.
- Simplify the code as a result of this inlining (some parameters have fixed values after inlining, so lots of branches can be removed).
- Remove the `pendingUnrollMemory` map by returning the amount of unrollMemory allocated when returning an iterator after a failed `putIterator` call.
- Change `putIterator` to return an instance of `PartiallyUnrolledIterator`, a special iterator subclass which will automatically free the unroll memory of its partially-unrolled elements when the iterator is consumed. To handle cases where the iterator is not consumed (e.g. when a MEMORY_ONLY put fails), `PartiallyUnrolledIterator` exposes a `close()` method which may be called to discard the unrolled values and free their memory.
This patch is marked WIP because it's currently rebased on top of #11534 and needs additional doc, comment, and test updates before it will be ready to merge. Here's a link to the actual diff: https://github.com/apache/spark/compare/66796b5...JoshRosen:cleanup-unroll-memory
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/JoshRosen/spark cleanup-unroll-memory
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11613.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #11613
----
commit 3d51b00b5724aa94bf5ed7967f881de5988c5b52
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:07:59Z
Delete the BlockStore interface.
commit 14d003ffe06ac763e128a1b3fbee8fb22fb2cc17
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:13:14Z
Remove unused DiskStore.getBytes() overloads.
commit 1a764c05a58c5ef299508569ac4096b5fa0468d9
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:14:36Z
DiskStore.getBytes() never returns None, so it shouldn't return an Option.
commit 9814ba6b479850ca9c198af1556e16017a769280
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:31:02Z
Simplify DiskStore.putIterator's return type.
commit 14d5652107cdfc8c32e94231cfc2919339d2a2a5
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:33:03Z
Remove MemoryStore.putIterator() overload.
commit 5c294ace7dcdabf4e325367866eac18aa6e16efb
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T01:39:06Z
DiskStore put() methods don't need to take a StorageLevel.
commit c8d0e695f473590bed1cc318f98c8445834c00a6
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:05:23Z
Factor common error-handling code in DiskStore.put*() into helper function.
commit 46b3877a462a19b9284990e05830f9782629235d
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:09:45Z
Remove DiskStore.putIterator().
commit 27cee47a6228e470ab54891fd1d2dec676a49938
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:21:57Z
Remove DiskStore's dependency on BlockManager.
commit 1a50c8115f31a1acd3b0a4dd5e1b3d28e81d05f5
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:25:31Z
Minor simplifications in DiskStoreSuite test.
commit f3b60052c42215d1440cc5660bdea443bed5598e
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:29:18Z
Remove outdated comment in DiskBlockManager.
commit 2d86e290f77c15aa8ba6e6e46f6c39987ee351a5
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T02:39:25Z
Remove DiskStore's dependency on BlockManager.
commit 9e3ae78f62310aa291833842113c9832ac520bfa
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T03:28:49Z
Shorten period of holding memoryManager lock.
commit d8487d4e8ee5bb5b64d60f1158fc7420ac6a2a54
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T03:48:35Z
MemoryStore.put() no longer handles dropping to disk.
This is now handled by the caller.
commit 10a667d62642ab478ab00b5e0267be93d6b01417
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T04:06:36Z
MemoryStore.putBytes() shouldn't perform deserialization.
commit 87e775d585d2db7c91af9c2587df2eb395040248
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T19:31:46Z
MemoryStore should take its own conf, not obtain it from BlockManager.
commit 2923850c27931cd8efb49449b19438e82763c39e
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T19:53:47Z
Move MemoryManager into new o.a.s.storage.memory package
commit 40f4e436e2d99eebc41b5f8703936f8497b9443c
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-04T22:33:28Z
getBytes() and getValues() no longer implicitly serialize / deserialize.
commit 495ad976699ab05a8b452c39c65ebcc13c1718db
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-05T00:00:00Z
Split doGetLocal() and getLocal() into smaller, simpler methods.
commit 032e3a3b62e70b653a97bb2353c85087f9e4f843
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-05T19:05:50Z
Fix scalastyle violations.
commit 988f00393676eabfc11e665f20f9ce26388e4c11
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-05T21:02:38Z
Fix leaked lock in getOrElseUpdate() when block already exists.
commit 31a500834bab30d3a162885bfc88b08d2c7ffb0f
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-08T19:13:44Z
Document lock requirements of doGetLocalBytes
commit ca5a3f30fdf74694fa9bf5e1352133df3051257e
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-08T19:15:26Z
Add clarifying comment to doGetLocalBytes()
commit 14857b3e53d1b496b7ff07099c53ab2d775f950a
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-08T19:17:51Z
Remove unnecessary putBlockInfo.synchronized call
commit 92c5125f2f4736335971e779fc39e9fa74f8c310
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-08T23:34:05Z
Remove effectiveStorageLevel from put() APIs.
commit 7a08a179f8951abbdcb7e70f6bfb53821fbc7352
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-08T23:59:34Z
Split doPut() into doPutBytes() and doPutIterator().
commit a16276e1cfb92fba88f2963b0d918aa14f2500a4
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-09T00:01:46Z
Remove unreachable level == StorageLevel.NONE case.
This is unreachable because we check whether level.isValid earlier in the same method.
commit 82886e03f60e2b4b0b97b0c6ae640ca10dede145
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-09T00:03:43Z
Fix statement without side-effects.
commit 66796b5bf89ddedc9644b9f5692441293c0c0aaa
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-09T00:15:11Z
Minor comment reword.
commit dbd164e5d7d147b39993b60390adf0a6b84c0ac8
Author: Josh Rosen <jo...@databricks.com>
Date: 2016-03-08T19:45:50Z
Make unrollSafely private.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195194620
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195647101
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52972/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898378
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
can we call `releaseUnrollMemoryUsedByThisTask` instead?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-194575378
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195591625
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52942/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55900156
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
it might be better to add an assert here:
```
val taskAttemptId = currentTaskAttemptId()
assert(unrollMemoryMap.contains(taskAttemptId))
assert(unrollMemoryMap(taskAttemptId) >= unrollMemoryUsedByThisBlock)
unrollMemoryMap(taskAttemptId) -= unrollMemoryUsedByThisBlock
```
not super necessary but it seems safer in case we screwed up the accounting elsewhere
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195571052
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55899254
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
actually, why do we release anything here? Before entering this if case we're guaranteed to hold `unroll memory == block size`. We still need to keep this storage memory because we now store the array in `entries`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55907029
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
Actually, it turns out that `unrollMemoryMap` is `@GuardedBy` `memoryManager` anyways, so we might as well synchronize and release then re-acquire to transfer.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196455250
**[Test build #53084 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53084/consoleFull)** for PR 11613 at commit [`5b3810c`](https://github.com/apache/spark/commit/5b3810cc220522098f5323c26a5e8e74594d2bdf).
* This patch **fails MiMa tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195644743
**[Test build #52971 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52971/consoleFull)** for PR 11613 at commit [`bc5c2b9`](https://github.com/apache/spark/commit/bc5c2b93edbe7f6a8f362a335fe1933c43634d98).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195194623
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52889/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55900567
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
--- End diff --
true, but it's a pretty important detail
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898706
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
--- End diff --
this should be in the java doc.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-194523821
**[Test build #52766 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52766/consoleFull)** for PR 11613 at commit [`9e11694`](https://github.com/apache/spark/commit/9e11694f1e14e0211806a48a5367fa4f83994a29).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-194575386
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52766/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55897722
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -520,3 +447,42 @@ private[spark] class MemoryStore(
logMemoryUsage()
}
}
+
+/**
+ * The result of a failed [[MemoryStore.putIterator()]] call.
+ *
+ * @param memoryStore the memoryStore, used for freeing memory.
+ * @param unrollMemory the amount of unroll memory used by the values in `unrolled`.
+ * @param unrolled an iterator for the partially-unrolled values.
+ * @param rest the rest of the original iterator passed to [[MemoryStore.putIterator()]].
+ */
+private[storage] class PartiallyUnrolledIterator(
+ memoryStore: MemoryStore,
+ unrollMemory: Long,
+ unrolled: Iterator[Any],
+ rest: Iterator[Any])
+ extends Iterator[Any] {
+
+ private[this] var unrolledIteratorIsCompleted: Boolean = false
+ private[this] var iter: Iterator[Any] = {
+ val completionIterator = CompletionIterator[Any, Iterator[Any]](unrolled, {
+ unrolledIteratorIsCompleted = true
+ memoryStore.releaseUnrollMemoryForThisTask(unrollMemory)
+ })
+ completionIterator ++ rest
--- End diff --
Yep, that's right.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898084
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
--- End diff --
Sure, can do. I didn't do this originally because I thought that `acquireStorageMemory` would complain about acquiring 0 bytes, but it looks like that's actually fine.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195582619
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195623515
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55906595
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
--- End diff --
Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195590566
@JoshRosen looks great. I did a close review and could not find anything serious. My suggestions are mainly to add more comments.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195619862
**[Test build #52972 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52972/consoleFull)** for PR 11613 at commit [`5b3810c`](https://github.com/apache/spark/commit/5b3810cc220522098f5323c26a5e8e74594d2bdf).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195623518
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52956/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196455327
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53084/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195517234
**[Test build #52935 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52935/consoleFull)** for PR 11613 at commit [`858aea2`](https://github.com/apache/spark/commit/858aea285a12b93c4a31c5ec3c44b42f81ea9be6).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196484795
LGTM once tests pass
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55900453
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -49,14 +49,6 @@ private[spark] class MemoryStore(
// A mapping from taskAttemptId to amount of memory used for unrolling a block (in bytes)
// All accesses of this map are assumed to have manually synchronized on `memoryManager`
private val unrollMemoryMap = mutable.HashMap[Long, Long]()
- // Same as `unrollMemoryMap`, but for pending unroll memory as defined below.
- // Pending unroll memory refers to the intermediate memory occupied by a task
- // after the unroll but before the actual putting of the block in the cache.
- // This chunk of memory is expected to be released *as soon as* we finish
- // caching the corresponding block as opposed to until after the task finishes.
- // This is only used if a block is successfully unrolled in its entirety in
- // memory (SPARK-4777).
- private val pendingUnrollMemoryMap = mutable.HashMap[Long, Long]()
--- End diff --
yasss!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898978
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
--- End diff --
For this location, you're right: we should call `releaseUnrollMemoryUsedByThisTask` since we _do_ want to update `releaseUnrollMemory`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195193991
**[Test build #52889 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52889/consoleFull)** for PR 11613 at commit [`5134139`](https://github.com/apache/spark/commit/51341398f4598f255cf8e0904a95b60b65e34310).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55899657
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
never mind, I get it now. No matter how much unroll memory we release up there in L197, we still want to update the bookkeeping in `unrollMemoryMap`. This is essentially "transferring" unroll memory to storage memory for storing the block. I think that's sufficiently subtle to warrant a comment to explain that all we're doing here is transferring memory.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55900897
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -520,3 +447,42 @@ private[spark] class MemoryStore(
logMemoryUsage()
}
}
+
+/**
+ * The result of a failed [[MemoryStore.putIterator()]] call.
+ *
+ * @param memoryStore the memoryStore, used for freeing memory.
+ * @param unrollMemory the amount of unroll memory used by the values in `unrolled`.
+ * @param unrolled an iterator for the partially-unrolled values.
+ * @param rest the rest of the original iterator passed to [[MemoryStore.putIterator()]].
+ */
+private[storage] class PartiallyUnrolledIterator(
+ memoryStore: MemoryStore,
+ unrollMemory: Long,
+ unrolled: Iterator[Any],
+ rest: Iterator[Any])
+ extends Iterator[Any] {
+
+ private[this] var unrolledIteratorIsCompleted: Boolean = false
--- End diff --
`unrolledIteratorIsConsumed`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55906570
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -793,7 +798,12 @@ private[spark] class BlockManager(
// We will drop it to disk later if the memory store can't hold it.
val putSucceeded = if (level.deserialized) {
val values = dataDeserialize(blockId, bytes.duplicate())
- memoryStore.putIterator(blockId, values, level).isRight
+ memoryStore.putIterator(blockId, values, level) match {
+ case Right(_) => true
+ case Left(iter) =>
+ iter.close()
--- End diff --
Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196525119
**[Test build #53090 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53090/consoleFull)** for PR 11613 at commit [`319fb55`](https://github.com/apache/spark/commit/319fb55b5f9b2ecf3cc58ce3ad812e365943bf54).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-194574850
**[Test build #52766 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52766/consoleFull)** for PR 11613 at commit [`9e11694`](https://github.com/apache/spark/commit/9e11694f1e14e0211806a48a5367fa4f83994a29).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898874
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
--- End diff --
Okay. I wasn't sure whether putting it there would have elicited the opposite comment because it's describing more of an implementation detail than the core of what this method does.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196465836
**[Test build #53090 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53090/consoleFull)** for PR 11613 at commit [`319fb55`](https://github.com/apache/spark/commit/319fb55b5f9b2ecf3cc58ce3ad812e365943bf54).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195536492
**[Test build #52942 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52942/consoleFull)** for PR 11613 at commit [`dbca8cf`](https://github.com/apache/spark/commit/dbca8cf7576dc889349a70fb3c3b29c84f5a4c7c).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196442324
Jenkins, retest this please.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195534728
@andrewor14 @nongli, this should now be ready for review.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196445488
**[Test build #53084 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/53084/consoleFull)** for PR 11613 at commit [`5b3810c`](https://github.com/apache/spark/commit/5b3810cc220522098f5323c26a5e8e74594d2bdf).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195581011
**[Test build #52941 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52941/consoleFull)** for PR 11613 at commit [`ebb25f9`](https://github.com/apache/spark/commit/ebb25f9d1e82f62dd1b4194e0945af49feb00d64).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196525513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/53090/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195644838
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52971/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196525511
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195524921
**[Test build #52941 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52941/consoleFull)** for PR 11613 at commit [`ebb25f9`](https://github.com/apache/spark/commit/ebb25f9d1e82f62dd1b4194e0945af49feb00d64).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11613
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195582626
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52941/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898638
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
--- End diff --
same here, it's better if we have a single place (i.e. `releaseUnrollMemoryUsedByThisTask`) instead of multiple places where we release unroll memory. Also this currently doesn't update `unrollMemoryMap`, which is a problem.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195571056
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/52935/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55784376
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -443,7 +443,7 @@ private[spark] class BlockManager(
// https://issues.apache.org/jira/browse/SPARK-6076
// If the file size is bigger than the free memory, OOM will happen. So if we
// cannot put it into MemoryStore, copyForMemory should not be created. That's why
- // this action is put into a `() => ByteBuffer` and created lazily.
+ // this action is put into a `() => ByteBuffer` and created sazily.
--- End diff --
whoops, will fix
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55897125
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
--- End diff --
containing
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196527457
Merging into master.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157][WIP] Remove pendingU...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195156377
**[Test build #52889 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52889/consoleFull)** for PR 11613 at commit [`5134139`](https://github.com/apache/spark/commit/51341398f4598f255cf8e0904a95b60b65e34310).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55908919
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
I've pushed an update which fixes a but that I found in the old logic. I'll walk through offline if you'd like, since it's kinda subtle.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195647098
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55900804
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
--- End diff --
oh actually, I think it does complain about that... :(
```
override def acquireStorageMemory(blockId: BlockId, numBytes: Long): Boolean = synchronized {
...
assert(numBytes >= 0)
...
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55896680
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -793,7 +798,12 @@ private[spark] class BlockManager(
// We will drop it to disk later if the memory store can't hold it.
val putSucceeded = if (level.deserialized) {
val values = dataDeserialize(blockId, bytes.duplicate())
- memoryStore.putIterator(blockId, values, level).isRight
+ memoryStore.putIterator(blockId, values, level) match {
+ case Right(_) => true
+ case Left(iter) =>
+ iter.close()
--- End diff --
```
// If putting deserialized values in memory failed, we will put the bytes directly to disk,
// so we don't need this iterator and we can close it to free resources earlier.
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55906566
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -717,8 +717,13 @@ private[spark] class BlockManager(
level: StorageLevel,
tellMaster: Boolean = true): Boolean = {
require(values != null, "Values is null")
- // If doPut() didn't hand work back to us, then block already existed or was successfully stored
- doPutIterator(blockId, () => values, level, tellMaster).isEmpty
+ doPutIterator(blockId, () => values, level, tellMaster) match {
+ case None =>
+ true
+ case Some(iter) =>
+ iter.close()
--- End diff --
Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55896316
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -520,3 +447,42 @@ private[spark] class MemoryStore(
logMemoryUsage()
}
}
+
+/**
+ * The result of a failed [[MemoryStore.putIterator()]] call.
+ *
+ * @param memoryStore the memoryStore, used for freeing memory.
+ * @param unrollMemory the amount of unroll memory used by the values in `unrolled`.
+ * @param unrolled an iterator for the partially-unrolled values.
+ * @param rest the rest of the original iterator passed to [[MemoryStore.putIterator()]].
+ */
+private[storage] class PartiallyUnrolledIterator(
+ memoryStore: MemoryStore,
+ unrollMemory: Long,
+ unrolled: Iterator[Any],
+ rest: Iterator[Any])
+ extends Iterator[Any] {
+
+ private[this] var unrolledIteratorIsCompleted: Boolean = false
+ private[this] var iter: Iterator[Any] = {
+ val completionIterator = CompletionIterator[Any, Iterator[Any]](unrolled, {
+ unrolledIteratorIsCompleted = true
+ memoryStore.releaseUnrollMemoryForThisTask(unrollMemory)
+ })
+ completionIterator ++ rest
--- End diff --
I see, the reason why we distinguish `rest` and `unrolled` here for the case where we unrolled just 10 values and ran out of memory, so we can release the array as soon as we consume those 10 values instead of everything else. Yes?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195623171
**[Test build #52956 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52956/consoleFull)** for PR 11613 at commit [`c5a23a4`](https://github.com/apache/spark/commit/c5a23a4d4aea88eae3af323501ad224db33d295d).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55896526
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala ---
@@ -717,8 +717,13 @@ private[spark] class BlockManager(
level: StorageLevel,
tellMaster: Boolean = true): Boolean = {
require(values != null, "Values is null")
- // If doPut() didn't hand work back to us, then block already existed or was successfully stored
- doPutIterator(blockId, () => values, level, tellMaster).isEmpty
+ doPutIterator(blockId, () => values, level, tellMaster) match {
+ case None =>
+ true
+ case Some(iter) =>
+ iter.close()
--- End diff --
can you add a short comment, something like:
```
// caller doesn't care about the iterator values, so we can close the iterator here to free resources earlier
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195615909
**[Test build #52971 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52971/consoleFull)** for PR 11613 at commit [`bc5c2b9`](https://github.com/apache/spark/commit/bc5c2b93edbe7f6a8f362a335fe1933c43634d98).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55897629
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
--- End diff --
you can probably merge these two cases; easier to read
```
if (unrollMemoryUsedByThisBlock <= size) {
memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
...
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195647035
**[Test build #52972 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52972/consoleFull)** for PR 11613 at commit [`5b3810c`](https://github.com/apache/spark/commit/5b3810cc220522098f5323c26a5e8e74594d2bdf).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195591366
**[Test build #52942 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52942/consoleFull)** for PR 11613 at commit [`dbca8cf`](https://github.com/apache/spark/commit/dbca8cf7576dc889349a70fb3c3b29c84f5a4c7c).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55906584
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
--- End diff --
Fixed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55897743
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -520,3 +447,42 @@ private[spark] class MemoryStore(
logMemoryUsage()
}
}
+
+/**
+ * The result of a failed [[MemoryStore.putIterator()]] call.
+ *
+ * @param memoryStore the memoryStore, used for freeing memory.
+ * @param unrollMemory the amount of unroll memory used by the values in `unrolled`.
+ * @param unrolled an iterator for the partially-unrolled values.
+ * @param rest the rest of the original iterator passed to [[MemoryStore.putIterator()]].
+ */
+private[storage] class PartiallyUnrolledIterator(
+ memoryStore: MemoryStore,
+ unrollMemory: Long,
+ unrolled: Iterator[Any],
+ rest: Iterator[Any])
+ extends Iterator[Any] {
+
+ private[this] var unrolledIteratorIsCompleted: Boolean = false
--- End diff --
`unrolledIteratorIsFullyConsumed`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55897952
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
--- End diff --
This method is getting big. We should add more inline comments throughout, e.g.
```
// Acquire storage memory if necessary to store this block in memory. If this task
// attempt already owns more unroll memory than is necessary to store the block,
// then release the memory that will not be used.
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55906575
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -520,3 +447,42 @@ private[spark] class MemoryStore(
logMemoryUsage()
}
}
+
+/**
+ * The result of a failed [[MemoryStore.putIterator()]] call.
+ *
+ * @param memoryStore the memoryStore, used for freeing memory.
+ * @param unrollMemory the amount of unroll memory used by the values in `unrolled`.
+ * @param unrolled an iterator for the partially-unrolled values.
+ * @param rest the rest of the original iterator passed to [[MemoryStore.putIterator()]].
+ */
+private[storage] class PartiallyUnrolledIterator(
+ memoryStore: MemoryStore,
+ unrollMemory: Long,
+ unrolled: Iterator[Any],
+ rest: Iterator[Any])
+ extends Iterator[Any] {
+
+ private[this] var unrolledIteratorIsCompleted: Boolean = false
--- End diff --
Done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-196455322
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898608
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
+ if (unrollMemoryUsedByThisBlock == size) {
+ true
+ } else if (unrollMemoryUsedByThisBlock < size) {
+ memoryManager.acquireStorageMemory(blockId, size - unrollMemoryUsedByThisBlock)
} else {
- val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
- if (tryToPut(blockId, () => bytes, bytes.limit, deserialized = false)) {
- Right(bytes.limit())
- } else {
- Left(arrayValues.toIterator)
- }
+ memoryManager.releaseUnrollMemory(unrollMemoryUsedByThisBlock - size)
+ true
}
- case Right(iteratorValues) =>
- Left(iteratorValues)
+ }
+ if (acquiredStorageMemory) {
+ // We acquired enough memory for the block, so go ahead and put it
+ memoryManager.synchronized {
+ unrollMemoryMap(currentTaskAttemptId()) -= unrollMemoryUsedByThisBlock
--- End diff --
We don't want to call `memoryManager.releaseUnrollMemory` because conceptually we're trying to transfer the unrollMemory to storageMemory. That's why I didn't call `releaseUnrollMemoryForThisTask`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898848
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
--- End diff --
I would also expand on the existing one to comment how the memory is "transferred" from the memory used for unrolling to store the block, so we never acquire 2X the memory needed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195592066
**[Test build #52956 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52956/consoleFull)** for PR 11613 at commit [`c5a23a4`](https://github.com/apache/spark/commit/c5a23a4d4aea88eae3af323501ad224db33d295d).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195616282
I ended up having to roll back that size estimation improvement because it broke some tests which relied on the old behavior. I'd still like to make that change but would prefer to do it later since it may impact a bunch of unrelated unit tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55895725
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -520,3 +447,42 @@ private[spark] class MemoryStore(
logMemoryUsage()
}
}
+
+/**
+ * The result of a failed [[MemoryStore.putIterator()]] call.
+ *
+ * @param memoryStore the memoryStore, used for freeing memory.
+ * @param unrollMemory the amount of unroll memory used by the values in `unrolled`.
+ * @param unrolled an iterator for the partially-unrolled values.
+ * @param rest the rest of the original iterator passed to [[MemoryStore.putIterator()]].
+ */
+private[storage] class PartiallyUnrolledIterator(
+ memoryStore: MemoryStore,
+ unrollMemory: Long,
+ unrolled: Iterator[Any],
+ rest: Iterator[Any])
+ extends Iterator[Any] {
+
+ private[this] var unrolledIteratorIsCompleted: Boolean = false
--- End diff --
how about `iteratorIsFullyConsumed`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by andrewor14 <gi...@git.apache.org>.
Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/11613#discussion_r55898188
--- Diff: core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala ---
@@ -100,48 +92,136 @@ private[spark] class MemoryStore(
*/
def putBytes(blockId: BlockId, size: Long, _bytes: () => ByteBuffer): Boolean = {
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- // Work on a duplicate - since the original input might be used elsewhere.
- lazy val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
- val putSuccess = tryToPut(blockId, () => bytes, size, deserialized = false)
- if (putSuccess) {
+ if (memoryManager.acquireStorageMemory(blockId, size)) {
+ // We acquired enough memory for the block, so go ahead and put it
+ // Work on a duplicate - since the original input might be used elsewhere.
+ val bytes = _bytes().duplicate().rewind().asInstanceOf[ByteBuffer]
assert(bytes.limit == size)
+ val entry = new MemoryEntry(bytes, size, deserialized = false)
+ entries.synchronized {
+ entries.put(blockId, entry)
+ }
+ logInfo("Block %s stored as bytes in memory (estimated size %s, free %s)".format(
+ blockId, Utils.bytesToString(size), Utils.bytesToString(blocksMemoryUsed)))
+ true
+ } else {
+ false
}
- putSuccess
}
/**
* Attempt to put the given block in memory store.
*
- * @return the estimated size of the stored data if the put() succeeded, or an iterator
- * in case the put() failed (the returned iterator lets callers fall back to the disk
- * store if desired).
+ * @return in case of success, the estimated the estimated size of the stored data. In case of
+ * failure, return an iterator contianing the values of the block. The returned iterator
+ * will be backed by the combination of the partially-unrolled block and the remaining
+ * elements of the original input iterator. The caller must either fully consume this
+ * iterator or call `close()` on it in order to free the storage memory consumed by the
+ * partially-unrolled block.
*/
private[storage] def putIterator(
blockId: BlockId,
values: Iterator[Any],
- level: StorageLevel): Either[Iterator[Any], Long] = {
+ level: StorageLevel): Either[PartiallyUnrolledIterator, Long] = {
+
require(!contains(blockId), s"Block $blockId is already present in the MemoryStore")
- val unrolledValues = unrollSafely(blockId, values)
- unrolledValues match {
- case Left(arrayValues) =>
- // Values are fully unrolled in memory, so store them as an array
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
- if (tryToPut(blockId, () => arrayValues, sizeEstimate, deserialized = true)) {
- Right(sizeEstimate)
- } else {
- Left(arrayValues.toIterator)
+
+ // It's possible that the iterator is too large to materialize and store in memory. To avoid
+ // OOM exceptions, this method will gradually unroll the iterator while periodically checking
+ // whether there is enough free memory.
+
+ // Number of elements unrolled so far
+ var elementsUnrolled = 0
+ // Whether there is still enough memory for us to continue unrolling this block
+ var keepUnrolling = true
+ // Initial per-task memory to request for unrolling blocks (bytes).
+ val initialMemoryThreshold = unrollMemoryThreshold
+ // How often to check whether we need to request more memory
+ val memoryCheckPeriod = 16
+ // Memory currently reserved by this task for this particular unrolling operation
+ var memoryThreshold = initialMemoryThreshold
+ // Memory to request as a multiple of current vector size
+ val memoryGrowthFactor = 1.5
+ // Keep track of unroll memory used by this particular block / putIterator() operation
+ var unrollMemoryUsedByThisBlock = 0L
+ // Underlying vector for unrolling the block
+ var vector = new SizeTrackingVector[Any]
+
+ // Request enough memory to begin unrolling
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold)
+
+ if (!keepUnrolling) {
+ logWarning(s"Failed to reserve initial memory threshold of " +
+ s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ } else {
+ unrollMemoryUsedByThisBlock += initialMemoryThreshold
+ }
+
+ // Unroll this block safely, checking whether we have exceeded our threshold periodically
+ while (values.hasNext && keepUnrolling) {
+ vector += values.next()
+ if (elementsUnrolled % memoryCheckPeriod == 0) {
+ // If our vector's size has exceeded the threshold, request more memory
+ val currentSize = vector.estimateSize()
+ if (currentSize >= memoryThreshold) {
+ val amountToRequest = (currentSize * memoryGrowthFactor - memoryThreshold).toLong
+ keepUnrolling = reserveUnrollMemoryForThisTask(blockId, amountToRequest)
+ if (keepUnrolling) {
+ unrollMemoryUsedByThisBlock += amountToRequest
}
+ // New threshold is currentSize * memoryGrowthFactor
+ memoryThreshold += amountToRequest
+ }
+ }
+ elementsUnrolled += 1
+ }
+
+ if (keepUnrolling) {
+ // We successfully unrolled the entirety of this block
+ val arrayValues = vector.toArray
+ vector = null
+ val entry = if (level.deserialized) {
+ val sizeEstimate = SizeEstimator.estimate(arrayValues.asInstanceOf[AnyRef])
+ new MemoryEntry(arrayValues, sizeEstimate, deserialized = true)
+ } else {
+ val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
+ new MemoryEntry(bytes, bytes.limit, deserialized = false)
+ }
+ val size = entry.size
+ val acquiredStorageMemory = {
--- End diff --
and maybe I would call this `enoughStorageMemory` or something; we're not really acquiring more storage memory here if we already have too much.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195591622
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195570778
**[Test build #52935 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/52935/consoleFull)** for PR 11613 at commit [`858aea2`](https://github.com/apache/spark/commit/858aea285a12b93c4a31c5ec3c44b42f81ea9be6).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-10907][SPARK-6157] Remove pendingUnroll...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11613#issuecomment-195644837
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org