fix several issues on allocation

Author: CodingCat

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

@@ -572,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 > app.desc.coresPerTask) {
+        for (app <- waitingApps if app.coresLeft >= app.desc.coresPerTask) {
           allocateWorkerResourceToExecutors(app, app.coresLeft, worker)
         }
       }
@@ -592,9 +592,10 @@ private[master] class Master(
     val cpuPerTask = app.desc.coresPerTask
     val memoryPerExecutor = app.desc.memoryPerExecutorMB
     var coresLeft = coresToAllocate - coresToAllocate % cpuPerTask
+    val maxCoresPerExecutor = app.desc.coresPerExecutor.getOrElse(coresLeft)
     val coresPerExecutor = math.max(
-      app.desc.coresPerExecutor.getOrElse(coresLeft),
-      app.desc.coresPerTask)
+      maxCoresPerExecutor - maxCoresPerExecutor % cpuPerTask,
+      cpuPerTask)
     while (coresLeft >= coresPerExecutor && worker.memoryFree >= memoryPerExecutor) {
       val exec = app.addExecutor(worker, coresPerExecutor)
       coresLeft -= coresPerExecutor

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

@@ -75,6 +75,13 @@ private[spark] class TaskSchedulerImpl(
       s"spark.task.cpus must be greater than 0! (was $CPUS_PER_TASK)")
   }
 
+  val maxCores = conf.get("spark.cores.max",  Int.MaxValue.toString).toInt
+
+  if (maxCores < CPUS_PER_TASK) {
+    throw new IllegalArgumentException(
+      s"spark.task.cpus ($CPUS_PER_TASK) should not be larger than spark.cores.max ($maxCores)")
+  }
+
   // 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

@@ -43,15 +43,9 @@ private[spark] class SparkDeploySchedulerBackend(
   private val registrationBarrier = new Semaphore(0)
 
   val maxCores = conf.getOption("spark.cores.max").map(_.toInt)
-  val coreNumPerTask = scheduler.CPUS_PER_TASK
-    
-  if (maxCores.isDefined && maxCores.get < coreNumPerTask) {
-    throw new IllegalArgumentException(
-      s"spark.task.cpus ($coreNumPerTask) should not be larger than spark.cores.max ($maxCores)")
-  }
-  
   val totalExpectedCores = maxCores.getOrElse(0)
-
+  val coresPerTask = scheduler.CPUS_PER_TASK
+    
   override def start() {
     super.start()
 
@@ -91,7 +85,7 @@ private[spark] class SparkDeploySchedulerBackend(
     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, coresPerExecutor, coreNumPerTask)
+      appUIAddress, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, coresPerTask)
     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

@@ -65,11 +65,6 @@ private[spark] class CoarseMesosSchedulerBackend(
 
   val coresPerTask = scheduler.CPUS_PER_TASK
 
-  if (maxCores < coresPerTask) {
-    throw new IllegalArgumentException(
-      s"spark.task.cpus ($coresPerTask) should not be larger than spark.cores.max ($maxCores)")
-  }
-
   // Cores we have acquired with each Mesos task ID
   val coresByTaskId = new HashMap[Int, Int]
   var totalCoresAcquired = 0