Skip to content

[SPARK-15992] [MESOS] Refactor MesosCoarseGrainedSchedulerBackend offer consideration #13715

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 2 commits into from
Closed

[SPARK-15992] [MESOS] Refactor MesosCoarseGrainedSchedulerBackend offer consideration #13715

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6b6fceb
Refactor MesosCoarseGrainedSchedulerBackend offer consideration
drcrallen Jun 16, 2016
9e0aedf
Don't switch around variables incorrectly
drcrallen Jun 17, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 84 additions & 46 deletions ...n/scala/org/apache/spark/scheduler/cluster/mesos/MesosCoarseGrainedSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -382,59 +382,97 @@ private[spark] class MesosCoarseGrainedSchedulerBackend(
for (offer <- offers) {
val slaveId = offer.getSlaveId.getValue
val offerId = offer.getId.getValue
val resources = remainingResources(offerId)

if (canLaunchTask(slaveId, resources)) {
// Create a task
launchTasks = true
val taskId = newMesosTaskId()
val offerCPUs = getResource(resources, "cpus").toInt

val taskCPUs = executorCores(offerCPUs)
val taskMemory = executorMemory(sc)

slaves.getOrElseUpdate(slaveId, new Slave(offer.getHostname)).taskIDs.add(taskId)

val (afterCPUResources, cpuResourcesToUse) =
partitionResources(resources, "cpus", taskCPUs)
val (resourcesLeft, memResourcesToUse) =
partitionResources(afterCPUResources.asJava, "mem", taskMemory)

val taskBuilder = MesosTaskInfo.newBuilder()
.setTaskId(TaskID.newBuilder().setValue(taskId.toString).build())
.setSlaveId(offer.getSlaveId)
.setCommand(createCommand(offer, taskCPUs + extraCoresPerExecutor, taskId))
.setName("Task " + taskId)
.addAllResources(cpuResourcesToUse.asJava)
.addAllResources(memResourcesToUse.asJava)

sc.conf.getOption("spark.mesos.executor.docker.image").foreach { image =>
MesosSchedulerBackendUtil
.setupContainerBuilderDockerInfo(image, sc.conf, taskBuilder.getContainerBuilder)
val availableResources = remainingResources(offerId)
val offerMem = getResource(availableResources, "mem")
val offerCpu = getResource(availableResources, "cpus")

// Catch offer limits
calculateUsableResources(
sc,
offerCpu.toInt,
offerMem.toInt
).flatMap(
{
// Catch "global" limits
case (taskCPUs: Int, taskMemory: Int) =>
if (numExecutors() >= executorLimit) {
logTrace(s"${numExecutors()} exceeds limit of $executorLimit")
None
} else if (
slaves.get(slaveId).map(_.taskFailures).getOrElse(0) >= MAX_SLAVE_FAILURES
) {
logTrace(s"Slave $slaveId exceeded limit of $MAX_SLAVE_FAILURES failures")
None
} else {
Some((taskCPUs, taskMemory))
}
}

tasks(offer.getId) ::= taskBuilder.build()
remainingResources(offerId) = resourcesLeft.asJava
totalCoresAcquired += taskCPUs
coresByTaskId(taskId) = taskCPUs
) match {
case Some((taskCPUs: Int, taskMemory: Int)) =>
// Create a task
launchTasks = true
val taskId = newMesosTaskId()

slaves.getOrElseUpdate(slaveId, new Slave(offer.getHostname)).taskIDs.add(taskId)

val (afterCPUResources, cpuResourcesToUse) =
partitionResources(availableResources, "cpus", taskCPUs)
val (resourcesLeft, memResourcesToUse) =
partitionResources(afterCPUResources.asJava, "mem", taskMemory)

val taskBuilder = MesosTaskInfo.newBuilder()
.setTaskId(TaskID.newBuilder().setValue(taskId.toString).build())
.setSlaveId(offer.getSlaveId)
.setCommand(createCommand(offer, taskCPUs + extraCoresPerExecutor, taskId))
.setName("Task " + taskId)
.addAllResources(cpuResourcesToUse.asJava)
.addAllResources(memResourcesToUse.asJava)

sc.conf.getOption("spark.mesos.executor.docker.image").foreach { image =>
MesosSchedulerBackendUtil
.setupContainerBuilderDockerInfo(image, sc.conf, taskBuilder.getContainerBuilder)
}

tasks(offer.getId) ::= taskBuilder.build()
remainingResources(offerId) = resourcesLeft.asJava
totalCoresAcquired += taskCPUs
coresByTaskId(taskId) = taskCPUs
case None => logDebug(s"Insufficient offer CPU: $offerCpu MEM: $offerMem")
}
}
}
tasks.toMap
}

private def canLaunchTask(slaveId: String, resources: JList[Resource]): Boolean = {
val offerMem = getResource(resources, "mem")
val offerCPUs = getResource(resources, "cpus").toInt
val cpus = executorCores(offerCPUs)
val mem = executorMemory(sc)

cpus > 0 &&
cpus <= offerCPUs &&
cpus + totalCoresAcquired <= maxCores &&
mem <= offerMem &&
numExecutors() < executorLimit &&
slaves.get(slaveId).map(_.taskFailures).getOrElse(0) < MAX_SLAVE_FAILURES
/**
* Try and fit the resources to the constraints. Return None if it does not match
*
* @param sc Spark context
* @param availableCpus The available CPUs
* @param availableMem The available Memory
* @return Tuple of CPU (integer cores) and Memory (integer MB) desired
*/
private[mesos]
def calculateUsableResources(sc: SparkContext, availableCpus: Int, availableMem: Int):
Option[(Int, Int)] = {
val desiredMemory = executorMemory(sc)
val desiredCpu = executorCores(availableCpus)
Copy link
Contributor

@tnachen tnachen Jun 22, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like the idea, but it does make it harder to read. Can we refactor so we can collapse global and resource limits, and introduce like a case class for each condition, where it knows how to verify and logs error message when it doesn't match?
I think it makes adding and understanding how offers are matched better.

Copy link
Contributor Author

@drcrallen drcrallen Aug 16, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can do that. Will be a few weeks before I can tackle this again though.

if (desiredCpu < 1) {
logTrace(s"Executor cores too low at $desiredCpu")
None
} else if (desiredMemory > availableMem) {
logTrace(s"Offer memory $availableMem is less than needed memory $desiredMemory")
None
} else if (desiredCpu > availableCpus) {
logTrace(s"Offer cpu $availableCpus is les than needed cpu $desiredCpu")
None
} else if (desiredCpu + totalCoresAcquired > maxCores) {
logTrace(s"Adding $desiredCpu more cpus to $totalCoresAcquired"
+ s" would go over $maxCores limit")
None
} else {
Some((desiredCpu, desiredMemory))
}
}

private def executorCores(offerCPUs: Int): Int = {
Expand Down