HADOOP-305. Fix tasktracker to ask for new tasks as soon as a task is finished. Contributed by Mahadev.

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

CHANGES.txt

@@ -52,9 +52,14 @@ Trunk (unreleased changes)
     error.  (omalley via cutting)
 
 13. HADOOP-304.  Improve error message for
-    UnregisterdDatanodeException, adding expected node name.
+    UnregisterdDatanodeException to include expected node name.
    (Konstantin Shvachko via cutting)
 
+14. HADOOP-305.  Fix TaskTracker to ask for new tasks as soon as a
+    task is finished, rather than waiting for the next heartbeat.
+    This improves performance when tasks are short.
+    (Mahadev Konar via cutting)
+
 
 Release 0.3.2 - 2006-06-09
 

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

@@ -79,7 +79,7 @@ public class TaskTracker
 
     private int maxCurrentTasks;
     private int failures;
-    
+    private int finishedCount[] = new int[1];
     /**
      * A list of tips that should be cleaned up.
      */
@@ -284,11 +284,16 @@ public class TaskTracker
             long waitTime = HEARTBEAT_INTERVAL - (now - lastHeartbeat);
             if (waitTime > 0) {
                 try {
-                    Thread.sleep(waitTime);
+                    // sleeps for the wait time, wakes up if a task is finished.
+		    synchronized(finishedCount) {
+                        if (finishedCount[0] == 0) {
+			    finishedCount.wait(waitTime);
+                        }
+                        finishedCount[0] = 0;
+                    }
                 } catch (InterruptedException ie) {
-                }
-                continue;
-            }
+               }
+	    }
             lastHeartbeat = now;
 
             //
@@ -856,6 +861,10 @@ public class TaskTracker
         }
         if (tip != null) {
           tip.taskFinished();
+          synchronized(finishedCount) {
+              finishedCount[0]++;
+              finishedCount.notifyAll();
+          }
         } else {
           LOG.warn("Unknown child task finshed: "+taskid+". Ignored.");
         }

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

@@ -114,10 +114,12 @@ class TaskTrackerStatus implements Writable {
         int mapCount = 0;
         for (Iterator it = taskReports.iterator(); it.hasNext(); ) {
             TaskStatus ts = (TaskStatus) it.next();
-            if (ts.getIsMap()) {
-                mapCount++;
-            }
-        }
+	    if (ts.getIsMap() &&
+		((ts.getRunState() == TaskStatus.RUNNING)
+                 || (ts.getRunState() == TaskStatus.UNASSIGNED))) {
+		mapCount++;
+	    }
+	}
         return mapCount;
     }
 
@@ -125,7 +127,16 @@ class TaskTrackerStatus implements Writable {
      * Return the current ReduceTask count
      */
     public int countReduceTasks() {
-        return taskReports.size() - countMapTasks();
+	int reduceCount = 0;
+        for (Iterator it = taskReports.iterator(); it.hasNext(); ) {
+            TaskStatus ts = (TaskStatus) it.next();
+            if ((!ts.getIsMap()) &&
+		((ts.getRunState() == TaskStatus.RUNNING)
+		 || (ts.getRunState() == TaskStatus.UNASSIGNED))) {
+                reduceCount++;
+            }
+        }
+        return reduceCount;
     }
 
     /**