Fix HADOOP-167. Reduce the number of Configuration and JobConf's allocated. Contributed by Owen.

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

src/examples/org/apache/hadoop/examples/Grep.java

@@ -27,8 +27,7 @@ import org.apache.hadoop.mapred.lib.LongSumReducer;
 import org.apache.hadoop.io.UTF8;
 import org.apache.hadoop.io.LongWritable;
 
-import org.apache.hadoop.conf.Configuration;
-
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
 import java.util.Random;
@@ -43,13 +42,11 @@ public class Grep {
       System.exit(-1);
     }
 
-    Configuration defaults = new Configuration();
-
     Path tempDir =
       new Path("grep-temp-"+
                Integer.toString(new Random().nextInt(Integer.MAX_VALUE)));
 
-    JobConf grepJob = new JobConf(defaults, Grep.class);
+    JobConf grepJob = new JobConf(Grep.class);
     grepJob.setJobName("grep-search");
 
     grepJob.setInputPath(new Path(args[0]));
@@ -69,7 +66,7 @@ public class Grep {
 
     JobClient.runJob(grepJob);
 
-    JobConf sortJob = new JobConf(defaults, Grep.class);
+    JobConf sortJob = new JobConf(Grep.class);
     sortJob.setJobName("grep-sort");
 
     sortJob.setInputPath(tempDir);
@@ -86,7 +83,7 @@ public class Grep {
 
     JobClient.runJob(sortJob);
 
-    new JobClient(defaults).getFs().delete(tempDir);
+    FileSystem.get(grepJob).delete(tempDir);
   }
 
 }

src/examples/org/apache/hadoop/examples/WordCount.java

