You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by re...@apache.org on 2023/03/29 02:23:17 UTC
[incubator-celeborn] 02/42: [CELEBORN-390][FLINK] Refine synchronization in FlinkShuffleClientImpl#updateFileGroup (#1320)
This is an automated email from the ASF dual-hosted git repository.
rexxiong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit 824122ce7623833af77804565c3f0cdd3733642d
Author: Keyong Zhou <zh...@apache.org>
AuthorDate: Thu Mar 9 16:49:18 2023 +0800
[CELEBORN-390][FLINK] Refine synchronization in FlinkShuffleClientImpl#updateFileGroup (#1320)
---
.../flink/readclient/FlinkShuffleClientImpl.java | 39 ++++++++++++----------
.../apache/celeborn/client/ShuffleClientImpl.java | 12 +++++--
2 files changed, 31 insertions(+), 20 deletions(-)
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
index ea0b6e88b..336c98a58 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/FlinkShuffleClientImpl.java
@@ -136,26 +136,31 @@ public class FlinkShuffleClientImpl extends ShuffleClientImpl {
String applicationId, String shuffleKey, int shuffleId, int partitionId) throws IOException {
ReduceFileGroups reduceFileGroups =
reduceFileGroupsMap.computeIfAbsent(shuffleId, (id) -> new ReduceFileGroups());
- synchronized (reduceFileGroups) {
- if (reduceFileGroups.partitionIds != null
- && reduceFileGroups.partitionIds.contains(partitionId)) {
- logger.debug(
- "use cached file groups for partition: {}",
- Utils.makeReducerKey(applicationId, shuffleId, partitionId));
- return reduceFileGroups;
- } else {
- // refresh file groups
- ReduceFileGroups newGroups = loadFileGroupInternal(applicationId, shuffleKey, shuffleId);
- if (newGroups == null || !newGroups.partitionIds.contains(partitionId)) {
- throw new IOException(
- "shuffle data lost for partition: "
- + Utils.makeReducerKey(applicationId, shuffleId, partitionId));
+ if (reduceFileGroups.partitionIds != null
+ && reduceFileGroups.partitionIds.contains(partitionId)) {
+ logger.debug(
+ "use cached file groups for partition: {}",
+ Utils.makeReducerKey(applicationId, shuffleId, partitionId));
+ } else {
+ synchronized (reduceFileGroups) {
+ if (reduceFileGroups.partitionIds != null
+ && reduceFileGroups.partitionIds.contains(partitionId)) {
+ logger.debug(
+ "use cached file groups for partition: {}",
+ Utils.makeReducerKey(applicationId, shuffleId, partitionId));
+ } else {
+ // refresh file groups
+ ReduceFileGroups newGroups = loadFileGroupInternal(applicationId, shuffleKey, shuffleId);
+ if (newGroups == null || !newGroups.partitionIds.contains(partitionId)) {
+ throw new IOException(
+ "shuffle data lost for partition: "
+ + Utils.makeReducerKey(applicationId, shuffleId, partitionId));
+ }
+ reduceFileGroups.update(newGroups);
}
-
- reduceFileGroupsMap.put(shuffleId, newGroups);
- return newGroups;
}
}
+ return reduceFileGroups;
}
public ReadClientHandler getReadClientHandler() {
diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
index 72f7ed8c0..20fc78284 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -111,9 +111,9 @@ public class ShuffleClientImpl extends ShuffleClient {
};
protected static class ReduceFileGroups {
- public final Map<Integer, Set<PartitionLocation>> partitionGroups;
- public final int[] mapAttempts;
- public final Set<Integer> partitionIds;
+ public Map<Integer, Set<PartitionLocation>> partitionGroups;
+ public int[] mapAttempts;
+ public Set<Integer> partitionIds;
ReduceFileGroups(
Map<Integer, Set<PartitionLocation>> partitionGroups,
@@ -129,6 +129,12 @@ public class ShuffleClientImpl extends ShuffleClient {
this.mapAttempts = null;
this.partitionIds = null;
}
+
+ public void update(ReduceFileGroups fileGroups) {
+ partitionGroups = fileGroups.partitionGroups;
+ mapAttempts = fileGroups.mapAttempts;
+ partitionIds = fileGroups.partitionIds;
+ }
}
// key: shuffleId