ProportionalCapacityPreemptionPolicy implements SchedulingEditPolicy { @Override public void editSchedule() { CSQueue root = scheduler.getRootQueue(); Resource clusterResources = Resources.clone(scheduler.getClusterResource()); clusterResources = getNonLabeledResources(clusterResources); setNodeLabels(scheduler.getRMContext().getNodeLabelManager() .getNodeLabels()); containerBasedPreemptOrKill(root, clusterResources); } ProportionalCapacityPreemptionPolicy();  ProportionalCapacityPreemptionPolicy(Configuration config,
      RMContext context, CapacityScheduler scheduler);  ProportionalCapacityPreemptionPolicy(Configuration config,
      RMContext context, CapacityScheduler scheduler, Clock clock); void init(Configuration config, RMContext context,
      PreemptableResourceScheduler sched); @VisibleForTesting ResourceCalculator getResourceCalculator(); @Override void editSchedule(); void setNodeLabels(Map<NodeId, Set<String>> nodelabels); @Override long getMonitoringInterval(); @Override String getPolicyName(); static final String OBSERVE_ONLY; static final String MONITORING_INTERVAL; static final String WAIT_TIME_BEFORE_KILL; static final String TOTAL_PREEMPTION_PER_ROUND; static final String MAX_IGNORED_OVER_CAPACITY; static final String NATURAL_TERMINATION_FACTOR; }
