hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From wan...@apache.org
Subject [2/2] hadoop git commit: YARN-5889. Improve and refactor user-limit calculation in Capacity Scheduler. (Sunil G via wangda)
Date Thu, 09 Feb 2017 18:26:48 GMT
YARN-5889. Improve and refactor user-limit calculation in Capacity Scheduler. (Sunil G via wangda)


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/5fb723bb
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/5fb723bb
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/5fb723bb

Branch: refs/heads/trunk
Commit: 5fb723bb77722d41df6959eee23e1b0cfeb5584e
Parents: b6bb99c
Author: Wangda Tan <wangda@apache.org>
Authored: Thu Feb 9 10:23:50 2017 -0800
Committer: Wangda Tan <wangda@apache.org>
Committed: Thu Feb 9 10:23:50 2017 -0800

----------------------------------------------------------------------
 .../FifoIntraQueuePreemptionPlugin.java         |   6 +-
 .../scheduler/AbstractUsersManager.java         |  54 +
 .../scheduler/ActiveUsersManager.java           |  23 +-
 .../scheduler/AppSchedulingInfo.java            |  16 +-
 .../server/resourcemanager/scheduler/Queue.java |   2 +-
 .../scheduler/SchedulerApplicationAttempt.java  |   4 +-
 .../scheduler/capacity/CSQueue.java             |   8 +-
 .../capacity/CapacityHeadroomProvider.java      |   4 +-
 .../scheduler/capacity/CapacityScheduler.java   |   3 +-
 .../scheduler/capacity/LeafQueue.java           | 589 +++--------
 .../scheduler/capacity/ParentQueue.java         |   2 +-
 .../scheduler/capacity/UsersManager.java        | 982 +++++++++++++++++++
 .../scheduler/common/fica/FiCaSchedulerApp.java |  14 +-
 .../scheduler/fair/FSLeafQueue.java             |   2 +-
 .../scheduler/fair/FSParentQueue.java           |   2 +-
 .../scheduler/fifo/FifoScheduler.java           |   2 +-
 .../dao/CapacitySchedulerLeafQueueInfo.java     |   2 +-
 ...alCapacityPreemptionPolicyMockFramework.java |   8 +-
 .../TestSchedulerApplicationAttempt.java        |  14 +-
 .../capacity/TestApplicationLimits.java         |  15 +-
 .../TestApplicationLimitsByPartition.java       |  23 +-
 .../TestCapacitySchedulerNodeLabelUpdate.java   |   4 +-
 .../scheduler/capacity/TestLeafQueue.java       |  98 +-
 .../TestNodeLabelContainerAllocation.java       |   7 +-
 24 files changed, 1302 insertions(+), 582 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/FifoIntraQueuePreemptionPlugin.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/FifoIntraQueuePreemptionPlugin.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/FifoIntraQueuePreemptionPlugin.java
index 757f567..5f1af1e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/FifoIntraQueuePreemptionPlugin.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/FifoIntraQueuePreemptionPlugin.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.IntraQueueCandidatesSelector.TAPriorityComparator;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
 import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.Resources;
