This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 4095c8cf7 [CELEBORN-1387] Allow more retries when requesting more
memory in sortbasedpusher
4095c8cf7 is described below
commit 4095c8cf789ff666daae1697d4537f58a3eff76d
Author: CodingCat <[email protected]>
AuthorDate: Fri Apr 19 20:07:26 2024 +0800
[CELEBORN-1387] Allow more retries when requesting more memory in
sortbasedpusher
### What changes were proposed in this pull request?
this PR proposes to have more retries when asking for more memory in
SortBasedPusher to increase the success rate
### Why are the changes needed?
in prod, we have a few jobs suffer from the following exception when asking
for more memory in sortbasedpusher,
```
org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire
1073741824 bytes of memory, got 760427315
```
we can see that it cannot allocate 1G memory but can still get 700+MB which
should be sufficient for handling more records, but still, it failed
we have used this change in our prod env, it successfully mitigate the issue
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
prod
Closes #2461 from CodingCat/oom_retry.
Authored-by: CodingCat <[email protected]>
Signed-off-by: zky.zhoukeyong <[email protected]>
---
.../spark/shuffle/celeborn/SortBasedPusher.java | 64 +++++++++++++++-------
1 file changed, 45 insertions(+), 19 deletions(-)
diff --git
a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
index 7147df6f3..01d852017 100644
---
a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
+++
b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
@@ -327,7 +327,7 @@ public class SortBasedPusher extends MemoryConsumer {
return true;
}
- private void growPointerArrayIfNecessary() throws IOException {
+ private void growPointerArrayIfNecessary(long required) throws IOException {
assert (inMemSorter != null);
if (!inMemSorter.hasSpaceForAnotherRecord()) {
if (inMemSorter.numRecords() <= 0) {
@@ -339,24 +339,50 @@ public class SortBasedPusher extends MemoryConsumer {
}
long used = inMemSorter.getMemoryUsage();
+ long requestedBytes = used / 8 * 2;
+ int allocateMemoryRetryCount = 0;
+ int maxMemoryAllocationRetry = 3;
LongArray array = null;
- try {
- // could trigger spilling
- array = allocateArray(used / 8 * 2);
- } catch (TooLargePageException e) {
- // The pointer array is too big to fix in a single page, spill.
- logger.info(
- "Pushdata in growPointerArrayIfNecessary, memory used {}",
- Utils.bytesToString(getUsed()));
- pushData(true);
- } catch (SparkOutOfMemoryError rethrow) {
- // should have trigger spilling
- if (inMemSorter.numRecords() > 0) {
- logger.error("OOM, unable to grow the pointer array");
- throw rethrow;
+ boolean continueRetry = true;
+ while (allocateMemoryRetryCount < maxMemoryAllocationRetry &&
continueRetry) {
+ try {
+ // could trigger spilling
+ logger.info("asking for " + requestedBytes + " more bytes to
accommodate more records");
+ array = allocateArray(requestedBytes);
+ continueRetry = false;
+ } catch (TooLargePageException e) {
+ // The pointer array is too big to fix in a single page, spill.
+ logger.info(
+ "Pushdata in growPointerArrayIfNecessary, memory used {}",
+ Utils.bytesToString(getUsed()));
+ pushData(true);
+ continueRetry = false;
+ } catch (SparkOutOfMemoryError rethrow) {
+ // should have trigger spilling
+ allocateMemoryRetryCount += 1;
+ if (inMemSorter.numRecords() > 0) {
+ if (allocateMemoryRetryCount == maxMemoryAllocationRetry) {
+ logger.error("OOM, unable to grow the pointer array");
+ throw rethrow;
+ } else {
+ long oldReq = requestedBytes;
+ requestedBytes = Math.max((long) (requestedBytes * 0.5),
required);
+ logger.warn(
+ "cannot allocate "
+ + oldReq
+ + " bytes, cut the request to "
+ + requestedBytes
+ + " bytes and retry",
+ rethrow);
+ pushData(true);
+ }
+ } else {
+ // The new array could not be allocated, but that is not an issue
as it is longer
+ // needed,
+ // as all records were spilled.
+ continueRetry = false;
+ }
}
- // The new array could not be allocated, but that is not an issue as
it is longer needed,
- // as all records were spilled.
}
if (inMemSorter.numRecords() <= 0) {
@@ -401,7 +427,7 @@ public class SortBasedPusher extends MemoryConsumer {
private void allocateMemoryForRecordIfNecessary(int required) throws
IOException {
// Step 1:
// Ensure that the pointer array has space for another record. This may
cause a spill.
- growPointerArrayIfNecessary();
+ growPointerArrayIfNecessary(required);
// Step 2:
// Ensure that the last page has space for another record. This may cause
a spill.
acquireNewPageIfNecessary(required);
@@ -419,7 +445,7 @@ public class SortBasedPusher extends MemoryConsumer {
// no-op that does not allocate any memory, and therefore can't cause a
spill event.
//
// Thus there is no need to call `acquireNewPageIfNecessary` again after
this step.
- growPointerArrayIfNecessary();
+ growPointerArrayIfNecessary(required);
}
@Override
