You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "advancedxy (via GitHub)" <gi...@apache.org> on 2023/03/08 03:07:17 UTC
[GitHub] [incubator-uniffle] advancedxy opened a new pull request, #693: [WIP]feat: support stage recompute for spark clients
advancedxy opened a new pull request, #693:
URL: https://github.com/apache/incubator-uniffle/pull/693
### What changes were proposed in this pull request?
1. add shuffle manager service definition
2. shuffle manager server and client impl
3. draft spark stage recompute support
### Why are the changes needed?
To improve stability of Spark application over RSS
Resolve #477
### Does this PR introduce _any_ user-facing change?
Yes. Updated lated
### How was this patch tested?
WIP
--
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] advancedxy commented on a diff in pull request #693: [WIP]feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1144933284
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleManagerGrpcClient.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.impl.grpc;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.proto.RssProtos;
+import org.apache.uniffle.proto.ShuffleManagerGrpc;
+
+public class ShuffleManagerGrpcClient extends GrpcClient implements ShuffleManagerClient {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcClient.class);
+ private static final long RPC_TIMEOUT_DEFAULT_MS = 60000;
+ private long rpcTimeout = RPC_TIMEOUT_DEFAULT_MS;
+ private ShuffleManagerGrpc.ShuffleManagerBlockingStub blockingStub;
+
+ public ShuffleManagerGrpcClient(String host, int port) {
+ this(host, port, 3);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts) {
+ this(host, port, maxRetryAttempts, true);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext) {
+ super(host, port, maxRetryAttempts, usePlaintext);
+ blockingStub = ShuffleManagerGrpc.newBlockingStub(channel);
+ }
+
+ public ShuffleManagerGrpc.ShuffleManagerBlockingStub getBlockingStub() {
+ return blockingStub.withDeadlineAfter(rpcTimeout, TimeUnit.MILLISECONDS);
+ }
+
+ public String getDesc() {
+ return "Shuffle manager grpc client ref " + host + ":" + port;
+ }
+ @Override
+ public RssReportShuffleFetchFailureResponse reportShuffleFetchFailure(RssReportShuffleFetchFailureRequest request) {
+ RssProtos.ReportShuffleFetchFailureRequest protoRequest = request.toProto();
+ int retryNum = 0;
+ while (retryNum <= maxRetryAttempts) {
Review Comment:
Will address it 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 #693: feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1149970963
##########
proto/src/main/proto/Rss.proto:
##########
@@ -488,4 +488,26 @@ message CancelDecommissionRequest {
message CancelDecommissionResponse {
StatusCode status = 1;
string retMsg = 2;
-}
\ No newline at end of file
+}
+// ShuffleManager service lives inside of compute-engine's application master, which handles rss shuffle specific logic
+// per application.
+service ShuffleManager {
+ rpc reportShuffleFetchFailure (ReportShuffleFetchFailureRequest) returns (ReportShuffleFetchFailureResponse);
+}
+
+message ReportShuffleFetchFailureRequest {
+ // appId normally should be omitted, it's used to avoid wrongly request issued from remaining executors of another
+ // app which accidentally has the same shuffle manager port with this app.
+ string appId = 1;
+ int32 shuffleId = 2;
+ int32 stageAttemptId = 3;
+ int32 partitionId = 4;
+ string exception = 5;
+ string serverId = 6;
Review Comment:
We already have a proto type called ShuffleServerId. Should we reuse 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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1149966880
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkConfig.java:
##########
@@ -265,6 +266,17 @@ public class RssSparkConfig {
+ " spark.rss.estimate.server.assignment.enabled"))
.createWithDefault(RssClientConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ public static final ConfigEntry<Integer> RSS_SHUFFLE_MANAGER_GRPC_PORT = createIntegerBuilder(
+ new ConfigBuilder(SPARK_RSS_CONFIG_PREFIX + RssClientConf.SHUFFLE_MANAGER_GRPC_PORT.key())
+ .internal()
+ .doc(RssClientConf.SHUFFLE_MANAGER_GRPC_PORT.description())
+ ).createWithDefault(-1);
+
+ public static final ConfigEntry<Boolean> RSS_STAGE_RECOMPUTE = createBooleanBuilder(
Review Comment:
This is only specific to Spark. So I don't think it's appropriate to put this configuration into `RSSClientConf`.
--
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 #693: feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1144969670
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkConfig.java:
##########
@@ -265,6 +266,17 @@ public class RssSparkConfig {
+ " spark.rss.estimate.server.assignment.enabled"))
.createWithDefault(RssClientConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ public static final ConfigEntry<Integer> RSS_SHUFFLE_MANAGER_GRPC_PORT = createIntegerBuilder(
Review Comment:
We need to add document for this feature.
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/ShuffleManagerGrpcService.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+import com.google.common.collect.Maps;
+import io.grpc.stub.StreamObserver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.proto.RssProtos;
+import org.apache.uniffle.proto.ShuffleManagerGrpc;
+
+public class ShuffleManagerGrpcService extends ShuffleManagerGrpc.ShuffleManagerImplBase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcService.class);
+ // todo: unregister shuffle status when shuffle is gced.
+ private final Map<Integer, RssShuffleStatus> shuffleStatus = Maps.newConcurrentMap();
Review Comment:
When do we clean this collection?
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/reader/ShuffleFetchFailedWrapper.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.reader;
+
+import java.io.IOException;
+
+import org.apache.spark.shuffle.FetchFailedException;
+import org.apache.spark.shuffle.RssSparkShuffleUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Product2;
+import scala.collection.AbstractIterator;
+import scala.collection.Iterator;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.config.RssConf;
+import org.apache.uniffle.common.exception.RssFetchFailedException;
+
+public class ShuffleFetchFailedWrapper {
Review Comment:
Could we have a better name? Is it a iterator actually?
##########
client/src/main/java/org/apache/uniffle/client/util/RssClientConfig.java:
##########
@@ -86,4 +86,6 @@ public class RssClientConfig {
public static final String RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER = "rss.estimate.task.concurrency.per.server";
public static final int RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE = 80;
+ public static final String RSS_STAGE_RECOMPUTE = "rss.stage.reCompute";
Review Comment:
`reCompute` seems weird.
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/reader/RssShuffleReader.java:
##########
@@ -173,6 +175,20 @@ public Void apply(TaskContext context) {
if (!(resultIter instanceof InterruptibleIterator)) {
resultIter = new InterruptibleIterator<>(context, resultIter);
}
+ // stage re-compute and shuffle manager server port are both set
+ if (rssConf.getBoolean(RssClientConfig.RSS_STAGE_RECOMPUTE, false)
Review Comment:
Is it right? Should we use Spark's config option?
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssFeatureMatrix.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle;
+
+public class RssFeatureMatrix {
Review Comment:
Could we move this method to `RssSparkShuffelUtils`?
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/AbstractRssShuffleManagerBase.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.apache.spark.MapOutputTracker;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.SparkException;
+import org.apache.spark.shuffle.RssFeatureMatrix;
+import org.apache.spark.shuffle.SparkVersionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractRssShuffleManagerBase implements RssShuffleManagerBase {
Review Comment:
Why do we choose to add interface?
##########
proto/src/main/proto/Rss.proto:
##########
@@ -488,4 +488,26 @@ message CancelDecommissionRequest {
message CancelDecommissionResponse {
StatusCode status = 1;
string retMsg = 2;
-}
\ No newline at end of file
+}
+// ShuffleManager service lives inside of compute-engine's application master, which handles rss shuffle specific logic
+// per application.
+service ShuffleManager {
+ rpc reportShuffleFetchFailure (ReportShuffleFetchFailureRequest) returns (ReportShuffleFetchFailureResponse);
+}
+
+message ReportShuffleFetchFailureRequest {
+ // appId normally should be omitted, it's used to avoid wrongly request issued from remaining executors of another
+ // app which accidentally has the same shuffle manager port with this app.
+ string appId = 1;
+ int32 shuffleId = 2;
+ int32 stageAttemptId = 3;
+ int32 partitionId = 4;
+ string exception = 5;
+ string serverId = 6;
Review Comment:
What do the serverId mean?
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/reader/ShuffleFetchFailedWrapper.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.reader;
+
+import java.io.IOException;
+
+import org.apache.spark.shuffle.FetchFailedException;
+import org.apache.spark.shuffle.RssSparkShuffleUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Product2;
+import scala.collection.AbstractIterator;
+import scala.collection.Iterator;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.config.RssConf;
+import org.apache.uniffle.common.exception.RssFetchFailedException;
+
+public class ShuffleFetchFailedWrapper {
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleFetchFailedWrapper.class);
+ private RssConf conf;
+ private String appId;
+ private int shuffleId;
+ private int partitionId;
+ private int stageAttemptId;
+ private String reportServerHost;
+ private int reportServerPort;
+
+ private ShuffleFetchFailedWrapper() {
+
+ }
+
+ static ShuffleFetchFailedWrapper newWrapper(RssConf conf) {
+ ShuffleFetchFailedWrapper wrapper = new ShuffleFetchFailedWrapper();
+ wrapper.conf = conf;
+ return wrapper;
+ }
+
+ ShuffleFetchFailedWrapper appId(String appId) {
+ this.appId = appId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper shuffleId(int shuffleId) {
+ this.shuffleId = shuffleId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper partitionId(int partitionId) {
+ this.partitionId = partitionId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper stageAttemptId(int stageAttemptId) {
+ this.stageAttemptId = stageAttemptId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper reportServerHost(String host) {
+ this.reportServerHost = host;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper port(int port) {
+ this.reportServerPort = port;
+ return this;
+ }
+
+
+ public <K, C> Iterator<Product2<K, C>> wrap(Iterator<Product2<K, C>> iter) {
+ return new IteratorImpl<>(iter, this);
+ }
+
+ private static class IteratorImpl<K, C> extends AbstractIterator<Product2<K, C>> {
+ private Iterator<Product2<K, C>> iter;
+ private ShuffleFetchFailedWrapper wrapper;
+
+ IteratorImpl(Iterator<Product2<K, C>> iter, ShuffleFetchFailedWrapper wrapper) {
+ this.iter = iter;
+ this.wrapper = wrapper;
+ }
+
+ private RuntimeException generateFetchFailedIfNecessary(RssFetchFailedException e) {
+ String driver = wrapper.reportServerHost;
+ int port = wrapper.reportServerPort;
+ // todo: reuse this manager client if this is a bottleneck.
+ try (ShuffleManagerClient client = RssSparkShuffleUtils.createShuffleManagerClient(driver, port)) {
Review Comment:
Do we need to process InterruptException?
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleManagerClientFactory.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.factory;
+
+import org.apache.uniffle.client.impl.grpc.ShuffleManagerGrpcClient;
+import org.apache.uniffle.common.ClientType;
+
+public class ShuffleManagerClientFactory {
+
+ private static final ShuffleManagerClientFactory INSTANCE = new ShuffleManagerClientFactory(ClientType.GRPC);
Review Comment:
For Singleton, our system prefer LazyHolder's style like class `ShuffleServerClientFactory`
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleManagerClientFactory.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.factory;
+
+import org.apache.uniffle.client.impl.grpc.ShuffleManagerGrpcClient;
+import org.apache.uniffle.common.ClientType;
+
+public class ShuffleManagerClientFactory {
+
+ private static final ShuffleManagerClientFactory INSTANCE = new ShuffleManagerClientFactory(ClientType.GRPC);
+
+ public static ShuffleManagerClientFactory getINSTANCE() {
+ return INSTANCE;
+ }
+
+ private ClientType clientType;
+
+ public ShuffleManagerClientFactory(ClientType clientType) {
+ this.clientType = clientType;
+ }
+
+ public ShuffleManagerGrpcClient createShuffleManagerClient(String host, int port) {
+ if (clientType.equals(ClientType.GRPC)) {
Review Comment:
Will it have problems when we use `ClientType.GRPC_NETTY`?
--
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] advancedxy commented on pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#issuecomment-1486131908
The CI fails hardly. Seems that the rss shuffle manager could not be extends another base class. I would investigate it further.
Also, I would split this PR into two or three parts:
1. for the ShuffleManager sever, client, protos and related changes.
2. Spark side changes, such as shuffle manager and data reader iterator
3. the user documentation and other related polishing changes.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1145709454
##########
client-spark/spark3/src/main/java/org/apache/spark/shuffle/reader/RssShuffleReader.java:
##########
@@ -173,6 +175,20 @@ public Void apply(TaskContext context) {
if (!(resultIter instanceof InterruptibleIterator)) {
resultIter = new InterruptibleIterator<>(context, resultIter);
}
+ // stage re-compute and shuffle manager server port are both set
+ if (rssConf.getBoolean(RssClientConfig.RSS_STAGE_RECOMPUTE, false)
Review Comment:
yes.. `RssClientConfig.RSS_STAGE_RECOMPUTE` is a string, the spark prefix is stripped.
--
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 #693: feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1149973040
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssFeatureMatrix.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle;
+
+public class RssFeatureMatrix {
Review Comment:
I'm a little confused about this class name. FeatureMatrix sounds like a concept of linear algebra.
--
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] advancedxy commented on a diff in pull request #693: [WIP]feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1131091779
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleManagerGrpcClient.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.impl.grpc;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.proto.RssProtos;
+import org.apache.uniffle.proto.ShuffleManagerGrpc;
+
+public class ShuffleManagerGrpcClient extends GrpcClient implements ShuffleManagerClient {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcClient.class);
+ private static final long RPC_TIMEOUT_DEFAULT_MS = 60000;
+ private long rpcTimeout = RPC_TIMEOUT_DEFAULT_MS;
+ private ShuffleManagerGrpc.ShuffleManagerBlockingStub blockingStub;
+
+ public ShuffleManagerGrpcClient(String host, int port) {
+ this(host, port, 3);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts) {
+ this(host, port, maxRetryAttempts, true);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext) {
+ super(host, port, maxRetryAttempts, usePlaintext);
+ blockingStub = ShuffleManagerGrpc.newBlockingStub(channel);
+ }
+
+ public ShuffleManagerGrpc.ShuffleManagerBlockingStub getBlockingStub() {
+ return blockingStub.withDeadlineAfter(rpcTimeout, TimeUnit.MILLISECONDS);
+ }
+
+ public String getDesc() {
+ return "Shuffle manager grpc client ref " + host + ":" + port;
+ }
+ @Override
+ public RssReportShuffleFetchFailureResponse reportShuffleFetchFailure(RssReportShuffleFetchFailureRequest request) {
+ RssProtos.ReportShuffleFetchFailureRequest protoRequest = request.toProto();
+ int retryNum = 0;
+ while (retryNum <= maxRetryAttempts) {
Review Comment:
Thanks for your suggestion. I will polish this in following commits. Let's first focus on the high level idea, do you think are there any arch/procedures should be adjusted?
--
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 #693: feat: support stage recompute for spark clients
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1145590005
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkConfig.java:
##########
@@ -265,6 +266,17 @@ public class RssSparkConfig {
+ " spark.rss.estimate.server.assignment.enabled"))
.createWithDefault(RssClientConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ public static final ConfigEntry<Integer> RSS_SHUFFLE_MANAGER_GRPC_PORT = createIntegerBuilder(
+ new ConfigBuilder(SPARK_RSS_CONFIG_PREFIX + RssClientConf.SHUFFLE_MANAGER_GRPC_PORT.key())
+ .internal()
+ .doc(RssClientConf.SHUFFLE_MANAGER_GRPC_PORT.description())
+ ).createWithDefault(-1);
+
+ public static final ConfigEntry<Boolean> RSS_STAGE_RECOMPUTE = createBooleanBuilder(
Review Comment:
Could you help put the new client config into `RssClientConf`?
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/reader/RssShuffleDataIterator.java:
##########
@@ -101,7 +101,8 @@ public boolean hasNext() {
// read next segment
long startFetch = System.currentTimeMillis();
// depends on spark.shuffle.compress, shuffled block may not be compressed
- CompressedShuffleBlock rawBlock = shuffleReadClient.readShuffleBlockData();
+ CompressedShuffleBlock rawBlock = null;
Review Comment:
Why?
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/AbstractRssShuffleManagerBase.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.apache.spark.MapOutputTracker;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.SparkException;
+import org.apache.spark.shuffle.RssFeatureMatrix;
+import org.apache.spark.shuffle.SparkVersionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractRssShuffleManagerBase implements RssShuffleManagerBase {
Review Comment:
As a shareable class for spark2.x and spark3.x ?
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1145708865
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssFeatureMatrix.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle;
+
+public class RssFeatureMatrix {
Review Comment:
I'm ok to move it to `RssSparkShuffelUtils`. I originally created it for further extensions.
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkConfig.java:
##########
@@ -265,6 +266,17 @@ public class RssSparkConfig {
+ " spark.rss.estimate.server.assignment.enabled"))
.createWithDefault(RssClientConfig.RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE);
+ public static final ConfigEntry<Integer> RSS_SHUFFLE_MANAGER_GRPC_PORT = createIntegerBuilder(
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1146006858
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleManagerClientFactory.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.factory;
+
+import org.apache.uniffle.client.impl.grpc.ShuffleManagerGrpcClient;
+import org.apache.uniffle.common.ClientType;
+
+public class ShuffleManagerClientFactory {
+
+ private static final ShuffleManagerClientFactory INSTANCE = new ShuffleManagerClientFactory(ClientType.GRPC);
+
+ public static ShuffleManagerClientFactory getINSTANCE() {
+ return INSTANCE;
+ }
+
+ private ClientType clientType;
+
+ public ShuffleManagerClientFactory(ClientType clientType) {
+ this.clientType = clientType;
+ }
+
+ public ShuffleManagerGrpcClient createShuffleManagerClient(String host, int port) {
+ if (clientType.equals(ClientType.GRPC)) {
Review Comment:
Yes, I think this client is grpc only.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1150123276
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/AbstractRssShuffleManagerBase.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.apache.spark.MapOutputTracker;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.SparkException;
+import org.apache.spark.shuffle.RssFeatureMatrix;
+import org.apache.spark.shuffle.SparkVersionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractRssShuffleManagerBase implements RssShuffleManagerBase {
Review Comment:
Do you have any suggestions?
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1150140939
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/RssFeatureMatrix.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle;
+
+public class RssFeatureMatrix {
Review Comment:
It's not a concept of linear algebra.
Quote from the ChatGPT's answer, part of it is as following:
> A feature matrix is a table or grid that presents the features or attributes of a product/service in comparison to other products/services in the same category. It is used to compare the different versions or variations of a product/service across a set of predefined criteria or features.
So the RssFeatureMatrix defines different function/features against different spark versions.
--
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 #693: [WIP]feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#issuecomment-1459613036
@smallzhongfeng Could you help me review this pr?
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1149968215
##########
proto/src/main/proto/Rss.proto:
##########
@@ -488,4 +488,26 @@ message CancelDecommissionRequest {
message CancelDecommissionResponse {
StatusCode status = 1;
string retMsg = 2;
-}
\ No newline at end of file
+}
+// ShuffleManager service lives inside of compute-engine's application master, which handles rss shuffle specific logic
+// per application.
+service ShuffleManager {
+ rpc reportShuffleFetchFailure (ReportShuffleFetchFailureRequest) returns (ReportShuffleFetchFailureResponse);
+}
+
+message ReportShuffleFetchFailureRequest {
+ // appId normally should be omitted, it's used to avoid wrongly request issued from remaining executors of another
+ // app which accidentally has the same shuffle manager port with this app.
+ string appId = 1;
+ int32 shuffleId = 2;
+ int32 stageAttemptId = 3;
+ int32 partitionId = 4;
+ string exception = 5;
+ string serverId = 6;
Review Comment:
It was originally about the shuffle server id that has fetched failures. But it looks like it might be a bit hard to retrieve this information. Let me comment out this field and add it if necessary in the future.
--
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 #693: feat: support stage recompute for spark clients
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#issuecomment-1479765561
## [Codecov](https://codecov.io/gh/apache/incubator-uniffle/pull/693?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 [#693](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (448e846) into [master](https://codecov.io/gh/apache/incubator-uniffle/commit/7cb0b000cdf7260db3ac77f9d1742b8496c24cb2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7cb0b00) will **increase** coverage by `1.05%`.
> The diff coverage is `8.03%`.
```diff
@@ Coverage Diff @@
## master #693 +/- ##
============================================
+ Coverage 60.65% 61.70% +1.05%
- Complexity 1900 1903 +3
============================================
Files 239 232 -7
Lines 13029 11287 -1742
Branches 1091 1107 +16
============================================
- Hits 7903 6965 -938
+ Misses 4687 3943 -744
+ Partials 439 379 -60
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...ava/org/apache/spark/shuffle/RssFeatureMatrix.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS9Sc3NGZWF0dXJlTWF0cml4LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...org/apache/spark/shuffle/RssSparkShuffleUtils.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS9Sc3NTcGFya1NodWZmbGVVdGlscy5qYXZh) | `55.90% <0.00%> (-12.37%)` | :arrow_down: |
| [...va/org/apache/spark/shuffle/SparkVersionUtils.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS9TcGFya1ZlcnNpb25VdGlscy5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...park/shuffle/reader/ShuffleFetchFailedWrapper.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS9yZWFkZXIvU2h1ZmZsZUZldGNoRmFpbGVkV3JhcHBlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...shuffle/manager/AbstractRssShuffleManagerBase.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zaHVmZmxlL21hbmFnZXIvQWJzdHJhY3RSc3NTaHVmZmxlTWFuYWdlckJhc2UuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...fle/shuffle/manager/ShuffleManagerGrpcService.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zaHVmZmxlL21hbmFnZXIvU2h1ZmZsZU1hbmFnZXJHcnBjU2VydmljZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...e/shuffle/manager/ShuffleManagerServerFactory.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvdW5pZmZsZS9zaHVmZmxlL21hbmFnZXIvU2h1ZmZsZU1hbmFnZXJTZXJ2ZXJGYWN0b3J5LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...rg/apache/uniffle/client/util/RssClientConfig.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NsaWVudC91dGlsL1Jzc0NsaWVudENvbmZpZy5qYXZh) | `0.00% <ø> (ø)` | |
| [...java/org/apache/uniffle/common/rpc/GrpcServer.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS91bmlmZmxlL2NvbW1vbi9ycGMvR3JwY1NlcnZlci5qYXZh) | `20.89% <0.00%> (-0.65%)` | :arrow_down: |
| [.../java/org/apache/spark/shuffle/RssSparkConfig.java](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y2xpZW50LXNwYXJrL2NvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvc3Bhcmsvc2h1ZmZsZS9Sc3NTcGFya0NvbmZpZy5qYXZh) | `98.02% <100.00%> (+0.10%)` | :arrow_up: |
| ... and [4 more](https://codecov.io/gh/apache/incubator-uniffle/pull/693?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
... and [18 files with indirect coverage changes](https://codecov.io/gh/apache/incubator-uniffle/pull/693/indirect-changes?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] advancedxy commented on pull request #693: [#477] feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#issuecomment-1498443054
Closed in favor of #787
--
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] smallzhongfeng commented on a diff in pull request #693: [WIP]feat: support stage recompute for spark clients
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1133479194
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleManagerGrpcClient.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.impl.grpc;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.proto.RssProtos;
+import org.apache.uniffle.proto.ShuffleManagerGrpc;
+
+public class ShuffleManagerGrpcClient extends GrpcClient implements ShuffleManagerClient {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcClient.class);
+ private static final long RPC_TIMEOUT_DEFAULT_MS = 60000;
+ private long rpcTimeout = RPC_TIMEOUT_DEFAULT_MS;
+ private ShuffleManagerGrpc.ShuffleManagerBlockingStub blockingStub;
+
+ public ShuffleManagerGrpcClient(String host, int port) {
+ this(host, port, 3);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts) {
+ this(host, port, maxRetryAttempts, true);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext) {
+ super(host, port, maxRetryAttempts, usePlaintext);
+ blockingStub = ShuffleManagerGrpc.newBlockingStub(channel);
+ }
+
+ public ShuffleManagerGrpc.ShuffleManagerBlockingStub getBlockingStub() {
+ return blockingStub.withDeadlineAfter(rpcTimeout, TimeUnit.MILLISECONDS);
+ }
+
+ public String getDesc() {
+ return "Shuffle manager grpc client ref " + host + ":" + port;
+ }
+ @Override
+ public RssReportShuffleFetchFailureResponse reportShuffleFetchFailure(RssReportShuffleFetchFailureRequest request) {
+ RssProtos.ReportShuffleFetchFailureRequest protoRequest = request.toProto();
+ int retryNum = 0;
+ while (retryNum <= maxRetryAttempts) {
Review Comment:
I think the overall design idea is OK. I think you can continue. But I still want to ask @xianjingfeng‘s opinion.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1146007513
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/ShuffleManagerGrpcService.java:
##########
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Supplier;
+
+import com.google.common.collect.Maps;
+import io.grpc.stub.StreamObserver;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.proto.RssProtos;
+import org.apache.uniffle.proto.ShuffleManagerGrpc;
+
+public class ShuffleManagerGrpcService extends ShuffleManagerGrpc.ShuffleManagerImplBase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcService.class);
+ // todo: unregister shuffle status when shuffle is gced.
+ private final Map<Integer, RssShuffleStatus> shuffleStatus = Maps.newConcurrentMap();
Review Comment:
This is a todo.. Let me addressed it in a later commit.
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/AbstractRssShuffleManagerBase.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.apache.spark.MapOutputTracker;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.SparkException;
+import org.apache.spark.shuffle.RssFeatureMatrix;
+import org.apache.spark.shuffle.SparkVersionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractRssShuffleManagerBase implements RssShuffleManagerBase {
Review Comment:
yes.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1146003700
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/reader/ShuffleFetchFailedWrapper.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.reader;
+
+import java.io.IOException;
+
+import org.apache.spark.shuffle.FetchFailedException;
+import org.apache.spark.shuffle.RssSparkShuffleUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Product2;
+import scala.collection.AbstractIterator;
+import scala.collection.Iterator;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.config.RssConf;
+import org.apache.uniffle.common.exception.RssFetchFailedException;
+
+public class ShuffleFetchFailedWrapper {
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleFetchFailedWrapper.class);
+ private RssConf conf;
+ private String appId;
+ private int shuffleId;
+ private int partitionId;
+ private int stageAttemptId;
+ private String reportServerHost;
+ private int reportServerPort;
+
+ private ShuffleFetchFailedWrapper() {
+
+ }
+
+ static ShuffleFetchFailedWrapper newWrapper(RssConf conf) {
+ ShuffleFetchFailedWrapper wrapper = new ShuffleFetchFailedWrapper();
+ wrapper.conf = conf;
+ return wrapper;
+ }
+
+ ShuffleFetchFailedWrapper appId(String appId) {
+ this.appId = appId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper shuffleId(int shuffleId) {
+ this.shuffleId = shuffleId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper partitionId(int partitionId) {
+ this.partitionId = partitionId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper stageAttemptId(int stageAttemptId) {
+ this.stageAttemptId = stageAttemptId;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper reportServerHost(String host) {
+ this.reportServerHost = host;
+ return this;
+ }
+
+ ShuffleFetchFailedWrapper port(int port) {
+ this.reportServerPort = port;
+ return this;
+ }
+
+
+ public <K, C> Iterator<Product2<K, C>> wrap(Iterator<Product2<K, C>> iter) {
+ return new IteratorImpl<>(iter, this);
+ }
+
+ private static class IteratorImpl<K, C> extends AbstractIterator<Product2<K, C>> {
+ private Iterator<Product2<K, C>> iter;
+ private ShuffleFetchFailedWrapper wrapper;
+
+ IteratorImpl(Iterator<Product2<K, C>> iter, ShuffleFetchFailedWrapper wrapper) {
+ this.iter = iter;
+ this.wrapper = wrapper;
+ }
+
+ private RuntimeException generateFetchFailedIfNecessary(RssFetchFailedException e) {
+ String driver = wrapper.reportServerHost;
+ int port = wrapper.reportServerPort;
+ // todo: reuse this manager client if this is a bottleneck.
+ try (ShuffleManagerClient client = RssSparkShuffleUtils.createShuffleManagerClient(driver, port)) {
Review Comment:
Don't need to. The InterruptedException is only thrown during closing.
--
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] advancedxy closed pull request #693: [#477] feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy closed pull request #693: [#477] feat: support stage recompute for spark clients
URL: https://github.com/apache/incubator-uniffle/pull/693
--
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] advancedxy commented on pull request #693: [WIP]feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#issuecomment-1459255491
Hi, this is still working in progress. But I'd like to open this draft PR early and get some feedbacks before I'm going further.
@zuston @jerqi @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] jerqi commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1150132095
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/AbstractRssShuffleManagerBase.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.apache.spark.MapOutputTracker;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.SparkException;
+import org.apache.spark.shuffle.RssFeatureMatrix;
+import org.apache.spark.shuffle.SparkVersionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractRssShuffleManagerBase implements RssShuffleManagerBase {
Review Comment:
How about
`RssShuffleManagerBase` -> `RssShuffleManagerInterface`?
`AbstractRssShuffleManagerBase` -> `AbstractRssShuffleManager`?
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1150131617
##########
proto/src/main/proto/Rss.proto:
##########
@@ -488,4 +488,26 @@ message CancelDecommissionRequest {
message CancelDecommissionResponse {
StatusCode status = 1;
string retMsg = 2;
-}
\ No newline at end of file
+}
+// ShuffleManager service lives inside of compute-engine's application master, which handles rss shuffle specific logic
+// per application.
+service ShuffleManager {
+ rpc reportShuffleFetchFailure (ReportShuffleFetchFailureRequest) returns (ReportShuffleFetchFailureResponse);
+}
+
+message ReportShuffleFetchFailureRequest {
+ // appId normally should be omitted, it's used to avoid wrongly request issued from remaining executors of another
+ // app which accidentally has the same shuffle manager port with this app.
+ string appId = 1;
+ int32 shuffleId = 2;
+ int32 stageAttemptId = 3;
+ int32 partitionId = 4;
+ string exception = 5;
+ string serverId = 6;
Review Comment:
Good idea. But I believe only the id part is needed in this context. To reduce the data transferred from client to rss shuffle manager, I don't think we need other fields.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1146006105
##########
internal-client/src/main/java/org/apache/uniffle/client/factory/ShuffleManagerClientFactory.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.factory;
+
+import org.apache.uniffle.client.impl.grpc.ShuffleManagerGrpcClient;
+import org.apache.uniffle.common.ClientType;
+
+public class ShuffleManagerClientFactory {
+
+ private static final ShuffleManagerClientFactory INSTANCE = new ShuffleManagerClientFactory(ClientType.GRPC);
Review Comment:
This class is referred with other similar ClientFactory implementation, such as CoordinatorClientFactory.
We can refactor this in a following PR.
--
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] smallzhongfeng commented on a diff in pull request #693: [WIP]feat: support stage recompute for spark clients
Posted by "smallzhongfeng (via GitHub)" <gi...@apache.org>.
smallzhongfeng commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1130905917
##########
internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/ShuffleManagerGrpcClient.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.client.impl.grpc;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.exception.RssException;
+import org.apache.uniffle.proto.RssProtos;
+import org.apache.uniffle.proto.ShuffleManagerGrpc;
+
+public class ShuffleManagerGrpcClient extends GrpcClient implements ShuffleManagerClient {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ShuffleManagerGrpcClient.class);
+ private static final long RPC_TIMEOUT_DEFAULT_MS = 60000;
+ private long rpcTimeout = RPC_TIMEOUT_DEFAULT_MS;
+ private ShuffleManagerGrpc.ShuffleManagerBlockingStub blockingStub;
+
+ public ShuffleManagerGrpcClient(String host, int port) {
+ this(host, port, 3);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts) {
+ this(host, port, maxRetryAttempts, true);
+ }
+
+ public ShuffleManagerGrpcClient(String host, int port, int maxRetryAttempts, boolean usePlaintext) {
+ super(host, port, maxRetryAttempts, usePlaintext);
+ blockingStub = ShuffleManagerGrpc.newBlockingStub(channel);
+ }
+
+ public ShuffleManagerGrpc.ShuffleManagerBlockingStub getBlockingStub() {
+ return blockingStub.withDeadlineAfter(rpcTimeout, TimeUnit.MILLISECONDS);
+ }
+
+ public String getDesc() {
+ return "Shuffle manager grpc client ref " + host + ":" + port;
+ }
+ @Override
+ public RssReportShuffleFetchFailureResponse reportShuffleFetchFailure(RssReportShuffleFetchFailureRequest request) {
+ RssProtos.ReportShuffleFetchFailureRequest protoRequest = request.toProto();
+ int retryNum = 0;
+ while (retryNum <= maxRetryAttempts) {
Review Comment:
What about using `RetryUtils` ?
--
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] advancedxy commented on pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#issuecomment-1479708067
@jerqi @smallzhongfeng @zuston I believe this pr is almost ready for review..
I will add UTs and integration tests in the following commits. The overall procedure should be ready, you guys can have a look at this code and get the overall impression how stage recompute works.
Note: this pr itself maybe a bit large, I may split it into small chunks.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1145999898
##########
client/src/main/java/org/apache/uniffle/client/util/RssClientConfig.java:
##########
@@ -86,4 +86,6 @@ public class RssClientConfig {
public static final String RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER = "rss.estimate.task.concurrency.per.server";
public static final int RSS_ESTIMATE_TASK_CONCURRENCY_PER_SERVER_DEFAULT_VALUE = 80;
+ public static final String RSS_STAGE_RECOMPUTE = "rss.stage.reCompute";
Review Comment:
Agree, let me consider a better name..
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1146003088
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/reader/ShuffleFetchFailedWrapper.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.reader;
+
+import java.io.IOException;
+
+import org.apache.spark.shuffle.FetchFailedException;
+import org.apache.spark.shuffle.RssSparkShuffleUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.Product2;
+import scala.collection.AbstractIterator;
+import scala.collection.Iterator;
+
+import org.apache.uniffle.client.api.ShuffleManagerClient;
+import org.apache.uniffle.client.request.RssReportShuffleFetchFailureRequest;
+import org.apache.uniffle.client.response.RssReportShuffleFetchFailureResponse;
+import org.apache.uniffle.common.config.RssConf;
+import org.apache.uniffle.common.exception.RssFetchFailedException;
+
+public class ShuffleFetchFailedWrapper {
Review Comment:
Actually not. The iterator is a private internal class. The wrapper class creates the iterator.
--
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] advancedxy commented on a diff in pull request #693: feat: support stage recompute for spark clients
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1146009921
##########
client-spark/common/src/main/java/org/apache/spark/shuffle/reader/RssShuffleDataIterator.java:
##########
@@ -101,7 +101,8 @@ public boolean hasNext() {
// read next segment
long startFetch = System.currentTimeMillis();
// depends on spark.shuffle.compress, shuffled block may not be compressed
- CompressedShuffleBlock rawBlock = shuffleReadClient.readShuffleBlockData();
+ CompressedShuffleBlock rawBlock = null;
Review Comment:
This is not intentional.. This change is introduced during a code refactor. let me remove it in a following commit.
--
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 #693: feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1149969105
##########
client-spark/common/src/main/java/org/apache/uniffle/shuffle/manager/AbstractRssShuffleManagerBase.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.uniffle.shuffle.manager;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.Optional;
+
+import org.apache.spark.MapOutputTracker;
+import org.apache.spark.MapOutputTrackerMaster;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.SparkException;
+import org.apache.spark.shuffle.RssFeatureMatrix;
+import org.apache.spark.shuffle.SparkVersionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractRssShuffleManagerBase implements RssShuffleManagerBase {
Review Comment:
Em .... could we give a better name for this class?
--
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 #693: feat: support stage recompute for spark clients
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #693:
URL: https://github.com/apache/incubator-uniffle/pull/693#discussion_r1150155834
##########
proto/src/main/proto/Rss.proto:
##########
@@ -488,4 +488,26 @@ message CancelDecommissionRequest {
message CancelDecommissionResponse {
StatusCode status = 1;
string retMsg = 2;
-}
\ No newline at end of file
+}
+// ShuffleManager service lives inside of compute-engine's application master, which handles rss shuffle specific logic
+// per application.
+service ShuffleManager {
+ rpc reportShuffleFetchFailure (ReportShuffleFetchFailureRequest) returns (ReportShuffleFetchFailureResponse);
+}
+
+message ReportShuffleFetchFailureRequest {
+ // appId normally should be omitted, it's used to avoid wrongly request issued from remaining executors of another
+ // app which accidentally has the same shuffle manager port with this app.
+ string appId = 1;
+ int32 shuffleId = 2;
+ int32 stageAttemptId = 3;
+ int32 partitionId = 4;
+ string exception = 5;
+ string serverId = 6;
Review Comment:
It's difficult for us to change old protocol field. If it won't bring too much cost, I prefer that we have a unify style.
--
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