HADOOP-173. Optimize allocation of tasks with local data.

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

CHANGES.txt

@@ -114,7 +114,9 @@ Trunk (unreleased)
 
 30. Permit specification of a higher replication levels for job
     submission files (job.xml and job.jar).  This helps with large
-    clusters, since these files are read by every node.
+    clusters, since these files are read by every node.  (cutting)
+
+31. HADOOP-173.  Optimize allocation of tasks with local data.  (cutting)
 
 
 Release 0.1.1 - 2006-04-08

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

@@ -44,7 +44,7 @@ class JobInProgress {
     int numReduceTasks = 0;
 
     JobTracker jobtracker = null;
-    TreeMap cachedHints = new TreeMap();
+    HashMap hostToMaps = new HashMap();
 
     long startTime;
     long finishTime;
@@ -159,29 +159,28 @@ class JobInProgress {
         // Obtain some tasktracker-cache information for the map task splits.
         //
         for (int i = 0; i < maps.length; i++) {
-            String hints[][] = fs.getFileCacheHints(splits[i].getPath(), splits[i].getStart(), splits[i].getLength());
-            cachedHints.put(maps[i].getTIPId(), hints);
+            String hints[][] =
+              fs.getFileCacheHints(splits[i].getPath(), splits[i].getStart(),
+                                   splits[i].getLength());
+
+            if (hints != null) {
+              for (int k = 0; k < hints.length; k++) {
+                for (int j = 0; j < hints[k].length; j++) {
+                  ArrayList hostMaps = (ArrayList)hostToMaps.get(hints[k][j]);
+                  if (hostMaps == null) {
+                    hostMaps = new ArrayList();
+                    hostToMaps.put(hints[k][j], hostMaps);
+                  }
+                  hostMaps.add(maps[i]);
+                }
+              }
+            }
         }
 
         this.status = new JobStatus(status.getJobId(), 0.0f, 0.0f, JobStatus.RUNNING);
         tasksInited = true;
     }
 
-    /**
-     * This is called by TaskInProgress objects.  The JobInProgress
-     * prefetches and caches a lot of these hints.  If the hint is
-     * not available, then we pass it through to the filesystem.
-     */
-    String[][] getFileCacheHints(String tipID, Path f, long start, long len) throws IOException {
-        String results[][] = (String[][]) cachedHints.get(tipID);
-        if (tipID == null) {
-            FileSystem fs = FileSystem.get(conf);
-            results = fs.getFileCacheHints(f, start, len);
-            cachedHints.put(tipID, results);
-        }
-        return results;
-    }
-
     /////////////////////////////////////////////////////
     // Accessors for the JobInProgress
     /////////////////////////////////////////////////////
@@ -301,20 +300,18 @@ class JobInProgress {
         // the TaskTracker checking in.  That means the block
         // doesn't have to be transmitted from another node.
         //
-        for (int i = 0; i < maps.length; i++) {
-            int realIdx = (i + firstMapToTry) % maps.length; 
-            if (maps[realIdx].hasTaskWithCacheHit(taskTracker, tts)) {
-                if (cacheTarget < 0) {
-                  if (maps[realIdx].hasFailedOnMachine(taskTracker)) {
-                    if (failedTarget < 0) {
-                      failedTarget = realIdx;
-                    }
-                  } else {
-                    cacheTarget = realIdx;
-                    break;
-                  }
-                }
+        ArrayList hostMaps = (ArrayList)hostToMaps.get(tts.getHost());
+        if (hostMaps != null) {
+          Iterator i = hostMaps.iterator();
+          while (i.hasNext()) {
+            TaskInProgress tip = (TaskInProgress)i.next();
+            if (tip.hasTask() && !tip.hasFailedOnMachine(taskTracker)) {
+              LOG.info("Found task with local split for "+tts.getHost());
+              cacheTarget = tip.getIdWithinJob();
+              i.remove();
+              break;
             }
+          }
         }
 
         //

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

@@ -55,7 +55,6 @@ class TaskInProgress {
     // Defines the TIP
     private String jobFile = null;
     private FileSplit split = null;
-    private String hints[][] = null;
     private TaskInProgress predecessors[] = null;
     private int partition;
     private JobTracker jobtracker;
@@ -387,34 +386,7 @@ class TaskInProgress {
     // "Action" methods that actually require the TIP
     // to do something.
     /////////////////////////////////////////////////
-    /**
-     * Return whether this TIP has an DFS cache-driven task 
-     * to run at the given taskTracker.
-     */
-    boolean hasTaskWithCacheHit(String taskTracker, TaskTrackerStatus tts) {
-        if (failed || isComplete() || recentTasks.size() > 0) {
-            return false;
-        } else {
-            try {
-                if (isMapTask()) {
-                    if (hints == null) {
-                        hints = job.getFileCacheHints(getTIPId(), split.getPath(), split.getStart(), split.getLength());
-                    }
-                    if (hints != null) {
-                        for (int i = 0; i < hints.length; i++) {
-                            for (int j = 0; j < hints[i].length; j++) {
-                                if (hints[i][j].equals(tts.getHost())) {
-                                    return true;
-                                }
-                            }
-                        }
-                    }
-                }
-            } catch (IOException ie) {
-            }
-            return false;
-        }
-    }
+
     /**
      * Return whether this TIP has a non-speculative task to run
      */
@@ -457,8 +429,7 @@ class TaskInProgress {
      */
     public Task getTaskToRun(String taskTracker, TaskTrackerStatus tts, double avgProgress) {
         Task t = null;
-        if (hasTaskWithCacheHit(taskTracker, tts) ||
-            hasTask() || 
+        if (hasTask() || 
             hasSpeculativeTask(avgProgress)) {
 
             String taskid = (String) usableTaskIds.first();