YARN-596. Use scheduling policies throughout the queue hierarchy to decide which containers to preempt (Wei Yan via Sandy Ryza)

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

Author: sryza

hadoop-yarn-project/CHANGES.txt

@@ -114,6 +114,9 @@ Release 2.5.0 - UNRELEASED
     YARN-2107. Refactored timeline classes into o.a.h.y.s.timeline package. (Vinod
     Kumar Vavilapalli via zjshen)
 
+    YARN-596. Use scheduling policies throughout the queue hierarchy to decide
+    which containers to preempt (Wei Yan via Sandy Ryza)
+
   OPTIMIZATIONS
 
   BUG FIXES 

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

@@ -18,8 +18,10 @@
 
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
 
+import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Comparator;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -31,8 +33,6 @@
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
-import org.apache.hadoop.yarn.factories.RecordFactory;
-import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
@@ -58,6 +58,8 @@ public class AppSchedulable extends Schedulable {
   private Priority priority;
   private ResourceWeights resourceWeights;
 
+  private RMContainerComparator comparator = new RMContainerComparator();
+
   public AppSchedulable(FairScheduler scheduler, FSSchedulerApp app, FSLeafQueue queue) {
     this.scheduler = scheduler;
     this.app = app;
@@ -111,7 +113,10 @@ public long getStartTime() {
 
   @Override
   public Resource getResourceUsage() {
-    return app.getCurrentConsumption();
+    // Here the getPreemptedResources() always return zero, except in
+    // a preemption round
+    return Resources.subtract(app.getCurrentConsumption(),
+        app.getPreemptedResources());
   }
 
 
@@ -383,6 +388,27 @@ public Resource assignContainer(FSSchedulerNode node) {
     return assignContainer(node, false);
   }
 
+  /**
+   * Preempt a running container according to the priority
+   */
+  @Override
+  public RMContainer preemptContainer() {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("App " + getName() + " is going to preempt a running " +
+          "container");
+    }
+
+    RMContainer toBePreempted = null;
+    for (RMContainer container : app.getLiveContainers()) {
+      if (! app.getPreemptionContainers().contains(container) &&
+          (toBePreempted == null ||
+              comparator.compare(toBePreempted, container) > 0)) {
+        toBePreempted = container;
+      }
+    }
+    return toBePreempted;
+  }
+
   /**
    * Whether this app has containers requests that could be satisfied on the
    * given node, if the node had full space.
@@ -407,4 +433,17 @@ public boolean hasContainerForNode(Priority prio, FSSchedulerNode node) {
         Resources.lessThanOrEqual(RESOURCE_CALCULATOR, null,
             anyRequest.getCapability(), node.getRMNode().getTotalCapability());
   }
+
+  static class RMContainerComparator implements Comparator<RMContainer>,
+      Serializable {
+    @Override
+    public int compare(RMContainer c1, RMContainer c2) {
+      int ret = c1.getContainer().getPriority().compareTo(
+          c2.getContainer().getPriority());
+      if (ret == 0) {
+        return c2.getContainerId().compareTo(c1.getContainerId());
+      }
+      return ret;
+      }
+    }
 }

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

@@ -33,10 +33,10 @@
 import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
 import org.apache.hadoop.yarn.api.records.Resource;
+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.SchedulerAppUtils;
 import org.apache.hadoop.yarn.util.resource.Resources;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplication;
 
 @Private
 @Unstable
@@ -208,6 +208,36 @@ public Resource assignContainer(FSSchedulerNode node) {
     return assigned;
   }
 
+  @Override
+  public RMContainer preemptContainer() {
+    RMContainer toBePreempted = null;
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Queue " + getName() + " is going to preempt a container " +
+          "from its applications.");
+    }
+
+    // If this queue is not over its fair share, reject
+    if (!preemptContainerPreCheck()) {
+      return toBePreempted;
+    }
+
+    // Choose the app that is most over fair share
+    Comparator<Schedulable> comparator = policy.getComparator();
+    AppSchedulable candidateSched = null;
+    for (AppSchedulable sched : runnableAppScheds) {
+      if (candidateSched == null ||
+          comparator.compare(sched, candidateSched) > 0) {
+        candidateSched = sched;
+      }
+    }
+
+    // Preempt from the selected app
+    if (candidateSched != null) {
+      toBePreempted = candidateSched.preemptContainer();
+    }
+    return toBePreempted;
+  }
+
   @Override
   public List<FSQueue> getChildQueues() {
     return new ArrayList<FSQueue>(1);

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

@@ -21,6 +21,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 
 import org.apache.commons.logging.Log;
@@ -32,6 +33,7 @@
 import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
 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.scheduler.ActiveUsersManager;
 
@@ -156,6 +158,32 @@ public Resource assignContainer(FSSchedulerNode node) {
     return assigned;
   }
 
+  @Override
+  public RMContainer preemptContainer() {
+    RMContainer toBePreempted = null;
+
+    // If this queue is not over its fair share, reject
+    if (!preemptContainerPreCheck()) {
+      return toBePreempted;
+    }
+
+    // Find the childQueue which is most over fair share
+    FSQueue candidateQueue = null;
+    Comparator<Schedulable> comparator = policy.getComparator();
+    for (FSQueue queue : childQueues) {
+      if (candidateQueue == null ||
+          comparator.compare(queue, candidateQueue) > 0) {
+        candidateQueue = queue;
+      }
+    }
+
+    // Let the selected queue choose which of its container to preempt
+    if (candidateQueue != null) {
+      toBePreempted = candidateQueue.preemptContainer();
+    }
+    return toBePreempted;
+  }
+
   @Override
   public List<FSQueue> getChildQueues() {
     return childQueues;

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

@@ -187,4 +187,17 @@ protected boolean assignContainerPreCheck(FSSchedulerNode node) {
     }
     return true;
   }
+
+  /**
+   * Helper method to check if the queue should preempt containers
+   *
+   * @return true if check passes (can preempt) or false otherwise
+   */
+  protected boolean preemptContainerPreCheck() {
+    if (this == scheduler.getQueueManager().getRootQueue()) {
+      return true;
+    }
+    return parent.getPolicy()
+        .checkIfUsageOverFairShare(getResourceUsage(), getFairShare());
+  }
 }

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

@@ -59,6 +59,8 @@ public class FSSchedulerApp extends SchedulerApplicationAttempt {
   private AppSchedulable appSchedulable;
 
   final Map<RMContainer, Long> preemptionMap = new HashMap<RMContainer, Long>();
+
+  private Resource preemptedResources = Resources.createResource(0);
 
   public FSSchedulerApp(ApplicationAttemptId applicationAttemptId, 
       String user, FSLeafQueue queue, ActiveUsersManager activeUsersManager,
@@ -316,6 +318,7 @@ public synchronized void resetAllowedLocalityLevel(Priority priority,
   public void addPreemption(RMContainer container, long time) {
     assert preemptionMap.get(container) == null;
     preemptionMap.put(container, time);
+    Resources.addTo(preemptedResources, container.getAllocatedResource());
   }
 
   public Long getContainerPreemptionTime(RMContainer container) {
@@ -330,4 +333,20 @@ public Set<RMContainer> getPreemptionContainers() {
   public FSLeafQueue getQueue() {
     return (FSLeafQueue)super.getQueue();
   }
+
+  public Resource getPreemptedResources() {
+    return preemptedResources;
+  }
+
+  public void resetPreemptedResources() {
+    preemptedResources = Resources.createResource(0);
+    for (RMContainer container : getPreemptionContainers()) {
+      Resources.addTo(preemptedResources, container.getAllocatedResource());
+    }
+  }
+
+  public void clearPreemptedResources() {
+    preemptedResources.setMemory(0);
+    preemptedResources.setVirtualCores(0);
+  }
 }