Copilot commented on code in PR #3716:
URL: https://github.com/apache/celeborn/pull/3716#discussion_r3341206619
##########
client/src/main/java/org/apache/celeborn/client/DummyShuffleClient.java:
##########
@@ -88,10 +99,40 @@ public int pushData(
int numMappers,
int numPartitions)
throws IOException {
+ String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
+ Map<Integer, List<byte[]>> partitionData =
+ pushDataByMapKey.computeIfAbsent(mapKey, k -> new HashMap<>());
+ partitionData
+ .computeIfAbsent(partitionId, k -> new ArrayList<>())
+ .add(Arrays.copyOfRange(data, offset, offset + length));
Review Comment:
`pushDataByMapKey` is a `ConcurrentHashMap`, but the per-mapKey value is
currently a plain `HashMap` and the per-partition value is a plain `ArrayList`.
Since `DummyShuffleClient.pushData()` can be invoked from the async
`DataPusher` thread while other threads update/read the same structures, this
can race (lost updates / `ConcurrentModificationException`) and make the new
integrity tests flaky. Use a concurrent map for the per-mapKey container and a
thread-safe list for the per-partition batches.
##########
client/src/main/java/org/apache/celeborn/client/DummyShuffleClient.java:
##########
@@ -88,10 +99,40 @@ public int pushData(
int numMappers,
int numPartitions)
throws IOException {
+ String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
+ Map<Integer, List<byte[]>> partitionData =
+ pushDataByMapKey.computeIfAbsent(mapKey, k -> new HashMap<>());
+ partitionData
+ .computeIfAbsent(partitionId, k -> new ArrayList<>())
+ .add(Arrays.copyOfRange(data, offset, offset + length));
+
os.write(data, offset, length);
return length;
}
+ @Override
+ public void computeBatchCRC(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ byte[] data,
+ int offset,
+ int length) {
+ if (!shuffleIntegrityCheckEnabled) {
+ return;
+ }
+ String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
+ PushState pushState = pushStateMap.computeIfAbsent(mapKey, k -> new
PushState(conf));
+ pushState.addDataWithOffsetAndLength(partitionId, data, offset, length);
+
+ Map<Integer, List<byte[]>> partitionData =
+ crcDataByMapKey.computeIfAbsent(mapKey, k -> new HashMap<>());
+ partitionData
+ .computeIfAbsent(partitionId, k -> new ArrayList<>())
+ .add(Arrays.copyOfRange(data, offset, offset + length));
+ }
Review Comment:
Same thread-safety issue as in `pushData()`: `crcDataByMapKey` stores a
non-thread-safe `HashMap` of `ArrayList`s, but `computeBatchCRC()` is called on
the writer thread while `pushData()` runs on the async push thread for the same
`mapKey`. This can corrupt the tracking structures and cause intermittent test
failures. Use `ConcurrentHashMap` + a thread-safe list here too.
##########
client/src/main/java/org/apache/celeborn/client/DummyShuffleClient.java:
##########
@@ -104,6 +145,13 @@ public int mergeData(
int numMappers,
int numPartitions)
throws IOException {
+ String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
+ Map<Integer, List<byte[]>> partitionData =
+ pushDataByMapKey.computeIfAbsent(mapKey, k -> new HashMap<>());
+ partitionData
+ .computeIfAbsent(partitionId, k -> new ArrayList<>())
+ .add(Arrays.copyOfRange(data, offset, offset + length));
Review Comment:
`mergeData()` records into `pushDataByMapKey` using a non-thread-safe
`HashMap`/`ArrayList` combo. Since merges can happen concurrently with async
pushes for the same `mapKey`, this tracking can race and lead to flaky tests.
Use concurrent structures as in `pushData()`.
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1404,6 +1398,26 @@ public int mergeData(
false);
}
+ @Override
+ public void computeBatchCRC(
+ int shuffleId,
+ int mapId,
+ int attemptId,
+ int partitionId,
+ byte[] data,
+ int offset,
+ int length) {
+ if (!shuffleIntegrityCheckEnabled) {
+ return;
+ }
+ final String mapKey = Utils.makeMapKey(shuffleId, mapId, attemptId);
+ if (mapperEnded(shuffleId, mapId) || isStageEnded(shuffleId)) {
+ return;
+ }
Review Comment:
`mapperEnded(shuffleId, mapId)` already returns `true` when
`isStageEnded(shuffleId)` is `true`, so the additional `||
isStageEnded(shuffleId)` check here is redundant. Dropping it avoids duplicated
logic and makes the early-return condition clearer.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
