xumingming commented on PR #3670:
URL: https://github.com/apache/celeborn/pull/3670#issuecomment-4495526184
@SteNicholas Made corresponding changes, and the following are the detailed
responses to Claude's comments:
> - ⚠️ attemptId encoding consistency. Hash writers pass encodedAttemptId;
SortBasedPusher passes
attemptNumber. computeBatchCRC does getPushState(makeMapKey(shuffleId,
mapId, attemptId)). If that
key doesn't match the key used by the actual push/commit path for the
same batch, CRC_M is
accumulated into a different PushState than the one read at commit →
integrity check silently
broken or always-mismatching. Verify both encodings resolve to the same
PushState (high priority).
This seems a mis-read. All the call site are using encodedAttemptId. The
current code does use attemptNumber and encodedAttemptId in difference places,
but it is just a naming issue, and not introduced by this PR.
> - ⚠️ Thread-safety / visibility. Accumulation moves from the DataPusher
thread to the writer
thread, while PushState.addDataWithOffsetAndLength was previously
single-threaded on the pusher.
Confirm: (a) SortBasedPusher.pushData can't run concurrently with another
path mutating the same
PushState (e.g., spill-triggered flush), and (b) there is a happens-before
between writer-thread
accumulation and the commit-time getCRC32PerPartition read (likely OK
since the writer also drives
mapperEnd, but it should be explicit/confirmed).
(a). Only the Spark task thread writes; the DataPusher thread only reads.
Spill is a no-op in SortBasedPusher, **there is no cross-thread visibility
issue.**
(b). In the new implementation, both the writer-thread and the read-thread
are the same spark task thread, **so there is a nature happens-before.**
> Test coverage — thin for the risk profile
Added e2e tests in CelebornShuffleWriterSuiteBase which covers the following
code path:
- Giant path: records > buffer size → pushGiantRecord → computeBatchCRC +
pushData
- Normal flush path: buffer fills up → computeBatchCRC +
dataPusher.addTask/mergeData
- Close residual path: remaining bytes at writer.stop() → computeBatchCRC +
mergeData
--
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]
