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

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -31,6 +31,7 @@ import org.apache.spark.TaskState.TaskState
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.deploy.worker.WorkerWatcher
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.EXECUTOR_ID
 import org.apache.spark.rpc._
 import org.apache.spark.scheduler.{ExecutorLossReason, TaskDescription}
 import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
@@ -239,6 +240,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
         SparkHadoopUtil.get.addDelegationTokens(tokens, driverConf)
       }
 
+      driverConf.set(EXECUTOR_ID, arguments.executorId)
       val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId, arguments.hostname,
         arguments.cores, cfg.ioEncryptionKey, isLocal = false)
 

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

@@ -36,16 +36,26 @@ 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)
+    // override threads configurations with role specific values if specified
+    // config order is role > module > default
+    Seq("serverThreads", "clientThreads").foreach { suffix =>
+      val value = role.flatMap { r => conf.getOption(s"spark.$r.$module.io.$suffix") }
+        .getOrElse(
+          conf.get(s"spark.$module.io.$suffix", numThreads.toString))
+      conf.set(s"spark.$module.io.$suffix", value)
+    }
 
     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,8 +24,9 @@ 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.EXECUTOR_ID
 import org.apache.spark.internal.config.Network.RPC_NETTY_DISPATCHER_NUM_THREADS
 import org.apache.spark.network.client.RpcResponseCallback
 import org.apache.spark.rpc._
@@ -194,12 +195,22 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) exte
     endpoints.containsKey(name)
   }
 
-  /** Thread pool used for dispatching messages. */
-  private val threadpool: ThreadPoolExecutor = {
+  private 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)
+
+    val modNumThreads = conf.get(RPC_NETTY_DISPATCHER_NUM_THREADS)
       .getOrElse(math.max(2, availableCores))
