address the comments

Author: CodingCat

core/src/main/scala/org/apache/spark/deploy/ApplicationDescription.scala

@@ -28,13 +28,12 @@ private[spark] class ApplicationDescription(
     val eventLogDir: Option[URI] = None,
     // short name of compression codec used when writing event logs, if any (e.g. lzf)
     val eventLogCodec: Option[String] = None,
-    val coresPerExecutor: Option[Int] = None)
+    val coresPerExecutor: Option[Int] = None,
+    val coresPerTask: Int = 1)
   extends Serializable {
 
   val user = System.getProperty("user.name", "<unknown>")
 
-  var coreNumPerTask = 1
-
   def copy(
       name: String = name,
       maxCores: Option[Int] = maxCores,

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -541,20 +541,26 @@ private[master] class Master(
     // in the queue, then the second app, etc.
     if (spreadOutApps) {
       // Try to spread out each app among all the nodes, until it has all its cores
-      for (app <- waitingApps if app.coresLeft >= app.desc.coreNumPerTask) {
-        val coreNumPerTask = app.desc.coreNumPerTask
+      for (app <- waitingApps if app.coresLeft >= app.desc.coresPerTask) {
+        val coreNumPerTask = app.desc.coresPerTask
         val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
           .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB &&
-            worker.coresFree >= app.desc.coresPerExecutor.getOrElse(1))
+            worker.coresFree >= math.max(app.desc.coresPerExecutor.getOrElse(1),
+              app.desc.coresPerTask))
           .sortBy(_.coresFree).reverse
         val numUsable = usableWorkers.length
         val assigned = new Array[Int](numUsable) // Number of cores to give on each node
         var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
         var pos = 0
-        while (toAssign >= coreNumPerTask) {
+        val availableWorkerSet = new HashSet[Int]
+        // Take into account the number of cores the application uses per task
+        // to avoid starving executors or wasting cluster resources (SPARK-5337)
+        while (toAssign >= coreNumPerTask && availableWorkerSet.size < usableWorkers.length) {
           if (usableWorkers(pos).coresFree - assigned(pos) >= coreNumPerTask) {
             toAssign -= coreNumPerTask
             assigned(pos) += coreNumPerTask
+          } else {
+            availableWorkerSet += pos
           }
           pos = (pos + 1) % numUsable
         }
@@ -566,7 +572,7 @@ private[master] class Master(
     } else {
       // Pack each app into as few workers as possible until we've assigned all its cores
       for (worker <- workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) {
-        for (app <- waitingApps if app.coresLeft > 0) {
+        for (app <- waitingApps if app.coresLeft > app.desc.coresPerTask) {
           allocateWorkerResourceToExecutors(app, app.coresLeft, worker)
         }
       }
@@ -583,9 +589,12 @@ private[master] class Master(
       app: ApplicationInfo,
       coresToAllocate: Int,
       worker: WorkerInfo): Unit = {
+    val cpuPerTask = app.desc.coresPerTask
     val memoryPerExecutor = app.desc.memoryPerExecutorMB
-    val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(coresToAllocate)
-    var coresLeft = coresToAllocate
+    var coresLeft = coresToAllocate - coresToAllocate % cpuPerTask
+    val coresPerExecutor = math.max(
+      app.desc.coresPerExecutor.getOrElse(coresLeft),
+      app.desc.coresPerTask)
     while (coresLeft >= coresPerExecutor && worker.memoryFree >= memoryPerExecutor) {
       val exec = app.addExecutor(worker, coresPerExecutor)
       coresLeft -= coresPerExecutor

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

@@ -70,6 +70,11 @@ private[spark] class TaskSchedulerImpl(
   // CPUs to request per task
   val CPUS_PER_TASK = conf.getInt("spark.task.cpus", 1)
 
+  if (CPUS_PER_TASK < 1) {
+    throw new IllegalArgumentException(
+      s"\"spark.task.cpus\" must be greater than 0! (was $CPUS_PER_TASK)")
+  }
+
   // TaskSetManagers are not thread safe, so any access to one should be synchronized
   // on this class.
   val activeTaskSets = new HashMap[String, TaskSetManager]

core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala

@@ -45,11 +45,6 @@ private[spark] class SparkDeploySchedulerBackend(
   val maxCores = conf.getOption("spark.cores.max").map(_.toInt)
   val coreNumPerTask = scheduler.CPUS_PER_TASK
 
-  if (coreNumPerTask < 1) {
-    throw new IllegalArgumentException(
-      s"spark.task.cpus is set to an invalid value $coreNumPerTask")
-  }
-
   if (maxCores.isDefined && maxCores.get < coreNumPerTask) {
     throw new IllegalArgumentException(
       s"spark.task.cpus ($coreNumPerTask) should not be larger than spark.cores.max ($maxCores)")
@@ -94,9 +89,9 @@ private[spark] class SparkDeploySchedulerBackend(
     val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",
       args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)
     val appUIAddress = sc.ui.map(_.appUIAddress).getOrElse("")
+    val coresPerExecutor = conf.getOption("spark.executor.cores").map(_.toInt)
     val appDesc = new ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
-      appUIAddress, sc.eventLogDir, sc.eventLogCodec)
-    appDesc.coreNumPerTask = coreNumPerTask
+      appUIAddress, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, coreNumPerTask)
     client = new AppClient(sc.env.actorSystem, masters, appDesc, this, conf)
     client.start()
     waitForRegistration()

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

@@ -63,16 +63,11 @@ private[spark] class CoarseMesosSchedulerBackend(
   // Maximum number of cores to acquire (TODO: we'll need more flexible controls here)
   val maxCores = conf.get("spark.cores.max",  Int.MaxValue.toString).toInt
 
-  val coreNumPerTask = conf.getInt("spark.task.cpus", 1)
-    
-  if (coreNumPerTask < 1) {
-    throw new IllegalArgumentException(
-      s"spark.task.cpus is set to an invalid value $coreNumPerTask")
-  }
+  val coresPerTask = scheduler.CPUS_PER_TASK
 
-  if (maxCores < coreNumPerTask) {
+  if (maxCores < coresPerTask) {
     throw new IllegalArgumentException(
-      s"spark.task.cpus ($coreNumPerTask) should not be larger than spark.cores.max ($maxCores)")
+      s"spark.task.cpus ($coresPerTask) should not be larger than spark.cores.max ($maxCores)")
   }
 
   // Cores we have acquired with each Mesos task ID
@@ -225,9 +220,10 @@ private[spark] class CoarseMesosSchedulerBackend(
         val slaveId = offer.getSlaveId.toString
         val mem = getResource(offer.getResourcesList, "mem")
         val cpus = getResource(offer.getResourcesList, "cpus").toInt
-        val cpusToUse = math.min(cpus, maxCores - totalCoresAcquired)
+        val demandCores = math.min(cpus, maxCores - totalCoresAcquired)
+        val cpusToUse = demandCores - demandCores % coresPerTask
         if (totalCoresAcquired < maxCores &&
-            cpusToUse >= coreNumPerTask &&
+            cpusToUse >= coresPerTask &&
             mem >= MemoryUtils.calculateTotalMemory(sc) &&
             failuresBySlaveId.getOrElse(slaveId, 0) < MAX_SLAVE_FAILURES &&
             !slaveIdsWithExecutors.contains(slaveId)) {