MAPREDUCE-5966. MR1 FairScheduler use of custom weight adjuster is not thread safe for comparisons. (Anubhav Dhoot via kasha)

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

CHANGES.txt

@@ -230,6 +230,9 @@ Release 1.3.0 - unreleased
     MAPREDUCE-5979. FairScheduler: zero weight can cause sort failures. 
     (Anubhav Dhoot via kasha)
 
+    MAPREDUCE-5966. MR1 FairScheduler use of custom weight adjuster is not 
+    thread safe for comparisons. (Anubhav Dhoot via kasha)
+
 Release 1.2.2 - unreleased
 
   INCOMPATIBLE CHANGES

src/contrib/fairscheduler/src/java/org/apache/hadoop/mapred/FairScheduler.java

@@ -701,11 +701,15 @@ public class FairScheduler extends TaskScheduler {
       
       updateRunnability(); // Set job runnability based on user/pool limits 
       
-      // Update demands of jobs and pools
+      // Update demands and weights of jobs and pools
       for (Pool pool: poolMgr.getPools()) {
         pool.getMapSchedulable().updateDemand();
         pool.getReduceSchedulable().updateDemand();
+
+        pool.getMapSchedulable().updateWeight();
+        pool.getReduceSchedulable().updateWeight();
       }
+
       
       // Compute fair shares based on updated demands
       List<PoolSchedulable> mapScheds = getPoolSchedulables(TaskType.MAP);

src/contrib/fairscheduler/src/java/org/apache/hadoop/mapred/JobSchedulable.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.mapred;
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.hadoop.mapred.FairScheduler.JobInfo;
 import org.apache.hadoop.mapreduce.TaskType;
@@ -29,6 +30,7 @@ public class JobSchedulable extends Schedulable {
   protected JobInProgress job;
   protected TaskType taskType;
   private int demand = 0;
+  private double weight = 1.0;
 
   public JobSchedulable(FairScheduler scheduler, JobInProgress job, 
       TaskType taskType) {
@@ -133,9 +135,14 @@ public class JobSchedulable extends Schedulable {
   
   @Override
   public double getWeight() {
-    return scheduler.getJobWeight(job, taskType);
+    return weight;
   }
-  
+
+  @Override
+  public void updateWeight() {
+    weight = scheduler.getJobWeight(job, taskType);
+  }
+
   @Override
   public int getMinShare() {
     return 0;
@@ -181,7 +188,6 @@ public class JobSchedulable extends Schedulable {
     }
   }
 
-  
   @Override
   protected String getMetricsContextName() {
     return "jobs";

src/contrib/fairscheduler/src/java/org/apache/hadoop/mapred/PoolSchedulable.java

@@ -95,7 +95,18 @@ public class PoolSchedulable extends Schedulable {
           + "; maxTasks is " + maxTasks);
     }
   }
-  
+
+  /**
+   * Ask jobs in the pool to update weights. Its own weight is computed on
+   * the fly
+   */
+  @Override
+  public void updateWeight() {
+    for (JobSchedulable sched: jobScheds) {
+      sched.updateWeight();
+    }
+  }
+
   /**
    * Distribute the pool's fair share among its jobs
    */
@@ -160,6 +171,9 @@ public class PoolSchedulable extends Schedulable {
     } else {
       throw new RuntimeException("Unsupported pool scheduling mode " + mode);
     }
+    for (JobSchedulable sched: jobScheds) {
+      sched.updateWeight();
+    }
     Collections.sort(jobScheds, comparator);
     for (JobSchedulable sched: jobScheds) {
       Task task = sched.assignTask(tts, currentTime, visited);

src/contrib/fairscheduler/src/java/org/apache/hadoop/mapred/Schedulable.java

@@ -95,7 +95,10 @@ abstract class Schedulable {
   
   /** Refresh the Schedulable's demand and those of its children if any. */
   public abstract void updateDemand();
-  
+
+  /** Refresh the Schedulable's weight and those of its children if any */
+  public abstract void updateWeight();
+
   /** 
    * Distribute the fair share assigned to this Schedulable among its 
    * children (used in pools where the internal scheduler is fair sharing). 

src/contrib/fairscheduler/src/test/org/apache/hadoop/mapred/FakeSchedulable.java

@@ -108,6 +108,9 @@ public class FakeSchedulable extends Schedulable {
   @Override
   public void updateDemand() {}
 
+  @Override
+  public void updateWeight() {}
+
   @Override
   public TaskType getTaskType() {
     return TaskType.MAP;

src/contrib/fairscheduler/src/test/org/apache/hadoop/mapred/TestFairScheduler.java

@@ -31,13 +31,14 @@ import java.util.IdentityHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 import java.util.Set;
 import java.util.TreeMap;
+import java.util.concurrent.LinkedBlockingDeque;
 
 import junit.framework.TestCase;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.mapred.FairScheduler.JobInfo;
 import org.apache.hadoop.mapred.JobInProgress.KillInterruptedException;
 import org.apache.hadoop.mapred.UtilsForTests.FakeClock;
@@ -3092,8 +3093,6 @@ public class TestFairScheduler extends TestCase {
     assertNull(scheduler.assignTasks(tracker("tt2")));
   }
 
-
-
   class TestJobSchedulableSort extends JobSchedulable {
 
     private final double testFairShare;
@@ -3154,10 +3153,6 @@ public class TestFairScheduler extends TestCase {
 
   public void testFairShareComparator()
   {
-    List<TestJobSchedulableSort> jobs = new ArrayList<TestJobSchedulableSort>();
-    final int iterations = 100;
-    int jobCount = 100;
-
     Comparator<Schedulable> comparator = new
         SchedulingAlgorithms.FairShareComparator();
 
@@ -3186,8 +3181,54 @@ public class TestFairScheduler extends TestCase {
     // s3 has a higher running task to weight ratio (infinity)
     assertTrue(comparator.compare(s1, s3) < 0);
   }
-  
-  
+
+  /**
+   * This test verifies that sorting of JobSchedulables with a custom
+   * weightadjuster that returns different values when called does not break
+   * the sorting. If the weight changes during the sort,
+   * the sort would fail in jdk7
+   */
+  public void testJobSchedulableSortingWithCustomWeightAdjuster() throws
+      IOException, InterruptedException {
+    final int iterations = 100, jobCount = 100, racks = 100, nodesPerRack = 2;
+    final int totalTaskTrackers = nodesPerRack * racks;
+
+    setUpCluster(racks, nodesPerRack, true);
+
+    scheduler.weightAdjuster = new WeightAdjuster() {
+      Random r = new Random();
+
+      @Override
+      public double adjustWeight(JobInProgress job, TaskType taskType, double
+          curWeight) {
+        return curWeight * r.nextInt(100);
+      }
+    };
+
+    for (int j = 0; j < jobCount; j++) {
+      advanceTime(100);
+      submitJob(JobStatus.RUNNING, 2 * iterations, iterations);
+      scheduler.updateMetrics();
+    }
+
+    final LinkedBlockingDeque<Task> tasks =
+        new LinkedBlockingDeque<Task>();
+
+    final String taskTrackerNamePrefix = "tt";
+
+    Random r1 = new Random();
+    for (int i = 0; i < iterations; i++) {
+
+      int randomTaskTrackerId = r1.nextInt(totalTaskTrackers) + 1;
+      String taskTrackerName = taskTrackerNamePrefix + randomTaskTrackerId;
+      List<Task> assignedTasks = scheduler.assignTasks(tracker
+          (taskTrackerName));
+      if (assignedTasks != null) {
+        tasks.addAll(assignedTasks);
+      }
+    }
+  }
+
   /**
    * Ask scheduler to update metrics and then verify that they're all
    * correctly published to the metrics context