+
+    conf.get(EXECUTOR_ID).map { id =>
+      val role = if (id == SparkContext.DRIVER_IDENTIFIER) "driver" else "executor"
+      conf.getInt(s"spark.$role.rpc.netty.dispatcher.numThreads", modNumThreads)
+    }.getOrElse(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,8 +29,9 @@ 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.EXECUTOR_ID
 import org.apache.spark.internal.config.Network._
 import org.apache.spark.network.TransportContext
 import org.apache.spark.network.client._
@@ -47,11 +48,15 @@ private[netty] class NettyRpcEnv(
     host: String,
     securityManager: SecurityManager,
     numUsableCores: Int) extends RpcEnv(conf) with Logging {
+  val role = conf.get(EXECUTOR_ID).map { id =>
+    if (id == SparkContext.DRIVER_IDENTIFIER) Some("driver") else Some("executor")
+  }.getOrElse(None)
 
   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)
 

core/src/test/scala/org/apache/spark/network/netty/SparkTransportConfSuite.scala

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.netty
+
+import org.scalatest.Matchers
+import org.scalatest.mockito.MockitoSugar
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.network.util.NettyUtils
+
+class SparkTransportConfSuite extends SparkFunSuite with MockitoSugar with Matchers{
+  val module = "rpc"
+  val serThreads = "serverThreads"
+  val cliThreads = "clientThreads"
+
+  test("default value is get when neither role nor module is set") {
+    val numUsableCores = 4
+    val conf = new SparkConf()
+    val sparkTransportConf = SparkTransportConf.fromSparkConf(conf, module, numUsableCores, None)
+    val expected = NettyUtils.defaultNumThreads(numUsableCores)
+    val serActual = sparkTransportConf.get(s"spark.$module.io.$serThreads", "")
+    val cliActual = sparkTransportConf.get(s"spark.$module.io.$cliThreads", "")
+    serActual should equal(expected.toString)
+    cliActual should equal(expected.toString)
+  }
+
+  test("module value is get when role is not set") {
+    val numUsableCores = 3
+    val serExpected = "7"
+    val cliExpected = "5"
+    val conf = new SparkConf()
+      .set(s"spark.$module.io.$serThreads", serExpected)
+      .set(s"spark.$module.io.$cliThreads", cliExpected)
+    val sparkTransportConf = SparkTransportConf.fromSparkConf(conf, module, numUsableCores, None)
+    val serActual = sparkTransportConf.get(s"spark.$module.io.$serThreads", "")
+    val cliActual = sparkTransportConf.get(s"spark.$module.io.$cliThreads", "")
+    serActual should equal(serExpected)
+    cliActual should equal(cliExpected)
+  }
+
+  test("role value is get when role is set") {
+    val role = Some("driver")
+    val numUsableCores = 10
+    val serModule = "7"
+    val cliModule = "5"
+    val serExpected = "8"
+    val cliExpected = "6"
+    val conf = new SparkConf()
+      .set(s"spark.$module.io.$serThreads", serModule)
+      .set(s"spark.$module.io.$cliThreads", cliModule)
+      .set(s"spark.${role.get}.$module.io.$serThreads", serExpected)
+      .set(s"spark.${role.get}.$module.io.$cliThreads", cliExpected)
+    val sparkTransportConf = SparkTransportConf.fromSparkConf(conf, module, numUsableCores, role)
+    val serActual = sparkTransportConf.get(s"spark.$module.io.$serThreads", "")
+    val cliActual = sparkTransportConf.get(s"spark.$module.io.$cliThreads", "")
+    serActual should equal(serExpected)
+    cliActual should equal(cliExpected)
+  }
+
+  test("module value is get when role other than mine is set") {
+    val role = Some("driver")
+    val otherRole = "executor"
+    val numUsableCores = 10
+    val serExpected = "7"
+    val cliExpected = "5"
+    val serRole = "8"
+    val cliRole = "6"
+    val conf = new SparkConf()
+      .set(s"spark.$module.io.$serThreads", serExpected)
+      .set(s"spark.$module.io.$cliThreads", cliExpected)
+      .set(s"spark.$otherRole.$module.io.$serThreads", serRole)
+      .set(s"spark.$otherRole.$module.io.$cliThreads", cliRole)
+    val sparkTransportConf = SparkTransportConf.fromSparkConf(conf, module, numUsableCores, role)
+    val serActual = sparkTransportConf.get(s"spark.$module.io.$serThreads", "")
+    val cliActual = sparkTransportConf.get(s"spark.$module.io.$cliThreads", "")
+    serActual should equal(serExpected)
+    cliActual should equal(cliExpected)
+  }
+}

docs/configuration.md

@@ -1954,6 +1954,46 @@ Apart from these, the following properties are also available, and may be useful
 </tr>
 </table>
 
+### Thread Configurations
+
+Depending on jobs and cluster configurations, we can set number of threads in several places in Spark to utilize 
+available resources efficiently to get better performance. Prior to Spark 3.0, these thread configurations apply 
+to all roles of Spark, such as driver, executor, worker and master. From Spark 3.0, we can configure threads in 
+finer granularity starting from driver and executor. Take RPC module as example in below table. For other modules,
+like shuffle, just replace "rpc" with "shuffle" in the property names except 
+spark.{driver|executor}.rpc.netty.dispatcher.numThreads, which is only for RPC module.
+
+<table class="table">
+<tr><th>Property Name</th><th>Default</th><th>Meaning</th></tr>
+<tr>
+  <td>spark.{driver|executor}.rpc.io.serverThreads</td>
+  <td>
+    Fall back on spark.rpc.io.serverThreads
+  </td>
+  <td>Number of threads used in the server thread pool</td>
+</tr>
+<tr>
+  <td>spark.{driver|executor}.rpc.io.clientThreads</td>
+  <td>
+    Fall back on spark.rpc.io.clientThreads
+  </td>
+  <td>Number of threads used in the client thread pool</td>
+</tr>
+<tr>
+  <td>spark.{driver|executor}.rpc.netty.dispatcher.numThreads</td>
+  <td>
+    Fall back on spark.rpc.netty.dispatcher.numThreads
+  </td>
+  <td>Number of threads used in RPC message dispatcher thread pool</td>
+</tr>
+</table>
+
+The default values of spark.rpc.io.serverThreads, spark.rpc.io.clientThreads and spark.rpc.netty.dispatcher.numThreads
+are same. It's <br>
+number of CPU cores if specified. Otherwise, the available processors to the JVM. In either cases, the default value 
+is limited to <code>org.apache.spark.network.util.NettyUtils.MAX_DEFAULT_NETTY_THREADS</code>.
+
+
 ### Security
 
 Please refer to the [Security](security.html) page for available options on how to secure different

project/MimaExcludes.scala

@@ -75,6 +75,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"),

resource-managers/mesos/src/main/scala/org/apache/spark/executor/MesosExecutorBackend.scala

@@ -29,6 +29,7 @@ import org.apache.spark.{SparkConf, SparkEnv, TaskState}
 import org.apache.spark.TaskState
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.EXECUTOR_ID
 import org.apache.spark.scheduler.TaskDescription
 import org.apache.spark.scheduler.cluster.mesos.MesosSchedulerUtils
 import org.apache.spark.util.Utils
@@ -74,6 +75,7 @@ private[spark] class MesosExecutorBackend
     val properties = Utils.deserialize[Array[(String, String)]](executorInfo.getData.toByteArray) ++
       Seq[(String, String)](("spark.app.id", frameworkInfo.getId.getValue))
     val conf = new SparkConf(loadDefaults = true).setAll(properties)
+    conf.set(EXECUTOR_ID, executorId)
     val env = SparkEnv.createExecutorEnv(
       conf, executorId, slaveInfo.getHostname, cpusPerTask, None, isLocal = false)