Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/23272#discussion_r240189245
--- Diff:
core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java
---
@@ -667,4 +669,54 @@ public void testPeakMemoryUsed() {
}
}
+ @Test
+ public void avoidDeadlock() throws InterruptedException {
+ memoryManager.limit(PAGE_SIZE_BYTES);
+ MemoryMode mode = useOffHeapMemoryAllocator() ? MemoryMode.OFF_HEAP:
MemoryMode.ON_HEAP;
+ TestMemoryConsumer c1 = new TestMemoryConsumer(taskMemoryManager,
mode);
+ BytesToBytesMap map =
+ new BytesToBytesMap(taskMemoryManager, blockManager,
serializerManager, 1, 0.5, 1024);
+
+ Runnable memoryConsumer = new Runnable() {
+ @Override
+ public void run() {
+ int i = 0;
+ long used = 0;
+ while (i < 10) {
+ c1.use(10000000);
+ used += 10000000;
+ i++;
+ }
+ c1.free(used);
+ }
+ };
+
+ Thread thread = new Thread(memoryConsumer);
+
+ try {
+ int i;
+ for (i = 0; i < 1024; i++) {
+ final long[] arr = new long[]{i};
+ final BytesToBytesMap.Location loc = map.lookup(arr,
Platform.LONG_ARRAY_OFFSET, 8);
+ loc.append(arr, Platform.LONG_ARRAY_OFFSET, 8, arr,
Platform.LONG_ARRAY_OFFSET, 8);
+ }
+
+ // Starts to require memory at another memory consumer.
+ thread.start();
+
+ BytesToBytesMap.MapIterator iter = map.destructiveIterator();
+ for (i = 0; i < 1024; i++) {
+ iter.next();
+ }
+ assertFalse(iter.hasNext());
+ } finally {
+ map.free();
+ thread.join();
--- End diff --
Is this line where the test hangs without the fix?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
