Skip to content

Commit df67bda

Browse files
author
Andrew Or
committed
Add test
1 parent 632ae14 commit df67bda

File tree

2 files changed

+38
-0
lines changed

2 files changed

+38
-0
lines changed

core/src/test/scala/org/apache/spark/memory/MemoryManagerSuite.scala

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,21 @@ private[memory] trait MemoryManagerSuite extends SparkFunSuite with BeforeAndAft
7878
ms
7979
}
8080

81+
/**
82+
* Make a mocked [[MemoryStore]] whose [[MemoryStore.evictBlocksToFreeSpace]] method is
83+
* stubbed to always throw [[RuntimeException]].
84+
*/
85+
protected def makeBadMemoryStore(mm: MemoryManager): MemoryStore = {
86+
val ms = mock(classOf[MemoryStore], RETURNS_SMART_NULLS)
87+
when(ms.evictBlocksToFreeSpace(any(), anyLong(), any())).thenAnswer(new Answer[Long] {
88+
override def answer(invocation: InvocationOnMock): Long = {
89+
throw new RuntimeException("bad memory store!")
90+
}
91+
})
92+
mm.setMemoryStore(ms)
93+
ms
94+
}
95+
8196
/**
8297
* Simulate the part of [[MemoryStore.evictBlocksToFreeSpace]] that releases storage memory.
8398
*

core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -280,4 +280,27 @@ class UnifiedMemoryManagerSuite extends MemoryManagerSuite with PrivateMethodTes
280280
assert(evictedBlocks.nonEmpty)
281281
}
282282

283+
test("SPARK-15260: atomically resize memory pools") {
284+
val conf = new SparkConf()
285+
.set("spark.memory.fraction", "1")
286+
.set("spark.memory.storageFraction", "0")
287+
.set("spark.testing.memory", "1000")
288+
val mm = UnifiedMemoryManager(conf, numCores = 2)
289+
makeBadMemoryStore(mm)
290+
val memoryMode = MemoryMode.ON_HEAP
291+
// Acquire 1000 then release 600 bytes of storage memory, leaving the
292+
// storage memory pool at 1000 bytes but only 400 bytes of which are used.
293+
assert(mm.acquireStorageMemory(dummyBlock, 1000L, memoryMode))
294+
mm.releaseStorageMemory(600L, memoryMode)
295+
// Before the fix for SPARK-15260, we would first shrink the storage pool by the amount of
296+
// unused storage memory (600 bytes), try to evict blocks, then enlarge the execution pool
297+
// by the same amount. If the eviction threw an exception, then we would shrink one pool
298+
// without enlarging the other, resulting in an assertion failure.
299+
intercept[RuntimeException] {
300+
mm.acquireExecutionMemory(1000L, 0, memoryMode)
301+
}
302+
val assertInvariants = PrivateMethod[Unit]('assertInvariants)
303+
mm.invokePrivate[Unit](assertInvariants())
304+
}
305+
283306
}

0 commit comments

Comments
 (0)