@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, numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); LeafQueue a = (LeafQueue)queues.get(A); LeafQueue b = (LeafQueue)queues.get(B); stubQueueAllocation(a, clusterResource, node_0, 0*GB); stubQueueAllocation(b, clusterResource, node_0, 1*GB); root.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource)); 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)); InOrder allocationOrder = inOrder(a, b); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); 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)); allocationOrder = inOrder(b, a); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); 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)); allocationOrder = inOrder(b, a); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); 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)); allocationOrder = inOrder(a, b); allocationOrder.verify(b).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); allocationOrder.verify(a).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); verifyQueueMetrics(a, 4*GB, clusterResource); verifyQueueMetrics(b, 9*GB, clusterResource); }
@Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(0, 0, 0), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.debug("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores(), getMetrics().getReservedGpuCores()))) { break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels); Resources.addTo(assignment.getResource(), assignedToChild.getResource()); 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 extends AbstractCSQueue { @Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(0, 0, 0), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.debug("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores(), getMetrics().getReservedGpuCores()))) { break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels); Resources.addTo(assignment.getResource(), assignedToChild.getResource()); 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 extends AbstractCSQueue { @Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(0, 0, 0), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.debug("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores(), getMetrics().getReservedGpuCores()))) { break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels); Resources.addTo(assignment.getResource(), assignedToChild.getResource()); 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); }
ParentQueue extends AbstractCSQueue { @Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(0, 0, 0), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.debug("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores(), getMetrics().getReservedGpuCores()))) { break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels); Resources.addTo(assignment.getResource(), assignedToChild.getResource()); 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); @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(); }
ParentQueue extends AbstractCSQueue { @Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(0, 0, 0), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.debug("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores(), getMetrics().getReservedGpuCores()))) { break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels); Resources.addTo(assignment.getResource(), assignedToChild.getResource()); 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); @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();  }