[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r360738965
##
File path: core/src/main/scala/org/apache/spark/scheduler/Pool.scala
##
@@ -119,4 +120,72 @@ private[spark] class Pool(
parent.decreaseRunningTasks(taskNum)
}
}
+
+ // Update the number of slots considered available for each TaskSetManager
whose ancestor
+ // in the tree is this pool
+ // For FAIR scheduling, slots are distributed among pools based on weights
and minshare.
+ // If a pool requires fewer slots than are available to it, the leftover
slots are redistributed
+ // to the remaining pools using the remaining pools' weights.
+ // For FIFO scheduling, the schedulable queue is iterated over in FIFO order,
+ // giving each schedulable the remaining slots,
+ // up to the number of remaining tasks for that schedulable.
+ override def updateAvailableSlots(numSlots: Float): Unit = {
+schedulingMode match {
+ case SchedulingMode.FAIR =>
+val queueCopy = new util.LinkedList[Schedulable](schedulableQueue)
+var shouldRedistribute = true
+var totalWeights = schedulableQueue.asScala.map(_.weight).sum
+var totalSlots = numSlots
+while (totalSlots > 0 && shouldRedistribute) {
+ shouldRedistribute = false
+ var nextWeights = totalWeights
+ var nextSlots = totalSlots
+ val iterator = queueCopy.iterator()
+ while (iterator.hasNext) {
+val schedulable = iterator.next()
+val numTasksRemaining = schedulable.getSortedTaskSetQueue
+ .map(tsm => tsm.tasks.length - tsm.tasksSuccessful).sum
+val allocatedSlots = Math.max(
+ totalSlots * schedulable.weight / totalWeights,
+ schedulable.minShare)
+if (numTasksRemaining < allocatedSlots) {
Review comment:
2. from above can be solved by utilizing the `SchedulingAlgorithm` as
mentioned, but 1. still remains. A couple more problematic areas:
- Slots can be rejected due to
[blacklisting](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L392)
- Slots can be rejected due to not meeting [resource
requirements](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala#L344)
I'm trying to think of an idea where the TSM will report directly if it
rejected due to delay scheduling, but I am having trouble thinking how to
utilize that data due to problem 1. in previous comment.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r360474081
##
File path: core/src/main/scala/org/apache/spark/scheduler/Pool.scala
##
@@ -119,4 +120,72 @@ private[spark] class Pool(
parent.decreaseRunningTasks(taskNum)
}
}
+
+ // Update the number of slots considered available for each TaskSetManager
whose ancestor
+ // in the tree is this pool
+ // For FAIR scheduling, slots are distributed among pools based on weights
and minshare.
+ // If a pool requires fewer slots than are available to it, the leftover
slots are redistributed
+ // to the remaining pools using the remaining pools' weights.
+ // For FIFO scheduling, the schedulable queue is iterated over in FIFO order,
+ // giving each schedulable the remaining slots,
+ // up to the number of remaining tasks for that schedulable.
+ override def updateAvailableSlots(numSlots: Float): Unit = {
+schedulingMode match {
+ case SchedulingMode.FAIR =>
+val queueCopy = new util.LinkedList[Schedulable](schedulableQueue)
+var shouldRedistribute = true
+var totalWeights = schedulableQueue.asScala.map(_.weight).sum
+var totalSlots = numSlots
+while (totalSlots > 0 && shouldRedistribute) {
+ shouldRedistribute = false
+ var nextWeights = totalWeights
+ var nextSlots = totalSlots
+ val iterator = queueCopy.iterator()
+ while (iterator.hasNext) {
+val schedulable = iterator.next()
+val numTasksRemaining = schedulable.getSortedTaskSetQueue
+ .map(tsm => tsm.tasks.length - tsm.tasksSuccessful).sum
+val allocatedSlots = Math.max(
+ totalSlots * schedulable.weight / totalWeights,
+ schedulable.minShare)
+if (numTasksRemaining < allocatedSlots) {
Review comment:
You got me thinking more about this by using the word "exactly". It is the
combination of TaskSchedulerImpl, Pool (including FAIR/FIFO scheduling algos),
and TaskSetManager (including delay scheduling), etc. which determine how
resources are assigned.
The goal for this approach is to simulate scheduling without delay
scheduling.
This helps determine how much you are underutilizing resources due to delay
scheduling.
So far the most recent diff seems to fall short due to at least a couple
reasons:
1. Scheduling is different depending on if
`TaskSchedulerImpl.resourceOffers` is called one by one with single offers vs
if it is called with all offers in one batch.
`Schedulable.getSortedTaskSetQueue` is called only once per `resourceOffers`
call, meaning that for a batch call, it only follows the scheduling algorithm
for the first task that is scheduled (seems like a bug).
2. The approach doesn't exactly follow FAIR ordering, such as the
minShareRatio and schedulable name based ordering found in
`FairSchedulingAlgorithm.`
I have a rough idea for an alternative implementation which does a more
direct simulation, utilizing the `SchedulingAlgorithm` trait directly. I'll do
more thinking in the coming days.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r360474081
##
File path: core/src/main/scala/org/apache/spark/scheduler/Pool.scala
##
@@ -119,4 +120,72 @@ private[spark] class Pool(
parent.decreaseRunningTasks(taskNum)
}
}
+
+ // Update the number of slots considered available for each TaskSetManager
whose ancestor
+ // in the tree is this pool
+ // For FAIR scheduling, slots are distributed among pools based on weights
and minshare.
+ // If a pool requires fewer slots than are available to it, the leftover
slots are redistributed
+ // to the remaining pools using the remaining pools' weights.
+ // For FIFO scheduling, the schedulable queue is iterated over in FIFO order,
+ // giving each schedulable the remaining slots,
+ // up to the number of remaining tasks for that schedulable.
+ override def updateAvailableSlots(numSlots: Float): Unit = {
+schedulingMode match {
+ case SchedulingMode.FAIR =>
+val queueCopy = new util.LinkedList[Schedulable](schedulableQueue)
+var shouldRedistribute = true
+var totalWeights = schedulableQueue.asScala.map(_.weight).sum
+var totalSlots = numSlots
+while (totalSlots > 0 && shouldRedistribute) {
+ shouldRedistribute = false
+ var nextWeights = totalWeights
+ var nextSlots = totalSlots
+ val iterator = queueCopy.iterator()
+ while (iterator.hasNext) {
+val schedulable = iterator.next()
+val numTasksRemaining = schedulable.getSortedTaskSetQueue
+ .map(tsm => tsm.tasks.length - tsm.tasksSuccessful).sum
+val allocatedSlots = Math.max(
+ totalSlots * schedulable.weight / totalWeights,
+ schedulable.minShare)
+if (numTasksRemaining < allocatedSlots) {
Review comment:
You got me thinking more about this by using the word "exactly". It is the
combination of TaskSchedulerImpl, Pool (including FAIR/FIFO scheduling algos),
and TaskManager (including delay scheduling), etc. which determine how
resources are assigned.
The goal for this approach is to simulate scheduling without delay
scheduling.
This helps determine how much you are underutilizing resources due to delay
scheduling.
So far the most recent diff seems to fall short due to at least a couple
reasons:
1. Scheduling is different depending on if
`TaskSchedulerImpl.resourceOffers` is called one by one with single offers vs
if it is called with all offers in one batch.
`Schedulable.getSortedTaskSetQueue` is called only once per `resourceOffers`
call, meaning that for a batch call, it only follows the scheduling algorithm
for the first task that is scheduled (seems like a bug).
2. The approach doesn't exactly follow FAIR ordering, such as the
minShareRatio and schedulable name based ordering found in
`FairSchedulingAlgorithm.`
I have a rough idea for an alternative implementation which does a more
direct simulation, utilizing the `SchedulingAlgorithm` trait directly. I'll do
more thinking in the coming days.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r358617814
##
File path:
core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
##
@@ -403,6 +406,9 @@ private[spark] class TaskSchedulerImpl(
if (!executorIdToRunningTaskIds.contains(o.executorId)) {
hostToExecutors(o.host) += o.executorId
executorAdded(o.executorId, o.host)
+// Assumes the first offer will include all cores (free cores == all
cores)
Review comment:
can anyone confirm whether it is true that the first resource offer for an
executor will include all cores?
even if this is true it feels odd to rely on it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r355960425
##
File path: core/src/main/scala/org/apache/spark/scheduler/Pool.scala
##
@@ -119,4 +119,28 @@ private[spark] class Pool(
parent.decreaseRunningTasks(taskNum)
}
}
+
+ override def updateAvailableSlots(numSlots: Float): Unit = {
+schedulingMode match {
+ case SchedulingMode.FAIR =>
+val usableWeights = schedulableQueue.asScala
+ .map(s => if (s.getSortedTaskSetQueue.nonEmpty) (s, s.weight) else
(s, 0))
+val totalWeights = usableWeights.map(_._2).sum
+usableWeights.foreach({case (schedulable, usableWeight) =>
+ schedulable.updateAvailableSlots(
+Math.max(numSlots * usableWeight / totalWeights,
schedulable.minShare))
+})
+ case SchedulingMode.FIFO =>
+val sortedSchedulableQueue =
+
schedulableQueue.asScala.toSeq.sortWith(taskSetSchedulingAlgorithm.comparator)
+var isFirst = true
+for (schedulable <- sortedSchedulableQueue) {
+ schedulable.updateAvailableSlots(if (isFirst) numSlots else 0)
Review comment:
thanks for pointing this out!
Instead of giving all slots to the first schedulable, I will change it so
that if a schedulable has n tasks, it will only give n slots, and proceed to
the next until no slots remain.
I will try to apply similar logic to the FAIR case by including the num of
tasks in calculation and distributing any "unused" slots to the remaining tasks
based on weight.
What are your thoughts here?
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized URL: https://github.com/apache/spark/pull/26696#discussion_r355951463 ## File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala ## @@ -808,6 +815,7 @@ private[spark] class TaskSchedulerImpl( // happen below in the rootPool.executorLost() call. taskIds.foreach(cleanupTaskState) } +executorIdToCores.remove(executorId).foreach(totalSlots -= _ / CPUS_PER_TASK) Review comment: updateAvailableSlots is dependent on both the total number of slots + what tasksets/pools are present. Will take a look to see how feasible doing the update incrementally is. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r355089923
##
File path: core/src/main/scala/org/apache/spark/scheduler/Schedulable.scala
##
@@ -39,6 +39,7 @@ private[spark] trait Schedulable {
def stageId: Int
def name: String
+ def updateAvailableSlots(numSlots: Float): Unit
Review comment:
slots is equivalent to cores/(cores per task)
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized URL: https://github.com/apache/spark/pull/26696#discussion_r355071000 ## File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala ## @@ -439,6 +441,9 @@ private[spark] class TaskSchedulerImpl( } } +val availableSlots = executorIdToCores.values.map(c => c / CPUS_PER_TASK).sum +rootPool.updateAvailableSlots(availableSlots) Review comment: yea, good suggestion. I'll look into optimizing this. Also need to take into account when an executor is removed. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r355089524
##
File path: core/src/main/scala/org/apache/spark/scheduler/Pool.scala
##
@@ -119,4 +119,28 @@ private[spark] class Pool(
parent.decreaseRunningTasks(taskNum)
}
}
+
+ override def updateAvailableSlots(numSlots: Float): Unit = {
+schedulingMode match {
+ case SchedulingMode.FAIR =>
+val usableWeights = schedulableQueue.asScala
+ .map(s => if (s.getSortedTaskSetQueue.nonEmpty) (s, s.weight) else
(s, 0))
+val totalWeights = usableWeights.map(_._2).sum
+usableWeights.foreach({case (schedulable, usableWeight) =>
Review comment:
this is computing how many slots each taskset/pool could ideally use if
distributed according to scheduling policy, not how many **more slots** it can
use in the next round of scheduling.
Say there are 10 total slots with two jobs running, each which have the same
weight, so they are assigned a potential of 5 slots each. Also assume taskset 1
is using 6 slots and taskset 2 is using 0.
This code will determine that each taskset's availableSlots is 5, and since
taskset 1 is using more than that, its timer will reset, whereas taskset 2's
will not. This means taskset 2 will increase its locality level if the locality
timer has expired, allowing it to utilize its "ideal" 5 slots.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r355088554
##
File path: core/src/main/scala/org/apache/spark/scheduler/Pool.scala
##
@@ -119,4 +119,28 @@ private[spark] class Pool(
parent.decreaseRunningTasks(taskNum)
}
}
+
+ override def updateAvailableSlots(numSlots: Float): Unit = {
+schedulingMode match {
+ case SchedulingMode.FAIR =>
+val usableWeights = schedulableQueue.asScala
+ .map(s => if (s.getSortedTaskSetQueue.nonEmpty) (s, s.weight) else
(s, 0))
+val totalWeights = usableWeights.map(_._2).sum
+usableWeights.foreach({case (schedulable, usableWeight) =>
+ schedulable.updateAvailableSlots(
+Math.max(numSlots * usableWeight / totalWeights,
schedulable.minShare))
+})
+ case SchedulingMode.FIFO =>
+val sortedSchedulableQueue =
+
schedulableQueue.asScala.toSeq.sortWith(taskSetSchedulingAlgorithm.comparator)
+var isFirst = true
+for (schedulable <- sortedSchedulableQueue) {
+ schedulable.updateAvailableSlots(if (isFirst) numSlots else 0)
Review comment:
@squito is correct here.
The idea is for FIFO we assign all "available slots" to the first
taskset/pool in the queue.
For FAIR we divide based on total slots based on weights, not counting
weights with pools with no tasksets, and ensuring we always give at least
minShare slot availability.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized URL: https://github.com/apache/spark/pull/26696#discussion_r355071000 ## File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala ## @@ -439,6 +441,9 @@ private[spark] class TaskSchedulerImpl( } } +val availableSlots = executorIdToCores.values.map(c => c / CPUS_PER_TASK).sum +rootPool.updateAvailableSlots(availableSlots) Review comment: yea, good suggestion. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE] Make locality wait time be the time since a TSM's available slots were fully utilized
bmarcott commented on a change in pull request #26696: [WIP][SPARK-18886][CORE]
Make locality wait time be the time since a TSM's available slots were fully
utilized
URL: https://github.com/apache/spark/pull/26696#discussion_r355070261
##
File path:
core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
##
@@ -195,6 +195,66 @@ class TaskSchedulerImplSuite extends SparkFunSuite with
LocalSparkContext with B
assert(!failedTaskSet)
}
+ test("executors should not sit idle for too long") {
+val LOCALITY_WAIT_MS = 3000
+val clock = new ManualClock
+val conf = new SparkConf()
+sc = new SparkContext("local", "TaskSchedulerImplSuite", conf)
+// import org.slf4j.{Logger, LoggerFactory}
+import org.apache.log4j.{Logger, Level}
+
Logger.getLogger("org.apache.spark.scheduler.TaskSetManager").setLevel(Level.DEBUG)
Review comment:
yes this was just for local testing. will remove
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
