YARN-3124. Fixed CS LeafQueue/ParentQueue to use QueueCapacities to track capacities-by-label. Contributed by Wangda Tan

(cherry picked from commit 18a594257e)

(cherry picked from commit 1be2d64ddd)
This commit is contained in:
Jian He 2015-02-12 14:58:09 -08:00 committed by Vinod Kumar Vavilapalli
parent 637e7f9e39
commit ee2b6bc248
17 changed files with 415 additions and 543 deletions

View File

@ -189,6 +189,9 @@ Release 2.6.1 - UNRELEASED
YARN-2694. Ensure only single node label specified in ResourceRequest. YARN-2694. Ensure only single node label specified in ResourceRequest.
(Wangda Tan via jianhe) (Wangda Tan via jianhe)
YARN-3124. Fixed CS LeafQueue/ParentQueue to use QueueCapacities to track
capacities-by-label. (Wangda Tan via jianhe)
Release 2.6.0 - 2014-11-18 Release 2.6.0 - 2014-11-18
INCOMPATIBLE CHANGES INCOMPATIBLE CHANGES

View File

@ -204,6 +204,8 @@
<Field name="absoluteNodeLabelCapacities" /> <Field name="absoluteNodeLabelCapacities" />
<Field name="reservationsContinueLooking" /> <Field name="reservationsContinueLooking" />
<Field name="absoluteCapacityByNodeLabels" /> <Field name="absoluteCapacityByNodeLabels" />
<Field name="authorizer" />
<Field name="parent" />
</Or> </Or>
<Bug pattern="IS2_INCONSISTENT_SYNC" /> <Bug pattern="IS2_INCONSISTENT_SYNC" />
</Match> </Match>

View File

