improve

Author: zhixingheyi-tian

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

@@ -296,7 +296,7 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       printUsageAndExit()
     }
 
-    logInfo(s"[YUQIANG] numaNodeId $numaNodeId")
+    logInfo(s"[NUMACHECK] numaNodeId $numaNodeId")
     run(driverUrl, executorId, numaNodeId, hostname,  cores, appId, workerUrl, userClassPath)
     System.exit(0)
   }

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ExecutorRunnable.scala

@@ -201,7 +201,7 @@ private[yarn] class ExecutorRunnable(
 
     val numaEnabled = sparkConf.get(SPARK_YARN_NUMA_ENABLED)
 
-    logInfo(s"[YUQIANG] numaEnabled $numaEnabled executorId $executorId")
+    logInfo(s"[NUMACHECK] numaEnabled $numaEnabled executorId $executorId")
     // Don't need numa binding for driver.
     val (numaCtlCommand, numaNodeOpts) = if (numaEnabled && executorId != "<executorId>"
       && numaNodeId.nonEmpty) {
@@ -212,7 +212,7 @@ private[yarn] class ExecutorRunnable(
       ("", Nil)
     }
 
-    logInfo(s"[YUQIANG] numactl command $numaCtlCommand")
+    logInfo(s"[NUMACHECK] numactl command $numaCtlCommand")
     YarnSparkHadoopUtil.addOutOfMemoryErrorArgument(javaOpts)
     val commands = prefixEnv ++
       Seq(numaCtlCommand  + Environment.JAVA_HOME.$$() + "/bin/java", "-server") ++
@@ -229,7 +229,7 @@ private[yarn] class ExecutorRunnable(
         s"1>${ApplicationConstants.LOG_DIR_EXPANSION_VAR}/stdout",
         s"2>${ApplicationConstants.LOG_DIR_EXPANSION_VAR}/stderr")
 
-    logInfo(s"[YUQIANG] container command $commands")
+    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
   }

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -161,10 +161,11 @@ private[yarn] class YarnAllocator(
     new LocalityPreferredContainerPlacementStrategy(sparkConf, conf, resource, resolver)
 
   // The total number of numa node
-  private[yarn] val totalNumaNumber = 2
-  // Mapping from host to executor counter, we use the counter with a round-robin mode to
-  // determine the numa node id that the executor should bind.
-  private[yarn] val hostToNuma = new mutable.HashMap[String, Int]()
+  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]()
 
   /**
    * Use a different clock for YarnAllocator. This is mainly used for testing.
@@ -501,13 +502,19 @@ private[yarn] class YarnAllocator(
     for (container <- containersToUse) {
       executorIdCounter += 1
       val executorHostname = container.getNodeId.getHost
-      // Setting the numa id that the executor should binding. Just round robin from 0 to
-      // totalNumaNumber for each host.
-      // TODO: This is very ugly, however this is should be processed in resource
-      // manager(such as yarn).
-      val preSize = hostToNuma.getOrElseUpdate(executorHostname, 0)
-      val numaNodeId = (preSize % totalNumaNumber).toString
-      hostToNuma.put(executorHostname, preSize + 1)
+      // 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
@@ -598,6 +605,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., " +
@@ -621,7 +639,9 @@ private[yarn] class YarnAllocator(
             failedExecutorsTimeStamps.enqueue(clock.getTimeMillis())
             (true, "Container marked as failed: " + containerId + onHostStr +
               ". Exit status: " + completedContainer.getExitStatus +
-              ". Diagnostics: " + completedContainer.getDiagnostics)
+              ". Diagnostics: " + completedContainer.getDiagnostics +
+              ". numaNodeId: " + numaNodeId +
+              ". hostName: " + hostName)
 
         }
         if (exitCausedByApp) {

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/config.scala

@@ -135,6 +135,11 @@ package object config {
     .booleanConf
     .createWithDefault(false)
 
+  private[spark] val SPARK_YARN_NUMA_NUMBER = ConfigBuilder("spark.yarn.numa.number")
+    .doc("Total number of numanodes in physical server")
+    .intConf
+    .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 " +
       "launcher process.")