@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); 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)); 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)); InOrder allocationOrder = inOrder(b2, b3); allocationOrder.verify(b2).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); allocationOrder.verify(b3).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); 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)); allocationOrder = inOrder(b3, b2); allocationOrder.verify(b3).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); allocationOrder.verify(b2).assignContainers(eq(clusterResource), any(FiCaSchedulerNode.class), anyResourceLimits()); verifyQueueMetrics(b2, 1*GB, clusterResource); verifyQueueMetrics(b3, 3*GB, clusterResource); }
@Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(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.info("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores()))) { LOG.info("over maximum-capacity of this queue "+getQueueName()); break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { if(assignedToChild.getContainersToResume().size() > 0){ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,true); }else{ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,false); } assignment.merge(assignedToChild); LOG.info("assignedContainer" + " queue=" + getQueueName() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster=" + clusterResource+ " resume size="+assignment.getContainersToResume().size()); } else { break; } if (LOG.isDebugEnabled()) { LOG.info("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), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.info("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores()))) { LOG.info("over maximum-capacity of this queue "+getQueueName()); break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { if(assignedToChild.getContainersToResume().size() > 0){ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,true); }else{ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,false); } assignment.merge(assignedToChild); LOG.info("assignedContainer" + " queue=" + getQueueName() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster=" + clusterResource+ " resume size="+assignment.getContainersToResume().size()); } else { break; } if (LOG.isDebugEnabled()) { LOG.info("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), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.info("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores()))) { LOG.info("over maximum-capacity of this queue "+getQueueName()); break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { if(assignedToChild.getContainersToResume().size() > 0){ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,true); }else{ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,false); } assignment.merge(assignedToChild); LOG.info("assignedContainer" + " queue=" + getQueueName() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster=" + clusterResource+ " resume size="+assignment.getContainersToResume().size()); } else { break; } if (LOG.isDebugEnabled()) { LOG.info("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), NodeType.NODE_LOCAL); Set<String> nodeLabels = node.getLabels(); if (!SchedulerUtils.checkQueueAccessToNode(accessibleLabels, nodeLabels)) { return assignment; } while (canAssign(clusterResource, node)) { if (LOG.isDebugEnabled()) { LOG.info("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores()))) { LOG.info("over maximum-capacity of this queue "+getQueueName()); break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { if(assignedToChild.getContainersToResume().size() > 0){ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,true); }else{ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,false); } assignment.merge(assignedToChild); LOG.info("assignedContainer" + " queue=" + getQueueName() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster=" + clusterResource+ " resume size="+assignment.getContainersToResume().size()); } else { break; } if (LOG.isDebugEnabled()) { LOG.info("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(); @Override Resource getPreemptedResource(); @Override void setFastResumption(boolean set); }
ParentQueue extends AbstractCSQueue { @Override public synchronized CSAssignment assignContainers(Resource clusterResource, FiCaSchedulerNode node, ResourceLimits resourceLimits) { CSAssignment assignment = new CSAssignment(Resources.createResource(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.info("Trying to assign containers to child-queue of " + getQueueName()); } if (!super.canAssignToThisQueue(clusterResource, nodeLabels, resourceLimits, minimumAllocation, Resources.createResource(getMetrics() .getReservedMB(), getMetrics().getReservedVirtualCores()))) { LOG.info("over maximum-capacity of this queue "+getQueueName()); break; } CSAssignment assignedToChild = assignContainersToChildQueues(clusterResource, node, resourceLimits); assignment.setType(assignedToChild.getType()); if (Resources.greaterThan( resourceCalculator, clusterResource, assignedToChild.getResource(), Resources.none())) { if(assignedToChild.getContainersToResume().size() > 0){ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,true); }else{ super.allocateResource(clusterResource, assignedToChild.getResource(), nodeLabels,false); } assignment.merge(assignedToChild); LOG.info("assignedContainer" + " queue=" + getQueueName() + " usedCapacity=" + getUsedCapacity() + " absoluteUsedCapacity=" + getAbsoluteUsedCapacity() + " used=" + queueUsage.getUsed() + " cluster=" + clusterResource+ " resume size="+assignment.getContainersToResume().size()); } else { break; } if (LOG.isDebugEnabled()) { LOG.info("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(); @Override Resource getPreemptedResource(); @Override void setFastResumption(boolean set);  }