modidy numa patch for Spark3.0 and delelte oap-common dependency

jikunshang · Aug 2, 2020 · 4888064 · 4888064
1 parent fdec769
commit 4888064
Show file tree

Hide file tree

Showing 9 changed files with 53 additions and 86 deletions.
diff --git a/core/pom.xml b/core/pom.xml
@@ -35,10 +35,6 @@
   </properties>
 
   <dependencies>
-    <dependency>
-      <groupId>com.intel</groupId>
-      <artifactId>oap-common</artifactId>
-    </dependency>
     <dependency>
       <groupId>com.thoughtworks.paranamer</groupId>
       <artifactId>paranamer</artifactId>

diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -179,7 +179,6 @@ object SparkEnv extends Logging {
     create(
       conf,
       SparkContext.DRIVER_IDENTIFIER,
-      None,
       bindAddress,
       advertiseAddress,
       Option(port),
@@ -198,7 +197,6 @@ object SparkEnv extends Logging {
   private[spark] def createExecutorEnv(
       conf: SparkConf,
       executorId: String,
-      numaNodeId: Option[String],
       bindAddress: String,
       hostname: String,
       numCores: Int,
@@ -207,7 +205,6 @@ object SparkEnv extends Logging {
     val env = create(
       conf,
       executorId,
-      numaNodeId,
       bindAddress,
       hostname,
       None,
@@ -222,12 +219,11 @@ object SparkEnv extends Logging {
   private[spark] def createExecutorEnv(
       conf: SparkConf,
       executorId: String,
-      numaNodeId: Option[String],
       hostname: String,
       numCores: Int,
       ioEncryptionKey: Option[Array[Byte]],
       isLocal: Boolean): SparkEnv = {
-    createExecutorEnv(conf, executorId, numaNodeId, hostname,
+    createExecutorEnv(conf, executorId, hostname,
       hostname, numCores, ioEncryptionKey, isLocal)
   }
 
@@ -237,7 +233,6 @@ object SparkEnv extends Logging {
   private def create(
       conf: SparkConf,
       executorId: String,
-      numaNodeId: Option[String],
       bindAddress: String,
       advertiseAddress: String,
       port: Option[Int],

diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -330,9 +330,10 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       }
 
       driverConf.set(EXECUTOR_ID, arguments.executorId)
-      val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId, arguments.numaNodeId,
+      val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId,
         arguments.bindAddress, arguments.hostname, arguments.cores, cfg.ioEncryptionKey,
         isLocal = false)
+      SparkEnv.get.conf.set("spark.executor.numa.id", s"${arguments.numaNodeId.getOrElse(-1)}")
 
       env.rpcEnv.setupEndpoint("Executor",
         backendCreateFn(env.rpcEnv, arguments, env, cfg.resourceProfile))

diff --git a/pom.xml b/pom.xml
@@ -110,7 +110,6 @@
   </modules>
 
   <properties>
-    <oap.common.version>0.8.0</oap.common.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
     <java.version>1.8</java.version>
@@ -348,11 +347,6 @@
   </dependencies>
   <dependencyManagement>
     <dependencies>
-      <dependency>
-        <groupId>com.intel</groupId>
-        <artifactId>oap-common</artifactId>
-        <version>${oap.common.version}</version>
-      </dependency>
       <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-tags_${scala.binary.version}</artifactId>

diff --git a/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala b/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala
@@ -204,21 +204,18 @@ private[yarn] class ExecutorRunnable(
 
     val numaEnabled = sparkConf.get(SPARK_YARN_NUMA_ENABLED)
 
+    logInfo(s"[NUMACHECK] numaEnabled $numaEnabled executorId $executorId")
     // Don't need numa binding for driver.
-    val numaCtlCommand = if (numaEnabled && executorId != "<executorId>" && numaNodeId.nonEmpty) {
+    val (numaCtlCommand, numaNodeOpts) = if (numaEnabled && executorId != "<executorId>"
+      && numaNodeId.nonEmpty) {
+      logInfo(s"numaNodeId ${numaNodeId.get}")
       val command = s"numactl --cpubind=${numaNodeId.get} --membind=${numaNodeId.get} "
-      command
+      (command, Seq("--numa-node-id", numaNodeId.get.toString))
     } else {
-      ""
-    }
-
-    val numaNodeOpts = if (executorId != "<executorId>" && numaNodeId.nonEmpty) {
-      val numanode = Seq("--numa-node-id", numaNodeId.get.toString)
-      numanode
-    } else {
-      Nil
+      ("", Nil)
     }
 
+    logInfo(s"[NUMACHECK] numactl command $numaCtlCommand")
     YarnSparkHadoopUtil.addOutOfMemoryErrorArgument(javaOpts)
     val commands = prefixEnv ++
       Seq(numaCtlCommand  + Environment.JAVA_HOME.$$() + "/bin/java", "-server") ++
@@ -236,6 +233,7 @@ private[yarn] class ExecutorRunnable(
         s"1>${ApplicationConstants.LOG_DIR_EXPANSION_VAR}/stdout",
         s"2>${ApplicationConstants.LOG_DIR_EXPANSION_VAR}/stderr")
 
+    logInfo(s"[NUMACHECK] container command $commands")
     // TODO: it would be nicer to just make sure there are no null commands here
     commands.map(s => if (s == null) "null" else s).toList
   }

diff --git a/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala b/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
@@ -176,8 +176,6 @@ private[yarn] class YarnAllocator(
   private[yarn] val containerPlacementStrategy =
     new LocalityPreferredContainerPlacementStrategy(sparkConf, conf, resource, resolver)
 
-  private[yarn] val numaManager = new NumaManager(sparkConf)
-
   def getNumExecutorsRunning: Int = runningExecutors.size()
 
   def getNumReleasedContainers: Int = releasedContainers.size()
@@ -186,6 +184,14 @@ private[yarn] class YarnAllocator(
 
   def isAllNodeBlacklisted: Boolean = allocatorBlacklistTracker.isAllNodeBlacklisted
 
+  // The total number of numa node
+  private[yarn] val totalNumaNumber = sparkConf.get(SPARK_YARN_NUMA_NUMBER)
+  // Mapping from host to executor counter
+  private[yarn] case class NumaInfo(cotainer2numa: mutable.HashMap[String, Int],
+                                    numaUsed: Array[Int])
+
+  private[yarn] val hostToNumaInfo = new mutable.HashMap[String, NumaInfo]()
+
   /**
    * A sequence of pending container requests that have not yet been fulfilled.
    */
@@ -534,12 +540,22 @@ private[yarn] class YarnAllocator(
     for (container <- containersToUse) {
       executorIdCounter += 1
       val executorHostname = container.getNodeId.getHost
+      // Setting the numa id that the executor should binding.
+      // new numaid binding method
+      val numaInfo = hostToNumaInfo.getOrElseUpdate(executorHostname,
+        NumaInfo(new mutable.HashMap[String, Int], new Array[Int](totalNumaNumber)))
+      val minUsed = numaInfo.numaUsed.min
+      val newNumaNodeId = numaInfo.numaUsed.indexOf(minUsed)
+      numaInfo.cotainer2numa.put(container.getId.toString, newNumaNodeId)
+      numaInfo.numaUsed(newNumaNodeId) += 1
+
+      val numaNodeId = newNumaNodeId.toString
+      logInfo(s"numaNodeId: $numaNodeId on host $executorHostname," +
+        "container: " + container.getId.toString +
+        ", minUsed: " + minUsed)
+
       val containerId = container.getId
       val executorId = executorIdCounter.toString
-
-      // Set the numa id that the executor should binding.
-      val numaNodeId = numaManager.assignNumaId(containerId, executorHostname)
-
       assert(container.getResource.getMemory >= resource.getMemory)
       logInfo(s"Launching container $containerId on host $executorHostname " +
         s"for executor with ID $executorId with numa ID $numaNodeId")
@@ -567,7 +583,7 @@ private[yarn] class YarnAllocator(
                 sparkConf,
                 driverUrl,
                 executorId,
-                numaNodeId,
+                Some(numaNodeId),
                 executorHostname,
                 executorMemory,
                 executorCores,
@@ -626,6 +642,17 @@ private[yarn] class YarnAllocator(
         // there are some exit status' we shouldn't necessarily count against us, but for
         // now I think its ok as none of the containers are expected to exit.
         val exitStatus = completedContainer.getExitStatus
+
+        var numaNodeId = -1
+        val hostName = hostOpt.getOrElse("nohost")
+        val numaInfoOp = hostToNumaInfo.get(hostName)
+        numaInfoOp match {
+          case Some(numaInfo) =>
+            numaNodeId = numaInfo.cotainer2numa.get(containerId.toString).getOrElse(-1)
+            if(-1 != numaNodeId) numaInfo.numaUsed(numaNodeId) -= 1
+          case _ => numaNodeId = -1
+        }
+
         val (exitCausedByApp, containerExitReason) = exitStatus match {
           case ContainerExitStatus.SUCCESS =>
             (false, s"Executor for container $containerId exited because of a YARN event (e.g., " +
@@ -696,7 +723,6 @@ private[yarn] class YarnAllocator(
         }
 
         allocatedContainerToHostMap.remove(containerId)
-        numaManager.releaseNuma(containerId, host)
       }
 
       containerIdToExecutorId.remove(containerId).foreach { eid =>
@@ -791,51 +817,6 @@ private[yarn] class YarnAllocator(
 
 }
 
-// scalastyle:off
-// Manage how to bind numa with an exector. No matter numa binding turn on/off
-// we should assign a numa id since Persistent Memory always be associate with a numa id
-private[yarn] class NumaManager(sparkConf: SparkConf) extends Logging {
-  private final val totalNumaNode = sparkConf.get(SPARK_YARN_NUMA_NUM)
-  private val hostToNumaIds = new ConcurrentHashMap[String, Array[Int]]()
-  private val hostToContainers = new ConcurrentHashMap[String, mutable.HashMap[ContainerId, String]]()
-
-  def assignNumaId(
-                    containerId: ContainerId,
-                    executorHostName: String): Option[String] = {
-    if (totalNumaNode == 0) return None
-    if (hostToContainers.containsKey(executorHostName)) {
-      if (!hostToContainers.get(executorHostName).contains(containerId)) {
-        this.synchronized {
-          val numaIds = hostToNumaIds.get(executorHostName)
-          val v = numaIds.zipWithIndex.min._2
-          logDebug(s"bind $containerId with $v on host $executorHostName")
-          hostToContainers.get(executorHostName) += (containerId -> v.toString)
-          numaIds(v) += 1
-          Some(v.toString)
-        }
-      } else {
-        hostToContainers.get(executorHostName).get(containerId)
-      }
-    } else {
-      logDebug(s"bind $containerId with 0 on host $executorHostName")
-      hostToNumaIds.put(executorHostName, Array.fill[Int](totalNumaNode)(0))
-      hostToNumaIds.get(executorHostName)(0) += 1
-      hostToContainers.putIfAbsent(executorHostName, mutable.HashMap[ContainerId, String](containerId -> "0"))
-      Some("0")
-    }
-  }
-
-  def releaseNuma(containerId: ContainerId, executorHostName: String): Unit = {
-    if (hostToContainers.get(executorHostName) != null) {
-      val numaIdToRelease = hostToContainers.get(executorHostName).getOrElseUpdate(containerId, "null")
-      logDebug(s"release $containerId with $numaIdToRelease on host $executorHostName")
-      hostToNumaIds.get(executorHostName)(numaIdToRelease.toInt) -= 1
-      hostToContainers.get(executorHostName).remove(containerId)
-    }
-  }
-
-}
-
 private object YarnAllocator {
   val MEM_REGEX = "[0-9.]+ [KMG]B"
   val VMEM_EXCEEDED_EXIT_CODE = -103

diff --git a/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/config.scala b/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/config.scala
@@ -164,10 +164,10 @@ package object config {
     .booleanConf
     .createWithDefault(false)
 
-  private[spark] val SPARK_YARN_NUMA_NUM = ConfigBuilder("spark.yarn.numa.num")
-    .doc("Specify numa node number in the host")
+  private[spark] val SPARK_YARN_NUMA_NUMBER = ConfigBuilder("spark.yarn.numa.number")
+    .doc("Total number of numanodes in physical server")
     .intConf
-    .createWithDefault(0)
+    .createWithDefault(2)
 
   private[spark] val WAIT_FOR_APP_COMPLETION = ConfigBuilder("spark.yarn.submit.waitAppCompletion")
     .doc("In cluster mode, whether to wait for the application to finish before exiting the " +

diff --git a/...gers/yarn/src/main/scala/org/apache/spark/executor/YarnCoarseGrainedExecutorBackend.scala b/...gers/yarn/src/main/scala/org/apache/spark/executor/YarnCoarseGrainedExecutorBackend.scala
@@ -35,6 +35,7 @@ private[spark] class YarnCoarseGrainedExecutorBackend(
     rpcEnv: RpcEnv,
     driverUrl: String,
     executorId: String,
+    numaNodeId: Option[String],
     bindAddress: String,
     hostname: String,
     cores: Int,
@@ -46,6 +47,7 @@ private[spark] class YarnCoarseGrainedExecutorBackend(
     rpcEnv,
     driverUrl,
     executorId,
+    numaNodeId,
     bindAddress,
     hostname,
     cores,
@@ -73,8 +75,8 @@ private[spark] object YarnCoarseGrainedExecutorBackend extends Logging {
     val createFn: (RpcEnv, CoarseGrainedExecutorBackend.Arguments, SparkEnv, ResourceProfile) =>
       CoarseGrainedExecutorBackend = { case (rpcEnv, arguments, env, resourceProfile) =>
       new YarnCoarseGrainedExecutorBackend(rpcEnv, arguments.driverUrl, arguments.executorId,
-        arguments.bindAddress, arguments.hostname, arguments.cores, arguments.userClassPath, env,
-        arguments.resourcesFileOpt, resourceProfile)
+        arguments.numaNodeId, arguments.bindAddress, arguments.hostname, arguments.cores,
+        arguments.userClassPath, env, arguments.resourcesFileOpt, resourceProfile)
     }
     val backendArgs = CoarseGrainedExecutorBackend.parseArguments(args,
       this.getClass.getCanonicalName.stripSuffix("$"))

diff --git a/scalastyle-config.xml b/scalastyle-config.xml
@@ -94,7 +94,7 @@ This file is divided into 3 sections:
   </check>
 
   <check customId="argcount" level="error" class="org.scalastyle.scalariform.ParameterNumberChecker" enabled="true">
-    <parameters><parameter name="maxParameters"><![CDATA[10]]></parameter></parameters>
+    <parameters><parameter name="maxParameters"><![CDATA[12]]></parameter></parameters>
   </check>
 
   <check level="error" class="org.scalastyle.scalariform.NoFinalizeChecker" enabled="true"></check>