riMap = new HashMap<>();
-
@Before
public void setup() {
-
- // Initialize mandatory resources
- ResourceInformation memory = ResourceInformation.newInstance(
- ResourceInformation.MEMORY_MB.getName(),
- ResourceInformation.MEMORY_MB.getUnits(),
- YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_MB);
- ResourceInformation vcores = ResourceInformation.newInstance(
- ResourceInformation.VCORES.getName(),
- ResourceInformation.VCORES.getUnits(),
- YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES,
- YarnConfiguration.DEFAULT_RM_SCHEDULER_MAXIMUM_ALLOCATION_VCORES);
- riMap.put(ResourceInformation.MEMORY_URI, memory);
- riMap.put(ResourceInformation.VCORES_URI, vcores);
-
- ResourceUtils.initializeResourcesFromResourceInformationMap(riMap);
-
+ rc = new DefaultResourceCalculator();
super.setup();
policy = new ProportionalCapacityPreemptionPolicy(rmContext, cs, mClock);
}
@@ -340,8 +321,8 @@ public class TestPreemptionForQueueWithPriorities
* - a2 (capacity=60), p=1
* - b (capacity=30), p=1
* - b1 (capacity=50), p=1
- * - b1 (capacity=50), p=2
- * - c (capacity=40), p=2
+ * - b2 (capacity=50), p=2
+ * - c (capacity=40), p=1
*
*/
String labelsConfig = "=100,true"; // default partition
@@ -349,11 +330,11 @@ public class TestPreemptionForQueueWithPriorities
String queuesConfig =
// guaranteed,max,used,pending
"root(=[100 100 100 100]);" + //root
- "-a(=[30 100 40 50]){priority=1};" + // a
+ "-a(=[29 100 40 50]){priority=1};" + // a
"--a1(=[12 100 20 50]){priority=1};" + // a1
- "--a2(=[18 100 20 50]){priority=1};" + // a2
- "-b(=[30 100 59 50]){priority=1};" + // b
- "--b1(=[15 100 30 50]){priority=1};" + // b1
+ "--a2(=[17 100 20 50]){priority=1};" + // a2
+ "-b(=[31 100 59 50]){priority=1};" + // b
+ "--b1(=[16 100 30 50]){priority=1};" + // b1
"--b2(=[15 100 29 50]){priority=2};" + // b2
"-c(=[40 100 1 30]){priority=1}"; // c
String appsConfig =
@@ -362,7 +343,7 @@ public class TestPreemptionForQueueWithPriorities
"a2\t(1,1,n1,,20,false);" + // app2 in a2
"b1\t(1,1,n1,,30,false);" + // app3 in b1
"b2\t(1,1,n1,,29,false);" + // app4 in b2
- "c\t(1,1,n1,,29,false)"; // app5 in c
+ "c\t(1,1,n1,,1,false)"; // app5 in c
buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
@@ -370,16 +351,16 @@ public class TestPreemptionForQueueWithPriorities
// Preemption should first divide capacities between a / b, and b2 should
// get less preemption than b1 (because b2 has higher priority)
- verify(mDisp, times(5)).handle(argThat(
+ verify(mDisp, times(6)).handle(argThat(
new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
getAppAttemptId(1))));
- verify(mDisp, never()).handle(argThat(
+ verify(mDisp, times(1)).handle(argThat(
new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
getAppAttemptId(2))));
- verify(mDisp, times(15)).handle(argThat(
+ verify(mDisp, times(13)).handle(argThat(
new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
getAppAttemptId(3))));
- verify(mDisp, times(9)).handle(argThat(
+ verify(mDisp, times(10)).handle(argThat(
new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
getAppAttemptId(4))));
}
@@ -426,7 +407,7 @@ public class TestPreemptionForQueueWithPriorities
// Preemption should first divide capacities between a / b, and b1 should
// get less preemption than b2 (because b1 has higher priority)
- verify(mDisp, never()).handle(argThat(
+ verify(mDisp, times(3)).handle(argThat(
new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
getAppAttemptId(1))));
verify(mDisp, never()).handle(argThat(
@@ -505,4 +486,56 @@ public class TestPreemptionForQueueWithPriorities
getAppAttemptId(3))));
}
+ @Test
+ public void test3ResourceTypesInterQueuePreemption() throws IOException {
+ rc = new DominantResourceCalculator();
+ when(cs.getResourceCalculator()).thenReturn(rc);
+
+ // Initialize resource map
+ String RESOURCE_1 = "res1";
+ riMap.put(RESOURCE_1, ResourceInformation.newInstance(RESOURCE_1, "", 0,
+ ResourceTypes.COUNTABLE, 0, Integer.MAX_VALUE));
+
+ ResourceUtils.initializeResourcesFromResourceInformationMap(riMap);
+
+ /**
+ * Queue structure is:
+ *
+ *
+ * root
+ * / \ \
+ * a b c
+ *
+ * A / B / C have 33.3 / 33.3 / 33.4 resources
+ * Total cluster resource have mem=30, cpu=18, GPU=6
+ * A uses mem=6, cpu=3, GPU=3
+ * B uses mem=6, cpu=3, GPU=3
+ * C is asking mem=1,cpu=1,GPU=1
+ *
+ * We expect it can preempt from one of the jobs
+ */
+ String labelsConfig =
+ "=30:18:6,true;";
+ String nodesConfig =
+ "n1= res=30:18:6;"; // n1 is default partition
+ String queuesConfig =
+ // guaranteed,max,used,pending
+ "root(=[30:18:6 30:18:6 12:12:6 1:1:1]){priority=1};" + //root
+ "-a(=[10:6:2 10:6:2 6:6:3 0:0:0]){priority=1};" + // a
+ "-b(=[10:6:2 10:6:2 6:6:3 0:0:0]){priority=1};" + // b
+ "-c(=[10:6:2 10:6:2 0:0:0 1:1:1]){priority=2}"; // c
+ String appsConfig=
+ //queueName\t(priority,resource,host,expression,#repeat,reserved)
+ "a\t" // app1 in a1
+ + "(1,2:2:1,n1,,3,false);" +
+ "b\t" // app2 in b2
+ + "(1,2:2:1,n1,,3,false)";
+
+ buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+ policy.editSchedule();
+
+ verify(mDisp, times(1)).handle(argThat(
+ new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+ getAppAttemptId(1))));
+ }
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java
index c8a1f0f70ce..14a3a9ad7d7 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyInterQueueWithDRF.java
@@ -18,11 +18,16 @@
package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
+import org.apache.hadoop.yarn.api.protocolrecords.ResourceTypes;
+import org.apache.hadoop.yarn.api.records.ResourceInformation;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.ResourceUtils;
import org.junit.Before;
import org.junit.Test;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import java.io.IOException;
+
import static org.mockito.Matchers.argThat;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -41,8 +46,7 @@ public class TestProportionalCapacityPreemptionPolicyInterQueueWithDRF
}
@Test
- public void testInterQueuePreemptionWithMultipleResource()
- throws Exception {
+ public void testInterQueuePreemptionWithMultipleResource() throws Exception {
/**
* Queue structure is:
*
@@ -121,4 +125,52 @@ public class TestProportionalCapacityPreemptionPolicyInterQueueWithDRF
new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
getAppAttemptId(1))));
}
-}
+
+ @Test
+ public void test3ResourceTypesInterQueuePreemption() throws IOException {
+ // Initialize resource map
+ String RESOURCE_1 = "res1";
+ riMap.put(RESOURCE_1, ResourceInformation
+ .newInstance(RESOURCE_1, "", 0, ResourceTypes.COUNTABLE, 0,
+ Integer.MAX_VALUE));
+
+ ResourceUtils.initializeResourcesFromResourceInformationMap(riMap);
+
+ /*
+ * root
+ * / \ \
+ * a b c
+ *
+ * A / B / C have 33.3 / 33.3 / 33.4 resources
+ * Total cluster resource have mem=30, cpu=18, GPU=6
+ * A uses mem=6, cpu=3, GPU=3
+ * B uses mem=6, cpu=3, GPU=3
+ * C is asking mem=1,cpu=1,GPU=1
+ *
+ * We expect it can preempt from one of the jobs
+ */
+ String labelsConfig = "=30:18:6,true;";
+ String nodesConfig = "n1= res=30:18:6;"; // n1 is default partition
+ String queuesConfig =
+ // guaranteed,max,used,pending
+ "root(=[30:18:6 30:18:6 12:12:6 1:1:1]);" + //root
+ "-a(=[10:7:2 10:6:3 6:6:3 0:0:0]);" + // a
+ "-b(=[10:6:2 10:6:3 6:6:3 0:0:0]);" + // b
+ "-c(=[10:5:2 10:6:2 0:0:0 1:1:1])"; // c
+ String appsConfig =
+ //queueName\t(priority,resource,host,expression,#repeat,reserved)
+ "a\t" // app1 in a1
+ + "(1,2:2:1,n1,,3,false);" + "b\t" // app2 in b2
+ + "(1,2:2:1,n1,,3,false)";
+
+ buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+ policy.editSchedule();
+
+ verify(mDisp, times(0)).handle(argThat(
+ new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+ getAppAttemptId(1))));
+ verify(mDisp, times(1)).handle(argThat(
+ new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+ getAppAttemptId(2))));
+ }
+}
\ No newline at end of file