/**
 * 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.IOException;
import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.LinkedHashMap;
import java.util.Set;
import java.util.LinkedHashSet;
import java.util.Iterator;
import java.util.Random;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.TaskStatus.Phase;
import org.apache.hadoop.mapred.TaskStatus.State;
import org.apache.hadoop.tools.rumen.TaskAttemptInfo;
import org.apache.hadoop.tools.rumen.ReduceTaskAttemptInfo;
// Explicitly use the new api, older o.a.h.mapred.TaskAttemptID is deprecated
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.conf.Configuration;

/**
 * This class simulates a {@link TaskTracker}. Its main purpose is to call heartbeat()
 * of the simulated Job Tracker with apropriately updated statuses of the 
 * tasks assigned to it.
 *
 * The events emitted and consumed are HeartbeatEvent and 
 * TaskAttemptCompletionEvent .
 *
 * Internal naming convention: accept() dispatches simulation events to 
 * process*Event() methods. heartbeat() dispactches task tracker actions to 
 * handle*Action() methods.
 */
public class SimulatorTaskTracker implements SimulatorEventListener {
  /** Default host name. */
  public static String DEFAULT_HOST_NAME = "unknown";
  /** Default task tracker name. */
  public static String DEFAULT_TRACKER_NAME = 
      "tracker_unknown:localhost/127.0.0.1:10000";
  /** Default number of map slots per task tracker. */
  public static final int DEFAULT_MAP_SLOTS = 2;
  /** Default number of reduce slots per task tracker. */
  public static final int DEFAULT_REDUCE_SLOTS = 2;
  /** Default range of heartbeat response perturbations + 1 in milliseconds. */
  public static final int DEFAULT_HEARTBEAT_FUZZ = 11;
  /** The name of the task tracker. */
  protected final String taskTrackerName;
  /** The name of the host the task tracker is running on. */
  protected final String hostName;
  /** The http port the simulated task tracker reports to the jobtracker. */  
  protected final int httpPort = 80;
  /** Number of map slots. */
  protected final int maxMapSlots;
  /** Number of reduce slots. */
  protected final int maxReduceSlots;
  /** The job tracker this task tracker is a slave of. */
  protected final InterTrackerProtocol jobTracker;

  /**
   * State of and bookkeeping information for all tasks assigned to the task 
   * tracker. Contains all the information about running or completed but
   * not yet reported tasks. We manage it in a mark & sweep garbage collector 
   * manner. We insert tasks on launch, mark them on completion, and remove
   * completed tasks at heartbeat() reports.
   * We use LinkedHashMap instead of HashMap so that the order of iteration
   * is deterministic.
   */
  protected Map<TaskAttemptID, SimulatorTaskInProgress> tasks = 
      new LinkedHashMap<TaskAttemptID, SimulatorTaskInProgress>();
  /** 
   * Number of map slots allocated to tasks in RUNNING state on this task 
   * tracker. Must be in sync with the tasks map above. 
   */
  private int usedMapSlots = 0;
  /**
   * Number of reduce slots allocated to tasks in RUNNING state on this task
   * tracker. Must be in sync with the tasks map above.
   */
  private int usedReduceSlots = 0;  
  /**
   * True if the jobTracker.heartbeat() call to be made is the first one.
   * We need this to mimick the InterTrackerProtocol properly.
   */
  private boolean firstHeartbeat = true;

  // last heartbeat response recieved
  private short heartbeatResponseId = -1;

  /**
   * Task attempt ids for which TaskAttemptCompletionEvent was created but the 
   * task attempt got killed.
   * We use LinkedHashSet to get deterministic iterators, should ever use one.
   */
  private Set<TaskAttemptID> orphanTaskCompletions = 
    new LinkedHashSet<TaskAttemptID>();

  /** The log object to send our messages to; only used for debugging. */
  private static final Log LOG = LogFactory.getLog(SimulatorTaskTracker.class);
  
  /** 
   * Number of milliseconds to perturb the requested heartbeat intervals with
   * to simulate network latency, etc.
   * If <= 1 then there is no pertrubation. This option is also useful for 
   * testing.
   * If > 1 then hearbeats are perturbed with a uniformly random integer in 
   * (-heartbeatIntervalFuzz,+heartbeatIntervalFuzz), not including 
   * the bounds.
   */
  private int heartbeatIntervalFuzz = -1;
  /** Used for randomly perturbing the heartbeat timings. */
  private Random random;
  
