YARN-789. Enable zero capabilities resource requests in fair scheduler. (tucu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1493220 13f79535-47bb-0310-9956-ffa450edef68

hadoop-yarn-project/CHANGES.txt

@@ -326,6 +326,9 @@ Release 2.1.0-beta - UNRELEASED
     YARN-803. factor out scheduler config validation from the ResourceManager 
     to each scheduler implementation. (tucu)
 
+    YARN-789. Enable zero capabilities resource requests in fair scheduler. 
+    (tucu)
+
   OPTIMIZATIONS
 
     YARN-512. Log aggregation root directory check is more expensive than it

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/resource/DefaultResourceCalculator.java

@@ -54,19 +54,25 @@ public class DefaultResourceCalculator extends ResourceCalculator {
 
   @Override
   public Resource normalize(Resource r, Resource minimumResource,
-      Resource maximumResource) {
+      Resource maximumResource, Resource stepFactor) {
     int normalizedMemory = Math.min(
         roundUp(
             Math.max(r.getMemory(), minimumResource.getMemory()),
-            minimumResource.getMemory()),
+            stepFactor.getMemory()),
             maximumResource.getMemory());
     return Resources.createResource(normalizedMemory);
   }
 
   @Override
-  public Resource roundUp(Resource r, Resource minimumResource) {
+  public Resource normalize(Resource r, Resource minimumResource,
+                            Resource maximumResource) {
+    return normalize(r, minimumResource, maximumResource, minimumResource);
+  }
+
+  @Override
+  public Resource roundUp(Resource r, Resource stepFactor) {
     return Resources.createResource(
-        roundUp(r.getMemory(),minimumResource.getMemory())
+        roundUp(r.getMemory(), stepFactor.getMemory())
         );
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/resource/DominantResourceCalculator.java

@@ -124,26 +124,26 @@ public class DominantResourceCalculator extends ResourceCalculator {
 
   @Override
   public Resource normalize(Resource r, Resource minimumResource,
-      Resource maximumResource) {
+                            Resource maximumResource, Resource stepFactor) {
     int normalizedMemory = Math.min(
-        roundUp(
-            Math.max(r.getMemory(), minimumResource.getMemory()),
-            minimumResource.getMemory()),
-            maximumResource.getMemory());
+      roundUp(
+        Math.max(r.getMemory(), minimumResource.getMemory()),
+        stepFactor.getMemory()),
+      maximumResource.getMemory());
     int normalizedCores = Math.min(
-        roundUp(
-            Math.max(r.getVirtualCores(), minimumResource.getVirtualCores()),
-            minimumResource.getVirtualCores()),
-            maximumResource.getVirtualCores());
+      roundUp(
+        Math.max(r.getVirtualCores(), minimumResource.getVirtualCores()),
+        stepFactor.getVirtualCores()),
+      maximumResource.getVirtualCores());
     return Resources.createResource(normalizedMemory,
-        normalizedCores);
+      normalizedCores);
   }
 
   @Override
-  public Resource roundUp(Resource r, Resource minimumResource) {
+  public Resource roundUp(Resource r, Resource stepFactor) {
     return Resources.createResource(
-        roundUp(r.getMemory(), minimumResource.getMemory()), 
-        roundUp(r.getVirtualCores(), minimumResource.getVirtualCores())
+        roundUp(r.getMemory(), stepFactor.getMemory()), 
+        roundUp(r.getVirtualCores(), stepFactor.getVirtualCores())
         );
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/resource/ResourceCalculator.java

@@ -96,8 +96,26 @@ public abstract class ResourceCalculator {
    * @param maximumResource the upper bound of the resource to be allocated
    * @return normalized resource
    */
+  public Resource normalize(Resource r, Resource minimumResource,
+      Resource maximumResource) {
+    return normalize(r, minimumResource, maximumResource, minimumResource);
+  }
+
+  /**
+   * Normalize resource <code>r</code> given the base 
+   * <code>minimumResource</code> and verify against max allowed
+   * <code>maximumResource</code> using a step factor for hte normalization.
+   *
+   * @param r resource
+   * @param minimumResource minimum value
+   * @param maximumResource the upper bound of the resource to be allocated
+   * @param stepFactor the increment for resources to be allocated
+   * @return normalized resource
+   */
   public abstract Resource normalize(Resource r, Resource minimumResource,
-      Resource maximumResource);
+                                     Resource maximumResource, 
+                                     Resource stepFactor);
+
 
   /**
    * Round-up resource <code>r</code> given factor <code>stepFactor</code>.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/resource/Resources.java

@@ -132,9 +132,9 @@ public class Resources {
   }
   
   public static Resource normalize(
-      ResourceCalculator calculator, Resource lhs, Resource factor,
-      Resource limit) {
-    return calculator.normalize(lhs, factor, limit);
+      ResourceCalculator calculator, Resource lhs, Resource min,
+      Resource max, Resource increment) {
+    return calculator.normalize(lhs, min, max, increment);
   }
   
   public static Resource roundUp(

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

@@ -82,6 +82,40 @@ public class SchedulerUtils {
     return containerStatus;
   }
 
+  /**
+   * Utility method to normalize a list of resource requests, by insuring that
+   * the memory for each request is a multiple of minMemory and is not zero.
+   */
+  public static void normalizeRequests(
+    List<ResourceRequest> asks,
+    ResourceCalculator resourceCalculator,
+    Resource clusterResource,
+    Resource minimumResource,
+    Resource maximumResource) {
+    for (ResourceRequest ask : asks) {
+      normalizeRequest(
+        ask, resourceCalculator, clusterResource, minimumResource,
+        maximumResource, minimumResource);
+    }
+  }
+
+  /**
+   * Utility method to normalize a resource request, by insuring that the
+   * requested memory is a multiple of minMemory and is not zero.
+   */
+  public static void normalizeRequest(
+    ResourceRequest ask,
+    ResourceCalculator resourceCalculator,
+    Resource clusterResource,
+    Resource minimumResource,
+    Resource maximumResource) {
+    Resource normalized =
+      Resources.normalize(
+        resourceCalculator, ask.getCapability(), minimumResource,
+        maximumResource, minimumResource);
+    ask.setCapability(normalized);
+  }
+
   /**
    * Utility method to normalize a list of resource requests, by insuring that
    * the memory for each request is a multiple of minMemory and is not zero.
@@ -91,11 +125,12 @@ public class SchedulerUtils {
       ResourceCalculator resourceCalculator, 
       Resource clusterResource,
       Resource minimumResource,
-      Resource maximumResource) {
+      Resource maximumResource,
+      Resource incrementResource) {
     for (ResourceRequest ask : asks) {
       normalizeRequest(
           ask, resourceCalculator, clusterResource, minimumResource,
-          maximumResource);
+          maximumResource, incrementResource);
     }
   }
 
@@ -108,11 +143,12 @@ public class SchedulerUtils {
       ResourceCalculator resourceCalculator, 
       Resource clusterResource,
       Resource minimumResource,
-      Resource maximumResource) {
+      Resource maximumResource,
+      Resource incrementResource) {
     Resource normalized = 
         Resources.normalize(
             resourceCalculator, ask.getCapability(), minimumResource,
-            maximumResource);
+            maximumResource, incrementResource);
     ask.setCapability(normalized);
   }
 

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

@@ -116,6 +116,7 @@ public class FairScheduler implements ResourceScheduler {
   private RMContext rmContext;
   private Resource minimumAllocation;
   private Resource maximumAllocation;
+  private Resource incrAllocation;
   private QueueManager queueMgr;
   private Clock clock;
 
@@ -300,7 +301,7 @@ public class FairScheduler implements ResourceScheduler {
    */
   boolean isStarvedForMinShare(FSLeafQueue sched) {
     Resource desiredShare = Resources.min(RESOURCE_CALCULATOR, clusterCapacity,
-        sched.getMinShare(), sched.getDemand());
+      sched.getMinShare(), sched.getDemand());
     return Resources.lessThan(RESOURCE_CALCULATOR, clusterCapacity,
         sched.getResourceUsage(), desiredShare);
   }
@@ -560,6 +561,10 @@ public class FairScheduler implements ResourceScheduler {
     return minimumAllocation;
   }
 
+  public Resource getIncrementResourceCapability() {
+    return incrAllocation;
+  }
+
   @Override
   public Resource getMaximumResourceCapability() {
     return maximumAllocation;
@@ -769,7 +774,7 @@ public class FairScheduler implements ResourceScheduler {
 
     // Sanity check
     SchedulerUtils.normalizeRequests(ask, new DominantResourceCalculator(),
-        clusterCapacity, minimumAllocation, maximumAllocation);
+        clusterCapacity, minimumAllocation, maximumAllocation, incrAllocation);
 
     // Release containers
     for (ContainerId releasedContainerId : release) {
@@ -1028,6 +1033,7 @@ public class FairScheduler implements ResourceScheduler {
     validateConf(this.conf);
     minimumAllocation = this.conf.getMinimumAllocation();
     maximumAllocation = this.conf.getMaximumAllocation();
+    incrAllocation = this.conf.getIncrementAllocation();
     userAsDefaultQueue = this.conf.getUserAsDefaultQueue();
     nodeLocalityThreshold = this.conf.getLocalityThresholdNode();
     rackLocalityThreshold = this.conf.getLocalityThresholdRack();

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

@@ -32,6 +32,16 @@ import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 @Private
 @Evolving
 public class FairSchedulerConfiguration extends Configuration {
+
+  /** Increment request grant-able by the RM scheduler. 
+   * These properties are looked up in the yarn-site.xml  */
+  public static final String RM_SCHEDULER_INCREMENT_ALLOCATION_MB =
+    YarnConfiguration.YARN_PREFIX + "scheduler.increment-allocation-mb";
+  public static final int DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_MB = 1024;
+  public static final String RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES =
+    YarnConfiguration.YARN_PREFIX + "scheduler.increment-allocation-vcores";
+  public static final int DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES = 1;
+  
   public static final String FS_CONFIGURATION_FILE = "fair-scheduler.xml";
 
   private static final String CONF_PREFIX =  "yarn.scheduler.fair.";
@@ -102,6 +112,16 @@ public class FairSchedulerConfiguration extends Configuration {
     return Resources.createResource(mem, cpu);
   }
 
+  public Resource getIncrementAllocation() {
+    int incrementMemory = getInt(
+      RM_SCHEDULER_INCREMENT_ALLOCATION_MB,
+      DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_MB);
+    int incrementCores = getInt(
+      RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES,
+      DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES);
+    return Resources.createResource(incrementMemory, incrementCores);
+  }
+
   public boolean getUserAsDefaultQueue() {
     return getBoolean(USER_AS_DEFAULT_QUEUE, DEFAULT_USER_AS_DEFAULT_QUEUE);
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java

@@ -62,6 +62,7 @@ import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService;
 import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
@@ -81,6 +82,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeRemoved
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.DominantResourceFairnessPolicy;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
+import org.apache.hadoop.yarn.server.resourcemanager.security.ClientToAMTokenSecretManagerInRM;
+import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.junit.After;
 import org.junit.Before;
@@ -120,7 +123,9 @@ public class TestFairScheduler {
   public void setUp() throws IOException {
     scheduler = new FairScheduler();
     Configuration conf = createConfiguration();
-    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 1024);
+    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 0);
+    conf.setInt(FairSchedulerConfiguration.RM_SCHEDULER_INCREMENT_ALLOCATION_MB,
+      1024);
     conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, 10240);
     // All tests assume only one assignment per node update
     conf.set(FairSchedulerConfiguration.ASSIGN_MULTIPLE, "false");
@@ -281,6 +286,8 @@ public class TestFairScheduler {
     conf.setFloat(FairSchedulerConfiguration.LOCALITY_THRESHOLD_RACK, .7f);
     conf.setInt(YarnConfiguration.RM_SCHEDULER_MAXIMUM_ALLOCATION_MB, 1024);
     conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 512);
+    conf.setInt(FairSchedulerConfiguration.RM_SCHEDULER_INCREMENT_ALLOCATION_MB, 
+      128);
     scheduler.reinitialize(conf, resourceManager.getRMContext());
     Assert.assertEquals(true, scheduler.assignMultiple);
     Assert.assertEquals(3, scheduler.maxAssign);
@@ -289,8 +296,44 @@ public class TestFairScheduler {
     Assert.assertEquals(.7, scheduler.rackLocalityThreshold, .01);
     Assert.assertEquals(1024, scheduler.getMaximumResourceCapability().getMemory());
     Assert.assertEquals(512, scheduler.getMinimumResourceCapability().getMemory());
+    Assert.assertEquals(128, 
+      scheduler.getIncrementResourceCapability().getMemory());
   }
   
+  @Test  
+  public void testNonMinZeroResourcesSettings() throws IOException {
+    FairScheduler fs = new FairScheduler();
+    YarnConfiguration conf = new YarnConfiguration();
+    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 256);
+    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES, 1);
+    conf.setInt(
+      FairSchedulerConfiguration.RM_SCHEDULER_INCREMENT_ALLOCATION_MB, 512);
+    conf.setInt(
+      FairSchedulerConfiguration.RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES, 2);
+    fs.reinitialize(conf, null);
+    Assert.assertEquals(256, fs.getMinimumResourceCapability().getMemory());
+    Assert.assertEquals(1, fs.getMinimumResourceCapability().getVirtualCores());
+    Assert.assertEquals(512, fs.getIncrementResourceCapability().getMemory());
+    Assert.assertEquals(2, fs.getIncrementResourceCapability().getVirtualCores());
+  }  
+  
+  @Test  
+  public void testMinZeroResourcesSettings() throws IOException {  
+    FairScheduler fs = new FairScheduler();  
+    YarnConfiguration conf = new YarnConfiguration();
+    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 0);
+    conf.setInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_VCORES, 0);
+    conf.setInt(
+      FairSchedulerConfiguration.RM_SCHEDULER_INCREMENT_ALLOCATION_MB, 512);
+    conf.setInt(
+      FairSchedulerConfiguration.RM_SCHEDULER_INCREMENT_ALLOCATION_VCORES, 2);
+    fs.reinitialize(conf, null);  
+    Assert.assertEquals(0, fs.getMinimumResourceCapability().getMemory());  
+    Assert.assertEquals(0, fs.getMinimumResourceCapability().getVirtualCores());
+    Assert.assertEquals(512, fs.getIncrementResourceCapability().getMemory());
+    Assert.assertEquals(2, fs.getIncrementResourceCapability().getVirtualCores());
+  }  
+  
   @Test
   public void testAggregateCapacityTracking() throws Exception {
     // Add a node
@@ -412,8 +455,8 @@ public class TestFairScheduler {
     NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1);
     scheduler.handle(updateEvent);
 
-    // Asked for less than min_allocation.
-    assertEquals(YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB,
+    // Asked for less than increment allocation.
+    assertEquals(FairSchedulerConfiguration.DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_MB,
         scheduler.getQueueManager().getQueue("queue1").
         getResourceUsage().getMemory());
 
@@ -571,7 +614,8 @@ public class TestFairScheduler {
     ApplicationAttemptId id22 = createAppAttemptId(2, 2);
     scheduler.addApplication(id22, "root.queue2", "user1");
 
-    int minReqSize = YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB;
+    int minReqSize = 
+        FairSchedulerConfiguration.DEFAULT_RM_SCHEDULER_INCREMENT_ALLOCATION_MB;
     
     // First ask, queue1 requests 1 large (minReqSize * 2).
     List<ResourceRequest> ask1 = new ArrayList<ResourceRequest>();