ParentQueue extends AbstractCSQueue { @Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits, SchedulingMode schedulingMode) { if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY && !accessibleToPartition(node.getPartition())) { return CSAssignment.NULL_ASSIGNMENT; } if (!super.hasPendingResourceRequest(node.getPartition(), clusterResource, schedulingMode)) { if (LOG.isDebugEnabled()) { LOG.debug("Skip this queue=" + getQueuePath() + ", because it doesn't need more resource, schedulingMode=" + schedulingMode.name() + " node-partition=" + node.getPartition()); } return CSAssignment.NULL_ASSIGNMENT; } CSAssignment assignment = new CSAssignment(Resources.createResource(0, 0), NodeType.NODE_LOCAL); while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.debug("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, node.getPartition(), resourceLimits, Resources.createResource( getMetrics().getReservedMB(), getMetrics() .getReservedVirtualCores()), schedulingMode)) { break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits, schedulingMode); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { super.allocateResource(clusterResource, assignedToChild.getResource(), node.getPartition(), assignedToChild.isIncreasedAllocation()); Resources .addTo(assignment.getResource(), assignedToChild.getResource()); Resources.addTo(assignment.getAssignmentInformation().getAllocated(), assignedToChild.getAssignmentInformation().getAllocated()); Resources.addTo(assignment.getAssignmentInformation().getReserved(), assignedToChild.getAssignmentInformation().getReserved()); assignment.getAssignmentInformation().incrAllocations( assignedToChild.getAssignmentInformation().getNumAllocations()); assignment.getAssignmentInformation().incrReservations( assignedToChild.getAssignmentInformation().getNumReservations()); assignment .getAssignmentInformation() .getAllocationDetails() .addAll( assignedToChild.getAssignmentInformation().getAllocationDetails()); assignment .getAssignmentInformation() .getReservationDetails() .addAll( assignedToChild.getAssignmentInformation() .getReservationDetails()); assignment.setIncreasedAllocation(assignedToChild .isIncreasedAllocation()); LOG.info("assignedContainer" + " queue=" + getQueueName() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster=" + clusterResource); } else { break; } if (LOG.isDebugEnabled()) { LOG.debug("ParentQ=" + getQueueName() + " assignedSoFarInThisIteration=" + assignment.getResource() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity()); } if (!rootQueue || assignment.getType() == NodeType.OFF_SWITCH) { if (LOG.isDebugEnabled()) { if (rootQueue && assignment.getType() == NodeType.OFF_SWITCH) { LOG.debug("Not assigning more than one off-switch container," + " assignments so far: " + assignment); } } break; } } return assignment; } ParentQueue(CapacitySchedulerContext cs, 
      String queueName, CSQueue parent, CSQueue old); @Override String getQueuePath(); @Override synchronized QueueInfo getQueueInfo( 
      boolean includeChildQueues, boolean recursive); @Override synchronized List<QueueUserACLInfo> getQueueUserAclInfo(
      UserGroupInformation user); String toString(); @Override synchronized void reinitialize(CSQueue newlyParsedQueue,
      Resource clusterResource); @Override void submitApplication(ApplicationId applicationId, String user,
      String queue); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName); @Override void finishApplicationAttempt(FiCaSchedulerApp application,
      String queue); @Override void finishApplication(ApplicationId application, String user); @Override synchronized CSAssignment assignContainers(Resource clusterResource,
      FiCaSchedulerNode node, ResourceLimits resourceLimits,
      SchedulingMode schedulingMode); @Override void decreaseContainer(Resource clusterResource,
      SchedContainerChangeRequest decreaseRequest, FiCaSchedulerApp app); @Override void unreserveIncreasedContainer(Resource clusterResource,
      FiCaSchedulerApp app, FiCaSchedulerNode node, RMContainer rmContainer); @Override void completedContainer(Resource clusterResource,
      FiCaSchedulerApp application, FiCaSchedulerNode node, 
      RMContainer rmContainer, ContainerStatus containerStatus, 
      RMContainerEventType event, CSQueue completedChildQueue,
      boolean sortQueues); @Override synchronized void updateClusterResource(Resource clusterResource,
      ResourceLimits resourceLimits); @Override synchronized List<CSQueue> getChildQueues(); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt attempt, RMContainer rmContainer); @Override ActiveUsersManager getActiveUsersManager(); @Override synchronized void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override void attachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); @Override void detachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); synchronized int getNumApplications();  }
