Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/10705#discussion_r54177134
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
---
@@ -452,7 +440,10 @@ private[spark] class BlockManager(
if (level.useMemory) {
logDebug(s"Getting block $blockId from memory")
val result = if (asBlockResult) {
- memoryStore.getValues(blockId).map(new BlockResult(_,
DataReadMethod.Memory, info.size))
+ memoryStore.getValues(blockId).map { iter =>
+ val ci = CompletionIterator[Any, Iterator[Any]](iter,
releaseLock(blockId))
+ new BlockResult(ci, DataReadMethod.Memory, info.size)
--- End diff --
Right now there's still a chance that the programmer forgets to wrap the
iter. I would actually push the `CompletionIterator` logic one step further
into `BlockResult` itself, e.g.
```
private[spark] class BlockResult(
iter: Iterator[Any],
blockId: BlockId,
val readMethod: DataReadMethod.Value,
val bytes: Long) {
/**
* Values of this block, to be consumed at most once.
*
* If this block was read locally, then we must have acquired a read lock
on this block.
* If so, release the lock once this iterator is drained. Otherwise, if
this block was read
* remotely from other executors, there is no need to do this because we
didn't acquire
* any locks on the block.
*/
val data: Iterator[Any] = {
if (readMethod != DataReadMethod.Network) {
CompletionIterator[Any, Iterator[Any]](iter, releaseLock(blockId))
} else {
iter
}
}
}
```
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
