diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java index 524a049..bdf60bd 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DefaultResourceCalculator.java @@ -67,6 +67,12 @@ public Resource divideAndCeil(Resource numerator, int denominator) { } @Override + public Resource divideAndCeil(Resource numerator, float denominator) { + return Resources.createResource( + divideAndCeil(numerator.getMemorySize(), denominator)); + } + + @Override public Resource normalize(Resource r, Resource minimumResource, Resource maximumResource, Resource stepFactor) { if (stepFactor.getMemorySize() == 0) { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java index 69fe716..ea9b927 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/DominantResourceCalculator.java @@ -155,6 +155,14 @@ public Resource divideAndCeil(Resource numerator, int denominator) { } @Override + public Resource divideAndCeil(Resource numerator, float denominator) { + return Resources.createResource( + divideAndCeil(numerator.getMemorySize(), denominator), + divideAndCeil(numerator.getVirtualCores(), denominator) + ); + } + + @Override public Resource normalize(Resource r, Resource minimumResource, Resource maximumResource, Resource stepFactor) { if (stepFactor.getMemorySize() == 0 || stepFactor.getVirtualCores() == 0) { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java index d219fe1..398dac5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceCalculator.java @@ -65,6 +65,13 @@ public static int divideAndCeil(int a, int b) { } return (a + (b - 1)) / b; } + + public static int divideAndCeil(int a, float b) { + if (b == 0) { + return 0; + } + return (int) Math.ceil(a / b); + } public static long divideAndCeil(long a, long b) { if (b == 0) { @@ -73,6 +80,13 @@ public static long divideAndCeil(long a, long b) { return (a + (b - 1)) / b; } + public static long divideAndCeil(long a, float b) { + if (b == 0) { + return 0; + } + return (long) Math.ceil(a/b); + } + public static int roundUp(int a, int b) { return divideAndCeil(a, b) * b; } @@ -198,6 +212,15 @@ public abstract float divide( * @return resultant resource */ public abstract Resource divideAndCeil(Resource numerator, int denominator); + + /** + * Divide-and-ceil numerator by denominator. + * + * @param numerator numerator resource + * @param denominator denominator + * @return resultant resource + */ + public abstract Resource divideAndCeil(Resource numerator, float denominator); /** * Check if a smaller resource can be contained by bigger resource. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/Resources.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/Resources.java index 91a5297..932fb82 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/Resources.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/Resources.java @@ -279,6 +279,11 @@ public static Resource divideAndCeil( ResourceCalculator resourceCalculator, Resource lhs, int rhs) { return resourceCalculator.divideAndCeil(lhs, rhs); } + + public static Resource divideAndCeil( + ResourceCalculator resourceCalculator, Resource lhs, float rhs) { + return resourceCalculator.divideAndCeil(lhs, rhs); + } public static boolean equals(Resource lhs, Resource rhs) { return lhs.equals(rhs); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ActiveUsersManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ActiveUsersManager.java index 36e6858..4723133 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ActiveUsersManager.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ActiveUsersManager.java @@ -43,6 +43,7 @@ private final QueueMetrics metrics; private int activeUsers = 0; + private boolean activeUsersChanged = false; private Map> usersApplications = new HashMap>(); @@ -65,6 +66,7 @@ synchronized public void activateApplication( usersApplications.put(user, userApps); ++activeUsers; metrics.incrActiveUsers(); + activeUsersChanged = true; LOG.debug("User " + user + " added to activeUsers, currently: " + activeUsers); } @@ -91,6 +93,7 @@ synchronized public void deactivateApplication( usersApplications.remove(user); --activeUsers; metrics.decrActiveUsers(); + activeUsersChanged = true; LOG.debug("User " + user + " removed from activeUsers, currently: " + activeUsers); } @@ -106,4 +109,30 @@ synchronized public void deactivateApplication( synchronized public int getNumActiveUsers() { return activeUsers; } + + /** + * Get list of active users + * @return a copy of the list of active users + */ + @Lock({Queue.class, SchedulerApplicationAttempt.class}) + synchronized public Set getActiveUsersSet() { + return new HashSet(usersApplications.keySet()); + } + + /** + * Get indicator of whether or not the active users list has changed. + * @return active users changed indicator + */ + @Lock({Queue.class, SchedulerApplicationAttempt.class}) + synchronized public boolean getActiveUsersChanged() { + return activeUsersChanged; + } + + /** + * Clear active users changed indicator + */ + @Lock({Queue.class, SchedulerApplicationAttempt.class}) + synchronized public boolean clearActiveUsersChanged() { + return activeUsersChanged = false; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java index 4fe008b..5fbdead 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/AbstractCSQueue.java @@ -112,6 +112,7 @@ protected ReentrantReadWriteLock.WriteLock writeLock; volatile Priority priority = Priority.newInstance(0); + private Map userWeights = new HashMap(); public AbstractCSQueue(CapacitySchedulerContext cs, String queueName, CSQueue parent, CSQueue old) throws IOException { @@ -333,11 +334,28 @@ void setupQueueConfigs(Resource clusterResource) this.priority = csContext.getConfiguration().getQueuePriority( getQueuePath()); + + this.userWeights = getUserWeightsFromHierarchy(); } finally { writeLock.unlock(); } } + private Map getUserWeightsFromHierarchy() throws IOException { + Map unionInheritedWeights = new HashMap(); + CSQueue parentQ = getParent(); + if (parentQ != null) { + // Inherit all of parent's user's weights + unionInheritedWeights.putAll(parentQ.getUserWeights()); + } + // Insert this queue's user's weights, overriding parent's user's weights if + // there is overlap. + CapacitySchedulerConfiguration csConf = csContext.getConfiguration(); + unionInheritedWeights.putAll( + csConf.getAllUserWeightsForQueue(getQueuePath())); + return unionInheritedWeights; + } + private void initializeQueueState(QueueState previousState, QueueState configuredState, QueueState parentState) { // verify that we can not any value for State other than RUNNING/STOPPED @@ -962,4 +980,9 @@ protected void appFinished() { public Priority getPriority() { return this.priority; } + + @Override + public Map getUserWeights() { + return userWeights; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java index b878e72..95aa6ee 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CSQueue.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.Collection; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.locks.ReentrantReadWriteLock; @@ -350,4 +351,10 @@ public void validateSubmitApplication(ApplicationId applicationId, * @return queue priority */ Priority getPriority(); + + /** + * Get a map of usernames and weights + * @return map of usernames and corresponding weight + */ + Map getUserWeights(); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java index 026dd82..d13e7a6 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacitySchedulerConfiguration.java @@ -108,6 +108,15 @@ public static final String USER_LIMIT_FACTOR = "user-limit-factor"; @Private + public static final String USER_WEIGHT = "weight"; + + @Private + public static final String USER_SETTINGS = "user-settings"; + + @Private + public static final float DEFAULT_USER_WEIGHT = 1.0f; + + @Private public static final String STATE = "state"; @Private @@ -1412,4 +1421,29 @@ public void setPUOrderingPolicyUnderUtilizedPreemptionMoveReservation( QUEUE_PRIORITY_UTILIZATION_ORDERING_POLICY, UNDER_UTILIZED_PREEMPTION_MOVE_RESERVATION), allowMoveReservation); } + + /** + * Get the weights of all users at this queue level from the configuration. + * Used in computing user-specific user limit, relative to other users. + * @param queuePath full queue path + * @return map of user weights, if they exists. Otherwise, return empty map. + */ + public Map getAllUserWeightsForQueue(String queuePath) { + Map userWeights = new HashMap (); + String qPathPlusPrefix = + getQueuePrefix(queuePath).replaceAll("\\.", "\\\\.") + + USER_SETTINGS + "\\."; + String weightKeyRegex = + qPathPlusPrefix + "\\w+\\." + USER_WEIGHT; + Map props = getValByRegex(weightKeyRegex); + for (Entry e : props.entrySet()) { + String userName = + e.getKey().replaceFirst(qPathPlusPrefix, "") + .replaceFirst("\\." + USER_WEIGHT, ""); + if (userName != null && !userName.isEmpty()) { + userWeights.put(userName, new Float(e.getValue())); + } + } + return userWeights; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java index 71225b8..8a5f798 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java @@ -88,6 +88,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Map.Entry; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; @@ -150,6 +151,10 @@ private Map> ignorePartitionExclusivityRMContainers = new ConcurrentHashMap<>(); + private Set activeUsersSet = new HashSet(); + private float activeUsersTimesWeights = 0.0f; + private float allUsersTimesWeights = 0.0f; + List priorityAcls = new ArrayList(); @@ -261,6 +266,20 @@ protected void setupQueueConfigs(Resource clusterResource) defaultAppPriorityPerQueue = Priority.newInstance( conf.getDefaultApplicationPriorityConfPerQueue(getQueuePath())); + // Validate leaf queue's user's weights. + int queueUL = Math.min(100, conf.getUserLimit(getQueuePath())); + for (Entry e : getUserWeights().entrySet()) { + float val = e.getValue().floatValue(); + if (val < 0.0f || val > (100.0f / queueUL)) { + throw new IOException("Weight (" + val + ") for user \"" + e.getKey() + + "\" must be between 0 and" + " 100 / " + queueUL + " (= " + + 100.0f/queueUL + ", the number of concurrent active users in " + + getQueuePath() + ")"); + } + } + + updateUserWeights(); + LOG.info( "Initializing " + queueName + "\n" + "capacity = " + queueCapacities .getCapacity() + " [= (float) configuredCapacity / 100 ]" + "\n" @@ -307,6 +326,20 @@ protected void setupQueueConfigs(Resource clusterResource) } } + private void updateUserWeights() { + try { + writeLock.lock(); + activeUsersSet = activeUsersManager.getActiveUsersSet(); + for (Map.Entry ue : users.entrySet()) { + ue.getValue().setWeight(getUserWeightFromQueue(ue.getKey())); + } + activeUsersTimesWeights = sumActiveUsersTimesWeights(); + allUsersTimesWeights = sumAllUsersTimesWeights(); + } finally { + writeLock.unlock(); + } + } + @Override public String getQueuePath() { return getParent().getQueuePath() + "." + getQueueName(); @@ -481,13 +514,16 @@ public User getUser(String userName) { } // Get and add user if absent - private User getUserAndAddIfAbsent(String userName) { + @VisibleForTesting + public User getUserAndAddIfAbsent(String userName) { try { writeLock.lock(); User u = users.get(userName); if (null == u) { u = new User(userName); users.put(userName, u); + u.setWeight(getUserWeightFromQueue(userName)); + allUsersTimesWeights = sumAllUsersTimesWeights(); } return u; } finally { @@ -495,6 +531,11 @@ private User getUserAndAddIfAbsent(String userName) { } } + private float getUserWeightFromQueue(String userName) { + Float weight = getUserWeights().get(userName); + return (weight == null) ? 1.0f : weight.floatValue(); + } + /** * @return an ArrayList of UserInfo objects who are active in this queue */ @@ -509,7 +550,8 @@ private User getUserAndAddIfAbsent(String userName) { user.getActiveApplications(), user.getPendingApplications(), Resources.clone(user.getConsumedAMResources()), Resources.clone(user.getUserResourceLimit()), - user.getResourceUsage())); + user.getResourceUsage(), user.getWeight(), + activeUsersSet.contains(user.userName))); } return usersToReturn; } finally { @@ -669,21 +711,38 @@ public Resource calculateAndGetAMResourceLimit() { @VisibleForTesting public Resource getUserAMResourceLimit() { - return getUserAMResourceLimitPerPartition(RMNodeLabelsManager.NO_LABEL); + return getUserAMResourceLimitPerPartition(RMNodeLabelsManager.NO_LABEL, + null); } public Resource getUserAMResourceLimitPerPartition( - String nodePartition) { + String nodePartition, String userName) { + float userWeight = 1.0f; + if (userName != null && getUser(userName) != null) { + userWeight = getUser(userName).getWeight(); + } try { readLock.lock(); + if (activeUsersManager.getActiveUsersChanged()) { + activeUsersSet = activeUsersManager.getActiveUsersSet(); + activeUsersTimesWeights = sumActiveUsersTimesWeights(); + activeUsersManager.clearActiveUsersChanged(); + } /* * The user am resource limit is based on the same approach as the user * limit (as it should represent a subset of that). This means that it uses * the absolute queue capacity (per partition) instead of the max and is * modified by the userlimit and the userlimit factor as is the userlimit */ - float effectiveUserLimit = Math.max(userLimit / 100.0f, + float effectiveUserLimit; + if (activeUsersTimesWeights > 0.0f) { + effectiveUserLimit = Math.max(userLimit / 100.0f, + 1.0f / activeUsersTimesWeights); + } else { + effectiveUserLimit = Math.max(userLimit / 100.0f, 1.0f / Math.max(getActiveUsersManager().getNumActiveUsers(), 1)); + } + effectiveUserLimit = Math.min(effectiveUserLimit * userWeight, 1.0f); Resource queuePartitionResource = Resources.multiplyAndNormalizeUp( resourceCalculator, @@ -822,7 +881,8 @@ private void activateApplications() { // Verify whether we already calculated user-am-limit for this label. if (userAMLimit == null) { - userAMLimit = getUserAMResourceLimitPerPartition(partitionName); + userAMLimit = getUserAMResourceLimitPerPartition(partitionName, + application.getUser()); userAmPartitionLimit.put(partitionName, userAMLimit); } @@ -951,6 +1011,7 @@ private void removeApplicationAttempt( user.finishApplication(wasActive); if (user.getTotalApplications() == 0) { users.remove(application.getUser()); + allUsersTimesWeights = sumAllUsersTimesWeights(); } // Check if we can activate more applications @@ -1463,20 +1524,25 @@ private Resource computeUserLimit(String userName, // Also, the queue's configured capacity should be higher than // queue-hard-limit * ulMin - final int usersCount; + float usersSummedByWeight; if (forActive) { - usersCount = activeUsersManager.getNumActiveUsers(); + if (activeUsersManager.getActiveUsersChanged()) { + activeUsersSet = activeUsersManager.getActiveUsersSet(); + activeUsersTimesWeights = sumActiveUsersTimesWeights(); + activeUsersManager.clearActiveUsersChanged(); + } + usersSummedByWeight = activeUsersTimesWeights; } else { - usersCount = users.size(); + usersSummedByWeight = allUsersTimesWeights; } // User limit resource is determined by: - // max{currentCapacity / #activeUsers, currentCapacity * + // max(currentCapacity / #activeUsers, currentCapacity * // user-limit-percentage%) Resource userLimitResource = Resources.max( resourceCalculator, partitionResource, Resources.divideAndCeil( - resourceCalculator, currentCapacity, usersCount), + resourceCalculator, currentCapacity, usersSummedByWeight), Resources.divideAndCeil( resourceCalculator, Resources.multiplyAndRoundDown( @@ -1524,18 +1590,49 @@ private Resource computeUserLimit(String userName, " qconsumed: " + queueUsage.getUsed() + " consumedRatio: " + totalUserConsumedRatio + " currentCapacity: " + currentCapacity + - " activeUsers: " + usersCount + + " activeUsers: " + usersSummedByWeight + " clusterCapacity: " + clusterResource + " resourceByLabel: " + partitionResource + " usageratio: " + qUsageRatios.getUsageRatio(nodePartition) + - " Partition: " + nodePartition + " Partition: " + nodePartition + + " maxUserLimit=" + maxUserLimit + + " userWeight=" + ((user != null) ? user.getWeight() : 1.0f) ); } + // Apply user's weight. + float weight = (user == null) ? 1.0f : user.getWeight(); + userLimitResource = + Resources.multiplyAndNormalizeDown(resourceCalculator, + userLimitResource, weight, minimumAllocation); + if (forActive) { user.setUserResourceLimit(userLimitResource); } return userLimitResource; } + + float sumActiveUsersTimesWeights() { + float count = 0.0f; + for (String userName : activeUsersSet) { + // Do the following instead of calling getUser so locking is not needed. + User user = users.get(userName); + count += (user != null) ? user.getWeight() : 1.0f; + } + return count; + } + + float sumAllUsersTimesWeights() { + float count = 0.0f; + try { + this.readLock.lock(); + for (String u : users.keySet()) { + count += (getUser(u) != null) ? getUser(u).getWeight() : 1.0f; + } + return count; + } finally { + this.readLock.unlock(); + } + } @Private protected boolean canAssignToUser(Resource clusterResource, @@ -1963,6 +2060,7 @@ public float getUsageRatio(String label) { private UsageRatios userUsageRatios = new UsageRatios(); private WriteLock writeLock; String userName; + float weight = 1.0f; @VisibleForTesting public User(String name) { @@ -2092,6 +2190,20 @@ public String getUserName() { public void setResourceUsage(ResourceUsage resourceUsage) { this.userResourceUsage = resourceUsage; } + + /** + * @return the weight + */ + public float getWeight() { + return weight; + } + + /** + * @param weight the weight to set + */ + public void setWeight(float weight) { + this.weight = weight; + } } @Override diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/UserInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/UserInfo.java index ff9d304..a1a8ecf 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/UserInfo.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/UserInfo.java @@ -37,11 +37,14 @@ protected ResourceInfo AMResourceUsed; protected ResourceInfo userResourceLimit; protected ResourcesInfo resources; + private float userWeight; + private boolean isActive; UserInfo() {} UserInfo(String username, Resource resUsed, int activeApps, int pendingApps, - Resource amResUsed, Resource resourceLimit, ResourceUsage resourceUsage) { + Resource amResUsed, Resource resourceLimit, ResourceUsage resourceUsage, + float weight, boolean isActive) { this.username = username; this.resourcesUsed = new ResourceInfo(resUsed); this.numActiveApplications = activeApps; @@ -49,6 +52,8 @@ this.AMResourceUsed = new ResourceInfo(amResUsed); this.userResourceLimit = new ResourceInfo(resourceLimit); this.resources = new ResourcesInfo(resourceUsage); + this.userWeight = weight; + this.isActive = isActive; } public String getUsername() { @@ -78,4 +83,12 @@ public ResourceInfo getUserResourceLimit() { public ResourcesInfo getResourceUsageInfo() { return resources; } + + public float getUserWeight() { + return userWeight; + } + + public boolean getIsActive() { + return isActive; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java index 3635398..81b4eb5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java @@ -881,8 +881,8 @@ protected void getPendingAppDiagnosticMessage( .append(queue.getAMResourceLimitPerPartition(appAMNodePartitionName)); diagnosticMessage.append("; "); diagnosticMessage.append("User AM Resource Limit of the queue = "); - diagnosticMessage.append( - queue.getUserAMResourceLimitPerPartition(appAMNodePartitionName)); + diagnosticMessage.append(queue.getUserAMResourceLimitPerPartition( + appAMNodePartitionName, getUser())); diagnosticMessage.append("; "); diagnosticMessage.append("Queue AM Resource Usage = "); diagnosticMessage.append( diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/CapacitySchedulerPage.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/CapacitySchedulerPage.java index b972428..292c5f3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/CapacitySchedulerPage.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/CapacitySchedulerPage.java @@ -68,6 +68,7 @@ "left:0%;background:none;border:1px dashed #BFBFBF"; static final String Q_OVER = "background:#FFA333"; static final String Q_UNDER = "background:#5BD75B"; + static final String ACTIVE_USER = "background:#FFFF00"; // Yellow highlight @RequestScoped static class CSQInfo { @@ -209,6 +210,7 @@ protected void render(Block html) { html.table("#userinfo").thead().$class("ui-widget-header").tr().th() .$class("ui-state-default")._("User Name")._().th() .$class("ui-state-default")._("Max Resource")._().th() + .$class("ui-state-default")._("Weight")._().th() .$class("ui-state-default")._("Used Resource")._().th() .$class("ui-state-default")._("Max AM Resource")._().th() .$class("ui-state-default")._("Used AM Resource")._().th() @@ -229,8 +231,11 @@ protected void render(Block html) { ResourceInfo amUsed = (resourceUsages.getAmUsed() == null) ? new ResourceInfo(Resources.none()) : resourceUsages.getAmUsed(); - tbody.tr().td(userInfo.getUsername()) + String highlightIfAsking = + userInfo.getIsActive() ? ACTIVE_USER : null; + tbody.tr().$style(highlightIfAsking).td(userInfo.getUsername()) .td(userInfo.getUserResourceLimit().toString()) + .td(String.valueOf(userInfo.getUserWeight())) .td(resourcesUsed.toString()) .td(resourceUsages.getAMLimit().toString()) .td(amUsed.toString()) @@ -399,6 +404,8 @@ public void render(Block html) { _("Used (over capacity)")._(). span().$class("qlegend ui-corner-all ui-state-default"). _("Max Capacity")._(). + span().$class("qlegend ui-corner-all").$style(ACTIVE_USER). + _("Users Requesting Resources")._(). _(); float used = 0; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java index f572ea3..42eb495 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java @@ -3869,4 +3869,128 @@ public void tearDown() throws Exception { cs.stop(); } } + + @Test + public void testUserSpecificUserLimits() throws Exception { + // Mock the queue + LeafQueue a = stubLeafQueue((LeafQueue)queues.get(A)); + // Set minimum-user-limit-percent for queue "a" in the configs. + csConf.setUserLimit(a.getQueuePath(), 50); + // Set weight for "user_0" to be 1.5 for the a queue in the configs. + csConf.setFloat("yarn.scheduler.capacity." + a.getQueuePath() + + ".user-settings.user_0." + CapacitySchedulerConfiguration.USER_WEIGHT, + 1.5f); + + when(csContext.getClusterResource()) + .thenReturn(Resources.createResource(16 * GB, 32)); + // Verify that configs were updated and parsed correctly. + Assert.assertNull(a.getUserWeights().get("user_0")); + a.reinitialize(a, csContext.getClusterResource()); + assertEquals(1.5, a.getUserWeights().get("user_0").floatValue(), 0.0); + + // set maxCapacity + a.setMaxCapacity(1.0f); + + // Set minimum user-limit-percent + a.setUserLimit(50); + a.setUserLimitFactor(2); + + // Users + final String user_0 = "user_0"; + final String user_1 = "user_1"; + + // Set user_0's weight to 1.5 in the a queue's object. + a.getUserAndAddIfAbsent(user_0).setWeight(1.5f); + + // Submit applications + final ApplicationAttemptId appAttemptId_0 = + TestUtils.getMockApplicationAttemptId(0, 0); + FiCaSchedulerApp app_0 = + new FiCaSchedulerApp(appAttemptId_0, user_0, a, + a.getActiveUsersManager(), spyRMContext); + a.submitApplicationAttempt(app_0, user_0); + + final ApplicationAttemptId appAttemptId_1 = + TestUtils.getMockApplicationAttemptId(1, 0); + FiCaSchedulerApp app_1 = + new FiCaSchedulerApp(appAttemptId_1, user_1, a, + a.getActiveUsersManager(), spyRMContext); + a.submitApplicationAttempt(app_1, user_1); // different user + + // Setup some nodes + 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), numNodes * 16); + when(csContext.getNumClusterNodes()).thenReturn(numNodes); + + // Setup resource-requests + // app_0 asks for 3 3-GB containers + Priority priority = TestUtils.createMockPriority(1); + app_0.updateResourceRequests(Collections.singletonList( + TestUtils.createResourceRequest(ResourceRequest.ANY, 4*GB, 3, true, + priority, recordFactory))); + + // app_1 asks for 2 1-GB containers + app_1.updateResourceRequests(Collections.singletonList( + TestUtils.createResourceRequest(ResourceRequest.ANY, 1*GB, 2, true, + priority, recordFactory))); + + Map apps = ImmutableMap.of( + app_0.getApplicationAttemptId(), app_0, app_1.getApplicationAttemptId(), + app_1); + Map nodes = ImmutableMap.of(node_0.getNodeID(), + node_0, node_1.getNodeID(), node_1); + + /** + * Start testing... + */ + + // There're two active users + assertEquals(2, a.getActiveUsersManager().getNumActiveUsers()); + + // 1 container to user_0. Since queue starts out empty, user limit would + // normally be calculated to be the minumum container size (1024GB). + // However, in this case, user_0 has a weight of 1.5, so the UL is 2048GB + // because 1024 * 1.5 rounded up to container size is 2048GB. + applyCSAssignment(clusterResource, + a.assignContainers(clusterResource, node_0, + new ResourceLimits(clusterResource), + SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), a, nodes, apps); + assertEquals(4*GB, a.getUsedResources().getMemorySize()); + assertEquals(4*GB, app_0.getCurrentConsumption().getMemorySize()); + assertEquals(0*GB, app_1.getCurrentConsumption().getMemorySize()); + + // At this point the queue-wide user limit is 3072GB, but since user_0 has a + // weight of 1.5, its user limit is 5120GB. So, even though user_0 already + // has 4096GB, it is under its user limit, so it gets another container. + applyCSAssignment(clusterResource, + a.assignContainers(clusterResource, node_0, + new ResourceLimits(clusterResource), + SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), a, nodes, apps); + assertEquals(8*GB, a.getUsedResources().getMemorySize()); + assertEquals(8*GB, app_0.getCurrentConsumption().getMemorySize()); + assertEquals(0*GB, app_1.getCurrentConsumption().getMemorySize()); + + // Queue-wide user limit at this point is 4069GB and user_0's user limit is + // 6144GB. user_0 has 8192GB. + // Now that user_0 is above its user limit, the next container should go to user_1 + applyCSAssignment(clusterResource, + a.assignContainers(clusterResource, node_1, + new ResourceLimits(clusterResource), + SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), a, nodes, apps); + assertEquals(9*GB, a.getUsedResources().getMemorySize()); + assertEquals(8*GB, app_0.getCurrentConsumption().getMemorySize()); + assertEquals(1*GB, app_1.getCurrentConsumption().getMemorySize()); + + assertEquals(4*GB, + app_0.getTotalPendingRequestsPerPartition().get("").getMemorySize()); + + assertEquals(1*GB, + app_1.getTotalPendingRequestsPerPartition().get("").getMemorySize()); + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/CapacityScheduler.md b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/CapacityScheduler.md index 737bdc2..f1d4535 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/CapacityScheduler.md +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/CapacityScheduler.md @@ -124,6 +124,7 @@ Configuration | `yarn.scheduler.capacity..user-limit-factor` | The multiple of the queue capacity which can be configured to allow a single user to acquire more resources. By default this is set to 1 which ensures that a single user can never take more than the queue's configured capacity irrespective of how idle the cluster is. Value is specified as a float. | | `yarn.scheduler.capacity..maximum-allocation-mb` | The per queue maximum limit of memory to allocate to each container request at the Resource Manager. This setting overrides the cluster configuration `yarn.scheduler.maximum-allocation-mb`. This value must be smaller than or equal to the cluster maximum. | | `yarn.scheduler.capacity..maximum-allocation-vcores` | The per queue maximum limit of virtual cores to allocate to each container request at the Resource Manager. This setting overrides the cluster configuration `yarn.scheduler.maximum-allocation-vcores`. This value must be smaller than or equal to the cluster maximum. | +| `yarn.scheduler.capacity..user-settings..weight` | This floating point value is used when calculating the user limit resource values for users in a queue. This value will weight each user more or less than the other users in the queue. For example, if user A should receive 50% more resources in a queue than users B and C, this property will be set to 1.5 for user A. Users B and C will default to 1.0. | * Running and Pending Application Limits