(celeborn) branch main updated: [CELEBORN-2345] Fix allocation for rpcAskTimeout

[nicholasjiang]

This is an automated email from the ASF dual-hosted git repository.

SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new c546a21fbb [CELEBORN-2345] Fix allocation for rpcAskTimeout
c546a21fbb is described below

commit c546a21fbbddd559f28966cf0b82d48116b984f5
Author: Sanskar Modi <sanskarmod...@gmail.com>
AuthorDate: Mon Jun 1 14:55:05 2026 +0800

    [CELEBORN-2345] Fix allocation for rpcAskTimeout
    
    ### What changes were proposed in this pull request?
    
    - Reuse the `rpcAskTimeout` variable in LifecycleManager
    - Make  `rpcAskTimeout` and `rpcRetryWait` lazy in `RpcEndpointRef`. Since 
many places are using `RpcEndpointRef` for just endpoint name. Examples are all 
the callers using `Dispatcher.postMessage`
    
    ### Why are the changes needed?
    
    `rpcAskTimeout` and `rpcRetryWait` are causing 4% of total allocations.
    
    <img width="644" height="379" alt="Screenshot 2026-05-29 at 3 23 53 PM" 
src="https://github.com/user-attachments/assets/d9d3d858-b56a-42c7-bdfb-cbf93cb18d07";
 />
    
    ### Does this PR resolve a correctness bug?
    
    - [ ] Yes
    
    ### Does this PR introduce _any_ user-facing change?
    
    - [ ] Yes
    
    ### How was this patch tested?
    
    Minor change
    
    Closes #3712 from s0nskar/fix_conf_allocation.
    
    Authored-by: Sanskar Modi <sanskarmod...@gmail.com>
    Signed-off-by: SteNicholas <programg...@163.com>
---
 .../main/scala/org/apache/celeborn/client/LifecycleManager.scala   | 7 ++++---
 .../main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala | 4 ++--
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git 
a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala 
b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
index a37513a236..d8c48420dd 100644
--- a/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala
@@ -118,6 +118,7 @@ class LifecycleManager(val appUniqueId: String, val conf: 
CelebornConf) extends
   private val rpcCacheConcurrencyLevel = conf.clientRpcCacheConcurrencyLevel
   private val rpcCacheExpireTime = conf.clientRpcCacheExpireTime
   private val rpcMaxRetires = conf.clientRpcMaxRetries
+  private val rpcAskTimeoutMs = conf.rpcAskTimeout.duration.toMillis
 
   private val batchRemoveExpiredShufflesEnabled = 
conf.batchHandleRemoveExpiredShufflesEnabled
 
@@ -575,7 +576,7 @@ class LifecycleManager(val appUniqueId: String, val conf: 
CelebornConf) extends
       futures.add((future, workerInfo))
     }
 
-    var timeout = conf.rpcAskTimeout.duration.toMillis
+    var timeout = rpcAskTimeoutMs
     val delta = 50
     while (timeout > 0 && !futures.isEmpty) {
       val iter = futures.iterator
@@ -1334,7 +1335,7 @@ class LifecycleManager(val appUniqueId: String, val conf: 
CelebornConf) extends
     val futureSeq = Future.sequence(outFutures)(cbf, ec)
     awaitResult(futureSeq, Duration.Inf)
 
-    var timeout = conf.rpcAskTimeout.duration.toMillis
+    var timeout = rpcAskTimeoutMs
     val delta = 50
     while (timeout >= 0 && !futures.isEmpty) {
       val iter = futures.iterator()
@@ -1721,7 +1722,7 @@ class LifecycleManager(val appUniqueId: String, val conf: 
CelebornConf) extends
         futures.add(DestroyFutureWithStatus(future, destroy, 
workerInfo.endpoint, 1, startTime))
     }
 
-    val timeout = conf.rpcAskTimeout.duration.toMillis
+    val timeout = rpcAskTimeoutMs
     var remainingTime = timeout * rpcMaxRetires
     val delta = 50
     while (remainingTime > 0 && !futures.isEmpty) {
diff --git 
a/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala 
b/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala
index 483144bdd3..f1f86c1b3d 100644
--- a/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpointRef.scala
@@ -30,8 +30,8 @@ import org.apache.celeborn.common.util.Utils
 abstract class RpcEndpointRef(conf: CelebornConf)
   extends Serializable with Logging {
 
-  private[this] val defaultAskTimeout = conf.rpcAskTimeout
-  private[this] val defaultRetryWait = conf.rpcRetryWait
+  private[this] lazy val defaultAskTimeout = conf.rpcAskTimeout
+  private[this] lazy val defaultRetryWait = conf.rpcRetryWait
 
   /**
    * return the address for the [[RpcEndpointRef]]