Skip to content
This repository has been archived by the owner on Sep 18, 2023. It is now read-only.

[NSE-725] change the code style for ExecutorManger #724

Merged
merged 2 commits into from
Mar 9, 2022
Merged

[NSE-725] change the code style for ExecutorManger #724

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ca76991
support test with local
jackylee-ch Jan 28, 2022
12b3757
change code style
jackylee-ch Mar 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 11 additions & 16 deletions native-sql-engine/core/src/main/scala/org/apache/spark/util/ExecutorManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,21 +16,19 @@
*/
package org.apache.spark.util

import org.apache.spark.{SparkConf, SparkContext, SparkEnv}
import java.io.File
import java.nio.file.Files
import java.nio.file.LinkOption
import java.nio.file.Path
import java.nio.file.Paths
import java.io.{InputStreamReader, BufferedReader}
import scala.collection.mutable.ListBuffer
import java.lang.management.ManagementFactory

import scala.util.Random

import com.intel.oap._

object ExecutorManager {
import org.apache.spark.{SparkContext, SparkEnv}
import org.apache.spark.internal.Logging

object ExecutorManager extends Logging {
def getExecutorIds(sc: SparkContext): Seq[String] = sc.getExecutorIds
var isTaskSet: Boolean = false
def tryTaskSet(numaInfo: GazelleNumaBindingInfo) = synchronized {
def tryTaskSet(numaInfo: GazelleNumaBindingInfo): Unit = synchronized {
if (numaInfo.enableNumaBinding && !isTaskSet) {
val cmd_output =
Utils.executeAndGetOutput(
Expand All @@ -41,19 +39,16 @@ object ExecutorManager {
tmp.toList.distinct
}
val executorId = SparkEnv.get.executorId
val numCorePerExecutor = numaInfo.numCoresPerExecutor
val coreRange = numaInfo.totalCoreRange
val shouldBindNumaIdx = executorIdOnLocalNode.indexOf(executorId) % coreRange.size
//val coreStartIdx = coreRange(shouldBindNumaIdx)._1
//val coreEndIdx = coreRange(shouldBindNumaIdx)._2
System.out.println(
s"executorId is ${executorId}, executorIdOnLocalNode is ${executorIdOnLocalNode}")
logInfo(s"executorId is $executorId, executorIdOnLocalNode is $executorIdOnLocalNode")
val taskSetCmd = s"taskset -cpa ${coreRange(shouldBindNumaIdx)} ${getProcessId()}"
System.out.println(taskSetCmd)
logInfo(s"taskSetCmd is $taskSetCmd")
isTaskSet = true
Utils.executeCommand(Seq("bash", "-c", taskSetCmd))
}
}

def getProcessId(): Int = {
val runtimeMXBean = ManagementFactory.getRuntimeMXBean()
runtimeMXBean.getName().split("@")(0).toInt
Expand Down