YARN-2393. FairScheduler: Add the notion of steady fair share. (Wei Yan via kasha)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1619845 13f79535-47bb-0310-9956-ffa450edef68

Author: kambatla

hadoop-yarn-project/CHANGES.txt

@@ -55,6 +55,9 @@ Release 2.6.0 - UNRELEASED
     YARN-2174. Enable HTTPs for the writer REST API of TimelineServer.
     (Zhijie Shen via jianhe)
 
+    YARN-2393. FairScheduler: Add the notion of steady fair share. 
+    (Wei Yan via kasha)
+
   IMPROVEMENTS
 
     YARN-2197. Add a link to YARN CHANGES.txt in the left side of doc

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java

@@ -717,12 +717,6 @@ public void setFairShare(Resource fairShare) {
     this.fairShare = fairShare;
   }
 
-  @Override
-  public boolean isActive() {
-    return true;
-  }
-
-
   @Override
   public void updateDemand() {
     demand = Resources.createResource(0);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSParentQueue.java

@@ -35,7 +35,6 @@
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.util.resource.Resources;
-import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
 
@@ -68,6 +67,16 @@ public void recomputeShares() {
     }
   }
 
+  public void recomputeSteadyShares() {
+    policy.computeSteadyShares(childQueues, getSteadyFairShare());
+    for (FSQueue childQueue : childQueues) {
+      childQueue.getMetrics().setSteadyFairShare(childQueue.getSteadyFairShare());
+      if (childQueue instanceof FSParentQueue) {
+        ((FSParentQueue) childQueue).recomputeSteadyShares();
+      }
+    }
+  }
+
   @Override
   public Resource getDemand() {
     return demand;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueue.java

@@ -41,6 +41,7 @@
 @Unstable
 public abstract class FSQueue implements Queue, Schedulable {
   private Resource fairShare = Resources.createResource(0, 0);
+  private Resource steadyFairShare = Resources.createResource(0, 0);
   private final String name;
   protected final FairScheduler scheduler;
   private final FSQueueMetrics metrics;
@@ -151,7 +152,17 @@ public void setFairShare(Resource fairShare) {
     this.fairShare = fairShare;
     metrics.setFairShare(fairShare);
   }
-  
+
+  /** Get the steady fair share assigned to this Schedulable. */
+  public Resource getSteadyFairShare() {
+    return steadyFairShare;
+  }
+
+  public void setSteadyFairShare(Resource steadyFairShare) {
+    this.steadyFairShare = steadyFairShare;
+    metrics.setSteadyFairShare(steadyFairShare);
+  }
+
   public boolean hasAccess(QueueACL acl, UserGroupInformation user) {
     return scheduler.getAllocationConfiguration().hasAccess(name, acl, user);
   }
@@ -161,7 +172,7 @@ public boolean hasAccess(QueueACL acl, UserGroupInformation user) {
    * queue's current share
    */
   public abstract void recomputeShares();
-  
+
   /**
    * Gets the children of this queue, if any.
    */
@@ -194,7 +205,9 @@ protected boolean assignContainerPreCheck(FSSchedulerNode node) {
     return true;
   }
 
-  @Override
+  /**
+   * Returns true if queue has at least one app running.
+   */
   public boolean isActive() {
     return getNumRunnableApps() > 0;
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSQueueMetrics.java

@@ -33,6 +33,8 @@ public class FSQueueMetrics extends QueueMetrics {
 
   @Metric("Fair share of memory in MB") MutableGaugeInt fairShareMB;
   @Metric("Fair share of CPU in vcores") MutableGaugeInt fairShareVCores;
+  @Metric("Steady fair share of memory in MB") MutableGaugeInt steadyFairShareMB;
+  @Metric("Steady fair share of CPU in vcores") MutableGaugeInt steadyFairShareVCores;
   @Metric("Minimum share of memory in MB") MutableGaugeInt minShareMB;
   @Metric("Minimum share of CPU in vcores") MutableGaugeInt minShareVCores;
   @Metric("Maximum share of memory in MB") MutableGaugeInt maxShareMB;
@@ -55,7 +57,20 @@ public int getFairShareMB() {
   public int getFairShareVirtualCores() {
     return fairShareVCores.value();
   }
-  
+
+  public void setSteadyFairShare(Resource resource) {
+    steadyFairShareMB.set(resource.getMemory());
+    steadyFairShareVCores.set(resource.getVirtualCores());
+  }
+
+  public int getSteadyFairShareMB() {
+    return steadyFairShareMB.value();
+  }
+
+  public int getSteadyFairShareVCores() {
+    return steadyFairShareVCores.value();
+  }
+
   public void setMinShare(Resource resource) {
     minShareMB.set(resource.getMemory());
     minShareVCores.set(resource.getVirtualCores());

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java

@@ -851,6 +851,8 @@ private synchronized void addNode(RMNode node) {
     Resources.addTo(clusterResource, node.getTotalCapability());
     updateRootQueueMetrics();
 
+    queueMgr.getRootQueue().setSteadyFairShare(clusterResource);
+    queueMgr.getRootQueue().recomputeSteadyShares();
     LOG.info("Added node " + node.getNodeAddress() +
         " cluster capacity: " + clusterResource);
   }
@@ -885,6 +887,8 @@ private synchronized void removeNode(RMNode rmNode) {
     }
 
     nodes.remove(rmNode.getNodeID());
+    queueMgr.getRootQueue().setSteadyFairShare(clusterResource);
+    queueMgr.getRootQueue().recomputeSteadyShares();
     LOG.info("Removed node " + rmNode.getNodeAddress() +
         " cluster capacity: " + clusterResource);
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/QueueManager.java

@@ -118,6 +118,11 @@ private FSQueue getQueue(String name, boolean create, FSQueueType queueType) {
       if (queue == null && create) {
         // if the queue doesn't exist,create it and return
         queue = createQueue(name, queueType);
+
+        // Update steady fair share for all queues
+        if (queue != null) {
+          rootQueue.recomputeSteadyShares();
+        }
       }
       return queue;
     }
@@ -190,7 +195,7 @@ private FSQueue createQueue(String name, FSQueueType queueType) {
         parent = newParent;
       }
     }
-    
+
     return parent;
   }
 
@@ -376,5 +381,8 @@ public void updateAllocationConfiguration(AllocationConfiguration queueConf) {
             + queue.getName(), ex);
       }
     }
+
+    // Update steady fair shares for all queues
+    rootQueue.recomputeSteadyShares();
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/Schedulable.java

@@ -24,7 +24,6 @@
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
-import org.apache.hadoop.yarn.util.resource.Resources;
 
 /**
  * A Schedulable represents an entity that can be scheduled such as an
@@ -102,10 +101,4 @@ public interface Schedulable {
 
   /** Assign a fair share to this Schedulable. */
   public void setFairShare(Resource fairShare);
-
-  /**
-   * Returns true if queue has atleast one app running. Always returns true for
-   * AppSchedulables.
-   */
-  public boolean isActive();
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/SchedulingPolicy.java

@@ -17,10 +17,6 @@
  */
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
 
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.concurrent.ConcurrentHashMap;
-
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Evolving;
 import org.apache.hadoop.util.ReflectionUtils;
@@ -29,6 +25,10 @@
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FairSharePolicy;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
 
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.concurrent.ConcurrentHashMap;
+
 @Public
 @Evolving
 public abstract class SchedulingPolicy {
@@ -131,15 +131,30 @@ public static boolean isApplicableTo(SchedulingPolicy policy, byte depth) {
   public abstract Comparator<Schedulable> getComparator();
 
   /**
-   * Computes and updates the shares of {@link Schedulable}s as per the
-   * {@link SchedulingPolicy}, to be used later at schedule time.
+   * Computes and updates the shares of {@link Schedulable}s as per
+   * the {@link SchedulingPolicy}, to be used later for scheduling decisions.
+   * The shares computed are instantaneous and only consider queues with
+   * running applications.
    * 
    * @param schedulables {@link Schedulable}s whose shares are to be updated
    * @param totalResources Total {@link Resource}s in the cluster
    */
   public abstract void computeShares(
       Collection<? extends Schedulable> schedulables, Resource totalResources);
 
+  /**
+   * Computes and updates the steady shares of {@link FSQueue}s as per the
+   * {@link SchedulingPolicy}. The steady share does not differentiate
+   * between queues with and without running applications under them. The
+   * steady share is not used for scheduling, it is displayed on the Web UI
+   * for better visibility.
+   *
+   * @param queues {@link FSQueue}s whose shares are to be updated
+   * @param totalResources Total {@link Resource}s in the cluster
+   */
+  public abstract void computeSteadyShares(
+      Collection<? extends FSQueue> queues, Resource totalResources);
+
   /**
    * Check if the resource usage is over the fair share under this policy
    *

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/policies/ComputeFairShares.java

@@ -22,6 +22,7 @@
 
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceType;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSQueue;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.Schedulable;
 
 /**
@@ -49,14 +50,29 @@ public static void computeShares(
       ResourceType type) {
     Collection<Schedulable> activeSchedulables = new ArrayList<Schedulable>();
     for (Schedulable sched : schedulables) {
-      if (sched.isActive()) {
-        activeSchedulables.add(sched);
-      } else {
+      if ((sched instanceof FSQueue) && !((FSQueue) sched).isActive()) {
         setResourceValue(0, sched.getFairShare(), type);
+      } else {
+        activeSchedulables.add(sched);
       }
     }
 
-    computeSharesInternal(activeSchedulables, totalResources, type);
+    computeSharesInternal(activeSchedulables, totalResources, type, false);
+  }
+
+  /**
+   * Compute the steady fair share of the given queues. The steady fair
+   * share is an allocation of shares considering all queues, i.e.,
+   * active and inactive.
+   *
+   * @param queues
+   * @param totalResources
+   * @param type
+   */
+  public static void computeSteadyShares(
+      Collection<? extends FSQueue> queues, Resource totalResources,
+      ResourceType type) {
+    computeSharesInternal(queues, totalResources, type, true);
   }
 
   /**
@@ -102,7 +118,7 @@ public static void computeShares(
    */
   private static void computeSharesInternal(
       Collection<? extends Schedulable> schedulables, Resource totalResources,
-      ResourceType type) {
+      ResourceType type, boolean isSteadyShare) {
     if (schedulables.isEmpty()) {
       return;
     }
@@ -145,7 +161,13 @@ private static void computeSharesInternal(
     }
     // Set the fair shares based on the value of R we've converged to
     for (Schedulable sched : schedulables) {
-      setResourceValue(computeShare(sched, right, type), sched.getFairShare(), type);
+      if (isSteadyShare) {
+        setResourceValue(computeShare(sched, right, type),
+            ((FSQueue) sched).getSteadyFairShare(), type);
+      } else {
+        setResourceValue(
+            computeShare(sched, right, type), sched.getFairShare(), type);
+      }
     }
   }