Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

MAPREDUCE-3683. Fixed maxCapacity of queues to be product of parent maxCapacities. 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1235858 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Arun Murthy 2012-01-25 18:17:55 +00:00
parent a02f07c4bc
commit 9d1621da52
8 changed files with 110 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-mapreduce-project/CHANGES.txt
View File

@ -558,6 +558,9 @@ Release 0.23.1 - Unreleased
MAPREDUCE-3630. Fixes a NullPointer exception while running TeraGen - if a
map is asked to generate 0 records. (Mahadev Konar via sseth)
MAPREDUCE-3683. Fixed maxCapacity of queues to be product of parent
maxCapacities. (acmurthy)
Release 0.23.0 - 2011-11-01
INCOMPATIBLE CHANGES

30
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueueUtils.java
View File

@ -17,12 +17,19 @@
*/
package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
import org.apache.hadoop.yarn.api.records.Resource;
class CSQueueUtils {
public static void checkMaxCapacity(String queueName,
float capacity, float maximumCapacity) {
if (Math.round(100 * maximumCapacity) != CapacitySchedulerConfiguration.UNDEFINED &&
if (maximumCapacity < 0.0f || maximumCapacity > 1.0f ||
maximumCapacity < capacity) {
throw new IllegalArgumentException(
"Illegal value of maximumCapacity " + maximumCapacity +
" used in call to setMaxCapacity for queue " + queueName);
}
if (maximumCapacity < capacity) {
throw new IllegalArgumentException(
"Illegal call to setMaxCapacity. " +
"Queue '" + queueName + "' has " +
@ -30,5 +37,26 @@ public static void checkMaxCapacity(String queueName,
"maximumCapacity (" + maximumCapacity + ")" );
}
}
public static float computeAbsoluteMaximumCapacity(
float maximumCapacity, CSQueue parent) {
float parentAbsMaxCapacity =
(parent == null) ? 1.0f : parent.getAbsoluteMaximumCapacity();
return (parentAbsMaxCapacity * maximumCapacity);
}
public static int computeMaxActiveApplications(Resource clusterResource,
float maxAMResourcePercent, float absoluteCapacity) {
return
Math.max(
(int)((clusterResource.getMemory() / (float)LeafQueue.DEFAULT_AM_RESOURCE) *
maxAMResourcePercent * absoluteCapacity),
1);
}
public static int computeMaxActiveApplicationsPerUser(
int maxActiveApplications, int userLimit, float userLimitFactor) {
return (int)(maxActiveApplications * (userLimit / 100.0f) * userLimitFactor);
}
}

8
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java
View File

@ -149,7 +149,7 @@ public int getCapacity(String queue) {
throw new IllegalArgumentException("Illegal " +
"capacity of " + capacity + " for queue " + queue);
}
LOG.debug("CSConf - setCapacity: queuePrefix=" + getQueuePrefix(queue) +
LOG.debug("CSConf - getCapacity: queuePrefix=" + getQueuePrefix(queue) +
", capacity=" + capacity);
return capacity;
}
@ -162,11 +162,15 @@ public void setCapacity(String queue, int capacity) {
public int getMaximumCapacity(String queue) {
int maxCapacity =
getInt(getQueuePrefix(queue) + MAXIMUM_CAPACITY, UNDEFINED);
getInt(getQueuePrefix(queue) + MAXIMUM_CAPACITY, MAXIMUM_CAPACITY_VALUE);
return maxCapacity;
}
public void setMaximumCapacity(String queue, int maxCapacity) {
if (maxCapacity > MAXIMUM_CAPACITY_VALUE) {
throw new IllegalArgumentException("Illegal " +
"maximum-capacity of " + maxCapacity + " for queue " + queue);
}
setInt(getQueuePrefix(queue) + MAXIMUM_CAPACITY, maxCapacity);
LOG.debug("CSConf - setMaxCapacity: queuePrefix=" + getQueuePrefix(queue) +
", maxCapacity=" + maxCapacity);

49
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java
View File

@ -144,10 +144,10 @@ public LeafQueue(CapacitySchedulerContext cs,
(float)cs.getConfiguration().getCapacity(getQueuePath()) / 100;
float absoluteCapacity = parent.getAbsoluteCapacity() * capacity;
float maximumCapacity = (float)cs.getConfiguration().getMaximumCapacity(getQueuePath()) / 100;
float maximumCapacity =
(float)cs.getConfiguration().getMaximumCapacity(getQueuePath()) / 100;
float absoluteMaxCapacity =
(Math.round(maximumCapacity * 100) == CapacitySchedulerConfiguration.UNDEFINED) ?
Float.MAX_VALUE : (parent.getAbsoluteCapacity() * maximumCapacity);
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
int userLimit = cs.getConfiguration().getUserLimit(getQueuePath());
float userLimitFactor =
@ -161,10 +161,10 @@ public LeafQueue(CapacitySchedulerContext cs,
this.maxAMResourcePercent =
cs.getConfiguration().getMaximumApplicationMasterResourcePercent();
int maxActiveApplications =
computeMaxActiveApplications(cs.getClusterResources(),
CSQueueUtils.computeMaxActiveApplications(cs.getClusterResources(),
maxAMResourcePercent, absoluteCapacity);
int maxActiveApplicationsPerUser =
computeMaxActiveApplicationsPerUser(maxActiveApplications, userLimit,
CSQueueUtils.computeMaxActiveApplicationsPerUser(maxActiveApplications, userLimit,
userLimitFactor);
this.queueInfo = recordFactory.newRecordInstance(QueueInfo.class);
@ -193,20 +193,6 @@ public LeafQueue(CapacitySchedulerContext cs,
this.activeApplications = new TreeSet<SchedulerApp>(applicationComparator);
}
private int computeMaxActiveApplications(Resource clusterResource,
float maxAMResourcePercent, float absoluteCapacity) {
return
Math.max(
(int)((clusterResource.getMemory() / (float)DEFAULT_AM_RESOURCE) *
maxAMResourcePercent * absoluteCapacity),
1);
}
private int computeMaxActiveApplicationsPerUser(int maxActiveApplications,
int userLimit, float userLimitFactor) {
return (int)(maxActiveApplications * (userLimit / 100.0f) * userLimitFactor);
}
private synchronized void setupQueueConfigs(
float capacity, float absoluteCapacity,
float maximumCapacity, float absoluteMaxCapacity,
@ -254,8 +240,8 @@ private synchronized void setupQueueConfigs(
"maxCapacity = " + maximumCapacity +
" [= configuredMaxCapacity ]" + "\n" +
"absoluteMaxCapacity = " + absoluteMaxCapacity +
" [= Float.MAX_VALUE if maximumCapacity undefined, " +
"(parentAbsoluteCapacity * maximumCapacity) / 100 otherwise ]" + "\n" +
" [= 1.0 maximumCapacity undefined, " +
"(parentAbsoluteMaxCapacity * maximumCapacity) / 100 otherwise ]" + "\n" +
"userLimit = " + userLimit +
" [= configuredUserLimit ]" + "\n" +
"userLimitFactor = " + userLimitFactor +
@ -400,9 +386,7 @@ synchronized void setMaxCapacity(float maximumCapacity) {
this.maximumCapacity = maximumCapacity;
this.absoluteMaxCapacity =
(Math.round(maximumCapacity * 100) == CapacitySchedulerConfiguration.UNDEFINED) ?
Float.MAX_VALUE :
(parent.getAbsoluteCapacity() * maximumCapacity);
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
}
/**
@ -835,13 +819,14 @@ private synchronized boolean assignToQueue(Resource clusterResource,
float potentialNewCapacity =
(float)(usedResources.getMemory() + required.getMemory()) /
clusterResource.getMemory();
LOG.info(getQueueName() +
" usedResources: " + usedResources.getMemory() +
" currentCapacity " + ((float)usedResources.getMemory())/clusterResource.getMemory() +
" required " + required.getMemory() +
" potentialNewCapacity: " + potentialNewCapacity + " ( " +
" max-capacity: " + absoluteMaxCapacity + ")");
if (potentialNewCapacity > absoluteMaxCapacity) {
LOG.info(getQueueName() +
" usedResources: " + usedResources.getMemory() +
" clusterResources: " + clusterResource.getMemory() +
" currentCapacity " + ((float)usedResources.getMemory())/clusterResource.getMemory() +
" required " + required.getMemory() +
" potentialNewCapacity: " + potentialNewCapacity + " ( " +
" max-capacity: " + absoluteMaxCapacity + ")");
return false;
}
return true;
@ -1308,10 +1293,10 @@ synchronized void releaseResource(Resource clusterResource,
public synchronized void updateClusterResource(Resource clusterResource) {
// Update queue properties
maxActiveApplications =
computeMaxActiveApplications(clusterResource, maxAMResourcePercent,
CSQueueUtils.computeMaxActiveApplications(clusterResource, maxAMResourcePercent,
absoluteCapacity);
maxActiveApplicationsPerUser =
computeMaxActiveApplicationsPerUser(maxActiveApplications, userLimit,
CSQueueUtils.computeMaxActiveApplicationsPerUser(maxActiveApplications, userLimit,
userLimitFactor);
// Update application properties

14
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
View File

@ -118,16 +118,14 @@ public ParentQueue(CapacitySchedulerContext cs,
}
float capacity = (float) rawCapacity / 100;
float parentAbsoluteCapacity =
(parent == null) ? 1.0f : parent.getAbsoluteCapacity();
(rootQueue) ? 1.0f : parent.getAbsoluteCapacity();
float absoluteCapacity = parentAbsoluteCapacity * capacity;
float maximumCapacity =
float maximumCapacity =
(float) cs.getConfiguration().getMaximumCapacity(getQueuePath()) / 100;
float absoluteMaxCapacity =
(Math.round(maximumCapacity * 100) == CapacitySchedulerConfiguration.UNDEFINED) ?
Float.MAX_VALUE : (parentAbsoluteCapacity * maximumCapacity);
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
QueueState state = cs.getConfiguration().getState(getQueuePath());
@ -497,12 +495,8 @@ synchronized void setMaxCapacity(float maximumCapacity) {
CSQueueUtils.checkMaxCapacity(getQueueName(), capacity, maximumCapacity);
this.maximumCapacity = maximumCapacity;
float parentAbsoluteCapacity =
(rootQueue) ? 100.0f : parent.getAbsoluteCapacity();
this.absoluteMaxCapacity =
(maximumCapacity == CapacitySchedulerConfiguration.UNDEFINED) ?
Float.MAX_VALUE :
(parentAbsoluteCapacity * maximumCapacity);
CSQueueUtils.computeAbsoluteMaximumCapacity(maximumCapacity, parent);
}
@Override

14
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java
View File

@ -255,7 +255,7 @@ public void testSingleQueueWithOneUser() throws Exception {
// Manipulate queue 'a'
LeafQueue a = stubLeafQueue((LeafQueue)queues.get(A));
//unset maxCapacity
a.setMaxCapacity(-0.01f);
a.setMaxCapacity(1.0f);
// Users
final String user_0 = "user_0";
@ -377,7 +377,7 @@ public void testSingleQueueWithMultipleUsers() throws Exception {
// Mock the queue
LeafQueue a = stubLeafQueue((LeafQueue)queues.get(A));
//unset maxCapacity
a.setMaxCapacity(-0.01f);
a.setMaxCapacity(1.0f);
// Users
final String user_0 = "user_0";
@ -491,7 +491,7 @@ public void testSingleQueueWithMultipleUsers() throws Exception {
// Revert max-capacity and user-limit-factor
// Now, allocations should goto app_3 since it's under user-limit
a.setMaxCapacity(-0.01f);
a.setMaxCapacity(1.0f);
a.setUserLimitFactor(1);
a.assignContainers(clusterResource, node_0);
assertEquals(7*GB, a.getUsedResources().getMemory());
@ -548,7 +548,7 @@ public void testReservation() throws Exception {
// Manipulate queue 'a'
LeafQueue a = stubLeafQueue((LeafQueue)queues.get(A));
//unset maxCapacity
a.setMaxCapacity(-0.01f);
a.setMaxCapacity(1.0f);
// Users
final String user_0 = "user_0";
@ -571,7 +571,7 @@ public void testReservation() throws Exception {
String host_0 = "host_0";
SchedulerNode node_0 = TestUtils.getMockNode(host_0, DEFAULT_RACK, 0, 4*GB);
final int numNodes = 1;
final int numNodes = 2;
Resource clusterResource = Resources.createResource(numNodes * (4*GB));
when(csContext.getNumClusterNodes()).thenReturn(numNodes);
@ -646,7 +646,7 @@ public void testReservationExchange() throws Exception {
// Manipulate queue 'a'
LeafQueue a = stubLeafQueue((LeafQueue)queues.get(A));
//unset maxCapacity
a.setMaxCapacity(-0.01f);
a.setMaxCapacity(1.0f);
a.setUserLimitFactor(10);
// Users
@ -673,7 +673,7 @@ public void testReservationExchange() throws Exception {
String host_1 = "host_1";
SchedulerNode node_1 = TestUtils.getMockNode(host_1, DEFAULT_RACK, 0, 4*GB);
final int numNodes = 2;
final int numNodes = 3;
Resource clusterResource = Resources.createResource(numNodes * (4*GB));
when(csContext.getNumClusterNodes()).thenReturn(numNodes);
when(csContext.getMaximumResourceCapability()).thenReturn(

32
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestQueueParsing.java
View File

@ -30,6 +30,8 @@ public class TestQueueParsing {
private static final Log LOG = LogFactory.getLog(TestQueueParsing.class);
private static final double DELTA = 0.000001;
@Test
public void testQueueParsing() throws Exception {
CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();
@ -37,6 +39,20 @@ public void testQueueParsing() throws Exception {
CapacityScheduler capacityScheduler = new CapacityScheduler();
capacityScheduler.reinitialize(conf, null, null);
CSQueue a = capacityScheduler.getQueue("a");
Assert.assertEquals(0.10, a.getAbsoluteCapacity(), DELTA);
Assert.assertEquals(0.15, a.getAbsoluteMaximumCapacity(), DELTA);
CSQueue b1 = capacityScheduler.getQueue("b1");
Assert.assertEquals(0.2 * 0.5, b1.getAbsoluteCapacity(), DELTA);
Assert.assertEquals("Parent B has no MAX_CAP",
0.85, b1.getAbsoluteMaximumCapacity(), DELTA);
CSQueue c12 = capacityScheduler.getQueue("c12");
Assert.assertEquals(0.7 * 0.5 * 0.45, c12.getAbsoluteCapacity(), DELTA);
Assert.assertEquals(0.7 * 0.55 * 0.7,
c12.getAbsoluteMaximumCapacity(), DELTA);
}
private void setupQueueConfiguration(CapacitySchedulerConfiguration conf) {
@ -47,12 +63,14 @@ private void setupQueueConfiguration(CapacitySchedulerConfiguration conf) {
final String A = CapacitySchedulerConfiguration.ROOT + ".a";
conf.setCapacity(A, 10);
conf.setMaximumCapacity(A, 15);
final String B = CapacitySchedulerConfiguration.ROOT + ".b";
conf.setCapacity(B, 20);
final String C = CapacitySchedulerConfiguration.ROOT + ".c";
conf.setCapacity(C, 70);
conf.setMaximumCapacity(C, 70);
LOG.info("Setup top-level queues");
@ -61,15 +79,20 @@ private void setupQueueConfiguration(CapacitySchedulerConfiguration conf) {
final String A2 = A + ".a2";
conf.setQueues(A, new String[] {"a1", "a2"});
conf.setCapacity(A1, 30);
conf.setMaximumCapacity(A1, 45);
conf.setCapacity(A2, 70);
conf.setMaximumCapacity(A2, 85);
final String B1 = B + ".b1";
final String B2 = B + ".b2";
final String B3 = B + ".b3";
conf.setQueues(B, new String[] {"b1", "b2", "b3"});
conf.setCapacity(B1, 50);
conf.setMaximumCapacity(B1, 85);
conf.setCapacity(B2, 30);
conf.setMaximumCapacity(B2, 35);
conf.setCapacity(B3, 20);
conf.setMaximumCapacity(B3, 35);
final String C1 = C + ".c1";
final String C2 = C + ".c2";
@ -77,9 +100,13 @@ private void setupQueueConfiguration(CapacitySchedulerConfiguration conf) {
final String C4 = C + ".c4";
conf.setQueues(C, new String[] {"c1", "c2", "c3", "c4"});
conf.setCapacity(C1, 50);
conf.setMaximumCapacity(C1, 55);
conf.setCapacity(C2, 10);
conf.setMaximumCapacity(C2, 25);
conf.setCapacity(C3, 35);
conf.setMaximumCapacity(C3, 38);
conf.setCapacity(C4, 5);
conf.setMaximumCapacity(C4, 5);
LOG.info("Setup 2nd-level queues");
@ -89,8 +116,11 @@ private void setupQueueConfiguration(CapacitySchedulerConfiguration conf) {
final String C13 = C1 + ".c13";
conf.setQueues(C1, new String[] {"c11", "c12", "c13"});
conf.setCapacity(C11, 15);
conf.setMaximumCapacity(C11, 30);
conf.setCapacity(C12, 45);
conf.setMaximumCapacity(C12, 70);
conf.setCapacity(C13, 40);
conf.setMaximumCapacity(C13, 40);
LOG.info("Setup 3rd-level queues");
}

23
hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesCapacitySched.java
View File

@ -235,12 +235,13 @@ public void verifyClusterSchedulerXML(NodeList nodes) throws Exception {
Element qElem = (Element) queues.item(j);
String qName = WebServicesTestUtils.getXmlString(qElem, "queueName");
String q = CapacitySchedulerConfiguration.ROOT + "." + qName;
verifySubQueueXML(qElem, q, 100);
verifySubQueueXML(qElem, q, 100, 100);
}
}
}
public void verifySubQueueXML(Element qElem, String q, float parentAbsCapacity)
public void verifySubQueueXML(Element qElem, String q,
float parentAbsCapacity, float parentAbsMaxCapacity)
throws Exception {
NodeList queues = qElem.getElementsByTagName("subQueues");
QueueInfo qi = (queues != null) ? new QueueInfo() : new LeafQueueInfo();
@ -258,14 +259,15 @@ public void verifySubQueueXML(Element qElem, String q, float parentAbsCapacity)
WebServicesTestUtils.getXmlString(qElem, "usedResources");
qi.queueName = WebServicesTestUtils.getXmlString(qElem, "queueName");
qi.state = WebServicesTestUtils.getXmlString(qElem, "state");
verifySubQueueGeneric(q, qi, parentAbsCapacity);
verifySubQueueGeneric(q, qi, parentAbsCapacity, parentAbsMaxCapacity);
if (queues != null) {
for (int j = 0; j < queues.getLength(); j++) {
Element subqElem = (Element) queues.item(j);
String qName = WebServicesTestUtils.getXmlString(subqElem, "queueName");
String q2 = q + "." + qName;
verifySubQueueXML(subqElem, q2, qi.absoluteCapacity);
verifySubQueueXML(subqElem, q2,
qi.absoluteCapacity, qi.absoluteMaxCapacity);
}
} else {
LeafQueueInfo lqi = (LeafQueueInfo) qi;
@ -309,7 +311,7 @@ private void verifyClusterScheduler(JSONObject json) throws JSONException,
for (int i = 0; i < arr.length(); i++) {
JSONObject obj = arr.getJSONObject(i);
String q = CapacitySchedulerConfiguration.ROOT + "." + obj.getString("queueName");
verifySubQueue(obj, q, 100);
verifySubQueue(obj, q, 100, 100);
}
}
@ -323,7 +325,8 @@ private void verifyClusterSchedulerGeneric(String type, float usedCapacity,
assertTrue("queueName doesn't match", "root".matches(queueName));
}
private void verifySubQueue(JSONObject info, String q, float parentAbsCapacity)
private void verifySubQueue(JSONObject info, String q,
float parentAbsCapacity, float parentAbsMaxCapacity)
throws JSONException, Exception {
int numExpectedElements = 11;
boolean isParentQueue = true;
@ -345,7 +348,7 @@ private void verifySubQueue(JSONObject info, String q, float parentAbsCapacity)
qi.queueName = info.getString("queueName");
qi.state = info.getString("state");
verifySubQueueGeneric(q, qi, parentAbsCapacity);
verifySubQueueGeneric(q, qi, parentAbsCapacity, parentAbsMaxCapacity);
if (isParentQueue) {
JSONArray arr = info.getJSONArray("subQueues");
@ -353,7 +356,7 @@ private void verifySubQueue(JSONObject info, String q, float parentAbsCapacity)
for (int i = 0; i < arr.length(); i++) {
JSONObject obj = arr.getJSONObject(i);
String q2 = q + "." + obj.getString("queueName");
verifySubQueue(obj, q2, qi.absoluteCapacity);
verifySubQueue(obj, q2, qi.absoluteCapacity, qi.absoluteMaxCapacity);
}
} else {
LeafQueueInfo lqi = (LeafQueueInfo) qi;
@ -371,7 +374,7 @@ private void verifySubQueue(JSONObject info, String q, float parentAbsCapacity)
}
private void verifySubQueueGeneric(String q, QueueInfo info,
float parentAbsCapacity) throws Exception {
float parentAbsCapacity, float parentAbsMaxCapacity) throws Exception {
String[] qArr = q.split("\\.");
assertTrue("q name invalid: " + q, qArr.length > 1);
String qshortName = qArr[qArr.length - 1];
@ -380,7 +383,7 @@ private void verifySubQueueGeneric(String q, QueueInfo info,
assertEquals("capacity doesn't match", csConf.getCapacity(q),
info.capacity, 1e-3f);
float expectCapacity = csConf.getMaximumCapacity(q);
float expectAbsMaxCapacity = parentAbsCapacity * (info.maxCapacity/100);
float expectAbsMaxCapacity = parentAbsMaxCapacity * (info.maxCapacity/100);
if (CapacitySchedulerConfiguration.UNDEFINED == expectCapacity) {
expectCapacity = 100;
expectAbsMaxCapacity = 100;