@@ -96,9 +96,7 @@ public class WordCount {
    *                     job tracker.
    */
   public static void main(String[] args) throws IOException {
-    Configuration defaults = new Configuration();
-    
-    JobConf conf = new JobConf(defaults, WordCount.class);
+    JobConf conf = new JobConf(WordCount.class);
     conf.setJobName("wordcount");
  
     // the keys are words (strings)

src/java/org/apache/hadoop/conf/Configuration.java

@@ -19,7 +19,7 @@ package org.apache.hadoop.conf;
 import java.util.*;
 import java.net.URL;
 import java.io.*;
-import java.util.logging.Logger;
+import java.util.logging.*;
 
 import javax.xml.parsers.*;
 
@@ -29,7 +29,7 @@ import javax.xml.transform.Transformer;
 import javax.xml.transform.dom.DOMSource;
 import javax.xml.transform.stream.StreamResult;
 
-import org.apache.hadoop.util.LogFormatter;
+import org.apache.hadoop.util.*;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
@@ -62,12 +62,19 @@ public class Configuration {
 
   /** A new configuration. */
   public Configuration() {
+    if (LOG.isLoggable(Level.FINE)) {
+      LOG.fine(StringUtils.stringifyException(new IOException("config()")));
+    }
     defaultResources.add("hadoop-default.xml");
     finalResources.add("hadoop-site.xml");
   }
 
   /** A new configuration with the same settings cloned from another. */
   public Configuration(Configuration other) {
+    if (LOG.isLoggable(Level.FINE)) {
+      LOG.fine(StringUtils.stringifyException
+                 (new IOException("config(config)")));
+    }
     this.defaultResources = (ArrayList)other.defaultResources.clone();
     this.finalResources = (ArrayList)other.finalResources.clone();
     if (other.properties != null)

src/java/org/apache/hadoop/ipc/RPC.java

@@ -237,7 +237,10 @@ public class RPC {
           implementation.getMethod(call.getMethodName(),
                                    call.getParameterClasses());
 
+        long startTime = System.currentTimeMillis();
         Object value = method.invoke(instance, call.getParameters());
+        long callTime = System.currentTimeMillis() - startTime;
+        LOG.fine("Call: " + call.getMethodName() + " " + callTime);
         if (verbose) log("Return: "+value);
 
         return new ObjectWritable(method.getReturnType(), value);

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

@@ -29,7 +29,6 @@ import java.net.URL;
 import java.net.URLDecoder;
 
 import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.conf.Configuration;
 
@@ -49,10 +48,34 @@ import org.apache.hadoop.mapred.lib.HashPartitioner;
  * of input files, and where the output files should be written. */
 public class JobConf extends Configuration {
 
+  private void initialize() {
+    addDefaultResource("mapred-default.xml");
+  }
+  
+  private void initialize(Class exampleClass) {
+    initialize();
+    String jar = findContainingJar(exampleClass);
+    if (jar != null) {
+      setJar(jar);
+    }   
+  }
+  
+  /**
+   * Construct a map/reduce job configuration.
+   */
   public JobConf() {
-    super();
+    initialize();
   }
-    
+
+  /** 
+   * Construct a map/reduce job configuration.
+   * @param conf a Configuration whose settings will be inherited.
+   * @param exampleClass a class whose containing jar is used as the job's jar.
+   */
+  public JobConf(Class exampleClass) {
+    initialize(exampleClass);
+  }
+  
   /**
    * Construct a map/reduce job configuration.
    * 
@@ -61,21 +84,18 @@ public class JobConf extends Configuration {
    */
   public JobConf(Configuration conf) {
     super(conf);
-    addDefaultResource("mapred-default.xml");
+    initialize();
   }
 
 
   /** Construct a map/reduce job configuration.
    * 
    * @param conf a Configuration whose settings will be inherited.
-   * @param aClass a class whose containing jar is used as the job's jar.
+   * @param exampleClass a class whose containing jar is used as the job's jar.
    */
-  public JobConf(Configuration conf, Class aClass) {
+  public JobConf(Configuration conf, Class exampleClass) {
     this(conf);
-    String jar = findContainingJar(aClass);
-    if (jar != null) {
-      setJar(jar);
-    }
+    initialize(exampleClass);
   }
 
 

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

@@ -172,7 +172,11 @@ class MapOutputFile implements Writable, Configurable {
   }
 
   public void setConf(Configuration conf) {
-    this.jobConf = new JobConf(conf);
+    if (conf instanceof JobConf) {
+      jobConf = (JobConf) conf;
+    } else {
+      this.jobConf = new JobConf(conf);
+    }
   }
 
   public Configuration getConf() {

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

@@ -34,8 +34,8 @@ class MapTask extends Task {
   }
 
   private FileSplit split;
-  private MapOutputFile mapOutputFile;
-  private Configuration conf;
+  private MapOutputFile mapOutputFile = new MapOutputFile();
+  private JobConf conf;
 
   public MapTask() {}
 
@@ -149,9 +149,12 @@ class MapTask extends Task {
   }
 
   public void setConf(Configuration conf) {
-    this.conf = conf;
-    this.mapOutputFile = new MapOutputFile();
-    this.mapOutputFile.setConf(conf);
+    if (conf instanceof JobConf) {
+      this.conf = (JobConf) conf;
+    } else {
+      this.conf = new JobConf(conf);
+    }
+    this.mapOutputFile.setConf(this.conf);
   }
 
   public Configuration getConf() {

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

@@ -23,7 +23,7 @@ import java.io.*;
 class MapTaskRunner extends TaskRunner {
   private MapOutputFile mapOutputFile;
 
-  public MapTaskRunner(Task task, TaskTracker tracker, Configuration conf) {
+  public MapTaskRunner(Task task, TaskTracker tracker, JobConf conf) {
     super(task, tracker, conf);
     this.mapOutputFile = new MapOutputFile();
     this.mapOutputFile.setConf(conf);

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

@@ -46,8 +46,8 @@ class ReduceTask extends Task {
   private Progress appendPhase = getProgress().addPhase("append");
   private Progress sortPhase  = getProgress().addPhase("sort");
   private Progress reducePhase = getProgress().addPhase("reduce");
-  private Configuration conf;
-  private MapOutputFile mapOutputFile;
+  private JobConf conf;
+  private MapOutputFile mapOutputFile = new MapOutputFile();
 
   public ReduceTask() {}
 
@@ -307,9 +307,12 @@ class ReduceTask extends Task {
   }
 
   public void setConf(Configuration conf) {
-    this.conf = conf;
-    this.mapOutputFile = new MapOutputFile();
-    this.mapOutputFile.setConf(conf);
+    if (conf instanceof JobConf) {
+      this.conf = (JobConf) conf;
+    } else {
+      this.conf = new JobConf(conf);
+    }
+    this.mapOutputFile.setConf(this.conf);
   }
 
   public Configuration getConf() {

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

@@ -31,7 +31,7 @@ class ReduceTaskRunner extends TaskRunner {
     LogFormatter.getLogger("org.apache.hadoop.mapred.ReduceTaskRunner");
   private MapOutputFile mapOutputFile;
 
-  public ReduceTaskRunner(Task task, TaskTracker tracker, Configuration conf) {
+  public ReduceTaskRunner(Task task, TaskTracker tracker, JobConf conf) {
     super(task, tracker, conf);
     this.mapOutputFile = new MapOutputFile();
     this.mapOutputFile.setConf(conf);

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

@@ -38,9 +38,9 @@ abstract class TaskRunner extends Thread {
   private Task t;
   private TaskTracker tracker;
 
-  protected Configuration conf;
+  protected JobConf conf;
 
-  public TaskRunner(Task t, TaskTracker tracker, Configuration conf) {
+  public TaskRunner(Task t, TaskTracker tracker, JobConf conf) {
     this.t = t;
     this.tracker = tracker;
     this.conf = conf;
@@ -76,8 +76,7 @@ abstract class TaskRunner extends Thread {
       classPath.append(System.getProperty("java.class.path"));
       classPath.append(sep);
 
-      JobConf job = new JobConf(t.getJobFile());
-      String jar = job.getJar();
+      String jar = conf.getJar();
       if (jar != null) {                      // if jar exists, it into workDir
         unJar(new File(jar), workDir);
         File[] libs = new File(workDir, "lib").listFiles();
@@ -124,10 +123,10 @@ abstract class TaskRunner extends Thread {
       //     </value>
       //
       String javaOpts = handleDeprecatedHeapSize(
-          job.get("mapred.child.java.opts", "-Xmx200m"),
-          job.get("mapred.child.heap.size"));
+          conf.get("mapred.child.java.opts", "-Xmx200m"),
+          conf.get("mapred.child.heap.size"));
       javaOpts = replaceAll(javaOpts, "@taskid@", t.getTaskId());
-      int port = job.getInt("mapred.task.tracker.report.port", 50050) + 1;
+      int port = conf.getInt("mapred.task.tracker.report.port", 50050) + 1;
       javaOpts = replaceAll(javaOpts, "@port@", Integer.toString(port));
       String [] javaOptsSplit = javaOpts.split(" ");
       for (int i = 0; i < javaOptsSplit.length; i++) {

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

@@ -65,7 +65,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
     FileSystem fs = null;
     static final String SUBDIR = "taskTracker";
 
-    private Configuration fConf;
+    private JobConf fConf;
     private MapOutputFile mapOutputFile;
 
     private int maxCurrentTasks;
@@ -83,14 +83,14 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
     /**
      * Start with the local machine name, and the default JobTracker
      */
-    public TaskTracker(Configuration conf) throws IOException {
+    public TaskTracker(JobConf conf) throws IOException {
       this(JobTracker.getAddress(conf), conf);
     }
 
     /**
      * Start with the local machine name, and the addr of the target JobTracker
      */
-    public TaskTracker(InetSocketAddress jobTrackAddr, Configuration conf) throws IOException {
+    public TaskTracker(InetSocketAddress jobTrackAddr, JobConf conf) throws IOException {
         maxCurrentTasks = conf.getInt("mapred.tasktracker.tasks.maximum", 2);
 
         this.fConf = conf;
@@ -112,7 +112,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
                                (Math.abs(r.nextInt()) % 100000);
         LOG.info("Starting tracker " + taskTrackerName);
 
-        new JobConf(this.fConf).deleteLocalFiles(SUBDIR);
+        fConf.deleteLocalFiles(SUBDIR);
 
         // Clear out state tables
         this.tasks = new TreeMap();
@@ -409,17 +409,19 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
         TaskRunner runner;
         boolean done = false;
         boolean wasKilled = false;
-        private JobConf jobConf;
+        private JobConf defaultJobConf;
+        private JobConf localJobConf;
 
         /**
          */
-        public TaskInProgress(Task task, Configuration conf) {
+        public TaskInProgress(Task task, JobConf conf) {
             this.task = task;
             this.progress = 0.0f;
             this.runstate = TaskStatus.UNASSIGNED;
             stateString = "initializing";
             this.lastProgressReport = System.currentTimeMillis();
-            this.jobConf = new JobConf(conf);
+            this.defaultJobConf = conf;
+            localJobConf = null;
         }
 
         /**
@@ -427,31 +429,35 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
          * So here, edit the Task's fields appropriately.
          */
         private void localizeTask(Task t) throws IOException {
-            this.jobConf.deleteLocalFiles(SUBDIR + "/" + task.getTaskId());
+            this.defaultJobConf.deleteLocalFiles(SUBDIR + "/" + 
+                                                 task.getTaskId());
             Path localJobFile =
-              this.jobConf.getLocalPath(SUBDIR+"/"+t.getTaskId()+"/"+"job.xml");
+              this.defaultJobConf.getLocalPath(SUBDIR+"/"+t.getTaskId()+"/"+"job.xml");
             Path localJarFile =
-              this.jobConf.getLocalPath(SUBDIR+"/"+t.getTaskId()+"/"+"job.jar");
+              this.defaultJobConf.getLocalPath(SUBDIR+"/"+t.getTaskId()+"/"+"job.jar");
 
             String jobFile = t.getJobFile();
             fs.copyToLocalFile(new Path(jobFile), localJobFile);
             t.setJobFile(localJobFile.toString());
 
-            JobConf jc = new JobConf(localJobFile);
-            jc.set("mapred.task.id", task.getTaskId());
-            String jarFile = jc.getJar();
+            localJobConf = new JobConf(localJobFile);
+            localJobConf.set("mapred.task.id", task.getTaskId());
+            String jarFile = localJobConf.getJar();
             if (jarFile != null) {
               fs.copyToLocalFile(new Path(jarFile), localJarFile);
-              jc.setJar(localJarFile.toString());
+              localJobConf.setJar(localJarFile.toString());
 
               FileSystem localFs = FileSystem.getNamed("local", fConf);
               OutputStream out = localFs.create(localJobFile);
               try {
-                jc.write(out);
+                localJobConf.write(out);
               } finally {
                 out.close();
               }
             }
+            // set the task's configuration to the local job conf
+            // rather than the default.
+            t.setConf(localJobConf);
         }
 
         /**
@@ -618,7 +624,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
                 runner.close();
             } catch (IOException ie) {
             }
-            this.jobConf.deleteLocalFiles(SUBDIR + "/" + task.getTaskId());
+            this.defaultJobConf.deleteLocalFiles(SUBDIR + "/" + task.getTaskId());
         }
     }
 
@@ -735,17 +741,18 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
           LogFormatter.showTime(false);
           LOG.info("Child starting");
 
-          Configuration conf = new Configuration();
+          JobConf defaultConf = new JobConf();
           int port = Integer.parseInt(args[0]);
           String taskid = args[1];
           TaskUmbilicalProtocol umbilical =
             (TaskUmbilicalProtocol)RPC.getProxy(TaskUmbilicalProtocol.class,
-                                                new InetSocketAddress(port), conf);
+                                                new InetSocketAddress(port), 
+                                                defaultConf);
             
           Task task = umbilical.getTask(taskid);
           JobConf job = new JobConf(task.getJobFile());
 
-          conf.addFinalResource(new Path(task.getJobFile()));
+          defaultConf.addFinalResource(new Path(task.getJobFile()));
 
           startPinging(umbilical, taskid);        // start pinging parent
 
@@ -809,7 +816,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol, MapOutpu
             System.exit(-1);
         }
 
-        TaskTracker tt = new TaskTracker(new Configuration());
+        TaskTracker tt = new TaskTracker(new JobConf());
         tt.run();
     }
 }