@ -39,7 +39,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator; import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets; import com.google.common.collect.Sets;
public abstract class AbstractCSQueue implements CSQueue { public abstract class AbstractCSQueue implements CSQueue {
@ -47,17 +46,10 @@ public abstract class AbstractCSQueue implements CSQueue {
CSQueue parent; CSQueue parent;
final String queueName; final String queueName;
float capacity;
float maximumCapacity;
float absoluteCapacity;
float absoluteMaxCapacity;
float absoluteUsedCapacity = 0.0f;
float usedCapacity = 0.0f;
volatile int numContainers; volatile int numContainers;
final Resource minimumAllocation; Resource minimumAllocation;
final Resource maximumAllocation; Resource maximumAllocation;
QueueState state; QueueState state;
final QueueMetrics metrics; final QueueMetrics metrics;
@ -65,10 +57,6 @@ public abstract class AbstractCSQueue implements CSQueue {
Set<String> accessibleLabels; Set<String> accessibleLabels;
RMNodeLabelsManager labelManager; RMNodeLabelsManager labelManager;
String defaultLabelExpression; String defaultLabelExpression;
Map<String, Float> absoluteCapacityByNodeLabels;
Map<String, Float> capacitiyByNodeLabels;
Map<String, Float> absoluteMaxCapacityByNodeLabels;
Map<String, Float> maxCapacityByNodeLabels;
Map<QueueACL, AccessControlList> acls = Map<QueueACL, AccessControlList> acls =
new HashMap<QueueACL, AccessControlList>(); new HashMap<QueueACL, AccessControlList>();
@ -77,13 +65,16 @@ public abstract class AbstractCSQueue implements CSQueue {
// Track resource usage-by-label like used-resource/pending-resource, etc. // Track resource usage-by-label like used-resource/pending-resource, etc.
ResourceUsage queueUsage; ResourceUsage queueUsage;
// Track capacities like used-capcity/abs-used-capacity/capacity/abs-capacity,
// etc.
QueueCapacities queueCapacities;
private final RecordFactory recordFactory = private final RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null); RecordFactoryProvider.getRecordFactory(null);
protected CapacitySchedulerContext csContext;
public AbstractCSQueue(CapacitySchedulerContext cs, public AbstractCSQueue(CapacitySchedulerContext cs,
String queueName, CSQueue parent, CSQueue old) throws IOException { String queueName, CSQueue parent, CSQueue old) throws IOException {
this.minimumAllocation = cs.getMinimumResourceCapability();
this.maximumAllocation = cs.getMaximumResourceCapability();
this.labelManager = cs.getRMContext().getNodeLabelManager(); this.labelManager = cs.getRMContext().getNodeLabelManager();
this.parent = parent; this.parent = parent;
this.queueName = queueName; this.queueName = queueName;
@ -94,65 +85,53 @@ public abstract class AbstractCSQueue implements CSQueue {
QueueMetrics.forQueue(getQueuePath(), parent, QueueMetrics.forQueue(getQueuePath(), parent,
cs.getConfiguration().getEnableUserMetrics(), cs.getConfiguration().getEnableUserMetrics(),
cs.getConf()); cs.getConf());
this.csContext = cs;
// get labels // initialize ResourceUsage
this.accessibleLabels = cs.getConfiguration().getAccessibleNodeLabels(getQueuePath());
this.defaultLabelExpression = cs.getConfiguration()
.getDefaultNodeLabelExpression(getQueuePath());
// inherit from parent if labels not set
if (this.accessibleLabels == null && parent != null) {
this.accessibleLabels = parent.getAccessibleNodeLabels();
}
SchedulerUtils.checkIfLabelInClusterNodeLabels(labelManager,
this.accessibleLabels);
// inherit from parent if labels not set
if (this.defaultLabelExpression == null && parent != null
&& this.accessibleLabels.containsAll(parent.getAccessibleNodeLabels())) {
this.defaultLabelExpression = parent.getDefaultNodeLabelExpression();
}
// set capacity by labels
capacitiyByNodeLabels =
cs.getConfiguration().getNodeLabelCapacities(getQueuePath(), accessibleLabels,
labelManager);
// set maximum capacity by labels
maxCapacityByNodeLabels =
cs.getConfiguration().getMaximumNodeLabelCapacities(getQueuePath(),
accessibleLabels, labelManager);
queueUsage = new ResourceUsage(); queueUsage = new ResourceUsage();
// initialize QueueCapacities
queueCapacities = new QueueCapacities(parent == null);
}
protected void setupConfigurableCapacities() {
CSQueueUtils.loadUpdateAndCheckCapacities(
getQueuePath(),
accessibleLabels,
csContext.getConfiguration(),
queueCapacities,
parent == null ? null : parent.getQueueCapacities(),
csContext.getRMContext().getNodeLabelManager());
} }
@Override @Override
public synchronized float getCapacity() { public synchronized float getCapacity() {
return capacity; return queueCapacities.getCapacity();
} }
@Override @Override
public synchronized float getAbsoluteCapacity() { public synchronized float getAbsoluteCapacity() {
return absoluteCapacity; return queueCapacities.getAbsoluteCapacity();
} }
@Override @Override
public float getAbsoluteMaximumCapacity() { public float getAbsoluteMaximumCapacity() {
return absoluteMaxCapacity; return queueCapacities.getAbsoluteMaximumCapacity();
} }
@Override @Override
public synchronized float getAbsoluteUsedCapacity() { public synchronized float getAbsoluteUsedCapacity() {
return absoluteUsedCapacity; return queueCapacities.getAbsoluteUsedCapacity();
} }
@Override @Override
public float getMaximumCapacity() { public float getMaximumCapacity() {
return maximumCapacity; return queueCapacities.getMaximumCapacity();
} }
@Override @Override
public synchronized float getUsedCapacity() { public synchronized float getUsedCapacity() {
return usedCapacity; return queueCapacities.getUsedCapacity();
} }
@Override @Override
@ -210,12 +189,12 @@ public abstract class AbstractCSQueue implements CSQueue {
@Override @Override
public synchronized void setUsedCapacity(float usedCapacity) { public synchronized void setUsedCapacity(float usedCapacity) {
this.usedCapacity = usedCapacity; queueCapacities.setUsedCapacity(usedCapacity);
} }
@Override @Override
public synchronized void setAbsoluteUsedCapacity(float absUsedCapacity) { public synchronized void setAbsoluteUsedCapacity(float absUsedCapacity) {
this.absoluteUsedCapacity = absUsedCapacity; queueCapacities.setAbsoluteUsedCapacity(absUsedCapacity);
} }
/** /**
@ -224,21 +203,16 @@ public abstract class AbstractCSQueue implements CSQueue {
*/ */
synchronized void setMaxCapacity(float maximumCapacity) { synchronized void setMaxCapacity(float maximumCapacity) {
// Sanity check // Sanity check
CSQueueUtils.checkMaxCapacity(getQueueName(), capacity, maximumCapacity); CSQueueUtils.checkMaxCapacity(getQueueName(),
queueCapacities.getCapacity(), maximumCapacity);
float absMaxCapacity = float absMaxCapacity =
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent); CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
CSQueueUtils.checkAbsoluteCapacity(getQueueName(), absoluteCapacity, CSQueueUtils.checkAbsoluteCapacity(getQueueName(),
queueCapacities.getAbsoluteCapacity(),
absMaxCapacity); absMaxCapacity);
this.maximumCapacity = maximumCapacity; queueCapacities.setMaximumCapacity(maximumCapacity);
this.absoluteMaxCapacity = absMaxCapacity; queueCapacities.setAbsoluteMaximumCapacity(absMaxCapacity);
}
@Override
public float getAbsActualCapacity() {
// for now, simply return actual capacity = guaranteed capacity for parent
// queue
return absoluteCapacity;
} }
@Override @Override
@ -246,39 +220,35 @@ public abstract class AbstractCSQueue implements CSQueue {
return defaultLabelExpression; return defaultLabelExpression;
} }
synchronized void setupQueueConfigs(Resource clusterResource, float capacity, synchronized void setupQueueConfigs(Resource clusterResource)
float absoluteCapacity, float maximumCapacity, float absoluteMaxCapacity,
QueueState state, Map<QueueACL, AccessControlList> acls,
Set<String> labels, String defaultLabelExpression,
Map<String, Float> nodeLabelCapacities,
Map<String, Float> maximumNodeLabelCapacities,
boolean reservationContinueLooking)
throws IOException { throws IOException {
// Sanity check // get labels
CSQueueUtils.checkMaxCapacity(getQueueName(), capacity, maximumCapacity); this.accessibleLabels =
CSQueueUtils.checkAbsoluteCapacity(getQueueName(), absoluteCapacity, csContext.getConfiguration().getAccessibleNodeLabels(getQueuePath());
absoluteMaxCapacity); this.defaultLabelExpression = csContext.getConfiguration()
.getDefaultNodeLabelExpression(getQueuePath());
this.capacity = capacity; // inherit from parent if labels not set
this.absoluteCapacity = absoluteCapacity; if (this.accessibleLabels == null && parent != null) {
this.accessibleLabels = parent.getAccessibleNodeLabels();
}
SchedulerUtils.checkIfLabelInClusterNodeLabels(labelManager,
this.accessibleLabels);
this.maximumCapacity = maximumCapacity; // inherit from parent if labels not set
this.absoluteMaxCapacity = absoluteMaxCapacity; if (this.defaultLabelExpression == null && parent != null
&& this.accessibleLabels.containsAll(parent.getAccessibleNodeLabels())) {
this.defaultLabelExpression = parent.getDefaultNodeLabelExpression();
}
this.state = state; // After we setup labels, we can setup capacities
setupConfigurableCapacities();
this.acls = acls; this.minimumAllocation = csContext.getMinimumResourceCapability();
this.maximumAllocation = csContext.getMaximumResourceCapability();
// set labels this.state = csContext.getConfiguration().getState(getQueuePath());
this.accessibleLabels = labels; this.acls = csContext.getConfiguration().getAcls(getQueuePath());
// set label expression
this.defaultLabelExpression = defaultLabelExpression;
// copy node label capacity
this.capacitiyByNodeLabels = new HashMap<String, Float>(nodeLabelCapacities);
this.maxCapacityByNodeLabels =
new HashMap<String, Float>(maximumNodeLabelCapacities);
// Update metrics // Update metrics
CSQueueUtils.updateQueueStatistics( CSQueueUtils.updateQueueStatistics(
@ -306,29 +276,17 @@ public abstract class AbstractCSQueue implements CSQueue {
} }
} }
// calculate absolute capacity by each node label this.reservationsContinueLooking = csContext.getConfiguration()
this.absoluteCapacityByNodeLabels = .getReservationContinueLook();
CSQueueUtils.computeAbsoluteCapacityByNodeLabels(
this.capacitiyByNodeLabels, parent);
// calculate maximum capacity by each node label
this.absoluteMaxCapacityByNodeLabels =
CSQueueUtils.computeAbsoluteMaxCapacityByNodeLabels(
maximumNodeLabelCapacities, parent);
// check absoluteMaximumNodeLabelCapacities is valid
CSQueueUtils.checkAbsoluteCapacitiesByLabel(getQueueName(),
absoluteCapacityByNodeLabels, absoluteCapacityByNodeLabels);
this.reservationsContinueLooking = reservationContinueLooking;
} }
protected QueueInfo getQueueInfo() { protected QueueInfo getQueueInfo() {
QueueInfo queueInfo = recordFactory.newRecordInstance(QueueInfo.class); QueueInfo queueInfo = recordFactory.newRecordInstance(QueueInfo.class);
queueInfo.setQueueName(queueName); queueInfo.setQueueName(queueName);
queueInfo.setAccessibleNodeLabels(accessibleLabels); queueInfo.setAccessibleNodeLabels(accessibleLabels);
queueInfo.setCapacity(capacity); queueInfo.setCapacity(queueCapacities.getCapacity());
queueInfo.setMaximumCapacity(maximumCapacity); queueInfo.setMaximumCapacity(queueCapacities.getMaximumCapacity());
queueInfo.setQueueState(state); queueInfo.setQueueState(state);
queueInfo.setDefaultNodeLabelExpression(defaultLabelExpression); queueInfo.setDefaultNodeLabelExpression(defaultLabelExpression);
queueInfo.setCurrentCapacity(getUsedCapacity()); queueInfo.setCurrentCapacity(getUsedCapacity());
@ -378,51 +336,6 @@ public abstract class AbstractCSQueue implements CSQueue {
--numContainers; --numContainers;
} }
@Private
public float getCapacityByNodeLabel(String label) {
if (StringUtils.equals(label, RMNodeLabelsManager.NO_LABEL)) {
if (null == parent) {
return 1f;
}
return getCapacity();
}
if (!capacitiyByNodeLabels.containsKey(label)) {
return 0f;
} else {
return capacitiyByNodeLabels.get(label);
}
}
@Private
public float getAbsoluteCapacityByNodeLabel(String label) {
if (StringUtils.equals(label, RMNodeLabelsManager.NO_LABEL)) {
if (null == parent) {
return 1f;
}
return getAbsoluteCapacity();
}
if (!absoluteCapacityByNodeLabels.containsKey(label)) {
return 0f;
} else {
return absoluteCapacityByNodeLabels.get(label);
}
}
@Private
public float getAbsoluteMaximumCapacityByNodeLabel(String label) {
if (StringUtils.equals(label, RMNodeLabelsManager.NO_LABEL)) {
return getAbsoluteMaximumCapacity();
}
if (!absoluteMaxCapacityByNodeLabels.containsKey(label)) {
return 0f;
} else {
return absoluteMaxCapacityByNodeLabels.get(label);
}
}
@Private @Private
public boolean getReservationContinueLooking() { public boolean getReservationContinueLooking() {
return reservationsContinueLooking; return reservationsContinueLooking;
@ -434,12 +347,12 @@ public abstract class AbstractCSQueue implements CSQueue {
} }
@Private @Private
public Resource getUsedResourceByLabel(String nodeLabel) { public QueueCapacities getQueueCapacities() {
return queueUsage.getUsed(nodeLabel); return queueCapacities;
} }
@VisibleForTesting @Private
public ResourceUsage getResourceUsage() { public ResourceUsage getQueueResourceUsage() {
return queueUsage; return queueUsage;
} }
} }

View File

@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer; 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.RMContainerEventType;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode; import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
@ -76,15 +77,6 @@ extends org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue {
*/ */
public float getCapacity(); public float getCapacity();
/**
* Get actual <em>capacity</em> of the queue, this may be different from
* configured capacity when mis-config take place, like add labels to the
* cluster
*
* @return actual queue capacity
*/
public float getAbsActualCapacity();
/** /**
* Get capacity of the parent of the queue as a function of the * Get capacity of the parent of the queue as a function of the
* cumulative capacity in the cluster. * cumulative capacity in the cluster.
@ -143,14 +135,6 @@ extends org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue {
*/ */
public Resource getUsedResources(); public Resource getUsedResources();
/**
* Get the currently utilized resources which allocated at nodes with label
* specified
*
* @return used resources by the queue and it's children
*/
public Resource getUsedResourceByLabel(String nodeLabel);
/** /**
* Get the current run-state of the queue * Get the current run-state of the queue
* @return current run-state * @return current run-state
@ -281,23 +265,14 @@ extends org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue {
FiCaSchedulerApp application, RMContainer container); FiCaSchedulerApp application, RMContainer container);
/** /**
* Get absolute capacity by label of this queue can use * Get QueueCapacities of this queue
* @param nodeLabel * @return queueCapacities
* @return absolute capacity by label of this queue can use
*/ */
public float getAbsoluteCapacityByNodeLabel(String nodeLabel); public QueueCapacities getQueueCapacities();
/** /**
* Get absolute max capacity by label of this queue can use * Get ResourceUsage of this queue
* @param nodeLabel * @return resourceUsage
* @return absolute capacity by label of this queue can use
*/ */
public float getAbsoluteMaximumCapacityByNodeLabel(String nodeLabel); public ResourceUsage getQueueResourceUsage();
/**
* Get capacity by node label
* @param nodeLabel
* @return capacity by node label
*/
public float getCapacityByNodeLabel(String nodeLabel);
} }

View File

@ -17,13 +17,14 @@
*/ */
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity; package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import java.util.HashMap; import java.util.HashSet;
import java.util.Map; import java.util.Set;
import java.util.Map.Entry;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.yarn.api.records.Resource; 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.server.utils.Lock; import org.apache.hadoop.yarn.server.utils.Lock;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator; import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources; import org.apache.hadoop.yarn.util.resource.Resources;
@ -34,6 +35,9 @@ class CSQueueUtils {
final static float EPSILON = 0.0001f; final static float EPSILON = 0.0001f;
/*
* Used only by tests
*/
public static void checkMaxCapacity(String queueName, public static void checkMaxCapacity(String queueName,
float capacity, float maximumCapacity) { float capacity, float maximumCapacity) {
if (maximumCapacity < 0.0f || maximumCapacity > 1.0f) { if (maximumCapacity < 0.0f || maximumCapacity > 1.0f) {
@ -43,6 +47,9 @@ class CSQueueUtils {
} }
} }
/*
* Used only by tests
*/
public static void checkAbsoluteCapacity(String queueName, public static void checkAbsoluteCapacity(String queueName,
float absCapacity, float absMaxCapacity) { float absCapacity, float absMaxCapacity) {
if (absMaxCapacity < (absCapacity - EPSILON)) { if (absMaxCapacity < (absCapacity - EPSILON)) {
@ -53,19 +60,33 @@ class CSQueueUtils {
} }
} }
public static void checkAbsoluteCapacitiesByLabel(String queueName, /**
Map<String, Float> absCapacities, * Check sanity of capacities:
Map<String, Float> absMaximumCapacities) { * - capacity <= maxCapacity
for (Entry<String, Float> entry : absCapacities.entrySet()) { * - absCapacity <= absMaximumCapacity
String label = entry.getKey(); */
float absCapacity = entry.getValue(); private static void capacitiesSanityCheck(String queueName,
float absMaxCapacity = absMaximumCapacities.get(label); QueueCapacities queueCapacities) {
if (absMaxCapacity < (absCapacity - EPSILON)) { for (String label : queueCapacities.getExistingNodeLabels()) {
throw new IllegalArgumentException("Illegal call to setMaxCapacity. " float capacity = queueCapacities.getCapacity(label);
+ "Queue '" + queueName + "' has " + "an absolute capacity (" float maximumCapacity = queueCapacities.getMaximumCapacity(label);
+ absCapacity + ") greater than " if (capacity > maximumCapacity) {
+ "its absolute maximumCapacity (" + absMaxCapacity + ") of label=" throw new IllegalArgumentException("Illegal queue capacity setting, "
+ label); + "(capacity=" + capacity + ") > (maximum-capacity="
+ maximumCapacity + "). When label=[" + label + "]");
}
// Actually, this may not needed since we have verified capacity <=
// maximumCapacity. And the way we compute absolute capacity (abs(x) =
// cap(x) * cap(x.parent) * ...) is a monotone increasing function. But
// just keep it here to make sure our compute abs capacity method works
// correctly.
float absCapacity = queueCapacities.getAbsoluteCapacity(label);
float absMaxCapacity = queueCapacities.getAbsoluteMaximumCapacity(label);
if (absCapacity > absMaxCapacity) {
throw new IllegalArgumentException("Illegal queue capacity setting, "
+ "(abs-capacity=" + absCapacity + ") > (abs-maximum-capacity="
+ absMaxCapacity + "). When label=[" + label + "]");
} }
} }
} }
@ -77,37 +98,94 @@ class CSQueueUtils {
return (parentAbsMaxCapacity * maximumCapacity); return (parentAbsMaxCapacity * maximumCapacity);
} }
public static Map<String, Float> computeAbsoluteCapacityByNodeLabels( /**
Map<String, Float> nodeLabelToCapacities, CSQueue parent) { * This method intends to be used by ReservationQueue, ReservationQueue will
if (parent == null) { * not appear in configuration file, so we shouldn't do load capacities
return nodeLabelToCapacities; * settings in configuration for reservation queue.
*/
public static void updateAndCheckCapacitiesByLabel(String queuePath,
QueueCapacities queueCapacities, QueueCapacities parentQueueCapacities) {
updateAbsoluteCapacitiesByNodeLabels(queueCapacities, parentQueueCapacities);
capacitiesSanityCheck(queuePath, queueCapacities);
} }
Map<String, Float> absoluteCapacityByNodeLabels = /**
new HashMap<String, Float>(); * Do following steps for capacities
for (Entry<String, Float> entry : nodeLabelToCapacities.entrySet()) { * - Load capacities from configuration
String label = entry.getKey(); * - Update absolute capacities for new capacities
float capacity = entry.getValue(); * - Check if capacities/absolute-capacities legal
absoluteCapacityByNodeLabels.put(label, */
capacity * parent.getAbsoluteCapacityByNodeLabel(label)); public static void loadUpdateAndCheckCapacities(String queuePath,
} Set<String> accessibleLabels, CapacitySchedulerConfiguration csConf,
return absoluteCapacityByNodeLabels; QueueCapacities queueCapacities, QueueCapacities parentQueueCapacities,
RMNodeLabelsManager nlm) {
loadCapacitiesByLabelsFromConf(queuePath, accessibleLabels, nlm,
queueCapacities, csConf);
updateAbsoluteCapacitiesByNodeLabels(queueCapacities, parentQueueCapacities);
capacitiesSanityCheck(queuePath, queueCapacities);
} }
public static Map<String, Float> computeAbsoluteMaxCapacityByNodeLabels( // Considered NO_LABEL, ANY and null cases
Map<String, Float> maximumNodeLabelToCapacities, CSQueue parent) { private static Set<String> normalizeAccessibleNodeLabels(Set<String> labels,
if (parent == null) { RMNodeLabelsManager mgr) {
return maximumNodeLabelToCapacities; Set<String> accessibleLabels = new HashSet<String>();
if (labels != null) {
accessibleLabels.addAll(labels);
}
if (accessibleLabels.contains(CommonNodeLabelsManager.ANY)) {
accessibleLabels.addAll(mgr.getClusterNodeLabels());
}
accessibleLabels.add(CommonNodeLabelsManager.NO_LABEL);
return accessibleLabels;
}
private static void loadCapacitiesByLabelsFromConf(String queuePath,
Set<String> labels, RMNodeLabelsManager mgr,
QueueCapacities queueCapacities, CapacitySchedulerConfiguration csConf) {
queueCapacities.clearConfigurableFields();
labels = normalizeAccessibleNodeLabels(labels, mgr);
for (String label : labels) {
if (label.equals(CommonNodeLabelsManager.NO_LABEL)) {
queueCapacities.setCapacity(CommonNodeLabelsManager.NO_LABEL,
csConf.getNonLabeledQueueCapacity(queuePath) / 100);
queueCapacities.setMaximumCapacity(CommonNodeLabelsManager.NO_LABEL,
csConf.getNonLabeledQueueMaximumCapacity(queuePath) / 100);
} else {
queueCapacities.setCapacity(label,
csConf.getLabeledQueueCapacity(queuePath, label) / 100);
queueCapacities.setMaximumCapacity(label,
csConf.getLabeledQueueMaximumCapacity(queuePath, label) / 100);
}
}
}
// Set absolute capacities for {capacity, maximum-capacity}
private static void updateAbsoluteCapacitiesByNodeLabels(
QueueCapacities queueCapacities, QueueCapacities parentQueueCapacities) {
for (String label : queueCapacities.getExistingNodeLabels()) {
float capacity = queueCapacities.getCapacity(label);
if (capacity > 0f) {
queueCapacities.setAbsoluteCapacity(
label,
capacity
* (parentQueueCapacities == null ? 1 : parentQueueCapacities
.getAbsoluteCapacity(label)));
}
float maxCapacity = queueCapacities.getMaximumCapacity(label);
if (maxCapacity > 0f) {
queueCapacities.setAbsoluteMaximumCapacity(
label,
maxCapacity
* (parentQueueCapacities == null ? 1 : parentQueueCapacities
.getAbsoluteMaximumCapacity(label)));
} }
Map<String, Float> absoluteMaxCapacityByNodeLabels =
new HashMap<String, Float>();
for (Entry<String, Float> entry : maximumNodeLabelToCapacities.entrySet()) {
String label = entry.getKey();
float maxCapacity = entry.getValue();
absoluteMaxCapacityByNodeLabels.put(label,
maxCapacity * parent.getAbsoluteMaximumCapacityByNodeLabel(label));
} }
return absoluteMaxCapacityByNodeLabels;
} }
@Lock(CSQueue.class) @Lock(CSQueue.class)

