svn commit: r227062 - /lucene/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java

[mc...@apache.org]

Author: mc
Date: Tue Aug  2 11:43:14 2005
New Revision: 227062

URL: http://svn.apache.org/viewcvs?rev=227062&view=rev
Log:

  The TaskTracker will now kill child tasks that fail to
report progress for a 60-second period.  This will take care
of hung tasks.


Modified:
    lucene/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java

Modified: lucene/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java
URL: http://svn.apache.org/viewcvs/lucene/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java?rev=227062&r1=227061&r2=227062&view=diff
==============================================================================
--- lucene/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java (original)
+++ lucene/nutch/branches/mapred/src/java/org/apache/nutch/mapred/TaskTracker.java Tue Aug  2 11:43:14 2005
@@ -35,7 +35,10 @@
 public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutputProtocol, Runnable {
     private static final int MAX_CURRENT_TASKS = 
     NutchConf.get().getInt("mapred.tasktracker.tasks.maximum", 2);
+
     static final long WAIT_FOR_DONE = 3 * 1000;
+    static final long TASK_MIN_PROGRESS_INTERVAL = 60 * 1000;
+
     static final int STALE_STATE = 1;
 
     public static final Logger LOG =
@@ -221,6 +224,18 @@
             }
 
             //
+            // Kill any tasks that have not reported progress in the last X seconds.
+            //
+            for (Iterator it = runningTasks.values().iterator(); it.hasNext(); ) {
+                TaskInProgress tip = (TaskInProgress) it.next();
+                if ((tip.getRunState() == TaskStatus.RUNNING) &&
+                    (System.currentTimeMillis() - tip.getLastProgressReport() > TASK_MIN_PROGRESS_INTERVAL)) {
+
+                    tip.cleanup();
+                }
+            }
+
+            //
             // Check for any Tasks whose job may have ended
             //
             String toCloseId = jobClient.pollForClosedTask(taskTrackerName);
@@ -280,6 +295,7 @@
         File localTaskDir;
         float progress;
         int runstate;
+        long lastProgressReport;
         StringBuffer diagnosticInfo = new StringBuffer();
         TaskRunner runner;
         boolean done = false;
@@ -289,6 +305,7 @@
          */
         public TaskInProgress(Task task) throws IOException {
             this.task = task;
+            this.lastProgressReport = System.currentTimeMillis();
             this.localTaskDir = new File(localDir, task.getTaskId());
             if (localTaskDir.exists()) {
                 FileUtil.fullyDelete(localTaskDir);
@@ -359,6 +376,19 @@
             LOG.info(task.getTaskId()+" "+p+"% "+state);
             this.progress = p;
             this.runstate = TaskStatus.RUNNING;
+            this.lastProgressReport = System.currentTimeMillis();
+        }
+
+        /**
+         */
+        public long getLastProgressReport() {
+            return lastProgressReport;
+        }
+
+        /**
+         */
+        public int getRunState() {
+            return runstate;
         }
 
         /**