[SPARK-26632][Core] Separate Thread Configurations of Driver and Executor

core/src/main/scala/org/apache/spark/network/netty/SparkTransportConf.scala

@@ -36,16 +36,30 @@ object SparkTransportConf {
    * @param numUsableCores if nonzero, this will restrict the server and client threads to only
    *                       use the given number of cores, rather than all of the machine's cores.
    *                       This restriction will only occur if these properties are not already set.
+   * @param role           optional role, could be driver, executor, worker and master. Default is
+   *                      [[None]], means no role specific configurations.
    */
-  def fromSparkConf(_conf: SparkConf, module: String, numUsableCores: Int = 0): TransportConf = {
+  def fromSparkConf(
+      _conf: SparkConf,
+      module: String,
+      numUsableCores: Int = 0,
+      role: Option[String] = None): TransportConf = {
     val conf = _conf.clone
-
-    // Specify thread configuration based on our JVM's allocation of cores (rather than necessarily
-    // assuming we have all the machine's cores).
-    // NB: Only set if serverThreads/clientThreads not already set.
+    // specify default thread configuration based on our JVM's allocation of cores (rather than
+    // necessarily assuming we have all the machine's cores).
     val numThreads = NettyUtils.defaultNumThreads(numUsableCores)
-    conf.setIfMissing(s"spark.$module.io.serverThreads", numThreads.toString)
-    conf.setIfMissing(s"spark.$module.io.clientThreads", numThreads.toString)
+    // module threads configuration
+    val (modServerThreads, modClientThreads) =
+      (conf.getInt(s"spark.$module.io.serverThreads", numThreads),
+       conf.getInt(s"spark.$module.io.clientThreads", numThreads))
+    // override threads configurations with role specific values if specified
+    val (serverThreads, clientThreads) = role match {
+      case Some(r) => (conf.getInt(s"spark.$r.$module.io.serverThreads", modServerThreads),
+        conf.getInt(s"spark.$r.$module.io.clientThreads", modClientThreads))
+      case None => (modServerThreads, modClientThreads)
+    }
+    conf.set(s"spark.$module.io.serverThreads", serverThreads.toString)
+    conf.set(s"spark.$module.io.clientThreads", clientThreads.toString)
 
     new TransportConf(module, new ConfigProvider {
       override def get(name: String): String = conf.get(name)

core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala

@@ -24,7 +24,7 @@ import scala.collection.JavaConverters._
 import scala.concurrent.Promise
 import scala.util.control.NonFatal
 
-import org.apache.spark.SparkException
+import org.apache.spark.{SparkConf, SparkContext, SparkException}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.Network.RPC_NETTY_DISPATCHER_NUM_THREADS
 import org.apache.spark.network.client.RpcResponseCallback
@@ -194,12 +194,32 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) exte
     endpoints.containsKey(name)
   }
 
-  /** Thread pool used for dispatching messages. */
-  private val threadpool: ThreadPoolExecutor = {
+  def getNumOfThreads(conf: SparkConf): Int = {
     val availableCores =
       if (numUsableCores > 0) numUsableCores else Runtime.getRuntime.availableProcessors()
-    val numThreads = nettyEnv.conf.get(RPC_NETTY_DISPATCHER_NUM_THREADS)
+    // module configuration
+    val modNumThreads = nettyEnv.conf.get(RPC_NETTY_DISPATCHER_NUM_THREADS)
       .getOrElse(math.max(2, availableCores))
+    // try to get specific threads configurations of driver and executor
+    // override module configurations if specified
+    val executorId = conf.get("spark.executor.id", "")
+    // neither driver nor executor if executor id is not set
+    val role = executorId match {
+      case "" => ""
+      case SparkContext.DRIVER_IDENTIFIER => "driver"
+      // any other non-empty values since executor must has "spark.executor.id" set
+      case _ => "executor"
+    }
+    if (role.isEmpty) {
+      modNumThreads
+    } else {
+      conf.getInt(s"spark.$role.rpc.netty.dispatcher.numThreads", modNumThreads)
+    }
+  }
+
+  /** Thread pool used for dispatching messages. */
+  private val threadpool: ThreadPoolExecutor = {
+    val numThreads = getNumOfThreads(nettyEnv.conf)
     val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, "dispatcher-event-loop")
     for (i <- 0 until numThreads) {
       pool.execute(new MessageLoop)

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

@@ -29,7 +29,7 @@ import scala.reflect.ClassTag
 import scala.util.{DynamicVariable, Failure, Success, Try}
 import scala.util.control.NonFatal
 
-import org.apache.spark.{SecurityManager, SparkConf}
+import org.apache.spark.{SecurityManager, SparkConf, SparkContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.config.Network._
 import org.apache.spark.network.TransportContext
@@ -47,11 +47,21 @@ private[netty] class NettyRpcEnv(
     host: String,
     securityManager: SecurityManager,
     numUsableCores: Int) extends RpcEnv(conf) with Logging {
+  // try to get specific threads configurations of driver and executor
+  val executorId = conf.get("spark.executor.id", "")
+  // neither driver nor executor if executor id is not set
+  val role = executorId match {
+    case "" => None
+    case SparkContext.DRIVER_IDENTIFIER => Some("driver")
+    // any other non-empty values since executor must has "spark.executor.id" set
+    case _ => Some("executor")
+  }
 
   private[netty] val transportConf = SparkTransportConf.fromSparkConf(
     conf.clone.set(RPC_IO_NUM_CONNECTIONS_PER_PEER, 1),
     "rpc",
-    conf.get(RPC_IO_THREADS).getOrElse(numUsableCores))
+    conf.get(RPC_IO_THREADS).getOrElse(numUsableCores),
+    role)
 
   private val dispatcher: Dispatcher = new Dispatcher(this, numUsableCores)
 

project/MimaExcludes.scala

@@ -44,6 +44,9 @@ object MimaExcludes {
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.SparkListenerApplicationStart.copy"),
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.SparkListenerApplicationStart.this"),
     ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.scheduler.SparkListenerApplicationStart$"),
+
+    // [SPARK-26632][Core] Separate Thread Configurations of Driver and Executor
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.network.netty.SparkTransportConf.fromSparkConf"),
 
     // [SPARK-25765][ML] Add training cost to BisectingKMeans summary
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.mllib.clustering.BisectingKMeansModel.this"),