  /**
   * Constructs a task tracker. 
   *
   * @param jobTracker the SimulatorJobTracker we talk to
   * @param conf Configuration object. Parameters read are:
   * <dl>
   * <dt> mumak.tasktracker.tracker.name <dd> 
   *      the task tracker name to report, otherwise unused
   * <dt> mumak.tasktracker.host.name <dd> 
   *      the host name to report, otherwise unused
   * <dt> mapred.tasktracker.map.tasks.maximum <dd> 
   *      the number of map slots
   * <dt> mapred.tasktracker.reduce.tasks.maximum <dd> 
   *      the number of reduce slots
   * <dt> mumak.tasktracker.heartbeat.fuzz <dd>
   *      Perturbation for the heartbeats. 
   *      None if <= 1 else perturbations are uniformly randomly generated 
   *      in (-heartbeat.fuzz,+heartbeat.fuzz), not including the bounds.
   * </dl>
   */
  public SimulatorTaskTracker(InterTrackerProtocol jobTracker,
                              Configuration conf) {
    this.taskTrackerName = conf.get(
        "mumak.tasktracker.tracker.name", DEFAULT_TRACKER_NAME);

    LOG.debug("SimulatorTaskTracker constructor, taskTrackerName=" +
              taskTrackerName);

    this.jobTracker = jobTracker;    
    this.hostName = conf.get(
        "mumak.tasktracker.host.name", DEFAULT_HOST_NAME);
    this.maxMapSlots = conf.getInt(
        "mapred.tasktracker.map.tasks.maximum", DEFAULT_MAP_SLOTS);
    this.maxReduceSlots = conf.getInt(
        "mapred.tasktracker.reduce.tasks.maximum", DEFAULT_REDUCE_SLOTS);
    this.heartbeatIntervalFuzz = conf.getInt(
        "mumak.tasktracker.heartbeat.fuzz", DEFAULT_HEARTBEAT_FUZZ);
    long seed = conf.getLong("mumak.tasktracker.random.seed", 
        System.nanoTime());
    this.random = new Random(seed);
  }
  
  /**
   * Processes a simulation event. 
   *
   * @param event the event to process, should be an instance of HeartbeatEvent
   *        or TaskAttemptCompletionEvent
   * @return the list of new events generated in response
   */
  @Override
  public List<SimulatorEvent> accept(SimulatorEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Accepted event " + event);
    }