@Test public void testIgnore() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 0, 60, 40 }, { 0, 0, 0, 0 }, { 0, 0, 0, 0 }, { 3, 1, 1, 1 }, { -1, 1, 1, 1 }, { 3, 0, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(isA(ContainerPreemptEvent.class)); }
@Test public void testProportionalPreemption() { int[][] qData = new int[][]{ { 100, 10, 40, 20, 30 }, { 100, 100, 100, 100, 100 }, { 100, 30, 60, 10, 0 }, { 45, 20, 5, 20, 0 }, { 0, 0, 0, 0, 0 }, { 3, 1, 1, 1, 0 }, { -1, 1, 1, 1, 1 }, { 4, 0, 0, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(16)).handle(argThat(new IsPreemptionRequestFor(appA))); }
@Test public void testMaxCap() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 45, 100 }, { 100, 55, 45, 0 }, { 20, 10, 10, 0 }, { 0, 0, 0, 0 }, { 2, 1, 1, 0 }, { -1, 1, 1, 0 }, { 3, 0, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appA))); }
@Test public void testPreemptCycle() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 0, 60, 40 }, { 10, 10, 0, 0 }, { 0, 0, 0, 0 }, { 3, 1, 1, 1 }, { -1, 1, 1, 1 }, { 3, 0, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(10)).handle(argThat(new IsPreemptionRequestFor(appC))); }
@Test public void testExpireKill() { final long killTime = 10000L; int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 0, 60, 40 }, { 10, 10, 0, 0 }, { 0, 0, 0, 0 }, { 3, 1, 1, 1 }, { -1, 1, 1, 1 }, { 3, 0, 0, 0 }, }; conf.setLong(WAIT_TIME_BEFORE_KILL, killTime); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); when(mClock.getTime()).thenReturn(0L); policy.editSchedule(); verify(mDisp, times(10)).handle(argThat(new IsPreemptionRequestFor(appC))); when(mClock.getTime()).thenReturn(killTime / 2); policy.editSchedule(); verify(mDisp, times(20)).handle(argThat(new IsPreemptionRequestFor(appC))); when(mClock.getTime()).thenReturn(killTime + 1); policy.editSchedule(); verify(mDisp, times(30)).handle(evtCaptor.capture()); List<ContainerPreemptEvent> events = evtCaptor.getAllValues(); for (ContainerPreemptEvent e : events.subList(20, 30)) { assertEquals(appC, e.getAppId()); assertEquals(KILL_CONTAINER, e.getType()); } }
@Test public void testDeadzone() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 39, 43, 21 }, { 10, 10, 0, 0 }, { 0, 0, 0, 0 }, { 3, 1, 1, 1 }, { -1, 1, 1, 1 }, { 3, 0, 0, 0 }, }; conf.setFloat(MAX_IGNORED_OVER_CAPACITY, (float) 0.1); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(isA(ContainerPreemptEvent.class)); }
@Test public void testPerQueueDisablePreemption() { int[][] qData = new int[][]{ { 100, 55, 25, 20 }, { 100, 100, 100, 100 }, { 100, 0, 54, 46 }, { 10, 10, 0, 0 }, { 0, 0, 0, 0 }, { 3, 1, 1, 1 }, { -1, 1, 1, 1 }, { 3, 0, 0, 0 }, }; schedConf.setPreemptionDisabled("root.queueB", true); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(10)).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appB))); setup(); schedConf.setPreemptionDisabled("root.queueB", false); ProportionalCapacityPreemptionPolicy policy2 = buildPolicy(qData); policy2.editSchedule(); verify(mDisp, times(4)).handle(argThat(new IsPreemptionRequestFor(appB))); verify(mDisp, times(6)).handle(argThat(new IsPreemptionRequestFor(appC))); }
@Test public void testPerQueueDisablePreemptionHierarchical() { int[][] qData = new int[][] { { 200, 100, 50, 50, 100, 10, 90 }, { 200, 200, 200, 200, 200, 200, 200 }, { 200, 110, 60, 50, 90, 90, 0 }, { 10, 0, 0, 0, 10, 0, 10 }, { 0, 0, 0, 0, 0, 0, 0 }, { 4, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1 }, { 2, 2, 0, 0, 2, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); ApplicationAttemptId expectedAttemptOnQueueB = ApplicationAttemptId.newInstance( appA.getApplicationId(), appA.getAttemptId()); assertTrue("appA should be running on queueB", mCS.getAppsInQueue("queueB").contains(expectedAttemptOnQueueB)); verify(mDisp, times(9)).handle(argThat(new IsPreemptionRequestFor(appA))); setup(); schedConf.setPreemptionDisabled("root.queueA.queueB", true); ProportionalCapacityPreemptionPolicy policy2 = buildPolicy(qData); policy2.editSchedule(); ApplicationAttemptId expectedAttemptOnQueueC = ApplicationAttemptId.newInstance( appB.getApplicationId(), appB.getAttemptId()); ApplicationAttemptId expectedAttemptOnQueueE = ApplicationAttemptId.newInstance( appC.getApplicationId(), appC.getAttemptId()); assertTrue("appB should be running on queueC", mCS.getAppsInQueue("queueC").contains(expectedAttemptOnQueueC)); assertTrue("appC should be running on queueE", mCS.getAppsInQueue("queueE").contains(expectedAttemptOnQueueE)); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appA))); verify(mDisp, times(9)).handle(argThat(new IsPreemptionRequestFor(appC))); }
@Test public void testPerQueueDisablePreemptionBroadHierarchical() { int[][] qData = new int[][] { {1000, 350, 150, 200, 400, 200, 200, 250, 100, 150 }, {1000,1000,1000,1000,1000,1000,1000,1000,1000,1000 }, {1000, 400, 200, 200, 400, 250, 150, 200, 150, 50 }, { 50, 0, 0, 0, 50, 0, 50, 0, 0, 0 }, { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }, { 6, 2, 1, 1, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1, -1, 1, 1 }, { 3, 2, 0, 0, 2, 0, 0, 2, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(28)).handle(argThat(new IsPreemptionRequestFor(appA))); verify(mDisp, times(22)).handle(argThat(new IsPreemptionRequestFor(appC))); setup(); schedConf.setPreemptionDisabled("root.queueA.queueB", true); ProportionalCapacityPreemptionPolicy policy2 = buildPolicy(qData); policy2.editSchedule(); verify(mDisp, times(50)).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appA))); setup(); schedConf.setPreemptionDisabled("root.queueD.queueE", true); schedConf.setPreemptionDisabled("root.queueA.queueB", true); ProportionalCapacityPreemptionPolicy policy3 = buildPolicy(qData); policy3.editSchedule(); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appA))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appB))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appE))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appF))); }
@Test public void testPerQueueDisablePreemptionInheritParent() { int[][] qData = new int[][] { {1000, 500, 200, 200, 100, 500, 200, 200, 100 }, {1000,1000,1000,1000,1000,1000,1000,1000,1000 }, {1000, 700, 0, 350, 350, 300, 0, 200, 100 }, { 200, 0, 0, 0, 0, 200, 200, 0, 0 }, { 0, 0, 0, 0, 0, 0, 0, 0, 0 }, { 5, 2, 0, 1, 1, 3, 1, 1, 1 }, { -1, -1, 1, 1, 1, -1, 1, 1, 1 }, { 2, 3, 0, 0, 0, 3, 0, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(16)).handle(argThat(new IsPreemptionRequestFor(appA))); verify(mDisp, times(182)).handle(argThat(new IsPreemptionRequestFor(appB))); setup(); schedConf.setPreemptionDisabled("root.queueA", true); ProportionalCapacityPreemptionPolicy policy2 = buildPolicy(qData); policy2.editSchedule(); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appA))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appB))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appD))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appE))); }
@Test public void testPerQueuePreemptionNotAllUntouchable() { int[][] qData = new int[][] { { 2000, 1000, 800, 100, 100, 1000, 500, 300, 200 }, { 2000, 2000, 2000, 2000, 2000, 2000, 2000, 2000, 2000 }, { 2000, 1300, 300, 800, 200, 700, 500, 0, 200 }, { 300, 0, 0, 0, 0, 300, 0, 300, 0 }, { 0, 0, 0, 0, 0, 0, 0, 0, 0 }, { 6, 3, 1, 1, 1, 3, 1, 1, 1 }, { -1, -1, 1, 1, 1, -1, 1, 1, 1 }, { 2, 3, 0, 0, 0, 3, 0, 0, 0 }, }; schedConf.setPreemptionDisabled("root.queueA.queueC", true); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(100)).handle(argThat(new IsPreemptionRequestFor(appC))); }
@Test public void testPerQueueDisablePreemptionRootDisablesAll() { int[][] qData = new int[][] { {1000, 500, 250, 250, 250, 100, 150, 250, 100, 150 }, {1000,1000,1000,1000,1000,1000,1000,1000,1000,1000 }, {1000, 20, 0, 20, 490, 240, 250, 490, 240, 250 }, { 200, 200, 200, 0, 0, 0, 0, 0, 0, 0 }, { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }, { 6, 2, 1, 1, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1, -1, 1, 1 }, { 3, 2, 0, 0, 2, 0, 0, 2, 0, 0 }, }; schedConf.setPreemptionDisabled("root", true); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appB))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appD))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appE))); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appF))); }
@Test public void testPerQueueDisablePreemptionOverAbsMaxCapacity() { int[][] qData = new int[][] { {1000, 725, 360, 365, 275, 17, 258 }, {1000,1000,1000,1000, 550, 109,1000 }, {1000, 741, 396, 345, 259, 110, 149 }, { 40, 20, 0, 20, 20, 20, 0 }, { 0, 0, 0, 0, 0, 0, 0 }, { 4, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1 }, { 2, 2, 0, 0, 2, 0, 0 }, }; schedConf.setPreemptionDisabled("root.queueD", true); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appC))); }
@Test public void testOverCapacityImbalance() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 55, 45, 0 }, { 20, 10, 10, 0 }, { 0, 0, 0, 0 }, { 2, 1, 1, 0 }, { -1, 1, 1, 0 }, { 3, 0, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(5)).handle(argThat(new IsPreemptionRequestFor(appA))); }
@Test public void testNaturalTermination() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 55, 45, 0 }, { 20, 10, 10, 0 }, { 0, 0, 0, 0 }, { 2, 1, 1, 0 }, { -1, 1, 1, 0 }, { 3, 0, 0, 0 }, }; conf.setFloat(NATURAL_TERMINATION_FACTOR, (float) 0.1); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(isA(ContainerPreemptEvent.class)); }
@Test public void testObserveOnly() { int[][] qData = new int[][]{ { 100, 40, 40, 20 }, { 100, 100, 100, 100 }, { 100, 90, 10, 0 }, { 80, 10, 20, 50 }, { 0, 0, 0, 0 }, { 2, 1, 1, 0 }, { -1, 1, 1, 0 }, { 3, 0, 0, 0 }, }; conf.setBoolean(OBSERVE_ONLY, true); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(isA(ContainerPreemptEvent.class)); }
@Test public void testHierarchical() { int[][] qData = new int[][] { { 200, 100, 50, 50, 100, 10, 90 }, { 200, 200, 200, 200, 200, 200, 200 }, { 200, 110, 60, 50, 90, 90, 0 }, { 10, 0, 0, 0, 10, 0, 10 }, { 0, 0, 0, 0, 0, 0, 0 }, { 4, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1 }, { 2, 2, 0, 0, 2, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(9)).handle(argThat(new IsPreemptionRequestFor(appA))); }
@Test public void testZeroGuar() { int[][] qData = new int[][] { { 200, 100, 0, 99, 100, 10, 90 }, { 200, 200, 200, 200, 200, 200, 200 }, { 170, 80, 60, 20, 90, 90, 0 }, { 10, 0, 0, 0, 10, 0, 10 }, { 0, 0, 0, 0, 0, 0, 0 }, { 4, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1 }, { 2, 2, 0, 0, 2, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, never()).handle(argThat(new IsPreemptionRequestFor(appA))); }
@Test public void testZeroGuarOverCap() { int[][] qData = new int[][] { { 200, 100, 0, 99, 0, 100, 100 }, { 200, 200, 200, 200, 200, 200, 200 }, { 170, 170, 60, 20, 90, 0, 0 }, { 85, 50, 30, 10, 10, 20, 20 }, { 0, 0, 0, 0, 0, 0, 0 }, { 4, 3, 1, 1, 1, 1, 1 }, { -1, -1, 1, 1, 1, -1, 1 }, { 2, 3, 0, 0, 0, 1, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(14)).handle(argThat(new IsPreemptionRequestFor(appC))); }
@Test public void testHierarchicalLarge() { int[][] qData = new int[][] { { 400, 200, 60, 140, 100, 70, 30, 100, 10, 90 }, { 400, 400, 400, 400, 400, 400, 400, 400, 400, 400 }, { 400, 210, 70, 140, 100, 50, 50, 90, 90, 0 }, { 15, 0, 0, 0, 0, 0, 0, 0, 0, 15 }, { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 }, { 6, 2, 1, 1, 2, 1, 1, 2, 1, 1 }, { -1, -1, 1, 1, -1, 1, 1, -1, 1, 1 }, { 3, 2, 0, 0, 2, 0, 0, 2, 0, 0 }, }; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(7)).handle(argThat(new IsPreemptionRequestFor(appA))); verify(mDisp, times(5)).handle(argThat(new IsPreemptionRequestFor(appE))); }
@Test public void testSkipAMContainer() { int[][] qData = new int[][] { { 100, 50, 50 }, { 100, 100, 100 }, { 100, 100, 0 }, { 70, 20, 50 }, { 0, 0, 0 }, { 5, 4, 1 }, { -1, 1, 1 }, { 2, 0, 0 }, }; setAMContainer = true; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(24)).handle(argThat(new IsPreemptionRequestFor(appD))); verify(mDisp, times(24)).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, times(2)).handle(argThat(new IsPreemptionRequestFor(appB))); setAMContainer = false; }
@Test public void testIdealAllocationForLabels() { int[][] qData = new int[][] { { 80, 40, 40 }, { 80, 80, 80 }, { 80, 80, 0 }, { 70, 20, 50 }, { 0, 0, 0 }, { 5, 4, 1 }, { -1, 1, 1 }, { 2, 0, 0 }, }; setAMContainer = true; setLabeledContainer = true; Map<NodeId, Set<String>> labels = new HashMap<NodeId, Set<String>>(); NodeId node = NodeId.newInstance("node1", 0); Set<String> labelSet = new HashSet<String>(); labelSet.add("x"); labels.put(node, labelSet); when(lm.getNodeLabels()).thenReturn(labels); ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); when(lm.getResourceByLabel(anyString(), any(Resource.class))).thenReturn( Resources.clone(Resource.newInstance(80, 0))); clusterResources.setMemory(100); policy.editSchedule(); verify(mDisp, times(19)).handle(argThat(new IsPreemptionRequestFor(appD))); verify(mDisp, times(19)).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, times(2)).handle(argThat(new IsPreemptionRequestFor(appB))); setAMContainer = false; setLabeledContainer = false; }
@Test public void testPreemptSkippedAMContainers() { int[][] qData = new int[][] { { 100, 10, 90 }, { 100, 100, 100 }, { 100, 100, 0 }, { 70, 20, 90 }, { 0, 0, 0 }, { 5, 4, 1 }, { -1, 5, 5 }, { 2, 0, 0 }, }; setAMContainer = true; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(5)).handle(argThat(new IsPreemptionRequestFor(appD))); verify(mDisp, times(5)).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, times(4)).handle(argThat(new IsPreemptionRequestFor(appB))); verify(mDisp, times(4)).handle(argThat(new IsPreemptionRequestFor(appA))); setAMContainer = false; }
@Test public void testAMResourcePercentForSkippedAMContainers() { int[][] qData = new int[][] { { 100, 10, 90 }, { 100, 100, 100 }, { 100, 100, 0 }, { 70, 20, 90 }, { 0, 0, 0 }, { 5, 4, 1 }, { -1, 5, 5 }, { 2, 0, 0 }, }; setAMContainer = true; setAMResourcePercent = 0.5f; ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData); policy.editSchedule(); verify(mDisp, times(5)).handle(argThat(new IsPreemptionRequestFor(appD))); verify(mDisp, times(5)).handle(argThat(new IsPreemptionRequestFor(appC))); verify(mDisp, times(4)).handle(argThat(new IsPreemptionRequestFor(appB))); verify(mDisp, times(4)).handle(argThat(new IsPreemptionRequestFor(appA))); setAMContainer = false; }