HADOOP-949. svn merge -r 500410:500397, reverting a patch from HADOOP-248, which was causing MapReduce programs to hang.

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

CHANGES.txt

@@ -87,8 +87,7 @@ Trunk (unreleased changes)
     with different versions of Lucene without worrying about CLASSPATH
     order.  (Milind Bhandarkar via cutting)
 
-27. HADOOP-248.  Optimize location of map outputs to no longer use
-    random probes.  (Devaraj Das via cutting)
+27. [ intentionally blank ]
 
 28. HADOOP-227.  Add support for backup namenodes, which periodically
     get snapshots of the namenode state.  (Dhruba Borthakur via cutting) 

src/java/org/apache/hadoop/mapred/InterTrackerProtocol.java

@@ -28,10 +28,11 @@ import org.apache.hadoop.ipc.VersionedProtocol;
  */ 
 interface InterTrackerProtocol extends VersionedProtocol {
   /**
-   * version 4 introduced that removes locateMapOutputs and instead uses
-   * getTaskCompletionEvents to figure finished maps and fetch the outputs
+   * version 3 introduced to replace 
+   * emitHearbeat/pollForNewTask/pollForTaskWithClosedJob with
+   * {@link #heartbeat(TaskTrackerStatus, boolean, boolean, short)}
    */
-  public static final long versionID = 4L;
+  public static final long versionID = 3L;
   
   public final static int TRACKERS_OK = 0;
   public final static int UNKNOWN_TASKTRACKER = 1;
@@ -61,6 +62,18 @@ interface InterTrackerProtocol extends VersionedProtocol {
           boolean initialContact, boolean acceptNewTasks, short responseId)
   throws IOException;
 
+  /** Called by a reduce task to find which map tasks are completed.
+   *
+   * @param jobId the job id
+   * @param mapTasksNeeded an array of the mapIds that we need
+   * @param partition the reduce's id
+   * @return an array of MapOutputLocation
+   */
+  MapOutputLocation[] locateMapOutputs(String jobId, 
+                                       int[] mapTasksNeeded,
+                                       int partition
+                                       ) throws IOException;
+
   /**
    * The task tracker calls this once, to discern where it can find
    * files referred to by the JobTracker
@@ -83,12 +96,11 @@ interface InterTrackerProtocol extends VersionedProtocol {
    * Returns empty aray if no events are available. 
    * @param jobid job id 
    * @param fromEventId event id to start from. 
-   * @param maxEvents the max number of events we want to look at
    * @return array of task completion events. 
    * @throws IOException
    */
   TaskCompletionEvent[] getTaskCompletionEvents(
-      String jobid, int fromEventId, int maxEvents) throws IOException;
+      String jobid, int fromEventId) throws IOException; 
   
 }
 

src/java/org/apache/hadoop/mapred/JobClient.java

