Skip to content

[CELEBORN-2243][FOLLOWUP][Spark-2] During the close phase of hashWriter, pushData and mergeData are sent in parallel #3584

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mcdull-zhang wants to merge 2 commits into from
+2 −5
Closed

[CELEBORN-2243][FOLLOWUP][Spark-2] During the close phase of hashWriter, pushData and mergeData are sent in parallel #3584

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e33e22c
fix spark-2
Jan 12, 2026
4e909f9
fix
Jan 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 2 additions & 5 deletions ...spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/HashBasedShuffleWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,11 +332,6 @@ private void cleanupPusher() throws IOException {
}

private void close() throws IOException, InterruptedException {
// here we wait for all the in-flight batches to return which sent by dataPusher thread
dataPusher.waitOnTermination();
sendBufferPool.returnPushTaskQueue(dataPusher.getAndResetIdleQueue());
shuffleClient.prepareForMergeData(shuffleId, mapId, encodedAttemptId);

// merge and push residual data to reduce network traffic
// NB: since dataPusher thread have no in-flight data at this point,
// we now push merged data by task thread will not introduce any contention
Expand Down Expand Up @@ -369,6 +364,8 @@ private void close() throws IOException, InterruptedException {
sendOffsets = null;

long waitStartTime = System.nanoTime();
dataPusher.waitOnTermination();
sendBufferPool.returnPushTaskQueue(dataPusher.getAndResetIdleQueue());
shuffleClient.mapperEnd(shuffleId, mapId, encodedAttemptId, numMappers, numPartitions);
writeMetrics.incWriteTime(System.nanoTime() - waitStartTime);

Expand Down
Loading