You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by zj...@apache.org on 2015/04/18 00:35:56 UTC
[28/50] [abbrv] hadoop git commit: YARN-3361. CapacityScheduler side
changes to support non-exclusive node labels. Contributed by Wangda Tan
YARN-3361. CapacityScheduler side changes to support non-exclusive node labels. Contributed by Wangda Tan
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/025787b3
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/025787b3
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/025787b3
Branch: refs/heads/YARN-2928
Commit: 025787b3d5b9bcefcb8f5ed422e3b1759599fd92
Parents: 77714c8
Author: Jian He <ji...@apache.org>
Authored: Tue Apr 14 11:36:37 2015 -0700
Committer: Zhijie Shen <zj...@apache.org>
Committed: Fri Apr 17 15:29:44 2015 -0700
----------------------------------------------------------------------
hadoop-yarn-project/CHANGES.txt | 3 +
.../hadoop/yarn/server/utils/BuilderUtils.java | 1 +
.../rmapp/attempt/RMAppAttemptImpl.java | 10 +-
.../scheduler/AppSchedulingInfo.java | 25 +-
.../scheduler/ResourceUsage.java | 8 +
.../scheduler/SchedulerApplicationAttempt.java | 44 +-
.../scheduler/SchedulerUtils.java | 87 +-
.../scheduler/capacity/AbstractCSQueue.java | 243 +++--
.../scheduler/capacity/CSQueue.java | 5 +-
.../scheduler/capacity/CapacityScheduler.java | 91 +-
.../CapacitySchedulerConfiguration.java | 5 +
.../scheduler/capacity/LeafQueue.java | 368 ++++---
.../scheduler/capacity/ParentQueue.java | 59 +-
.../scheduler/capacity/SchedulingMode.java | 44 +
.../server/resourcemanager/Application.java | 4 +
.../yarn/server/resourcemanager/MockAM.java | 8 +-
.../yarn/server/resourcemanager/MockRM.java | 35 +-
.../capacity/TestApplicationLimits.java | 8 +-
.../scheduler/capacity/TestChildQueueOrder.java | 41 +-
.../capacity/TestContainerAllocation.java | 390 +------
.../scheduler/capacity/TestLeafQueue.java | 148 +--
.../TestNodeLabelContainerAllocation.java | 1027 ++++++++++++++++++
.../scheduler/capacity/TestParentQueue.java | 111 +-
.../scheduler/capacity/TestReservations.java | 101 +-
.../scheduler/capacity/TestUtils.java | 2 +
25 files changed, 1914 insertions(+), 954 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt
index 6ca51c9..c0bc974 100644
--- a/hadoop-yarn-project/CHANGES.txt
+++ b/hadoop-yarn-project/CHANGES.txt
@@ -131,6 +131,9 @@ Release 2.8.0 - UNRELEASED
YARN-3443. Create a 'ResourceHandler' subsystem to ease addition of support
for new resource types on the NM. (Sidharta Seethana via junping_du)
+ YARN-3361. CapacityScheduler side changes to support non-exclusive node
+ labels. (Wangda Tan via jianhe)
+
IMPROVEMENTS
YARN-1880. Cleanup TestApplicationClientProtocolOnHA
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java
index 68d4ef9..f2146c8 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/utils/BuilderUtils.java
@@ -313,6 +313,7 @@ public class BuilderUtils {
request.setResourceName(r.getResourceName());
request.setCapability(r.getCapability());
request.setNumContainers(r.getNumContainers());
+ request.setNodeLabelExpression(r.getNodeLabelExpression());
return request;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
index 1be1727..1071831 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
@@ -146,7 +146,7 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
private ConcurrentMap<NodeId, List<ContainerStatus>>
finishedContainersSentToAM =
new ConcurrentHashMap<NodeId, List<ContainerStatus>>();
- private Container masterContainer;
+ private volatile Container masterContainer;
private float progress = 0;
private String host = "N/A";
@@ -762,13 +762,7 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
@Override
public Container getMasterContainer() {
- this.readLock.lock();
-
- try {
- return this.masterContainer;
- } finally {
- this.readLock.unlock();
- }
+ return this.masterContainer;
}
@InterfaceAudience.Private
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
index 5521d47..5604f0f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
@@ -73,10 +73,11 @@ public class AppSchedulingInfo {
/* Allocated by scheduler */
boolean pending = true; // for app metrics
+ private ResourceUsage appResourceUsage;
public AppSchedulingInfo(ApplicationAttemptId appAttemptId,
String user, Queue queue, ActiveUsersManager activeUsersManager,
- long epoch) {
+ long epoch, ResourceUsage appResourceUsage) {
this.applicationAttemptId = appAttemptId;
this.applicationId = appAttemptId.getApplicationId();
this.queue = queue;
@@ -84,6 +85,7 @@ public class AppSchedulingInfo {
this.user = user;
this.activeUsersManager = activeUsersManager;
this.containerIdCounter = new AtomicLong(epoch << EPOCH_BIT_SHIFT);
+ this.appResourceUsage = appResourceUsage;
}
public ApplicationId getApplicationId() {
@@ -191,13 +193,19 @@ public class AppSchedulingInfo {
lastRequestCapability);
// update queue:
+ Resource increasedResource = Resources.multiply(request.getCapability(),
+ request.getNumContainers());
queue.incPendingResource(
request.getNodeLabelExpression(),
- Resources.multiply(request.getCapability(),
- request.getNumContainers()));
+ increasedResource);
+ appResourceUsage.incPending(request.getNodeLabelExpression(), increasedResource);
if (lastRequest != null) {
+ Resource decreasedResource =
+ Resources.multiply(lastRequestCapability, lastRequestContainers);
queue.decPendingResource(lastRequest.getNodeLabelExpression(),
- Resources.multiply(lastRequestCapability, lastRequestContainers));
+ decreasedResource);
+ appResourceUsage.decPending(lastRequest.getNodeLabelExpression(),
+ decreasedResource);
}
}
}
@@ -385,6 +393,8 @@ public class AppSchedulingInfo {
checkForDeactivation();
}
+ appResourceUsage.decPending(offSwitchRequest.getNodeLabelExpression(),
+ offSwitchRequest.getCapability());
queue.decPendingResource(offSwitchRequest.getNodeLabelExpression(),
offSwitchRequest.getCapability());
}
@@ -492,9 +502,10 @@ public class AppSchedulingInfo {
}
public ResourceRequest cloneResourceRequest(ResourceRequest request) {
- ResourceRequest newRequest = ResourceRequest.newInstance(
- request.getPriority(), request.getResourceName(),
- request.getCapability(), 1, request.getRelaxLocality());
+ ResourceRequest newRequest =
+ ResourceRequest.newInstance(request.getPriority(),
+ request.getResourceName(), request.getCapability(), 1,
+ request.getRelaxLocality(), request.getNodeLabelExpression());
return newRequest;
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ResourceUsage.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ResourceUsage.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ResourceUsage.java
index 36ee4da..5169b78 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ResourceUsage.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ResourceUsage.java
@@ -27,6 +27,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
+import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import org.apache.hadoop.yarn.util.resource.Resources;
/**
@@ -250,6 +251,10 @@ public class ResourceUsage {
}
private Resource _get(String label, ResourceType type) {
+ if (label == null) {
+ label = RMNodeLabelsManager.NO_LABEL;
+ }
+
try {
readLock.lock();
UsageByLabel usage = usages.get(label);
@@ -263,6 +268,9 @@ public class ResourceUsage {
}
private UsageByLabel getAndAddIfMissing(String label) {
+ if (label == null) {
+ label = RMNodeLabelsManager.NO_LABEL;
+ }
if (!usages.containsKey(label)) {
UsageByLabel u = new UsageByLabel(label);
usages.put(label, u);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
index 5e0bbc7..fccf766 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
@@ -56,6 +56,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerReservedEvent;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanContainerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
+import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import com.google.common.base.Preconditions;
@@ -108,14 +110,24 @@ public class SchedulerApplicationAttempt {
private Set<ContainerId> pendingRelease = null;
/**
- * Count how many times the application has been given an opportunity
- * to schedule a task at each priority. Each time the scheduler
- * asks the application for a task at this priority, it is incremented,
- * and each time the application successfully schedules a task, it
+ * Count how many times the application has been given an opportunity to
+ * schedule a task at each priority. Each time the scheduler asks the
+ * application for a task at this priority, it is incremented, and each time
+ * the application successfully schedules a task (at rack or node local), it
* is reset to 0.
*/
Multiset<Priority> schedulingOpportunities = HashMultiset.create();
+ /**
+ * Count how many times the application has been given an opportunity to
+ * schedule a non-partitioned resource request at each priority. Each time the
+ * scheduler asks the application for a task at this priority, it is
+ * incremented, and each time the application successfully schedules a task,
+ * it is reset to 0 when schedule any task at corresponding priority.
+ */
+ Multiset<Priority> missedNonPartitionedRequestSchedulingOpportunity =
+ HashMultiset.create();
+
// Time of the last container scheduled at the current allowed level
protected Map<Priority, Long> lastScheduledContainer =
new HashMap<Priority, Long>();
@@ -132,7 +144,7 @@ public class SchedulerApplicationAttempt {
this.rmContext = rmContext;
this.appSchedulingInfo =
new AppSchedulingInfo(applicationAttemptId, user, queue,
- activeUsersManager, rmContext.getEpoch());
+ activeUsersManager, rmContext.getEpoch(), attemptResourceUsage);
this.queue = queue;
this.pendingRelease = new HashSet<ContainerId>();
this.attemptId = applicationAttemptId;
@@ -489,6 +501,18 @@ public class SchedulerApplicationAttempt {
return this.appSchedulingInfo.isBlacklisted(resourceName);
}
+ public synchronized int addMissedNonPartitionedRequestSchedulingOpportunity(
+ Priority priority) {
+ missedNonPartitionedRequestSchedulingOpportunity.add(priority);
+ return missedNonPartitionedRequestSchedulingOpportunity.count(priority);
+ }
+
+ public synchronized void
+ resetMissedNonPartitionedRequestSchedulingOpportunity(Priority priority) {
+ missedNonPartitionedRequestSchedulingOpportunity.setCount(priority, 0);
+ }
+
+
public synchronized void addSchedulingOpportunity(Priority priority) {
schedulingOpportunities.setCount(priority,
schedulingOpportunities.count(priority) + 1);
@@ -518,6 +542,7 @@ public class SchedulerApplicationAttempt {
public synchronized void resetSchedulingOpportunities(Priority priority) {
resetSchedulingOpportunities(priority, System.currentTimeMillis());
}
+
// used for continuous scheduling
public synchronized void resetSchedulingOpportunities(Priority priority,
long currentTimeMs) {
@@ -669,4 +694,13 @@ public class SchedulerApplicationAttempt {
public Set<String> getBlacklistedNodes() {
return this.appSchedulingInfo.getBlackListCopy();
}
+
+ @Private
+ public boolean hasPendingResourceRequest(ResourceCalculator rc,
+ String nodePartition, Resource cluster,
+ SchedulingMode schedulingMode) {
+ return SchedulerUtils.hasPendingResourceRequest(rc,
+ this.attemptResourceUsage, nodePartition, cluster,
+ schedulingMode);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
index 248cc08..7a1a528 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerUtils.java
@@ -37,11 +37,10 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.security.AccessType;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
-import com.google.common.collect.Sets;
-
/**
* Utilities shared by schedulers.
*/
@@ -235,9 +234,13 @@ public class SchedulerUtils {
if (labelExp == null && queueInfo != null
&& ResourceRequest.ANY.equals(resReq.getResourceName())) {
labelExp = queueInfo.getDefaultNodeLabelExpression();
- resReq.setNodeLabelExpression(labelExp);
}
+ // If labelExp still equals to null, set it to be NO_LABEL
+ resReq
+ .setNodeLabelExpression(labelExp == null ? RMNodeLabelsManager.NO_LABEL
+ : labelExp);
+
// we don't allow specify label expression other than resourceName=ANY now
if (!ResourceRequest.ANY.equals(resReq.getResourceName())
&& labelExp != null && !labelExp.trim().isEmpty()) {
@@ -273,25 +276,6 @@ public class SchedulerUtils {
}
}
- public static boolean checkQueueAccessToNode(Set<String> queueLabels,
- Set<String> nodeLabels) {
- // if queue's label is *, it can access any node
- if (queueLabels != null && queueLabels.contains(RMNodeLabelsManager.ANY)) {
- return true;
- }
- // any queue can access to a node without label
- if (nodeLabels == null || nodeLabels.isEmpty()) {
- return true;
- }
- // a queue can access to a node only if it contains any label of the node
- if (queueLabels != null
- && Sets.intersection(queueLabels, nodeLabels).size() > 0) {
- return true;
- }
- // sorry, you cannot access
- return false;
- }
-
public static void checkIfLabelInClusterNodeLabels(RMNodeLabelsManager mgr,
Set<String> labels) throws IOException {
if (mgr == null) {
@@ -311,26 +295,6 @@ public class SchedulerUtils {
}
}
}
-
- public static boolean checkNodeLabelExpression(Set<String> nodeLabels,
- String labelExpression) {
- // empty label expression can only allocate on node with empty labels
- if (labelExpression == null || labelExpression.trim().isEmpty()) {
- if (!nodeLabels.isEmpty()) {
- return false;
- }
- }
-
- if (labelExpression != null) {
- for (String str : labelExpression.split("&&")) {
- if (!str.trim().isEmpty()
- && (nodeLabels == null || !nodeLabels.contains(str.trim()))) {
- return false;
- }
- }
- }
- return true;
- }
public static boolean checkQueueLabelExpression(Set<String> queueLabels,
String labelExpression) {
@@ -360,4 +324,43 @@ public class SchedulerUtils {
}
return null;
}
+
+ public static boolean checkResourceRequestMatchingNodePartition(
+ ResourceRequest offswitchResourceRequest, String nodePartition,
+ SchedulingMode schedulingMode) {
+ // We will only look at node label = nodeLabelToLookAt according to
+ // schedulingMode and partition of node.
+ String nodePartitionToLookAt = null;
+ if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY) {
+ nodePartitionToLookAt = nodePartition;
+ } else {
+ nodePartitionToLookAt = RMNodeLabelsManager.NO_LABEL;
+ }
+
+ String askedNodePartition = offswitchResourceRequest.getNodeLabelExpression();
+ if (null == askedNodePartition) {
+ askedNodePartition = RMNodeLabelsManager.NO_LABEL;
+ }
+ return askedNodePartition.equals(nodePartitionToLookAt);
+ }
+
+ private static boolean hasPendingResourceRequest(ResourceCalculator rc,
+ ResourceUsage usage, String partitionToLookAt, Resource cluster) {
+ if (Resources.greaterThan(rc, cluster,
+ usage.getPending(partitionToLookAt), Resources.none())) {
+ return true;
+ }
+ return false;
+ }
+
+ @Private
+ public static boolean hasPendingResourceRequest(ResourceCalculator rc,
+ ResourceUsage usage, String nodePartition, Resource cluster,
+ SchedulingMode schedulingMode) {
+ String partitionToLookAt = nodePartition;
+ if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) {
+ partitionToLookAt = RMNodeLabelsManager.NO_LABEL;
+ }
+ return hasPendingResourceRequest(rc, usage, partitionToLookAt, cluster);
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java
index 42ea089..d95c45c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import java.io.IOException;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -38,12 +37,12 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
import org.apache.hadoop.yarn.security.AccessType;
import org.apache.hadoop.yarn.security.PrivilegedEntity;
import org.apache.hadoop.yarn.security.PrivilegedEntity.EntityType;
import org.apache.hadoop.yarn.security.YarnAuthorizationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
@@ -56,6 +55,11 @@ import com.google.common.collect.Sets;
public abstract class AbstractCSQueue implements CSQueue {
private static final Log LOG = LogFactory.getLog(AbstractCSQueue.class);
+ static final CSAssignment NULL_ASSIGNMENT =
+ new CSAssignment(Resources.createResource(0, 0), NodeType.NODE_LOCAL);
+
+ static final CSAssignment SKIP_ASSIGNMENT = new CSAssignment(true);
+
CSQueue parent;
final String queueName;
volatile int numContainers;
@@ -343,16 +347,8 @@ public abstract class AbstractCSQueue implements CSQueue {
}
synchronized void allocateResource(Resource clusterResource,
- Resource resource, Set<String> nodeLabels) {
-
- // Update usedResources by labels
- if (nodeLabels == null || nodeLabels.isEmpty()) {
- queueUsage.incUsed(resource);
- } else {
- for (String label : Sets.intersection(accessibleLabels, nodeLabels)) {
- queueUsage.incUsed(label, resource);
- }
- }
+ Resource resource, String nodePartition) {
+ queueUsage.incUsed(nodePartition, resource);
++numContainers;
CSQueueUtils.updateQueueStatistics(resourceCalculator, this, getParent(),
@@ -360,15 +356,8 @@ public abstract class AbstractCSQueue implements CSQueue {
}
protected synchronized void releaseResource(Resource clusterResource,
- Resource resource, Set<String> nodeLabels) {
- // Update usedResources by labels
- if (null == nodeLabels || nodeLabels.isEmpty()) {
- queueUsage.decUsed(resource);
- } else {
- for (String label : Sets.intersection(accessibleLabels, nodeLabels)) {
- queueUsage.decUsed(label, resource);
- }
- }
+ Resource resource, String nodePartition) {
+ queueUsage.decUsed(nodePartition, resource);
CSQueueUtils.updateQueueStatistics(resourceCalculator, this, getParent(),
clusterResource, minimumAllocation);
@@ -434,103 +423,108 @@ public abstract class AbstractCSQueue implements CSQueue {
parentQ.getPreemptionDisabled());
}
- private Resource getCurrentLimitResource(String nodeLabel,
- Resource clusterResource, ResourceLimits currentResourceLimits) {
- /*
- * Current limit resource: For labeled resource: limit = queue-max-resource
- * (TODO, this part need update when we support labeled-limit) For
- * non-labeled resource: limit = min(queue-max-resource,
- * limit-set-by-parent)
- */
- Resource queueMaxResource =
- Resources.multiplyAndNormalizeDown(resourceCalculator,
- labelManager.getResourceByLabel(nodeLabel, clusterResource),
- queueCapacities.getAbsoluteMaximumCapacity(nodeLabel), minimumAllocation);
- if (nodeLabel.equals(RMNodeLabelsManager.NO_LABEL)) {
- return Resources.min(resourceCalculator, clusterResource,
- queueMaxResource, currentResourceLimits.getLimit());
+ private Resource getCurrentLimitResource(String nodePartition,
+ Resource clusterResource, ResourceLimits currentResourceLimits,
+ SchedulingMode schedulingMode) {
+ if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY) {
+ /*
+ * Current limit resource: For labeled resource: limit = queue-max-resource
+ * (TODO, this part need update when we support labeled-limit) For
+ * non-labeled resource: limit = min(queue-max-resource,
+ * limit-set-by-parent)
+ */
+ Resource queueMaxResource =
+ Resources.multiplyAndNormalizeDown(resourceCalculator,
+ labelManager.getResourceByLabel(nodePartition, clusterResource),
+ queueCapacities.getAbsoluteMaximumCapacity(nodePartition), minimumAllocation);
+ if (nodePartition.equals(RMNodeLabelsManager.NO_LABEL)) {
+ return Resources.min(resourceCalculator, clusterResource,
+ queueMaxResource, currentResourceLimits.getLimit());
+ }
+ return queueMaxResource;
+ } else if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) {
+ // When we doing non-exclusive resource allocation, maximum capacity of
+ // all queues on this label equals to total resource with the label.
+ return labelManager.getResourceByLabel(nodePartition, clusterResource);
}
- return queueMaxResource;
+
+ return Resources.none();
}
synchronized boolean canAssignToThisQueue(Resource clusterResource,
- Set<String> nodeLabels, ResourceLimits currentResourceLimits,
- Resource nowRequired, Resource resourceCouldBeUnreserved) {
- // Get label of this queue can access, it's (nodeLabel AND queueLabel)
- Set<String> labelCanAccess;
- if (null == nodeLabels || nodeLabels.isEmpty()) {
- labelCanAccess = new HashSet<String>();
- // Any queue can always access any node without label
- labelCanAccess.add(RMNodeLabelsManager.NO_LABEL);
- } else {
- labelCanAccess = new HashSet<String>(
- accessibleLabels.contains(CommonNodeLabelsManager.ANY) ? nodeLabels
- : Sets.intersection(accessibleLabels, nodeLabels));
- }
-
- for (String label : labelCanAccess) {
- // New total resource = used + required
- Resource newTotalResource =
- Resources.add(queueUsage.getUsed(label), nowRequired);
-
- Resource currentLimitResource =
- getCurrentLimitResource(label, clusterResource, currentResourceLimits);
-
- // if reservation continous looking enabled, check to see if could we
- // potentially use this node instead of a reserved node if the application
- // has reserved containers.
- // TODO, now only consider reservation cases when the node has no label
- if (this.reservationsContinueLooking
- && label.equals(RMNodeLabelsManager.NO_LABEL)
- && Resources.greaterThan(resourceCalculator, clusterResource,
- resourceCouldBeUnreserved, Resources.none())) {
- // resource-without-reserved = used - reserved
- Resource newTotalWithoutReservedResource =
- Resources.subtract(newTotalResource, resourceCouldBeUnreserved);
-
- // when total-used-without-reserved-resource < currentLimit, we still
- // have chance to allocate on this node by unreserving some containers
- if (Resources.lessThan(resourceCalculator, clusterResource,
- newTotalWithoutReservedResource, currentLimitResource)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("try to use reserved: " + getQueueName()
- + " usedResources: " + queueUsage.getUsed()
- + ", clusterResources: " + clusterResource
- + ", reservedResources: " + resourceCouldBeUnreserved
- + ", capacity-without-reserved: "
- + newTotalWithoutReservedResource + ", maxLimitCapacity: "
- + currentLimitResource);
- }
- return true;
+ String nodePartition, ResourceLimits currentResourceLimits,
+ Resource nowRequired, Resource resourceCouldBeUnreserved,
+ SchedulingMode schedulingMode) {
+ // New total resource = used + required
+ Resource newTotalResource =
+ Resources.add(queueUsage.getUsed(nodePartition), nowRequired);
+
+ // Get current limited resource:
+ // - When doing RESPECT_PARTITION_EXCLUSIVITY allocation, we will respect
+ // queues' max capacity.
+ // - When doing IGNORE_PARTITION_EXCLUSIVITY allocation, we will not respect
+ // queue's max capacity, queue's max capacity on the partition will be
+ // considered to be 100%. Which is a queue can use all resource in the
+ // partition.
+ // Doing this because: for non-exclusive allocation, we make sure there's
+ // idle resource on the partition, to avoid wastage, such resource will be
+ // leveraged as much as we can, and preemption policy will reclaim it back
+ // when partitoned-resource-request comes back.
+ Resource currentLimitResource =
+ getCurrentLimitResource(nodePartition, clusterResource,
+ currentResourceLimits, schedulingMode);
+
+ // if reservation continous looking enabled, check to see if could we
+ // potentially use this node instead of a reserved node if the application
+ // has reserved containers.
+ // TODO, now only consider reservation cases when the node has no label
+ if (this.reservationsContinueLooking
+ && nodePartition.equals(RMNodeLabelsManager.NO_LABEL)
+ && Resources.greaterThan(resourceCalculator, clusterResource,
+ resourceCouldBeUnreserved, Resources.none())) {
+ // resource-without-reserved = used - reserved
+ Resource newTotalWithoutReservedResource =
+ Resources.subtract(newTotalResource, resourceCouldBeUnreserved);
+
+ // when total-used-without-reserved-resource < currentLimit, we still
+ // have chance to allocate on this node by unreserving some containers
+ if (Resources.lessThan(resourceCalculator, clusterResource,
+ newTotalWithoutReservedResource, currentLimitResource)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("try to use reserved: " + getQueueName()
+ + " usedResources: " + queueUsage.getUsed()
+ + ", clusterResources: " + clusterResource
+ + ", reservedResources: " + resourceCouldBeUnreserved
+ + ", capacity-without-reserved: "
+ + newTotalWithoutReservedResource + ", maxLimitCapacity: "
+ + currentLimitResource);
}
+ return true;
}
-
- // Otherwise, if any of the label of this node beyond queue limit, we
- // cannot allocate on this node. Consider a small epsilon here.
- if (Resources.greaterThan(resourceCalculator, clusterResource,
- newTotalResource, currentLimitResource)) {
- return false;
- }
+ }
- if (LOG.isDebugEnabled()) {
- LOG.debug(getQueueName()
- + "Check assign to queue, label=" + label
- + " usedResources: " + queueUsage.getUsed(label)
- + " clusterResources: " + clusterResource
- + " currentUsedCapacity "
- + Resources.divide(resourceCalculator, clusterResource,
- queueUsage.getUsed(label),
- labelManager.getResourceByLabel(label, clusterResource))
- + " max-capacity: "
- + queueCapacities.getAbsoluteMaximumCapacity(label)
- + ")");
- }
- return true;
+ // Check if we over current-resource-limit computed.
+ if (Resources.greaterThan(resourceCalculator, clusterResource,
+ newTotalResource, currentLimitResource)) {
+ return false;
}
-
- // Actually, this will not happen, since labelCanAccess will be always
- // non-empty
- return false;
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(getQueueName()
+ + "Check assign to queue, nodePartition="
+ + nodePartition
+ + " usedResources: "
+ + queueUsage.getUsed(nodePartition)
+ + " clusterResources: "
+ + clusterResource
+ + " currentUsedCapacity "
+ + Resources.divide(resourceCalculator, clusterResource,
+ queueUsage.getUsed(nodePartition),
+ labelManager.getResourceByLabel(nodePartition, clusterResource))
+ + " max-capacity: "
+ + queueCapacities.getAbsoluteMaximumCapacity(nodePartition) + ")");
+ }
+ return true;
}
@Override
@@ -556,4 +550,33 @@ public abstract class AbstractCSQueue implements CSQueue {
parent.decPendingResource(nodeLabel, resourceToDec);
}
}
+
+ /**
+ * Return if the queue has pending resource on given nodePartition and
+ * schedulingMode.
+ */
+ boolean hasPendingResourceRequest(String nodePartition,
+ Resource cluster, SchedulingMode schedulingMode) {
+ return SchedulerUtils.hasPendingResourceRequest(resourceCalculator,
+ queueUsage, nodePartition, cluster, schedulingMode);
+ }
+
+ boolean accessibleToPartition(String nodePartition) {
+ // if queue's label is *, it can access any node
+ if (accessibleLabels != null
+ && accessibleLabels.contains(RMNodeLabelsManager.ANY)) {
+ return true;
+ }
+ // any queue can access to a node without label
+ if (nodePartition == null
+ || nodePartition.equals(RMNodeLabelsManager.NO_LABEL)) {
+ return true;
+ }
+ // a queue can access to a node only if it contains any label of the node
+ if (accessibleLabels != null && accessibleLabels.contains(nodePartition)) {
+ return true;
+ }
+ // sorry, you cannot access
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java
index 1a9448a..b06a646 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java
@@ -190,10 +190,13 @@ extends org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue {
* @param clusterResource the resource of the cluster.
* @param node node on which resources are available
* @param resourceLimits how much overall resource of this queue can use.
+ * @param schedulingMode Type of exclusive check when assign container on a
+ * NodeManager, see {@link SchedulingMode}.
* @return the assignment
*/
public CSAssignment assignContainers(Resource clusterResource,
- FiCaSchedulerNode node, ResourceLimits resourceLimits);
+ FiCaSchedulerNode node, ResourceLimits resourceLimits,
+ SchedulingMode schedulingMode);
/**
* A container assigned to the queue has completed.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
index e93c529..cfeee37 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
@@ -35,6 +35,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
@@ -1114,28 +1115,30 @@ public class CapacityScheduler extends
if (reservedContainer != null) {
FiCaSchedulerApp reservedApplication =
getCurrentAttemptForContainer(reservedContainer.getContainerId());
-
+
// Try to fulfill the reservation
- LOG.info("Trying to fulfill reservation for application " +
- reservedApplication.getApplicationId() + " on node: " +
- node.getNodeID());
-
- LeafQueue queue = ((LeafQueue)reservedApplication.getQueue());
- assignment = queue.assignContainers(
+ LOG.info("Trying to fulfill reservation for application "
+ + reservedApplication.getApplicationId() + " on node: "
+ + node.getNodeID());
+
+ LeafQueue queue = ((LeafQueue) reservedApplication.getQueue());
+ assignment =
+ queue.assignContainers(
clusterResource,
node,
// TODO, now we only consider limits for parent for non-labeled
// resources, should consider labeled resources as well.
new ResourceLimits(labelManager.getResourceByLabel(
- RMNodeLabelsManager.NO_LABEL, clusterResource)));
+ RMNodeLabelsManager.NO_LABEL, clusterResource)),
+ SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
if (assignment.isFulfilledReservation()) {
CSAssignment tmp =
new CSAssignment(reservedContainer.getReservedResource(),
- assignment.getType());
+ assignment.getType());
Resources.addTo(assignment.getAssignmentInformation().getAllocated(),
- reservedContainer.getReservedResource());
+ reservedContainer.getReservedResource());
tmp.getAssignmentInformation().addAllocationDetails(
- reservedContainer.getContainerId(), queue.getQueuePath());
+ reservedContainer.getContainerId(), queue.getQueuePath());
tmp.getAssignmentInformation().incrAllocations();
updateSchedulerHealth(lastNodeUpdateTime, node, tmp);
schedulerHealth.updateSchedulerFulfilledReservationCounts(1);
@@ -1143,16 +1146,13 @@ public class CapacityScheduler extends
RMContainer excessReservation = assignment.getExcessReservation();
if (excessReservation != null) {
- Container container = excessReservation.getContainer();
- queue.completedContainer(
- clusterResource, assignment.getApplication(), node,
- excessReservation,
- SchedulerUtils.createAbnormalContainerStatus(
- container.getId(),
- SchedulerUtils.UNRESERVED_CONTAINER),
- RMContainerEventType.RELEASED, null, true);
+ Container container = excessReservation.getContainer();
+ queue.completedContainer(clusterResource, assignment.getApplication(),
+ node, excessReservation, SchedulerUtils
+ .createAbnormalContainerStatus(container.getId(),
+ SchedulerUtils.UNRESERVED_CONTAINER),
+ RMContainerEventType.RELEASED, null, true);
}
-
}
// Try to schedule more if there are no reservations to fulfill
@@ -1163,22 +1163,61 @@ public class CapacityScheduler extends
LOG.debug("Trying to schedule on node: " + node.getNodeName() +
", available: " + node.getAvailableResource());
}
+
assignment = root.assignContainers(
clusterResource,
node,
// TODO, now we only consider limits for parent for non-labeled
// resources, should consider labeled resources as well.
new ResourceLimits(labelManager.getResourceByLabel(
- RMNodeLabelsManager.NO_LABEL, clusterResource)));
+ RMNodeLabelsManager.NO_LABEL, clusterResource)),
+ SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
+ if (Resources.greaterThan(calculator, clusterResource,
+ assignment.getResource(), Resources.none())) {
+ updateSchedulerHealth(lastNodeUpdateTime, node, assignment);
+ return;
+ }
+
+ // Only do non-exclusive allocation when node has node-labels.
+ if (StringUtils.equals(node.getPartition(),
+ RMNodeLabelsManager.NO_LABEL)) {
+ return;
+ }
+
+ // Only do non-exclusive allocation when the node-label supports that
+ try {
+ if (rmContext.getNodeLabelManager().isExclusiveNodeLabel(
+ node.getPartition())) {
+ return;
+ }
+ } catch (IOException e) {
+ LOG.warn("Exception when trying to get exclusivity of node label="
+ + node.getPartition(), e);
+ return;
+ }
+
+ // Try to use NON_EXCLUSIVE
+ assignment = root.assignContainers(
+ clusterResource,
+ node,
+ // TODO, now we only consider limits for parent for non-labeled
+ // resources, should consider labeled resources as well.
+ new ResourceLimits(labelManager.getResourceByLabel(
+ RMNodeLabelsManager.NO_LABEL, clusterResource)),
+ SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY);
updateSchedulerHealth(lastNodeUpdateTime, node, assignment);
+ if (Resources.greaterThan(calculator, clusterResource,
+ assignment.getResource(), Resources.none())) {
+ return;
+ }
}
} else {
- LOG.info("Skipping scheduling since node " + node.getNodeID() +
- " is reserved by application " +
- node.getReservedContainer().getContainerId().getApplicationAttemptId()
- );
+ LOG.info("Skipping scheduling since node "
+ + node.getNodeID()
+ + " is reserved by application "
+ + node.getReservedContainer().getContainerId()
+ .getApplicationAttemptId());
}
-
}
@Override
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
index 102e553..4e8d617 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
@@ -319,6 +319,11 @@ public class CapacitySchedulerConfiguration extends ReservationSchedulerConfigur
getMaximumApplicationMasterResourcePercent());
}
+ public void setMaximumApplicationMasterResourcePerQueuePercent(String queue,
+ float percent) {
+ setFloat(getQueuePrefix(queue) + MAXIMUM_AM_RESOURCE_SUFFIX, percent);
+ }
+
public float getNonLabeledQueueCapacity(String queue) {
float capacity = queue.equals("root") ? 100.0f : getFloat(
getQueuePrefix(queue) + CAPACITY, UNDEFINED);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/025787b3/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
index 59a016f..8a6a601 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
@@ -24,7 +24,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -58,6 +57,7 @@ import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
import org.apache.hadoop.yarn.security.AccessType;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
@@ -718,39 +718,11 @@ public class LeafQueue extends AbstractCSQueue {
ApplicationAttemptId applicationAttemptId) {
return applicationAttemptMap.get(applicationAttemptId);
}
-
- private static final CSAssignment NULL_ASSIGNMENT =
- new CSAssignment(Resources.createResource(0, 0), NodeType.NODE_LOCAL);
-
- private static final CSAssignment SKIP_ASSIGNMENT = new CSAssignment(true);
-
- private static Set<String> getRequestLabelSetByExpression(
- String labelExpression) {
- Set<String> labels = new HashSet<String>();
- if (null == labelExpression) {
- return labels;
- }
- for (String l : labelExpression.split("&&")) {
- if (l.trim().isEmpty()) {
- continue;
- }
- labels.add(l.trim());
- }
- return labels;
- }
-
- private boolean checkResourceRequestMatchingNodeLabel(ResourceRequest offswitchResourceRequest,
- FiCaSchedulerNode node) {
- String askedNodeLabel = offswitchResourceRequest.getNodeLabelExpression();
- if (null == askedNodeLabel) {
- askedNodeLabel = RMNodeLabelsManager.NO_LABEL;
- }
- return askedNodeLabel.equals(node.getPartition());
- }
@Override
public synchronized CSAssignment assignContainers(Resource clusterResource,
- FiCaSchedulerNode node, ResourceLimits currentResourceLimits) {
+ FiCaSchedulerNode node, ResourceLimits currentResourceLimits,
+ SchedulingMode schedulingMode) {
updateCurrentResourceLimits(currentResourceLimits, clusterResource);
if(LOG.isDebugEnabled()) {
@@ -758,12 +730,6 @@ public class LeafQueue extends AbstractCSQueue {
+ " #applications=" + activeApplications.size());
}
- // if our queue cannot access this node, just return
- if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels,
- node.getLabels())) {
- return NULL_ASSIGNMENT;
- }
-
// Check for reserved resources
RMContainer reservedContainer = node.getReservedContainer();
if (reservedContainer != null) {
@@ -771,8 +737,26 @@ public class LeafQueue extends AbstractCSQueue {
getApplication(reservedContainer.getApplicationAttemptId());
synchronized (application) {
return assignReservedContainer(application, node, reservedContainer,
- clusterResource);
+ clusterResource, schedulingMode);
+ }
+ }
+
+ // if our queue cannot access this node, just return
+ if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY
+ && !accessibleToPartition(node.getPartition())) {
+ return NULL_ASSIGNMENT;
+ }
+
+ // Check if this queue need more resource, simply skip allocation if this
+ // queue doesn't need more resources.
+ if (!hasPendingResourceRequest(node.getPartition(),
+ clusterResource, schedulingMode)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Skip this queue=" + getQueuePath()
+ + ", because it doesn't need more resource, schedulingMode="
+ + schedulingMode.name() + " node-partition=" + node.getPartition());
}
+ return NULL_ASSIGNMENT;
}
// Try to assign containers to applications in order
@@ -783,6 +767,17 @@ public class LeafQueue extends AbstractCSQueue {
+ application.getApplicationId());
application.showRequests();
}
+
+ // Check if application needs more resource, skip if it doesn't need more.
+ if (!application.hasPendingResourceRequest(resourceCalculator,
+ node.getPartition(), clusterResource, schedulingMode)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Skip app_attempt=" + application.getApplicationAttemptId()
+ + ", because it doesn't need more resource, schedulingMode="
+ + schedulingMode.name() + " node-label=" + node.getPartition());
+ }
+ continue;
+ }
synchronized (application) {
// Check if this resource is on the blacklist
@@ -806,10 +801,27 @@ public class LeafQueue extends AbstractCSQueue {
continue;
}
+ // AM container allocation doesn't support non-exclusive allocation to
+ // avoid painful of preempt an AM container
+ if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) {
+ RMAppAttempt rmAppAttempt =
+ csContext.getRMContext().getRMApps()
+ .get(application.getApplicationId()).getCurrentAppAttempt();
+ if (null == rmAppAttempt.getMasterContainer()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Skip allocating AM container to app_attempt="
+ + application.getApplicationAttemptId()
+ + ", don't allow to allocate AM container in non-exclusive mode");
+ }
+ break;
+ }
+ }
+
// Is the node-label-expression of this offswitch resource request
// matches the node's label?
// If not match, jump to next priority.
- if (!checkResourceRequestMatchingNodeLabel(anyRequest, node)) {
+ if (!SchedulerUtils.checkResourceRequestMatchingNodePartition(
+ anyRequest, node.getPartition(), schedulingMode)) {
continue;
}
@@ -822,10 +834,6 @@ public class LeafQueue extends AbstractCSQueue {
}
}
- Set<String> requestedNodeLabels =
- getRequestLabelSetByExpression(anyRequest
- .getNodeLabelExpression());
-
// Compute user-limit & set headroom
// Note: We compute both user-limit & headroom with the highest
// priority request as the target.
@@ -833,27 +841,61 @@ public class LeafQueue extends AbstractCSQueue {
// before all higher priority ones are serviced.
Resource userLimit =
computeUserLimitAndSetHeadroom(application, clusterResource,
- required, requestedNodeLabels);
+ required, node.getPartition(), schedulingMode);
// Check queue max-capacity limit
- if (!super.canAssignToThisQueue(clusterResource, node.getLabels(),
- this.currentResourceLimits, required, application.getCurrentReservation())) {
+ if (!super.canAssignToThisQueue(clusterResource, node.getPartition(),
+ this.currentResourceLimits, required,
+ application.getCurrentReservation(), schedulingMode)) {
return NULL_ASSIGNMENT;
}
// Check user limit
if (!canAssignToUser(clusterResource, application.getUser(), userLimit,
- application, true, requestedNodeLabels)) {
+ application, true, node.getPartition())) {
break;
}
// Inform the application it is about to get a scheduling opportunity
application.addSchedulingOpportunity(priority);
+ // Increase missed-non-partitioned-resource-request-opportunity.
+ // This is to make sure non-partitioned-resource-request will prefer
+ // to be allocated to non-partitioned nodes
+ int missedNonPartitionedRequestSchedulingOpportunity = 0;
+ if (anyRequest.getNodeLabelExpression().equals(
+ RMNodeLabelsManager.NO_LABEL)) {
+ missedNonPartitionedRequestSchedulingOpportunity =
+ application
+ .addMissedNonPartitionedRequestSchedulingOpportunity(priority);
+ }
+
+ if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) {
+ // Before doing allocation, we need to check scheduling opportunity to
+ // make sure : non-partitioned resource request should be scheduled to
+ // non-partitioned partition first.
+ if (missedNonPartitionedRequestSchedulingOpportunity < scheduler
+ .getNumClusterNodes()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Skip app_attempt="
+ + application.getApplicationAttemptId()
+ + " priority="
+ + priority
+ + " because missed-non-partitioned-resource-request"
+ + " opportunity under requred:"
+ + " Now=" + missedNonPartitionedRequestSchedulingOpportunity
+ + " required="
+ + scheduler.getNumClusterNodes());
+ }
+
+ break;
+ }
+ }
+
// Try to schedule
CSAssignment assignment =
assignContainersOnNode(clusterResource, node, application, priority,
- null);
+ null, schedulingMode);
// Did the application skip this node?
if (assignment.getSkipped()) {
@@ -870,9 +912,9 @@ public class LeafQueue extends AbstractCSQueue {
// Book-keeping
// Note: Update headroom to account for current allocation too...
allocateResource(clusterResource, application, assigned,
- node.getLabels());
+ node.getPartition());
- // Don't reset scheduling opportunities for non-local assignments
+ // Don't reset scheduling opportunities for offswitch assignments
// otherwise the app will be delayed for each non-local assignment.
// This helps apps with many off-cluster requests schedule faster.
if (assignment.getType() != NodeType.OFF_SWITCH) {
@@ -881,6 +923,10 @@ public class LeafQueue extends AbstractCSQueue {
}
application.resetSchedulingOpportunities(priority);
}
+ // Non-exclusive scheduling opportunity is different: we need reset
+ // it every time to make sure non-labeled resource request will be
+ // most likely allocated on non-labeled nodes first.
+ application.resetMissedNonPartitionedRequestSchedulingOpportunity(priority);
// Done
return assignment;
@@ -904,7 +950,8 @@ public class LeafQueue extends AbstractCSQueue {
private synchronized CSAssignment assignReservedContainer(
FiCaSchedulerApp application, FiCaSchedulerNode node,
- RMContainer rmContainer, Resource clusterResource) {
+ RMContainer rmContainer, Resource clusterResource,
+ SchedulingMode schedulingMode) {
// Do we still need this reservation?
Priority priority = rmContainer.getReservedPriority();
if (application.getTotalRequiredResources(priority) == 0) {
@@ -915,7 +962,7 @@ public class LeafQueue extends AbstractCSQueue {
// Try to assign if we have sufficient resources
CSAssignment tmp =
assignContainersOnNode(clusterResource, node, application, priority,
- rmContainer);
+ rmContainer, schedulingMode);
// Doesn't matter... since it's already charged for at time of reservation
// "re-reservation" is *free*
@@ -929,7 +976,8 @@ public class LeafQueue extends AbstractCSQueue {
protected Resource getHeadroom(User user, Resource queueCurrentLimit,
Resource clusterResource, FiCaSchedulerApp application, Resource required) {
return getHeadroom(user, queueCurrentLimit, clusterResource,
- computeUserLimit(application, clusterResource, required, user, null));
+ computeUserLimit(application, clusterResource, required, user,
+ RMNodeLabelsManager.NO_LABEL, SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY));
}
private Resource getHeadroom(User user, Resource currentResourceLimit,
@@ -973,7 +1021,8 @@ public class LeafQueue extends AbstractCSQueue {
@Lock({LeafQueue.class, FiCaSchedulerApp.class})
Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application,
- Resource clusterResource, Resource required, Set<String> requestedLabels) {
+ Resource clusterResource, Resource required, String nodePartition,
+ SchedulingMode schedulingMode) {
String user = application.getUser();
User queueUser = getUser(user);
@@ -981,7 +1030,7 @@ public class LeafQueue extends AbstractCSQueue {
// TODO, need consider headroom respect labels also
Resource userLimit =
computeUserLimit(application, clusterResource, required,
- queueUser, requestedLabels);
+ queueUser, nodePartition, schedulingMode);
setQueueResourceLimitsInfo(clusterResource);
@@ -1010,34 +1059,18 @@ public class LeafQueue extends AbstractCSQueue {
@Lock(NoLock.class)
private Resource computeUserLimit(FiCaSchedulerApp application,
Resource clusterResource, Resource required, User user,
- Set<String> requestedLabels) {
+ String nodePartition, SchedulingMode schedulingMode) {
// What is our current capacity?
// * It is equal to the max(required, queue-capacity) if
// we're running below capacity. The 'max' ensures that jobs in queues
// with miniscule capacity (< 1 slot) make progress
// * If we're running over capacity, then its
// (usedResources + required) (which extra resources we are allocating)
- Resource queueCapacity = Resource.newInstance(0, 0);
- if (requestedLabels != null && !requestedLabels.isEmpty()) {
- // if we have multiple labels to request, we will choose to use the first
- // label
- String firstLabel = requestedLabels.iterator().next();
- queueCapacity =
- Resources
- .max(resourceCalculator, clusterResource, queueCapacity,
- Resources.multiplyAndNormalizeUp(resourceCalculator,
- labelManager.getResourceByLabel(firstLabel,
- clusterResource),
- queueCapacities.getAbsoluteCapacity(firstLabel),
- minimumAllocation));
- } else {
- // else there's no label on request, just to use absolute capacity as
- // capacity for nodes without label
- queueCapacity =
- Resources.multiplyAndNormalizeUp(resourceCalculator, labelManager
- .getResourceByLabel(CommonNodeLabelsManager.NO_LABEL, clusterResource),
- queueCapacities.getAbsoluteCapacity(), minimumAllocation);
- }
+ Resource queueCapacity =
+ Resources.multiplyAndNormalizeUp(resourceCalculator,
+ labelManager.getResourceByLabel(nodePartition, clusterResource),
+ queueCapacities.getAbsoluteCapacity(nodePartition),
+ minimumAllocation);
// Allow progress for queues with miniscule capacity
queueCapacity =
@@ -1047,33 +1080,56 @@ public class LeafQueue extends AbstractCSQueue {
required);
Resource currentCapacity =
- Resources.lessThan(resourceCalculator, clusterResource,
- queueUsage.getUsed(), queueCapacity) ?
- queueCapacity : Resources.add(queueUsage.getUsed(), required);
+ Resources.lessThan(resourceCalculator, clusterResource,
+ queueUsage.getUsed(nodePartition), queueCapacity) ? queueCapacity
+ : Resources.add(queueUsage.getUsed(nodePartition), required);
// Never allow a single user to take more than the
// queue's configured capacity * user-limit-factor.
// Also, the queue's configured capacity should be higher than
// queue-hard-limit * ulMin
- final int activeUsers = activeUsersManager.getNumActiveUsers();
-
- Resource limit =
+ final int activeUsers = activeUsersManager.getNumActiveUsers();
+
+ // User limit resource is determined by:
+ // max{currentCapacity / #activeUsers, currentCapacity * user-limit-percentage%)
+ Resource userLimitResource = Resources.max(
+ resourceCalculator, clusterResource,
+ Resources.divideAndCeil(
+ resourceCalculator, currentCapacity, activeUsers),
+ Resources.divideAndCeil(
+ resourceCalculator,
+ Resources.multiplyAndRoundDown(
+ currentCapacity, userLimit),
+ 100)
+ );
+
+ // User limit is capped by maxUserLimit
+ // - maxUserLimit = queueCapacity * user-limit-factor (RESPECT_PARTITION_EXCLUSIVITY)
+ // - maxUserLimit = total-partition-resource (IGNORE_PARTITION_EXCLUSIVITY)
+ //
+ // In IGNORE_PARTITION_EXCLUSIVITY mode, if a queue cannot access a
+ // partition, its guaranteed resource on that partition is 0. And
+ // user-limit-factor computation is based on queue's guaranteed capacity. So
+ // we will not cap user-limit as well as used resource when doing
+ // IGNORE_PARTITION_EXCLUSIVITY allocation.
+ Resource maxUserLimit = Resources.none();
+ if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY) {
+ maxUserLimit =
+ Resources.multiplyAndRoundDown(queueCapacity, userLimitFactor);
+ } else if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) {
+ maxUserLimit =
+ labelManager.getResourceByLabel(nodePartition, clusterResource);
+ }
+
+ // Cap final user limit with maxUserLimit
+ userLimitResource =
Resources.roundUp(
resourceCalculator,
Resources.min(
resourceCalculator, clusterResource,
- Resources.max(
- resourceCalculator, clusterResource,
- Resources.divideAndCeil(
- resourceCalculator, currentCapacity, activeUsers),
- Resources.divideAndCeil(
- resourceCalculator,
- Resources.multiplyAndRoundDown(
- currentCapacity, userLimit),
- 100)
- ),
- Resources.multiplyAndRoundDown(queueCapacity, userLimitFactor)
+ userLimitResource,
+ maxUserLimit
),
minimumAllocation);
@@ -1081,11 +1137,11 @@ public class LeafQueue extends AbstractCSQueue {
String userName = application.getUser();
LOG.debug("User limit computation for " + userName +
" in queue " + getQueueName() +
- " userLimit=" + userLimit +
+ " userLimitPercent=" + userLimit +
" userLimitFactor=" + userLimitFactor +
" required: " + required +
" consumed: " + user.getUsed() +
- " limit: " + limit +
+ " user-limit-resource: " + userLimitResource +
" queueCapacity: " + queueCapacity +
" qconsumed: " + queueUsage.getUsed() +
" currentCapacity: " + currentCapacity +
@@ -1093,31 +1149,26 @@ public class LeafQueue extends AbstractCSQueue {
" clusterCapacity: " + clusterResource
);
}
- user.setUserResourceLimit(limit);
- return limit;
+ user.setUserResourceLimit(userLimitResource);
+ return userLimitResource;
}
@Private
protected synchronized boolean canAssignToUser(Resource clusterResource,
String userName, Resource limit, FiCaSchedulerApp application,
- boolean checkReservations, Set<String> requestLabels) {
+ boolean checkReservations, String nodePartition) {
User user = getUser(userName);
-
- String label = CommonNodeLabelsManager.NO_LABEL;
- if (requestLabels != null && !requestLabels.isEmpty()) {
- label = requestLabels.iterator().next();
- }
// Note: We aren't considering the current request since there is a fixed
// overhead of the AM, but it's a > check, not a >= check, so...
if (Resources
.greaterThan(resourceCalculator, clusterResource,
- user.getUsed(label),
+ user.getUsed(nodePartition),
limit)) {
// if enabled, check to see if could we potentially use this node instead
// of a reserved node if the application has reserved containers
if (this.reservationsContinueLooking && checkReservations
- && label.equals(CommonNodeLabelsManager.NO_LABEL)) {
+ && nodePartition.equals(CommonNodeLabelsManager.NO_LABEL)) {
if (Resources.lessThanOrEqual(
resourceCalculator,
clusterResource,
@@ -1136,7 +1187,7 @@ public class LeafQueue extends AbstractCSQueue {
if (LOG.isDebugEnabled()) {
LOG.debug("User " + userName + " in queue " + getQueueName()
+ " will exceed limit - " + " consumed: "
- + user.getUsed() + " limit: " + limit);
+ + user.getUsed(nodePartition) + " limit: " + limit);
}
return false;
}
@@ -1176,7 +1227,7 @@ public class LeafQueue extends AbstractCSQueue {
private CSAssignment assignContainersOnNode(Resource clusterResource,
FiCaSchedulerNode node, FiCaSchedulerApp application, Priority priority,
- RMContainer reservedContainer) {
+ RMContainer reservedContainer, SchedulingMode schedulingMode) {
CSAssignment assigned;
@@ -1190,7 +1241,7 @@ public class LeafQueue extends AbstractCSQueue {
assigned =
assignNodeLocalContainers(clusterResource, nodeLocalResourceRequest,
node, application, priority, reservedContainer,
- allocatedContainer);
+ allocatedContainer, schedulingMode);
if (Resources.greaterThan(resourceCalculator, clusterResource,
assigned.getResource(), Resources.none())) {
@@ -1219,7 +1270,7 @@ public class LeafQueue extends AbstractCSQueue {
assigned =
assignRackLocalContainers(clusterResource, rackLocalResourceRequest,
node, application, priority, reservedContainer,
- allocatedContainer);
+ allocatedContainer, schedulingMode);
if (Resources.greaterThan(resourceCalculator, clusterResource,
assigned.getResource(), Resources.none())) {
@@ -1248,7 +1299,7 @@ public class LeafQueue extends AbstractCSQueue {
assigned =
assignOffSwitchContainers(clusterResource, offSwitchResourceRequest,
node, application, priority, reservedContainer,
- allocatedContainer);
+ allocatedContainer, schedulingMode);
// update locality statistics
if (allocatedContainer.getValue() != null) {
@@ -1314,16 +1365,17 @@ public class LeafQueue extends AbstractCSQueue {
@Private
protected boolean checkLimitsToReserve(Resource clusterResource,
- FiCaSchedulerApp application, Resource capability) {
+ FiCaSchedulerApp application, Resource capability, String nodePartition,
+ SchedulingMode schedulingMode) {
// we can't reserve if we got here based on the limit
// checks assuming we could unreserve!!!
Resource userLimit = computeUserLimitAndSetHeadroom(application,
- clusterResource, capability, null);
+ clusterResource, capability, nodePartition, schedulingMode);
// Check queue max-capacity limit,
// TODO: Consider reservation on labels
- if (!canAssignToThisQueue(clusterResource, null,
- this.currentResourceLimits, capability, Resources.none())) {
+ if (!canAssignToThisQueue(clusterResource, RMNodeLabelsManager.NO_LABEL,
+ this.currentResourceLimits, capability, Resources.none(), schedulingMode)) {
if (LOG.isDebugEnabled()) {
LOG.debug("was going to reserve but hit queue limit");
}
@@ -1332,7 +1384,7 @@ public class LeafQueue extends AbstractCSQueue {
// Check user limit
if (!canAssignToUser(clusterResource, application.getUser(), userLimit,
- application, false, null)) {
+ application, false, nodePartition)) {
if (LOG.isDebugEnabled()) {
LOG.debug("was going to reserve but hit user limit");
}
@@ -1345,12 +1397,13 @@ public class LeafQueue extends AbstractCSQueue {
private CSAssignment assignNodeLocalContainers(Resource clusterResource,
ResourceRequest nodeLocalResourceRequest, FiCaSchedulerNode node,
FiCaSchedulerApp application, Priority priority,
- RMContainer reservedContainer, MutableObject allocatedContainer) {
+ RMContainer reservedContainer, MutableObject allocatedContainer,
+ SchedulingMode schedulingMode) {
if (canAssign(application, priority, node, NodeType.NODE_LOCAL,
reservedContainer)) {
return assignContainer(clusterResource, node, application, priority,
nodeLocalResourceRequest, NodeType.NODE_LOCAL, reservedContainer,
- allocatedContainer);
+ allocatedContainer, schedulingMode);
}
return new CSAssignment(Resources.none(), NodeType.NODE_LOCAL);
@@ -1359,12 +1412,13 @@ public class LeafQueue extends AbstractCSQueue {
private CSAssignment assignRackLocalContainers(Resource clusterResource,
ResourceRequest rackLocalResourceRequest, FiCaSchedulerNode node,
FiCaSchedulerApp application, Priority priority,
- RMContainer reservedContainer, MutableObject allocatedContainer) {
+ RMContainer reservedContainer, MutableObject allocatedContainer,
+ SchedulingMode schedulingMode) {
if (canAssign(application, priority, node, NodeType.RACK_LOCAL,
reservedContainer)) {
return assignContainer(clusterResource, node, application, priority,
rackLocalResourceRequest, NodeType.RACK_LOCAL, reservedContainer,
- allocatedContainer);
+ allocatedContainer, schedulingMode);
}
return new CSAssignment(Resources.none(), NodeType.RACK_LOCAL);
@@ -1373,16 +1427,21 @@ public class LeafQueue extends AbstractCSQueue {
private CSAssignment assignOffSwitchContainers(Resource clusterResource,
ResourceRequest offSwitchResourceRequest, FiCaSchedulerNode node,
FiCaSchedulerApp application, Priority priority,
- RMContainer reservedContainer, MutableObject allocatedContainer) {
+ RMContainer reservedContainer, MutableObject allocatedContainer,
+ SchedulingMode schedulingMode) {
if (canAssign(application, priority, node, NodeType.OFF_SWITCH,
reservedContainer)) {
return assignContainer(clusterResource, node, application, priority,
offSwitchResourceRequest, NodeType.OFF_SWITCH, reservedContainer,
- allocatedContainer);
+ allocatedContainer, schedulingMode);
}
return new CSAssignment(Resources.none(), NodeType.OFF_SWITCH);
}
+
+ private int getActualNodeLocalityDelay() {
+ return Math.min(scheduler.getNumClusterNodes(), getNodeLocalityDelay());
+ }
boolean canAssign(FiCaSchedulerApp application, Priority priority,
FiCaSchedulerNode node, NodeType type, RMContainer reservedContainer) {
@@ -1417,10 +1476,7 @@ public class LeafQueue extends AbstractCSQueue {
if (type == NodeType.RACK_LOCAL) {
// 'Delay' rack-local just a little bit...
long missedOpportunities = application.getSchedulingOpportunities(priority);
- return (
- Math.min(scheduler.getNumClusterNodes(), getNodeLocalityDelay()) <
- missedOpportunities
- );
+ return getActualNodeLocalityDelay() < missedOpportunities;
}
// Check if we need containers on this host
@@ -1460,7 +1516,7 @@ public class LeafQueue extends AbstractCSQueue {
private CSAssignment assignContainer(Resource clusterResource, FiCaSchedulerNode node,
FiCaSchedulerApp application, Priority priority,
ResourceRequest request, NodeType type, RMContainer rmContainer,
- MutableObject createdContainer) {
+ MutableObject createdContainer, SchedulingMode schedulingMode) {
if (LOG.isDebugEnabled()) {
LOG.debug("assignContainers: node=" + node.getNodeName()
+ " application=" + application.getApplicationId()
@@ -1469,9 +1525,8 @@ public class LeafQueue extends AbstractCSQueue {
}
// check if the resource request can access the label
- if (!SchedulerUtils.checkNodeLabelExpression(
- node.getLabels(),
- request.getNodeLabelExpression())) {
+ if (!SchedulerUtils.checkResourceRequestMatchingNodePartition(request,
+ node.getPartition(), schedulingMode)) {
// this is a reserved container, but we cannot allocate it now according
// to label not match. This can be caused by node label changed
// We should un-reserve this container.
@@ -1576,8 +1631,8 @@ public class LeafQueue extends AbstractCSQueue {
// If we're trying to reserve a container here, not container will be
// unreserved for reserving the new one. Check limits again before
// reserve the new container
- if (!checkLimitsToReserve(clusterResource,
- application, capability)) {
+ if (!checkLimitsToReserve(clusterResource,
+ application, capability, node.getPartition(), schedulingMode)) {
return new CSAssignment(Resources.none(), type);
}
}
@@ -1666,7 +1721,7 @@ public class LeafQueue extends AbstractCSQueue {
// Book-keeping
if (removed) {
releaseResource(clusterResource, application,
- container.getResource(), node.getLabels());
+ container.getResource(), node.getPartition());
LOG.info("completedContainer" +
" container=" + container +
" queue=" + this +
@@ -1684,13 +1739,13 @@ public class LeafQueue extends AbstractCSQueue {
synchronized void allocateResource(Resource clusterResource,
SchedulerApplicationAttempt application, Resource resource,
- Set<String> nodeLabels) {
- super.allocateResource(clusterResource, resource, nodeLabels);
+ String nodePartition) {
+ super.allocateResource(clusterResource, resource, nodePartition);
// Update user metrics
String userName = application.getUser();
User user = getUser(userName);
- user.assignContainer(resource, nodeLabels);
+ user.assignContainer(resource, nodePartition);
// Note this is a bit unconventional since it gets the object and modifies
// it here, rather then using set routine
Resources.subtractFrom(application.getHeadroom(), resource); // headroom
@@ -1707,13 +1762,13 @@ public class LeafQueue extends AbstractCSQueue {
}
synchronized void releaseResource(Resource clusterResource,
- FiCaSchedulerApp application, Resource resource, Set<String> nodeLabels) {
- super.releaseResource(clusterResource, resource, nodeLabels);
+ FiCaSchedulerApp application, Resource resource, String nodePartition) {
+ super.releaseResource(clusterResource, resource, nodePartition);
// Update user metrics
String userName = application.getUser();
User user = getUser(userName);
- user.releaseContainer(resource, nodeLabels);
+ user.releaseContainer(resource, nodePartition);
metrics.setAvailableResourcesToUser(userName, application.getHeadroom());
LOG.info(getQueueName() +
@@ -1723,7 +1778,8 @@ public class LeafQueue extends AbstractCSQueue {
private void updateAbsoluteCapacityResource(Resource clusterResource) {
absoluteCapacityResource =
- Resources.multiplyAndNormalizeUp(resourceCalculator, clusterResource,
+ Resources.multiplyAndNormalizeUp(resourceCalculator, labelManager
+ .getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource),
queueCapacities.getAbsoluteCapacity(), minimumAllocation);
}
@@ -1769,8 +1825,9 @@ public class LeafQueue extends AbstractCSQueue {
// Update application properties
for (FiCaSchedulerApp application : activeApplications) {
synchronized (application) {
- computeUserLimitAndSetHeadroom(application, clusterResource,
- Resources.none(), null);
+ computeUserLimitAndSetHeadroom(application, clusterResource,
+ Resources.none(), RMNodeLabelsManager.NO_LABEL,
+ SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
}
}
}
@@ -1828,25 +1885,12 @@ public class LeafQueue extends AbstractCSQueue {
}
}
- public void assignContainer(Resource resource,
- Set<String> nodeLabels) {
- if (nodeLabels == null || nodeLabels.isEmpty()) {
- userResourceUsage.incUsed(resource);
- } else {
- for (String label : nodeLabels) {
- userResourceUsage.incUsed(label, resource);
- }
- }
+ public void assignContainer(Resource resource, String nodePartition) {
+ userResourceUsage.incUsed(nodePartition, resource);
}
- public void releaseContainer(Resource resource, Set<String> nodeLabels) {
- if (nodeLabels == null || nodeLabels.isEmpty()) {
- userResourceUsage.decUsed(resource);
- } else {
- for (String label : nodeLabels) {
- userResourceUsage.decUsed(label, resource);
- }
- }
+ public void releaseContainer(Resource resource, String nodePartition) {
+ userResourceUsage.decUsed(nodePartition, resource);
}
public Resource getUserResourceLimit() {
@@ -1869,7 +1913,7 @@ public class LeafQueue extends AbstractCSQueue {
FiCaSchedulerNode node =
scheduler.getNode(rmContainer.getContainer().getNodeId());
allocateResource(clusterResource, attempt, rmContainer.getContainer()
- .getResource(), node.getLabels());
+ .getResource(), node.getPartition());
}
getParent().recoverContainer(clusterResource, attempt, rmContainer);
}
@@ -1909,7 +1953,7 @@ public class LeafQueue extends AbstractCSQueue {
FiCaSchedulerNode node =
scheduler.getNode(rmContainer.getContainer().getNodeId());
allocateResource(clusterResource, application, rmContainer.getContainer()
- .getResource(), node.getLabels());
+ .getResource(), node.getPartition());
LOG.info("movedContainer" + " container=" + rmContainer.getContainer()
+ " resource=" + rmContainer.getContainer().getResource()
+ " queueMoveIn=" + this + " usedCapacity=" + getUsedCapacity()
@@ -1927,7 +1971,7 @@ public class LeafQueue extends AbstractCSQueue {
FiCaSchedulerNode node =
scheduler.getNode(rmContainer.getContainer().getNodeId());
releaseResource(clusterResource, application, rmContainer.getContainer()
- .getResource(), node.getLabels());
+ .getResource(), node.getPartition());
LOG.info("movedContainer" + " container=" + rmContainer.getContainer()
+ " resource=" + rmContainer.getContainer().getResource()
+ " queueMoveOut=" + this + " usedCapacity=" + getUsedCapacity()