Fix head of line blocking in scheduling drivers.

Author: tnachen

core/src/main/scala/org/apache/spark/deploy/mesos/ui/MesosClusterPage.scala

@@ -58,18 +58,19 @@ private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage(
 
   private def queuedRow(submission: MesosDriverDescription): Seq[Node] = {
     <tr>
-      <td>{submission.submissionId}</td>
-      <td>{submission.submissionDate}</td>
+      <td>{submission.submissionId.get}</td>
+      <td>{submission.submissionDate.get}</td>
       <td>{submission.command.mainClass}</td>
       <td>cpus: {submission.cores}, mem: {submission.mem}</td>
     </tr>
   }
 
   private def driverRow(state: MesosClusterTaskState): Seq[Node] = {
     <tr>
-      <td>{state.submission.submissionId}</td>
-      <td>{state.submission.submissionDate}</td>
+      <td>{state.submission.submissionId.get}</td>
+      <td>{state.submission.submissionDate.get}</td>
       <td>{state.submission.command.mainClass}</td>
+      <td>cpus: {state.submission.cores}, mem: {state.submission.mem}</td>
       <td>{state.startDate}</td>
       <td>{state.slaveId.getValue}</td>
       <td>{stateString(state.taskState)}</td>
@@ -78,8 +79,8 @@ private[mesos] class MesosClusterPage(parent: MesosClusterUI) extends WebUIPage(
 
   private def retryRow(state: RetryState): Seq[Node] = {
     <tr>
-      <td>{state.submission.submissionId}</td>
-      <td>{state.submission.submissionDate}</td>
+      <td>{state.submission.submissionId.get}</td>
+      <td>{state.submission.submissionDate.get}</td>
       <td>{state.submission.command.mainClass}</td>
       <td>{state.lastFailureStatus}</td>
       <td>{state.nextRetry}</td>

core/src/main/scala/org/apache/spark/deploy/rest/SubmitRestProtocolRequest.scala

@@ -61,7 +61,7 @@ private[rest] class CreateSubmissionRequest extends SubmitRestProtocolRequest {
     assertProperty[Boolean](key, "boolean", _.toBoolean)
 
   private def assertPropertyIsNumeric(key: String): Unit =
-    assertProperty[Int](key, "numeric", _.toInt)
+    assertProperty[Double](key, "numeric", _.toDouble)
 
   private def assertPropertyIsMemory(key: String): Unit =
     assertProperty[Int](key, "memory", Utils.memoryStringToMb)

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

@@ -62,30 +62,17 @@ private[mesos] class DriverQueue(state: MesosClusterPersistenceEngine, capacity:
   }
 
   def remove(submissionId: String): Boolean = {
-    val removed = queue.dequeueFirst(d => d.submissionId.equals(submissionId))
-    if (removed.isDefined) {
-      state.expunge(removed.get.submissionId.get)
+    val removed = queue.dequeueFirst { d =>
+      d.submissionId.map(_.equals(submissionId)).getOrElse(false)
     }
-
+    if (removed.isDefined) state.expunge(removed.get.submissionId.get)
     removed.isDefined
   }
 
-  def peek(): Option[MesosDriverDescription] = {
-    queue.headOption
-  }
-
-  def poll(): Option[MesosDriverDescription] = {
-    if (queue.isEmpty) {
-      None
-    } else {
-      val item = queue.dequeue()
-      state.expunge(item.submissionId.get)
-      Some(item)
-    }
-  }
+  def drivers: Seq[MesosDriverDescription] = queue
 
   // Returns a copy of the queued drivers.
-  def drivers: Iterable[MesosDriverDescription] = {
+  def copyDrivers: Iterable[MesosDriverDescription] = {
     val buffer = new Array[MesosDriverDescription](queue.size)
     queue.copyToArray(buffer)
     buffer

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

@@ -76,7 +76,7 @@ private[spark] class MesosClusterTaskState(
  */
 private[spark] class MesosClusterSchedulerState(
     val frameworkId: String,
-    val masterUrl: String,
+    val masterUrl: Option[String],
     val queuedDrivers: Iterable[MesosDriverDescription],
     val launchedDrivers: Iterable[MesosClusterTaskState],
     val finishedDrivers: Iterable[MesosClusterTaskState],
@@ -135,7 +135,7 @@ private[spark] class MesosClusterSchedulerDriver(
   // All supervised drivers that are waiting to retry after termination.
   var superviseRetryList: SuperviseRetryList = _
 
-  private var masterInfo: MasterInfo = _
+  private var masterInfo: Option[MasterInfo] = None
 
   private def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")  // For application IDs
   private def newDriverId(submitDate: Date): String = {
@@ -252,7 +252,7 @@ private[spark] class MesosClusterSchedulerDriver(
     markRegistered()
 
     stateLock.synchronized {
-      this.masterInfo = masterInfo
+      this.masterInfo = Some(masterInfo)
       if (!launchedDrivers.pendingRecover.isEmpty) {
         // Start task reconciliation if we need to recover.
         val statuses = launchedDrivers.pendingRecover.collect {
@@ -350,20 +350,14 @@ private[spark] class MesosClusterSchedulerDriver(
         getResource(o.getResourcesList, "cpus"),
         getResource(o.getResourcesList, "mem"))
     }
-    logTrace(s"Received offers from Mesos: ${printOffers(currentOffers)}")
+    logTrace(s"Received offers from Mesos: \n${printOffers(currentOffers)}")
     val tasks = new mutable.HashMap[OfferID, ArrayBuffer[TaskInfo]]()
     val currentTime = new Date()
-
     def scheduleTasks(
-        taskFunc: () => (Option[MesosDriverDescription], Option[RetryState]),
+        tasksFunc: () => Seq[(MesosDriverDescription, Option[RetryState])],
         scheduledCallback: (String) => Unit): Unit = {
-      var nextItem = taskFunc()
-      // TODO: We should not stop scheduling at the very first task
-      // that cannot be scheduled. Instead we should exhaust the
-      // candidate list and remove drivers that cannot scheduled
-      // over a configurable period of time.
-      while (nextItem._1.isDefined) {
-        val (submission, retryState) = (nextItem._1.get, nextItem._2)
+      val candidates = tasksFunc()
+      for ((submission, retryState) <- candidates) {
         val driverCpu = submission.cores
         val driverMem = submission.mem
         logTrace(s"Finding offer to launch driver with cpu: $driverCpu, mem: $driverMem")
@@ -374,56 +368,54 @@ private[spark] class MesosClusterSchedulerDriver(
         if (offerOption.isEmpty) {
           logDebug(s"Unable to find offer to launch driver id: ${submission.submissionId.get}," +
               s"cpu: $driverCpu, mem: $driverMem")
-          return
-        }
-
-        val offer = offerOption.get
-        offer.cpu -= driverCpu
-        offer.mem -= driverMem
-        val taskId = TaskID.newBuilder().setValue(submission.submissionId.get).build()
-        val cpuResource = Resource.newBuilder()
-          .setName("cpus").setType(Value.Type.SCALAR)
-          .setScalar(Value.Scalar.newBuilder().setValue(driverCpu)).build()
-        val memResource = Resource.newBuilder()
-          .setName("mem").setType(Value.Type.SCALAR)
-          .setScalar(Value.Scalar.newBuilder().setValue(driverMem)).build()
-        val commandInfo = buildCommand(submission)
-        val appName = submission.schedulerProperties("spark.app.name")
-        val taskInfo = TaskInfo.newBuilder()
-          .setTaskId(taskId)
-          .setName(s"Driver for $appName")
-          .setSlaveId(offer.offer.getSlaveId)
-          .setCommand(commandInfo)
-          .addResources(cpuResource)
-          .addResources(memResource)
-          .build
-        val queuedTasks = if (!tasks.contains(offer.offer.getId)) {
-          val buffer = new ArrayBuffer[TaskInfo]
-          tasks(offer.offer.getId) = buffer
-          buffer
         } else {
-          tasks(offer.offer.getId)
+          val offer = offerOption.get
+          offer.cpu -= driverCpu
+          offer.mem -= driverMem
+          val taskId = TaskID.newBuilder().setValue(submission.submissionId.get).build()
+          val cpuResource = Resource.newBuilder()
+            .setName("cpus").setType(Value.Type.SCALAR)
+            .setScalar(Value.Scalar.newBuilder().setValue(driverCpu)).build()
+          val memResource = Resource.newBuilder()
+            .setName("mem").setType(Value.Type.SCALAR)
+            .setScalar(Value.Scalar.newBuilder().setValue(driverMem)).build()
+          val commandInfo = buildCommand(submission)
+          val appName = submission.schedulerProperties("spark.app.name")
+          val taskInfo = TaskInfo.newBuilder()
+            .setTaskId(taskId)
+            .setName(s"Driver for $appName")
+            .setSlaveId(offer.offer.getSlaveId)
+            .setCommand(commandInfo)
+            .addResources(cpuResource)
+            .addResources(memResource)
+            .build
+          val queuedTasks = if (!tasks.contains(offer.offer.getId)) {
+            val buffer = new ArrayBuffer[TaskInfo]
+            tasks(offer.offer.getId) = buffer
+            buffer
+          } else {
+            tasks(offer.offer.getId)
+          }
+          queuedTasks += taskInfo
+          logTrace(s"Using offer ${offer.offer.getId.getValue} to launch driver " +
+            submission.submissionId.get)
+
+          launchedDrivers.set(
+            submission.submissionId.get,
+            new MesosClusterTaskState(submission, taskId, offer.offer.getSlaveId,
+              None, new Date(), retryState))
+          scheduledCallback(submission.submissionId.get)
         }
-        queuedTasks += taskInfo
-        logTrace(s"Using offer ${offer.offer.getId.getValue} to launch driver " +
-          submission.submissionId.get)
-
-        launchedDrivers.set(
-          submission.submissionId.get,
-          new MesosClusterTaskState(submission, taskId, offer.offer.getSlaveId,
-            None, new Date(), retryState))
-        scheduledCallback(submission.submissionId.get)
-        nextItem = taskFunc()
       }
     }
 
     stateLock.synchronized {
       scheduleTasks(() => {
-        superviseRetryList.getNextRetry(currentTime)
+        superviseRetryList.getNextRetries(currentTime)
       }, (id: String) => {
         superviseRetryList.remove(id)
       })
-      scheduleTasks(() => (queue.peek(), None), (_) => queue.poll())
+      scheduleTasks(() => queue.drivers.map(d => (d, None)), (id) => queue.remove(id))
     }
 
     tasks.foreach { case (offerId, tasks) =>
@@ -439,8 +431,8 @@ private[spark] class MesosClusterSchedulerDriver(
     stateLock.synchronized {
       new MesosClusterSchedulerState(
         frameworkId,
-        s"http://${masterInfo.getIp}:${masterInfo.getPort}",
-        queue.drivers,
+        masterInfo.map(m => s"http://${m.getIp}:${m.getPort}"),
+        queue.copyDrivers,
         launchedDrivers.states,
         finishedDrivers.collect { case s => s.copy() },
         superviseRetryList.retries)

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

@@ -20,7 +20,7 @@ package org.apache.spark.scheduler.cluster.mesos
 import java.util.List
 import java.util.concurrent.CountDownLatch
 
-import org.apache.mesos.Protos.{FrameworkInfo, Resource}
+import org.apache.mesos.Protos.{Status, FrameworkInfo, Resource}
 import org.apache.mesos.{MesosSchedulerDriver, Scheduler, SchedulerDriver}
 import org.apache.spark.Logging
 
@@ -64,6 +64,7 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
           try {
             val ret = driver.run()
             logInfo("driver.run() returned with code " + ret)
+            onDriverExit(ret)
           } catch {
             case e: Exception => logError("driver.run() failed", e)
           }
@@ -74,6 +75,14 @@ private[mesos] trait MesosSchedulerUtils extends Logging {
     }
   }
 
+  def onDriverExit(status: Status): Unit = {
+    // Exit the process when the Mesos framework driver was aborted.
+    // This behavior can be overriden by the scheduler.
+    if (status.equals(Status.DRIVER_ABORTED)) {
+      System.exit(1)
+    }
+  }
+
   /**
    * Waits for the scheduler to be registered, which the scheduler will signal by calling
    * markRegistered().

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

@@ -59,28 +59,32 @@ private[mesos] class SuperviseRetryList(state: MesosClusterPersistenceEngine) {
 
   def size: Int = drivers.size
 
-  def contains(submissionId: String): Boolean =
-    drivers.exists(d => d.submission.submissionId.equals(submissionId))
-
-  def getNextRetry(currentTime: Date): (Option[MesosDriverDescription], Option[RetryState]) = {
-    val retry = drivers.find(d => d.nextRetry.before(currentTime))
-    if (retry.isDefined) {
-      (Some(retry.get.submission), retry)
-    } else {
-      (None, None)
+  def contains(submissionId: String): Boolean = {
+    drivers.exists { d =>
+      d.submission.submissionId.map(_.equals(submissionId)).getOrElse(false)
+    }
+  }
+
+  def getNextRetries(currentTime: Date): Seq[(MesosDriverDescription, Option[RetryState])] = {
+    drivers.filter(d => d.nextRetry.before(currentTime)).map { d =>
+      (d.submission, Some(d))
     }
   }
 
   def get(submissionId: String): Option[RetryState] = {
-    drivers.find(d => d.submission.submissionId.equals(submissionId))
+    drivers.find { d =>
+      d.submission.submissionId.map(_.equals(submissionId)).getOrElse(false)
+    }
   }
 
   def retries: Iterable[RetryState] = {
     drivers.map(d => d.copy).toList
   }
 
   def remove(submissionId: String): Boolean = {
-    val index = drivers.indexWhere(s => s.submission.submissionId.equals(submissionId))
+    val index = drivers.indexWhere { s =>
+      s.submission.submissionId.map(_.equals(submissionId)).getOrElse(false)
+    }
 
     if (index != -1) {
       drivers.remove(index)

core/src/test/scala/org/apache/spark/scheduler/mesos/MesosClusterSchedulerSuite.scala

@@ -49,8 +49,8 @@ class MesosClusterSchedulerSuite extends FunSuite with LocalSparkContext with Mo
     assert(response2.success)
 
     val state = scheduler.getState()
-    assert(state.queuedDrivers.exists(d => d.submissionId == response.submissionId))
-    assert(state.queuedDrivers.exists(d => d.submissionId == response2.submissionId))
+    assert(state.queuedDrivers.exists(d => d.submissionId.get == response.submissionId))
+    assert(state.queuedDrivers.exists(d => d.submissionId.get == response2.submissionId))
   }
 
   test("can kill queued drivers") {