@@ -264,8 +265,9 @@ public class FifoIntraQueuePreemptionPlugin
 
       // Verify whether we already calculated headroom for this user.
       if (userLimitResource == null) {
-        userLimitResource = Resources.clone(tq.leafQueue
-            .getUserLimitPerUser(userName, partitionBasedResource, partition));
+        userLimitResource = Resources.clone(
+            tq.leafQueue.getResourceLimitForAllUsers(userName, clusterResource,
+                partition, SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY));
 
         Resource amUsed = perUserAMUsed.get(userName);
         if (null == amUsed) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractUsersManager.java
----------------------------------------------------------------------
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/AbstractUsersManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractUsersManager.java
new file mode 100644
index 0000000..4db3584
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AbstractUsersManager.java
@@ -0,0 +1,54 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+
+/**
+ * {@link AbstractUsersManager} tracks users in the system.
+ */
+@Private
+public interface AbstractUsersManager {
+  /**
+   * An application has new outstanding requests.
+   *
+   * @param user
+   *          application user
+   * @param applicationId
+   *          activated application
+   */
+  void activateApplication(String user, ApplicationId applicationId);
+  /**
+   * An application has no more outstanding requests.
+   *
+   * @param user
+   *          application user
+   * @param applicationId
+   *          deactivated application
+   */
+  void deactivateApplication(String user, ApplicationId applicationId);
+
+  /**
+   * Get number of active users i.e. users with applications which have pending
+   * resource requests.
+   *
+   * @return number of active users
+   */
+  int getNumActiveUsers();
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/ActiveUsersManager.java
----------------------------------------------------------------------
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..049f324 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
@@ -36,8 +36,8 @@ import org.apache.hadoop.yarn.server.utils.Lock;
  * An active user is defined as someone with outstanding resource requests.
  */
 @Private
-public class ActiveUsersManager {
-  
+public class ActiveUsersManager implements AbstractUsersManager {
+
   private static final Log LOG = LogFactory.getLog(ActiveUsersManager.class);
   
   private final QueueMetrics metrics;
@@ -45,7 +45,7 @@ public class ActiveUsersManager {
   private int activeUsers = 0;
   private Map<String, Set<ApplicationId>> usersApplications = 
       new HashMap<String, Set<ApplicationId>>();
-  
+
   public ActiveUsersManager(QueueMetrics metrics) {
     this.metrics = metrics;
   }
@@ -57,6 +57,7 @@ public class ActiveUsersManager {
    * @param applicationId activated application
    */
   @Lock({Queue.class, SchedulerApplicationAttempt.class})
+  @Override
   synchronized public void activateApplication(
       String user, ApplicationId applicationId) {
     Set<ApplicationId> userApps = usersApplications.get(user);
@@ -65,8 +66,10 @@ public class ActiveUsersManager {
       usersApplications.put(user, userApps);
       ++activeUsers;
       metrics.incrActiveUsers();
-      LOG.debug("User " + user + " added to activeUsers, currently: " + 
-          activeUsers);
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("User " + user + " added to activeUsers, currently: "
+            + activeUsers);
+      }
     }
     if (userApps.add(applicationId)) {
       metrics.activateApp(user);
@@ -80,6 +83,7 @@ public class ActiveUsersManager {
    * @param applicationId deactivated application
    */
   @Lock({Queue.class, SchedulerApplicationAttempt.class})
+  @Override
   synchronized public void deactivateApplication(
       String user, ApplicationId applicationId) {
     Set<ApplicationId> userApps = usersApplications.get(user);
@@ -91,18 +95,21 @@ public class ActiveUsersManager {
         usersApplications.remove(user);
         --activeUsers;
         metrics.decrActiveUsers();
-        LOG.debug("User " + user + " removed from activeUsers, currently: " + 
-            activeUsers);
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("User " + user + " removed from activeUsers, currently: "
+              + activeUsers);
+        }
       }
     }
   }
-  
+
   /**
    * Get number of active users i.e. users with applications which have pending
    * resource requests.
    * @return number of active users
    */
   @Lock({Queue.class, SchedulerApplicationAttempt.class})
+  @Override
   synchronized public int getNumActiveUsers() {
     return activeUsers;
   }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
----------------------------------------------------------------------
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/AppSchedulingInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
index e2ff082..48ecd2e 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/AppSchedulingInfo.java
@@ -73,7 +73,7 @@ public class AppSchedulingInfo {
   private final String user;
 
   private Queue queue;
-  private ActiveUsersManager activeUsersManager;
+  private AbstractUsersManager abstractUsersManager;
   // whether accepted/allocated by scheduler
   private volatile boolean pending = true;
   private ResourceUsage appResourceUsage;
@@ -99,13 +99,13 @@ public class AppSchedulingInfo {
   public final ContainerUpdateContext updateContext;
 
   public AppSchedulingInfo(ApplicationAttemptId appAttemptId,
-      String user, Queue queue, ActiveUsersManager activeUsersManager,
+      String user, Queue queue, AbstractUsersManager abstractUsersManager,
       long epoch, ResourceUsage appResourceUsage) {
     this.applicationAttemptId = appAttemptId;
     this.applicationId = appAttemptId.getApplicationId();
     this.queue = queue;
     this.user = user;
-    this.activeUsersManager = activeUsersManager;
+    this.abstractUsersManager = abstractUsersManager;
     this.containerIdCounter = new AtomicLong(
         epoch << ResourceManager.EPOCH_BIT_SHIFT);
     this.appResourceUsage = appResourceUsage;
@@ -477,7 +477,7 @@ public class AppSchedulingInfo {
       // Activate application. Metrics activation is done here.
       if (lastRequestContainers <= 0) {
         incrementSchedulerKeyReference(schedulerKey);
-        activeUsersManager.activateApplication(user, applicationId);
+        abstractUsersManager.activateApplication(user, applicationId);
       }
     }
 
@@ -735,7 +735,7 @@ public class AppSchedulingInfo {
 
   public void checkForDeactivation() {
     if (schedulerKeys.isEmpty()) {
-      activeUsersManager.deactivateApplication(user, applicationId);
+      abstractUsersManager.deactivateApplication(user, applicationId);
     }
   }
   
@@ -763,9 +763,9 @@ public class AppSchedulingInfo {
       }
       oldMetrics.moveAppFrom(this);
       newMetrics.moveAppTo(this);
-      activeUsersManager.deactivateApplication(user, applicationId);
-      activeUsersManager = newQueue.getActiveUsersManager();
-      activeUsersManager.activateApplication(user, applicationId);
+      abstractUsersManager.deactivateApplication(user, applicationId);
+      abstractUsersManager = newQueue.getAbstractUsersManager();
+      abstractUsersManager.activateApplication(user, applicationId);
       this.queue = newQueue;
     } finally {
       this.writeLock.unlock();

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Queue.java
----------------------------------------------------------------------
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/Queue.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Queue.java
index ada2a0b..d166e5f 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Queue.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Queue.java
@@ -63,7 +63,7 @@ public interface Queue {
 
   boolean hasAccess(QueueACL acl, UserGroupInformation user);
   
-  public ActiveUsersManager getActiveUsersManager();
+  public AbstractUsersManager getAbstractUsersManager();
 
   /**
    * Recover the state of the queue for a given container.

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
----------------------------------------------------------------------
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/SchedulerApplicationAttempt.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
index 3e41cb4..0e79838 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/SchedulerApplicationAttempt.java
@@ -197,13 +197,13 @@ public class SchedulerApplicationAttempt implements SchedulableEntity {
   private AtomicInteger unconfirmedAllocatedVcores = new AtomicInteger();
 
   public SchedulerApplicationAttempt(ApplicationAttemptId applicationAttemptId, 
-      String user, Queue queue, ActiveUsersManager activeUsersManager,
+      String user, Queue queue, AbstractUsersManager abstractUsersManager,
       RMContext rmContext) {
     Preconditions.checkNotNull(rmContext, "RMContext should not be null");
     this.rmContext = rmContext;
     this.appSchedulingInfo = 
         new AppSchedulingInfo(applicationAttemptId, user, queue,  
-            activeUsersManager, rmContext.getEpoch(), attemptResourceUsage);
+            abstractUsersManager, rmContext.getEpoch(), attemptResourceUsage);
     this.queue = queue;
     this.pendingRelease = Collections.newSetFromMap(
         new ConcurrentHashMap<ContainerId, Boolean>());

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/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
----------------------------------------------------------------------
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 2e3ced5..a65b3d2 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
@@ -40,7 +40,7 @@ import org.apache.hadoop.yarn.exceptions.InvalidResourceRequestException;
 import org.apache.hadoop.yarn.security.PrivilegedEntity;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AbstractUsersManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
@@ -262,10 +262,10 @@ public interface CSQueue extends SchedulerQueue<CSQueue> {
       ResourceLimits resourceLimits);
   
   /**
-   * Get the {@link ActiveUsersManager} for the queue.
-   * @return the <code>ActiveUsersManager</code> for the queue
+   * Get the {@link AbstractUsersManager} for the queue.
+   * @return the <code>AbstractUsersManager</code> for the queue
    */
-  public ActiveUsersManager getActiveUsersManager();
+  public AbstractUsersManager getAbstractUsersManager();
   
   /**
    * Adds all applications in the queue and its subqueues to the given collection.

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityHeadroomProvider.java
----------------------------------------------------------------------
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/CapacityHeadroomProvider.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/CapacityHeadroomProvider.java
index 5605f18..140a2ac 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/CapacityHeadroomProvider.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/CapacityHeadroomProvider.java
@@ -26,12 +26,12 @@ import org.apache.hadoop.yarn.util.resource.Resources;
 
 public class CapacityHeadroomProvider {
   
-  LeafQueue.User user;
+  UsersManager.User user;
   LeafQueue queue;
   FiCaSchedulerApp application;
   LeafQueue.QueueResourceLimitsInfo queueResourceLimitsInfo;
   
-  public CapacityHeadroomProvider(LeafQueue.User user, LeafQueue queue,
+  public CapacityHeadroomProvider(UsersManager.User user, LeafQueue queue,
       FiCaSchedulerApp application,
       LeafQueue.QueueResourceLimitsInfo queueResourceLimitsInfo) {
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java
----------------------------------------------------------------------
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/CapacityScheduler.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/CapacityScheduler.java
index 03bdd3a..3517764 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/CapacityScheduler.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/CapacityScheduler.java
@@ -23,7 +23,6 @@ import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.List;
@@ -751,7 +750,7 @@ public class CapacityScheduler extends
       CSQueue queue = (CSQueue) application.getQueue();
 
       FiCaSchedulerApp attempt = new FiCaSchedulerApp(applicationAttemptId,
-          application.getUser(), queue, queue.getActiveUsersManager(),
+          application.getUser(), queue, queue.getAbstractUsersManager(),
           rmContext, application.getPriority(), isAttemptRecovering,
           activitiesManager);
       if (transferStateFromPreviousAttempt) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/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
----------------------------------------------------------------------
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 bacfeef..b65f16a 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
@@ -21,9 +21,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity;
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
@@ -56,7 +53,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsMana
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
 
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
@@ -67,6 +63,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.Activi
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesLogger;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt.AMState;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.UsersManager.User;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.KillableContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ContainerAllocationProposal;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCommitRequest;
@@ -80,7 +77,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.OrderingPo
 import org.apache.hadoop.yarn.server.utils.Lock;
 import org.apache.hadoop.yarn.server.utils.Lock.NoLock;
 import org.apache.hadoop.yarn.util.SystemClock;
-import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.Resources;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -92,8 +88,6 @@ public class LeafQueue extends AbstractCSQueue {
   private static final Log LOG = LogFactory.getLog(LeafQueue.class);
 
   private float absoluteUsedCapacity = 0.0f;
-  private volatile int userLimit;
-  private volatile float userLimitFactor;
 
   protected int maxApplications;
   protected volatile int maxApplicationsPerUser;
@@ -112,14 +106,12 @@ public class LeafQueue extends AbstractCSQueue {
 
   private volatile float minimumAllocationFactor;
 
-  private Map<String, User> users = new ConcurrentHashMap<>();
-
   private final RecordFactory recordFactory = 
     RecordFactoryProvider.getRecordFactory(null);
 
   private CapacitySchedulerContext scheduler;
   
-  private final ActiveUsersManager activeUsersManager;
+  private final UsersManager usersManager;
 
   // cache last cluster resource to compute actual capacity
   private Resource lastClusterResource = Resources.none();
@@ -131,10 +123,6 @@ public class LeafQueue extends AbstractCSQueue {
 
   private volatile OrderingPolicy<FiCaSchedulerApp> orderingPolicy = null;
 
-  // Summation of consumed ratios for all users in queue
-  private float totalUserConsumedRatio = 0;
-  private UsageRatios qUsageRatios;
-
   // record all ignore partition exclusivityRMContainer, this will be used to do
   // preemption, key is the partition of the RMContainer allocated on
   private Map<String, TreeSet<RMContainer>> ignorePartitionExclusivityRMContainers =
@@ -149,13 +137,12 @@ public class LeafQueue extends AbstractCSQueue {
     super(cs, queueName, parent, old);
     this.scheduler = cs;
 
-    this.activeUsersManager = new ActiveUsersManager(metrics);
+    this.usersManager = new UsersManager(metrics, this, labelManager, scheduler,
+        resourceCalculator);
 
     // One time initialization is enough since it is static ordering policy
     this.pendingOrderingPolicy = new FifoOrderingPolicyForPendingApps();
 
-    qUsageRatios = new UsageRatios();
-
     if(LOG.isDebugEnabled()) {
       LOG.debug("LeafQueue:" + " name=" + queueName
         + ", fullname=" + getQueuePath());
@@ -187,8 +174,8 @@ public class LeafQueue extends AbstractCSQueue {
       setOrderingPolicy(
           conf.<FiCaSchedulerApp>getAppOrderingPolicy(getQueuePath()));
 
-      userLimit = conf.getUserLimit(getQueuePath());
-      userLimitFactor = conf.getUserLimitFactor(getQueuePath());
+      usersManager.setUserLimit(conf.getUserLimit(getQueuePath()));
+      usersManager.setUserLimitFactor(conf.getUserLimitFactor(getQueuePath()));
 
       maxApplications = conf.getMaximumApplicationsPerQueue(getQueuePath());
       if (maxApplications < 0) {
@@ -202,7 +189,8 @@ public class LeafQueue extends AbstractCSQueue {
         }
       }
       maxApplicationsPerUser = Math.min(maxApplications,
-          (int) (maxApplications * (userLimit / 100.0f) * userLimitFactor));
+          (int) (maxApplications * (usersManager.getUserLimit() / 100.0f)
+              * usersManager.getUserLimitFactor()));
 
       maxAMResourcePerQueuePercent =
           conf.getMaximumApplicationMasterResourcePerQueuePercent(
@@ -260,8 +248,9 @@ public class LeafQueue extends AbstractCSQueue {
               + queueCapacities.getAbsoluteMaximumCapacity()
               + " [= 1.0 maximumCapacity undefined, "
               + "(parentAbsoluteMaxCapacity * maximumCapacity) / 100 otherwise ]"
-              + "\n" + "userLimit = " + userLimit + " [= configuredUserLimit ]"
-              + "\n" + "userLimitFactor = " + userLimitFactor
+              + "\n" + "userLimit = " + usersManager.getUserLimit()
+              + " [= configuredUserLimit ]" + "\n" + "userLimitFactor = "
+              + usersManager.getUserLimitFactor()
               + " [= configuredUserLimitFactor ]" + "\n" + "maxApplications = "
               + maxApplications
               + " [= configuredMaximumSystemApplicationsPerQueue or"
@@ -322,9 +311,17 @@ public class LeafQueue extends AbstractCSQueue {
     return maxApplicationsPerUser;
   }
 
+  /**
+   *
+   * @return UsersManager instance.
+   */
+  public UsersManager getUsersManager() {
+    return usersManager;
+  }
+
   @Override
-  public ActiveUsersManager getActiveUsersManager() {
-    return activeUsersManager;
+  public AbstractUsersManager getAbstractUsersManager() {
+    return usersManager;
   }
 
   @Override
@@ -338,7 +335,8 @@ public class LeafQueue extends AbstractCSQueue {
    */
   @VisibleForTesting
   void setUserLimit(int userLimit) {
-    this.userLimit = userLimit;
+    usersManager.setUserLimit(userLimit);
+    usersManager.userLimitNeedsRecompute();
   }
 
   /**
@@ -347,7 +345,8 @@ public class LeafQueue extends AbstractCSQueue {
    */
   @VisibleForTesting
   void setUserLimitFactor(float userLimitFactor) {
-    this.userLimitFactor = userLimitFactor;
+    usersManager.setUserLimitFactor(userLimitFactor);
+    usersManager.userLimitNeedsRecompute();
   }
 
   @Override
@@ -408,12 +407,12 @@ public class LeafQueue extends AbstractCSQueue {
 
   @Private
   public int getUserLimit() {
-    return userLimit;
+    return usersManager.getUserLimit();
   }
 
   @Private
   public float getUserLimitFactor() {
-    return userLimitFactor;
+    return usersManager.getUserLimitFactor();
   }
 
   @Override
@@ -463,44 +462,7 @@ public class LeafQueue extends AbstractCSQueue {
 
   @VisibleForTesting
   public User getUser(String userName) {
-    return users.get(userName);
-  }
-
-  // Get and add user if absent
-  private User getUserAndAddIfAbsent(String userName) {
-    try {
-      writeLock.lock();
-      User u = users.get(userName);
-      if (null == u) {
-        u = new User();
-        users.put(userName, u);
-      }
-      return u;
-    } finally {
-      writeLock.unlock();
-    }
-  }
-
-  /**
-   * @return an ArrayList of UserInfo objects who are active in this queue
-   */
-  public ArrayList<UserInfo> getUsers() {
-    try {
-      readLock.lock();
-      ArrayList<UserInfo> usersToReturn = new ArrayList<UserInfo>();
-      for (Map.Entry<String, User> entry : users.entrySet()) {
-        User user = entry.getValue();
-        usersToReturn.add(
-            new UserInfo(entry.getKey(), Resources.clone(user.getAllUsed()),
-                user.getActiveApplications(), user.getPendingApplications(),
-                Resources.clone(user.getConsumedAMResources()),
-                Resources.clone(user.getUserResourceLimit()),
-                user.getResourceUsage()));
-      }
-      return usersToReturn;
-    } finally {
-      readLock.unlock();
-    }
+    return usersManager.getUser(userName);
   }
 
   @Private
@@ -561,7 +523,7 @@ public class LeafQueue extends AbstractCSQueue {
 
       // TODO, should use getUser, use this method just to avoid UT failure
       // which is caused by wrong invoking order, will fix UT separately
-      User user = getUserAndAddIfAbsent(userName);
+      User user = usersManager.getUserAndAddIfAbsent(userName);
 
       // Add the attempt to our data-structures
       addApplicationAttempt(application, user);
@@ -618,7 +580,7 @@ public class LeafQueue extends AbstractCSQueue {
       }
 
       // Check submission limits for the user on this queue
-      User user = getUserAndAddIfAbsent(userName);
+      User user = usersManager.getUserAndAddIfAbsent(userName);
       if (user.getTotalApplications() >= getMaxApplicationsPerUser()) {
         String msg = "Queue " + getQueuePath() + " already has " + user
             .getTotalApplications() + " applications from user " + userName
@@ -668,19 +630,21 @@ public class LeafQueue extends AbstractCSQueue {
        * 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,
-          1.0f / Math.max(getActiveUsersManager().getNumActiveUsers(), 1));
+      float effectiveUserLimit = Math.max(usersManager.getUserLimit() / 100.0f,
+          1.0f / Math.max(getAbstractUsersManager().getNumActiveUsers(), 1));
 
-      Resource queuePartitionResource = Resources.multiplyAndNormalizeUp(
-          resourceCalculator,
-          labelManager.getResourceByLabel(nodePartition, lastClusterResource),
-          queueCapacities.getAbsoluteCapacity(nodePartition),
-          minimumAllocation);
+      Resource queuePartitionResource = Resources
+          .multiplyAndNormalizeUp(resourceCalculator,
+              labelManager.getResourceByLabel(nodePartition,
+                  lastClusterResource),
+              queueCapacities.getAbsoluteCapacity(nodePartition),
+              minimumAllocation);
 
       Resource userAMLimit = Resources.multiplyAndNormalizeUp(
           resourceCalculator, queuePartitionResource,
           queueCapacities.getMaxAMResourcePercentage(nodePartition)
-              * effectiveUserLimit * userLimitFactor, minimumAllocation);
+              * effectiveUserLimit * usersManager.getUserLimitFactor(),
+          minimumAllocation);
       return Resources.lessThanOrEqual(resourceCalculator, lastClusterResource,
           userAMLimit, getAMResourceLimitPerPartition(nodePartition)) ?
           userAMLimit :
@@ -895,7 +859,7 @@ public class LeafQueue extends AbstractCSQueue {
   @Override
   public void finishApplication(ApplicationId application, String user) {
     // Inform the activeUsersManager
-    activeUsersManager.deactivateApplication(user, application);
+    usersManager.deactivateApplication(user, application);
 
     appFinished();
 
@@ -917,7 +881,7 @@ public class LeafQueue extends AbstractCSQueue {
 
       // TODO, should use getUser, use this method just to avoid UT failure
       // which is caused by wrong invoking order, will fix UT separately
-      User user = getUserAndAddIfAbsent(userName);
+      User user = usersManager.getUserAndAddIfAbsent(userName);
 
       String partitionName = application.getAppAMNodePartitionName();
       boolean wasActive = orderingPolicy.removeSchedulableEntity(application);
@@ -935,7 +899,7 @@ public class LeafQueue extends AbstractCSQueue {
 
       user.finishApplication(wasActive);
       if (user.getTotalApplications() == 0) {
-        users.remove(application.getUser());
+        usersManager.removeUser(application.getUser());
       }
 
       // Check if we can activate more applications
@@ -1282,7 +1246,7 @@ public class LeafQueue extends AbstractCSQueue {
       Resource clusterResource, FiCaSchedulerApp application,
       String partition) {
     return getHeadroom(user, queueCurrentLimit, clusterResource,
-        computeUserLimit(application.getUser(), clusterResource, user,
+        getResourceLimitForActiveUsers(application.getUser(), clusterResource,
             partition, SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY),
         partition);
   }
@@ -1356,7 +1320,7 @@ public class LeafQueue extends AbstractCSQueue {
     // Compute user limit respect requested labels,
     // TODO, need consider headroom respect labels also
     Resource userLimit =
-        computeUserLimit(application.getUser(), clusterResource, queueUser,
+        getResourceLimitForActiveUsers(application.getUser(), clusterResource,
             nodePartition, schedulingMode);
 
     setQueueResourceLimitsInfo(clusterResource);
@@ -1366,11 +1330,11 @@ public class LeafQueue extends AbstractCSQueue {
             clusterResource, userLimit, nodePartition);
     
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Headroom calculation for user " + user + ": " + 
-          " userLimit=" + userLimit + 
-          " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() +
-          " consumed=" + queueUser.getUsed() + 
-          " headroom=" + headroom);
+      LOG.debug("Headroom calculation for user " + user + ": " + " userLimit="
+          + userLimit + " queueMaxAvailRes="
+          + cachedResourceLimitsForHeadroom.getLimit() + " consumed="
+          + queueUser.getUsed() + " headroom=" + headroom + " partition="
+          + nodePartition);
     }
     
     CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider(
@@ -1393,129 +1357,46 @@ public class LeafQueue extends AbstractCSQueue {
     return rackLocalityFullReset;
   }
 
-  @Lock(NoLock.class)
-  private Resource computeUserLimit(String userName,
-      Resource clusterResource, User user,
-      String nodePartition, SchedulingMode schedulingMode) {
-    Resource partitionResource = labelManager.getResourceByLabel(nodePartition,
-        clusterResource);
-
-    // What is our current capacity? 
-    // * It is equal to the max(required, queue-capacity) if
-    //   we're running below capacity. The 'max' ensures that jobs in queues
-    //   with miniscule capacity (< 1 slot) make progress
-    // * If we're running over capacity, then its
-    //   (usedResources + required) (which extra resources we are allocating)
-    Resource queueCapacity =
-        Resources.multiplyAndNormalizeUp(resourceCalculator,
-            partitionResource,
-            queueCapacities.getAbsoluteCapacity(nodePartition),
-            minimumAllocation);
-
-    // Assume we have required resource equals to minimumAllocation, this can
-    // make sure user limit can continuously increase till queueMaxResource
-    // reached.
-    Resource required = minimumAllocation;
-
-    // Allow progress for queues with miniscule capacity
-    queueCapacity =
-        Resources.max(
-            resourceCalculator, partitionResource,
-            queueCapacity, 
-            required);
-
-
-    /* We want to base the userLimit calculation on
-     * max(queueCapacity, usedResources+required). However, we want
-     * usedResources to be based on the combined ratios of all the users in the
-     * queue so we use consumedRatio to calculate such.
-     * The calculation is dependent on how the resourceCalculator calculates the
-     * ratio between two Resources. DRF Example: If usedResources is
-     * greater than queueCapacity and users have the following [mem,cpu] usages:
-     * User1: [10%,20%] - Dominant resource is 20%
-     * User2: [30%,10%] - Dominant resource is 30%
-     * Then total consumedRatio is then 20+30=50%. Yes, this value can be
-     * larger than 100% but for the purposes of making sure all users are
-     * getting their fair share, it works.
-     */
-    Resource consumed = Resources.multiplyAndNormalizeUp(resourceCalculator,
-        partitionResource, qUsageRatios.getUsageRatio(nodePartition),
-        minimumAllocation);
-    Resource currentCapacity =
-        Resources.lessThan(resourceCalculator, partitionResource, consumed,
-            queueCapacity) ? queueCapacity : Resources.add(consumed, required);
-    // Never allow a single user to take more than the 
-    // queue's configured capacity * user-limit-factor.
-    // Also, the queue's configured capacity should be higher than 
-    // queue-hard-limit * ulMin
-    
-    final int activeUsers = activeUsersManager.getNumActiveUsers();
-    
-    // User limit resource is determined by:
-    // max{currentCapacity / #activeUsers, currentCapacity *
-    // user-limit-percentage%)
-    Resource userLimitResource = Resources.max(
-        resourceCalculator, partitionResource,
-        Resources.divideAndCeil(
-            resourceCalculator, currentCapacity, activeUsers),
-        Resources.divideAndCeil(
-            resourceCalculator, 
-            Resources.multiplyAndRoundDown(
-                currentCapacity, userLimit), 
-            100)
-        );
-    
-    // User limit is capped by maxUserLimit
-    // - maxUserLimit = queueCapacity * user-limit-factor (RESPECT_PARTITION_EXCLUSIVITY)
-    // - maxUserLimit = total-partition-resource (IGNORE_PARTITION_EXCLUSIVITY)
-    //
-    // In IGNORE_PARTITION_EXCLUSIVITY mode, if a queue cannot access a
-    // partition, its guaranteed resource on that partition is 0. And
-    // user-limit-factor computation is based on queue's guaranteed capacity. So
-    // we will not cap user-limit as well as used resource when doing
-    // IGNORE_PARTITION_EXCLUSIVITY allocation.
-    Resource maxUserLimit = Resources.none();
-    if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY) {
-      maxUserLimit =
-          Resources.multiplyAndRoundDown(queueCapacity, userLimitFactor);
-    } else if (schedulingMode == SchedulingMode.IGNORE_PARTITION_EXCLUSIVITY) {
-      maxUserLimit = partitionResource;
-    }
-    
-    // Cap final user limit with maxUserLimit
-    userLimitResource =
-        Resources.roundUp(
-            resourceCalculator, 
-            Resources.min(
-                resourceCalculator, partitionResource,
-                  userLimitResource,
-                  maxUserLimit
-                ), 
-            minimumAllocation);
+  /**
+   *
+   * @param userName
+   *          Name of user who has submitted one/more app to given queue.
+   * @param clusterResource
+   *          total cluster resource
+   * @param nodePartition
+   *          partition name
+   * @param schedulingMode
+   *          scheduling mode
+   *          RESPECT_PARTITION_EXCLUSIVITY/IGNORE_PARTITION_EXCLUSIVITY
+   * @return Computed User Limit
+   */
+  public Resource getResourceLimitForActiveUsers(String userName,
+      Resource clusterResource, String nodePartition,
+      SchedulingMode schedulingMode) {
+    return usersManager.getComputedResourceLimitForActiveUsers(userName,
+        clusterResource, nodePartition, schedulingMode);
+  }
 
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("User limit computation for " + userName +
-          " in queue " + getQueueName() +
-          " userLimitPercent=" + userLimit +
-          " userLimitFactor=" + userLimitFactor +
-          " required: " + required +
-          " consumed: " + consumed +
-          " user-limit-resource: " + userLimitResource +
-          " queueCapacity: " + queueCapacity +
-          " qconsumed: " + queueUsage.getUsed() +
-          " consumedRatio: " + totalUserConsumedRatio +
-          " currentCapacity: " + currentCapacity +
-          " activeUsers: " + activeUsers +
-          " clusterCapacity: " + clusterResource +
-          " resourceByLabel: " + partitionResource +
-          " usageratio: " + qUsageRatios.getUsageRatio(nodePartition) +
-          " Partition: " + nodePartition
-      );
-    }
-    user.setUserResourceLimit(userLimitResource);
-    return userLimitResource;
+  /**
+   *
+   * @param userName
+   *          Name of user who has submitted one/more app to given queue.
+   * @param clusterResource
+   *          total cluster resource
+   * @param nodePartition
+   *          partition name
+   * @param schedulingMode
+   *          scheduling mode
+   *          RESPECT_PARTITION_EXCLUSIVITY/IGNORE_PARTITION_EXCLUSIVITY
+   * @return Computed User Limit
+   */
+  public Resource getResourceLimitForAllUsers(String userName,
+      Resource clusterResource, String nodePartition,
+      SchedulingMode schedulingMode) {
+    return usersManager.getComputedResourceLimitForAllUsers(userName,
+        clusterResource, nodePartition, schedulingMode);
   }
-  
+
   @Private
   protected boolean canAssignToUser(Resource clusterResource,
       String userName, Resource limit, FiCaSchedulerApp application,
@@ -1620,52 +1501,34 @@ public class LeafQueue extends AbstractCSQueue {
     }
   }
 
-  private float calculateUserUsageRatio(Resource clusterResource,
-      String nodePartition) {
-    try {
-      writeLock.lock();
-      Resource resourceByLabel = labelManager.getResourceByLabel(nodePartition,
-          clusterResource);
-      float consumed = 0;
-      User user;
-      for (Map.Entry<String, User> entry : users.entrySet()) {
-        user = entry.getValue();
-        consumed += user.resetAndUpdateUsageRatio(resourceCalculator,
-            resourceByLabel, nodePartition);
-      }
-      return consumed;
-    } finally {
-      writeLock.unlock();
-    }
-  }
-
-  private void recalculateQueueUsageRatio(Resource clusterResource,
+  /**
+   * Recalculate QueueUsage Ratio.
+   *
+   * @param clusterResource
+   *          Total Cluster Resource
+   * @param nodePartition
+   *          Partition
+   */
+  public void recalculateQueueUsageRatio(Resource clusterResource,
       String nodePartition) {
     try {
       writeLock.lock();
-      ResourceUsage queueResourceUsage = this.getQueueResourceUsage();
+      ResourceUsage queueResourceUsage = getQueueResourceUsage();
 
       if (nodePartition == null) {
         for (String partition : Sets.union(
-            queueCapacities.getNodePartitionsSet(),
+            getQueueCapacities().getNodePartitionsSet(),
             queueResourceUsage.getNodePartitionsSet())) {
-          qUsageRatios.setUsageRatio(partition,
-              calculateUserUsageRatio(clusterResource, partition));
+          usersManager.updateUsageRatio(partition, clusterResource);
         }
-      } else{
-        qUsageRatios.setUsageRatio(nodePartition,
-            calculateUserUsageRatio(clusterResource, nodePartition));
+      } else {
+        usersManager.updateUsageRatio(nodePartition, clusterResource);
       }
     } finally {
       writeLock.unlock();
     }
   }
 
-  private void updateQueueUsageRatio(String nodePartition,
-      float delta) {
-    qUsageRatios.incUsageRatio(nodePartition, delta);
-  }
-
   @Override
   public void completedContainer(Resource clusterResource, 
       FiCaSchedulerApp application, FiCaSchedulerNode node, RMContainer rmContainer, 
@@ -1740,8 +1603,6 @@ public class LeafQueue extends AbstractCSQueue {
       writeLock.lock();
       super.allocateResource(clusterResource, resource, nodePartition,
           isIncreasedAllocation);
-      Resource resourceByLabel = labelManager.getResourceByLabel(nodePartition,
-          clusterResource);
 
       // handle ignore exclusivity container
       if (null != rmContainer && rmContainer.getNodeLabelExpression().equals(
@@ -1760,16 +1621,9 @@ public class LeafQueue extends AbstractCSQueue {
       // Update user metrics
       String userName = application.getUser();
 
-      // TODO, should use getUser, use this method just to avoid UT failure
-      // which is caused by wrong invoking order, will fix UT separately
-      User user = getUserAndAddIfAbsent(userName);
-
-      user.assignContainer(resource, nodePartition);
-
-      // Update usage ratios
-      updateQueueUsageRatio(nodePartition,
-          user.updateUsageRatio(resourceCalculator, resourceByLabel,
-              nodePartition));
+      // Increment user's resource usage.
+      User user = usersManager.updateUserResourceUsage(userName, resource,
+          nodePartition, true);
 
       // Note this is a bit unconventional since it gets the object and modifies
       // it here, rather then using set routine
@@ -1777,9 +1631,10 @@ public class LeafQueue extends AbstractCSQueue {
       metrics.setAvailableResourcesToUser(userName, application.getHeadroom());
 
       if (LOG.isDebugEnabled()) {
-        LOG.debug(getQueueName() + " user=" + userName + " used=" + queueUsage
-            .getUsed() + " numContainers=" + numContainers + " headroom = "
-            + application.getHeadroom() + " user-resources=" + user.getUsed());
+        LOG.debug(getQueueName() + " user=" + userName + " used="
+            + queueUsage.getUsed(nodePartition) + " numContainers="
+            + numContainers + " headroom = " + application.getHeadroom()
+            + " user-resources=" + user.getUsed());
       }
     } finally {
       writeLock.unlock();
@@ -1793,8 +1648,6 @@ public class LeafQueue extends AbstractCSQueue {
       writeLock.lock();
       super.releaseResource(clusterResource, resource, nodePartition,
           isChangeResource);
-      Resource resourceByLabel = labelManager.getResourceByLabel(nodePartition,
-          clusterResource);
 
       // handle ignore exclusivity container
       if (null != rmContainer && rmContainer.getNodeLabelExpression().equals(
@@ -1812,13 +1665,8 @@ public class LeafQueue extends AbstractCSQueue {
 
       // Update user metrics
       String userName = application.getUser();
-      User user = getUserAndAddIfAbsent(userName);
-      user.releaseContainer(resource, nodePartition);
-
-      // Update usage ratios
-      updateQueueUsageRatio(nodePartition,
-          user.updateUsageRatio(resourceCalculator, resourceByLabel,
-              nodePartition));
+      User user = usersManager.updateUserResourceUsage(userName, resource,
+          nodePartition, false);
 
       metrics.setAvailableResourcesToUser(userName, application.getHeadroom());
 
@@ -1877,6 +1725,10 @@ public class LeafQueue extends AbstractCSQueue {
       // activate the pending applications if possible
       activateApplications();
 
+      // In case of any resource change, invalidate recalculateULCount to clear
+      // the computed user-limit.
+      usersManager.userLimitNeedsRecompute();
+
       // Update application properties
       for (FiCaSchedulerApp application : orderingPolicy
           .getSchedulableEntities()) {
@@ -1892,16 +1744,16 @@ public class LeafQueue extends AbstractCSQueue {
   @Override
   public void incUsedResource(String nodeLabel, Resource resourceToInc,
       SchedulerApplicationAttempt application) {
-    getUser(application.getUser()).getResourceUsage().incUsed(nodeLabel,
-        resourceToInc);
+    usersManager.updateUserResourceUsage(application.getUser(), resourceToInc,
+        nodeLabel, true);
     super.incUsedResource(nodeLabel, resourceToInc, application);
   }
 
   @Override
   public void decUsedResource(String nodeLabel, Resource resourceToDec,
       SchedulerApplicationAttempt application) {
-    getUser(application.getUser()).getResourceUsage().decUsed(nodeLabel,
-        resourceToDec);
+    usersManager.updateUserResourceUsage(application.getUser(), resourceToDec,
+        nodeLabel, false);
     super.decUsedResource(nodeLabel, resourceToDec, application);
   }
 
@@ -1921,191 +1773,6 @@ public class LeafQueue extends AbstractCSQueue {
     queueUsage.decAMUsed(nodeLabel, resourceToDec);
   }
 
-  /*
-   * Usage Ratio
-   */
-  static private class UsageRatios {
-    private Map<String, Float> usageRatios;
-    private ReadLock readLock;
-    private WriteLock writeLock;
-
-    public UsageRatios() {
-      ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
-      readLock = lock.readLock();
-      writeLock = lock.writeLock();
-      usageRatios = new HashMap<String, Float>();
-    }
-
-    private void incUsageRatio(String label, float delta) {
-      try {
-        writeLock.lock();
-        Float fl = usageRatios.get(label);
-        if (null == fl) {
-          fl = new Float(0.0);
-        }
-        fl += delta;
-        usageRatios.put(label, new Float(fl));
-      } finally {
-        writeLock.unlock();
-      }
-    }
-
-    float getUsageRatio(String label) {
-      try {
-        readLock.lock();
-        Float f = usageRatios.get(label);
-        if (null == f) {
-          return 0.0f;
-        }
-        return f;
-      } finally {
-        readLock.unlock();
-      }
-    }
-
-    private void setUsageRatio(String label, float ratio) {
-      try {
-        writeLock.lock();
-        usageRatios.put(label, new Float(ratio));
-      } finally {
-        writeLock.unlock();
-      }
-    }
-  }
-
-  @VisibleForTesting
-  public float getUsageRatio(String label) {
-    return qUsageRatios.getUsageRatio(label);
-  }
-
-  @VisibleForTesting
-  public static class User {
-    ResourceUsage userResourceUsage = new ResourceUsage();
-    volatile Resource userResourceLimit = Resource.newInstance(0, 0);
-    volatile int pendingApplications = 0;
-    volatile int activeApplications = 0;
-    private UsageRatios userUsageRatios = new UsageRatios();
-    private WriteLock writeLock;
-
-    User() {
-      ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
-      // Nobody uses read-lock now, will add it when necessary
-      writeLock = lock.writeLock();
-    }
-
-    public ResourceUsage getResourceUsage() {
-      return userResourceUsage;
-    }
-    
-    public float resetAndUpdateUsageRatio(
-        ResourceCalculator resourceCalculator,
-        Resource resource, String nodePartition) {
-      try {
-        writeLock.lock();
-        userUsageRatios.setUsageRatio(nodePartition, 0);
-        return updateUsageRatio(resourceCalculator, resource, nodePartition);
-      } finally {
-        writeLock.unlock();
-      }
-    }
-
-    public float updateUsageRatio(
-        ResourceCalculator resourceCalculator,
-        Resource resource, String nodePartition) {
-      try {
-        writeLock.lock();
-        float delta;
-        float newRatio = Resources.ratio(resourceCalculator,
-            getUsed(nodePartition), resource);
-        delta = newRatio - userUsageRatios.getUsageRatio(nodePartition);
-        userUsageRatios.setUsageRatio(nodePartition, newRatio);
-        return delta;
-      } finally {
-        writeLock.unlock();
-      }
-    }
-
-    public Resource getUsed() {
-      return userResourceUsage.getUsed();
-    }
-
-    public Resource getAllUsed() {
-      return userResourceUsage.getAllUsed();
-    }
-
-    public Resource getUsed(String label) {
-      return userResourceUsage.getUsed(label);
-    }
-
-    public int getPendingApplications() {
-      return pendingApplications;
-    }
-
-    public int getActiveApplications() {
-      return activeApplications;
-    }
-    
-    public Resource getConsumedAMResources() {
-      return userResourceUsage.getAMUsed();
-    }
-
-    public Resource getConsumedAMResources(String label) {
-      return userResourceUsage.getAMUsed(label);
-    }
-
-    public int getTotalApplications() {
-      return getPendingApplications() + getActiveApplications();
-    }
-    
-    public void submitApplication() {
-      try {
-        writeLock.lock();
-        ++pendingApplications;
-      } finally {
-        writeLock.unlock();
-      }
-    }
-    
-    public void activateApplication() {
-      try {
-        writeLock.lock();
-        --pendingApplications;
-        ++activeApplications;
-      } finally {
-        writeLock.unlock();
-      }
-    }
-
-    public void finishApplication(boolean wasActive) {
-      try {
-        writeLock.lock();
-        if (wasActive) {
-          --activeApplications;
-        } else{
-          --pendingApplications;
-        }
-      } finally {
-        writeLock.unlock();
-      }
-    }
-
-    public void assignContainer(Resource resource, String nodePartition) {
-      userResourceUsage.incUsed(nodePartition, resource);
-    }
-
-    public void releaseContainer(Resource resource, String nodePartition) {
-      userResourceUsage.decUsed(nodePartition, resource);
-    }
-
-    public Resource getUserResourceLimit() {
-      return userResourceLimit;
-    }
-
-    public void setUserResourceLimit(Resource userResourceLimit) {
-      this.userResourceLimit = userResourceLimit;
-    }
-  }
-
   @Override
   public void recoverContainer(Resource clusterResource,
       SchedulerApplicationAttempt attempt, RMContainer rmContainer) {
@@ -2175,9 +1842,9 @@ public class LeafQueue extends AbstractCSQueue {
    *                                  excessive preemption.
    * @return Total pending resource considering user limit
    */
-
   public Resource getTotalPendingResourcesConsideringUserLimit(
-      Resource clusterResources, String partition, boolean deductReservedFromPending) {
+      Resource clusterResources, String partition,
+      boolean deductReservedFromPending) {
     try {
       readLock.lock();
       Map<String, Resource> userNameToHeadroom =
@@ -2188,8 +1855,8 @@ public class LeafQueue extends AbstractCSQueue {
         if (!userNameToHeadroom.containsKey(userName)) {
           User user = getUser(userName);
           Resource headroom = Resources.subtract(
-              computeUserLimit(app.getUser(), clusterResources, user, partition,
-                  SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY),
+              getResourceLimitForActiveUsers(app.getUser(), clusterResources,
+                  partition, SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY),
               user.getUsed(partition));
           // Make sure headroom is not negative.
           headroom = Resources.componentwiseMax(headroom, Resources.none());
@@ -2219,16 +1886,6 @@ public class LeafQueue extends AbstractCSQueue {
 
   }
 
-  public synchronized Resource getUserLimitPerUser(String userName,
-      Resource resources, String partition) {
-
-    // Check user resource limit
-    User user = getUser(userName);
-
-    return computeUserLimit(userName, resources, user, partition,
-        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
-  }
-
   @Override
   public void collectSchedulerApplications(
       Collection<ApplicationAttemptId> apps) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/5fb723bb/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/ParentQueue.java
----------------------------------------------------------------------
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/ParentQueue.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/ParentQueue.java
index 0e5b884..9c42c61 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/ParentQueue.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/ParentQueue.java
@@ -897,7 +897,7 @@ public class ParentQueue extends AbstractCSQueue {
   }
   
   @Override
-  public ActiveUsersManager getActiveUsersManager() {
+  public ActiveUsersManager getAbstractUsersManager() {
     // Should never be called since all applications are submitted to LeafQueues
     return null;
   }


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message