FMX commented on code in PR #2462:
URL: https://github.com/apache/celeborn/pull/2462#discussion_r1574219186
##########
client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala:
##########
@@ -151,15 +160,21 @@ class ChangePartitionManager(
oldPartition,
cause)
- requests.synchronized {
- if (requests.containsKey(partitionId)) {
- requests.get(partitionId).add(changePartition)
+ locks(partitionId % locks.length).synchronized {
Review Comment:
@CodingCat I think the "partition Id" of different shuffles can be
repeated. The lock is for the same "shuffleId" in the previous implementation
but the lock can be contended by the same "partition Id" of different stages in
your new implementation. Although a spark application won't run too many stages
concurrently, but the spark thrift server might run many stages.
The locks variable can be changed to avoid the lock contention of different
stages.
`
private val locks = JavaUtils.newConcurrentHashMap[Int,Array[AnyRef]]()
`
I think creating an array of AnyRef won't cost more than the contended
locks. 256 AnyRef objects will consume 2 kb of memory, this suggestion won't
introduce memory pressure.
--
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]
