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:57 UTC
[spark] branch branch-3.0 updated (9b268122f68 -> c653d287b15)
This is an automated email from the ASF dual-hosted git repository.
joshrosen pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/spark.git
from 9b268122f68 [SPARK-39293][SQL] Fix the accumulator of ArrayAggregate to handle complex types properly
new 02488e0df30 [SPARK-32911][CORE] Free memory in UnsafeExternalSorter.SpillableIterator.spill() when all records have been read
new c653d287b15 [SPARK-39283][CORE] Fix deadlock between TaskMemoryManager and UnsafeExternalSorter.SpillableIterator
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../unsafe/sort/UnsafeExternalSorter.java | 157 ++++++++++++++-------
.../unsafe/sort/UnsafeExternalSorterSuite.java | 30 ++++
2 files changed, 139 insertions(+), 48 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
[spark] 01/02: [SPARK-32911][CORE] Free memory in UnsafeExternalSorter.SpillableIterator.spill() when all records have been read
Posted by jo...@apache.org.
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
[spark] 02/02: [SPARK-39283][CORE] Fix deadlock between TaskMemoryManager and UnsafeExternalSorter.SpillableIterator
Posted by jo...@apache.org.
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 c653d287b15db3e50fa206071a5435028879f15f
Author: sandeepvinayak <sa...@outlook.com>
AuthorDate: Tue May 31 15:28:07 2022 -0700
[SPARK-39283][CORE] Fix deadlock between TaskMemoryManager and UnsafeExternalSorter.SpillableIterator
### What changes were proposed in this pull request?
This PR fixes a deadlock between TaskMemoryManager and UnsafeExternalSorter.SpillableIterator.
### Why are the changes needed?
We are facing the deadlock issue b/w TaskMemoryManager and UnsafeExternalSorter.SpillableIterator during the join. It turns out that in UnsafeExternalSorter.SpillableIterator#spill() function, it tries to get lock on UnsafeExternalSorter`SpillableIterator` and UnsafeExternalSorter and call `freePage` to free all allocated pages except the last one which takes the lock on TaskMemoryManager.
At the same time, there can be another `MemoryConsumer` using `UnsafeExternalSorter` as part of sorting can try to allocatePage needs to get lock on `TaskMemoryManager` which can cause spill to happen which requires lock on `UnsafeExternalSorter` again causing deadlock.
There is a similar fix here as well:
https://issues.apache.org/jira/browse/SPARK-27338
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing unit tests.
Closes #36680 from sandeepvinayak/SPARK-39283.
Authored-by: sandeepvinayak <sa...@outlook.com>
Signed-off-by: Josh Rosen <jo...@databricks.com>
---
.../unsafe/sort/UnsafeExternalSorter.java | 160 +++++++++++++--------
1 file changed, 104 insertions(+), 56 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 c38327cae8c..ac8170c9d97 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
@@ -21,6 +21,7 @@ import javax.annotation.Nullable;
import java.io.File;
import java.io.IOException;
import java.util.LinkedList;
+import java.util.List;
import java.util.Queue;
import java.util.function.Supplier;
@@ -298,16 +299,30 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
* @return the number of bytes freed.
*/
private long freeMemory() {
- updatePeakMemoryUsed();
+ List<MemoryBlock> pagesToFree = clearAndGetAllocatedPagesToFree();
long memoryFreed = 0;
- for (MemoryBlock block : allocatedPages) {
+ for (MemoryBlock block : pagesToFree) {
memoryFreed += block.size();
freePage(block);
}
+ return memoryFreed;
+ }
+
+ /**
+ * Clear the allocated pages and return the list of allocated pages to let
+ * the caller free the page. This is to prevent the deadlock by nested locks
+ * if the caller locks the UnsafeExternalSorter and call freePage which locks the
+ * TaskMemoryManager and cause nested locks.
+ *
+ * @return list of allocated pages to free
+ */
+ private List<MemoryBlock> clearAndGetAllocatedPagesToFree() {
+ updatePeakMemoryUsed();
+ List<MemoryBlock> pagesToFree = new LinkedList<>(allocatedPages);
allocatedPages.clear();
currentPage = null;
pageCursor = 0;
- return memoryFreed;
+ return pagesToFree;
}
/**
@@ -328,12 +343,27 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
* Frees this sorter's in-memory data structures and cleans up its spill files.
*/
public void cleanupResources() {
- synchronized (this) {
- deleteSpillFiles();
- freeMemory();
- if (inMemSorter != null) {
- inMemSorter.freeMemory();
- inMemSorter = null;
+ // To avoid deadlocks, we can't call methods that lock the TaskMemoryManager
+ // (such as various free() methods) while synchronizing on the UnsafeExternalSorter.
+ // Instead, we will manipulate UnsafeExternalSorter state inside the synchronized
+ // lock and perform the actual free() calls outside it.
+ UnsafeInMemorySorter inMemSorterToFree = null;
+ List<MemoryBlock> pagesToFree = null;
+ try {
+ synchronized (this) {
+ deleteSpillFiles();
+ pagesToFree = clearAndGetAllocatedPagesToFree();
+ if (inMemSorter != null) {
+ inMemSorterToFree = inMemSorter;
+ inMemSorter = null;
+ }
+ }
+ } finally {
+ for (MemoryBlock pageToFree : pagesToFree) {
+ freePage(pageToFree);
+ }
+ if (inMemSorterToFree != null) {
+ inMemSorterToFree.freeMemory();
}
}
}
@@ -576,58 +606,76 @@ public final class UnsafeExternalSorter extends MemoryConsumer {
}
public long spill() throws IOException {
- synchronized (this) {
- if (inMemSorter == null) {
- return 0L;
- }
-
- long currentPageNumber = upstream.getCurrentPageNumber();
+ UnsafeInMemorySorter inMemSorterToFree = null;
+ List<MemoryBlock> pagesToFree = new LinkedList<>();
+ try {
+ synchronized (this) {
+ if (inMemSorter == null) {
+ return 0L;
+ }
- ShuffleWriteMetrics writeMetrics = new ShuffleWriteMetrics();
- 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 currentPageNumber = upstream.getCurrentPageNumber();
+
+ ShuffleWriteMetrics writeMetrics = new ShuffleWriteMetrics();
+ 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) {
- // release the pages except the one that is used. There can still be a caller that
- // is accessing the current record. We free this page in that caller's next loadNext()
- // call.
- for (MemoryBlock page : allocatedPages) {
- if (!loaded || page.pageNumber != currentPageNumber) {
- released += page.size();
- freePage(page);
- } else {
- lastPage = page;
+ long released = 0L;
+ synchronized (UnsafeExternalSorter.this) {
+ // release the pages except the one that is used. There can still be a caller that
+ // is accessing the current record. We free this page in that caller's next loadNext()
+ // call.
+ for (MemoryBlock page : allocatedPages) {
+ if (!loaded || page.pageNumber != currentPageNumber) {
+ released += page.size();
+ // Do not free the page, while we are locking `SpillableIterator`. The `freePage`
+ // method locks the `TaskMemoryManager`, and it's not a good idea to lock 2 objects
+ // in sequence. We may hit dead lock if another thread locks `TaskMemoryManager`
+ // and `SpillableIterator` in sequence, which may happen in
+ // `TaskMemoryManager.acquireExecutionMemory`.
+ pagesToFree.add(page);
+ } else {
+ lastPage = page;
+ }
+ }
+ 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);
}
}
- 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
- assert(inMemSorter != null);
- released += inMemSorter.getMemoryUsage();
- totalSortTimeNanos += inMemSorter.getSortTimeNanos();
- inMemSorter.freeMemory();
- inMemSorter = null;
- taskContext.taskMetrics().incMemoryBytesSpilled(released);
- taskContext.taskMetrics().incDiskBytesSpilled(writeMetrics.bytesWritten());
- totalSpillBytes += released;
- return released;
+ // in-memory sorter will not be used after spilling
+ assert (inMemSorter != null);
+ released += inMemSorter.getMemoryUsage();
+ totalSortTimeNanos += inMemSorter.getSortTimeNanos();
+ // Do not free the sorter while we are locking `SpillableIterator`,
+ // as this can cause a deadlock.
+ inMemSorterToFree = inMemSorter;
+ inMemSorter = null;
+ taskContext.taskMetrics().incMemoryBytesSpilled(released);
+ taskContext.taskMetrics().incDiskBytesSpilled(writeMetrics.bytesWritten());
+ totalSpillBytes += released;
+ return released;
+ }
+ } finally {
+ for (MemoryBlock pageToFree : pagesToFree) {
+ freePage(pageToFree);
+ }
+ if (inMemSorterToFree != null) {
+ inMemSorterToFree.freeMemory();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org