@Test public void testComputeUserLimitAndSetHeadroom(){ LeafQueue qb = stubLeafQueue((LeafQueue)queues.get(B)); qb.setMaxCapacity(1.0f); final String user_0 = "user_0"; final String user_1 = "user_1"; String host_0 = "127.0.0.1"; FiCaSchedulerNode node_0 = TestUtils.getMockNode(host_0, DEFAULT_RACK, 0, 8*GB); String host_1 = "127.0.0.2"; FiCaSchedulerNode node_1 = TestUtils.getMockNode(host_1, DEFAULT_RACK, 0, 8*GB); final int numNodes = 2; Resource clusterResource = Resources.createResource(numNodes * (8*GB), 1); when(csContext.getNumClusterNodes()).thenReturn(numNodes); qb.setUserLimit(100); qb.setUserLimitFactor(1); final ApplicationAttemptId appAttemptId_0 = TestUtils.getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, qb, qb.getActiveUsersManager(), spyRMContext); qb.submitApplicationAttempt(app_0, user_0); Priority u0Priority = TestUtils.createMockPriority(1); app_0.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 4*GB, 1, true, u0Priority, recordFactory))); assertEquals("There should only be 1 active user!", 1, qb.getActiveUsersManager().getNumActiveUsers()); qb.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource)); qb.computeUserLimitAndSetHeadroom(app_0, clusterResource, app_0 .getResourceRequest(u0Priority, ResourceRequest.ANY).getCapability(), null); assertEquals(9*GB,app_0.getHeadroom().getMemory()); final ApplicationAttemptId appAttemptId_2 = TestUtils.getMockApplicationAttemptId(2, 0); FiCaSchedulerApp app_2 = new FiCaSchedulerApp(appAttemptId_2, user_1, qb, qb.getActiveUsersManager(), spyRMContext); Priority u1Priority = TestUtils.createMockPriority(2); app_2.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 4*GB, 1, true, u1Priority, recordFactory))); qb.submitApplicationAttempt(app_2, user_1); qb.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource)); qb.computeUserLimitAndSetHeadroom(app_0, clusterResource, app_0 .getResourceRequest(u0Priority, ResourceRequest.ANY).getCapability(), null); assertEquals(8*GB, qb.getUsedResources().getMemory()); assertEquals(4*GB, app_0.getCurrentConsumption().getMemory()); assertEquals(8*GB, app_0.getHeadroom().getMemory()); assertEquals(4*GB, app_2.getCurrentConsumption().getMemory()); assertEquals(8*GB, app_2.getHeadroom().getMemory()); qb.finishApplication(app_0.getApplicationId(), user_0); qb.finishApplication(app_2.getApplicationId(), user_1); qb.releaseResource(clusterResource, app_0, app_0.getResource(u0Priority), null); qb.releaseResource(clusterResource, app_2, app_2.getResource(u1Priority), null); qb.setUserLimit(50); qb.setUserLimitFactor(1); final ApplicationAttemptId appAttemptId_1 = TestUtils.getMockApplicationAttemptId(1, 0); FiCaSchedulerApp app_1 = new FiCaSchedulerApp(appAttemptId_1, user_0, qb, qb.getActiveUsersManager(), spyRMContext); final ApplicationAttemptId appAttemptId_3 = TestUtils.getMockApplicationAttemptId(3, 0); FiCaSchedulerApp app_3 = new FiCaSchedulerApp(appAttemptId_3, user_1, qb, qb.getActiveUsersManager(), spyRMContext); app_1.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 2*GB, 1, true, u0Priority, recordFactory))); app_3.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 2*GB, 1, true, u1Priority, recordFactory))); qb.submitApplicationAttempt(app_1, user_0); qb.submitApplicationAttempt(app_3, user_1); qb.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource)); qb.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource)); qb.computeUserLimitAndSetHeadroom(app_3, clusterResource, app_3 .getResourceRequest(u1Priority, ResourceRequest.ANY).getCapability(), null); assertEquals(4*GB, qb.getUsedResources().getMemory()); assertEquals(5*GB, app_3.getHeadroom().getMemory()); assertEquals(5*GB, app_1.getHeadroom().getMemory()); final ApplicationAttemptId appAttemptId_4 = TestUtils.getMockApplicationAttemptId(4, 0); FiCaSchedulerApp app_4 = new FiCaSchedulerApp(appAttemptId_4, user_0, qb, qb.getActiveUsersManager(), spyRMContext); qb.submitApplicationAttempt(app_4, user_0); app_4.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 6*GB, 1, true, u0Priority, recordFactory))); qb.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource)); qb.computeUserLimitAndSetHeadroom(app_4, clusterResource, app_4 .getResourceRequest(u0Priority, ResourceRequest.ANY).getCapability(), null); qb.computeUserLimitAndSetHeadroom(app_3, clusterResource, app_3 .getResourceRequest(u1Priority, ResourceRequest.ANY).getCapability(), null); assertEquals(6*GB, app_3.getHeadroom().getMemory()); assertEquals(10*GB, qb.getUsedResources().getMemory()); assertEquals(5*GB, app_4.getHeadroom().getMemory()); }
@Lock({LeafQueue.class, FiCaSchedulerApp.class}) Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application, Resource clusterResource, Resource required, Set<String> requestedLabels) { String user = application.getUser(); User queueUser = getUser(user); Resource userLimit = computeUserLimit(application, clusterResource, required, queueUser, requestedLabels); setQueueResourceLimitsInfo(clusterResource); Resource headroom = getHeadroom(queueUser, cachedResourceLimitsForHeadroom.getLimit(), clusterResource, userLimit); if (LOG.isDebugEnabled()) { LOG.debug("Headroom calculation for user " + user + ": " + " userLimit=" + userLimit + " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() + " consumed=" + queueUser.getUsed() + " headroom=" + headroom); } CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider( queueUser, this, application, required, queueResourceLimitsInfo); application.setHeadroomProvider(headroomProvider); metrics.setAvailableResourcesToUser(user, headroom); return userLimit; }
LeafQueue extends AbstractCSQueue { @Lock({LeafQueue.class, FiCaSchedulerApp.class}) Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application, Resource clusterResource, Resource required, Set<String> requestedLabels) { String user = application.getUser(); User queueUser = getUser(user); Resource userLimit = computeUserLimit(application, clusterResource, required, queueUser, requestedLabels); setQueueResourceLimitsInfo(clusterResource); Resource headroom = getHeadroom(queueUser, cachedResourceLimitsForHeadroom.getLimit(), clusterResource, userLimit); if (LOG.isDebugEnabled()) { LOG.debug("Headroom calculation for user " + user + ": " + " userLimit=" + userLimit + " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() + " consumed=" + queueUser.getUsed() + " headroom=" + headroom); } CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider( queueUser, this, application, required, queueResourceLimitsInfo); application.setHeadroomProvider(headroomProvider); metrics.setAvailableResourcesToUser(user, headroom); return userLimit; } }
LeafQueue extends AbstractCSQueue { @Lock({LeafQueue.class, FiCaSchedulerApp.class}) Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application, Resource clusterResource, Resource required, Set<String> requestedLabels) { String user = application.getUser(); User queueUser = getUser(user); Resource userLimit = computeUserLimit(application, clusterResource, required, queueUser, requestedLabels); setQueueResourceLimitsInfo(clusterResource); Resource headroom = getHeadroom(queueUser, cachedResourceLimitsForHeadroom.getLimit(), clusterResource, userLimit); if (LOG.isDebugEnabled()) { LOG.debug("Headroom calculation for user " + user + ": " + " userLimit=" + userLimit + " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() + " consumed=" + queueUser.getUsed() + " headroom=" + headroom); } CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider( queueUser, this, application, required, queueResourceLimitsInfo); application.setHeadroomProvider(headroomProvider); metrics.setAvailableResourcesToUser(user, headroom); return userLimit; } LeafQueue(CapacitySchedulerContext cs, 
      String queueName, CSQueue parent, CSQueue old); }
LeafQueue extends AbstractCSQueue { @Lock({LeafQueue.class, FiCaSchedulerApp.class}) Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application, Resource clusterResource, Resource required, Set<String> requestedLabels) { String user = application.getUser(); User queueUser = getUser(user); Resource userLimit = computeUserLimit(application, clusterResource, required, queueUser, requestedLabels); setQueueResourceLimitsInfo(clusterResource); Resource headroom = getHeadroom(queueUser, cachedResourceLimitsForHeadroom.getLimit(), clusterResource, userLimit); if (LOG.isDebugEnabled()) { LOG.debug("Headroom calculation for user " + user + ": " + " userLimit=" + userLimit + " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() + " consumed=" + queueUser.getUsed() + " headroom=" + headroom); } CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider( queueUser, this, application, required, queueResourceLimitsInfo); application.setHeadroomProvider(headroomProvider); metrics.setAvailableResourcesToUser(user, headroom); return userLimit; } LeafQueue(CapacitySchedulerContext cs, 
      String queueName, CSQueue parent, CSQueue old); @Override String getQueuePath(); @Private float getMinimumAllocationFactor(); @Private float getMaxAMResourcePerQueuePercent(); int getMaxApplications(); synchronized int getMaxApplicationsPerUser(); @Override ActiveUsersManager getActiveUsersManager(); @Override List<CSQueue> getChildQueues(); @Override synchronized int getNumApplications(); synchronized int getNumPendingApplications(); synchronized int getNumActiveApplications(); @Private synchronized int getNumApplications(String user); @Private synchronized int getNumPendingApplications(String user); @Private synchronized int getNumActiveApplications(String user); synchronized int getNumContainers(); @Override synchronized QueueState getState(); @Private synchronized int getUserLimit(); @Private synchronized float getUserLimitFactor(); @Override synchronized QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive); @Override synchronized List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); @Private int getNodeLocalityDelay(); String toString(); @VisibleForTesting synchronized void setNodeLabelManager(RMNodeLabelsManager mgr); @VisibleForTesting synchronized User getUser(String userName); synchronized ArrayList<UserInfo> getUsers(); @Override synchronized void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName); @Override void submitApplication(ApplicationId applicationId, String userName,
      String queue); synchronized Resource getAMResourceLimit(); synchronized Resource getUserAMResourceLimit(); synchronized Resource getUserAMResourceLimitPerPartition(
      String nodePartition); synchronized Resource getAMResourceLimitPerPartition(
      String nodePartition); @Override void finishApplication(ApplicationId application, String user); @Override void finishApplicationAttempt(FiCaSchedulerApp application, String queue); synchronized void removeApplicationAttempt(
      FiCaSchedulerApp application, User user); @Override synchronized CSAssignment assignContainers(Resource clusterResource,
      FiCaSchedulerNode node, ResourceLimits currentResourceLimits); @Override void completedContainer(Resource clusterResource, 
      FiCaSchedulerApp application, FiCaSchedulerNode node, RMContainer rmContainer, 
      ContainerStatus containerStatus, RMContainerEventType event, CSQueue childQueue,
      boolean sortQueues); @Override synchronized void updateClusterResource(Resource clusterResource,
      ResourceLimits currentResourceLimits); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt attempt, RMContainer rmContainer); Set<FiCaSchedulerApp> getApplications(); synchronized Resource getTotalResourcePending(); @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); void setCapacity(float capacity); void setAbsoluteCapacity(float absoluteCapacity); void setMaxApplications(int maxApplications); }
LeafQueue extends AbstractCSQueue { @Lock({LeafQueue.class, FiCaSchedulerApp.class}) Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application, Resource clusterResource, Resource required, Set<String> requestedLabels) { String user = application.getUser(); User queueUser = getUser(user); Resource userLimit = computeUserLimit(application, clusterResource, required, queueUser, requestedLabels); setQueueResourceLimitsInfo(clusterResource); Resource headroom = getHeadroom(queueUser, cachedResourceLimitsForHeadroom.getLimit(), clusterResource, userLimit); if (LOG.isDebugEnabled()) { LOG.debug("Headroom calculation for user " + user + ": " + " userLimit=" + userLimit + " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() + " consumed=" + queueUser.getUsed() + " headroom=" + headroom); } CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider( queueUser, this, application, required, queueResourceLimitsInfo); application.setHeadroomProvider(headroomProvider); metrics.setAvailableResourcesToUser(user, headroom); return userLimit; } LeafQueue(CapacitySchedulerContext cs, 
      String queueName, CSQueue parent, CSQueue old); @Override String getQueuePath(); @Private float getMinimumAllocationFactor(); @Private float getMaxAMResourcePerQueuePercent(); int getMaxApplications(); synchronized int getMaxApplicationsPerUser(); @Override ActiveUsersManager getActiveUsersManager(); @Override List<CSQueue> getChildQueues(); @Override synchronized int getNumApplications(); synchronized int getNumPendingApplications(); synchronized int getNumActiveApplications(); @Private synchronized int getNumApplications(String user); @Private synchronized int getNumPendingApplications(String user); @Private synchronized int getNumActiveApplications(String user); synchronized int getNumContainers(); @Override synchronized QueueState getState(); @Private synchronized int getUserLimit(); @Private synchronized float getUserLimitFactor(); @Override synchronized QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive); @Override synchronized List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); @Private int getNodeLocalityDelay(); String toString(); @VisibleForTesting synchronized void setNodeLabelManager(RMNodeLabelsManager mgr); @VisibleForTesting synchronized User getUser(String userName); synchronized ArrayList<UserInfo> getUsers(); @Override synchronized void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName); @Override void submitApplication(ApplicationId applicationId, String userName,
      String queue); synchronized Resource getAMResourceLimit(); synchronized Resource getUserAMResourceLimit(); synchronized Resource getUserAMResourceLimitPerPartition(
      String nodePartition); synchronized Resource getAMResourceLimitPerPartition(
      String nodePartition); @Override void finishApplication(ApplicationId application, String user); @Override void finishApplicationAttempt(FiCaSchedulerApp application, String queue); synchronized void removeApplicationAttempt(
      FiCaSchedulerApp application, User user); @Override synchronized CSAssignment assignContainers(Resource clusterResource,
      FiCaSchedulerNode node, ResourceLimits currentResourceLimits); @Override void completedContainer(Resource clusterResource, 
      FiCaSchedulerApp application, FiCaSchedulerNode node, RMContainer rmContainer, 
      ContainerStatus containerStatus, RMContainerEventType event, CSQueue childQueue,
      boolean sortQueues); @Override synchronized void updateClusterResource(Resource clusterResource,
      ResourceLimits currentResourceLimits); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt attempt, RMContainer rmContainer); Set<FiCaSchedulerApp> getApplications(); synchronized Resource getTotalResourcePending(); @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); void setCapacity(float capacity); void setAbsoluteCapacity(float absoluteCapacity); void setMaxApplications(int maxApplications);  }