/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.mapred;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

import junit.framework.TestCase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.FairScheduler.JobInfo;
import org.apache.hadoop.mapreduce.Cluster.JobTrackerStatus;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapred.FakeObjectUtilities.FakeJobHistory;
import org.apache.hadoop.mapred.JobInProgress.KillInterruptedException;
import org.apache.hadoop.mapred.UtilsForTests.FakeClock;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;
import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker;
import org.apache.hadoop.mapreduce.split.JobSplit;
import org.apache.hadoop.net.Node;
import org.mortbay.log.Log;

public class TestFairScheduler extends TestCase {
  final static String TEST_DIR = new File(System.getProperty("test.build.data",
      "build/contrib/streaming/test/data")).getAbsolutePath();
  final static String ALLOC_FILE = new File(TEST_DIR, 
      "test-pools").getAbsolutePath();
  
  private static final String POOL_PROPERTY = "pool";
  private static final String EXPLICIT_POOL_PROPERTY = "mapred.fairscheduler.pool";
  
  private static int jobCounter;
  
  class FakeJobInProgress extends JobInProgress {
    
    private FakeTaskTrackerManager taskTrackerManager;
    private int mapCounter = 0;
    private int reduceCounter = 0;
    private final String[][] mapInputLocations; // Array of hosts for each map
    private boolean initialized;
    
    public FakeJobInProgress(JobConf jobConf,
        FakeTaskTrackerManager taskTrackerManager, 
        String[][] mapInputLocations, JobTracker jt) throws IOException {
      super(new JobID("test", ++jobCounter), jobConf, jt);
      this.taskTrackerManager = taskTrackerManager;
      this.mapInputLocations = mapInputLocations;
      this.startTime = System.currentTimeMillis();
      this.status = new JobStatus();
      this.status.setRunState(JobStatus.PREP);
      this.nonLocalMaps = new LinkedList<TaskInProgress>();
      this.nonLocalRunningMaps = new LinkedHashSet<TaskInProgress>();
      this.runningMapCache = new IdentityHashMap<Node, Set<TaskInProgress>>();
      this.nonRunningReduces = new LinkedList<TaskInProgress>();   
      this.runningReduces = new LinkedHashSet<TaskInProgress>();
      this.jobHistory = new FakeJobHistory();
      this.initialized = false;
    }
    
    @Override
    public synchronized void initTasks() throws IOException {
      // initTasks is needed to create non-empty cleanup and setup TIP
      // arrays, otherwise calls such as job.getTaskInProgress will fail
      JobID jobId = getJobID();
      JobConf conf = getJobConf();
      String jobFile = "";
      // create two cleanup tips, one map and one reduce.
      cleanup = new TaskInProgress[2];
      // cleanup map tip.
      cleanup[0] = new TaskInProgress(jobId, jobFile, null, 
              jobtracker, conf, this, numMapTasks, 1);
      cleanup[0].setJobCleanupTask();
      // cleanup reduce tip.
      cleanup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,
                         numReduceTasks, jobtracker, conf, this, 1);
      cleanup[1].setJobCleanupTask();
      // create two setup tips, one map and one reduce.
      setup = new TaskInProgress[2];
      // setup map tip.
      setup[0] = new TaskInProgress(jobId, jobFile, null, 
              jobtracker, conf, this, numMapTasks + 1, 1);
      setup[0].setJobSetupTask();
      // setup reduce tip.
      setup[1] = new TaskInProgress(jobId, jobFile, numMapTasks,
                         numReduceTasks + 1, jobtracker, conf, this, 1);
      setup[1].setJobSetupTask();
      // create maps
      numMapTasks = conf.getNumMapTasks();
      maps = new TaskInProgress[numMapTasks];
      // empty format
      JobSplit.TaskSplitMetaInfo split = JobSplit.EMPTY_TASK_SPLIT;
      for (int i = 0; i < numMapTasks; i++) {
        String[] inputLocations = null;
        if (mapInputLocations != null)
          inputLocations = mapInputLocations[i];
        maps[i] = new FakeTaskInProgress(getJobID(), i,
            getJobConf(), this, inputLocations, split);
        if (mapInputLocations == null) // Job has no locality info
          nonLocalMaps.add(maps[i]);
      }
      // create reduces
      numReduceTasks = conf.getNumReduceTasks();
      reduces = new TaskInProgress[numReduceTasks];
      for (int i = 0; i < numReduceTasks; i++) {
        reduces[i] = new FakeTaskInProgress(getJobID(), i,
            getJobConf(), this);
      }
      
      initialized = true;
    }
    
    @Override
    public boolean inited() {
      return initialized;
    }

    @Override
    public Task obtainNewMapTask(final TaskTrackerStatus tts, int clusterSize,
        int numUniqueHosts, int localityLevel) throws IOException {
      for (int map = 0; map < maps.length; map++) {
        FakeTaskInProgress tip = (FakeTaskInProgress) maps[map];
        if (!tip.isRunning() && !tip.isComplete() &&
            getLocalityLevel(tip, tts) < localityLevel) {
          TaskAttemptID attemptId = getTaskAttemptID(tip);
          JobSplit.TaskSplitMetaInfo split = JobSplit.EMPTY_TASK_SPLIT;
          Task task = new MapTask("", attemptId, 0, split.getSplitIndex(), 1) {
            @Override
            public String toString() {
              return String.format("%s on %s", getTaskID(), tts.getTrackerName());
            }
          };
          runningMapTasks++;
          tip.createTaskAttempt(task, tts.getTrackerName());
          nonLocalRunningMaps.add(tip);
          taskTrackerManager.startTask(tts.getTrackerName(), task, tip);
          return task;
        }
      }
      return null;
    }
    
    @Override
    public Task obtainNewReduceTask(final TaskTrackerStatus tts,
        int clusterSize, int ignored) throws IOException {
      for (int reduce = 0; reduce < reduces.length; reduce++) {
        FakeTaskInProgress tip = 
          (FakeTaskInProgress) reduces[reduce];
        if (!tip.isRunning() && !tip.isComplete()) {
          TaskAttemptID attemptId = getTaskAttemptID(tip);
          Task task = new ReduceTask("", attemptId, 0, maps.length, 1) {
            @Override
            public String toString() {
              return String.format("%s on %s", getTaskID(), tts.getTrackerName());
            }
          };
          runningReduceTasks++;
          tip.createTaskAttempt(task, tts.getTrackerName());
          runningReduces.add(tip);
          taskTrackerManager.startTask(tts.getTrackerName(), task, tip);
          return task;
        }
      }
      return null;
    }
    
    public void mapTaskFinished(TaskInProgress tip) {
      runningMapTasks--;
      finishedMapTasks++;
      nonLocalRunningMaps.remove(tip);
    }
    
    public void reduceTaskFinished(TaskInProgress tip) {
      runningReduceTasks--;
      finishedReduceTasks++;
      runningReduces.remove(tip);
    }
    
    private TaskAttemptID getTaskAttemptID(TaskInProgress tip) {
      JobID jobId = getJobID();
      TaskType type = tip.isMapTask() ? TaskType.MAP : TaskType.REDUCE;
      return new TaskAttemptID(jobId.getJtIdentifier(),
          jobId.getId(), type, tip.getIdWithinJob(), tip.nextTaskId++);
    }
    
