Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10705#discussion_r53877407
--- Diff: core/src/main/scala/org/apache/spark/storage/BlockManager.scala
---
@@ -989,77 +1014,62 @@ private[spark] class BlockManager(
* store reaches its limit and needs to free up space.
*
* If `data` is not put on disk, it won't be created.
+ *
+ * The caller of this method must hold a write lock on the block before
calling this method.
+ *
+ * @return the block's new effective StorageLevel.
*/
def dropFromMemory(
blockId: BlockId,
- data: () => Either[Array[Any], ByteBuffer]): Unit = {
-
+ data: () => Either[Array[Any], ByteBuffer]): StorageLevel = {
logInfo(s"Dropping block $blockId from memory")
- val info = blockInfo.get(blockId)
-
- // If the block has not already been dropped
- if (info != null) {
- info.synchronized {
- // required ? As of now, this will be invoked only for blocks
which are ready
- // But in case this changes in future, adding for consistency sake.
- if (!info.waitForReady()) {
- // If we get here, the block write failed.
- logWarning(s"Block $blockId was marked as failure. Nothing to
drop")
- return
- } else if (blockInfo.asScala.get(blockId).isEmpty) {
- logWarning(s"Block $blockId was already dropped.")
- return
- }
- var blockIsUpdated = false
- val level = info.level
-
- // Drop to disk, if storage level requires
- if (level.useDisk && !diskStore.contains(blockId)) {
- logInfo(s"Writing block $blockId to disk")
- data() match {
- case Left(elements) =>
- diskStore.putArray(blockId, elements, level, returnValues =
false)
- case Right(bytes) =>
- diskStore.putBytes(blockId, bytes, level)
- }
- blockIsUpdated = true
- }
+ val info = blockInfoManager.assertBlockIsLockedForWriting(blockId)
+ var blockIsUpdated = false
+ val level = info.level
+
+ // Drop to disk, if storage level requires
+ if (level.useDisk && !diskStore.contains(blockId)) {
+ logInfo(s"Writing block $blockId to disk")
+ data() match {
+ case Left(elements) =>
+ diskStore.putArray(blockId, elements, level, returnValues =
false)
+ case Right(bytes) =>
+ diskStore.putBytes(blockId, bytes, level)
+ }
+ blockIsUpdated = true
+ }
- // Actually drop from memory store
- val droppedMemorySize =
- if (memoryStore.contains(blockId)) memoryStore.getSize(blockId)
else 0L
- val blockIsRemoved = memoryStore.remove(blockId)
- if (blockIsRemoved) {
- blockIsUpdated = true
- } else {
- logWarning(s"Block $blockId could not be dropped from memory as
it does not exist")
- }
+ // Actually drop from memory store
+ val droppedMemorySize =
+ if (memoryStore.contains(blockId)) memoryStore.getSize(blockId) else
0L
+ val blockIsRemoved = memoryStore.remove(blockId)
+ if (blockIsRemoved) {
+ blockIsUpdated = true
+ } else {
+ logWarning(s"Block $blockId could not be dropped from memory as it
does not exist")
--- End diff --
Is this still possible?
---
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]
