CSQueueUtils { public static float getAbsoluteMaxAvailCapacity( ResourceCalculator resourceCalculator, Resource clusterResource, CSQueue queue) { CSQueue parent = queue.getParent(); if (parent == null) { return queue.getAbsoluteMaximumCapacity(); } float parentMaxAvail = getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, parent); Resource parentResource = Resources.multiply(clusterResource, parentMaxAvail); if (Resources.isInvalidDivisor(resourceCalculator, parentResource)) { return 0.0f; } float siblingUsedCapacity = Resources.ratio( resourceCalculator, Resources.subtract(parent.getUsedResources(), queue.getUsedResources()), parentResource); float maxAvail = Math.min( queue.getMaximumCapacity(), 1.0f - siblingUsedCapacity); float absoluteMaxAvail = maxAvail * parentMaxAvail; if (LOG.isDebugEnabled()) { LOG.debug("qpath " + queue.getQueuePath()); LOG.debug("parentMaxAvail " + parentMaxAvail); LOG.debug("siblingUsedCapacity " + siblingUsedCapacity); LOG.debug("getAbsoluteMaximumCapacity " + queue.getAbsoluteMaximumCapacity()); LOG.debug("maxAvail " + maxAvail); LOG.debug("absoluteMaxAvail " + absoluteMaxAvail); } if (absoluteMaxAvail < 0.0f) { absoluteMaxAvail = 0.0f; } else if (absoluteMaxAvail > 1.0f) { absoluteMaxAvail = 1.0f; } return absoluteMaxAvail; }  static void checkMaxCapacity(String queueName, 
      float capacity, float maximumCapacity); static void checkAbsoluteCapacity(String queueName,
      float absCapacity, float absMaxCapacity); static float computeAbsoluteMaximumCapacity(
      float maximumCapacity, CSQueue parent); static void updateAndCheckCapacitiesByLabel(String queuePath,
      QueueCapacities queueCapacities, QueueCapacities parentQueueCapacities); static void loadUpdateAndCheckCapacities(String queuePath,
      CapacitySchedulerConfiguration csConf,
      QueueCapacities queueCapacities, QueueCapacities parentQueueCapacities); @Lock(CSQueue.class) static void updateQueueStatistics(
      final ResourceCalculator calculator,
      final CSQueue childQueue, final CSQueue parentQueue, 
      final Resource clusterResource, final Resource minimumAllocation); static float getAbsoluteMaxAvailCapacity(
      ResourceCalculator resourceCalculator, Resource clusterResource, CSQueue queue);  }
