Skip to content

[SPARK-7200] Check that memory is not leaked in TaskMemoryManagerSuite's after() #17402

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jsoltren wants to merge 4 commits into from
Closed

[SPARK-7200] Check that memory is not leaked in TaskMemoryManagerSuite's after() #17402

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c44fc87
[SPARK-7200] Add cleanup for memory leak tests to after() in TaskMemo…
jsoltren Mar 10, 2017
8677a99
changes in response to code review from vanzin on 2017-03-28
jsoltren May 16, 2017
b3ffb7d
Merge remote-tracking branch 'upstream/master' into jose-tungsten
jsoltren May 16, 2017
e6ca027
minor cleanup to get rid of unused temp variable
jsoltren May 16, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 19 additions & 10 deletions core/src/test/java/org/apache/spark/memory/TaskMemoryManagerSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package org.apache.spark.memory;

import org.junit.After;
import org.junit.Assert;
import org.junit.Test;

Expand All @@ -25,27 +26,37 @@

public class TaskMemoryManagerSuite {

static TaskMemoryManager manager;
Copy link
Contributor

@vanzin vanzin Mar 28, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does not need to be static. Also should be private.


@After
public void after() {
Assert.assertEquals(0, manager.getMemoryConsumptionForThisTask());
Copy link
Contributor

@vanzin vanzin Mar 28, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add Assert.assertNotNull(manager).

Also, probably minor, but if a test fails, these checks will probably fail in weird ways and might mask the original error... I'd try to insert a failure in a test and seeing what happens.

Copy link
Author

@jsoltren jsoltren May 16, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@squito had the same concern. I purposefully leaked some memory at the end of leakedPageMemoryIsDetected and got this very helpful trace:

[info] Test org.apache.spark.memory.TaskMemoryManagerSuite.leakedPageMemoryIsDetected started
[error] Test org.apache.spark.memory.TaskMemoryManagerSuite.leakedPageMemoryIsDetected failed: java.lang.AssertionError: expected:<0> but was:<4096>, took 0.007 sec
[error] at org.apache.spark.memory.TaskMemoryManagerSuite.after(TaskMemoryManagerSuite.java:34)
[error] ...

Copy link
Contributor

@vanzin vanzin May 17, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This sounds like this assert triggering, not the actual test failing?

I'm talking about is leaking memory and inserting a fail("blah") in a test and seeing what's reported.

Assert.assertEquals(0, manager.cleanUpAllAllocatedMemory());
manager = null;
}

@Test
public void leakedPageMemoryIsDetected() {
final TaskMemoryManager manager = new TaskMemoryManager(
manager = new TaskMemoryManager(
new StaticMemoryManager(
new SparkConf().set("spark.memory.offHeap.enabled", "false"),
Long.MAX_VALUE,
Long.MAX_VALUE,
1),
0);
final MemoryConsumer c = new TestMemoryConsumer(manager);
manager.allocatePage(4096, c); // leak memory
final MemoryBlock block = manager.allocatePage(4096, c); // leak memory
Assert.assertEquals(4096, manager.getMemoryConsumptionForThisTask());
Assert.assertEquals(4096, manager.cleanUpAllAllocatedMemory());
manager.freePage(block, c);
Copy link
Contributor

@vanzin vanzin Mar 28, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this really needed? Why isn't the call above freeing this memory?

Copy link
Author

@jsoltren jsoltren May 16, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should. We can check for that instead with:
Assert.assertEquals(0, manager.cleanUpAllAllocatedMemory());

}

@Test
public void encodePageNumberAndOffsetOffHeap() {
final SparkConf conf = new SparkConf()
.set("spark.memory.offHeap.enabled", "true")
.set("spark.memory.offHeap.size", "1000");
final TaskMemoryManager manager = new TaskMemoryManager(new TestMemoryManager(conf), 0);
manager = new TaskMemoryManager(new TestMemoryManager(conf), 0);
final MemoryConsumer c = new TestMemoryConsumer(manager, MemoryMode.OFF_HEAP);
final MemoryBlock dataPage = manager.allocatePage(256, c);
// In off-heap mode, an offset is an absolute address that may require more than 51 bits to
Expand All @@ -58,7 +69,7 @@ public void encodePageNumberAndOffsetOffHeap() {

@Test
public void encodePageNumberAndOffsetOnHeap() {
final TaskMemoryManager manager = new TaskMemoryManager(
manager = new TaskMemoryManager(
new TestMemoryManager(new SparkConf().set("spark.memory.offHeap.enabled", "false")), 0);
final MemoryConsumer c = new TestMemoryConsumer(manager, MemoryMode.ON_HEAP);
final MemoryBlock dataPage = manager.allocatePage(256, c);
Expand All @@ -71,7 +82,7 @@ public void encodePageNumberAndOffsetOnHeap() {
public void cooperativeSpilling() {
final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
memoryManager.limit(100);
final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
manager = new TaskMemoryManager(memoryManager, 0);

TestMemoryConsumer c1 = new TestMemoryConsumer(manager);
TestMemoryConsumer c2 = new TestMemoryConsumer(manager);
Expand Down Expand Up @@ -106,14 +117,13 @@ public void cooperativeSpilling() {

c1.free(0);
c2.free(100);
Assert.assertEquals(0, manager.cleanUpAllAllocatedMemory());
}

@Test
public void cooperativeSpilling2() {
final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
memoryManager.limit(100);
final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
manager = new TaskMemoryManager(memoryManager, 0);

TestMemoryConsumer c1 = new TestMemoryConsumer(manager);
TestMemoryConsumer c2 = new TestMemoryConsumer(manager);
Expand Down Expand Up @@ -141,14 +151,13 @@ public void cooperativeSpilling2() {
c1.free(0);
c2.free(80);
c3.free(10);
Assert.assertEquals(0, manager.cleanUpAllAllocatedMemory());
}

@Test
public void shouldNotForceSpillingInDifferentModes() {
final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
memoryManager.limit(100);
final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
manager = new TaskMemoryManager(memoryManager, 0);

TestMemoryConsumer c1 = new TestMemoryConsumer(manager, MemoryMode.ON_HEAP);
TestMemoryConsumer c2 = new TestMemoryConsumer(manager, MemoryMode.OFF_HEAP);
Expand All @@ -170,7 +179,7 @@ public void offHeapConfigurationBackwardsCompatibility() {
final SparkConf conf = new SparkConf()
.set("spark.unsafe.offHeap", "true")
.set("spark.memory.offHeap.size", "1000");
final TaskMemoryManager manager = new TaskMemoryManager(new TestMemoryManager(conf), 0);
manager = new TaskMemoryManager(new TestMemoryManager(conf), 0);
Assert.assertSame(MemoryMode.OFF_HEAP, manager.tungstenMemoryMode);
}

Expand Down