@@ -157,7 +157,7 @@ public class JobClient extends ToolBase implements MRConstants  {
         public synchronized TaskCompletionEvent[] getTaskCompletionEvents(
             int startFrom) throws IOException{
           return jobSubmitClient.getTaskCompletionEvents(
-              getJobID(), startFrom, 10); 
+              getJobID(), startFrom); 
         }
 
         /**

src/java/org/apache/hadoop/mapred/JobInProgress.java

@@ -291,40 +291,24 @@ class JobInProgress {
           
           if (state == TaskStatus.State.SUCCEEDED) {
             this.taskCompletionEvents.add( new TaskCompletionEvent(
-                taskCompletionEventTracker, 
-                status.getTaskId(),
-                tip.idWithinJob(),
-                status.getIsMap(),
+                taskCompletionEventTracker++, 
+                status.getTaskId(), 
                 TaskCompletionEvent.Status.SUCCEEDED,
                 httpTaskLogLocation ));
-            tip.setSuccessEventNumber(taskCompletionEventTracker);
             completedTask(tip, status, metrics);
           } else if (state == TaskStatus.State.FAILED ||
                      state == TaskStatus.State.KILLED) {
             this.taskCompletionEvents.add( new TaskCompletionEvent(
-                taskCompletionEventTracker, 
-                status.getTaskId(),
-                tip.idWithinJob(),
-                status.getIsMap(),
+                taskCompletionEventTracker++, 
+                status.getTaskId(), 
                 TaskCompletionEvent.Status.FAILED, 
                 httpTaskLogLocation ));
-            // Get the event number for the (possibly) previously successful
-            // task. If there exists one, then set that status to OBSOLETE 
-            int eventNumber;
-            if ((eventNumber = tip.getSuccessEventNumber()) != -1) {
-              TaskCompletionEvent t = 
-                this.taskCompletionEvents.get(eventNumber);
-              if (t.getTaskId().equals(status.getTaskId()))
-                t.setTaskStatus(TaskCompletionEvent.Status.OBSOLETE);
-            }
             // Tell the job to fail the relevant task
             failedTask(tip, status.getTaskId(), status, status.getTaskTracker(),
                        wasRunning, wasComplete);
           }          
         }
 
-        taskCompletionEventTracker++;
-        
         //
         // Update JobInProgress status
         //
@@ -794,14 +778,12 @@ class JobInProgress {
        return null;
     }
     
-    public TaskCompletionEvent[] getTaskCompletionEvents(int fromEventId, 
-        int maxEvents) {
+    public TaskCompletionEvent[] getTaskCompletionEvents(int fromEventId) {
       TaskCompletionEvent[] events = TaskCompletionEvent.EMPTY_ARRAY;
       if( taskCompletionEvents.size() > fromEventId) {
-        int actualMax = Math.min(maxEvents, 
-            (taskCompletionEvents.size() - fromEventId));
         events = (TaskCompletionEvent[])taskCompletionEvents.subList(
-            fromEventId, actualMax + fromEventId).toArray(events);        
+            fromEventId, taskCompletionEvents.size()).
+            toArray(events);        
       }
       return events; 
     }

src/java/org/apache/hadoop/mapred/JobSubmissionProtocol.java

@@ -85,12 +85,11 @@ public interface JobSubmissionProtocol extends VersionedProtocol {
      * Get task completion events for the jobid, starting from fromEventId. 
      * Returns empty aray if no events are available. 
      * @param jobid job id 
-     * @param fromEventId event id to start from.
-     * @param maxEvents the max number of events we want to look at 
+     * @param fromEventId event id to start from. 
      * @return array of task completion events. 
      * @throws IOException
      */
     public TaskCompletionEvent[] getTaskCompletionEvents(
-              String jobid, int fromEventId, int maxEvents) throws IOException;
+                String jobid, int fromEventId) throws IOException;
     
 }

src/java/org/apache/hadoop/mapred/JobTracker.java

@@ -1262,6 +1262,48 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
         return null;
     }
 
+    /**
+     * A TaskTracker wants to know the physical locations of completed, but not
+     * yet closed, tasks.  This exists so the reduce task thread can locate
+     * map task outputs.
+     */
+    public synchronized MapOutputLocation[] 
+             locateMapOutputs(String jobId, int[] mapTasksNeeded, int reduce) 
+    throws IOException {
+        // Check to make sure that the job hasn't 'completed'.
+        JobInProgress job = getJob(jobId);
+        if (job.status.getRunState() != JobStatus.RUNNING) {
+          return new MapOutputLocation[0];
+        }
+        
+        ArrayList result = new ArrayList(mapTasksNeeded.length);
+        for (int i = 0; i < mapTasksNeeded.length; i++) {
+          TaskStatus status = job.findFinishedMap(mapTasksNeeded[i]);
+          if (status != null) {
+             String trackerId = 
+               (String) taskidToTrackerMap.get(status.getTaskId());
+             // Safety check, if we can't find the taskid in 
+             // taskidToTrackerMap and job isn't 'running', then just
+             // return an empty array
+             if (trackerId == null && 
+                     job.status.getRunState() != JobStatus.RUNNING) {
+               return new MapOutputLocation[0];
+             }
+             
+             TaskTrackerStatus tracker;
+             synchronized (taskTrackers) {
+               tracker = (TaskTrackerStatus) taskTrackers.get(trackerId);
+             }
+             result.add(new MapOutputLocation(status.getTaskId(), 
+                                              mapTasksNeeded[i],
+                                              tracker.getHost(), 
+                                              tracker.getHttpPort()));
+          }
+        }
+        return (MapOutputLocation[]) 
+               result.toArray(new MapOutputLocation[result.size()]);
+    }
+
     /**
      * Grab the local fs name
      */
