[incubator-celeborn] branch main updated: [CELEBORN-774] Pullout celeborn.rpc.dispatcher.threads to CelebornConf

[ch...@apache.org]

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

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


The following commit(s) were added to refs/heads/main by this push:
     new 4f8e72f21 [CELEBORN-774] Pullout celeborn.rpc.dispatcher.threads to CelebornConf
4f8e72f21 is described below

commit 4f8e72f217e30ba547370c0ffcb5f52dd09aad85
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Thu Jul 6 16:23:32 2023 +0800

    [CELEBORN-774] Pullout celeborn.rpc.dispatcher.threads to CelebornConf
    
    ### What changes were proposed in this pull request?
    
    Pullout hardcoded `celeborn.rpc.dispatcher.numThreads` to `CelebornConf` and rename it to `celeborn.rpc.dispatcher.threads` to align with existing configuration style
    
    ### Why are the changes needed?
    
    Pullout inline configuration to `CelebornConf`, and expose it in configuration docs
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    Pass GA.
    
    Closes #1684 from pan3793/CELEBORN-774.
    
    Authored-by: Cheng Pan <ch...@apache.org>
    Signed-off-by: Cheng Pan <ch...@apache.org>
---
 .../scala/org/apache/celeborn/common/CelebornConf.scala    | 11 +++++++++++
 .../org/apache/celeborn/common/rpc/netty/Dispatcher.scala  | 14 ++++++--------
 docs/configuration/network.md                              |  1 +
 3 files changed, 18 insertions(+), 8 deletions(-)

diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index ef2eb44fd..84262b5cb 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -381,6 +381,8 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
     new RpcTimeout(get(RPC_LOOKUP_TIMEOUT).milli, RPC_LOOKUP_TIMEOUT.key)
   def rpcAskTimeout: RpcTimeout =
     new RpcTimeout(get(RPC_ASK_TIMEOUT).milli, RPC_ASK_TIMEOUT.key)
+  def rpcDispatcherNumThreads(availableCores: Int): Int =
+    get(RPC_DISPATCHER_THREADS).getOrElse(availableCores)
   def networkIoMode(module: String): String = {
     val key = NETWORK_IO_MODE.key.replace("<module>", module)
     get(key, NETWORK_IO_MODE.defaultValue.get)
@@ -1280,6 +1282,15 @@ object CelebornConf extends Logging {
       .timeConf(TimeUnit.MILLISECONDS)
       .createWithDefaultString("30s")
 
+  val RPC_DISPATCHER_THREADS: OptionalConfigEntry[Int] =
+    buildConf("celeborn.rpc.dispatcher.threads")
+      .withAlternative("celeborn.rpc.dispatcher.numThreads")
+      .categories("network")
+      .doc("Threads number of message dispatcher event loop")
+      .version("0.3.0")
+      .intConf
+      .createOptional
+
   val NETWORK_IO_MODE: ConfigEntry[String] =
     buildConf("celeborn.<module>.io.mode")
       .categories("network")
diff --git a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala
index da597c05f..9ea65aef3 100644
--- a/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/rpc/netty/Dispatcher.scala
@@ -115,11 +115,9 @@ private[celeborn] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) e
       postMessage(
         name,
         message,
-        (e) => {
-          e match {
-            case e: RpcEnvStoppedException => logDebug(s"Message $message dropped. ${e.getMessage}")
-            case e: Throwable => logWarning(s"Message $message dropped. ${e.getMessage}")
-          }
+        {
+          case e: RpcEnvStoppedException => logDebug(s"Message $message dropped. ${e.getMessage}")
+          case e: Throwable => logWarning(s"Message $message dropped. ${e.getMessage}")
         })
     }
   }
@@ -145,7 +143,7 @@ private[celeborn] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) e
     postMessage(
       message.receiver.name,
       OneWayMessage(message.senderAddress, message.content),
-      (e) => throw e)
+      e => throw e)
   }
 
   /**
@@ -158,7 +156,7 @@ private[celeborn] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) e
   def postMessage(
       endpointName: String,
       message: InboxMessage,
-      callbackIfStopped: (Exception) => Unit): Unit = {
+      callbackIfStopped: Exception => Unit): Unit = {
     val error = synchronized {
       val data = endpoints.get(endpointName)
       if (stopped) {
@@ -205,7 +203,7 @@ private[celeborn] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) e
     val availableCores =
       if (numUsableCores > 0) numUsableCores
       else Math.max(16, Runtime.getRuntime.availableProcessors())
-    val numThreads = nettyEnv.conf.getInt("celeborn.rpc.dispatcher.numThreads", availableCores)
+    val numThreads = nettyEnv.conf.rpcDispatcherNumThreads(availableCores)
     val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, "dispatcher-event-loop")
     logInfo(s"Dispatcher numThreads: $numThreads")
     for (i <- 0 until numThreads) {
diff --git a/docs/configuration/network.md b/docs/configuration/network.md
index 690ad518a..6cde65c5b 100644
--- a/docs/configuration/network.md
+++ b/docs/configuration/network.md
@@ -46,6 +46,7 @@ license: |
 | celeborn.port.maxRetries | 1 | When port is occupied, we will retry for max retry times. | 0.2.0 | 
 | celeborn.rpc.askTimeout | 30s | Timeout for RPC ask operations. It's recommended to set at least `240s` when `HDFS` is enabled in `celeborn.storage.activeTypes` | 0.2.0 | 
 | celeborn.rpc.connect.threads | 64 |  | 0.2.0 | 
+| celeborn.rpc.dispatcher.threads | &lt;undefined&gt; | Threads number of message dispatcher event loop | 0.3.0 | 
 | celeborn.rpc.io.threads | &lt;undefined&gt; | Netty IO thread number of NettyRpcEnv to handle RPC request. The default threads number is the number of runtime available processors. | 0.2.0 | 
 | celeborn.rpc.lookupTimeout | 30s | Timeout for RPC lookup operations. | 0.2.0 | 
 | celeborn.shuffle.io.maxChunksBeingTransferred | 9223372036854775807 | The max number of chunks allowed to be transferred at the same time on shuffle service. Note that new incoming connections will be closed when the max number is hit. The client will retry according to the shuffle retry configs (see `celeborn.<module>.io.maxRetries` and `celeborn.<module>.io.retryWait`), if those limits are reached the task will fail with fetch failure. | 0.2.0 |