You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uniffle.apache.org by xi...@apache.org on 2023/09/14 08:53:27 UTC
[incubator-uniffle] branch branch-0.8 updated (4b21eca5a -> 547f61fe4)
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a change to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
from 4b21eca5a [MINOR] docs: Fix spark.serializer in README and client_guide (#1180)
new a785d3a7e [#1177] improvement: Reduce the write time of tasks (#1179)
new 547f61fe4 [#1175] fix(netty): Retry failed with StacklessClosedChannelException after channel closed (#1181)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../spark/shuffle/writer/RssShuffleWriter.java | 51 +++++++++++++++++-----
.../impl/grpc/ShuffleServerGrpcNettyClient.java | 9 +++-
2 files changed, 48 insertions(+), 12 deletions(-)
[incubator-uniffle] 01/02: [#1177] improvement: Reduce the write time of tasks (#1179)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit a785d3a7e934f432fc0966dfa24d14de7a9fe319
Author: roryqi <ro...@apache.org>
AuthorDate: Thu Aug 31 10:05:12 2023 +0800
[#1177] improvement: Reduce the write time of tasks (#1179)
### What changes were proposed in this pull request?
Use callback and queue to optimize the write time.
### Why are the changes needed?
Fix: #1177
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
CI passed.
(cherry picked from commit 164e0d02456763f661fb55cc47459312bf4e80b4)
---
.../spark/shuffle/writer/RssShuffleWriter.java | 51 +++++++++++++++++-----
1 file changed, 41 insertions(+), 10 deletions(-)
diff --git a/client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java b/client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java
index fb0c78502..330f56c8d 100644
--- a/client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java
+++ b/client-spark/spark3/src/main/java/org/apache/spark/shuffle/writer/RssShuffleWriter.java
@@ -24,10 +24,12 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
@@ -93,6 +95,8 @@ public class RssShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
protected final ShuffleWriteMetrics shuffleWriteMetrics;
+ private final BlockingQueue<Object> finishEventQueue = new LinkedBlockingQueue<>();
+
// Only for tests
@VisibleForTesting
public RssShuffleWriter(
@@ -293,6 +297,13 @@ public class RssShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
List<ShuffleBlockInfo> shuffleBlockInfoList) {
List<CompletableFuture<Long>> futures = new ArrayList<>();
for (AddBlockEvent event : bufferManager.buildBlockEvents(shuffleBlockInfoList)) {
+ event.addCallback(
+ () -> {
+ boolean ret = finishEventQueue.add(new Object());
+ if (!ret) {
+ LOG.error("Add event " + event + " to finishEventQueue fail");
+ }
+ });
futures.add(shuffleManager.sendData(event));
}
return futures;
@@ -300,17 +311,33 @@ public class RssShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
@VisibleForTesting
protected void checkBlockSendResult(Set<Long> blockIds) {
- long start = System.currentTimeMillis();
- while (true) {
- checkIfBlocksFailed();
- Set<Long> successBlockIds = shuffleManager.getSuccessBlockIds(taskId);
- blockIds.removeAll(successBlockIds);
- if (blockIds.isEmpty()) {
- break;
+ boolean interrupted = false;
+
+ try {
+ long remainingMs = sendCheckTimeout;
+ long end = System.currentTimeMillis() + remainingMs;
+
+ while (true) {
+ try {
+ finishEventQueue.clear();
+ checkIfBlocksFailed();
+ Set<Long> successBlockIds = shuffleManager.getSuccessBlockIds(taskId);
+ blockIds.removeAll(successBlockIds);
+ if (blockIds.isEmpty()) {
+ break;
+ }
+ if (finishEventQueue.isEmpty()) {
+ remainingMs = Math.max(end - System.currentTimeMillis(), 0);
+ Object event = finishEventQueue.poll(remainingMs, TimeUnit.MILLISECONDS);
+ if (event == null) {
+ break;
+ }
+ }
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
}
- LOG.info("Wait " + blockIds.size() + " blocks sent to shuffle server");
- Uninterruptibles.sleepUninterruptibly(sendCheckInterval, TimeUnit.MILLISECONDS);
- if (System.currentTimeMillis() - start > sendCheckTimeout) {
+ if (!blockIds.isEmpty()) {
String errorMsg =
"Timeout: Task["
+ taskId
@@ -322,6 +349,10 @@ public class RssShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
LOG.error(errorMsg);
throw new RssException(errorMsg);
}
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
}
[incubator-uniffle] 02/02: [#1175] fix(netty): Retry failed with StacklessClosedChannelException after channel closed (#1181)
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xianjingfeng pushed a commit to branch branch-0.8
in repository https://gitbox.apache.org/repos/asf/incubator-uniffle.git
commit 547f61fe4d22362637c3d25c3f77f6322a5371a0
Author: xumanbu <ma...@163.com>
AuthorDate: Fri Sep 1 16:57:32 2023 +0800
[#1175] fix(netty): Retry failed with StacklessClosedChannelException after channel closed (#1181)
### What changes were proposed in this pull request?
1、reget a transportClient in ShuffleServerGrpcNettyClient.sendShuffleData RetryCmd
2、add ShuffleServerGrpcNettyClient construct with maxRetryAttempts (but in this pr init ShuffleServerGrpcNettyClient with default maxRetryAttempts(3))
### Why are the changes needed?
Fix: #1175
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing UT.
---------
Co-authored-by: jam.xu <ja...@vipshop.com>
(cherry picked from commit 9622df2d9e46e534e8f5d935f334fdbf761cd12e)
---
.../uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java | 9 +++++++--
1 file changed, 7 insertions(+), 2 deletions(-)
diff --git a/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java b/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java
index 1a867fa22..2fe15aae7 100644
--- a/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java
+++ b/internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleServerGrpcNettyClient.java
@@ -58,7 +58,12 @@ public class ShuffleServerGrpcNettyClient extends ShuffleServerGrpcClient {
private TransportClientFactory clientFactory;
public ShuffleServerGrpcNettyClient(RssConf rssConf, String host, int grpcPort, int nettyPort) {
- super(host, grpcPort);
+ this(rssConf, host, grpcPort, nettyPort, 3);
+ }
+
+ public ShuffleServerGrpcNettyClient(
+ RssConf rssConf, String host, int grpcPort, int nettyPort, int maxRetryAttempts) {
+ super(host, grpcPort, maxRetryAttempts);
this.nettyPort = nettyPort;
TransportContext transportContext = new TransportContext(new TransportConf(rssConf));
this.clientFactory = new TransportClientFactory(transportContext);
@@ -66,7 +71,6 @@ public class ShuffleServerGrpcNettyClient extends ShuffleServerGrpcClient {
@Override
public RssSendShuffleDataResponse sendShuffleData(RssSendShuffleDataRequest request) {
- TransportClient transportClient = getTransportClient();
Map<Integer, Map<Integer, List<ShuffleBlockInfo>>> shuffleIdToBlocks =
request.getShuffleIdToBlocks();
boolean isSuccessful = true;
@@ -88,6 +92,7 @@ public class ShuffleServerGrpcNettyClient extends ShuffleServerGrpcClient {
try {
RetryUtils.retry(
() -> {
+ TransportClient transportClient = getTransportClient();
long requireId =
requirePreAllocation(
request.getAppId(),