[CELEBORN-2406] Avoid blocking GetReducerFileGroup RPC under ConcurrentHashMap bin lock in updateFileGroup - #3776
Closed
venkata91 wants to merge 3 commits into
Closed
Conversation
venkata91
changed the title
…in updateFileGroup updateFileGroup loaded the file group via reduceFileGroupsMap.compute(), which holds ConcurrentHashMap's per-key bin lock for the whole remapping function. So the first reduce task of a shuffle held the bin lock while doing the blocking GetReducerFileGroup round-trip to the LifecycleManager, and every other reduce task for that shuffleId blocked on the same bin lock, even cache hits. On a large shuffle this convoyed the entire stage for minutes with no fetch activity. Move the RPC off the bin lock: a lock-free get() fast path for cache hits, and a per-shuffle monitor (double-checked) that serializes cold loads to a single RPC. Caching semantics are unchanged (a failed load caches null and reloads next call). Flink overrides updateFileGroup and is unaffected; the Spark client uses the base ShuffleClientImpl and was exposed. Standalone convoy fix extracted from apache#3687 (closed), independent of the broader partition-range metadata optimization in apache#3745 (CELEBORN-2370). Co-authored-by: Chao Sun <sunchao@apache.org>
venkata91
force-pushed
the
celeborn-updatefilegroup-convoy-fix
branch
from
August 4, 2026 16:03
6be0d76 to
29572d8
Compare
venkata91
marked this pull request as ready for review
August 4, 2026 16:09
There was a problem hiding this comment.
Pull request overview
This PR refactors ShuffleClientImpl.updateFileGroup to prevent a blocking GetReducerFileGroup RPC from running under a ConcurrentHashMap.compute per-key/bin lock, which can cause thread convoys at high reduce parallelism. The change introduces a lock-free cache-hit fast path and serializes cold loads with a dedicated per-shuffle monitor so only one RPC is in flight per shuffle.
Changes:
- Replace
reduceFileGroupsMap.compute(shuffleId, ...)with a lock-freeget()fast path plus double-checked synchronized cold-load path. - Add a per-shuffle lock map (
fileGroupLoadLocks) and clear it duringcleanupShuffle. - Add a concurrency test asserting concurrent first-time loads issue exactly one RPC and complete within a timeout.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java | Moves blocking reducer file-group RPC out from under ConcurrentHashMap.compute by using a per-shuffle monitor and double-checking cached state. |
| client/src/test/java/org/apache/celeborn/client/ShuffleClientSuiteJ.java | Adds a multi-threaded regression test to verify concurrent loads deduplicate to a single RPC. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
RexXiong
approved these changes
Aug 5, 2026
RexXiong
left a comment
RexXiong
left a comment
Contributor
There was a problem hiding this comment.
LGTM — but please address Copilot's suggestions. 👍
Author
Thanks for the review! Will do. |
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Author
|
@RexXiong Addressed the copilot's suggestions. Thanks! |
Author
|
Looks like some transient CI failures. Can you please trigger it? I don't have permissions looks like. |
Member
|
@venkata91, please create jira ticket for this pull request and update the title of this pull request. |
venkata91
changed the title
Author
@SteNicholas Created a JIRA ticket and updated the PR title. Thanks! |
zaynt4606
pushed a commit
that referenced
this pull request
Aug 10, 2026
…ntHashMap bin lock in updateFileGroup ### What changes were proposed in this pull request? `ShuffleClientImpl.updateFileGroup` loaded the reducer file group via `reduceFileGroupsMap.compute(shuffleId, ...)`. `ConcurrentHashMap.compute` holds the per-key bin lock for the entire remapping function, so the blocking `GetReducerFileGroup` RPC to the `LifecycleManager` ran while holding that lock. At high reduce parallelism the first task of a shuffle takes the bin lock and issues the RPC; every other reduce task for the same `shuffleId` then blocks on that bin lock, even cache hits. On a large shuffle the stage sits idle for minutes with reduce threads BLOCKED in `updateFileGroup` and no fetch activity, i.e. a load convoy. This PR moves the RPC out from under the bin lock: - Lock-free `get()` fast path, so cache hits never take a lock. - Cold loads serialize on a dedicated per-shuffle monitor (`fileGroupLoadLocks`, cleared in `cleanupShuffle`) with a double-check, so only one RPC is issued per shuffle. - Caching semantics unchanged: a cached tuple whose `_1()` is null still reloads. `FlinkShuffleClientImpl` overrides `updateFileGroup` and is unaffected; the Spark plugin uses the base `ShuffleClientImpl` and was exposed. ### Why are the changes needed? Observed in production (tens of thousands of reduce partitions, ~900 executors): a stage stalled over 20 minutes with reduce threads blocked in `updateFileGroup` -> `ConcurrentHashMap.compute`. A slow RPC under the bin lock turns a transient lookup into a multi-minute wedge. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? New `ShuffleClientSuiteJ#testUpdateReducerFileGroupConcurrentLoadIssuesSingleRpc`: 16 threads concurrently load the same shuffle against a mock RPC that blocks 500ms; asserts all callers finish and exactly one RPC is issued. Closes #3776 from venkata91/celeborn-updatefilegroup-convoy-fix. Lead-authored-by: Venkata Krishnan Sowrirajan <vsowrirajan@microsoft.com> Co-authored-by: Venkata krishnan Sowrirajan <vsowrirajan@linkedin.com> Signed-off-by: zhengtao <shuaizhentao.szt@alibaba-inc.com> (cherry picked from commit d1d69cb) Signed-off-by: zhengtao <shuaizhentao.szt@alibaba-inc.com> AI-Contributed/Feature: 0/28 AI-Contributed/UT: 0/54
Contributor
|
Thanks. Merged to main(v1.0.0) and branch-0.7. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
ShuffleClientImpl.updateFileGrouploaded the reducer file group viareduceFileGroupsMap.compute(shuffleId, ...).ConcurrentHashMap.computeholds the per-key bin lock for the entire remapping function, so the blockingGetReducerFileGroupRPC to theLifecycleManagerran while holding that lock.At high reduce parallelism the first task of a shuffle takes the bin lock and issues the RPC; every other reduce task for the same
shuffleIdthen blocks on that bin lock, even cache hits. On a large shuffle the stage sits idle for minutes with reduce threads BLOCKED inupdateFileGroupand no fetch activity, i.e. a load convoy.This PR moves the RPC out from under the bin lock:
get()fast path, so cache hits never take a lock.fileGroupLoadLocks, cleared incleanupShuffle) with a double-check, so only one RPC is issued per shuffle._1()is null still reloads.FlinkShuffleClientImploverridesupdateFileGroupand is unaffected; the Spark plugin uses the baseShuffleClientImpland was exposed.Why are the changes needed?
Observed in production (tens of thousands of reduce partitions, ~900 executors): a stage stalled over 20 minutes with reduce threads blocked in
updateFileGroup->ConcurrentHashMap.compute. A slow RPC under the bin lock turns a transient lookup into a multi-minute wedge.Does this PR introduce any user-facing change?
No.
How was this patch tested?
New
ShuffleClientSuiteJ#testUpdateReducerFileGroupConcurrentLoadIssuesSingleRpc: 16 threads concurrently load the same shuffle against a mock RPC that blocks 500ms; asserts all callers finish and exactly one RPC is issued.