Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/4960#discussion_r33731738
--- Diff:
core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerUtils.scala
---
@@ -79,17 +119,59 @@ private[mesos] trait MesosSchedulerUtils extends
Logging {
/**
* Signal that the scheduler has registered with Mesos.
*/
+ protected def getResource(res: List[Resource], name: String): Double = {
+ // A resource can have multiple values in the offer since it can
either be from
+ // a specific role or wildcard.
+ res.filter(_.getName == name).map(_.getScalar.getValue).sum
+ }
+
protected def markRegistered(): Unit = {
registerLatch.countDown()
}
/**
- * Get the amount of resources for the specified type from the resource
list
+ * Partition the existing set of resources into two groups, those
remaining to be
+ * scheduled and those requested to be used for a new task.
+ * @param resources The full list of available resources
+ * @param resourceName The name of the resource to take from the
available resources
+ * @param count The amount of resources to take from the available
resources
+ * @return The remaining resources list and the used resources list.
*/
- protected def getResource(res: List[Resource], name: String): Double = {
- for (r <- res if r.getName == name) {
- return r.getScalar.getValue
+ def partitionResources(
+ resources: List[Resource],
+ resourceName: String,
+ count: Double): (List[Resource], List[Resource]) = {
+ var remain = count
+ var requestedResources = new ArrayBuffer[Resource]
+ val remainingResources = resources.collect {
+ case r => {
+ if (remain > 0 &&
+ r.getType == Value.Type.SCALAR &&
+ r.getScalar.getValue > 0.0 &&
+ r.getName == resourceName) {
+ val usage = Math.min(remain, r.getScalar.getValue)
+ requestedResources += Resource.newBuilder()
+ .setName(resourceName)
+ .setRole(r.getRole)
+ .setType(Value.Type.SCALAR)
+ .setScalar(Value.Scalar.newBuilder().setValue(usage).build())
+ .build()
+ remain -= usage
+ Resource.newBuilder()
+ .setName(resourceName)
+ .setRole(r.getRole)
+ .setType(Value.Type.SCALAR)
+
.setScalar(Value.Scalar.newBuilder().setValue(r.getScalar.getValue -
usage).build())
+ .build()
+ } else {
+ r
+ }
+ }
}
- 0.0
+
+ // Filter any resource that has depleted.
+ (remainingResources.filter(r => r.getType != Value.Type.SCALAR ||
r.getScalar.getValue > 0.0)
+ .toList,
+ requestedResources.toList)
--- End diff --
minor: can you put the long thing in a variable so it's easier to read
```
val filteredRemainingResources = remainingResources.filter { r =>
r.getType ...
}
(filteredRemainingResources.toList, requestedResources.toList)
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
