venkata91 opened a new pull request, #3776:
URL: https://github.com/apache/celeborn/pull/3776

   ### What changes were proposed in this pull request?
   
   `ShuffleClientImpl.updateFileGroup` issued the blocking 
`GetReducerFileGroup` RPC inside `reduceFileGroupsMap.compute(shuffleId, ...)`. 
`ConcurrentHashMap.compute` holds the per-key bin lock for the whole remapping 
function, so the RPC round-trip to the `LifecycleManager` ran while holding 
that lock.
   
   This moves the RPC out from under the bin lock: a lock-free `get()` fast 
path so cache hits take no lock, and first-time loads serialize on a dedicated 
per-shuffle monitor (`fileGroupLoadLocks`, cleared in `cleanupShuffle`) with a 
double-check so still only one RPC is issued per shuffle. Caching semantics are 
unchanged: a cached tuple whose `_1()` is `null` (failed load) still triggers a 
reload, exactly as the old `compute()` branch did.
   
   `FlinkShuffleClientImpl` already overrides `updateFileGroup` and is 
unaffected; the Spark plugin uses the base `ShuffleClientImpl` directly 
(`ShuffleClient.get()` -> `new ShuffleClientImpl(...)`, via 
`CelebornShuffleReader`), so it was exposed. This is the standalone convoy fix 
extracted from #3687 (closed), independent of the partition-range metadata 
optimization in #3745 (CELEBORN-2370) that supersedes it. Credit to @sunchao 
for #3687.
   
   ### Why are the changes needed?
   
   At high reduce parallelism, the first reduce task takes the bin lock and 
fires `GetReducerFileGroup` (seconds under load for a large shuffle); every 
other reduce task for the same `shuffleId` then blocks on that bin lock inside 
`compute`, even though they are cache hits. The result is a thundering-herd 
convoy: thousands of reduce tasks BLOCKED on `ConcurrentHashMap.compute` with 
no fetch activity and no fetch failures, so the stage appears frozen with 
nothing on fetch dashboards. Because a failed load is cached as a `null` tuple 
and `CelebornShuffleReader` retries until stage end, a slow RPC re-forms the 
convoy each retry, turning a transient stall into a multi-minute wedge.
   
   Observed in production (tens of thousands of reduce partitions, ~900 
executors): a stage sat idle 20+ min; the driver thread dump showed reduce 
threads BLOCKED in `updateFileGroup` -> `ConcurrentHashMap.compute`.
   
   ### Does this PR introduce _any_ user-facing change?
   
   No.
   
   ### How was this patch tested?
   
   New unit test 
`ShuffleClientSuiteJ#testUpdateReducerFileGroupConcurrentLoadIssuesSingleRpc`: 
16 threads call `updateFileGroup` for one shuffle against a mock RPC that 
blocks 500ms; asserts all callers complete and exactly one RPC is issued. 
Existing `ShuffleClientSuiteJ` passes 14/14 (incl. interrupt, 
non-fetch-failure, timeout paths). `build/mvn -pl client -am test 
-Dtest=ShuffleClientSuiteJ` and `spotless:check` green.
   


-- 
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]

Reply via email to