@Test public void testSingleLevelQueues() throws Exception { setupSingleLevelQueues(csConf); Map<String, CSQueue> queues = new HashMap<String, CSQueue>(); CSQueue root = CapacityScheduler.parseQueue(csContext, csConf, null, CapacitySchedulerConfiguration.ROOT, queues, queues, TestUtils.spyHook); final int memoryPerNode = 10; final int coresPerNode = 16; final int numNodes = 2; FiCaSchedulerNode node_0 = TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB); FiCaSchedulerNode node_1 = TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB); final Resource clusterResource = Resources.createResource(numNodes * (memoryPerNode*GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); LeafQueue a = (LeafQueue)queues.get(A); LeafQueue b = (LeafQueue)queues.get(B); a.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage() .incPending(Resources.createResource(1 * GB)); stubQueueAllocation(a, clusterResource, node_0, 0*GB); stubQueueAllocation(b, clusterResource, node_0, 1*GB); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); verifyQueueMetrics(a, 0*GB, clusterResource); verifyQueueMetrics(b, 1*GB, clusterResource); stubQueueAllocation(a, clusterResource, node_1, 2*GB); stubQueueAllocation(b, clusterResource, node_1, 1*GB); root.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); InOrder allocationOrder = inOrder(a, b); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 2*GB, clusterResource); verifyQueueMetrics(b, 2*GB, clusterResource); stubQueueAllocation(a, clusterResource, node_0, 1*GB); stubQueueAllocation(b, clusterResource, node_0, 2*GB); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); allocationOrder = inOrder(b, a); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 3*GB, clusterResource); verifyQueueMetrics(b, 4*GB, clusterResource); stubQueueAllocation(a, clusterResource, node_0, 0*GB); stubQueueAllocation(b, clusterResource, node_0, 4*GB); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); allocationOrder = inOrder(b, a); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 3*GB, clusterResource); verifyQueueMetrics(b, 8*GB, clusterResource); stubQueueAllocation(a, clusterResource, node_1, 1*GB); stubQueueAllocation(b, clusterResource, node_1, 1*GB); root.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); allocationOrder = inOrder(a, b); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 4*GB, clusterResource); verifyQueueMetrics(b, 9*GB, clusterResource); }
@Test public void testMultiLevelQueues() throws Exception { setupMultiLevelQueues(csConf); Map<String, CSQueue> queues = new HashMap<String, CSQueue>(); CSQueue root = CapacityScheduler.parseQueue(csContext, csConf, null, CapacitySchedulerConfiguration.ROOT, queues, queues, TestUtils.spyHook); final int memoryPerNode = 10; final int coresPerNode = 16; final int numNodes = 3; FiCaSchedulerNode node_0 = TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB); FiCaSchedulerNode node_1 = TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB); FiCaSchedulerNode node_2 = TestUtils.getMockNode("host_2", DEFAULT_RACK, 0, memoryPerNode*GB); final Resource clusterResource = Resources.createResource(numNodes * (memoryPerNode*GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); CSQueue a = queues.get(A); a.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue b = queues.get(B); b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue c = queues.get(C); c.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue d = queues.get(D); d.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue a1 = queues.get(A1); a1.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue a2 = queues.get(A2); a2.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue b1 = queues.get(B1); b1.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue b2 = queues.get(B2); b2.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); CSQueue b3 = queues.get(B3); b3.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage() .incPending(Resources.createResource(1 * GB)); stubQueueAllocation(a, clusterResource, node_0, 0*GB); stubQueueAllocation(b, clusterResource, node_0, 0*GB); stubQueueAllocation(c, clusterResource, node_0, 1*GB); stubQueueAllocation(d, clusterResource, node_0, 0*GB); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); verifyQueueMetrics(a, 0*GB, clusterResource); verifyQueueMetrics(b, 0*GB, clusterResource); verifyQueueMetrics(c, 1*GB, clusterResource); verifyQueueMetrics(d, 0*GB, clusterResource); reset(a); reset(b); reset(c); stubQueueAllocation(a, clusterResource, node_1, 0*GB); stubQueueAllocation(b2, clusterResource, node_1, 4*GB); stubQueueAllocation(c, clusterResource, node_1, 0*GB); root.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); verifyQueueMetrics(a, 0*GB, clusterResource); verifyQueueMetrics(b, 4*GB, clusterResource); verifyQueueMetrics(c, 1*GB, clusterResource); reset(a); reset(b); reset(c); stubQueueAllocation(a1, clusterResource, node_0, 1*GB); stubQueueAllocation(b3, clusterResource, node_0, 2*GB); stubQueueAllocation(c, clusterResource, node_0, 2*GB); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); InOrder allocationOrder = inOrder(a, c, b); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(c).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 1*GB, clusterResource); verifyQueueMetrics(b, 6*GB, clusterResource); verifyQueueMetrics(c, 3*GB, clusterResource); reset(a); reset(b); reset(c); LOG.info("here"); ((ParentQueue)a).setMaxCapacity(.1f); stubQueueAllocation(a1, clusterResource, node_2, 1*GB); stubQueueAllocation(a2, clusterResource, node_2, 2*GB); stubQueueAllocation(b3, clusterResource, node_2, 1*GB); stubQueueAllocation(b1, clusterResource, node_2, 1*GB); stubQueueAllocation(c, clusterResource, node_2, 1*GB); root.assignContainers(clusterResource, node_2, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); allocationOrder = inOrder(a, a2, a1, b, c); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(a2).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(c).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 3*GB, clusterResource); verifyQueueMetrics(b, 8*GB, clusterResource); verifyQueueMetrics(c, 4*GB, clusterResource); reset(a); reset(b); reset(c); }
@Test public void testOffSwitchScheduling() throws Exception { setupSingleLevelQueues(csConf); Map<String, CSQueue> queues = new HashMap<String, CSQueue>(); CSQueue root = CapacityScheduler.parseQueue(csContext, csConf, null, CapacitySchedulerConfiguration.ROOT, queues, queues, TestUtils.spyHook); final int memoryPerNode = 10; final int coresPerNode = 16; final int numNodes = 2; FiCaSchedulerNode node_0 = TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB); FiCaSchedulerNode node_1 = TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB); final Resource clusterResource = Resources.createResource(numNodes * (memoryPerNode*GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); LeafQueue a = (LeafQueue)queues.get(A); LeafQueue b = (LeafQueue)queues.get(B); a.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage() .incPending(Resources.createResource(1 * GB)); stubQueueAllocation(a, clusterResource, node_0, 0*GB, NodeType.OFF_SWITCH); stubQueueAllocation(b, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); verifyQueueMetrics(a, 0*GB, clusterResource); verifyQueueMetrics(b, 1*GB, clusterResource); stubQueueAllocation(a, clusterResource, node_1, 2*GB, NodeType.RACK_LOCAL); stubQueueAllocation(b, clusterResource, node_1, 1*GB, NodeType.OFF_SWITCH); root.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); InOrder allocationOrder = inOrder(a, b); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 2*GB, clusterResource); verifyQueueMetrics(b, 2*GB, clusterResource); stubQueueAllocation(a, clusterResource, node_0, 1*GB, NodeType.NODE_LOCAL); stubQueueAllocation(b, clusterResource, node_0, 2*GB, NodeType.OFF_SWITCH); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); allocationOrder = inOrder(b, a); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(a, 2*GB, clusterResource); verifyQueueMetrics(b, 4*GB, clusterResource); }
@Test public void testOffSwitchSchedulingMultiLevelQueues() throws Exception { setupMultiLevelQueues(csConf); Map<String, CSQueue> queues = new HashMap<String, CSQueue>(); CSQueue root = CapacityScheduler.parseQueue(csContext, csConf, null, CapacitySchedulerConfiguration.ROOT, queues, queues, TestUtils.spyHook); final int memoryPerNode = 10; final int coresPerNode = 10; final int numNodes = 2; FiCaSchedulerNode node_0 = TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB); FiCaSchedulerNode node_1 = TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB); final Resource clusterResource = Resources.createResource(numNodes * (memoryPerNode*GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); LeafQueue b3 = (LeafQueue)queues.get(B3); LeafQueue b2 = (LeafQueue)queues.get(B2); b2.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); b3.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage() .incPending(Resources.createResource(1 * GB)); CSQueue b = queues.get(B); b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB)); stubQueueAllocation(b2, clusterResource, node_0, 0*GB, NodeType.OFF_SWITCH); stubQueueAllocation(b3, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); verifyQueueMetrics(b2, 0*GB, clusterResource); verifyQueueMetrics(b3, 1*GB, clusterResource); stubQueueAllocation(b2, clusterResource, node_1, 1*GB, NodeType.RACK_LOCAL); stubQueueAllocation(b3, clusterResource, node_1, 1*GB, NodeType.OFF_SWITCH); root.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); InOrder allocationOrder = inOrder(b2, b3); allocationOrder.verify(b2).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(b3).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(b2, 1*GB, clusterResource); verifyQueueMetrics(b3, 2*GB, clusterResource); stubQueueAllocation(b2, clusterResource, node_0, 1*GB, NodeType.NODE_LOCAL); stubQueueAllocation(b3, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY); allocationOrder = inOrder(b3, b2); allocationOrder.verify(b3).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); allocationOrder.verify(b2).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits(), any(SchedulingMode.class)); verifyQueueMetrics(b2, 1*GB, clusterResource); verifyQueueMetrics(b3, 3*GB, clusterResource); }