View File

@ -287,6 +287,9 @@ public class CapacitySchedulerConfiguration extends Configuration {
} }
private String getNodeLabelPrefix(String queue, String label) { private String getNodeLabelPrefix(String queue, String label) {
if (label.equals(CommonNodeLabelsManager.NO_LABEL)) {
return getQueuePrefix(queue);
}
return getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS + DOT + label + DOT; return getQueuePrefix(queue) + ACCESSIBLE_NODE_LABELS + DOT + label + DOT;
} }
@ -325,7 +328,7 @@ public class CapacitySchedulerConfiguration extends Configuration {
getMaximumApplicationMasterResourcePercent()); getMaximumApplicationMasterResourcePercent());
} }
public float getCapacity(String queue) { public float getNonLabeledQueueCapacity(String queue) {
float capacity = queue.equals("root") ? 100.0f : getFloat( float capacity = queue.equals("root") ? 100.0f : getFloat(
getQueuePrefix(queue) + CAPACITY, UNDEFINED); getQueuePrefix(queue) + CAPACITY, UNDEFINED);
if (capacity < MINIMUM_CAPACITY_VALUE || capacity > MAXIMUM_CAPACITY_VALUE) { if (capacity < MINIMUM_CAPACITY_VALUE || capacity > MAXIMUM_CAPACITY_VALUE) {
@ -347,7 +350,7 @@ public class CapacitySchedulerConfiguration extends Configuration {
", capacity=" + capacity); ", capacity=" + capacity);
} }
public float getMaximumCapacity(String queue) { public float getNonLabeledQueueMaximumCapacity(String queue) {
float maxCapacity = getFloat(getQueuePrefix(queue) + MAXIMUM_CAPACITY, float maxCapacity = getFloat(getQueuePrefix(queue) + MAXIMUM_CAPACITY,
MAXIMUM_CAPACITY_VALUE); MAXIMUM_CAPACITY_VALUE);
maxCapacity = (maxCapacity == DEFAULT_MAXIMUM_CAPACITY_VALUE) ? maxCapacity = (maxCapacity == DEFAULT_MAXIMUM_CAPACITY_VALUE) ?
@ -451,18 +454,10 @@ public class CapacitySchedulerConfiguration extends Configuration {
return Collections.unmodifiableSet(set); return Collections.unmodifiableSet(set);
} }
public Map<String, Float> getNodeLabelCapacities(String queue, private float internalGetLabeledQueueCapacity(String queue, String label, String suffix,
Set<String> labels, RMNodeLabelsManager mgr) { float defaultValue) {
Map<String, Float> nodeLabelCapacities = new HashMap<String, Float>(); String capacityPropertyName = getNodeLabelPrefix(queue, label) + suffix;
float capacity = getFloat(capacityPropertyName, defaultValue);
if (labels == null) {
return nodeLabelCapacities;
}
for (String label : labels.contains(CommonNodeLabelsManager.ANY) ? mgr
.getClusterNodeLabels() : labels) {
String capacityPropertyName = getNodeLabelPrefix(queue, label) + CAPACITY;
float capacity = getFloat(capacityPropertyName, 0f);
if (capacity < MINIMUM_CAPACITY_VALUE if (capacity < MINIMUM_CAPACITY_VALUE
|| capacity > MAXIMUM_CAPACITY_VALUE) { || capacity > MAXIMUM_CAPACITY_VALUE) {
throw new IllegalArgumentException("Illegal capacity of " + capacity throw new IllegalArgumentException("Illegal capacity of " + capacity
@ -473,35 +468,15 @@ public class CapacitySchedulerConfiguration extends Configuration {
LOG.debug("CSConf - getCapacityOfLabel: prefix=" LOG.debug("CSConf - getCapacityOfLabel: prefix="
+ getNodeLabelPrefix(queue, label) + ", capacity=" + capacity); + getNodeLabelPrefix(queue, label) + ", capacity=" + capacity);
} }
return capacity;
nodeLabelCapacities.put(label, capacity / 100f);
}
return nodeLabelCapacities;
} }
public Map<String, Float> getMaximumNodeLabelCapacities(String queue, public float getLabeledQueueCapacity(String queue, String label) {
Set<String> labels, RMNodeLabelsManager mgr) { return internalGetLabeledQueueCapacity(queue, label, CAPACITY, 0f);
Map<String, Float> maximumNodeLabelCapacities = new HashMap<String, Float>();
if (labels == null) {
return maximumNodeLabelCapacities;
} }
for (String label : labels.contains(CommonNodeLabelsManager.ANY) ? mgr public float getLabeledQueueMaximumCapacity(String queue, String label) {
.getClusterNodeLabels() : labels) { return internalGetLabeledQueueCapacity(queue, label, MAXIMUM_CAPACITY, 100f);
float maxCapacity =
getFloat(getNodeLabelPrefix(queue, label) + MAXIMUM_CAPACITY,
100f);
if (maxCapacity < MINIMUM_CAPACITY_VALUE
|| maxCapacity > MAXIMUM_CAPACITY_VALUE) {
throw new IllegalArgumentException("Illegal " + "capacity of "
+ maxCapacity + " for label=" + label + " in queue=" + queue);
}
LOG.debug("CSConf - getCapacityOfLabel: prefix="
+ getNodeLabelPrefix(queue, label) + ", capacity=" + maxCapacity);
maximumNodeLabelCapacities.put(label, maxCapacity / 100f);
}
return maximumNodeLabelCapacities;
} }
public String getDefaultNodeLabelExpression(String queue) { public String getDefaultNodeLabelExpression(String queue) {

View File

@ -97,7 +97,7 @@ public class LeafQueue extends AbstractCSQueue {
Set<FiCaSchedulerApp> pendingApplications; Set<FiCaSchedulerApp> pendingApplications;
private final float minimumAllocationFactor; private float minimumAllocationFactor;
private Map<String, User> users = new HashMap<String, User>(); private Map<String, User> users = new HashMap<String, User>();
@ -124,52 +124,6 @@ public class LeafQueue extends AbstractCSQueue {
this.scheduler = cs; this.scheduler = cs;
this.activeUsersManager = new ActiveUsersManager(metrics); this.activeUsersManager = new ActiveUsersManager(metrics);
this.minimumAllocationFactor =
Resources.ratio(resourceCalculator,
Resources.subtract(maximumAllocation, minimumAllocation),
maximumAllocation);
float capacity = getCapacityFromConf();
float absoluteCapacity = parent.getAbsoluteCapacity() * capacity;
float maximumCapacity =
(float)cs.getConfiguration().getMaximumCapacity(getQueuePath()) / 100;
float absoluteMaxCapacity =
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
// Initially set to absoluteMax, will be updated to more accurate
// max avail value during assignContainers
absoluteMaxAvailCapacity = absoluteMaxCapacity;
int userLimit = cs.getConfiguration().getUserLimit(getQueuePath());
float userLimitFactor =
cs.getConfiguration().getUserLimitFactor(getQueuePath());
int maxApplications =
cs.getConfiguration().getMaximumApplicationsPerQueue(getQueuePath());
if (maxApplications < 0) {
int maxSystemApps = cs.getConfiguration().getMaximumSystemApplications();
maxApplications = (int)(maxSystemApps * absoluteCapacity);
}
maxApplicationsPerUser =
(int)(maxApplications * (userLimit / 100.0f) * userLimitFactor);
float maxAMResourcePerQueuePercent = cs.getConfiguration()
.getMaximumApplicationMasterResourcePerQueuePercent(getQueuePath());
QueueState state = cs.getConfiguration().getState(getQueuePath());
Map<QueueACL, AccessControlList> acls =
cs.getConfiguration().getAcls(getQueuePath());
setupQueueConfigs(cs.getClusterResource(), capacity, absoluteCapacity,
maximumCapacity, absoluteMaxCapacity, userLimit, userLimitFactor,
maxApplications, maxAMResourcePerQueuePercent, maxApplicationsPerUser,
state, acls, cs.getConfiguration().getNodeLocalityDelay(), accessibleLabels,
defaultLabelExpression, this.capacitiyByNodeLabels,
this.maxCapacityByNodeLabels,
cs.getConfiguration().getReservationContinueLook());
if(LOG.isDebugEnabled()) { if(LOG.isDebugEnabled()) {
LOG.debug("LeafQueue:" + " name=" + queueName LOG.debug("LeafQueue:" + " name=" + queueName
+ ", fullname=" + getQueuePath()); + ", fullname=" + getQueuePath());
@ -180,34 +134,13 @@ public class LeafQueue extends AbstractCSQueue {
this.pendingApplications = this.pendingApplications =
new TreeSet<FiCaSchedulerApp>(applicationComparator); new TreeSet<FiCaSchedulerApp>(applicationComparator);
this.activeApplications = new TreeSet<FiCaSchedulerApp>(applicationComparator); this.activeApplications = new TreeSet<FiCaSchedulerApp>(applicationComparator);
setupQueueConfigs(cs.getClusterResource());
} }
// externalizing in method, to allow overriding protected synchronized void setupQueueConfigs(Resource clusterResource)
protected float getCapacityFromConf() { throws IOException {
return (float)scheduler.getConfiguration().getCapacity(getQueuePath()) / 100; super.setupQueueConfigs(clusterResource);
}
protected synchronized void setupQueueConfigs(
Resource clusterResource,
float capacity, float absoluteCapacity,
float maximumCapacity, float absoluteMaxCapacity,
int userLimit, float userLimitFactor,
int maxApplications, float maxAMResourcePerQueuePercent,
int maxApplicationsPerUser, QueueState state,
Map<QueueACL, AccessControlList> acls, int nodeLocalityDelay,
Set<String> labels, String defaultLabelExpression,
Map<String, Float> capacitieByLabel,
Map<String, Float> maximumCapacitiesByLabel,
boolean revervationContinueLooking) throws IOException {
super.setupQueueConfigs(clusterResource, capacity, absoluteCapacity,
maximumCapacity, absoluteMaxCapacity, state, acls, labels,
defaultLabelExpression, capacitieByLabel, maximumCapacitiesByLabel,
revervationContinueLooking);
// Sanity check
CSQueueUtils.checkMaxCapacity(getQueueName(), capacity, maximumCapacity);
float absCapacity = getParent().getAbsoluteCapacity() * capacity;
CSQueueUtils.checkAbsoluteCapacity(getQueueName(), absCapacity,
absoluteMaxCapacity);
this.lastClusterResource = clusterResource; this.lastClusterResource = clusterResource;
updateAbsoluteCapacityResource(clusterResource); updateAbsoluteCapacityResource(clusterResource);
@ -217,16 +150,28 @@ public class LeafQueue extends AbstractCSQueue {
// and all queues may not be realized yet, we'll use (optimistic) // and all queues may not be realized yet, we'll use (optimistic)
// absoluteMaxCapacity (it will be replaced with the more accurate // absoluteMaxCapacity (it will be replaced with the more accurate
// absoluteMaxAvailCapacity during headroom/userlimit/allocation events) // absoluteMaxAvailCapacity during headroom/userlimit/allocation events)
updateHeadroomInfo(clusterResource, absoluteMaxCapacity); updateHeadroomInfo(clusterResource,
queueCapacities.getAbsoluteMaximumCapacity());
this.absoluteCapacity = absCapacity; CapacitySchedulerConfiguration conf = csContext.getConfiguration();
userLimit = conf.getUserLimit(getQueuePath());
userLimitFactor = conf.getUserLimitFactor(getQueuePath());
this.userLimit = userLimit; // Initially set to absoluteMax, will be updated to more accurate
this.userLimitFactor = userLimitFactor; // max avail value during assignContainers
absoluteMaxAvailCapacity = queueCapacities.getAbsoluteMaximumCapacity();
this.maxApplications = maxApplications; maxApplications = conf.getMaximumApplicationsPerQueue(getQueuePath());
this.maxAMResourcePerQueuePercent = maxAMResourcePerQueuePercent; if (maxApplications < 0) {
this.maxApplicationsPerUser = maxApplicationsPerUser; int maxSystemApps = conf.getMaximumSystemApplications();
maxApplications =
(int) (maxSystemApps * queueCapacities.getAbsoluteCapacity());
}
maxApplicationsPerUser =
(int)(maxApplications * (userLimit / 100.0f) * userLimitFactor);
maxAMResourcePerQueuePercent =
conf.getMaximumApplicationMasterResourcePerQueuePercent(getQueuePath());
if (!SchedulerUtils.checkQueueLabelExpression(this.accessibleLabels, if (!SchedulerUtils.checkQueueLabelExpression(this.accessibleLabels,
this.defaultLabelExpression)) { this.defaultLabelExpression)) {
@ -242,7 +187,12 @@ public class LeafQueue extends AbstractCSQueue {
getAccessibleNodeLabels().iterator(), ','))); getAccessibleNodeLabels().iterator(), ',')));
} }
this.nodeLocalityDelay = nodeLocalityDelay; nodeLocalityDelay = conf.getNodeLocalityDelay();
this.minimumAllocationFactor =
Resources.ratio(resourceCalculator,
Resources.subtract(maximumAllocation, minimumAllocation),
maximumAllocation);
StringBuilder aclsString = new StringBuilder(); StringBuilder aclsString = new StringBuilder();
for (Map.Entry<QueueACL, AccessControlList> e : acls.entrySet()) { for (Map.Entry<QueueACL, AccessControlList> e : acls.entrySet()) {
@ -250,21 +200,21 @@ public class LeafQueue extends AbstractCSQueue {
} }
StringBuilder labelStrBuilder = new StringBuilder(); StringBuilder labelStrBuilder = new StringBuilder();
if (labels != null) { if (accessibleLabels != null) {
for (String s : labels) { for (String s : accessibleLabels) {
labelStrBuilder.append(s); labelStrBuilder.append(s);
labelStrBuilder.append(","); labelStrBuilder.append(",");
} }
} }
LOG.info("Initializing " + queueName + "\n" + LOG.info("Initializing " + queueName + "\n" +
"capacity = " + capacity + "capacity = " + queueCapacities.getCapacity() +
" [= (float) configuredCapacity / 100 ]" + "\n" + " [= (float) configuredCapacity / 100 ]" + "\n" +
"asboluteCapacity = " + absoluteCapacity + "asboluteCapacity = " + queueCapacities.getAbsoluteCapacity() +
" [= parentAbsoluteCapacity * capacity ]" + "\n" + " [= parentAbsoluteCapacity * capacity ]" + "\n" +
"maxCapacity = " + maximumCapacity + "maxCapacity = " + queueCapacities.getMaximumCapacity() +
" [= configuredMaxCapacity ]" + "\n" + " [= configuredMaxCapacity ]" + "\n" +
"absoluteMaxCapacity = " + absoluteMaxCapacity + "absoluteMaxCapacity = " + queueCapacities.getAbsoluteMaximumCapacity() +
" [= 1.0 maximumCapacity undefined, " + " [= 1.0 maximumCapacity undefined, " +
"(parentAbsoluteMaxCapacity * maximumCapacity) / 100 otherwise ]" + "(parentAbsoluteMaxCapacity * maximumCapacity) / 100 otherwise ]" +
"\n" + "\n" +
@ -279,7 +229,7 @@ public class LeafQueue extends AbstractCSQueue {
"maxApplicationsPerUser = " + maxApplicationsPerUser + "maxApplicationsPerUser = " + maxApplicationsPerUser +
" [= (int)(maxApplications * (userLimit / 100.0f) * " + " [= (int)(maxApplications * (userLimit / 100.0f) * " +
"userLimitFactor) ]" + "\n" + "userLimitFactor) ]" + "\n" +
"usedCapacity = " + usedCapacity + "usedCapacity = " + queueCapacities.getUsedCapacity() +
" [= usedResourcesMemory / " + " [= usedResourcesMemory / " +
"(clusterResourceMemory * absoluteCapacity)]" + "\n" + "(clusterResourceMemory * absoluteCapacity)]" + "\n" +
"absoluteUsedCapacity = " + absoluteUsedCapacity + "absoluteUsedCapacity = " + absoluteUsedCapacity +
@ -435,8 +385,8 @@ public class LeafQueue extends AbstractCSQueue {
public String toString() { public String toString() {
return queueName + ": " + return queueName + ": " +
"capacity=" + capacity + ", " + "capacity=" + queueCapacities.getCapacity() + ", " +
"absoluteCapacity=" + absoluteCapacity + ", " + "absoluteCapacity=" + queueCapacities.getAbsoluteCapacity() + ", " +
"usedResources=" + queueUsage.getUsed() + ", " + "usedResources=" + queueUsage.getUsed() + ", " +
"usedCapacity=" + getUsedCapacity() + ", " + "usedCapacity=" + getUsedCapacity() + ", " +
"absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + ", " + "absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + ", " +
@ -483,23 +433,7 @@ public class LeafQueue extends AbstractCSQueue {
" from " + newlyParsedQueue.getQueuePath()); " from " + newlyParsedQueue.getQueuePath());
} }
LeafQueue newlyParsedLeafQueue = (LeafQueue)newlyParsedQueue; setupQueueConfigs(clusterResource);
setupQueueConfigs(
clusterResource,
newlyParsedLeafQueue.capacity, newlyParsedLeafQueue.absoluteCapacity,
newlyParsedLeafQueue.maximumCapacity,
newlyParsedLeafQueue.absoluteMaxCapacity,
newlyParsedLeafQueue.userLimit, newlyParsedLeafQueue.userLimitFactor,
newlyParsedLeafQueue.maxApplications,
newlyParsedLeafQueue.maxAMResourcePerQueuePercent,
newlyParsedLeafQueue.getMaxApplicationsPerUser(),
newlyParsedLeafQueue.state, newlyParsedLeafQueue.acls,
newlyParsedLeafQueue.getNodeLocalityDelay(),
newlyParsedLeafQueue.accessibleLabels,
newlyParsedLeafQueue.defaultLabelExpression,
newlyParsedLeafQueue.capacitiyByNodeLabels,
newlyParsedLeafQueue.maxCapacityByNodeLabels,
newlyParsedLeafQueue.reservationsContinueLooking);
// queue metrics are updated, more resource may be available // queue metrics are updated, more resource may be available
// activate the pending applications if possible // activate the pending applications if possible
@ -1022,7 +956,8 @@ public class LeafQueue extends AbstractCSQueue {
application.getCurrentReservation()), application.getCurrentReservation()),
labelManager.getResourceByLabel(label, clusterResource)); labelManager.getResourceByLabel(label, clusterResource));
if (potentialNewWithoutReservedCapacity <= absoluteMaxCapacity) { if (potentialNewWithoutReservedCapacity <= queueCapacities
.getAbsoluteMaximumCapacity()) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("try to use reserved: " LOG.debug("try to use reserved: "
+ getQueueName() + getQueueName()
@ -1036,8 +971,9 @@ public class LeafQueue extends AbstractCSQueue {
+ Resources.divide(resourceCalculator, clusterResource, + Resources.divide(resourceCalculator, clusterResource,
queueUsage.getUsed(), clusterResource) + " required " + required queueUsage.getUsed(), clusterResource) + " required " + required
+ " potentialNewWithoutReservedCapacity: " + " potentialNewWithoutReservedCapacity: "
+ potentialNewWithoutReservedCapacity + " ( " + " max-capacity: " + potentialNewWithoutReservedCapacity + " ( "
+ absoluteMaxCapacity + ")"); + " max-capacity: "
+ queueCapacities.getAbsoluteMaximumCapacity() + ")");
} }
// we could potentially use this node instead of reserved node // we could potentially use this node instead of reserved node
return true; return true;
@ -1046,7 +982,8 @@ public class LeafQueue extends AbstractCSQueue {
// Otherwise, if any of the label of this node beyond queue limit, we // Otherwise, if any of the label of this node beyond queue limit, we
// cannot allocate on this node. Consider a small epsilon here. // cannot allocate on this node. Consider a small epsilon here.
if (potentialNewCapacity > getAbsoluteMaximumCapacityByNodeLabel(label) + 1e-4) { if (potentialNewCapacity > queueCapacities
.getAbsoluteMaximumCapacity(label) + 1e-4) {
canAssign = false; canAssign = false;
break; break;
} }
@ -1061,7 +998,8 @@ public class LeafQueue extends AbstractCSQueue {
queueUsage.getUsed(label), queueUsage.getUsed(label),
labelManager.getResourceByLabel(label, clusterResource)) labelManager.getResourceByLabel(label, clusterResource))
+ " potentialNewCapacity: " + potentialNewCapacity + " ( " + " potentialNewCapacity: " + potentialNewCapacity + " ( "
+ " max-capacity: " + absoluteMaxCapacity + ")"); + " max-capacity: " + queueCapacities.getAbsoluteMaximumCapacity()
+ ")");
} }
} }
@ -1144,7 +1082,7 @@ public class LeafQueue extends AbstractCSQueue {
Resources.multiplyAndNormalizeUp(resourceCalculator, Resources.multiplyAndNormalizeUp(resourceCalculator,
labelManager.getResourceByLabel(firstLabel, labelManager.getResourceByLabel(firstLabel,
clusterResource), clusterResource),
getAbsoluteCapacityByNodeLabel(firstLabel), queueCapacities.getAbsoluteCapacity(firstLabel),
minimumAllocation)); minimumAllocation));
} else { } else {
// else there's no label on request, just to use absolute capacity as // else there's no label on request, just to use absolute capacity as
@ -1152,7 +1090,7 @@ public class LeafQueue extends AbstractCSQueue {
queueCapacity = queueCapacity =
Resources.multiplyAndNormalizeUp(resourceCalculator, labelManager Resources.multiplyAndNormalizeUp(resourceCalculator, labelManager
.getResourceByLabel(CommonNodeLabelsManager.NO_LABEL, clusterResource), .getResourceByLabel(CommonNodeLabelsManager.NO_LABEL, clusterResource),
absoluteCapacity, minimumAllocation); queueCapacities.getAbsoluteCapacity(), minimumAllocation);
} }
// Allow progress for queues with miniscule capacity // Allow progress for queues with miniscule capacity
@ -1797,12 +1735,9 @@ public class LeafQueue extends AbstractCSQueue {
} }
private void updateAbsoluteCapacityResource(Resource clusterResource) { private void updateAbsoluteCapacityResource(Resource clusterResource) {
absoluteCapacityResource =
absoluteCapacityResource = Resources.multiplyAndNormalizeUp( Resources.multiplyAndNormalizeUp(resourceCalculator, clusterResource,
resourceCalculator, queueCapacities.getAbsoluteCapacity(), minimumAllocation);
clusterResource,
absoluteCapacity, minimumAllocation);
} }
@Override @Override
@ -1813,7 +1748,8 @@ public class LeafQueue extends AbstractCSQueue {
// Update headroom info based on new cluster resource value // Update headroom info based on new cluster resource value
// absoluteMaxCapacity now, will be replaced with absoluteMaxAvailCapacity // absoluteMaxCapacity now, will be replaced with absoluteMaxAvailCapacity
// during allocation // during allocation
updateHeadroomInfo(clusterResource, absoluteMaxCapacity); updateHeadroomInfo(clusterResource,
queueCapacities.getAbsoluteMaximumCapacity());
// Update metrics // Update metrics
CSQueueUtils.updateQueueStatistics( CSQueueUtils.updateQueueStatistics(
@ -1987,34 +1923,12 @@ public class LeafQueue extends AbstractCSQueue {
} }
} }
@Override
public float getAbsActualCapacity() {
//? Is this actually used by anything at present?
// There is a findbugs warning -re lastClusterResource (now excluded),
// when this is used, verify that the access is mt correct and remove
// the findbugs exclusion if possible
if (Resources.lessThanOrEqual(resourceCalculator, lastClusterResource,
lastClusterResource, Resources.none())) {
return absoluteCapacity;
}
Resource resourceRespectLabels =
labelManager == null ? lastClusterResource : labelManager
.getQueueResource(queueName, accessibleLabels, lastClusterResource);
float absActualCapacity =
Resources.divide(resourceCalculator, lastClusterResource,
resourceRespectLabels, lastClusterResource);
return absActualCapacity > absoluteCapacity ? absoluteCapacity
: absActualCapacity;
}
public void setCapacity(float capacity) { public void setCapacity(float capacity) {
this.capacity = capacity; queueCapacities.setCapacity(capacity);
} }
public void setAbsoluteCapacity(float absoluteCapacity) { public void setAbsoluteCapacity(float absoluteCapacity) {
this.absoluteCapacity = absoluteCapacity; queueCapacities.setAbsoluteCapacity(absoluteCapacity);
} }
public void setMaxApplications(int maxApplications) { public void setMaxApplications(int maxApplications) {

View File

@ -86,7 +86,7 @@ public class ParentQueue extends AbstractCSQueue {
this.rootQueue = (parent == null); this.rootQueue = (parent == null);
float rawCapacity = cs.getConfiguration().getCapacity(getQueuePath()); float rawCapacity = cs.getConfiguration().getNonLabeledQueueCapacity(getQueuePath());
if (rootQueue && if (rootQueue &&
(rawCapacity != CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE)) { (rawCapacity != CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE)) {
@ -95,44 +95,18 @@ public class ParentQueue extends AbstractCSQueue {
". Must be " + CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE); ". Must be " + CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE);
} }
float capacity = (float) rawCapacity / 100;
float parentAbsoluteCapacity =
(rootQueue) ? 1.0f : parent.getAbsoluteCapacity();
float absoluteCapacity = parentAbsoluteCapacity * capacity;
float maximumCapacity =
(float) cs.getConfiguration().getMaximumCapacity(getQueuePath()) / 100;
float absoluteMaxCapacity =
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
QueueState state = cs.getConfiguration().getState(getQueuePath());
Map<QueueACL, AccessControlList> acls =
cs.getConfiguration().getAcls(getQueuePath());
setupQueueConfigs(cs.getClusterResource(), capacity, absoluteCapacity,
maximumCapacity, absoluteMaxCapacity, state, acls, accessibleLabels,
defaultLabelExpression, capacitiyByNodeLabels, maxCapacityByNodeLabels,
cs.getConfiguration().getReservationContinueLook());
this.childQueues = new TreeSet<CSQueue>(queueComparator); this.childQueues = new TreeSet<CSQueue>(queueComparator);
setupQueueConfigs(cs.getClusterResource());
LOG.info("Initialized parent-queue " + queueName + LOG.info("Initialized parent-queue " + queueName +
" name=" + queueName + " name=" + queueName +
", fullname=" + getQueuePath()); ", fullname=" + getQueuePath());
} }
synchronized void setupQueueConfigs(Resource clusterResource, float capacity, synchronized void setupQueueConfigs(Resource clusterResource)
float absoluteCapacity, float maximumCapacity, float absoluteMaxCapacity, throws IOException {
QueueState state, Map<QueueACL, AccessControlList> acls, super.setupQueueConfigs(clusterResource);
Set<String> accessibleLabels, String defaultLabelExpression,
Map<String, Float> nodeLabelCapacities,
Map<String, Float> maximumCapacitiesByLabel,
boolean reservationContinueLooking) throws IOException {
super.setupQueueConfigs(clusterResource, capacity, absoluteCapacity,
maximumCapacity, absoluteMaxCapacity, state, acls, accessibleLabels,
defaultLabelExpression, nodeLabelCapacities, maximumCapacitiesByLabel,
reservationContinueLooking);
StringBuilder aclsString = new StringBuilder(); StringBuilder aclsString = new StringBuilder();
for (Map.Entry<QueueACL, AccessControlList> e : acls.entrySet()) { for (Map.Entry<QueueACL, AccessControlList> e : acls.entrySet()) {
aclsString.append(e.getKey() + ":" + e.getValue().getAclString()); aclsString.append(e.getKey() + ":" + e.getValue().getAclString());
@ -147,10 +121,10 @@ public class ParentQueue extends AbstractCSQueue {
} }
LOG.info(queueName + LOG.info(queueName +
", capacity=" + capacity + ", capacity=" + this.queueCapacities.getCapacity() +
", asboluteCapacity=" + absoluteCapacity + ", asboluteCapacity=" + this.queueCapacities.getAbsoluteCapacity() +
", maxCapacity=" + maximumCapacity + ", maxCapacity=" + this.queueCapacities.getMaximumCapacity() +
", asboluteMaxCapacity=" + absoluteMaxCapacity + ", asboluteMaxCapacity=" + this.queueCapacities.getAbsoluteMaximumCapacity() +
", state=" + state + ", state=" + state +
", acls=" + aclsString + ", acls=" + aclsString +
", labels=" + labelStrBuilder.toString() + "\n" + ", labels=" + labelStrBuilder.toString() + "\n" +
@ -166,19 +140,19 @@ public class ParentQueue extends AbstractCSQueue {
} }
float delta = Math.abs(1.0f - childCapacities); // crude way to check float delta = Math.abs(1.0f - childCapacities); // crude way to check
// allow capacities being set to 0, and enforce child 0 if parent is 0 // allow capacities being set to 0, and enforce child 0 if parent is 0
if (((capacity > 0) && (delta > PRECISION)) || if (((queueCapacities.getCapacity() > 0) && (delta > PRECISION)) ||
((capacity == 0) && (childCapacities > 0))) { ((queueCapacities.getCapacity() == 0) && (childCapacities > 0))) {
throw new IllegalArgumentException("Illegal" + throw new IllegalArgumentException("Illegal" +
" capacity of " + childCapacities + " capacity of " + childCapacities +
" for children of queue " + queueName); " for children of queue " + queueName);
} }
// check label capacities // check label capacities
for (String nodeLabel : labelManager.getClusterNodeLabels()) { for (String nodeLabel : labelManager.getClusterNodeLabels()) {
float capacityByLabel = getCapacityByNodeLabel(nodeLabel); float capacityByLabel = queueCapacities.getCapacity(nodeLabel);
// check children's labels // check children's labels
float sum = 0; float sum = 0;
for (CSQueue queue : childQueues) { for (CSQueue queue : childQueues) {
sum += queue.getCapacityByNodeLabel(nodeLabel); sum += queue.getQueueCapacities().getCapacity(nodeLabel);
} }
if ((capacityByLabel > 0 && Math.abs(1.0f - sum) > PRECISION) if ((capacityByLabel > 0 && Math.abs(1.0f - sum) > PRECISION)
|| (capacityByLabel == 0) && (sum > 0)) { || (capacityByLabel == 0) && (sum > 0)) {
@ -254,8 +228,8 @@ public class ParentQueue extends AbstractCSQueue {
public String toString() { public String toString() {
return queueName + ": " + return queueName + ": " +
"numChildQueue= " + childQueues.size() + ", " + "numChildQueue= " + childQueues.size() + ", " +
"capacity=" + capacity + ", " + "capacity=" + queueCapacities.getCapacity() + ", " +
"absoluteCapacity=" + absoluteCapacity + ", " + "absoluteCapacity=" + queueCapacities.getAbsoluteCapacity() + ", " +
"usedResources=" + queueUsage.getUsed() + "usedResources=" + queueUsage.getUsed() +
"usedCapacity=" + getUsedCapacity() + ", " + "usedCapacity=" + getUsedCapacity() + ", " +
"numApps=" + getNumApplications() + ", " + "numApps=" + getNumApplications() + ", " +
@ -263,9 +237,8 @@ public class ParentQueue extends AbstractCSQueue {
} }
@Override @Override
public synchronized void reinitialize( public synchronized void reinitialize(CSQueue newlyParsedQueue,
CSQueue newlyParsedQueue, Resource clusterResource) Resource clusterResource) throws IOException {
throws IOException {
// Sanity check // Sanity check
if (!(newlyParsedQueue instanceof ParentQueue) || if (!(newlyParsedQueue instanceof ParentQueue) ||
!newlyParsedQueue.getQueuePath().equals(getQueuePath())) { !newlyParsedQueue.getQueuePath().equals(getQueuePath())) {
@ -276,18 +249,7 @@ public class ParentQueue extends AbstractCSQueue {
ParentQueue newlyParsedParentQueue = (ParentQueue)newlyParsedQueue; ParentQueue newlyParsedParentQueue = (ParentQueue)newlyParsedQueue;
// Set new configs // Set new configs
setupQueueConfigs(clusterResource, setupQueueConfigs(clusterResource);
newlyParsedParentQueue.capacity,
newlyParsedParentQueue.absoluteCapacity,
newlyParsedParentQueue.maximumCapacity,
newlyParsedParentQueue.absoluteMaxCapacity,
newlyParsedParentQueue.state,
newlyParsedParentQueue.acls,
newlyParsedParentQueue.accessibleLabels,
newlyParsedParentQueue.defaultLabelExpression,
newlyParsedParentQueue.capacitiyByNodeLabels,
newlyParsedParentQueue.maxCapacityByNodeLabels,
newlyParsedParentQueue.reservationsContinueLooking);
// Re-configure existing child queues and add new ones // Re-configure existing child queues and add new ones
// The CS has already checked to ensure all existing child queues are present! // The CS has already checked to ensure all existing child queues are present!
@ -512,7 +474,7 @@ public class ParentQueue extends AbstractCSQueue {
labelManager.getResourceByLabel(label, clusterResource)); labelManager.getResourceByLabel(label, clusterResource));
// if any of the label doesn't beyond limit, we can allocate on this node // if any of the label doesn't beyond limit, we can allocate on this node
if (currentAbsoluteLabelUsedCapacity >= if (currentAbsoluteLabelUsedCapacity >=
getAbsoluteMaximumCapacityByNodeLabel(label)) { queueCapacities.getAbsoluteMaximumCapacity(label)) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug(getQueueName() + " used=" + queueUsage.getUsed() LOG.debug(getQueueName() + " used=" + queueUsage.getUsed()
+ " current-capacity (" + queueUsage.getUsed(label) + ") " + " current-capacity (" + queueUsage.getUsed(label) + ") "
@ -540,7 +502,8 @@ public class ParentQueue extends AbstractCSQueue {
Resources.subtract(queueUsage.getUsed(), reservedResources), Resources.subtract(queueUsage.getUsed(), reservedResources),
clusterResource); clusterResource);
if (capacityWithoutReservedCapacity <= absoluteMaxCapacity) { if (capacityWithoutReservedCapacity <= queueCapacities
.getAbsoluteMaximumCapacity()) {
if (LOG.isDebugEnabled()) { if (LOG.isDebugEnabled()) {
LOG.debug("parent: try to use reserved: " + getQueueName() LOG.debug("parent: try to use reserved: " + getQueueName()
+ " usedResources: " + queueUsage.getUsed().getMemory() + " usedResources: " + queueUsage.getUsed().getMemory()
@ -550,7 +513,7 @@ public class ParentQueue extends AbstractCSQueue {
/ clusterResource.getMemory() / clusterResource.getMemory()
+ " potentialNewWithoutReservedCapacity: " + " potentialNewWithoutReservedCapacity: "
+ capacityWithoutReservedCapacity + " ( " + " max-capacity: " + capacityWithoutReservedCapacity + " ( " + " max-capacity: "
+ absoluteMaxCapacity + ")"); + queueCapacities.getAbsoluteMaximumCapacity() + ")");
} }
// we could potentially use this node instead of reserved node // we could potentially use this node instead of reserved node
return true; return true;
@ -761,13 +724,6 @@ public class ParentQueue extends AbstractCSQueue {
} }
} }
@Override
public float getAbsActualCapacity() {
// for now, simply return actual capacity = guaranteed capacity for parent
// queue
return absoluteCapacity;
}
public synchronized int getNumApplications() { public synchronized int getNumApplications() {
return numApplications; return numApplications;
} }

View File

@ -19,8 +19,6 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity; package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import java.io.IOException; import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator; import java.util.Iterator;
import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.api.records.Resource;
@ -101,16 +99,7 @@ public class PlanQueue extends ParentQueue {
} }
// Set new configs // Set new configs
setupQueueConfigs(clusterResource, newlyParsedParentQueue.getCapacity(), setupQueueConfigs(clusterResource);
newlyParsedParentQueue.getAbsoluteCapacity(),
newlyParsedParentQueue.getMaximumCapacity(),
newlyParsedParentQueue.getAbsoluteMaximumCapacity(),
newlyParsedParentQueue.getState(), newlyParsedParentQueue.getACLs(),
newlyParsedParentQueue.accessibleLabels,
newlyParsedParentQueue.defaultLabelExpression,
newlyParsedParentQueue.capacitiyByNodeLabels,
newlyParsedParentQueue.maxCapacityByNodeLabels,
newlyParsedParentQueue.getReservationContinueLooking());
updateQuotas(newlyParsedParentQueue.userLimit, updateQuotas(newlyParsedParentQueue.userLimit,
newlyParsedParentQueue.userLimitFactor, newlyParsedParentQueue.userLimitFactor,

View File

@ -19,12 +19,18 @@
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity; package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import java.util.HashMap; import java.util.HashMap;
import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock; import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager; import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
import com.google.common.collect.Sets;
public class QueueCapacities { public class QueueCapacities {
private static final String NL = CommonNodeLabelsManager.NO_LABEL; private static final String NL = CommonNodeLabelsManager.NO_LABEL;
@ -32,13 +38,15 @@ public class QueueCapacities {
private Map<String, Capacities> capacitiesMap; private Map<String, Capacities> capacitiesMap;
private ReadLock readLock; private ReadLock readLock;
private WriteLock writeLock; private WriteLock writeLock;
private final boolean isRoot;
public QueueCapacities() { public QueueCapacities(boolean isRoot) {
ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
readLock = lock.readLock(); readLock = lock.readLock();
writeLock = lock.writeLock(); writeLock = lock.writeLock();
capacitiesMap = new HashMap<String, Capacities>(); capacitiesMap = new HashMap<String, Capacities>();
this.isRoot = isRoot;
} }
// Usage enum here to make implement cleaner // Usage enum here to make implement cleaner
@ -58,6 +66,18 @@ public class QueueCapacities {
public Capacities() { public Capacities() {
capacitiesArr = new float[CapacityType.values().length]; capacitiesArr = new float[CapacityType.values().length];
} }
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("{used=" + capacitiesArr[0] + "%, ");
sb.append("abs_used=" + capacitiesArr[1] + "%, ");
sb.append("max_cap=" + capacitiesArr[2] + "%, ");
sb.append("abs_max_cap=" + capacitiesArr[3] + "%, ");
sb.append("cap=" + capacitiesArr[4] + "%, ");
sb.append("abs_cap=" + capacitiesArr[5] + "%}");
return sb.toString();
}
} }
private float _get(String label, CapacityType type) { private float _get(String label, CapacityType type) {
@ -127,6 +147,10 @@ public class QueueCapacities {
} }
public float getCapacity(String label) { public float getCapacity(String label) {
if (StringUtils.equals(label, RMNodeLabelsManager.NO_LABEL) && isRoot) {
return 1f;
}
return _get(label, CapacityType.CAP); return _get(label, CapacityType.CAP);
} }
@ -144,6 +168,9 @@ public class QueueCapacities {
} }
public float getAbsoluteCapacity(String label) { public float getAbsoluteCapacity(String label) {
if (StringUtils.equals(label, RMNodeLabelsManager.NO_LABEL) && isRoot) {
return 1f;
}
return _get(label, CapacityType.ABS_CAP); return _get(label, CapacityType.ABS_CAP);
} }
@ -188,4 +215,43 @@ public class QueueCapacities {
public void setAbsoluteMaximumCapacity(String label, float value) { public void setAbsoluteMaximumCapacity(String label, float value) {
_set(label, CapacityType.ABS_MAX_CAP, value); _set(label, CapacityType.ABS_MAX_CAP, value);
} }
/**
* Clear configurable fields, like
* (absolute)capacity/(absolute)maximum-capacity, this will be used by queue
* reinitialize, when we reinitialize a queue, we will first clear all
* configurable fields, and load new values
*/
public void clearConfigurableFields() {
try {
writeLock.lock();
for (String label : capacitiesMap.keySet()) {
_set(label, CapacityType.CAP, 0);
_set(label, CapacityType.MAX_CAP, 0);
_set(label, CapacityType.ABS_CAP, 0);
_set(label, CapacityType.ABS_MAX_CAP, 0);
}
} finally {
writeLock.unlock();
}
}
public Set<String> getExistingNodeLabels() {
try {
readLock.lock();
return new HashSet<String>(capacitiesMap.keySet());
} finally {
readLock.unlock();
}
}
@Override
public String toString() {
try {
readLock.lock();
return this.capacitiesMap.toString();
} finally {
readLock.unlock();
}
}
} }

View File

@ -62,11 +62,11 @@ public class ReservationQueue extends LeafQueue {
throw new IOException("Trying to reinitialize " + getQueuePath() throw new IOException("Trying to reinitialize " + getQueuePath()
+ " from " + newlyParsedQueue.getQueuePath()); + " from " + newlyParsedQueue.getQueuePath());
} }
super.reinitialize(newlyParsedQueue, clusterResource);
CSQueueUtils.updateQueueStatistics( CSQueueUtils.updateQueueStatistics(
parent.schedulerContext.getResourceCalculator(), newlyParsedQueue, parent.schedulerContext.getResourceCalculator(), newlyParsedQueue,
parent, parent.schedulerContext.getClusterResource(), parent, parent.schedulerContext.getClusterResource(),
parent.schedulerContext.getMinimumResourceCapability()); parent.schedulerContext.getMinimumResourceCapability());
super.reinitialize(newlyParsedQueue, clusterResource);
updateQuotas(parent.getUserLimitForReservation(), updateQuotas(parent.getUserLimitForReservation(),
parent.getUserLimitFactor(), parent.getUserLimitFactor(),
parent.getMaxApplicationsForReservations(), parent.getMaxApplicationsForReservations(),
@ -108,9 +108,9 @@ public class ReservationQueue extends LeafQueue {
maxApplicationsPerUser = maxAppsPerUserForReservation; maxApplicationsPerUser = maxAppsPerUserForReservation;
} }
// used by the super constructor, we initialize to zero @Override
protected float getCapacityFromConf() { protected void setupConfigurableCapacities() {
return 0f; CSQueueUtils.updateAndCheckCapacitiesByLabel(getQueuePath(),
queueCapacities, parent == null ? null : parent.getQueueCapacities());
} }
} }

View File

@ -202,33 +202,33 @@ public class TestCSQueueUtils {
LOG.info("t2 l2q2 " + result); LOG.info("t2 l2q2 " + result);
//some usage, but below the base capacity //some usage, but below the base capacity
root.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f));
l1q2.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); l1q2.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f));
result = CSQueueUtils.getAbsoluteMaxAvailCapacity( result = CSQueueUtils.getAbsoluteMaxAvailCapacity(
resourceCalculator, clusterResource, l2q2); resourceCalculator, clusterResource, l2q2);
assertEquals( 0.4f, result, 0.000001f); assertEquals( 0.4f, result, 0.000001f);
LOG.info("t2 l2q2 " + result); LOG.info("t2 l2q2 " + result);
//usage gt base on parent sibling //usage gt base on parent sibling
root.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.3f)); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.3f));
l1q2.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.3f)); l1q2.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.3f));
result = CSQueueUtils.getAbsoluteMaxAvailCapacity( result = CSQueueUtils.getAbsoluteMaxAvailCapacity(
resourceCalculator, clusterResource, l2q2); resourceCalculator, clusterResource, l2q2);
assertEquals( 0.3f, result, 0.000001f); assertEquals( 0.3f, result, 0.000001f);
LOG.info("t2 l2q2 " + result); LOG.info("t2 l2q2 " + result);
//same as last, but with usage also on direct parent //same as last, but with usage also on direct parent
root.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f));
l1q1.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); l1q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f));
result = CSQueueUtils.getAbsoluteMaxAvailCapacity( result = CSQueueUtils.getAbsoluteMaxAvailCapacity(
resourceCalculator, clusterResource, l2q2); resourceCalculator, clusterResource, l2q2);
assertEquals( 0.3f, result, 0.000001f); assertEquals( 0.3f, result, 0.000001f);
LOG.info("t2 l2q2 " + result); LOG.info("t2 l2q2 " + result);
//add to direct sibling, below the threshold of effect at present //add to direct sibling, below the threshold of effect at present
root.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f));
l1q1.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l1q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f));
l2q1.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l2q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f));
result = CSQueueUtils.getAbsoluteMaxAvailCapacity( result = CSQueueUtils.getAbsoluteMaxAvailCapacity(
resourceCalculator, clusterResource, l2q2); resourceCalculator, clusterResource, l2q2);
assertEquals( 0.3f, result, 0.000001f); assertEquals( 0.3f, result, 0.000001f);
@ -236,9 +236,9 @@ public class TestCSQueueUtils {
//add to direct sibling, now above the threshold of effect //add to direct sibling, now above the threshold of effect
//(it's cumulative with prior tests) //(it's cumulative with prior tests)
root.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f));
l1q1.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l1q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f));
l2q1.getResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l2q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f));
result = CSQueueUtils.getAbsoluteMaxAvailCapacity( result = CSQueueUtils.getAbsoluteMaxAvailCapacity(
resourceCalculator, clusterResource, l2q2); resourceCalculator, clusterResource, l2q2);
assertEquals( 0.1f, result, 0.000001f); assertEquals( 0.1f, result, 0.000001f);