@Test public void testAbsoluteMaxAvailCapacityNoUse() throws Exception { ResourceCalculator resourceCalculator = new DefaultResourceCalculator(); Resource clusterResource = Resources.createResource(100 * 16 * GB, 100 * 32); YarnConfiguration conf = new YarnConfiguration(); CapacitySchedulerConfiguration csConf = new CapacitySchedulerConfiguration(); CapacitySchedulerContext csContext = mock(CapacitySchedulerContext.class); when(csContext.getConf()).thenReturn(conf); when(csContext.getConfiguration()).thenReturn(csConf); when(csContext.getClusterResource()).thenReturn(clusterResource); when(csContext.getResourceCalculator()).thenReturn(resourceCalculator); when(csContext.getMinimumResourceCapability()). thenReturn(Resources.createResource(GB, 1)); when(csContext.getMaximumResourceCapability()). thenReturn(Resources.createResource(16*GB, 32)); RMContext rmContext = TestUtils.getMockRMContext(); when(csContext.getRMContext()).thenReturn(rmContext); final String L1Q1 = "L1Q1"; csConf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {L1Q1}); final String L1Q1P = CapacitySchedulerConfiguration.ROOT + "." + L1Q1; csConf.setCapacity(L1Q1P, 90); csConf.setMaximumCapacity(L1Q1P, 90); ParentQueue root = new ParentQueue(csContext, CapacitySchedulerConfiguration.ROOT, null, null); LeafQueue l1q1 = new LeafQueue(csContext, L1Q1, root, null); LOG.info("t1 root " + CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, root)); LOG.info("t1 l1q1 " + CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l1q1)); assertEquals(1.0f, CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, root), 0.000001f); assertEquals(0.9f, CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l1q1), 0.000001f); }
@Test public void testAbsoluteMaxAvailCapacityWithUse() throws Exception { ResourceCalculator resourceCalculator = new DefaultResourceCalculator(); Resource clusterResource = Resources.createResource(100 * 16 * GB, 100 * 32); YarnConfiguration conf = new YarnConfiguration(); CapacitySchedulerConfiguration csConf = new CapacitySchedulerConfiguration(); CapacitySchedulerContext csContext = mock(CapacitySchedulerContext.class); when(csContext.getConf()).thenReturn(conf); when(csContext.getConfiguration()).thenReturn(csConf); when(csContext.getClusterResource()).thenReturn(clusterResource); when(csContext.getResourceCalculator()).thenReturn(resourceCalculator); when(csContext.getMinimumResourceCapability()). thenReturn(Resources.createResource(GB, 1)); when(csContext.getMaximumResourceCapability()). thenReturn(Resources.createResource(16*GB, 32)); RMContext rmContext = TestUtils.getMockRMContext(); when(csContext.getRMContext()).thenReturn(rmContext); final String L1Q1 = "L1Q1"; final String L1Q2 = "L1Q2"; final String L2Q1 = "L2Q1"; final String L2Q2 = "L2Q2"; csConf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {L1Q1, L1Q2, L2Q1, L2Q2}); final String L1Q1P = CapacitySchedulerConfiguration.ROOT + "." + L1Q1; csConf.setCapacity(L1Q1P, 80); csConf.setMaximumCapacity(L1Q1P, 80); final String L1Q2P = CapacitySchedulerConfiguration.ROOT + "." + L1Q2; csConf.setCapacity(L1Q2P, 20); csConf.setMaximumCapacity(L1Q2P, 100); final String L2Q1P = L1Q1P + "." + L2Q1; csConf.setCapacity(L2Q1P, 50); csConf.setMaximumCapacity(L2Q1P, 50); final String L2Q2P = L1Q1P + "." + L2Q2; csConf.setCapacity(L2Q2P, 50); csConf.setMaximumCapacity(L2Q2P, 50); float result; ParentQueue root = new ParentQueue(csContext, CapacitySchedulerConfiguration.ROOT, null, null); LeafQueue l1q1 = new LeafQueue(csContext, L1Q1, root, null); LeafQueue l1q2 = new LeafQueue(csContext, L1Q2, root, null); LeafQueue l2q2 = new LeafQueue(csContext, L2Q2, l1q1, null); LeafQueue l2q1 = new LeafQueue(csContext, L2Q1, l1q1, null); result = CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l2q2); assertEquals( 0.4f, result, 0.000001f); LOG.info("t2 l2q2 " + result); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); l1q2.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); result = CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l2q2); assertEquals( 0.4f, result, 0.000001f); LOG.info("t2 l2q2 " + result); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.3f)); l1q2.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.3f)); result = CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l2q2); assertEquals( 0.3f, result, 0.000001f); LOG.info("t2 l2q2 " + result); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); l1q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.1f)); result = CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l2q2); assertEquals( 0.3f, result, 0.000001f); LOG.info("t2 l2q2 " + result); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l1q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l2q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); result = CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l2q2); assertEquals( 0.3f, result, 0.000001f); LOG.info("t2 l2q2 " + result); root.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l1q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); l2q1.getQueueResourceUsage().incUsed(Resources.multiply(clusterResource, 0.2f)); result = CSQueueUtils.getAbsoluteMaxAvailCapacity( resourceCalculator, clusterResource, l2q2); assertEquals( 0.1f, result, 0.000001f); LOG.info("t2 l2q2 " + result); }