Merge -r 1433525:1433526 from trunk to branch-2. Fixes: YARN-336. Fair scheduler FIFO scheduling within a queue only allows 1 app at a time. Contributed by Sandy Ryza.

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

hadoop-yarn-project/CHANGES.txt

@@ -175,6 +175,9 @@ Release 2.0.3-alpha - Unreleased
     
     YARN-335. Fair scheduler doesn't check whether rack needs containers
     before assigning to node. (Sandy Ryza via tomwhite)
+    
+    YARN-336. Fair scheduler FIFO scheduling within a queue only allows 1
+    app at a time. (Sandy Ryza via tomwhite)
 
 Release 2.0.2-alpha - 2012-09-07 
 

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

@@ -275,7 +275,7 @@ public class AppSchedulable extends Schedulable {
       // The desired container won't fit here, so reserve
       reserve(application, priority, node, container, reserved);
 
-      return Resources.none();
+      return FairScheduler.CONTAINER_RESERVED;
     }
   }
 

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

@@ -177,7 +177,10 @@ public class FSLeafQueue extends FSQueue {
       Collections.sort(appScheds, comparator);
       for (AppSchedulable sched: appScheds) {
         if (sched.getRunnable()) {
-          return sched.assignContainer(node, reserved);
+          Resource assignedResource = sched.assignContainer(node, reserved);
+          if (!assignedResource.equals(Resources.none())) {
+            return assignedResource;
+          }
         }
       }
 

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

@@ -109,6 +109,10 @@ public class FairScheduler implements ResourceScheduler {
   private Clock clock;
 
   private static final Log LOG = LogFactory.getLog(FairScheduler.class);
+  
+  // Value that container assignment methods return when a container is
+  // reserved
+  public static final Resource CONTAINER_RESERVED = Resources.createResource(-1);
 
   // How often fair shares are re-calculated (ms)
   protected long UPDATE_INTERVAL = 500;

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

@@ -1317,4 +1317,44 @@ public class TestFairScheduler {
     // should assign rack local
     assertEquals(2, scheduler.applications.get(appId).getLiveContainers().size());
   }
+  
+  @Test
+  public void testFifoWithinQueue() throws Exception {
+    RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(3072));
+    NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
+    scheduler.handle(nodeEvent1);
+    
+    // Even if submitted at exact same time, apps will be deterministically
+    // ordered by name.
+    ApplicationAttemptId attId1 = createSchedulingRequest(1024, "queue1",
+        "user1", 2);
+    ApplicationAttemptId attId2 = createSchedulingRequest(1024, "queue1",
+        "user1", 2);
+    FSSchedulerApp app1 = scheduler.applications.get(attId1);
+    FSSchedulerApp app2 = scheduler.applications.get(attId2);
+    
+    FSLeafQueue queue1 = scheduler.getQueueManager().getLeafQueue("queue1");
+    queue1.setSchedulingMode(SchedulingMode.FIFO);
+    
+    scheduler.update();
+
+    // First two containers should go to app 1, third should go to app 2.
+    // Because tests set assignmultiple to false, each heartbeat assigns a single
+    // container.
+    
+    NodeUpdateSchedulerEvent updateEvent = new NodeUpdateSchedulerEvent(node1,
+      new ArrayList<ContainerStatus>(), new ArrayList<ContainerStatus>());
+
+    scheduler.handle(updateEvent);
+    assertEquals(1, app1.getLiveContainers().size());
+    assertEquals(0, app2.getLiveContainers().size());
+    
+    scheduler.handle(updateEvent);
+    assertEquals(2, app1.getLiveContainers().size());
+    assertEquals(0, app2.getLiveContainers().size());
+    
+    scheduler.handle(updateEvent);
+    assertEquals(2, app1.getLiveContainers().size());
+    assertEquals(1, app2.getLiveContainers().size());
+  }
 }