    @Override
    int getLocalityLevel(TaskInProgress tip, TaskTrackerStatus tts) {
      FakeTaskInProgress ftip = (FakeTaskInProgress) tip;
      if (ftip.inputLocations != null) {
        // Check whether we're on the same host as an input split
        for (String location: ftip.inputLocations) {
          if (location.equals(tts.host)) {
            return 0;
          }
        }
        // Check whether we're on the same rack as an input split
        for (String location: ftip.inputLocations) {
          if (getRack(location).equals(getRack(tts.host))) {
            return 1;
          }
        }
        // Not on same rack or host
        return 2;
      } else {
        // Job has no locality info  
        return -1;
      }
    }
  }
  
  class FakeTaskInProgress extends TaskInProgress {
    private boolean isMap;
    private FakeJobInProgress fakeJob;
    private TreeMap<TaskAttemptID, String> activeTasks;
    private TaskStatus taskStatus;
    private boolean isComplete = false;
    private String[] inputLocations;
    
    // Constructor for map
    FakeTaskInProgress(JobID jId, int id, JobConf jobConf,
        FakeJobInProgress job, String[] inputLocations, 
        JobSplit.TaskSplitMetaInfo split) {
      super(jId, "", split, null, jobConf, job, id, 1);
      this.isMap = true;
      this.fakeJob = job;
      this.inputLocations = inputLocations;
      activeTasks = new TreeMap<TaskAttemptID, String>();
      taskStatus = TaskStatus.createTaskStatus(isMap);
      taskStatus.setRunState(TaskStatus.State.UNASSIGNED);
    }

    // Constructor for reduce
    FakeTaskInProgress(JobID jId, int id, JobConf jobConf,
        FakeJobInProgress job) {
      super(jId, "", jobConf.getNumMapTasks(), id, null, jobConf, job, 1);
      this.isMap = false;
      this.fakeJob = job;
      activeTasks = new TreeMap<TaskAttemptID, String>();
      taskStatus = TaskStatus.createTaskStatus(isMap);
      taskStatus.setRunState(TaskStatus.State.UNASSIGNED);
    }
    
    private void createTaskAttempt(Task task, String taskTracker) {
      activeTasks.put(task.getTaskID(), taskTracker);
      taskStatus = TaskStatus.createTaskStatus(isMap, task.getTaskID(),
          0.5f, 1, TaskStatus.State.RUNNING, "", "", "", 
          TaskStatus.Phase.STARTING, new Counters());
      taskStatus.setStartTime(clock.getTime());
    }
    
    @Override
    TreeMap<TaskAttemptID, String> getActiveTasks() {
      return activeTasks;
    }
    
    public synchronized boolean isComplete() {
      return isComplete;
    }
    
    public boolean isRunning() {
      return activeTasks.size() > 0;
    }
    
    @Override
    public TaskStatus getTaskStatus(TaskAttemptID taskid) {
      return taskStatus;
    }
    
    void killAttempt() {
      if (isMap) {
        fakeJob.mapTaskFinished(this);
      }
      else {
        fakeJob.reduceTaskFinished(this);
      }
      activeTasks.clear();
      taskStatus.setRunState(TaskStatus.State.UNASSIGNED);
    }
    
    void finishAttempt() {
      isComplete = true;
      if (isMap) {
        fakeJob.mapTaskFinished(this);
      }
      else {
        fakeJob.reduceTaskFinished(this);
      }
      activeTasks.clear();
      taskStatus.setRunState(TaskStatus.State.UNASSIGNED);
    }
  }
  
  static class FakeQueueManager extends QueueManager {
    private Set<String> queues = null;
    FakeQueueManager() {
      super(new Configuration());
    }
    void setQueues(Set<String> queues) {
      this.queues = queues;
    }
    public synchronized Set<String> getLeafQueueNames() {
      return queues;
    }
  }
  
  static class FakeTaskTrackerManager implements TaskTrackerManager {
    int maps = 0;
    int reduces = 0;
    int maxMapTasksPerTracker = 2;
    int maxReduceTasksPerTracker = 2;
    long ttExpiryInterval = 10 * 60 * 1000L; // default interval
    List<JobInProgressListener> listeners =
      new ArrayList<JobInProgressListener>();
    Map<JobID, JobInProgress> jobs = new HashMap<JobID, JobInProgress>();
    
    private Map<String, TaskTracker> trackers =
      new HashMap<String, TaskTracker>();
    private Map<String, TaskStatus> statuses = 
      new HashMap<String, TaskStatus>();
    private Map<String, FakeTaskInProgress> tips = 
      new HashMap<String, FakeTaskInProgress>();
    private Map<String, TaskTrackerStatus> trackerForTip =
      new HashMap<String, TaskTrackerStatus>();
    
    public FakeTaskTrackerManager(int numRacks, int numTrackersPerRack) {
      int nextTrackerId = 1;
      for (int rack = 1; rack <= numRacks; rack++) {
        for (int node = 1; node <= numTrackersPerRack; node++) {
          int id = nextTrackerId++;
          String host = "rack" + rack + ".node" + node;
          System.out.println("Creating TaskTracker tt" + id + " on " + host);
          TaskTracker tt = new TaskTracker("tt" + id);
          tt.setStatus(new TaskTrackerStatus("tt" + id, host, 0,
              new ArrayList<TaskStatus>(), 0,
              maxMapTasksPerTracker, maxReduceTasksPerTracker));
          trackers.put("tt" + id, tt);
        }
      }
    }
    
    @Override
    public ClusterStatus getClusterStatus() {
      int numTrackers = trackers.size();

      return new ClusterStatus(numTrackers, 0,
          ttExpiryInterval, maps, reduces,
          numTrackers * maxMapTasksPerTracker,
          numTrackers * maxReduceTasksPerTracker,
          JobTrackerStatus.RUNNING);
    }

    @Override
    public QueueManager getQueueManager() {
      return null;
    }
    
    @Override
    public int getNumberOfUniqueHosts() {
      return trackers.size();
    }

    @Override
    public Collection<TaskTrackerStatus> taskTrackers() {
      List<TaskTrackerStatus> statuses = new ArrayList<TaskTrackerStatus>();
      for (TaskTracker tt : trackers.values()) {
        statuses.add(tt.getStatus());
      }
      return statuses;
    }


    @Override
    public void addJobInProgressListener(JobInProgressListener listener) {
      listeners.add(listener);
    }

    @Override
    public void removeJobInProgressListener(JobInProgressListener listener) {
      listeners.remove(listener);
    }
    
    @Override
    public int getNextHeartbeatInterval() {
      return JTConfig.JT_HEARTBEAT_INTERVAL_MIN_DEFAULT;
    }

    @Override
    public void killJob(JobID jobid) {
      return;
    }

    @Override
    public JobInProgress getJob(JobID jobid) {
      return jobs.get(jobid);
    }

    public void initJob (JobInProgress job) {
      try {
        job.initTasks();
      } catch (KillInterruptedException e) {
      } catch (IOException e) {
      }
    }
    
    public void failJob (JobInProgress job) {
      // do nothing
    }
    
    // Test methods
    
    public void submitJob(JobInProgress job) throws IOException {
      jobs.put(job.getJobID(), job);
      for (JobInProgressListener listener : listeners) {
        listener.jobAdded(job);
      }
    }
    
    public TaskTracker getTaskTracker(String trackerID) {
      return trackers.get(trackerID);
    }
    
    public void startTask(String trackerName, Task t, FakeTaskInProgress tip) {
      final boolean isMap = t.isMapTask();
      if (isMap) {
        maps++;
      } else {
        reduces++;
      }
      String attemptId = t.getTaskID().toString();
      TaskStatus status = tip.getTaskStatus(t.getTaskID());
      TaskTrackerStatus trackerStatus = trackers.get(trackerName).getStatus();
      tips.put(attemptId, tip);
      statuses.put(attemptId, status);
      trackerForTip.put(attemptId, trackerStatus);
      status.setRunState(TaskStatus.State.RUNNING);
      trackerStatus.getTaskReports().add(status);
    }
    
    public void finishTask(String taskTrackerName, String attemptId) {
      FakeTaskInProgress tip = tips.get(attemptId);
      if (tip.isMapTask()) {
        maps--;
      } else {
        reduces--;
      }
      tip.finishAttempt();
      TaskStatus status = statuses.get(attemptId);
      trackers.get(taskTrackerName).getStatus().getTaskReports().remove(status);
    }

    @Override
    public boolean killTask(TaskAttemptID attemptId, boolean shouldFail) {
      String attemptIdStr = attemptId.toString();
      FakeTaskInProgress tip = tips.get(attemptIdStr);
      if (tip.isMapTask()) {
        maps--;
      } else {
        reduces--;
      }
      tip.killAttempt();
      TaskStatus status = statuses.get(attemptIdStr);
      trackerForTip.get(attemptIdStr).getTaskReports().remove(status);
      return true;
    }
  }
  
  protected JobConf conf;
  protected FairScheduler scheduler;
  private FakeTaskTrackerManager taskTrackerManager;
  private FakeClock clock;

  @Override
  protected void setUp() throws Exception {
    jobCounter = 0;
    new File(TEST_DIR).mkdirs(); // Make sure data directory exists
    // Create an empty pools file (so we can add/remove pools later)
    FileWriter fileWriter = new FileWriter(ALLOC_FILE);
    fileWriter.write("<?xml version=\"1.0\"?>\n");
    fileWriter.write("<allocations />\n");
    fileWriter.close();
    setUpCluster(1, 2, false);
  }

  public String getRack(String hostname) {
    // Host names are of the form rackN.nodeM, so split at the dot.
    return hostname.split("\\.")[0];
  }

  private void setUpCluster(int numRacks, int numNodesPerRack,
      boolean assignMultiple) {
    conf = new JobConf();
    conf.set("mapred.fairscheduler.allocation.file", ALLOC_FILE);
    conf.set("mapred.fairscheduler.poolnameproperty", POOL_PROPERTY);
    conf.setBoolean("mapred.fairscheduler.assignmultiple", assignMultiple);
    // Manually set locality delay because we aren't using a JobTracker so
    // we can't auto-compute it from the heartbeat interval.
    conf.setLong("mapred.fairscheduler.locality.delay.node", 5000);
    conf.setLong("mapred.fairscheduler.locality.delay.rack", 10000);
    taskTrackerManager = new FakeTaskTrackerManager(numRacks, numNodesPerRack);
    clock = new FakeClock();
    scheduler = new FairScheduler(clock, true);
    scheduler.waitForMapsBeforeLaunchingReduces = false;
    scheduler.setConf(conf);
    scheduler.setTaskTrackerManager(taskTrackerManager);
    scheduler.start();
  }
  
  @Override
  protected void tearDown() throws Exception {
    if (scheduler != null) {
      scheduler.terminate();
    }
  }
  
  private JobInProgress submitJobNotInitialized(int state, int maps, int reduces)
	    throws IOException {
    return submitJob(state, maps, reduces, null, null, false);
  }

  private JobInProgress submitJob(int state, int maps, int reduces)
      throws IOException {
    return submitJob(state, maps, reduces, null, null, true);
  }
  
  private JobInProgress submitJob(int state, int maps, int reduces, String pool)
      throws IOException {
    return submitJob(state, maps, reduces, pool, null, true);
  }
  
  private JobInProgress submitJob(int state, int maps, int reduces, String pool,
      String[][] mapInputLocations, boolean initializeJob) throws IOException {
    JobConf jobConf = new JobConf(conf);
    jobConf.setNumMapTasks(maps);
    jobConf.setNumReduceTasks(reduces);
    if (pool != null)
      jobConf.set(POOL_PROPERTY, pool);
    JobInProgress job = new FakeJobInProgress(jobConf, taskTrackerManager,
        mapInputLocations, UtilsForTests.getJobTracker());
    if (initializeJob) {
      taskTrackerManager.initJob(job);
    }
    job.getStatus().setRunState(state);
    taskTrackerManager.submitJob(job);
    job.startTime = clock.time;
    return job;
  }
  
  protected void submitJobs(int number, int state, int maps, int reduces)
    throws IOException {
    for (int i = 0; i < number; i++) {
      submitJob(state, maps, reduces);
    }
  }

  public void testAllocationFileParsing() throws Exception {
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>"); 
    // Give pool A a minimum of 1 map, 2 reduces
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>1</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("</pool>");
    // Give pool B a minimum of 2 maps, 1 reduce
    out.println("<pool name=\"poolB\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>1</minReduces>");
    out.println("</pool>");
    // Give pool C min maps but no min reduces
    out.println("<pool name=\"poolC\">");
    out.println("<minMaps>2</minMaps>");
    out.println("</pool>");
    // Give pool D a limit of 3 running jobs
    out.println("<pool name=\"poolD\">");
    out.println("<maxRunningJobs>3</maxRunningJobs>");
    out.println("</pool>");
    // Give pool E a preemption timeout of one minute
    out.println("<pool name=\"poolE\">");
    out.println("<minSharePreemptionTimeout>60</minSharePreemptionTimeout>");
    out.println("</pool>");
    // Set default limit of jobs per pool to 15
    out.println("<poolMaxJobsDefault>15</poolMaxJobsDefault>");
    // Set default limit of jobs per user to 5
    out.println("<userMaxJobsDefault>5</userMaxJobsDefault>");
    // Give user1 a limit of 10 jobs
    out.println("<user name=\"user1\">");
    out.println("<maxRunningJobs>10</maxRunningJobs>");
    out.println("</user>");
    // Set default min share preemption timeout to 2 minutes
    out.println("<defaultMinSharePreemptionTimeout>120" 
        + "</defaultMinSharePreemptionTimeout>"); 
    // Set fair share preemption timeout to 5 minutes
    out.println("<fairSharePreemptionTimeout>300</fairSharePreemptionTimeout>"); 
    out.println("</allocations>"); 
    out.close();
    
    PoolManager poolManager = scheduler.getPoolManager();
    poolManager.reloadAllocs();
    
    assertEquals(6, poolManager.getPools().size()); // 5 in file + default pool
    assertEquals(0, poolManager.getAllocation(Pool.DEFAULT_POOL_NAME,
        TaskType.MAP));
    assertEquals(0, poolManager.getAllocation(Pool.DEFAULT_POOL_NAME,
        TaskType.REDUCE));
    assertEquals(1, poolManager.getAllocation("poolA", TaskType.MAP));
    assertEquals(2, poolManager.getAllocation("poolA", TaskType.REDUCE));
    assertEquals(2, poolManager.getAllocation("poolB", TaskType.MAP));
    assertEquals(1, poolManager.getAllocation("poolB", TaskType.REDUCE));
    assertEquals(2, poolManager.getAllocation("poolC", TaskType.MAP));
    assertEquals(0, poolManager.getAllocation("poolC", TaskType.REDUCE));
    assertEquals(0, poolManager.getAllocation("poolD", TaskType.MAP));
    assertEquals(0, poolManager.getAllocation("poolD", TaskType.REDUCE));
    assertEquals(0, poolManager.getAllocation("poolE", TaskType.MAP));
    assertEquals(0, poolManager.getAllocation("poolE", TaskType.REDUCE));
    assertEquals(15, poolManager.getPoolMaxJobs(Pool.DEFAULT_POOL_NAME));
    assertEquals(15, poolManager.getPoolMaxJobs("poolA"));
    assertEquals(15, poolManager.getPoolMaxJobs("poolB"));
    assertEquals(15, poolManager.getPoolMaxJobs("poolC"));
    assertEquals(3, poolManager.getPoolMaxJobs("poolD"));
    assertEquals(15, poolManager.getPoolMaxJobs("poolE"));
    assertEquals(10, poolManager.getUserMaxJobs("user1"));
    assertEquals(5, poolManager.getUserMaxJobs("user2"));
    assertEquals(120000, poolManager.getMinSharePreemptionTimeout(
        Pool.DEFAULT_POOL_NAME));
    assertEquals(120000, poolManager.getMinSharePreemptionTimeout("poolA"));
    assertEquals(120000, poolManager.getMinSharePreemptionTimeout("poolB"));
    assertEquals(120000, poolManager.getMinSharePreemptionTimeout("poolC"));
    assertEquals(120000, poolManager.getMinSharePreemptionTimeout("poolD"));
    assertEquals(120000, poolManager.getMinSharePreemptionTimeout("poolA"));
    assertEquals(60000, poolManager.getMinSharePreemptionTimeout("poolE"));
    assertEquals(300000, poolManager.getFairSharePreemptionTimeout());
  }
  
  public void testTaskNotAssignedWhenNoJobsArePresent() throws IOException {
    assertNull(scheduler.assignTasks(tracker("tt1")));
  }

  public void testNonRunningJobsAreIgnored() throws IOException {
    submitJobs(1, JobStatus.SUCCEEDED, 10, 10);
    submitJobs(1, JobStatus.FAILED, 10, 10);
    submitJobs(1, JobStatus.KILLED, 10, 10);
    assertNull(scheduler.assignTasks(tracker("tt1")));
    advanceTime(100); // Check that we still don't assign jobs after an update
    assertNull(scheduler.assignTasks(tracker("tt1")));
  }

  /**
   * This test contains two jobs with fewer required tasks than there are slots.
   * We check that all tasks are assigned, but job 1 gets them first because it
   * was submitted earlier.
   */
  public void testSmallJobs() throws IOException {
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 2, 1);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(2,    info1.mapSchedulable.getDemand());
    assertEquals(1,    info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(1.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time before submitting another job j2, to make j1 run before j2
    // deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 1, 2);
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(2,    info1.mapSchedulable.getDemand());
    assertEquals(1,    info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(1.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(0,    info2.mapSchedulable.getRunningTasks());
    assertEquals(0,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(1,    info2.mapSchedulable.getDemand());
    assertEquals(2,    info2.reduceSchedulable.getDemand());
    assertEquals(1.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    
    // Assign tasks and check that jobs alternate in filling slots
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt2")));
    
    // Check that the scheduler has started counting the tasks as running
    // as soon as it launched them.
    assertEquals(2,  info1.mapSchedulable.getRunningTasks());
    assertEquals(1,  info1.reduceSchedulable.getRunningTasks());
    assertEquals(2,  info1.mapSchedulable.getDemand());
    assertEquals(1,  info1.reduceSchedulable.getDemand());
    assertEquals(1,  info2.mapSchedulable.getRunningTasks());
    assertEquals(2,  info2.reduceSchedulable.getRunningTasks());
    assertEquals(1, info2.mapSchedulable.getDemand());
    assertEquals(2, info2.reduceSchedulable.getDemand());
  }
  
  /**
   * This test is identical to testSmallJobs but sets assignMultiple to
   * true so that multiple tasks can be assigned per heartbeat.
   */
  public void testSmallJobsWithAssignMultiple() throws IOException {
    setUpCluster(1, 2, true);
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 2, 1);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(2,    info1.mapSchedulable.getDemand());
    assertEquals(1,    info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(1.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time before submitting another job j2, to make j1 run before j2
    // deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 1, 2);
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(2,    info1.mapSchedulable.getDemand());
    assertEquals(1,    info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(1.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(0,    info2.mapSchedulable.getRunningTasks());
    assertEquals(0,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(1,    info2.mapSchedulable.getDemand());
    assertEquals(2,    info2.reduceSchedulable.getDemand());
    assertEquals(1.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    
    // Assign tasks and check that jobs alternate in filling slots
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1",
                           "attempt_test_0001_r_000000_0 on tt1",
                           "attempt_test_0002_m_000000_0 on tt1",
                           "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2",
                           "attempt_test_0002_r_000001_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt2")));
    
    // Check that the scheduler has started counting the tasks as running
    // as soon as it launched them.
    assertEquals(2,  info1.mapSchedulable.getRunningTasks());
    assertEquals(1,  info1.reduceSchedulable.getRunningTasks());
    assertEquals(2,  info1.mapSchedulable.getDemand());
    assertEquals(1,  info1.reduceSchedulable.getDemand());
    assertEquals(1,  info2.mapSchedulable.getRunningTasks());
    assertEquals(2,  info2.reduceSchedulable.getRunningTasks());
    assertEquals(1, info2.mapSchedulable.getDemand());
    assertEquals(2, info2.reduceSchedulable.getDemand());
  }
  
  /**
   * This test begins by submitting two jobs with 10 maps and reduces each.
   * The first job is submitted 100ms after the second, to make it get slots
   * first deterministically. We then assign a wave of tasks and check that
   * they are given alternately to job1, job2, job1, job2, etc. We finish
   * these tasks and assign a second wave, which should continue to be
   * allocated in this manner.
   */
  public void testLargeJobs() throws IOException {
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(4.0,  info1.mapSchedulable.getFairShare());
    assertEquals(4.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time before submitting another job j2, to make j1 run before j2
    // deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(0,    info2.mapSchedulable.getRunningTasks());
    assertEquals(0,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info2.mapSchedulable.getDemand());
    assertEquals(10,   info2.reduceSchedulable.getDemand());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    
    // Check that tasks are filled alternately by the jobs
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    
    // Check that no new tasks can be launched once the tasktrackers are full
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    
    // Check that the scheduler has started counting the tasks as running
    // as soon as it launched them.
    assertEquals(2,  info1.mapSchedulable.getRunningTasks());
    assertEquals(2,  info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,  info1.mapSchedulable.getDemand());
    assertEquals(10,  info1.reduceSchedulable.getDemand());
    assertEquals(2,  info2.mapSchedulable.getRunningTasks());
    assertEquals(2,  info2.reduceSchedulable.getRunningTasks());
    assertEquals(10, info2.mapSchedulable.getDemand());
    assertEquals(10, info2.reduceSchedulable.getDemand());
    
    // Finish up the tasks and advance time again. Note that we must finish
    // the task since FakeJobInProgress does not properly maintain running
    // tasks, so the scheduler will always get an empty task list from
    // the JobInProgress's getTasks(TaskType.MAP)/getTasks(TaskType.REDUCE) and 
    // think they finished.
    taskTrackerManager.finishTask("tt1", "attempt_test_0001_m_000000_0");
    taskTrackerManager.finishTask("tt1", "attempt_test_0002_m_000000_0");
    taskTrackerManager.finishTask("tt1", "attempt_test_0001_r_000000_0");
    taskTrackerManager.finishTask("tt1", "attempt_test_0002_r_000000_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0001_m_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0002_m_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0001_r_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0002_r_000001_0");
    advanceTime(200);
    assertEquals(0,   info1.mapSchedulable.getRunningTasks());
    assertEquals(0,   info1.reduceSchedulable.getRunningTasks());
    assertEquals(0,   info2.mapSchedulable.getRunningTasks());
    assertEquals(0,   info2.reduceSchedulable.getRunningTasks());

    // Check that tasks are filled alternately by the jobs
    checkAssignment("tt1", "attempt_test_0001_m_000002_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000002_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_m_000002_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000002_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000003_0 on tt2");
    
    // Check scheduler variables; the demands should now be 8 because 2 tasks
    // of each type have finished in each job
    assertEquals(2,    info1.mapSchedulable.getRunningTasks());
    assertEquals(2,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(8,   info1.mapSchedulable.getDemand());
    assertEquals(8,   info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(2,    info2.mapSchedulable.getRunningTasks());
    assertEquals(2,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(8,   info2.mapSchedulable.getDemand());
    assertEquals(8,   info2.reduceSchedulable.getDemand());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
  }
  
  /**
   * A copy of testLargeJobs that enables the assignMultiple feature to launch
   * multiple tasks per heartbeat. Results should be the same as testLargeJobs.
   */
  public void testLargeJobsWithAssignMultiple() throws IOException {
    setUpCluster(1, 2, true);
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(4.0,  info1.mapSchedulable.getFairShare());
    assertEquals(4.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time before submitting another job j2, to make j1 run before j2
    // deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check scheduler variables; the fair shares should now have been allocated
    // equally between j1 and j2, but j1 should have (4 slots)*(100 ms) deficit
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(0,    info2.mapSchedulable.getRunningTasks());
    assertEquals(0,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info2.mapSchedulable.getDemand());
    assertEquals(10,   info2.reduceSchedulable.getDemand());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    
    // Check that tasks are filled alternately by the jobs
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1",
                           "attempt_test_0001_r_000000_0 on tt1",
                           "attempt_test_0002_m_000000_0 on tt1",
                           "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2",
                           "attempt_test_0001_r_000001_0 on tt2",
                           "attempt_test_0002_m_000001_0 on tt2",
                           "attempt_test_0002_r_000001_0 on tt2");
    
    // Check that no new tasks can be launched once the tasktrackers are full
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    
    // Check that the scheduler has started counting the tasks as running
    // as soon as it launched them.
    assertEquals(2,  info1.mapSchedulable.getRunningTasks());
    assertEquals(2,  info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,  info1.mapSchedulable.getDemand());
    assertEquals(10,  info1.reduceSchedulable.getDemand());
    assertEquals(2,  info2.mapSchedulable.getRunningTasks());
    assertEquals(2,  info2.reduceSchedulable.getRunningTasks());
    assertEquals(10, info2.mapSchedulable.getDemand());
    assertEquals(10, info2.reduceSchedulable.getDemand());
    
    // Finish up the tasks and advance time again. Note that we must finish
    // the task since FakeJobInProgress does not properly maintain running
    // tasks, so the scheduler will always get an empty task list from
    // the JobInProgress's getTasks(TaskType.MAP)/getTasks(TaskType.REDUCE) and
    // think they finished.
    taskTrackerManager.finishTask("tt1", "attempt_test_0001_m_000000_0");
    taskTrackerManager.finishTask("tt1", "attempt_test_0002_m_000000_0");
    taskTrackerManager.finishTask("tt1", "attempt_test_0001_r_000000_0");
    taskTrackerManager.finishTask("tt1", "attempt_test_0002_r_000000_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0001_m_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0002_m_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0001_r_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0002_r_000001_0");
    advanceTime(200);
    assertEquals(0,   info1.mapSchedulable.getRunningTasks());
    assertEquals(0,   info1.reduceSchedulable.getRunningTasks());
    assertEquals(0,   info2.mapSchedulable.getRunningTasks());
    assertEquals(0,   info2.reduceSchedulable.getRunningTasks());

    // Check that tasks are filled alternately by the jobs
    checkAssignment("tt1", "attempt_test_0001_m_000002_0 on tt1",
                           "attempt_test_0001_r_000002_0 on tt1",
                           "attempt_test_0002_m_000002_0 on tt1",
                           "attempt_test_0002_r_000002_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2",
                           "attempt_test_0001_r_000003_0 on tt2",
                           "attempt_test_0002_m_000003_0 on tt2",
                           "attempt_test_0002_r_000003_0 on tt2");
    
    // Check scheduler variables; the demands should now be 8 because 2 tasks
    // of each type have finished in each job
    assertEquals(2,    info1.mapSchedulable.getRunningTasks());
    assertEquals(2,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(8,   info1.mapSchedulable.getDemand());
    assertEquals(8,   info1.reduceSchedulable.getDemand());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(2,    info2.mapSchedulable.getRunningTasks());
    assertEquals(2,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(8,   info2.mapSchedulable.getDemand());
    assertEquals(8,   info2.reduceSchedulable.getDemand());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
  }

  /**
   * We submit two jobs such that one has 2x the priority of the other to 
   * a cluster of 3 nodes, wait for 100 ms, and check that the weights/shares 
   * the high-priority job gets 4 tasks while the normal-priority job gets 2.
   */
  public void testJobsWithPriorities() throws IOException {
    setUpCluster(1, 3, false);
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info2 = scheduler.infos.get(job2);
    job2.setPriority(JobPriority.HIGH);
    scheduler.update();
    
    // Check scheduler variables
    assertEquals(0,   info1.mapSchedulable.getRunningTasks());
    assertEquals(0,   info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,  info1.mapSchedulable.getDemand());
    assertEquals(10,  info1.reduceSchedulable.getDemand());
    assertEquals(2.0, info1.mapSchedulable.getFairShare(), 0.1);
    assertEquals(2.0, info1.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0,   info2.mapSchedulable.getRunningTasks());
    assertEquals(0,   info2.reduceSchedulable.getRunningTasks());
    assertEquals(10,  info2.mapSchedulable.getDemand());
    assertEquals(10,  info2.reduceSchedulable.getDemand());
    assertEquals(4.0, info2.mapSchedulable.getFairShare(), 0.1);
    assertEquals(4.0, info2.reduceSchedulable.getFairShare(), 0.1);
    
    // Advance time
    advanceTime(100);
    
    // Assign tasks and check that j2 gets 2x more tasks than j1. In addition,
    // whenever the jobs' runningTasks/weight ratios are tied, j1 should get
    // the new task first because it started first; thus the tasks of each
    // type should be handed out alternately to 1, 2, 2, 1, 2, 2, etc.
    System.out.println("HEREEEE");
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
    checkAssignment("tt3", "attempt_test_0002_m_000002_0 on tt3");
    checkAssignment("tt3", "attempt_test_0002_r_000002_0 on tt3");
    checkAssignment("tt3", "attempt_test_0002_m_000003_0 on tt3");
    checkAssignment("tt3", "attempt_test_0002_r_000003_0 on tt3");
  }
  
  /**
   * This test starts by submitting three large jobs:
   * - job1 in the default pool, at time 0
   * - job2 in poolA, with an allocation of 1 map / 2 reduces, at time 200
   * - job3 in poolB, with an allocation of 2 maps / 1 reduce, at time 300
   * 
   * We then assign tasks to all slots. The maps should be assigned in the
   * order job2, job3, job 3, job1 because jobs 3 and 2 have guaranteed slots
   * (1 and 2 respectively). Job2 comes before job3 when they are both at 0
   * slots because it has an earlier start time. In a similar manner,
   * reduces should be assigned as job2, job3, job2, job1.
   */
  public void testLargeJobsWithPools() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a minimum of 1 map, 2 reduces
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>1</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("</pool>");
    // Give pool B a minimum of 2 maps, 1 reduce
    out.println("<pool name=\"poolB\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>1</minReduces>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool defaultPool = scheduler.getPoolManager().getPool("default");
    Pool poolA = scheduler.getPoolManager().getPool("poolA");
    Pool poolB = scheduler.getPoolManager().getPool("poolB");
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(4.0,  info1.mapSchedulable.getFairShare());
    assertEquals(4.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time 200ms and submit jobs 2 and 3
    advanceTime(200);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    JobInfo info2 = scheduler.infos.get(job2);
    advanceTime(100);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    JobInfo info3 = scheduler.infos.get(job3);
    
    // Check that minimum and fair shares have been allocated
    assertEquals(0,    defaultPool.getMapSchedulable().getMinShare());
    assertEquals(0,    defaultPool.getReduceSchedulable().getMinShare());
    assertEquals(1.0,  info1.mapSchedulable.getFairShare());
    assertEquals(1.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(1,    poolA.getMapSchedulable().getMinShare());
    assertEquals(2,    poolA.getReduceSchedulable().getMinShare());
    assertEquals(1.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    assertEquals(2,    poolB.getMapSchedulable().getMinShare());
    assertEquals(1,    poolB.getReduceSchedulable().getMinShare());
    assertEquals(2.0,  info3.mapSchedulable.getFairShare());
    assertEquals(1.0,  info3.reduceSchedulable.getFairShare());
    
    // Advance time 100ms
    advanceTime(100);
    
    // Assign tasks and check that slots are first given to needy jobs
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0003_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000000_0 on tt2");
  }

  /**
   * This test starts by submitting three large jobs:
   * - job1 in the default pool, at time 0
   * - job2 in poolA, with an allocation of 2 maps / 2 reduces, at time 200
   * - job3 in poolA, with an allocation of 2 maps / 2 reduces, at time 300
   * 
   * After this, we start assigning tasks. The first two tasks of each type
   * should be assigned to job2 and job3 since they are in a pool with an
   * allocation guarantee, but the next two slots should be assigned to job 3
   * because the pool will no longer be needy.
   */
  public void testLargeJobsWithExcessCapacity() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a minimum of 2 maps, 2 reduces
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool poolA = scheduler.getPoolManager().getPool("poolA");
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(4.0,  info1.mapSchedulable.getFairShare());
    assertEquals(4.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time 200ms and submit job 2
    advanceTime(200);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check that minimum and fair shares have been allocated
    assertEquals(2,    poolA.getMapSchedulable().getMinShare());
    assertEquals(2,    poolA.getReduceSchedulable().getMinShare());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    
    // Advance time 100ms and submit job 3
    advanceTime(100);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    JobInfo info3 = scheduler.infos.get(job3);
    
    // Check that minimum and fair shares have been allocated
    assertEquals(2,    poolA.getMapSchedulable().getMinShare());
    assertEquals(2,    poolA.getReduceSchedulable().getMinShare());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(1.0,  info2.mapSchedulable.getFairShare());
    assertEquals(1.0,  info2.reduceSchedulable.getFairShare());
    assertEquals(1.0,  info3.mapSchedulable.getFairShare());
    assertEquals(1.0,  info3.reduceSchedulable.getFairShare());
    
    // Advance time
    advanceTime(100);
    
    // Assign tasks and check that slots are first given to needy jobs, but
    // that job 1 gets two tasks after due to having a larger share.
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
  }
  
  /**
   * A copy of testLargeJobsWithExcessCapacity that enables assigning multiple
   * tasks per heartbeat. Results should match testLargeJobsWithExcessCapacity.
   */
  public void testLargeJobsWithExcessCapacityAndAssignMultiple() 
      throws Exception {
    setUpCluster(1, 2, true);
    
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a minimum of 2 maps, 2 reduces
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool poolA = scheduler.getPoolManager().getPool("poolA");
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(4.0,  info1.mapSchedulable.getFairShare());
    assertEquals(4.0,  info1.reduceSchedulable.getFairShare());
    
    // Advance time 200ms and submit job 2
    advanceTime(200);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check that minimum and fair shares have been allocated
    assertEquals(2,    poolA.getMapSchedulable().getMinShare());
    assertEquals(2,    poolA.getReduceSchedulable().getMinShare());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    
    // Advance time 100ms and submit job 3
    advanceTime(100);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    JobInfo info3 = scheduler.infos.get(job3);
    
    // Check that minimum and fair shares have been allocated
    assertEquals(2,    poolA.getMapSchedulable().getMinShare());
    assertEquals(2,    poolA.getReduceSchedulable().getMinShare());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(1.0,  info2.mapSchedulable.getFairShare());
    assertEquals(1.0,  info2.reduceSchedulable.getFairShare());
    assertEquals(1.0,  info3.mapSchedulable.getFairShare());
    assertEquals(1.0,  info3.reduceSchedulable.getFairShare());
    
    // Advance time
    advanceTime(100);
    
    // Assign tasks and check that slots are first given to needy jobs, but
    // that job 1 gets two tasks after due to having a larger share.
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1",
                           "attempt_test_0002_r_000000_0 on tt1",
                           "attempt_test_0003_m_000000_0 on tt1",
                           "attempt_test_0003_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000000_0 on tt2",
                           "attempt_test_0001_r_000000_0 on tt2",
                           "attempt_test_0001_m_000001_0 on tt2",
                           "attempt_test_0001_r_000001_0 on tt2");
  }
  
  /**
   * This test starts by submitting two jobs at time 0:
   * - job1 in the default pool
   * - job2, with 1 map and 1 reduce, in poolA, which has an alloc of 4
   *   maps and 4 reduces
   * 
   * When we assign the slots, job2 should only get 1 of each type of task.
   */
  public void testSmallJobInLargePool() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a minimum of 4 maps, 4 reduces
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>4</minMaps>");
    out.println("<minReduces>4</minReduces>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 1, 1, "poolA");
    JobInfo info2 = scheduler.infos.get(job2);
    
    // Check scheduler variables
    assertEquals(0,    info1.mapSchedulable.getRunningTasks());
    assertEquals(0,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(10,   info1.mapSchedulable.getDemand());
    assertEquals(10,   info1.reduceSchedulable.getDemand());
    assertEquals(3.0,  info1.mapSchedulable.getFairShare());
    assertEquals(3.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(0,    info2.mapSchedulable.getRunningTasks());
    assertEquals(0,    info2.reduceSchedulable.getRunningTasks());
    assertEquals(1,    info2.mapSchedulable.getDemand());
    assertEquals(1,    info2.reduceSchedulable.getDemand());
    assertEquals(1.0,  info2.mapSchedulable.getFairShare());
    assertEquals(1.0,  info2.reduceSchedulable.getFairShare());
    
    // Assign tasks and check that slots are first given to needy jobs
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
  }
  
  /**
   * This test starts by submitting four jobs in the default pool. However, the
   * maxRunningJobs limit for this pool has been set to two. We should see only
   * the first two jobs get scheduled, each with half the total slots.
   */
  public void testPoolMaxJobs() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<pool name=\"default\">");
    out.println("<maxRunningJobs>2</maxRunningJobs>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJobNotInitialized(JobStatus.PREP, 10, 10);
    assertTrue(((FakeJobInProgress)job1).inited());
    job1.getStatus().setRunState(JobStatus.RUNNING);
    JobInfo info1 = scheduler.infos.get(job1);
    advanceTime(10);
    JobInProgress job2 = submitJobNotInitialized(JobStatus.PREP, 10, 10);
    assertTrue(((FakeJobInProgress)job2).inited());
    job2.getStatus().setRunState(JobStatus.RUNNING);
    JobInfo info2 = scheduler.infos.get(job2);
    advanceTime(10);
    JobInProgress job3 = submitJobNotInitialized(JobStatus.PREP, 10, 10);
    JobInfo info3 = scheduler.infos.get(job3);
    advanceTime(10);
    JobInProgress job4 = submitJobNotInitialized(JobStatus.PREP, 10, 10);
    JobInfo info4 = scheduler.infos.get(job4);
    
    // Only two of the jobs should be initialized.
    assertTrue(((FakeJobInProgress)job1).inited());
    assertTrue(((FakeJobInProgress)job2).inited());
    assertFalse(((FakeJobInProgress)job3).inited());
    assertFalse(((FakeJobInProgress)job4).inited());
    
    // Check scheduler variables
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(2.0,  info2.mapSchedulable.getFairShare());
    assertEquals(2.0,  info2.reduceSchedulable.getFairShare());
    assertEquals(0.0,  info3.mapSchedulable.getFairShare());
    assertEquals(0.0,  info3.reduceSchedulable.getFairShare());
    assertEquals(0.0,  info4.mapSchedulable.getFairShare());
    assertEquals(0.0,  info4.reduceSchedulable.getFairShare());
    
    // Assign tasks and check that only jobs 1 and 2 get them
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
  }

  /**
   * This test starts by submitting two jobs by user "user1" to the default
   * pool, and two jobs by "user2". We set user1's job limit to 1. We should
   * see one job from user1 and two from user2. 
   */
  public void testUserMaxJobs() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<user name=\"user1\">");
    out.println("<maxRunningJobs>1</maxRunningJobs>");
    out.println("</user>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    job1.getJobConf().set(JobContext.USER_NAME, "user1");
    JobInfo info1 = scheduler.infos.get(job1);
    advanceTime(10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10);
    job2.getJobConf().set(JobContext.USER_NAME, "user1");
    JobInfo info2 = scheduler.infos.get(job2);
    advanceTime(10);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10);
    job3.getJobConf().set(JobContext.USER_NAME, "user2");
    JobInfo info3 = scheduler.infos.get(job3);
    advanceTime(10);
    JobInProgress job4 = submitJob(JobStatus.RUNNING, 10, 10);
    job4.getJobConf().set(JobContext.USER_NAME, "user2");
    JobInfo info4 = scheduler.infos.get(job4);
    
    // Check scheduler variables
    assertEquals(1.33,  info1.mapSchedulable.getFairShare(), 0.1);
    assertEquals(1.33,  info1.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.0,   info2.mapSchedulable.getFairShare());
    assertEquals(0.0,   info2.reduceSchedulable.getFairShare());
    assertEquals(1.33,  info3.mapSchedulable.getFairShare(), 0.1);
    assertEquals(1.33,  info3.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(1.33,  info4.mapSchedulable.getFairShare(), 0.1);
    assertEquals(1.33,  info4.reduceSchedulable.getFairShare(), 0.1);
    
    // Assign tasks and check that slots are given only to jobs 1, 3 and 4
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_r_000000_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0004_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0004_r_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
  }
  
  /**
   * Test a combination of pool job limits and user job limits, the latter
   * specified through both the userMaxJobsDefaults (for some users) and
   * user-specific &lt;user&gt; elements in the allocations file. 
   */
  public void testComplexJobLimits() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<pool name=\"poolA\">");
    out.println("<maxRunningJobs>1</maxRunningJobs>");
    out.println("</pool>");
    out.println("<user name=\"user1\">");
    out.println("<maxRunningJobs>1</maxRunningJobs>");
    out.println("</user>");
    out.println("<user name=\"user2\">");
    out.println("<maxRunningJobs>10</maxRunningJobs>");
    out.println("</user>");
    out.println("<userMaxJobsDefault>2</userMaxJobsDefault>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    
    // Two jobs for user1; only one should get to run
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    job1.getJobConf().set(JobContext.USER_NAME, "user1");
    JobInfo info1 = scheduler.infos.get(job1);
    advanceTime(10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10);
    job2.getJobConf().set(JobContext.USER_NAME, "user1");
    JobInfo info2 = scheduler.infos.get(job2);
    advanceTime(10);
    
    // Three jobs for user2; all should get to run
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10);
    job3.getJobConf().set(JobContext.USER_NAME, "user2");
    JobInfo info3 = scheduler.infos.get(job3);
    advanceTime(10);
    JobInProgress job4 = submitJob(JobStatus.RUNNING, 10, 10);
    job4.getJobConf().set(JobContext.USER_NAME, "user2");
    JobInfo info4 = scheduler.infos.get(job4);
    advanceTime(10);
    JobInProgress job5 = submitJob(JobStatus.RUNNING, 10, 10);
    job5.getJobConf().set(JobContext.USER_NAME, "user2");
    JobInfo info5 = scheduler.infos.get(job5);
    advanceTime(10);
    
    // Three jobs for user3; only two should get to run
    JobInProgress job6 = submitJob(JobStatus.RUNNING, 10, 10);
    job6.getJobConf().set(JobContext.USER_NAME, "user3");
    JobInfo info6 = scheduler.infos.get(job6);
    advanceTime(10);
    JobInProgress job7 = submitJob(JobStatus.RUNNING, 10, 10);
    job7.getJobConf().set(JobContext.USER_NAME, "user3");
    JobInfo info7 = scheduler.infos.get(job7);
    advanceTime(10);
    JobInProgress job8 = submitJob(JobStatus.RUNNING, 10, 10);
    job8.getJobConf().set(JobContext.USER_NAME, "user3");
    JobInfo info8 = scheduler.infos.get(job8);
    advanceTime(10);
    
    // Two jobs for user4, in poolA; only one should get to run
    JobInProgress job9 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    job9.getJobConf().set(JobContext.USER_NAME, "user4");
    JobInfo info9 = scheduler.infos.get(job9);
    advanceTime(10);
    JobInProgress job10 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    job10.getJobConf().set(JobContext.USER_NAME, "user4");
    JobInfo info10 = scheduler.infos.get(job10);
    advanceTime(10);
    
    // Check scheduler variables. The jobs in poolA should get half
    // the total share, while those in the default pool should get
    // the other half. This works out to 2 slots each for the jobs
    // in poolA and 1/3 each for the jobs in the default pool because
    // there are 2 runnable jobs in poolA and 6 jobs in the default pool.
    assertEquals(0.33,   info1.mapSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info1.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.0,    info2.mapSchedulable.getFairShare());
    assertEquals(0.0,    info2.reduceSchedulable.getFairShare());
    assertEquals(0.33,   info3.mapSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info3.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info4.mapSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info4.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info5.mapSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info5.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info6.mapSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info6.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info7.mapSchedulable.getFairShare(), 0.1);
    assertEquals(0.33,   info7.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.0,    info8.mapSchedulable.getFairShare());
    assertEquals(0.0,    info8.reduceSchedulable.getFairShare());
    assertEquals(2.0,    info9.mapSchedulable.getFairShare(), 0.1);
    assertEquals(2.0,    info9.reduceSchedulable.getFairShare(), 0.1);
    assertEquals(0.0,    info10.mapSchedulable.getFairShare());
    assertEquals(0.0,    info10.reduceSchedulable.getFairShare());
  }
  
  public void testSizeBasedWeight() throws Exception {
    scheduler.sizeBasedWeight = true;
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 2, 10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 20, 1);
    assertTrue(scheduler.infos.get(job2).mapSchedulable.getFairShare() >
               scheduler.infos.get(job1).mapSchedulable.getFairShare());
    assertTrue(scheduler.infos.get(job1).reduceSchedulable.getFairShare() >
               scheduler.infos.get(job2).reduceSchedulable.getFairShare());
  }

  /**
   * This test submits jobs in three pools: poolA, which has a weight
   * of 2.0; poolB, which has a weight of 0.5; and the default pool, which
   * should have a weight of 1.0. It then checks that the map and reduce
   * fair shares are given out accordingly. We then submit a second job to
   * pool B and check that each gets half of the pool (weight of 0.25).
   */
  public void testPoolWeights() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<pool name=\"poolA\">");
    out.println("<weight>2.0</weight>");
    out.println("</pool>");
    out.println("<pool name=\"poolB\">");
    out.println("<weight>0.5</weight>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    JobInfo info1 = scheduler.infos.get(job1);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    JobInfo info2 = scheduler.infos.get(job2);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    JobInfo info3 = scheduler.infos.get(job3);
    advanceTime(10);
    
    assertEquals(1.14,  info1.mapSchedulable.getFairShare(), 0.01);
    assertEquals(1.14,  info1.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(2.28,  info2.mapSchedulable.getFairShare(), 0.01);
    assertEquals(2.28,  info2.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(0.57,  info3.mapSchedulable.getFairShare(), 0.01);
    assertEquals(0.57,  info3.reduceSchedulable.getFairShare(), 0.01);
    
    JobInProgress job4 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    JobInfo info4 = scheduler.infos.get(job4);
    advanceTime(10);
    
    assertEquals(1.14,  info1.mapSchedulable.getFairShare(), 0.01);
    assertEquals(1.14,  info1.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(2.28,  info2.mapSchedulable.getFairShare(), 0.01);
    assertEquals(2.28,  info2.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(0.28,  info3.mapSchedulable.getFairShare(), 0.01);
    assertEquals(0.28,  info3.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(0.28,  info4.mapSchedulable.getFairShare(), 0.01);
    assertEquals(0.28,  info4.reduceSchedulable.getFairShare(), 0.01);
  }

  /**
   * This test submits jobs in two pools, poolA and poolB. None of the
   * jobs in poolA have maps, but this should not affect their reduce
   * share.
   */
  public void testPoolWeightsWhenNoMaps() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<pool name=\"poolA\">");
    out.println("<weight>2.0</weight>");
    out.println("</pool>");
    out.println("<pool name=\"poolB\">");
    out.println("<weight>1.0</weight>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 0, 10, "poolA");
    JobInfo info1 = scheduler.infos.get(job1);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 0, 10, "poolA");
    JobInfo info2 = scheduler.infos.get(job2);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    JobInfo info3 = scheduler.infos.get(job3);
    advanceTime(10);
    
    /*
    assertEquals(0,     info1.mapWeight, 0.01);
    assertEquals(1.0,   info1.reduceWeight, 0.01);
    assertEquals(0,     info2.mapWeight, 0.01);
    assertEquals(1.0,   info2.reduceWeight, 0.01);
    assertEquals(1.0,   info3.mapWeight, 0.01);
    assertEquals(1.0,   info3.reduceWeight, 0.01);
    */
    
    assertEquals(0,     info1.mapSchedulable.getFairShare(), 0.01);
    assertEquals(1.33,  info1.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(0,     info2.mapSchedulable.getFairShare(), 0.01);
    assertEquals(1.33,  info2.reduceSchedulable.getFairShare(), 0.01);
    assertEquals(4,     info3.mapSchedulable.getFairShare(), 0.01);
    assertEquals(1.33,  info3.reduceSchedulable.getFairShare(), 0.01);
  }

  public void testPoolMaxMapsReduces() throws Exception {
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Pool with upper bound
    out.println("<pool name=\"poolLimited\">");
    out.println("<weight>1.0</weight>");
    out.println("<maxMaps>2</maxMaps>");
    out.println("<maxReduces>1</maxReduces>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    // Create two jobs with ten maps
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 5, "poolLimited");
    advanceTime(10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 5);
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0002_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000002_0 on tt2");

    Pool limited = scheduler.getPoolManager().getPool("poolLimited");
    assertEquals(2, limited.getSchedulable(TaskType.MAP).getRunningTasks());
    assertEquals(1, limited.getSchedulable(TaskType.REDUCE).getRunningTasks());
    Pool defaultPool = scheduler.getPoolManager().getPool("default");
    assertEquals(2, defaultPool.getSchedulable(TaskType.MAP).getRunningTasks());
    assertEquals(3, defaultPool.getSchedulable(TaskType.REDUCE)
        .getRunningTasks());
    assertEquals(2, job1.runningMapTasks);
    assertEquals(1, job1.runningReduceTasks);
    assertEquals(2, job2.runningMapTasks);
    assertEquals(3, job2.runningReduceTasks);
  }

  /**
   * Tests that max-running-tasks per node are set by assigning load
   * equally accross the cluster in CapBasedLoadManager.
   */
  public void testCapBasedLoadManager() {
    CapBasedLoadManager loadMgr = new CapBasedLoadManager();
    // Arguments to getCap: totalRunnableTasks, nodeCap, totalSlots
    // Desired behavior: return ceil(nodeCap * min(1, runnableTasks/totalSlots))
    assertEquals(1, loadMgr.getCap(1, 1, 100));
    assertEquals(1, loadMgr.getCap(1, 2, 100));
    assertEquals(1, loadMgr.getCap(1, 10, 100));
    assertEquals(1, loadMgr.getCap(200, 1, 100));
    assertEquals(1, loadMgr.getCap(1, 5, 100));
    assertEquals(3, loadMgr.getCap(50, 5, 100));
    assertEquals(5, loadMgr.getCap(100, 5, 100));
    assertEquals(5, loadMgr.getCap(200, 5, 100));
  }

  /**
   * This test starts by launching a job in the default pool that takes
   * all the slots in the cluster. We then submit a job in a pool with
   * min share of 2 maps and 1 reduce task. After the min share preemption
   * timeout, this pool should be allowed to preempt tasks. 
   */
  public void testMinSharePreemption() throws Exception {
    // Enable preemption in scheduler
    scheduler.preemptionEnabled = true;
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a min share of 2 maps and 1 reduce, and a preemption
    // timeout of 1 minute
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>1</minReduces>");
    out.println("<minSharePreemptionTimeout>60</minSharePreemptionTimeout>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool poolA = scheduler.getPoolManager().getPool("poolA");

    // Submit job 1 and assign all slots to it. Sleep a bit before assigning
    // tasks on tt1 and tt2 to ensure that the ones on tt2 get preempted first.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    
    // Ten seconds later, submit job 2.
    advanceTime(10000);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    
    // Ten seconds later, check that job 2 is not able to preempt tasks.
    advanceTime(10000);
    assertEquals(0, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));
    
    // Advance time by 49 more seconds, putting us at 59s after the
    // submission of job 2. It should still not be able to preempt.
    advanceTime(49000);
    assertEquals(0, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));
    
    // Advance time by 2 seconds, putting us at 61s after the submission
    // of job 2. It should now be able to preempt 2 maps and 1 reduce.
    advanceTime(2000);
    assertEquals(2, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(1, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));
    
    // Test that the tasks actually get preempted and we can assign new ones
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(2, job1.runningMaps());
    assertEquals(3, job1.runningReduces());
    checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000000_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
  }

  /**
   * This test starts by launching a job in the default pool that takes
   * all the slots in the cluster. We then submit a job in a pool with
   * min share of 3 maps and 3 reduce tasks, but which only actually
   * needs 1 map and 2 reduces. We check that this pool does not prempt
   * more than this many tasks despite its min share being higher. 
   */
  public void testMinSharePreemptionWithSmallJob() throws Exception {
    // Enable preemption in scheduler
    scheduler.preemptionEnabled = true;
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a min share of 2 maps and 1 reduce, and a preemption
    // timeout of 1 minute
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>3</minMaps>");
    out.println("<minReduces>3</minReduces>");
    out.println("<minSharePreemptionTimeout>60</minSharePreemptionTimeout>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool poolA = scheduler.getPoolManager().getPool("poolA");

    // Submit job 1 and assign all slots to it. Sleep a bit before assigning
    // tasks on tt1 and tt2 to ensure that the ones on tt2 get preempted first.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    
    // Ten seconds later, submit job 2.
    advanceTime(10000);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 1, 2, "poolA");
    
    // Advance time by 59 seconds and check that no preemption occurs.
    advanceTime(59000);
    assertEquals(0, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));
    
    // Advance time by 2 seconds, putting us at 61s after the submission
    // of job 2. Job 2 should now preempt 1 map and 2 reduces.
    advanceTime(2000);
    assertEquals(1, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(2, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));

    // Test that the tasks actually get preempted and we can assign new ones
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(3, job1.runningMaps());
    assertEquals(2, job1.runningReduces());
    checkAssignment("tt2", "attempt_test_0002_r_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
  }

  /**
   * This test runs on a 4-node (8-slot) cluster to allow 3 pools with fair
   * shares greater than 2 slots to coexist (which makes the half-fair-share 
   * of each pool more than 1 so that fair share preemption can kick in). 
   * 
   * The test first starts job 1, which takes 6 map slots and 6 reduce slots,
   * in pool 1.  We then submit job 2 in pool 2, which takes 2 slots of each
   * type. Finally, we submit a third job, job 3 in pool3, which gets no slots. 
   * At this point the fair share of each pool will be 8/3 ~= 2.7 slots. 
   * Pool 1 will be above its fair share, pool 2 will be below it but at half
   * fair share, and pool 3 will be below half fair share. Therefore pool 3 
   * should preempt a task (after a timeout) but pools 1 and 2 shouldn't. 
   */
  public void testFairSharePreemption() throws Exception {
    // Create a bigger cluster than normal (4 tasktrackers instead of 2)
    setUpCluster(1, 4, false);
    // Enable preemption in scheduler
    scheduler.preemptionEnabled = true;
    // Set up pools file with a fair share preemtion timeout of 1 minute
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<fairSharePreemptionTimeout>60</fairSharePreemptionTimeout>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Grab pools (they'll be created even though they're not in the alloc file)
    Pool pool1 = scheduler.getPoolManager().getPool("pool1");
    Pool pool2 = scheduler.getPoolManager().getPool("pool2");
    Pool pool3 = scheduler.getPoolManager().getPool("pool3");

    // Submit job 1. We advance time by 100 between each task tracker
    // assignment stage to ensure that the tasks from job1 on tt3 are the ones
    // that are deterministically preempted first (being the latest launched
    // tasks in an over-allocated job).
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 6, 6, "pool1");
    advanceTime(100);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    advanceTime(100);
    checkAssignment("tt3", "attempt_test_0001_m_000004_0 on tt3");
    checkAssignment("tt3", "attempt_test_0001_r_000004_0 on tt3");
    checkAssignment("tt3", "attempt_test_0001_m_000005_0 on tt3");
    checkAssignment("tt3", "attempt_test_0001_r_000005_0 on tt3");
    advanceTime(100);
    
    // Submit job 2. It should get the last 2 slots.
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "pool2");
    advanceTime(100);
    checkAssignment("tt4", "attempt_test_0002_m_000000_0 on tt4");
    checkAssignment("tt4", "attempt_test_0002_r_000000_0 on tt4");
    checkAssignment("tt4", "attempt_test_0002_m_000001_0 on tt4");
    checkAssignment("tt4", "attempt_test_0002_r_000001_0 on tt4");
    
    // Submit job 3.
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "pool3");
    
    // Check that after 59 seconds, neither pool can preempt
    advanceTime(59000);
    assertEquals(0, scheduler.tasksToPreempt(pool2.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(pool2.getReduceSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(pool3.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(pool3.getReduceSchedulable(),
        clock.getTime()));
    
    // Wait 2 more seconds, so that job 3 has now been in the system for 61s.
    // Now pool 3 should be able to preempt 2 tasks (its share of 2.7 rounded
    // down to its floor), but pool 2 shouldn't.
    advanceTime(2000);
    assertEquals(0, scheduler.tasksToPreempt(pool2.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(pool2.getReduceSchedulable(),
        clock.getTime()));
    assertEquals(2, scheduler.tasksToPreempt(pool3.getMapSchedulable(),
        clock.getTime()));
    assertEquals(2, scheduler.tasksToPreempt(pool3.getReduceSchedulable(),
        clock.getTime()));
    
    // Test that the tasks actually get preempted and we can assign new ones
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(4, job1.runningMaps());
    assertEquals(4, job1.runningReduces());
    checkAssignment("tt3", "attempt_test_0003_m_000000_0 on tt3");
    checkAssignment("tt3", "attempt_test_0003_r_000000_0 on tt3");
    checkAssignment("tt3", "attempt_test_0003_m_000001_0 on tt3");
    checkAssignment("tt3", "attempt_test_0003_r_000001_0 on tt3");
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    assertNull(scheduler.assignTasks(tracker("tt3")));
    assertNull(scheduler.assignTasks(tracker("tt4")));
  }
  
  /**
   * This test runs on a 3-node (6-slot) cluster to allow 3 pools with fair
   * shares equal 2 slots to coexist (which makes the half-fair-share 
   * of each pool equal to 1 so that fair share preemption can kick in). 
   * 
   * The test first starts job 1, which takes 3 map slots and 0 reduce slots,
   * in pool 1.  We then submit job 2 in pool 2, which takes 3 map slots and zero
   * reduce slots. Finally, we submit a third job, job 3 in pool3, which gets no slots. 
   * At this point the fair share of each pool will be 6/3 = 2 slots. 
   * Pool 1 and 2 will be above their fair share and pool 3 will be below half fair share. 
   * Therefore pool 3 should preempt tasks from both pool 1 & 2 (after a timeout) but 
   * pools 1 and 2 shouldn't. 
   */
  public void testFairSharePreemptionFromMultiplePools() throws Exception {
	// Create a bigger cluster than normal (3 tasktrackers instead of 2)
	setUpCluster(1, 3, false);
	// Enable preemption in scheduler
	scheduler.preemptionEnabled = true;
	// Set up pools file with a fair share preemtion timeout of 1 minute
	PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
	out.println("<?xml version=\"1.0\"?>");
	out.println("<allocations>");
	out.println("<fairSharePreemptionTimeout>60</fairSharePreemptionTimeout>");
	out.println("</allocations>");
	out.close();
	scheduler.getPoolManager().reloadAllocs();
	 
	// Grab pools (they'll be created even though they're not in the alloc file)
	Pool pool1 = scheduler.getPoolManager().getPool("pool1");
	Pool pool2 = scheduler.getPoolManager().getPool("pool2");
	Pool pool3 = scheduler.getPoolManager().getPool("pool3");

	// Submit job 1. We advance time by 100 between each task tracker
	// assignment stage to ensure that the tasks from job1 on tt3 are the ones
	// that are deterministically preempted first (being the latest launched
	// tasks in an over-allocated job).
	JobInProgress job1 = submitJob(JobStatus.RUNNING, 12, 0, "pool1");
	advanceTime(100);
	checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
	checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
	advanceTime(100);
	checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
	advanceTime(100);
	    
	// Submit job 2. It should get the last 3 slots.
	JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 0, "pool2");
	advanceTime(100);
	checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2");
	checkAssignment("tt3", "attempt_test_0002_m_000001_0 on tt3");
	advanceTime(100);
	checkAssignment("tt3", "attempt_test_0002_m_000002_0 on tt3");

	advanceTime(100);
	    
	// Submit job 3.
	JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 0, "pool3");
	    
	// Check that after 59 seconds, neither pool can preempt
	advanceTime(59000);
	assertEquals(0, scheduler.tasksToPreempt(pool2.getMapSchedulable(),
			clock.getTime()));
	assertEquals(0, scheduler.tasksToPreempt(pool2.getReduceSchedulable(),
	        clock.getTime()));
	assertEquals(0, scheduler.tasksToPreempt(pool3.getMapSchedulable(),
	        clock.getTime()));
	assertEquals(0, scheduler.tasksToPreempt(pool3.getReduceSchedulable(),
	        clock.getTime()));
	    
	// Wait 2 more seconds, so that job 3 has now been in the system for 61s.
	// Now pool 3 should be able to preempt 2 tasks (its share of 2 rounded
	// down to its floor), but pool 1 & 2 shouldn't.
	advanceTime(2000);
	assertEquals(0, scheduler.tasksToPreempt(pool2.getMapSchedulable(),
	        clock.getTime()));
	assertEquals(0, scheduler.tasksToPreempt(pool2.getReduceSchedulable(),
	        clock.getTime()));
	assertEquals(2, scheduler.tasksToPreempt(pool3.getMapSchedulable(),
	        clock.getTime()));
	assertEquals(0, scheduler.tasksToPreempt(pool3.getReduceSchedulable(),
	        clock.getTime()));
	    
	// Test that the tasks actually get preempted and we can assign new ones.
	// This should preempt one task each from pool1 and pool2
	scheduler.preemptTasksIfNecessary();
	scheduler.update();
	assertEquals(2, job2.runningMaps());  
	assertEquals(2, job1.runningMaps());  
	checkAssignment("tt2", "attempt_test_0003_m_000000_0 on tt2");
	checkAssignment("tt3", "attempt_test_0003_m_000001_0 on tt3");
	assertNull(scheduler.assignTasks(tracker("tt1")));
	assertNull(scheduler.assignTasks(tracker("tt2")));
	assertNull(scheduler.assignTasks(tracker("tt3")));
  }
  
  
  /**
   * This test submits a job that takes all 4 slots, and then a second job in
   * a pool that has both a min share of 2 slots with a 60s timeout and a
   * fair share timeout of 60s. After 60 seconds, this pool will be starved
   * of both min share (2 slots of each type) and fair share (2 slots of each
   * type), and we test that it does not kill more than 2 tasks of each type
   * in total.
   */
  public void testMinAndFairSharePreemption() throws Exception {
    // Enable preemption in scheduler
    scheduler.preemptionEnabled = true;
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a min share of 2 maps and 1 reduce, and a preemption
    // timeout of 1 minute
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("<minSharePreemptionTimeout>60</minSharePreemptionTimeout>");
    out.println("</pool>");
    out.println("<fairSharePreemptionTimeout>60</fairSharePreemptionTimeout>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool poolA = scheduler.getPoolManager().getPool("poolA");

    // Submit job 1 and assign all slots to it. Sleep a bit before assigning
    // tasks on tt1 and tt2 to ensure that the ones on tt2 get preempted first.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    
    // Ten seconds later, submit job 2.
    advanceTime(10000);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    
    // Ten seconds later, check that job 2 is not able to preempt tasks.
    advanceTime(10000);
    assertEquals(0, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));
    
    // Advance time by 49 more seconds, putting us at 59s after the
    // submission of job 2. It should still not be able to preempt.
    advanceTime(49000);
    assertEquals(0, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));
    
    // Advance time by 2 seconds, putting us at 61s after the submission
    // of job 2. It should now be able to preempt 2 maps and 1 reduce.
    advanceTime(2000);
    assertEquals(2, scheduler.tasksToPreempt(poolA.getMapSchedulable(),
        clock.getTime()));
    assertEquals(2, scheduler.tasksToPreempt(poolA.getReduceSchedulable(),
        clock.getTime()));

    // Test that the tasks actually get preempted and we can assign new ones
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(2, job1.runningMaps());
    assertEquals(2, job1.runningReduces());
    checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
  }
  
  /**
   * This is a copy of testMinAndFairSharePreemption that turns preemption
   * off and verifies that no tasks get killed.
   */
  public void testNoPreemptionIfDisabled() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a min share of 2 maps and 1 reduce, and a preemption
    // timeout of 1 minute
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("<minSharePreemptionTimeout>60</minSharePreemptionTimeout>");
    out.println("</pool>");
    out.println("<fairSharePreemptionTimeout>60</fairSharePreemptionTimeout>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();

    // Submit job 1 and assign all slots to it. Sleep a bit before assigning
    // tasks on tt1 and tt2 to ensure that the ones on tt2 get preempted first.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    
    // Ten seconds later, submit job 2.
    advanceTime(10000);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    
    // Advance time by 61s, putting us past the preemption timeout,
    // and check that no tasks get preempted.
    advanceTime(61000);
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(4, job1.runningMaps());
    assertEquals(4, job1.runningReduces());
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
  }

  /**
   * This is a copy of testMinAndFairSharePreemption that turns preemption
   * on but also turns on mapred.fairscheduler.preemption.only.log (the
   * "dry run" parameter for testing out preemption) and verifies that no
   * tasks get killed.
   */
  public void testNoPreemptionIfOnlyLogging() throws Exception {
    // Turn on preemption, but for logging only
    scheduler.preemptionEnabled = true;
    scheduler.onlyLogPreemption = true;
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    // Give pool A a min share of 2 maps and 1 reduce, and a preemption
    // timeout of 1 minute
    out.println("<pool name=\"poolA\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("<minSharePreemptionTimeout>60</minSharePreemptionTimeout>");
    out.println("</pool>");
    out.println("<fairSharePreemptionTimeout>60</fairSharePreemptionTimeout>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();

    // Submit job 1 and assign all slots to it. Sleep a bit before assigning
    // tasks on tt1 and tt2 to ensure that the ones on tt2 get preempted first.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");
    
    // Ten seconds later, submit job 2.
    advanceTime(10000);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    
    // Advance time by 61s, putting us past the preemption timeout,
    // and check that no tasks get preempted.
    advanceTime(61000);
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(4, job1.runningMaps());
    assertEquals(4, job1.runningReduces());
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
  }

  /**
   * This test exercises delay scheduling at the node level. We submit a job
   * with data on rack1.node2 and check that it doesn't get assigned on earlier
   * nodes. A second job with no locality info should get assigned instead.
   * 
   * TaskTracker names in this test map to nodes as follows:
   * - tt1 = rack1.node1
   * - tt2 = rack1.node2
   * - tt3 = rack2.node1
   * - tt4 = rack2.node2
   */
  public void testDelaySchedulingAtNodeLevel() throws IOException {
    setUpCluster(2, 2, true);
    scheduler.assignMultiple = true;
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 1, 0, "pool1",
        new String[][] {
          {"rack2.node2"}
        }, true);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Advance time before submitting another job j2, to make j1 be ahead
    // of j2 in the queue deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 0);
    
    // Assign tasks on nodes 1-3 and check that j2 gets them
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1", 
                           "attempt_test_0002_m_000001_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000002_0 on tt2",
                           "attempt_test_0002_m_000003_0 on tt2");
    checkAssignment("tt3", "attempt_test_0002_m_000004_0 on tt3",
                           "attempt_test_0002_m_000005_0 on tt3");
    
    // Assign a task on node 4 now and check that j1 gets it. The other slot
    // on the node should be given to j2 because j1 will be out of tasks.
    checkAssignment("tt4", "attempt_test_0001_m_000000_0 on tt4",
                           "attempt_test_0002_m_000006_0 on tt4");
    
    // Check that delay scheduling info is properly set
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.NODE);
    assertEquals(info1.timeWaitedForLocalMap, 0);
    assertEquals(info1.skippedAtLastHeartbeat, false);
  }
  
  /**
   * This test submits a job and causes it to exceed its node-level delay,
   * and thus to go on to launch a rack-local task. We submit one job with data
   * on rack2.node4 and check that it does not get assigned on any of the other
   * nodes until 10 seconds (the delay configured in setUpCluster) pass.
   * Finally, after some delay, we let the job assign local tasks and check
   * that it has returned to waiting for node locality.
   * 
   * TaskTracker names in this test map to nodes as follows:
   * - tt1 = rack1.node1
   * - tt2 = rack1.node2
   * - tt3 = rack2.node1
   * - tt4 = rack2.node2
   */
  public void testDelaySchedulingAtRackLevel() throws IOException {
    setUpCluster(2, 2, true);
    scheduler.assignMultiple = true;
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 4, 0, "pool1",
        new String[][] {
          {"rack2.node2"}, {"rack2.node2"}, {"rack2.node2"}, {"rack2.node2"}
        }, true);
    JobInfo info1 = scheduler.infos.get(job1);
    
    // Advance time before submitting another job j2, to make j1 be ahead
    // of j2 in the queue deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 20, 0);
    
    // Assign tasks on nodes 1-3 and check that j2 gets them
    checkAssignment("tt1", "attempt_test_0002_m_000000_0 on tt1", 
                           "attempt_test_0002_m_000001_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000002_0 on tt2",
                           "attempt_test_0002_m_000003_0 on tt2");
    checkAssignment("tt3", "attempt_test_0002_m_000004_0 on tt3",
                           "attempt_test_0002_m_000005_0 on tt3");
    
    // Advance time by 6 seconds to put us past the 5-second node locality delay
    advanceTime(6000);
    
    // Finish some tasks on each node
    taskTrackerManager.finishTask("tt1", "attempt_test_0002_m_000000_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0002_m_000002_0");
    taskTrackerManager.finishTask("tt3", "attempt_test_0002_m_000004_0");
    advanceTime(100);
    
    // Check that job 1 is only assigned on node 3 (which is rack-local)
    checkAssignment("tt1", "attempt_test_0002_m_000006_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000007_0 on tt2");
    checkAssignment("tt3", "attempt_test_0001_m_000000_0 on tt3");
    
    // Check that delay scheduling info is properly set
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.RACK);
    assertEquals(info1.timeWaitedForLocalMap, 0);
    assertEquals(info1.skippedAtLastHeartbeat, false);
    
    // Also give job 1 some tasks on node 4. Its lastMapLocalityLevel
    // should go back to 0 after it gets assigned these.
    checkAssignment("tt4", "attempt_test_0001_m_000001_0 on tt4",
                           "attempt_test_0001_m_000002_0 on tt4");
    
    // Check that delay scheduling info is properly set
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.NODE);
    assertEquals(info1.timeWaitedForLocalMap, 0);
    assertEquals(info1.skippedAtLastHeartbeat, false);
    
    // Check that job 1 no longer assigns tasks in the same rack now
    // that it has obtained a node-local task
    taskTrackerManager.finishTask("tt1", "attempt_test_0002_m_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0002_m_000003_0");
    taskTrackerManager.finishTask("tt3", "attempt_test_0002_m_000005_0");
    advanceTime(100);
    checkAssignment("tt1", "attempt_test_0002_m_000008_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000009_0 on tt2");
    checkAssignment("tt3", "attempt_test_0002_m_000010_0 on tt3");
    advanceTime(100);
    
    // However, job 1 should still be able to launch tasks on node 4
    taskTrackerManager.finishTask("tt4", "attempt_test_0001_m_000001_0");
    advanceTime(100);
    checkAssignment("tt4", "attempt_test_0001_m_000003_0 on tt4");
  }
  
  /**
   * This test submits a job and causes it to exceed its node-level delay,
   * then its rack-level delay. It should then launch tasks off-rack.
   * However, once the job gets a rack-local slot it should stay in-rack,
   * and once it gets a node-local slot it should stay in-node.
   * For simplicity, we don't submit a second job in this test.
   * 
   * TaskTracker names in this test map to nodes as follows:
   * - tt1 = rack1.node1
   * - tt2 = rack1.node2
   * - tt3 = rack2.node1
   * - tt4 = rack2.node2
   */
  public void testDelaySchedulingOffRack() throws IOException {
    setUpCluster(2, 2, true);
    scheduler.assignMultiple = true;
    
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 8, 0, "pool1",
        new String[][] {
          {"rack2.node2"}, {"rack2.node2"}, {"rack2.node2"}, {"rack2.node2"},
          {"rack2.node2"}, {"rack2.node2"}, {"rack2.node2"}, {"rack2.node2"},
        }, true);
    JobInfo info1 = scheduler.infos.get(job1);
    advanceTime(100);
    
    // Check that nothing is assigned on trackers 1-3
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    assertNull(scheduler.assignTasks(tracker("tt3")));
    
    // Advance time by 6 seconds to put us past the 5-sec node locality delay
    advanceTime(6000);

    // Check that nothing is assigned on trackers 1-2; the job would assign
    // a task on tracker 3 (rack1.node2) so we skip that one 
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    
    // Repeat to see that receiving multiple heartbeats works
    advanceTime(100);
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    advanceTime(100);
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));

    // Check that delay scheduling info is properly set
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.NODE);
    assertEquals(info1.timeWaitedForLocalMap, 6200);
    assertEquals(info1.skippedAtLastHeartbeat, true);
    
    // Advance time by 11 seconds to put us past the 10-sec rack locality delay
    advanceTime(11000);
    
    // Now the job should be able to assign tasks on tt1 and tt2
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1",
                           "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2",
                           "attempt_test_0001_m_000003_0 on tt2");

    // Check that delay scheduling info is properly set
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.ANY);
    assertEquals(info1.timeWaitedForLocalMap, 0);
    assertEquals(info1.skippedAtLastHeartbeat, false);
    
    // Now assign a task on tt3. This should make the job stop assigning
    // on tt1 and tt2 (checked after we finish some tasks there)
    checkAssignment("tt3", "attempt_test_0001_m_000004_0 on tt3",
                           "attempt_test_0001_m_000005_0 on tt3");

    // Check that delay scheduling info is properly set
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.RACK);
    assertEquals(info1.timeWaitedForLocalMap, 0);
    assertEquals(info1.skippedAtLastHeartbeat, false);
    
    // Check that j1 no longer assigns tasks on rack 1 now
    taskTrackerManager.finishTask("tt1", "attempt_test_0001_m_000001_0");
    taskTrackerManager.finishTask("tt2", "attempt_test_0001_m_000003_0");
    advanceTime(100);
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
    
    // However, tasks on rack 2 should still be assigned
    taskTrackerManager.finishTask("tt3", "attempt_test_0001_m_000004_0");
    advanceTime(100);
    checkAssignment("tt3", "attempt_test_0001_m_000006_0 on tt3");
    
    // Now assign a task on node 4
    checkAssignment("tt4", "attempt_test_0001_m_000007_0 on tt4");

    // Check that delay scheduling info is set so we are looking for node-local
    // tasks at this point
    assertEquals(info1.lastMapLocalityLevel, LocalityLevel.NODE);
    assertEquals(info1.timeWaitedForLocalMap, 0);
    assertEquals(info1.skippedAtLastHeartbeat, false);
  }
  
  /**
   * This test submits two jobs with 4 maps and 3 reduces in total to a
   * 4-node cluster. Although the cluster has 2 map slots and 2 reduce
   * slots per node, it should only launch one map and one reduce on each
   * node to balance the load. We check that this happens even if
   * assignMultiple is set to true so the scheduler has the opportunity
   * to launch multiple tasks per heartbeat.
   */
  public void testAssignMultipleWithUnderloadedCluster() throws IOException {
    setUpCluster(1, 4, true);
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 2, 2);
    
    // Advance to make j1 be scheduled before j2 deterministically.
    advanceTime(100);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 2, 1);
    
    // Assign tasks and check that at most one map and one reduce slot is used
    // on each node, and that no tasks are assigned on subsequent heartbeats
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1",
                           "attempt_test_0001_r_000000_0 on tt1");
    assertNull(scheduler.assignTasks(tracker("tt1")));
    checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2",
                           "attempt_test_0002_r_000000_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt2")));
    checkAssignment("tt3", "attempt_test_0001_m_000001_0 on tt3",
                           "attempt_test_0001_r_000001_0 on tt3");
    assertNull(scheduler.assignTasks(tracker("tt3")));
    checkAssignment("tt4", "attempt_test_0002_m_000001_0 on tt4");
    assertNull(scheduler.assignTasks(tracker("tt4")));
  }
  
  /**
   * This test submits four jobs in the default pool, which is set to FIFO mode:
   * - job1, with 1 map and 1 reduce
   * - job2, with 3 maps and 3 reduces
   * - job3, with 1 map, 1 reduce, and priority set to HIGH
   * - job4, with 3 maps and 3 reduces
   * 
   * We check that the scheduler assigns tasks first to job3 (because it is
   * high priority), then to job1, then to job2.
   */
  public void testFifoPool() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<pool name=\"default\">");
    out.println("<schedulingMode>fifo</schedulingMode>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 1, 1);
    advanceTime(10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 3, 3);
    advanceTime(10);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 1, 1);
    job3.setPriority(JobPriority.HIGH);
    advanceTime(10);
    JobInProgress job4 = submitJob(JobStatus.RUNNING, 3, 3);
    
    // Assign tasks and check that they're given first to job3 (because it is
    // high priority), then to job1, then to job2.
    checkAssignment("tt1", "attempt_test_0003_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
  }
  
  /**
   * This test submits 2 large jobs each to 2 pools, which are both set to FIFO
   * mode through the global defaultPoolSchedulingMode setting. We check that
   * the scheduler assigns tasks only to the first job within each pool, but
   * alternates between the pools to give each pool a fair share.
   */
  public void testMultipleFifoPools() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<defaultPoolSchedulingMode>fifo</defaultPoolSchedulingMode>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    advanceTime(10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    advanceTime(10);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    advanceTime(10);
    JobInProgress job4 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    
    // Assign tasks and check that they alternate between jobs 1 and 3, the
    // head-of-line jobs in their respective pools.
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0003_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0003_r_000001_0 on tt2");
  }
  
  /**
   * This test submits 2 large jobs each to 2 pools, one of which is set to FIFO
   * mode through the global defaultPoolSchedulingMode setting, and one of which
   * is set to fair mode. We check that the scheduler assigns tasks only to the
   * first job in the FIFO pool but to both jobs in the fair sharing pool.
   */
  public void testFifoAndFairPools() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<defaultPoolSchedulingMode>fifo</defaultPoolSchedulingMode>");
    out.println("<pool name=\"poolB\">");
    out.println("<schedulingMode>fair</schedulingMode>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    
    // Submit jobs, advancing time in-between to make sure that they are
    // all submitted at distinct times.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    advanceTime(10);
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "poolA");
    advanceTime(10);
    JobInProgress job3 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    advanceTime(10);
    JobInProgress job4 = submitJob(JobStatus.RUNNING, 10, 10, "poolB");
    
    // Assign tasks and check that only job 1 gets tasks in pool A, but
    // jobs 3 and 4 both get tasks in pool B.
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0003_r_000000_0 on tt1");
    checkAssignment("tt2", "attempt_test_0001_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0004_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0004_r_000000_0 on tt2");
  }

  /**
   * This test uses the mapred.fairscheduler.pool property to assign jobs to pools.
   */
  public void testPoolAssignment() throws Exception {
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<pool name=\"default\">");
    out.println("<schedulingMode>fair</schedulingMode>");
    out.println("</pool>");
    out.println("<pool name=\"poolA\">");
    out.println("<schedulingMode>fair</schedulingMode>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool defaultPool = scheduler.getPoolManager().getPool("default");
    Pool poolA = scheduler.getPoolManager().getPool("poolA");
 
    // Submit a job to the default pool.  All specifications take default values.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 1, 3);

    assertEquals(1,    defaultPool.getMapSchedulable().getDemand());
    assertEquals(3,    defaultPool.getReduceSchedulable().getDemand());
    assertEquals(0,    poolA.getMapSchedulable().getDemand());
    assertEquals(0,    poolA.getReduceSchedulable().getDemand());

    // Submit a job to the default pool and move it to poolA using setPool.
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 5, 7);

    assertEquals(6,    defaultPool.getMapSchedulable().getDemand());
    assertEquals(10,   defaultPool.getReduceSchedulable().getDemand());
    assertEquals(0,    poolA.getMapSchedulable().getDemand());
    assertEquals(0,    poolA.getReduceSchedulable().getDemand());

    scheduler.getPoolManager().setPool(job2, "poolA");
    assertEquals("poolA", scheduler.getPoolManager().getPoolName(job2));

    defaultPool.getMapSchedulable().updateDemand();
    defaultPool.getReduceSchedulable().updateDemand();
    poolA.getMapSchedulable().updateDemand();
    poolA.getReduceSchedulable().updateDemand();

    assertEquals(1,    defaultPool.getMapSchedulable().getDemand());
    assertEquals(3,    defaultPool.getReduceSchedulable().getDemand());
    assertEquals(5,    poolA.getMapSchedulable().getDemand());
    assertEquals(7,    poolA.getReduceSchedulable().getDemand());

    // Submit a job to poolA by specifying mapred.fairscheduler.pool
    JobConf jobConf = new JobConf(conf);
    jobConf.setNumMapTasks(11);
    jobConf.setNumReduceTasks(13);
    jobConf.set(POOL_PROPERTY, "nonsense"); // test that this is overridden
    jobConf.set(EXPLICIT_POOL_PROPERTY, "poolA");
    JobInProgress job3 = new FakeJobInProgress(jobConf, taskTrackerManager,
        null, UtilsForTests.getJobTracker());
    job3.initTasks();
    job3.getStatus().setRunState(JobStatus.RUNNING);
    taskTrackerManager.submitJob(job3);

    assertEquals(1,    defaultPool.getMapSchedulable().getDemand());
    assertEquals(3,    defaultPool.getReduceSchedulable().getDemand());
    assertEquals(16,   poolA.getMapSchedulable().getDemand());
    assertEquals(20,   poolA.getReduceSchedulable().getDemand());

    // Submit a job to poolA by specifying pool and not mapred.fairscheduler.pool
    JobConf jobConf2 = new JobConf(conf);
    jobConf2.setNumMapTasks(17);
    jobConf2.setNumReduceTasks(19);
    jobConf2.set(POOL_PROPERTY, "poolA");
    JobInProgress job4 = new FakeJobInProgress(jobConf2, taskTrackerManager,
        null, UtilsForTests.getJobTracker());
    job4.initTasks();
    job4.getStatus().setRunState(JobStatus.RUNNING);
    taskTrackerManager.submitJob(job4);

    assertEquals(1,    defaultPool.getMapSchedulable().getDemand());
    assertEquals(3,    defaultPool.getReduceSchedulable().getDemand());
    assertEquals(33,   poolA.getMapSchedulable().getDemand());
    assertEquals(39,   poolA.getReduceSchedulable().getDemand());
  }
  
  private void advanceTime(long time) {
    clock.advance(time);
    scheduler.update();
  }

  protected TaskTracker tracker(String taskTrackerName) {
    return taskTrackerManager.getTaskTracker(taskTrackerName);
  }
  
  protected void checkAssignment(String taskTrackerName,
      String... expectedTasks) throws IOException {
    List<Task> tasks = scheduler.assignTasks(tracker(taskTrackerName));
    assertNotNull(tasks);
    System.out.println("Assigned tasks:");
    for (int i = 0; i < tasks.size(); i++)
      System.out.println("- " + tasks.get(i));
    assertEquals(expectedTasks.length, tasks.size());
    for (int i = 0; i < tasks.size(); i++)
      assertEquals("assignment " + i, expectedTasks[i], tasks.get(i).toString());
  }

  /**
   * This test submits a job that takes all 2 slots in a pool has both a min
   * share of 2 slots with minshare timeout of 5s, and then a second job in
   * default pool with a fair share timeout of 5s. After 60 seconds, this pool
   * will be starved of fair share (2 slots of each type), and we test that it
   * does not kill more than 2 tasks of each type.
   */
  public void testFairSharePreemptionWithShortTimeout() throws Exception {
    // Enable preemption in scheduler
    scheduler.preemptionEnabled = true;
    // Set up pools file
    PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE));
    out.println("<?xml version=\"1.0\"?>");
    out.println("<allocations>");
    out.println("<fairSharePreemptionTimeout>5</fairSharePreemptionTimeout>");
    out.println("<pool name=\"pool1\">");
    out.println("<minMaps>2</minMaps>");
    out.println("<minReduces>2</minReduces>");
    out.println("<minSharePreemptionTimeout>5</minSharePreemptionTimeout>");
    out.println("</pool>");
    out.println("</allocations>");
    out.close();
    scheduler.getPoolManager().reloadAllocs();
    Pool pool1 = scheduler.getPoolManager().getPool("pool1");
    Pool defaultPool = scheduler.getPoolManager().getPool("default");

    // Submit job 1 and assign all slots to it. Sleep a bit before assigning
    // tasks on tt1 and tt2 to ensure that the ones on tt2 get preempted first.
    JobInProgress job1 = submitJob(JobStatus.RUNNING, 10, 10, "pool1");
    JobInfo info1 = scheduler.infos.get(job1);
    checkAssignment("tt1", "attempt_test_0001_m_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000000_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_m_000001_0 on tt1");
    checkAssignment("tt1", "attempt_test_0001_r_000001_0 on tt1");
    advanceTime(100);
    checkAssignment("tt2", "attempt_test_0001_m_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000002_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_m_000003_0 on tt2");
    checkAssignment("tt2", "attempt_test_0001_r_000003_0 on tt2");

    advanceTime(10000);
    assertEquals(4,    info1.mapSchedulable.getRunningTasks());
    assertEquals(4,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(4.0,  info1.mapSchedulable.getFairShare());
    assertEquals(4.0,  info1.reduceSchedulable.getFairShare());
    // Ten seconds later, submit job 2.
    JobInProgress job2 = submitJob(JobStatus.RUNNING, 10, 10, "default");

    // Advance time by 6 seconds without update the scheduler.
    // This simulates the time gap between update and task preemption.
    clock.advance(6000);
    assertEquals(4,    info1.mapSchedulable.getRunningTasks());
    assertEquals(4,    info1.reduceSchedulable.getRunningTasks());
    assertEquals(2.0,  info1.mapSchedulable.getFairShare());
    assertEquals(2.0,  info1.reduceSchedulable.getFairShare());
    assertEquals(0, scheduler.tasksToPreempt(pool1.getMapSchedulable(),
        clock.getTime()));
    assertEquals(0, scheduler.tasksToPreempt(pool1.getReduceSchedulable(),
        clock.getTime()));
    assertEquals(2, scheduler.tasksToPreempt(defaultPool.getMapSchedulable(),
        clock.getTime()));
    assertEquals(2, scheduler.tasksToPreempt(defaultPool.getReduceSchedulable(),
        clock.getTime()));

    // Test that the tasks actually get preempted and we can assign new ones
    scheduler.preemptTasksIfNecessary();
    scheduler.update();
    assertEquals(2, job1.runningMaps());
    assertEquals(2, job1.runningReduces());
    checkAssignment("tt2", "attempt_test_0002_m_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000000_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_m_000001_0 on tt2");
    checkAssignment("tt2", "attempt_test_0002_r_000001_0 on tt2");
    assertNull(scheduler.assignTasks(tracker("tt1")));
    assertNull(scheduler.assignTasks(tracker("tt2")));
  }
}