View File

@ -389,11 +389,11 @@ public class TestCapacityScheduler {
public void testMaximumCapacitySetup() { public void testMaximumCapacitySetup() {
float delta = 0.0000001f; float delta = 0.0000001f;
CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration(); CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
assertEquals(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE,conf.getMaximumCapacity(A),delta); assertEquals(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE,conf.getNonLabeledQueueMaximumCapacity(A),delta);
conf.setMaximumCapacity(A, 50.0f); conf.setMaximumCapacity(A, 50.0f);
assertEquals(50.0f, conf.getMaximumCapacity(A),delta); assertEquals(50.0f, conf.getNonLabeledQueueMaximumCapacity(A),delta);
conf.setMaximumCapacity(A, -1); conf.setMaximumCapacity(A, -1);
assertEquals(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE,conf.getMaximumCapacity(A),delta); assertEquals(CapacitySchedulerConfiguration.MAXIMUM_CAPACITY_VALUE,conf.getNonLabeledQueueMaximumCapacity(A),delta);
} }

View File

@ -91,7 +91,8 @@ public class TestCapacitySchedulerNodeLabelUpdate {
String label) { String label) {
CapacityScheduler scheduler = (CapacityScheduler) rm.getResourceScheduler(); CapacityScheduler scheduler = (CapacityScheduler) rm.getResourceScheduler();
CSQueue queue = scheduler.getQueue(queueName); CSQueue queue = scheduler.getQueue(queueName);
Assert.assertEquals(memory, queue.getUsedResourceByLabel(label).getMemory()); Assert.assertEquals(memory, queue.getQueueResourceUsage().getUsed(label)
.getMemory());
} }
@Test (timeout = 30000) @Test (timeout = 30000)

