Skip to content

[SPARK-10342] [SPARK-10309] [SPARK-10474] [SPARK-10929] [SQL] Cooperative memory management #9241

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
davies wants to merge 17 commits into from
Closed

[SPARK-10342] [SPARK-10309] [SPARK-10474] [SPARK-10929] [SQL] Cooperative memory management #9241

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
d0ada7b
force spilling
Oct 26, 2015
ee6b9a4
fix build
Oct 27, 2015
49b8135
fix Spillable
Oct 27, 2015
7087f2f
fix tests
Oct 27, 2015
ce24f03
add more tests
Oct 28, 2015
8470fc9
address coments
Oct 28, 2015
827d4f0
add tests for TaskMemoryManager
Oct 28, 2015
a3e01d0
counting the longArray in BytesToBytesMap
Oct 29, 2015
7bf76e5
preserve the memory in BytsToBytesMap
Oct 29, 2015
51278f8
use the memory of longArray for in-memory sorter
Oct 29, 2015
27ff4fc
improve spilling
Oct 29, 2015
4491013
release from other consumers first
Oct 29, 2015
c044afe
fix npe
Oct 29, 2015
afc8c7c
address comments
Oct 29, 2015
cda4b2a
address more comments
Oct 29, 2015
4ee1f42
improve OOM handling
Oct 30, 2015
e943e74
fix tests
Oct 30, 2015
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
Prev Previous commit
Next Next commit
improve OOM handling
  • Loading branch information
Davies Liu committed Oct 30, 2015
commit 4ee1f42b08b46469d8631a56b01898d251ebf3e7
28 changes: 16 additions & 12 deletions core/src/main/java/org/apache/spark/shuffle/sort/ShuffleExternalSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,22 +320,26 @@ public void cleanupResources() {
private void growPointerArrayIfNecessary() throws IOException {
assert(inMemSorter != null);
if (!inMemSorter.hasSpaceForAnotherRecord()) {
logger.debug("Attempting to expand sort pointer array");
long used = inMemSorter.getMemoryUsage();
long needed = inMemSorter.getMemoryToExpand();
long needed = used + inMemSorter.getMemoryToExpand();
try {
acquireMemory(used + needed); // could trigger spilling
if (inMemSorter.hasSpaceForAnotherRecord()) {
releaseMemory(used + needed);
} else {
logger.debug("Expand sort pointer array");
acquireMemory(needed); // could trigger spilling
} catch (OutOfMemoryError e) {
// should have trigger spilling
assert(inMemSorter.hasSpaceForAnotherRecord());
return;
}
// check if spilling is triggered or not
if (inMemSorter.hasSpaceForAnotherRecord()) {
releaseMemory(needed);
} else {
try {
inMemSorter.expandPointerArray();
releaseMemory(used);
}
} catch (OutOfMemoryError oom) {
// spilling should be triggered
if (!inMemSorter.hasSpaceForAnotherRecord()) {
spill(); // Just in case that JVM had run out of memory
} catch (OutOfMemoryError oom) {
// Just in case that JVM had run out of memory
releaseMemory(needed);
spill();
}
}
}
Expand Down
3 changes: 1 addition & 2 deletions core/src/main/java/org/apache/spark/shuffle/sort/ShuffleInMemorySorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,7 @@ public void reset() {

private int newLength() {
// Guard against overflow:
return array.length <= Integer.MAX_VALUE / 2 ?
(array.length * 2) : Integer.MAX_VALUE;
return array.length <= Integer.MAX_VALUE / 2 ? (array.length * 2) : Integer.MAX_VALUE;
}

/**
Expand Down
16 changes: 11 additions & 5 deletions core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -710,13 +710,13 @@ public boolean putNewKey(Object keyBase, long keyOffset, int keyLength,
private boolean acquireNewPage(long required) {
try {
currentPage = allocatePage(required);
dataPages.add(currentPage);
Platform.putInt(currentPage.getBaseObject(), currentPage.getBaseOffset(), 0);
pageCursor = 4;
return true;
} catch (OutOfMemoryError e) {
Copy link
Contributor

@JoshRosen JoshRosen Oct 30, 2015

Choose a reason for hiding this comment

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

If we want to catch OOME here, I think that we should do it at a much smaller scope (in the assignment to currentPage but not for adding to dataPages or modifying the page cursor. Given the risks of catching OOME that I mentioned above, the scope of the catch should be as narrow as possible.

return false;
}
dataPages.add(currentPage);
Platform.putInt(currentPage.getBaseObject(), currentPage.getBaseOffset(), 0);
pageCursor = 4;
return true;
}

@Override
Expand Down Expand Up @@ -862,7 +862,13 @@ void growAndRehash() {
final int oldCapacity = (int) oldBitSet.capacity();

// Allocate the new data structures
allocate(Math.min(growthStrategy.nextCapacity(oldCapacity), MAX_CAPACITY));
try {
allocate(Math.min(growthStrategy.nextCapacity(oldCapacity), MAX_CAPACITY));
} catch (OutOfMemoryError oom) {
longArray = oldLongArray;
bitset = oldBitSet;
throw oom;
}

// Re-mask (we don't recompute the hashcode because we stored all 32 bits of it)
for (int pos = oldBitSet.nextSetBit(0); pos >= 0; pos = oldBitSet.nextSetBit(pos + 1)) {
Expand Down
26 changes: 16 additions & 10 deletions core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,19 +290,25 @@ private void growPointerArrayIfNecessary() throws IOException {
assert(inMemSorter != null);
if (!inMemSorter.hasSpaceForAnotherRecord()) {
long used = inMemSorter.getMemoryUsage();
long needed = inMemSorter.getMemoryToExpand();
long needed = used + inMemSorter.getMemoryToExpand();
try {
acquireMemory(used + needed); // could trigger spilling
if (inMemSorter.hasSpaceForAnotherRecord()) {
releaseMemory(used + needed);
} else {
acquireMemory(needed); // could trigger spilling
} catch (OutOfMemoryError e) {
// should have trigger spilling
assert(inMemSorter.hasSpaceForAnotherRecord());
return;
}
// check if spilling is triggered or not
if (inMemSorter.hasSpaceForAnotherRecord()) {
releaseMemory(needed);
} else {
try {
inMemSorter.expandPointerArray();
releaseMemory(used);
}
} catch (OutOfMemoryError oom) {
// spilling should be triggered
if (!inMemSorter.hasSpaceForAnotherRecord()) {
spill(); // Just in case that JVM had run out of memory
} catch (OutOfMemoryError oom) {
// Just in case that JVM had run out of memory
releaseMemory(needed);
spill();
}
}
}
Expand Down