Fix HADOOP-150. Improved task names that include job name.

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

CHANGES.txt

@@ -80,6 +80,8 @@ Trunk (unreleased)
     retries when creating a file long enough for locks on abandoned
     files to expire.  (omalley via cutting)
 
+22. Fix HADOOP-150.  Improved task names that include job
+    names. (omalley via cutting)
 
 Release 0.1.1 - 2006-04-08
 

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

@@ -55,6 +55,7 @@ class JobInProgress {
     boolean tasksInited = false;
 
     private LocalFileSystem localFs;
+    private String uniqueString;
   
     /**
      * Create a JobInProgress with the given job file, plus a handle
@@ -62,7 +63,8 @@ class JobInProgress {
      */
     public JobInProgress(String jobFile, JobTracker jobtracker, 
                          Configuration default_conf) throws IOException {
-        String jobid = "job_" + jobtracker.createUniqueId();
+        uniqueString = jobtracker.createUniqueId();
+        String jobid = "job_" + uniqueString;
         String url = "http://" + jobtracker.getJobTrackerMachine() + ":" + jobtracker.getInfoPort() + "/jobdetails.jsp?jobid=" + jobid;
         this.jobtracker = jobtracker;
         this.status = new JobStatus(jobid, 0.0f, 0.0f, JobStatus.PREP);
@@ -101,7 +103,6 @@ class JobInProgress {
         //
         // construct input splits
         //
-        String jobid = profile.getJobId();
         String jobFile = profile.getJobFile();
 
         JobConf jd = new JobConf(localJobFile);
@@ -141,8 +142,8 @@ class JobInProgress {
         // create a map task for each split
         this.maps = new TaskInProgress[numMapTasks];
         for (int i = 0; i < numMapTasks; i++) {
-            maps[i] = new TaskInProgress(jobFile, splits[i], jobtracker, conf, 
-                                         this, i);
+            maps[i] = new TaskInProgress(uniqueString, jobFile, splits[i], 
+                                         jobtracker, conf, this, i);
         }
 
         //
@@ -150,7 +151,8 @@ class JobInProgress {
         //
         this.reduces = new TaskInProgress[numReduceTasks];
         for (int i = 0; i < numReduceTasks; i++) {
-            reduces[i] = new TaskInProgress(jobFile, maps, i, jobtracker, conf, this);
+            reduces[i] = new TaskInProgress(uniqueString, jobFile, maps, i, 
+                                            jobtracker, conf, this);
         }
 
         //

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

@@ -23,6 +23,7 @@ import org.apache.hadoop.util.LogFormatter;
 
 import java.io.*;
 import java.net.*;
+import java.text.NumberFormat;
 import java.util.*;
 import java.util.logging.*;
 
@@ -40,6 +41,16 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     static float PAD_FRACTION;
     static float MIN_SLOTS_FOR_PADDING;
 
+    /**
+     * Used for formatting the id numbers
+     */
+    private static NumberFormat idFormat = NumberFormat.getInstance();
+    static {
+      idFormat.setMinimumIntegerDigits(4);
+      idFormat.setGroupingUsed(false);
+    }
+    private int nextJobId = 1;
+
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.mapred.JobTracker");
 
     private static JobTracker tracker = null;
@@ -847,10 +858,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
         return (JobInProgress) jobs.get(jobid);
     }
     /**
-     * Grab random num for task id
+     * Grab random num for job id
      */
     String createUniqueId() {
-        return "" + Integer.toString(Math.abs(r.nextInt()),36);
+        return idFormat.format(nextJobId++);
     }
 
     ////////////////////////////////////////////////////

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

@@ -20,6 +20,7 @@ import org.apache.hadoop.conf.*;
 import org.apache.hadoop.util.LogFormatter;
 
 import java.io.*;
+import java.text.NumberFormat;
 import java.util.*;
 import java.util.logging.*;
 
@@ -43,6 +44,11 @@ class TaskInProgress {
     static final int MAX_TASK_FAILURES = 4;    
     static final double SPECULATIVE_GAP = 0.2;
     static final long SPECULATIVE_LAG = 60 * 1000;
+    private static NumberFormat idFormat = NumberFormat.getInstance();
+    static {
+      idFormat.setMinimumIntegerDigits(6);
+      idFormat.setGroupingUsed(false);
+    }
 
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.mapred.TaskInProgress");
 
@@ -77,7 +83,7 @@ class TaskInProgress {
     /**
      * Constructor for MapTask
      */
-    public TaskInProgress(String jobFile, FileSplit split, 
+    public TaskInProgress(String uniqueString, String jobFile, FileSplit split, 
                           JobTracker jobtracker, JobConf conf, 
                           JobInProgress job, int partition) {
         this.jobFile = jobFile;
@@ -86,13 +92,14 @@ class TaskInProgress {
         this.job = job;
         this.conf = conf;
         this.partition = partition;
-        init();
+        init(uniqueString);
     }
         
     /**
      * Constructor for ReduceTask
      */
-    public TaskInProgress(String jobFile, TaskInProgress predecessors[], 
+    public TaskInProgress(String uniqueString, String jobFile, 
+                          TaskInProgress predecessors[], 
                           int partition, JobTracker jobtracker, JobConf conf,
                           JobInProgress job) {
         this.jobFile = jobFile;
@@ -101,23 +108,37 @@ class TaskInProgress {
         this.jobtracker = jobtracker;
         this.job = job;
         this.conf = conf;
-        init();
+        init(uniqueString);
     }
 
+    /**
+     * Make a unique name for this TIP.
+     * @param uniqueBase The unique name of the job
+     * @return The unique string for this tip
+     */
+    private String makeUniqueString(String uniqueBase) {
+      StringBuffer result = new StringBuffer();
+      result.append(uniqueBase);
+      if (isMapTask()) {
+        result.append("_m_");
+      } else {
+        result.append("_r_");
+      }
+      result.append(idFormat.format(partition));
+      return result.toString();
+    }
+    
     /**
      * Initialization common to Map and Reduce
      */
-    void init() {
+    void init(String jobUniqueString) {
         this.startTime = System.currentTimeMillis();
-        this.id = "tip_" + jobtracker.createUniqueId();
+        String uniqueString = makeUniqueString(jobUniqueString);
+        this.id = "tip_" + uniqueString;
         this.totalTaskIds = new String[MAX_TASK_EXECS + MAX_TASK_FAILURES];
         for (int i = 0; i < totalTaskIds.length; i++) {
-            if (isMapTask()) {
-                totalTaskIds[i] = "task_m_" + jobtracker.createUniqueId();
-            } else {
-                totalTaskIds[i] = "task_r_" + jobtracker.createUniqueId();
-            }
-            usableTaskIds.add(totalTaskIds[i]);
+          totalTaskIds[i] = "task_" + uniqueString + "_" + i;
+          usableTaskIds.add(totalTaskIds[i]);
         }
     }