You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "yew1eb (via GitHub)" <gi...@apache.org> on 2026/01/19 16:07:37 UTC
[PR] [CELEBORN #2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
yew1eb opened a new pull request, #3586:
URL: https://github.com/apache/celeborn/pull/3586
### What changes were proposed in this pull request?
Add `shuffleIdLocks` (fine-grained locks per shuffleId), replace global `shuffleMapperAttempts` lock in `initMapperAttempts` and `finishMapperAttempt`.
### Why are the changes needed?
High concurrency causes lock contention on `shuffleMapperAttempts` in `finishMapperAttempt`, leading to abnormally long shuffle write time for small queries in Kyuubi Shared Mode. Fine-grained locks eliminate cross-shuffle blocking and improve concurrency.
### Does this PR resolve a correctness bug?
No.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
CI.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN #2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #3586:
URL: https://github.com/apache/celeborn/pull/3586#discussion_r2730828888
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -77,6 +77,8 @@ class ReducePartitionCommitHandler(
private val shuffleMapperAttempts = JavaUtils.newConcurrentHashMap[Int, Array[Int]]()
// TODO: Move this to native Int -> Int Map
private val shuffleToCompletedMappers = JavaUtils.newConcurrentHashMap[Int, Int]()
+ private val shuffleIdLocks = JavaUtils.newConcurrentHashMap[Int, Object]()
Review Comment:
Better remove the lock object when the shuffle expired.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN #2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on code in PR #3586:
URL: https://github.com/apache/celeborn/pull/3586#discussion_r2730874204
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -77,6 +77,8 @@ class ReducePartitionCommitHandler(
private val shuffleMapperAttempts = JavaUtils.newConcurrentHashMap[Int, Array[Int]]()
// TODO: Move this to native Int -> Int Map
private val shuffleToCompletedMappers = JavaUtils.newConcurrentHashMap[Int, Int]()
+ private val shuffleIdLocks = JavaUtils.newConcurrentHashMap[Int, Object]()
Review Comment:
updated. PTAL
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN #2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on code in PR #3586:
URL: https://github.com/apache/celeborn/pull/3586#discussion_r2731754354
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -314,7 +317,8 @@ class ReducePartitionCommitHandler(
numPartitions: Int,
crc32PerPartition: Array[Int],
bytesWrittenPerPartition: Array[Long]): (Boolean, Boolean) = {
- val (mapperAttemptFinishedSuccess, allMapperFinished) = shuffleMapperAttempts.synchronized {
+ val shuffleLock = shuffleIdLocks.computeIfAbsent(shuffleId, _ => new Object())
Review Comment:
updated
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -428,7 +432,8 @@ class ReducePartitionCommitHandler(
}
private def initMapperAttempts(shuffleId: Int, numMappers: Int, numPartitions: Int): Unit = {
- shuffleMapperAttempts.synchronized {
+ val shuffleLock = shuffleIdLocks.computeIfAbsent(shuffleId, _ => new Object())
Review Comment:
updated
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3586:
URL: https://github.com/apache/celeborn/pull/3586#issuecomment-3805169132
Merged to main(v0.7.0).
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN #2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3586:
URL: https://github.com/apache/celeborn/pull/3586#discussion_r2731674192
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -314,7 +317,8 @@ class ReducePartitionCommitHandler(
numPartitions: Int,
crc32PerPartition: Array[Int],
bytesWrittenPerPartition: Array[Long]): (Boolean, Boolean) = {
- val (mapperAttemptFinishedSuccess, allMapperFinished) = shuffleMapperAttempts.synchronized {
+ val shuffleLock = shuffleIdLocks.computeIfAbsent(shuffleId, _ => new Object())
Review Comment:
`shuffleIdLocks.computeIfAbsent(shuffleId, _ => new Object())` relies on Scala SAM-conversion of a Scala lambda to `java.util.function.Function`, which is not available in Scala 2.11 without the `FunctionConverter` implicits; since this project still cross-builds for 2.11 (see `CelebornBuild.scala`), this will fail to compile for that Scala version. To keep 2.11 compatibility, either import `org.apache.celeborn.common.util.FunctionConverter._` in this file (as done in `CommitManager.scala`) or replace the lambda with an explicit `java.util.function.Function[Int, Object]` implementation.
```suggestion
val shuffleLock = shuffleIdLocks.computeIfAbsent(
shuffleId,
new function.Function[Int, Object]() {
override def apply(id: Int): Object = new Object()
})
```
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -428,7 +432,8 @@ class ReducePartitionCommitHandler(
}
private def initMapperAttempts(shuffleId: Int, numMappers: Int, numPartitions: Int): Unit = {
- shuffleMapperAttempts.synchronized {
+ val shuffleLock = shuffleIdLocks.computeIfAbsent(shuffleId, _ => new Object())
Review Comment:
Same issue as in `finishMapperAttempt`: using the Scala lambda `_ => new Object()` as the second argument to `shuffleIdLocks.computeIfAbsent` depends on SAM-conversion that is not available under Scala 2.11 without `FunctionConverter` implicits, causing compilation failures for that cross-build. Please either add `import org.apache.celeborn.common.util.FunctionConverter._` to this file or construct an explicit `java.util.function.Function[Int, Object]` instance here.
```suggestion
val shuffleLock = shuffleIdLocks.computeIfAbsent(
shuffleId,
new java.util.function.Function[Int, Object] {
override def apply(key: Int): Object = new Object()
})
```
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN #2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3586:
URL: https://github.com/apache/celeborn/pull/3586#discussion_r2731715760
##########
client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala:
##########
@@ -314,7 +317,8 @@ class ReducePartitionCommitHandler(
numPartitions: Int,
crc32PerPartition: Array[Int],
bytesWrittenPerPartition: Array[Long]): (Boolean, Boolean) = {
- val (mapperAttemptFinishedSuccess, allMapperFinished) = shuffleMapperAttempts.synchronized {
+ val shuffleLock = shuffleIdLocks.computeIfAbsent(shuffleId, _ => new Object())
Review Comment:
@yew1eb, PTAL.
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3586: [CELEBORN-2250] Fix lock contention in ReducePartitionCommitHandler.finishMapperAttempt via fine-grained locks
URL: https://github.com/apache/celeborn/pull/3586
--
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: issues-unsubscribe@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org