Avoid unnecessary ensureFreeSpace() calls

Author: JoshRosen

core/src/main/scala/org/apache/spark/memory/StorageMemoryPool.scala

@@ -105,14 +105,18 @@ class StorageMemoryPool extends MemoryPool with Logging {
   }
 
   // TODO(josh): comment
-  def shrinkPoolByEvictingBlocks(spaceToEnsure: Long): Long = synchronized {
+  def shrinkPoolToFreeSpace(spaceToEnsure: Long): Long = synchronized {
     val spaceFreedByReleasingUnusedMemory = Math.min(spaceToEnsure, memoryFree)
     decrementPoolSize(spaceFreedByReleasingUnusedMemory)
-    val evictedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]
-    memoryStore.ensureFreeSpace(spaceToEnsure - spaceFreedByReleasingUnusedMemory, evictedBlocks)
-    val spaceFreedByEviction = evictedBlocks.map(_._2.memSize).sum
-    _memoryUsed -= spaceFreedByEviction
-    decrementPoolSize(spaceFreedByEviction)
-    spaceFreedByReleasingUnusedMemory + spaceFreedByEviction
+    if (spaceFreedByReleasingUnusedMemory == spaceToEnsure) {
+      spaceFreedByReleasingUnusedMemory
+    } else {
+      val evictedBlocks = new ArrayBuffer[(BlockId, BlockStatus)]
+      memoryStore.ensureFreeSpace(spaceToEnsure - spaceFreedByReleasingUnusedMemory, evictedBlocks)
+      val spaceFreedByEviction = evictedBlocks.map(_._2.memSize).sum
+      _memoryUsed -= spaceFreedByEviction
+      decrementPoolSize(spaceFreedByEviction)
+      spaceFreedByReleasingUnusedMemory + spaceFreedByEviction
+    }
   }
 }

core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala

@@ -89,7 +89,7 @@ private[spark] class UnifiedMemoryManager private[memory] (
         // If there is not enough free memory AND storage has borrowed some execution memory,
         // then evict as much memory borrowed by storage as needed to grant this request
         if (numBytes > onHeapExecutionMemoryPool.memoryFree && memoryBorrowedByStorage > 0) {
-          val spaceReclaimed = storageMemoryPool.shrinkPoolByEvictingBlocks(
+          val spaceReclaimed = storageMemoryPool.shrinkPoolToFreeSpace(
             math.min(numBytes, memoryBorrowedByStorage))
           onHeapExecutionMemoryPool.incrementPoolSize(spaceReclaimed)
         }