You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by jo...@apache.org on 2022/05/31 22:57:58 UTC
[spark] 01/02: [SPARK-32911][CORE] Free memory in UnsafeExternalSorter.SpillableIterator.spill() when all records have been read
This is an automated email from the ASF dual-hosted git repository.
joshrosen pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/spark.git
commit 02488e0df30a25fd235e2dd25e9b1b3404150125
Author: Tom van Bussel <to...@databricks.com>
AuthorDate: Fri Sep 18 11:49:26 2020 +0000
[SPARK-32911][CORE] Free memory in UnsafeExternalSorter.SpillableIterator.spill() when all records have been read
### What changes were proposed in this pull request?
This PR changes `UnsafeExternalSorter.SpillableIterator` to free its memory (except for the page holding the last record) if it is forced to spill after all of its records have been read. It also makes sure that `lastPage` is freed if `loadNext` is never called the again. The latter was necessary to get my test case to succeed (otherwise it would complain about a leak).
### Why are the changes needed?
No memory is freed after calling `UnsafeExternalSorter.SpillableIterator.spill()` when all records have been read, even though it is still holding onto some memory. This may cause a `SparkOutOfMemoryError` to be thrown, even though we could have just freed the memory instead.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
A test was added to `UnsafeExternalSorterSuite`.
Closes #29787 from tomvanbussel/SPARK-32911.
Authored-by: Tom van Bussel <to...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../unsafe/sort/UnsafeExternalSorter.java | 31 +++++++++++++++-------
.../unsafe/sort/UnsafeExternalSorterSuite.java | 30 +++++++++++++++++++++
2 files changed, 52 insertions(+), 9 deletions(-)
diff --git a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
index e4e369baf9d..c38327cae8c 100644
--- a/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
+++ b/core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
@@ -553,7 +553,7 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
private UnsafeSorterIterator upstream;
private MemoryBlock lastPage = null;
private boolean loaded = false;
- private int numRecords = 0;
+ private int numRecords;
private Object currentBaseObject;
private long currentBaseOffset;
@@ -577,19 +577,25 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
public long spill() throws IOException {
synchronized (this) {
- if (inMemSorter == null || numRecords <= 0) {
+ if (inMemSorter == null) {
return 0L;
}
long currentPageNumber = upstream.getCurrentPageNumber();
ShuffleWriteMetrics writeMetrics = new ShuffleWriteMetrics();
- // Iterate over the records that have not been returned and spill them.
- final UnsafeSorterSpillWriter spillWriter =
- new UnsafeSorterSpillWriter(blockManager, fileBufferSizeBytes, writeMetrics, numRecords);
- spillIterator(upstream, spillWriter);
- spillWriters.add(spillWriter);
- upstream = spillWriter.getReader(serializerManager);
+ if (numRecords > 0) {
+ // Iterate over the records that have not been returned and spill them.
+ final UnsafeSorterSpillWriter spillWriter = new UnsafeSorterSpillWriter(
+ blockManager, fileBufferSizeBytes, writeMetrics, numRecords);
+ spillIterator(upstream, spillWriter);
+ spillWriters.add(spillWriter);
+ upstream = spillWriter.getReader(serializerManager);
+ } else {
+ // Nothing to spill as all records have been read already, but do not return yet, as the
+ // memory still has to be freed.
+ upstream = null;
+ }
long released = 0L;
synchronized (UnsafeExternalSorter.this) {
@@ -605,6 +611,11 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
}
}
allocatedPages.clear();
+ if (lastPage != null) {
+ // Add the last page back to the list of allocated pages to make sure it gets freed in
+ // case loadNext() never gets called again.
+ allocatedPages.add(lastPage);
+ }
}
// in-memory sorter will not be used after spilling
@@ -627,11 +638,12 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
@Override
public void loadNext() throws IOException {
+ assert upstream != null;
MemoryBlock pageToFree = null;
try {
synchronized (this) {
loaded = true;
- // Just consumed the last record from in memory iterator
+ // Just consumed the last record from the in-memory iterator.
if (lastPage != null) {
// Do not free the page here, while we are locking `SpillableIterator`. The `freePage`
// method locks the `TaskMemoryManager`, and it's a bad idea to lock 2 objects in
@@ -639,6 +651,7 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
// `SpillableIterator` in sequence, which may happen in
// `TaskMemoryManager.acquireExecutionMemory`.
pageToFree = lastPage;
+ allocatedPages.clear();
lastPage = null;
}
numRecords--;
diff --git a/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java b/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java
index 9904a0a113f..dc2b4814c82 100644
--- a/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java
+++ b/core/src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java
@@ -390,6 +390,36 @@ public class UnsafeExternalSorterSuite {
assertSpillFilesWereCleanedUp();
}
+ @Test
+ public void forcedSpillingWithFullyReadIterator() throws Exception {
+ final UnsafeExternalSorter sorter = newSorter();
+ long[] record = new long[100];
+ final int recordSize = record.length * 8;
+ final int n = (int) pageSizeBytes / recordSize * 3;
+ for (int i = 0; i < n; i++) {
+ record[0] = i;
+ sorter.insertRecord(record, Platform.LONG_ARRAY_OFFSET, recordSize, 0, false);
+ }
+ assertTrue(sorter.getNumberOfAllocatedPages() >= 2);
+
+ UnsafeExternalSorter.SpillableIterator iter =
+ (UnsafeExternalSorter.SpillableIterator) sorter.getSortedIterator();
+ for (int i = 0; i < n; i++) {
+ assertTrue(iter.hasNext());
+ iter.loadNext();
+ assertEquals(i, Platform.getLong(iter.getBaseObject(), iter.getBaseOffset()));
+ }
+ assertFalse(iter.hasNext());
+
+ assertTrue(iter.spill() > 0);
+ assertEquals(0, iter.spill());
+ assertEquals(n - 1, Platform.getLong(iter.getBaseObject(), iter.getBaseOffset()));
+ assertFalse(iter.hasNext());
+
+ sorter.cleanupResources();
+ assertSpillFilesWereCleanedUp();
+ }
+
@Test
public void forcedSpillingWithNotReadIterator() throws Exception {
final UnsafeExternalSorter sorter = newSorter();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org