View File

@ -99,7 +99,7 @@ public class TestQueueCapacities {
} }
private void internalTestModifyAndRead(String label) throws Exception { private void internalTestModifyAndRead(String label) throws Exception {
QueueCapacities qc = new QueueCapacities(); QueueCapacities qc = new QueueCapacities(false);
// First get returns 0 always // First get returns 0 always
Assert.assertEquals(0f, get(qc, suffix, label), 1e-8); Assert.assertEquals(0f, get(qc, suffix, label), 1e-8);

View File

@ -418,18 +418,18 @@ public class TestQueueParsing {
// check capacity of A2 // check capacity of A2
CSQueue qA2 = capacityScheduler.getQueue("a2"); CSQueue qA2 = capacityScheduler.getQueue("a2");
Assert.assertEquals(0.7, qA2.getCapacity(), DELTA); Assert.assertEquals(0.7, qA2.getCapacity(), DELTA);
Assert.assertEquals(0.5, qA2.getCapacityByNodeLabel("red"), DELTA); Assert.assertEquals(0.5, qA2.getQueueCapacities().getCapacity("red"), DELTA);
Assert.assertEquals(0.07, qA2.getAbsoluteCapacity(), DELTA); Assert.assertEquals(0.07, qA2.getAbsoluteCapacity(), DELTA);
Assert.assertEquals(0.25, qA2.getAbsoluteCapacityByNodeLabel("red"), DELTA); Assert.assertEquals(0.25, qA2.getQueueCapacities().getAbsoluteCapacity("red"), DELTA);
Assert.assertEquals(0.1275, qA2.getAbsoluteMaximumCapacity(), DELTA); Assert.assertEquals(0.1275, qA2.getAbsoluteMaximumCapacity(), DELTA);
Assert.assertEquals(0.3, qA2.getAbsoluteMaximumCapacityByNodeLabel("red"), DELTA); Assert.assertEquals(0.3, qA2.getQueueCapacities().getAbsoluteMaximumCapacity("red"), DELTA);
// check capacity of B3 // check capacity of B3
CSQueue qB3 = capacityScheduler.getQueue("b3"); CSQueue qB3 = capacityScheduler.getQueue("b3");
Assert.assertEquals(0.18, qB3.getAbsoluteCapacity(), DELTA); Assert.assertEquals(0.18, qB3.getAbsoluteCapacity(), DELTA);
Assert.assertEquals(0.125, qB3.getAbsoluteCapacityByNodeLabel("red"), DELTA); Assert.assertEquals(0.125, qB3.getQueueCapacities().getAbsoluteCapacity("red"), DELTA);
Assert.assertEquals(0.35, qB3.getAbsoluteMaximumCapacity(), DELTA); Assert.assertEquals(0.35, qB3.getAbsoluteMaximumCapacity(), DELTA);
Assert.assertEquals(1, qB3.getAbsoluteMaximumCapacityByNodeLabel("red"), DELTA); Assert.assertEquals(1, qB3.getQueueCapacities().getAbsoluteMaximumCapacity("red"), DELTA);
} }
private void private void
@ -641,8 +641,8 @@ public class TestQueueParsing {
// check root queue's capacity by label -- they should be all zero // check root queue's capacity by label -- they should be all zero
CSQueue root = capacityScheduler.getQueue(CapacitySchedulerConfiguration.ROOT); CSQueue root = capacityScheduler.getQueue(CapacitySchedulerConfiguration.ROOT);
Assert.assertEquals(0, root.getCapacityByNodeLabel("red"), DELTA); Assert.assertEquals(0, root.getQueueCapacities().getCapacity("red"), DELTA);
Assert.assertEquals(0, root.getCapacityByNodeLabel("blue"), DELTA); Assert.assertEquals(0, root.getQueueCapacities().getCapacity("blue"), DELTA);
CSQueue a = capacityScheduler.getQueue("a"); CSQueue a = capacityScheduler.getQueue("a");
Assert.assertEquals(0.10, a.getAbsoluteCapacity(), DELTA); Assert.assertEquals(0.10, a.getAbsoluteCapacity(), DELTA);

View File

@ -395,9 +395,9 @@ public class TestRMWebServicesCapacitySched extends JerseyTest {
String qshortName = qArr[qArr.length - 1]; String qshortName = qArr[qArr.length - 1];
assertEquals("usedCapacity doesn't match", 0, info.usedCapacity, 1e-3f); assertEquals("usedCapacity doesn't match", 0, info.usedCapacity, 1e-3f);
assertEquals("capacity doesn't match", csConf.getCapacity(q), assertEquals("capacity doesn't match", csConf.getNonLabeledQueueCapacity(q),
info.capacity, 1e-3f); info.capacity, 1e-3f);
float expectCapacity = csConf.getMaximumCapacity(q); float expectCapacity = csConf.getNonLabeledQueueMaximumCapacity(q);
float expectAbsMaxCapacity = parentAbsMaxCapacity * (info.maxCapacity/100); float expectAbsMaxCapacity = parentAbsMaxCapacity * (info.maxCapacity/100);
if (CapacitySchedulerConfiguration.UNDEFINED == expectCapacity) { if (CapacitySchedulerConfiguration.UNDEFINED == expectCapacity) {
expectCapacity = 100; expectCapacity = 100;