    if (event instanceof HeartbeatEvent) {
      return processHeartbeatEvent((HeartbeatEvent)event);
    } else if (event instanceof TaskAttemptCompletionEvent) {
      return processTaskAttemptCompletionEvent((TaskAttemptCompletionEvent)
                                               event);
    } else {
      throw new IllegalArgumentException("Unhandled event "+event);
    }
  }
  
  /**
   * Called once at the start of the simulation.
   *
   * @param when Time when the task tracker starts.
   * @return the initial HeartbeatEvent for ourselves.
   */ 
  public List<SimulatorEvent> init(long when) {
    LOG.debug("TaskTracker starting up, current simulation time=" + when);

    return Collections.<SimulatorEvent>singletonList(new HeartbeatEvent(this, when));  
  }
 
  /**
   * Stops running a task attempt on the task tracker. It also updates the 
   * number of available slots accordingly.
   * 
   * @param finalStatus the TaskStatus containing the task id and final 
   *        status of the task attempt. This rountine asserts a lot of the
   *        finalStatus params, in case it is coming from a task attempt
   *        completion event sent to ourselves. Only the run state, finish time,
   *        and progress fields of the task attempt are updated.
   * @param now Current simulation time, used for assert only
   */
  private void finishRunningTask(TaskStatus finalStatus, long now) {
    TaskAttemptID taskId = finalStatus.getTaskID();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Finishing running task id=" + taskId + ", now=" + now);
    }

    SimulatorTaskInProgress tip = tasks.get(taskId);
    if (tip == null) {
      throw new IllegalArgumentException("Unknown task attempt " + taskId
          + " completed");
    }
    TaskStatus currentStatus = tip.getTaskStatus();
    if (currentStatus.getRunState() != State.RUNNING) {
      throw new IllegalArgumentException(
          "Task attempt to finish is not running: " + tip);
    }

    // Check that finalStatus describes a task attempt that has just been
    // completed
    State finalRunState = finalStatus.getRunState();
    if (finalRunState != State.SUCCEEDED && finalRunState != State.FAILED
        && finalRunState != State.KILLED) {
      throw new IllegalArgumentException(
          "Final run state for completed task can't be : " + finalRunState
              + " " + tip);
    }

    if (now != finalStatus.getFinishTime()) {
      throw new IllegalArgumentException(
          "Current time does not match task finish time: now=" + now
              + ", finish=" + finalStatus.getFinishTime());
    }

    if (currentStatus.getIsMap() != finalStatus.getIsMap()
        || currentStatus.getNumSlots() != finalStatus.getNumSlots()
        || currentStatus.getPhase() != finalStatus.getPhase()
        || currentStatus.getStartTime() != finalStatus.getStartTime()) {
      throw new IllegalArgumentException(
          "Current status does not match final status");
    }

    // We can't assert getShuffleFinishTime() and getSortFinishTime() for
    // reduces as those were unknown when the task attempt completion event
    // was created. We have not called setMapFinishTime() for maps either.
    // If we were really thorough we could update the progress of the task
    // and check if it is consistent with finalStatus.

    // If we've got this far it is safe to update the task status
    currentStatus.setRunState(finalStatus.getRunState());
    currentStatus.setFinishTime(finalStatus.getFinishTime());
    currentStatus.setProgress(finalStatus.getProgress());

    // and update the free slots
    int numSlots = currentStatus.getNumSlots();
    if (tip.isMapTask()) {
      usedMapSlots -= numSlots;
      if (usedMapSlots < 0) {
        throw new IllegalStateException(
            "TaskTracker reaches negative map slots: " + usedMapSlots);
      }
    } else {
      usedReduceSlots -= numSlots;
      if (usedReduceSlots < 0) {
        throw new IllegalStateException(
            "TaskTracker reaches negative reduce slots: " + usedReduceSlots);
      }
    }
  }

  /**
   * Records that a task attempt has completed. Ignores the event for tasks
   * that got killed after the creation of the completion event.
   * 
   * @param event the TaskAttemptCompletionEvent the tracker sent to itself
   * @return the list of response events, empty
   */ 
  private List<SimulatorEvent> processTaskAttemptCompletionEvent(
      TaskAttemptCompletionEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Processing task attempt completion event" + event);
    }
    
    long now = event.getTimeStamp();  
    TaskStatus finalStatus = event.getStatus();
    TaskAttemptID taskID = finalStatus.getTaskID();
    boolean killedEarlier = orphanTaskCompletions.remove(taskID);
    if (!killedEarlier) {
      finishRunningTask(finalStatus, now);
    }
    return SimulatorEngine.EMPTY_EVENTS;
  }
  
 
  /** 
   * Creates a signal for itself marking the completion of a task attempt. 
   * It assumes that the task attempt hasn't made any progress in the user 
   * space code so far, i.e. it is called right at launch for map tasks and
   * immediately after all maps completed for reduce tasks.
   *
   * @param tip the simulator task in progress
   * @param now the current simulation time
   * @return the TaskAttemptCompletionEvent we are sending to ourselves
   */
  private TaskAttemptCompletionEvent createTaskAttemptCompletionEvent(
                                       SimulatorTaskInProgress tip, long now) {
    // We need to clone() status as we modify and it goes into an Event
    TaskStatus status = (TaskStatus)tip.getTaskStatus().clone();
    long delta = tip.getUserSpaceRunTime();
    assert delta >= 0 : "TaskAttempt " + tip.getTaskStatus().getTaskID()
        + " has negative UserSpaceRunTime = " + delta;
    long finishTime = now + delta;
    status.setFinishTime(finishTime);
    if (tip.isMapTask()) {
      status.setMapFinishTime(finishTime);
    }
    status.setProgress(1.0f);
    status.setRunState(tip.getFinalRunState());
    TaskAttemptCompletionEvent event = 
        new TaskAttemptCompletionEvent(this, status);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Created task attempt completion event " + event);
    }
    return event;
  }
  
  /**
   * Launches a task on the simulated task tracker. 
   * 
   * @param action SimulatorLaunchTaskAction sent by the job tracker
   * @param now current simulation time
   * @return new events generated, a TaskAttemptCompletionEvent for map
   *         tasks, empty otherwise
   */
  private List<SimulatorEvent> handleSimulatorLaunchTaskAction(
                         SimulatorLaunchTaskAction action, long now) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Handling launch task action " + action);
    }
    // First, create statuses and update used slots for map and reduce 
    // task separately
    Task task = action.getTask();
    TaskAttemptID taskId = task.getTaskID();
    if (tasks.containsKey(taskId)) {
      throw new IllegalArgumentException("Multiple launch of task id =" + taskId);
    }

    // Ctor of MapTaskStatus and ReduceTaskStatus need deprecated 
    // o.a.h.mapred.TaskAttemptID, hence the downgrade
    org.apache.hadoop.mapred.TaskAttemptID taskIdOldApi = 
        org.apache.hadoop.mapred.TaskAttemptID.downgrade(taskId);
    TaskStatus status;
    int numSlotsRequired = task.getNumSlotsRequired();    
    Counters emptyCounters = new Counters();
    if (task.isMapTask()) {     
      status = new MapTaskStatus(taskIdOldApi, 0f, numSlotsRequired,
                                 State.RUNNING, "", "", taskTrackerName, 
                                 Phase.MAP, emptyCounters);
      usedMapSlots += numSlotsRequired;
      if (usedMapSlots > maxMapSlots) {
        throw new IllegalStateException("usedMapSlots exceeds maxMapSlots: " + 
          usedMapSlots + " > " + maxMapSlots);
      }
    } else {
      status = new ReduceTaskStatus(taskIdOldApi, 0f, numSlotsRequired, 
                                    State.RUNNING, "", "", taskTrackerName,
                                    Phase.SHUFFLE, emptyCounters);
      usedReduceSlots += numSlotsRequired;
      if (usedReduceSlots > maxReduceSlots) {
        throw new IllegalStateException("usedReduceSlots exceeds usedReduceSlots: " + 
            usedReduceSlots + " > " + usedReduceSlots);
      }
    }
    
    //  Second, create and store a TIP
    status.setStartTime(now);
    SimulatorTaskInProgress tip = 
      new SimulatorTaskInProgress(action, status, now);
    tasks.put(taskId, tip);

    // Third, schedule events for ourselves
    if (task.isMapTask()) {
      // we know when this task attempts ends iff it is a map 
      TaskAttemptCompletionEvent e = createTaskAttemptCompletionEvent(tip, now);
      return Collections.<SimulatorEvent>singletonList(e);
    } else { 
      // reduce, completion time can only be determined when all maps are done
      return SimulatorEngine.EMPTY_EVENTS;
    }
  }
  
  /** 
   * Kills a task attempt.
   *
   * @param action contains the task attempt to kill
   * @param now current simulation time
   * @return new events generated in response, empty
   */
  private List<SimulatorEvent> handleKillTaskAction(KillTaskAction action, long now) {
    TaskAttemptID taskId = action.getTaskID();
    // we don't have a nice(r) toString() in Hadoop's TaskActions 
    if (LOG.isDebugEnabled()) {
      LOG.debug("Handling kill task action, taskId=" + taskId + ", now=" + now);
    }
    
    SimulatorTaskInProgress tip = tasks.get(taskId);
    
    // Safety check: We might get a KillTaskAction even for completed reduces
    if (tip == null) {
      return SimulatorEngine.EMPTY_EVENTS;
    }
    
    progressTaskStatus(tip, now); // make progress up to date
    TaskStatus finalStatus = (TaskStatus)tip.getTaskStatus().clone();
    finalStatus.setFinishTime(now);
    finalStatus.setRunState(State.KILLED);
    finishRunningTask(finalStatus, now);
   
    if (finalStatus.getIsMap() || finalStatus.getPhase() == Phase.REDUCE) {
      // if we have already created a task attempt completion event we remember
      // the task id, so that we can safely ignore the event when its delivered
      orphanTaskCompletions.add(taskId);
    }
    return SimulatorEngine.EMPTY_EVENTS;
  }  

  /** 
   * Starts "running" the REDUCE phase of reduce upon being notified that 
   * all map tasks are (successfully) done.
   *
   * @param action contains the notification for one of the reduce tasks
   * @param now current simulation time
   * @return new events generated, a single TaskAttemptCompletionEvent for the
   *         reduce
   */
  private List<SimulatorEvent> handleAllMapsCompletedTaskAction(
                        AllMapsCompletedTaskAction action, long now) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Handling all maps completed task action " + action);
    }
    
    TaskAttemptID taskId = action.getTaskID();
    SimulatorTaskInProgress tip = tasks.get(taskId);
    // If tip is null here it is because the task attempt to be notified is
    // unknown to this TaskTracker.
    TaskStatus status = tip.getTaskStatus();
    if (status.getIsMap()) {
      throw new IllegalStateException(
          "Invalid AllMapsCompletedTaskAction, task attempt "
              + "to be notified is a map: " + taskId + " " + status);
    }
    if (status.getPhase() != Phase.SHUFFLE) {
      throw new IllegalArgumentException(
          "Reducer task attempt already notified: " + taskId + " " + status);
    }
           
    // Warning: setPhase() uses System.currentTimeMillis() internally to
    // set shuffle and sort times, but we overwrite that manually anyway
    status.setPhase(Phase.REDUCE);
    status.setShuffleFinishTime(now);
    status.setSortFinishTime(now);
    
    // Forecast the completion of this reduce
    TaskAttemptCompletionEvent e = createTaskAttemptCompletionEvent(tip, now);
    return Collections.<SimulatorEvent>singletonList(e);
  }
  
  /** 
   * Updates the progress indicator of a task if it is running.
   * 
   * @param tip simulator task in progress whose progress is to be updated
   * @param now current simulation time
   */
  private void progressTaskStatus(SimulatorTaskInProgress tip, long now) {
    TaskStatus status = tip.getTaskStatus();
    if (status.getRunState() != State.RUNNING) {
      return; // nothing to be done
    }

    boolean isMap = tip.isMapTask();
    // Time when the user space code started
    long startTime = -1;
    // Time spent in map or just in the REDUCE phase of a reduce task
    long runTime = tip.getUserSpaceRunTime();
    float progress = 0.0f;
    if (isMap) {
      // We linearly estimate the progress of maps since their start 
      startTime = status.getStartTime();
      progress = ((float)(now - startTime)) / runTime;
    } else {
      // We don't model reduce progress in the SHUFFLE or SORT phases
      // We use linear estimate for the 3rd, REDUCE phase
      Phase reducePhase = status.getPhase();
      switch (reducePhase) {
      case SHUFFLE:
        progress = 0.0f; // 0 phase is done out of 3
        break;
      case SORT:
        progress = 1.0f/3; // 1 phase is done out of 3
        break;
      case REDUCE: {
        // REDUCE phase with the user code started when sort finished
        startTime = status.getSortFinishTime();
        // 0.66f : 2 phases are done out of of 3
        progress = 2.0f/3 + (((float) (now - startTime)) / runTime) / 3.0f;
      }
        break;
      default:
        // should never get here
        throw new IllegalArgumentException("Invalid reducePhase=" + reducePhase);
      }
    }
    
    final float EPSILON = 0.0001f;
    if (progress < -EPSILON || progress > 1 + EPSILON) {
      throw new IllegalStateException("Task progress out of range: " + progress);
    }
    progress = Math.max(Math.min(1.0f, progress), 0.0f);
    status.setProgress(progress);
    
    if (LOG.isDebugEnabled()) {
      LOG.debug("Updated task progress, taskId=" + status.getTaskID()
          + ", progress=" + status.getProgress());
    }
  }
  
  /**
   * Updates the progress indicator of all running tasks. 
   *
   * @param now current simulation time
   */
  private void progressTaskStatuses(long now) {
    for (SimulatorTaskInProgress tip : tasks.values()) {
      progressTaskStatus(tip, now);
    }
  }

  /** 
   * Frees up bookkeping memory used by completed tasks. 
   * Has no effect on the events or logs produced by the SimulatorTaskTracker.
   * We need this in order not to report completed task multiple times and 
   * to ensure that we do not run out of Java heap memory in larger 
   * simulations.
   */
  private void garbageCollectCompletedTasks() {
    for (Iterator<TaskAttemptID> iter = tasks.keySet().iterator();
         iter.hasNext();) {
      TaskAttemptID taskId = iter.next();
      SimulatorTaskInProgress tip = tasks.get(taskId);
      if (tip.getTaskStatus().getRunState() != State.RUNNING) {
        iter.remove();
        if (LOG.isDebugEnabled()) {
          LOG.debug("Garbage collected SimulatorTIP, taskId=" + taskId);
        }
        // We don't have to / must not touch usedMapSlots and usedReduceSlots
        // as those were already updated by processTaskAttemptCompletionEvent() 
        // when the task switched its state from running
      }
    }
  }
  
  /**
   * Creates a list of task statuses suitable for transmission via heartbeat().
   * The task statuses are cloned() so that the heartbeat() callee, the job 
   * tracker, can't mess up the SimulatorTaskTracker's internal data.
   *
   * @return the list of running and recently completed task statuses 
   * on the tracker
   */
  private List<TaskStatus> collectAndCloneTaskStatuses() {
    ArrayList<TaskStatus> statuses = new ArrayList<TaskStatus>();
    for (SimulatorTaskInProgress tip : tasks.values()) {
      statuses.add((TaskStatus)tip.getTaskStatus().clone());
    }
    return statuses;
  }

  /**
   * Handles the HeartbeatResponse received from the job tracker upon 
   * heartbeat(). Dispatches to handle*Action() methods.
   *
   * @param response HeartbeatResponse received from the job tracker
   * @param now current simulation time
   * @return list of new events generated in response to the task actions
   */
  private List<SimulatorEvent> handleHeartbeatResponse(HeartbeatResponse response,
                                              long now) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Handling heartbeat response " + response);
    }

    List<SimulatorEvent> events = new ArrayList<SimulatorEvent>();  
    TaskTrackerAction[] actions = response.getActions();
    for (TaskTrackerAction action : actions) {
      List<SimulatorEvent> actionEvents;
      if (action instanceof SimulatorLaunchTaskAction) {
        actionEvents = handleSimulatorLaunchTaskAction(
            (SimulatorLaunchTaskAction)action, now);            
      } else if(action instanceof KillTaskAction) {
        actionEvents = handleKillTaskAction((KillTaskAction)action, now);
      } else if(action instanceof AllMapsCompletedTaskAction) {
        // our extra task action for notifying the reducers
        actionEvents = handleAllMapsCompletedTaskAction(
            (AllMapsCompletedTaskAction)action, now);
      } else {
        // Should never get here.
        // CommitTaskAction is not implemented in the simulator
        // LaunchTaskAction has to be SimulatorLaunchTaskAction
        throw new UnsupportedOperationException("Unimplemented TaskAction: "
            + action);
      }
      events.addAll(actionEvents);
    }
    return events;
  }
  
  /** 
   * Transmits a heartbeat event to the jobtracker and processes the response.
   *
   * @param event HeartbeatEvent to process
   * @return list of new events generated in response
   */
  private List<SimulatorEvent> processHeartbeatEvent(HeartbeatEvent event) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Processing heartbeat event " + event);
    }
    
    long now = event.getTimeStamp();
    
    // Create the TaskTrackerStatus to report
    progressTaskStatuses(now);
    List<TaskStatus> taskStatuses = collectAndCloneTaskStatuses();
    boolean askForNewTask = (usedMapSlots < maxMapSlots ||
                             usedReduceSlots < maxReduceSlots);

    // 0 means failures == 0 here. Undocumented in TaskTracker, but does not 
    // seem to be used at all in org.apache.hadoop.mapred .
    TaskTrackerStatus taskTrackerStatus = 
      new SimulatorTaskTrackerStatus(taskTrackerName, hostName, httpPort, 
          taskStatuses, 0, 
          maxMapSlots, maxReduceSlots, now);
    
    // This is the right, and only, place to release bookkeping memory held 
    // by completed tasks: after collectAndCloneTaskStatuses() and before 
    // heartbeat().
    // The status of TIPs to be purged is already cloned & copied to
    // taskStatuses for reporting
    // We shouldn't run the gc after heartbeat() since  KillTaskAction might 
    // produce new completed tasks that we have not yet reported back and 
    // don't want to purge immediately.
    garbageCollectCompletedTasks();
    
    // Transmit the heartbeat 
    HeartbeatResponse response = null;
    try {
      response = 
        jobTracker.heartbeat(taskTrackerStatus, false, firstHeartbeat, 
                             askForNewTask, heartbeatResponseId);
    } catch (IOException ioe) {
      throw new IllegalStateException("Internal error", ioe);
    }
    firstHeartbeat = false;

    // The heartbeat got through successfully!
    heartbeatResponseId = response.getResponseId();

    // Process the heartbeat response
    List<SimulatorEvent> events = handleHeartbeatResponse(response, now);
    
    // Next heartbeat
    int heartbeatInterval = response.getHeartbeatInterval();
    if (heartbeatIntervalFuzz > 1) {
      // Add some randomness to heartbeat timings to simulate network latency, 
      // time spent servicing this heartbeat request, etc.
      // randomFuzz is in (-heartbeatIntervalFuzz,+heartbeatIntervalFuzz)
      int randomFuzz = random.nextInt(2*heartbeatIntervalFuzz-1) - 
                       heartbeatIntervalFuzz;
      heartbeatInterval += randomFuzz;
      // make sure we never schedule a heartbeat in the past
      heartbeatInterval = Math.max(1, heartbeatInterval); 
    }
    events.add(new HeartbeatEvent(this, now + heartbeatInterval));

    return events;
  }
  
  /**
   * Internal helper class used for storing the current status and other
   * auxilliary information associated with a task attempt assigned to
   * a simulator task tracker.
   * WARNING: This is a completely different inner class than the one with
   *          the same name in SimulatorJobTracker.
   */
  static class SimulatorTaskInProgress {
  
    /**
     * Current status of the task attempt. 
     * We store the start time, the start time of reduce phases and the
     * run state of the task in this object.
     */
    private TaskStatus taskStatus;
    
    /** 
     * Object storing the run time and the final state of the task attempt.
     * It is never read directly by the SimulatorTaskTracker.
     */
    private TaskAttemptInfo taskAttempInfo;

    /**
     * Runtime of the user-space code of the task attempt. This is the full
     * runtime for map tasks, and only that of the REDUCE phase for reduce
     * tasks.
     */ 
    private final long userSpaceRunTime;
    
    /** 
     * Constructs an object by copying most of the fields from a
     * SimulatorTaskAction.
     */
    public SimulatorTaskInProgress(SimulatorLaunchTaskAction action,
                                   TaskStatus taskStatus, long now) {
      this.taskStatus = taskStatus;
      this.taskAttempInfo = action.getTaskAttemptInfo();
      if (taskStatus.getIsMap()) {
        this.userSpaceRunTime = taskAttempInfo.getRuntime();
      } else {
        this.userSpaceRunTime = 
          ((ReduceTaskAttemptInfo)taskAttempInfo).getReduceRuntime();
      }
    }
    
    /**
     * Returns whether the task attempt is a map. 
     * 
     * @return true iff the task attempt is a map
     */
    public boolean isMapTask() {
      return taskStatus.getIsMap();
    }

    /*
     * Returns the current status of the task attempt. 
     *
     * @return current task status
     */
    public TaskStatus getTaskStatus() {
      return taskStatus;
    }
    
    /** 
     * Sets the status of the task attempt.
     *
     * @param status the new task status
     */
    public void setTaskStatus(TaskStatus status) {
      this.taskStatus = status;
    }
        
    /** 
     * Returns the final state of the completed task.
     * 
     * @return the final state of the completed task; 
     *        it is either State.SUCCEEDED or State.FAILED
     */
    public State getFinalRunState() {
      return taskAttempInfo.getRunState();
    }
    
    /**
     * Gets the time spent in the user space code of the task attempt.
     * This is the full runtime for map tasks, and only that of the REDUCE 
     * phase for reduce tasks.
     *
     * @return the user space runtime 
     */
    public long getUserSpaceRunTime() {
      return userSpaceRunTime;
    }          
  }
}