YARN-23. FairScheduler: FSQueueSchedulable#updateDemand() - potential redundant aggregation. (kkambatl via tucu)

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

hadoop-yarn-project/CHANGES.txt

@@ -43,6 +43,9 @@ Release 2.0.3-alpha - Unreleased
     YARN-116. Add the ability to change the RM include/exclude file without
     a restart. (xieguiming and Harsh J via sseth)
 
+    YARN-23. FairScheduler: FSQueueSchedulable#updateDemand() - potential 
+    redundant aggregation. (kkambatl via tucu)
+
   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/FSQueueSchedulable.java

@@ -65,6 +65,17 @@ public class FSQueueSchedulable extends Schedulable implements Queue {
   long lastTimeAtMinShare;
   long lastTimeAtHalfFairShare;
 
+  // Constructor for tests
+  protected FSQueueSchedulable(FairScheduler scheduler, FSQueue fsQueue,
+      QueueManager qMgr, QueueMetrics metrics, long minShare, long fairShare) {
+    this.scheduler = scheduler;
+    this.queueMgr = qMgr;
+    this.queue = fsQueue;
+    this.metrics = metrics;
+    this.lastTimeAtMinShare = minShare;
+    this.lastTimeAtHalfFairShare = fairShare;
+  }
+
   public FSQueueSchedulable(FairScheduler scheduler, FSQueue queue) {
     this.scheduler = scheduler;
     this.queue = queue;
@@ -93,19 +104,27 @@ public class FSQueueSchedulable extends Schedulable implements Queue {
    */
   @Override
   public void updateDemand() {
+    // Compute demand by iterating through apps in the queue
+    // Limit demand to maxResources
+    Resource maxRes = queueMgr.getMaxResources(queue.getName());
     demand = Resources.createResource(0);
     for (AppSchedulable sched: appScheds) {
       sched.updateDemand();
       Resource toAdd = sched.getDemand();
-      LOG.debug("Counting resource from " + sched.getName() + " " + toAdd.toString());
-      LOG.debug("Total resource consumption for " + this.getName() + " now " + demand.toString());
-      demand = Resources.add(demand, toAdd);
-
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Counting resource from " + sched.getName() + " " + toAdd
+            + "; Total resource consumption for " + this.getName() + " now "
+            + demand);
       }
-    // if demand exceeds the cap for this queue, limit to the max
-    Resource maxRes = queueMgr.getMaxResources(queue.getName());
-    if(Resources.greaterThan(demand, maxRes)) {
+      demand = Resources.add(demand, toAdd);
+      if (Resources.greaterThanOrEqual(demand, maxRes)) {
         demand = maxRes;
+        break;
+      }
+    }
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("The updated demand for " + this.getName() + " is " + demand
+          + "; the max is " + maxRes);
     }
   }
 

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

@@ -0,0 +1,42 @@
+package org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair;
+
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.server.resourcemanager.resource.Resources;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class TestFSQueueSchedulable {
+  private FSQueueSchedulable schedulable = null;
+  private Resource maxResource = Resources.createResource(10);
+
+  @Before
+  public void setup() {
+    String queueName = "testFSQueue";
+    FSQueue mockQueue = mock(FSQueue.class);
+    when(mockQueue.getName()).thenReturn(queueName);
+
+    QueueManager mockMgr = mock(QueueManager.class);
+    when(mockMgr.getMaxResources(queueName)).thenReturn(maxResource);
+
+    schedulable = new FSQueueSchedulable(null, mockQueue, mockMgr, null, 0, 0);
+  }
+
+  @Test
+  public void testUpdateDemand() {
+    AppSchedulable app = mock(AppSchedulable.class);
+    Mockito.when(app.getDemand()).thenReturn(maxResource);
+
+    schedulable.addApp(app);
+    schedulable.addApp(app);
+
+    schedulable.updateDemand();
+
+    assertTrue("Demand is greater than max allowed ",
+        Resources.equals(schedulable.getDemand(), maxResource));
+  }
+}