@@ -1393,14 +1435,14 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     /* 
      * Returns a list of TaskCompletionEvent for the given job, 
      * starting from fromEventId.
-     * @see org.apache.hadoop.mapred.JobSubmissionProtocol#getTaskCompletionEvents(java.lang.String, int, int)
+     * @see org.apache.hadoop.mapred.JobSubmissionProtocol#getTaskCompletionEvents(java.lang.String, int)
      */
     public synchronized TaskCompletionEvent[] getTaskCompletionEvents(
-        String jobid, int fromEventId, int maxEvents) throws IOException{
+        String jobid, int fromEventId) throws IOException{
       TaskCompletionEvent[] events = TaskCompletionEvent.EMPTY_ARRAY;
       JobInProgress job = (JobInProgress)this.jobs.get(jobid);
       if (null != job) {
-        events = job.getTaskCompletionEvents(fromEventId, maxEvents);
+        events = job.getTaskCompletionEvents(fromEventId);
       }
       return events;
     }

src/java/org/apache/hadoop/mapred/LocalJobRunner.java

@@ -244,7 +244,7 @@ class LocalJobRunner implements JobSubmissionProtocol {
 
   public JobStatus[] jobsToComplete() {return null;}
   public TaskCompletionEvent[] getTaskCompletionEvents(
-      String jobid, int fromEventId, int maxEvents) throws IOException{
+      String jobid, int fromEventId) throws IOException{
     return TaskCompletionEvent.EMPTY_ARRAY;
   }
   

src/java/org/apache/hadoop/mapred/ReduceTaskRunner.java

@@ -122,19 +122,9 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
   
   /**
    * the number of map output locations to poll for at one time
-   */  
-  private int probe_sample_size = 50;
-  
-  /**
-   * a Random used during the map output fetching
    */
-  private Random randForProbing;
-  
-  /**
-   * a hashmap from mapId to MapOutputLocation for retrials
-   */
-  private Map<Integer, MapOutputLocation> retryFetches = new HashMap();
-  
+  private static final int PROBE_SAMPLE_SIZE = 50;
+
   /** Represents the result of an attempt to copy a map output */
   private class CopyResult {
     
@@ -378,10 +368,6 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
     Random         backoff = new Random();
     final Progress copyPhase = getTask().getProgress().phase();
     
-    //tweak the probe sample size (make it a function of numCopiers)
-    probe_sample_size = Math.max(numCopiers*5, 50);
-    randForProbing = new Random(reduceTask.getPartition() * 100);
-    
     for (int i = 0; i < numOutputs; i++) {
       neededOutputs.add(new Integer(i));
       copyPhase.addPhase();       // add sub-phase per file
@@ -399,8 +385,6 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
     // start the clock for bandwidth measurement
     long startTime = System.currentTimeMillis();
     long currentTime = startTime;
-    int fromEventId = 0;
-    
     PingTimer pingTimer = new PingTimer();
     pingTimer.setName("Map output copy reporter for task " + 
                       reduceTask.getTaskId());
@@ -417,36 +401,17 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
         LOG.info(reduceTask.getTaskId() + " Need " + neededOutputs.size() +
                  " map output location(s)");
         try {
-          // the call to queryJobTracker will modify fromEventId to a value
-          // that it should be for the next call to queryJobTracker
-          MapOutputLocation[] locs = queryJobTracker(fromEventId, jobClient);
+          MapOutputLocation[] locs = queryJobTracker(neededOutputs, jobClient);
           
           // remove discovered outputs from needed list
           // and put them on the known list
-          int gotLocs = (locs == null ? 0 : locs.length);
           for (int i=0; i < locs.length; i++) {
-            // check whether we actually need an output. It could happen
-            // that a map task that successfully ran earlier got lost, but
-            // if we already have copied the output of that unfortunate task
-            // we need not copy it again from the new TT (we will ignore 
-            // the event for the new rescheduled execution)
-            if(neededOutputs.remove(new Integer(locs[i].getMapId()))) {
-              // remove the mapId from the retryFetches hashmap since we now
-              // prefer the new location instead of what we saved earlier
-              retryFetches.remove(new Integer(locs[i].getMapId()));
-              knownOutputs.add(locs[i]);
-              gotLocs--;
-            }
-            
+            neededOutputs.remove(new Integer(locs[i].getMapId()));
+            knownOutputs.add(locs[i]);
           }
-          // now put the remaining hash entries for the failed fetches 
-          // and clear the hashmap
-          knownOutputs.addAll(retryFetches.values());
           LOG.info(reduceTask.getTaskId() +
-                " Got " + gotLocs + 
-                " new map outputs from jobtracker and " + retryFetches.size() +
-                " map outputs from previous failures");
-          retryFetches.clear();
+                   " Got " + (locs == null ? 0 : locs.length) + 
+                   " map outputs from jobtracker");
         }
         catch (IOException ie) {
           LOG.warn(reduceTask.getTaskId() +
@@ -520,7 +485,6 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
           } else {
             // this copy failed, put it back onto neededOutputs
             neededOutputs.add(new Integer(cr.getMapId()));
-            retryFetches.put(new Integer(cr.getMapId()), cr.getLocation());
           
             // wait a random amount of time for next contact
             currentTime = System.currentTimeMillis();
@@ -540,7 +504,6 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
             while (locIt.hasNext()) {
               MapOutputLocation loc = (MapOutputLocation)locIt.next();
               if (cr.getHost().equals(loc.getHost())) {
-                retryFetches.put(new Integer(loc.getMapId()), loc);
                 locIt.remove();
                 neededOutputs.add(new Integer(loc.getMapId()));
               }
@@ -551,7 +514,7 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
         }
         
         // ensure we have enough to keep us busy
-        if (numInFlight < lowThreshold && (numOutputs-numCopied) > probe_sample_size) {
+        if (numInFlight < lowThreshold && (numOutputs-numCopied) > PROBE_SAMPLE_SIZE) {
           break;
         }
       }
@@ -645,16 +608,28 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
   }
   
   /** Queries the job tracker for a set of outputs ready to be copied
-   * @param fromEventId the first event ID we want to start from, this will be
-   * modified by the call to this method
+   * @param neededOutputs the list of currently unknown outputs
    * @param jobClient the job tracker
    * @return a set of locations to copy outputs from
    * @throws IOException
    */  
-  private MapOutputLocation[] queryJobTracker(int fromEventId, 
+  private MapOutputLocation[] queryJobTracker(List neededOutputs,
                                               InterTrackerProtocol jobClient)
   throws IOException {
     
+    // query for a just a random subset of needed segments so that we don't
+    // overwhelm jobtracker.  ideally perhaps we could send a more compact
+    // representation of all needed, i.e., a bit-vector
+    int     checkSize = Math.min(PROBE_SAMPLE_SIZE, neededOutputs.size());
+    int     neededIds[] = new int[checkSize];
+      
+    Collections.shuffle(neededOutputs);
+      
+    ListIterator itr = neededOutputs.listIterator();
+    for (int i=0; i < checkSize; i++) {
+      neededIds[i] = ((Integer)itr.next()).intValue();
+    }
+
     long currentTime = System.currentTimeMillis();    
     long pollTime = lastPollTime + MIN_POLL_INTERVAL;
     while (currentTime < pollTime) {
@@ -665,28 +640,9 @@ class ReduceTaskRunner extends TaskRunner implements MRConstants {
     }
     lastPollTime = currentTime;
 
-    TaskCompletionEvent t[] = jobClient.getTaskCompletionEvents(
-                                      reduceTask.getJobId().toString(),
-                                      fromEventId,
-                                      probe_sample_size);
-    
-    List <MapOutputLocation> mapOutputsList = new ArrayList();
-    for (int i = 0; i < t.length; i++) {
-      if (t[i].isMap && 
-          t[i].getTaskStatus() == TaskCompletionEvent.Status.SUCCEEDED) {
-        URI u = URI.create(t[i].getTaskTrackerHttp());
-        String host = u.getHost();
-        int port = u.getPort();
-        String taskId = t[i].getTaskId();
-        int mId = t[i].idWithinJob();
-        mapOutputsList.add(new MapOutputLocation(taskId, mId, host, port));
-      }
-    }
-    Collections.shuffle(mapOutputsList, randForProbing);
-    MapOutputLocation[] locations =
-                        new MapOutputLocation[mapOutputsList.size()];
-    fromEventId += t.length;
-    return mapOutputsList.toArray(locations);
+    return jobClient.locateMapOutputs(reduceTask.getJobId().toString(), 
+                                      neededIds,
+                                      reduceTask.getPartition());
   }
 
   

src/java/org/apache/hadoop/mapred/TaskCompletionEvent.java

@@ -12,14 +12,12 @@ import org.apache.hadoop.io.WritableUtils;
  *
  */
 public class TaskCompletionEvent implements Writable{
-    static public enum Status {FAILED, SUCCEEDED, OBSOLETE};
+    static public enum Status {FAILED, SUCCEEDED};
     
     private int eventId ; 
     private String taskTrackerHttp ;
     private String taskId ;
     Status status ; 
-    boolean isMap = false ;
-    private int idWithinJob;
     public static final TaskCompletionEvent[] EMPTY_ARRAY = 
         new TaskCompletionEvent[0];
     /**
@@ -37,15 +35,11 @@ public class TaskCompletionEvent implements Writable{
      * @param taskTrackerHttp task tracker's host:port for http. 
      */
     public TaskCompletionEvent(int eventId, 
-        String taskId,
-        int idWithinJob,
-        boolean isMap,
+        String taskId, 
         Status status, 
         String taskTrackerHttp){
       
-        this.taskId = taskId ;
-        this.idWithinJob = idWithinJob ;
-        this.isMap = isMap ;
+        this.taskId = taskId ; 
         this.eventId = eventId ; 
         this.status =status ; 
         this.taskTrackerHttp = taskTrackerHttp ;
@@ -120,28 +114,17 @@ public class TaskCompletionEvent implements Writable{
         return buf.toString();
     }
     
-    public boolean isMapTask() {
-        return isMap;
-    }
-    
-    public int idWithinJob() {
-      return idWithinJob;
-    }
     //////////////////////////////////////////////
     // Writable
     //////////////////////////////////////////////
     public void write(DataOutput out) throws IOException {
         WritableUtils.writeString(out, taskId); 
-        WritableUtils.writeVInt(out, idWithinJob);
-        out.writeBoolean(isMap);
         WritableUtils.writeEnum(out, status); 
         WritableUtils.writeString(out, taskTrackerHttp);
     }
   
     public void readFields(DataInput in) throws IOException {
         this.taskId = WritableUtils.readString(in) ; 
-        this.idWithinJob = WritableUtils.readVInt(in);
-        this.isMap = in.readBoolean();
         this.status = WritableUtils.readEnum(in, Status.class);
         this.taskTrackerHttp = WritableUtils.readString(in);
     }

src/java/org/apache/hadoop/mapred/TaskInProgress.java

@@ -60,7 +60,6 @@ class TaskInProgress {
     private JobInProgress job;
 
     // Status of the TIP
-    private int successEventNumber = -1;
     private int numTaskFailures = 0;
     private double progress = 0;
     private String state = "";
@@ -139,15 +138,6 @@ class TaskInProgress {
       return result.toString();
     }
     
-    /**
-     * Return the index of the tip within the job, so "tip_0002_m_012345"
-     * would return 12345;
-     * @return int the tip index
-     */
-     public int idWithinJob() {
-       return partition;
-     }    
-
     /**
      * Initialization common to Map and Reduce
      */
@@ -578,18 +568,4 @@ class TaskInProgress {
     public int getIdWithinJob() {
       return partition;
     }
-    
-    /**
-     * Set the event number that was raised for this tip
-     */
-    public void setSuccessEventNumber(int eventNumber) {
-      successEventNumber = eventNumber;
-    }
-       
-    /**
-     * Get the event number that was raised for this tip
-     */
-    public int getSuccessEventNumber() {
-      return successEventNumber;
-    }
 }