You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by GitBox <gi...@apache.org> on 2022/12/02 03:25:37 UTC
[GitHub] [incubator-uniffle] zuston opened a new pull request, #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
zuston opened a new pull request, #383:
URL: https://github.com/apache/incubator-uniffle/pull/383
### What changes were proposed in this pull request?
1. Avoid partial events being pushed into pending queue when using `MultipleStorageManager`
2. Remove the cache in `MultipleStorageManager` and change the invoking sequence, especially when using the fallback strategy
3. Make the concrete under storage bound to data-flush event.
### Why are the changes needed?
To fix the fallback invalid when data-flush event enters into pending queue
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Will be added later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038099041
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
Review Comment:
while (retryTimes < maxRetry)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038112289
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
Review Comment:
OK, the exit condition in while will be more clear.
I will correct it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038110257
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
I know. I don’t insist on keeping this into event. I just explain the cache is also unreasonable.
And this is also a potential bug due to memory leak.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038725263
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
I can't get your point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038963402
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039224142
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,97 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Exception e) {
Review Comment:
> @jerqi Do you mean need to catch `Error`s?
Yes, I want to catch more exceptions and errors because we don't use `finally` to release our resource. If errors will happen, the resource will leak.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039582613
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
like
```
ShuffleDataFlushEvent event =
buffer.toFlushEvent(
appId,
shuffleId,
startPartition,
endPartition,
() -> bufferPool.containsKey(appId),
shuffleFlushManager.getDataDistributionType(appId));
event.addCleanupCallback(
xxxxreleaseMemoryxxxx
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038997571
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
Is this necessary? Maybe we could use reflection in our test code. I just have some concern about it. I don't know the best practice. @LuciferYang Could you give some suggestion?
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
Is this necessary? Maybe we could use reflection in our test code. I just have some concern about it. I don't know the best practice. @LuciferYang Could you give me some suggestion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038061107
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
Review Comment:
Why do we need `isPended`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335013152
> We should refactor the ShuffleFlushEvent. We need a method `getShuffleWriteHandler(ShuffleFlushEvent)`. So we can remove `Storage.getCreateWriterHandlerRequest()`.
For what?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038105246
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -105,6 +109,22 @@ public void increaseRetryTimes() {
retryTimes.incrementAndGet();
}
+ public boolean isPended() {
+ return isPended;
+ }
+
+ public void markPended() {
+ isPended = true;
+ }
+
+ public Storage getUnderStorage() {
+ return underStorage;
Review Comment:
If you could fetch this PR into your local IDEA, you could find the invoking chain. This will help understand. Although this PR scope may be a little bit scattered, to fix some unreasonable design.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335152420
If we can't make the process clear, I would better to remove pending queue. There are too many bugs caused by pending queue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038168185
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Sounds great!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335012605
> To implement what you want We just modify here
>
> ```
> private StorageManager selectStorageManager(ShuffleDataFlushEvent event) {
> StorageManager storageManager = storageManagerCache.getIfPresent(event);
> if (storageManager != null) {
> // modify here if storageManager is not canWrite we fallback.
> return storageManager;
> }
> if (event.getSize() > flushColdStorageThresholdSize) {
> storageManager = coldStorageManager;
> } else {
> storageManager = warmStorageManager;
> }
>
> if (!storageManager.canWrite(event)) {
> storageManager = storageManagerFallbackStrategy.tryFallback(
> storageManager, event, warmStorageManager, coldStorageManager);
> }
> storageManagerCache.put(event, storageManager);
> return storageManager;
> }
> ```
Make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038089426
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -105,6 +109,22 @@ public void increaseRetryTimes() {
retryTimes.incrementAndGet();
}
+ public boolean isPended() {
+ return isPended;
+ }
+
+ public void markPended() {
+ isPended = true;
+ }
+
+ public Storage getUnderStorage() {
+ return underStorage;
Review Comment:
Why do we need `underStorage`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037807824
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Emm.. But I just want to bound the concrete `storageManager` into event here for writing in `MultiStorageManager`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1338893902
Merge. thanks @zuston
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038998472
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,97 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Exception e) {
Review Comment:
Exception -> Throwable ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038748521
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
If the event has been pushed into pending queue and re-enter the flush queue, when its disk has problems again, it will be re-pushed to pending queue, that will make the system too much pressure. We should drop these events directly. Right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039498341
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
Sorry. Could you help check this question again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038704672
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
For an event, avoid re-entering the pending queue and wait too long.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039582613
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
like
```
ShuffleDataFlushEvent event =
buffer.toFlushEvent(
appId,
shuffleId,
startPartition,
endPartition,
() -> bufferPool.containsKey(appId),
shuffleFlushManager.getDataDistributionType(appId));
event.addCleanupCallback(
this.releaseMemory
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038095660
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Emmm…. I still think cache is not suitable. Because it holds the reference of events and will cause memory leak. We should implicitly remove this event reference when event is flushed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038232253
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
Why do we need `break`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038097657
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
Review Comment:
If not, how to handle?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335166671
> If we can't make the process clear, I would better to remove pending queue. There are too many bugs caused by pending queue.
Removing is OK for me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1061353355
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
After supporting multiple disk selection #435 , I think we could remove this pending queue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037754781
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Could we avoid using storageManager in a ShuffleDataFlushEvent? It's a little weird.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] codecov-commenter commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335008047
# [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/383?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#383](https://codecov.io/gh/apache/incubator-uniffle/pull/383?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (42f883f) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/897613b8022ea73d5664904bbf43fc27f93505cc?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (897613b) will **decrease** coverage by `0.57%`.
> The diff coverage is `67.60%`.
```diff
@@ Coverage Diff @@
## master #383 +/- ##
============================================
- Coverage 58.59% 58.02% -0.58%
+ Complexity 1589 1381 -208
============================================
Files 193 171 -22
Lines 10910 9067 -1843
Branches 957 792 -165
============================================
- Hits 6393 5261 -1132
+ Misses 4139 3476 -663
+ Partials 378 330 -48
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-uniffle/pull/383?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...rg/apache/uniffle/storage/common/LocalStorage.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c3RvcmFnZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zdG9yYWdlL2NvbW1vbi9Mb2NhbFN0b3JhZ2UuamF2YQ==) | `43.22% <0.00%> (-0.57%)` | :arrow_down: |
| [...org/apache/uniffle/server/ShuffleFlushManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9TaHVmZmxlRmx1c2hNYW5hZ2VyLmphdmE=) | `79.89% <66.00%> (+0.75%)` | :arrow_up: |
| [...g/apache/uniffle/server/ShuffleDataFlushEvent.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9TaHVmZmxlRGF0YUZsdXNoRXZlbnQuamF2YQ==) | `91.89% <70.00%> (-8.11%)` | :arrow_down: |
| [...he/uniffle/server/storage/MultiStorageManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9zdG9yYWdlL011bHRpU3RvcmFnZU1hbmFnZXIuamF2YQ==) | `62.22% <75.00%> (+14.46%)` | :arrow_up: |
| [...che/uniffle/server/storage/HdfsStorageManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9zdG9yYWdlL0hkZnNTdG9yYWdlTWFuYWdlci5qYXZh) | `95.23% <100.00%> (+0.15%)` | :arrow_up: |
| [...he/uniffle/server/storage/LocalStorageManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9zdG9yYWdlL0xvY2FsU3RvcmFnZU1hbmFnZXIuamF2YQ==) | `91.24% <100.00%> (+0.06%)` | :arrow_up: |
| [...e/uniffle/server/storage/SingleStorageManager.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9zdG9yYWdlL1NpbmdsZVN0b3JhZ2VNYW5hZ2VyLmphdmE=) | `69.11% <100.00%> (+1.47%)` | :arrow_up: |
| [.../java/org/apache/uniffle/server/ShuffleServer.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2VydmVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL3NlcnZlci9TaHVmZmxlU2VydmVyLmphdmE=) | `64.49% <0.00%> (-2.90%)` | :arrow_down: |
| [...che/spark/shuffle/writer/BufferManagerOptions.java](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS93cml0ZXIvQnVmZmVyTWFuYWdlck9wdGlvbnMuamF2YQ==) | | |
| ... and [25 more](https://codecov.io/gh/apache/incubator-uniffle/pull/383/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335006295
To implement what you want
We just modify here
```
private StorageManager selectStorageManager(ShuffleDataFlushEvent event) {
StorageManager storageManager = storageManagerCache.getIfPresent(event);
if (storageManager != null) {
return storageManager;
}
if (event.getSize() > flushColdStorageThresholdSize) {
storageManager = coldStorageManager;
} else {
storageManager = warmStorageManager;
}
if (!storageManager.canWrite(event)) {
storageManager = storageManagerFallbackStrategy.tryFallback(
storageManager, event, warmStorageManager, coldStorageManager);
}
storageManagerCache.put(event, storageManager);
return storageManager;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037819492
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> Emm.. But I just want to bound the concrete `storageManager` into event here for writing in `MultiStorageManager`.
Relation is not good idea. But it's better than putting StorageManager into event. If StorageManager is ok, should we put ShuffleFlushManager into the ShuffleFlushEvent instead of `Supplier<Boolean> valid`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037892543
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> Relation is not good idea. But it's better than putting StorageManager into event. If StorageManager is ok, should we put ShuffleFlushManager into the ShuffleFlushEvent instead of Supplier<Boolean> valid?
Sounds reasonable. The only thing is that how to clean up this relation when event is flushed.
> @zuston If we use this solution you proposed in https://github.com/apache/incubator-uniffle/pull/235, i think we don't need maintain this relation
The relation will help event find the concrete under-storage-manager selected by method of `selectStorage` when invoked by `write` method. Right? @xianjingfeng
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039231169
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
Do you mean that we need to create a callback of `ShuffleBufferManager#releaseMemory` in `shuffleBuffer` for the usage of event's `cleanupCallback`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] LuciferYang commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039168645
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
I think it is OK to add a test only method or reflection, but this method seems not robust enough. It just reassigns `clients`, but doesn't clear the state of the original one, such as close shuffleServerClient
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1040354026
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
Got your thought. Thanks for your advice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038111430
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
Review Comment:
For me, it is the same with this PR’s current implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039205542
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
Got your point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335075130
> I think Storage.getOrCreateWriterHandlerRequest isn't reasonable.
This is not related with this PR. Let's focus on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039216409
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,97 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Exception e) {
Review Comment:
For catch more exceptions. Throwable will be more safe.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1061252293
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
Is there any process about this optimization?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039497544
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,97 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Exception e) {
Review Comment:
Done
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037761639
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> Currently it can't. But this solution sounds great. We could extend the interface of `storage`
I think we need add a interface for StorageManager#isBelongto(Storage storage). Storage is in the module `storage`. StorageManager is in the module server. Storage shouldn't depend on StorageManager.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039007763
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
Should we print some logs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037758561
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Currently it can't. But this solution sounds great. We could extend the interface of `storage`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038081999
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
Review Comment:
Why do we need to modify the flush process?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038084207
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -99,7 +100,7 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
try {
ShuffleServerMetrics.gaugeEventQueueSize.set(flushQueue.size());
ShuffleServerMetrics.gaugeWriteHandler.inc();
- flushToFile(event);
+ flushToFileImpl(event);
Review Comment:
Why do we rename to `flushToFIleImpl`?
Usually we use `Impl` method , because we have already had a `interface` method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038669777
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
Let us leave a todo comment for later optimization, and make current PR little.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039178995
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
The original intention of this method is to solve the problem of client closed by other test cases. This will make some test cases flaky.
> It just reassigns clients, but doesn't clear the state of the original one, such as close shuffleServerClient
You mean we'd better to close the client when cleaning the cache. Right? @LuciferYang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039188027
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,97 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Exception e) {
Review Comment:
For what? I can't get this point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039239974
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
Got it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039007865
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
Could we put the logic into `doCleanup`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038125988
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
Could we avoid judging the `MultiStorageManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037757255
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Could we only have `Storage` field. Could we use `Storage` to find its StorageManager?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038098321
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -99,7 +100,7 @@ public ShuffleFlushManager(ShuffleServerConf shuffleServerConf, String shuffleSe
try {
ShuffleServerMetrics.gaugeEventQueueSize.set(flushQueue.size());
ShuffleServerMetrics.gaugeWriteHandler.inc();
- flushToFile(event);
+ flushToFileImpl(event);
Review Comment:
Yes. I’ll correct it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335168366
> > If we can't make the process clear, I would better to remove pending queue. There are too many bugs caused by pending queue.
>
> Removing is OK for me.
Let's remove it. It's more simple.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038158015
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
Emm... It's weird, let me think again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] LuciferYang commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039220948
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,97 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Exception e) {
Review Comment:
@jerqi Do you mean need to catch `Error`s?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039178995
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
The original intention of this method is to solve the problem of client closed by other test cases due to existence of cache. This will make some test cases flaky.
> It just reassigns clients, but doesn't clear the state of the original one, such as close shuffleServerClient
You mean we'd better to close the client when cleaning the cache. Right? @LuciferYang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] LuciferYang commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039182041
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleServerClientFactory.java:
##########
@@ -59,4 +59,9 @@ public synchronized ShuffleServerClient getShuffleServerClient(
}
return serverToClients.get(shuffleServerInfo);
}
+
+ // Only for tests
+ public synchronized void cleanupCache() {
Review Comment:
Yes, if we implement such method for `ShuffleServerClientFactory`, it should be more robust. Otherwise, when `cleanupCache` is called, the resources in previous `clients` may leak?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi merged pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi merged PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1040520557
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,98 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
+ LOG.error("Drop this event directly due to already having entered pending queue. event: {}", event);
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Throwable throwable) {
+ // just log the error, don't throw the exception and stop the flush thread
+ LOG.error("Exception happened when process flush shuffle data for {}", event, throwable);
+ event.increaseRetryTimes();
}
}
- }
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for {} in {} times, shuffle data will be lost", event, retryMax);
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
Review Comment:
`getUnderStorage` may be null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038036121
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Do we have ideas on this problem?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038149403
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
Could we add an interface `supportPending` for storageManager?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037756450
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Yes. I also want to avoid this, but if we don't, we have to maintain this relation mapping in multipleStorageManager. However we couldn't remove this structure when event's lifecycle finished, because we don't have similar interface in storage manager
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
xianjingfeng commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038688836
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> The relation will help event find the concrete under-storage-manager selected by method of `selectStorage` when invoked by `write` method. Right? @xianjingfeng
Yes, if we remove pending queue, we don't need maintain this relation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037756450
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Yes. I also want to avoid this, but if we don't, we have to maintain this relation mapping in multipleStorageManager. However we couldn't remove this structure when event's lifecycle finished, because we don't similar interface in storage manager
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335050022
> > We should refactor the ShuffleFlushEvent. We need a method `getShuffleWriteHandler(ShuffleFlushEvent)`. So we can remove `Storage.getCreateWriterHandlerRequest()`.
>
> For what?
I think `Storage.getOrCreateWriterHandlerRequest` isn't reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038098561
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> Emmm…. I still think cache is not suitable. Because it holds the reference of events and will cause memory leak. We should implicitly remove this event reference when event is flushed.
It's not acceptable for me to put StorageManager into ShuffleDataFlushEvent. I would rather give up this strategy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038750016
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ if (event.isPended()) {
+ // add metrics
break;
Review Comment:
Maybe you need some comments. I think it's your strategy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039193124
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
Yes. But the `shuffleBufferManager` is not a member variable for `ShuffleBuffer`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039580654
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
> Do you mean that we need to create a callback of `ShuffleBufferManager#releaseMemory` in `shuffleBuffer` for the usage of event's `cleanupCallback`
No. I mean we could add a cleanupCallback for ShuffleFlushDataEvent in
https://github.com/apache/incubator-uniffle/blob/9a552b42f3b5073600b995d0da38204c2fe65e58/server/src/main/java/org/apache/uniffle/server/buffer/ShuffleBufferManager.java#L211
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038044003
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> Relation is not good idea. But it's better than putting StorageManager into event. If StorageManager is ok, should we put ShuffleFlushManager into the ShuffleFlushEvent instead of Supplier<Boolean> valid?
I have removed the `storageManager` params in event and reuse the cache to keep relations. Although this is a little bit weird.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038669777
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
Let us leave a todo comment for later optimization, and make current PR small.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038104132
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -105,6 +109,22 @@ public void increaseRetryTimes() {
retryTimes.incrementAndGet();
}
+ public boolean isPended() {
+ return isPended;
+ }
+
+ public void markPended() {
+ isPended = true;
+ }
+
+ public Storage getUnderStorage() {
+ return underStorage;
Review Comment:
OK.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] xianjingfeng commented on a diff in pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
xianjingfeng commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1037814703
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
> I have two directions to improve this part of MultiStorageManager
>
> 1. Remove the `storageManagerCache` , it looks unused and bring some problems. One possible problem I have seen is that, when one event enters into pending queue due to storage cannot write, it will not have a chance to get a new fallback strategy due to cache.
> 2. Avoid changing the external reference object in `write` method. If we want to fallback write to other storage. We could use the invoking sequence like. (selectStorage -> write) if failed and then (selectStorage -> write), instead of selectStorage -> write (failed) -> write(choose other storage in this method). That means we should change the fallback strategy invoked from `write` method to `selectStorage` method.
@zuston If we use this solution you proposed in #235, i think we don't need maintain this relation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038061874
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
Review Comment:
Why do need `while(true)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335129691
> > > We should refactor the ShuffleFlushEvent. We need a method `getShuffleWriteHandler(ShuffleFlushEvent)`. So we can remove `Storage.getCreateWriterHandlerRequest()`.
> >
> >
> > For what?
>
> I think `Storage.getOrCreateWriterHandlerRequest` isn't reasonable.
My mistake. It's necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038101787
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -105,6 +109,22 @@ public void increaseRetryTimes() {
retryTimes.incrementAndGet();
}
+ public boolean isPended() {
+ return isPended;
+ }
+
+ public void markPended() {
+ isPended = true;
+ }
+
+ public Storage getUnderStorage() {
+ return underStorage;
Review Comment:
For the getting concrete storage when updating metrics. In original logic, the storage will be gotten by invoking selectStorage method. This is unreasonable.
We should guarantee that StorageManager#selectStorage only could be invoked before writing. After written, the storage has been bound to event, we shouldn’t invoke selectStorage again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038098000
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
Review Comment:
I think this logic need to be refactored.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on pull request #383: [WIP][BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#issuecomment-1335000686
We should refactor the ShuffleFlushEvent. We need a method `getShuffleWriteHandler(ShuffleFlushEvent)`. So we can remove `Storage.getCreateWriterHandlerRequest()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038113069
##########
server/src/main/java/org/apache/uniffle/server/ShuffleDataFlushEvent.java:
##########
@@ -36,6 +38,9 @@ public class ShuffleDataFlushEvent {
private final Supplier<Boolean> valid;
private final ShuffleBuffer shuffleBuffer;
private final AtomicInteger retryTimes = new AtomicInteger();
+ private boolean isPended = false;
+ private StorageManager underStorageManager;
Review Comment:
Could we add a callback for ShuffleFlushEvent. In the callback, it will remove the data. When we call the cleanupFlushEventData(event) method. We call event.clearCallback();
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [BUG] Overlook fallback when data-flush event enters into pending queue
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1038138326
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -142,87 +143,96 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
}
- private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
+ private void flushToFileImpl(ShuffleDataFlushEvent event) {
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (true) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
+
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: " + event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ if (storageManager instanceof MultiStorageManager) {
Review Comment:
If we want to remove pending queue, this is unnecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1040562696
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -143,86 +143,98 @@ public void addToFlushQueue(ShuffleDataFlushEvent event) {
}
private void flushToFile(ShuffleDataFlushEvent event) {
-
- Storage storage = storageManager.selectStorage(event);
- if (storage != null && !storage.canWrite()) {
- addPendingEvents(event);
- return;
- }
-
long start = System.currentTimeMillis();
- List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
boolean writeSuccess = false;
- try {
- // storage info maybe null if the application cache was cleared already
- if (storage != null) {
- if (blocks == null || blocks.isEmpty()) {
- LOG.info("There is no block to be flushed: " + event);
- } else if (!event.isValid()) {
- // avoid printing error log
+
+ while (event.getRetryTimes() <= retryMax) {
+ try {
+ if (!event.isValid()) {
writeSuccess = true;
LOG.warn("AppId {} was removed already, event {} should be dropped", event.getAppId(), event);
- } else {
- String user = StringUtils.defaultString(
- shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
- StringUtils.EMPTY
- );
- CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
- storageType,
- event.getAppId(),
- event.getShuffleId(),
- event.getStartPartition(),
- event.getEndPartition(),
- storageBasePaths.toArray(new String[storageBasePaths.size()]),
- shuffleServerId,
- hadoopConf,
- storageDataReplica,
- user);
- ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
- do {
- if (event.getRetryTimes() > retryMax) {
- LOG.error("Failed to write data for " + event + " in " + retryMax + " times, shuffle data will be lost");
- ShuffleServerMetrics.incStorageFailedCounter(storage.getStorageHost());
- break;
- }
- if (!event.isValid()) {
- LOG.warn("AppId {} was removed already, event {} should be dropped, may leak one handler",
- event.getAppId(), event);
- // avoid printing error log
- writeSuccess = true;
- break;
- }
+ break;
+ }
- writeSuccess = storageManager.write(storage, handler, event);
+ List<ShufflePartitionedBlock> blocks = event.getShuffleBlocks();
+ if (blocks == null || blocks.isEmpty()) {
+ LOG.info("There is no block to be flushed: {}", event);
+ break;
+ }
- if (writeSuccess) {
- updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
- ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ Storage storage = storageManager.selectStorage(event);
+ if (storage == null) {
+ break;
+ }
+
+ if (!storage.canWrite()) {
+ // todo: Could we add an interface supportPending for storageManager
+ // to unify following logic of multiple different storage managers
+ if (storageManager instanceof MultiStorageManager) {
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
+ continue;
+ } else {
+ // To avoid being re-pushed to pending queue and make the server too much pressure,
+ // it's better to drop directly.
+ if (event.isPended()) {
+ LOG.error("Drop this event directly due to already having entered pending queue. event: {}", event);
break;
- } else {
- event.increaseRetryTimes();
- ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
- } while (event.getRetryTimes() <= retryMax);
+ event.increaseRetryTimes();
+ event.markPended();
+ addPendingEvents(event);
+ return;
+ }
}
- }
- } catch (Exception e) {
- // just log the error, don't throw the exception and stop the flush thread
- LOG.error("Exception happened when process flush shuffle data for " + event, e);
- } finally {
- cleanupFlushEventData(event);
- if (shuffleServer != null) {
- long duration = System.currentTimeMillis() - start;
+
+ String user = StringUtils.defaultString(
+ shuffleServer.getShuffleTaskManager().getUserByAppId(event.getAppId()),
+ StringUtils.EMPTY
+ );
+ CreateShuffleWriteHandlerRequest request = new CreateShuffleWriteHandlerRequest(
+ storageType,
+ event.getAppId(),
+ event.getShuffleId(),
+ event.getStartPartition(),
+ event.getEndPartition(),
+ storageBasePaths.toArray(new String[storageBasePaths.size()]),
+ shuffleServerId,
+ hadoopConf,
+ storageDataReplica,
+ user);
+ ShuffleWriteHandler handler = storage.getOrCreateWriteHandler(request);
+ writeSuccess = storageManager.write(storage, handler, event);
if (writeSuccess) {
- LOG.debug("Flush to file success in " + duration + " ms and release " + event.getSize() + " bytes");
+ updateCommittedBlockIds(event.getAppId(), event.getShuffleId(), blocks);
+ ShuffleServerMetrics.incStorageSuccessCounter(storage.getStorageHost());
+ break;
} else {
- LOG.error("Flush to file for " + event + " failed in "
- + duration + " ms and release " + event.getSize() + " bytes");
+ event.increaseRetryTimes();
+ ShuffleServerMetrics.incStorageRetryCounter(storage.getStorageHost());
}
+ } catch (Throwable throwable) {
+ // just log the error, don't throw the exception and stop the flush thread
+ LOG.error("Exception happened when process flush shuffle data for {}", event, throwable);
+ event.increaseRetryTimes();
}
}
- }
+ if (event.getRetryTimes() > retryMax) {
+ LOG.error("Failed to write data for {} in {} times, shuffle data will be lost", event, retryMax);
+ ShuffleServerMetrics.incStorageFailedCounter(event.getUnderStorage().getStorageHost());
Review Comment:
Nice catch. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] zuston commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
zuston commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1040365710
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
[GitHub] [incubator-uniffle] jerqi commented on a diff in pull request #383: [ISSUE-380] Refactor the flush process to fix fallback fail
Posted by GitBox <gi...@apache.org>.
jerqi commented on code in PR #383:
URL: https://github.com/apache/incubator-uniffle/pull/383#discussion_r1039217307
##########
server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java:
##########
@@ -309,10 +320,7 @@ void processPendingEvents() throws Exception {
}
private void cleanupFlushEventData(ShuffleDataFlushEvent event) {
- ShuffleBuffer shuffleBuffer = event.getShuffleBuffer();
- if (shuffleBuffer != null) {
- shuffleBuffer.clearInFlushBuffer(event.getEventId());
- }
+ event.doCleanup();
if (shuffleServer != null) {
Review Comment:
> Yes. But the `shuffleBufferManager` is not a member variable for `ShuffleBuffer`
We can add this logic as a cleanup callback. It's ok although ShuffleBufferManager is not a member variable for ShuffleBuffer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org