Copilot commented on code in PR #3670:
URL: https://github.com/apache/celeborn/pull/3670#discussion_r3232406483
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1404,6 +1398,23 @@ 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);
+ PushState pushState = getPushState(mapKey);
+ pushState.addDataWithOffsetAndLength(partitionId, data, offset, length);
+ }
Review Comment:
This change moves CRC accumulation onto the writer thread, while
`pushOrMergeData()` (running in the async `DataPusher` thread) still mutates
the same `PushState` (e.g., batch id progression and potentially other
bookkeeping). That introduces real cross-thread access that didn’t exist when
CRC accumulation lived solely in `pushOrMergeData()`. To prevent races and
inconsistent per-partition CRC/bytes, ensure
`PushState.addDataWithOffsetAndLength(...)` and the async-thread mutations are
coordinated (e.g., internal synchronization in `PushState`, a shared lock
around all `PushState` updates, or separate thread-confined structures merged
deterministically).
##########
client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java:
##########
@@ -1404,6 +1398,23 @@ 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);
+ PushState pushState = getPushState(mapKey);
Review Comment:
`computeBatchCRC()` allocates/derives `mapKey` on every batch CRC update.
With CRC now computed at multiple writer-side flush points, this can become a
hot-path allocation cost. Consider reducing repeated `mapKey` construction
(e.g., cache `mapKey` in the writer/pusher and pass it through, or provide an
overload taking a `PushState`/precomputed key), while keeping the current API
available if you need it for compatibility.
--
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]
