use config spark.scheduler.priority for specifying TaskSet's priority on DAGScheduler

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

@@ -46,6 +46,7 @@ private[spark] class Pool(
 
   // A pool's stage id is used to break the tie in scheduling.
   var stageId = -1
+  var jobId = -1
   var name = poolName
   var parent: Pool = null
 

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

@@ -37,6 +37,7 @@ private[spark] trait Schedulable {
   def runningTasks: Int
   def priority: Int
   def stageId: Int
+  def jobId: Int
   def name: String
 
   def addSchedulable(schedulable: Schedulable): Unit

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

@@ -28,19 +28,8 @@ private[spark] trait SchedulingAlgorithm {
 
 private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {
   override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
-    val priority1 = s1.priority
-    val priority2 = s2.priority
-    var res = math.signum(priority1 - priority2)
-    if (res == 0) {
-      val stageId1 = s1.stageId
-      val stageId2 = s2.stageId
-      res = math.signum(stageId1 - stageId2)
-    }
-    if (res < 0) {
-      true
-    } else {
-      false
-    }
+    import scala.math.Ordering.Implicits._
+    (s1.priority, s2.jobId, s2.stageId) > (s2.priority, s1.jobId, s1.stageId)
   }
 }
 

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

@@ -27,10 +27,14 @@ private[spark] class TaskSet(
     val tasks: Array[Task[_]],
     val stageId: Int,
     val attempt: Int,
-    val priority: Int,
+    val jobId: Int,
     val properties: Properties) {
     val id: String = stageId + "." + attempt
 
+  val priority = if (properties != null) {
+    properties.getProperty("spark.scheduler.priority", "0").toInt
+  } else 0
+
   def kill(interruptThread: Boolean) {
     tasks.foreach(_.kill(interruptThread))
   }

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

@@ -86,6 +86,7 @@ private[spark] class TaskSetManager(
   var minShare = 0
   var priority = taskSet.priority
   var stageId = taskSet.stageId
+  var jobId = taskSet.jobId
   var name = "TaskSet_" + taskSet.stageId.toString
   var parent: Pool = null