YARN-223. Update process tree instead of getting new process trees. (Radim Kolar via llu)

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

Author: vicaya

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/Task.java

@@ -61,8 +61,7 @@
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer;
 import org.apache.hadoop.mapreduce.task.ReduceContextImpl;
-import org.apache.hadoop.yarn.util.ResourceCalculatorPlugin;
-import org.apache.hadoop.yarn.util.ResourceCalculatorPlugin.*;
+import org.apache.hadoop.yarn.util.ResourceCalculatorProcessTree;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.util.Progress;
 import org.apache.hadoop.util.Progressable;
@@ -169,7 +168,7 @@ static synchronized String getOutputName(int partition) {
   private Iterator<Long> currentRecIndexIterator = 
     skipRanges.skipRangeIterator();
 
-  private ResourceCalculatorPlugin resourceCalculator = null;
+  private ResourceCalculatorProcessTree pTree;
   private long initCpuCumulativeTime = 0;
 
   protected JobConf conf;
@@ -372,7 +371,7 @@ public void setSkipping(boolean skipping) {
    * Return current state of the task. 
    * needs to be synchronized as communication thread 
    * sends the state every second
-   * @return
+   * @return task state
    */
   synchronized TaskStatus.State getState(){
     return this.taskStatus.getRunState(); 
@@ -558,15 +557,15 @@ public void initialize(JobConf job, JobID id,
       }
     }
     committer.setupTask(taskContext);
-    Class<? extends ResourceCalculatorPlugin> clazz =
-        conf.getClass(MRConfig.RESOURCE_CALCULATOR_PLUGIN,
-            null, ResourceCalculatorPlugin.class);
-    resourceCalculator = ResourceCalculatorPlugin
-            .getResourceCalculatorPlugin(clazz, conf);
-    LOG.info(" Using ResourceCalculatorPlugin : " + resourceCalculator);
-    if (resourceCalculator != null) {
-      initCpuCumulativeTime =
-        resourceCalculator.getProcResourceValues().getCumulativeCpuTime();
+    Class<? extends ResourceCalculatorProcessTree> clazz =
+        conf.getClass(MRConfig.RESOURCE_CALCULATOR_PROCESS_TREE,
+            null, ResourceCalculatorProcessTree.class);
+    pTree = ResourceCalculatorProcessTree
+            .getResourceCalculatorProcessTree(System.getenv().get("JVM_PID"), clazz, conf);
+    LOG.info(" Using ResourceCalculatorProcessTree : " + pTree);
+    if (pTree != null) {
+      pTree.updateProcessTree();
+      initCpuCumulativeTime = pTree.getCumulativeCpuTime();
     }
   }
 
@@ -817,14 +816,14 @@ void updateResourceCounters() {
     // Update generic resource counters
     updateHeapUsageCounter();
 
-    // Updating resources specified in ResourceCalculatorPlugin
-    if (resourceCalculator == null) {
+    // Updating resources specified in ResourceCalculatorProcessTree
+    if (pTree == null) {
       return;
     }
-    ProcResourceValues res = resourceCalculator.getProcResourceValues();
-    long cpuTime = res.getCumulativeCpuTime();
-    long pMem = res.getPhysicalMemorySize();
-    long vMem = res.getVirtualMemorySize();
+    pTree.updateProcessTree();
+    long cpuTime = pTree.getCumulativeCpuTime();
+    long pMem = pTree.getCumulativeRssmem();
+    long vMem = pTree.getCumulativeVmem();
     // Remove the CPU time consumed previously by JVM reuse
     cpuTime -= initCpuCumulativeTime;
     counters.findCounter(TaskCounter.CPU_MILLISECONDS).setValue(cpuTime);

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRConfig.java

@@ -55,8 +55,8 @@ public interface MRConfig {
   public static final long    DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT = 
     7*24*60*60*1000; // 7 days
 
-  public static final String RESOURCE_CALCULATOR_PLUGIN = 
-    "mapreduce.job.resourcecalculatorplugin";
+  public static final String RESOURCE_CALCULATOR_PROCESS_TREE =
+    "mapreduce.job.process-tree.class";
   public static final String STATIC_RESOLUTIONS = 
     "mapreduce.job.net.static.resolutions";
 

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/LinuxResourceCalculatorPlugin.java

@@ -409,7 +409,7 @@ public static void main(String[] args) {
 
   @Override
   public ProcResourceValues getProcResourceValues() {
-    pTree = pTree.getProcessTree();
+    pTree.updateProcessTree();
     long cpuTime = pTree.getCumulativeCpuTime();
     long pMem = pTree.getCumulativeRssmem();
     long vMem = pTree.getCumulativeVmem();

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/ProcfsBasedProcessTree.java

@@ -166,12 +166,10 @@ public static boolean isAvailable() {
   }
 
   /**
-   * Get the process-tree with latest state. If the root-process is not alive,
-   * an empty tree will be returned.
-   * 
-   * @return the process-tree with latest state.
+   * Update the process-tree with latest state. If the root-process is not alive,
+   * tree will become empty.
    */
-  public ProcfsBasedProcessTree getProcessTree() {
+  public void updateProcessTree() {
     if (!pid.equals(deadPid)) {
       // Get the list of processes
       List<String> processList = getProcessList();
@@ -197,7 +195,7 @@ public ProcfsBasedProcessTree getProcessTree() {
       }
 
       if (me == null) {
-        return this; 
+        return;
       }
 
       // Add each process to its parent.
@@ -239,7 +237,6 @@ public ProcfsBasedProcessTree getProcessTree() {
         LOG.debug(this.toString());
       }
     }
-    return this;
   }
 
   /**

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/util/TestProcfsBasedProcessTree.java

@@ -151,7 +151,7 @@ public void testProcessTree() {
     ProcfsBasedProcessTree p = new ProcfsBasedProcessTree(pid,
                                ProcessTree.isSetsidAvailable,
                                ProcessTree.DEFAULT_SLEEPTIME_BEFORE_SIGKILL);
-    p = p.getProcessTree(); // initialize
+    p.updateProcessTree(); // initialize
     LOG.info("ProcessTree: " + p.toString());
 
     File leaf = new File(lowestDescendant);
@@ -164,7 +164,7 @@ public void testProcessTree() {
       }
     }
 
-    p = p.getProcessTree(); // reconstruct
+    p.updateProcessTree(); // reconstruct
     LOG.info("ProcessTree: " + p.toString());
 
     // Get the process-tree dump
@@ -203,7 +203,7 @@ public void testProcessTree() {
     }
 
     // ProcessTree is gone now. Any further calls should be sane.
-    p = p.getProcessTree();
+    p.updateProcessTree();
     assertFalse("ProcessTree must have been gone", p.isAlive());
     assertTrue("Cumulative vmem for the gone-process is "
         + p.getCumulativeVmem() + " . It should be zero.", p
@@ -336,7 +336,7 @@ public void testCpuAndMemoryForProcessTree() throws IOException {
           new ProcfsBasedProcessTree("100", true, 100L, 
                                   procfsRootDir.getAbsolutePath());
       // build the process tree.
-      processTree.getProcessTree();
+      processTree.updateProcessTree();
 
       // verify cumulative memory
       assertEquals("Cumulative virtual memory does not match", 600000L,
@@ -362,7 +362,7 @@ public void testCpuAndMemoryForProcessTree() throws IOException {
       writeStatFiles(procfsRootDir, pids, procInfos);
 
       // build the process tree.
-      processTree.getProcessTree();
+      processTree.updateProcessTree();
 
       // verify cumulative cpu time again
       cumuCpuTime = ProcfsBasedProcessTree.JIFFY_LENGTH_IN_MILLIS > 0 ?
@@ -409,7 +409,7 @@ public void testMemForOlderProcesses() throws IOException {
           new ProcfsBasedProcessTree("100", true, 100L, 
                                   procfsRootDir.getAbsolutePath());
       // build the process tree.
-      processTree.getProcessTree();
+      processTree.updateProcessTree();
 
       // verify cumulative memory
       assertEquals("Cumulative memory does not match",
@@ -425,7 +425,7 @@ public void testMemForOlderProcesses() throws IOException {
       writeStatFiles(procfsRootDir, newPids, newProcInfos);
 
       // check memory includes the new process.
-      processTree.getProcessTree();
+      processTree.updateProcessTree();
       assertEquals("Cumulative vmem does not include new process",
                    1200000L, processTree.getCumulativeVmem());
       long cumuRssMem = ProcfsBasedProcessTree.PAGE_SIZE > 0 ?
@@ -451,7 +451,7 @@ public void testMemForOlderProcesses() throws IOException {
       writeStatFiles(procfsRootDir, newPids, newProcInfos);
 
       // refresh process tree
-      processTree.getProcessTree();
+      processTree.updateProcessTree();
 
       // processes older than 2 iterations should be same as before.
       assertEquals("Cumulative vmem shouldn't have included new processes",
@@ -555,7 +555,7 @@ public void testProcessTreeDump()
           new ProcfsBasedProcessTree("100", true, 100L, procfsRootDir
               .getAbsolutePath());
       // build the process tree.
-      processTree.getProcessTree();
+      processTree.updateProcessTree();
 
       // Get the process-tree dump
       String processTreeDump = processTree.getProcessTreeDump();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/LinuxResourceCalculatorPlugin.java

@@ -91,7 +91,6 @@ public class LinuxResourceCalculatorPlugin extends ResourceCalculatorPlugin {
   private float cpuUsage = UNAVAILABLE;
   private long sampleTime = UNAVAILABLE;
   private long lastSampleTime = UNAVAILABLE;
-  private ResourceCalculatorProcessTree pTree = null;
 
   boolean readMemInfoFile = false;
   boolean readCpuInfoFile = false;
@@ -109,8 +108,6 @@ public LinuxResourceCalculatorPlugin() {
     procfsCpuFile = PROCFS_CPUINFO;
     procfsStatFile = PROCFS_STAT;
     jiffyLengthInMillis = ProcfsBasedProcessTree.JIFFY_LENGTH_IN_MILLIS;
-    String pid = System.getenv().get("JVM_PID");
-    pTree = new ProcfsBasedProcessTree(pid);
   }
 
   /**
@@ -129,8 +126,6 @@ public LinuxResourceCalculatorPlugin(String procfsMemFile,
     this.procfsCpuFile = procfsCpuFile;
     this.procfsStatFile = procfsStatFile;
     this.jiffyLengthInMillis = jiffyLengthInMillis;
-    String pid = System.getenv().get("JVM_PID");
-    pTree = new ProcfsBasedProcessTree(pid);
   }
 
   /**
@@ -400,13 +395,4 @@ public static void main(String[] args) {
     }
     System.out.println("CPU usage % : " + plugin.getCpuUsage());
   }
-
-  @Override
-  public ProcResourceValues getProcResourceValues() {
-    pTree = pTree.getProcessTree();
-    long cpuTime = pTree.getCumulativeCpuTime();
-    long pMem = pTree.getCumulativeRssmem();
-    long vMem = pTree.getCumulativeVmem();
-    return new ProcResourceValues(cpuTime, pMem, vMem);
-  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ProcfsBasedProcessTree.java

@@ -140,13 +140,12 @@ public static boolean isAvailable() {
   }
 
   /**
-   * Get the process-tree with latest state. If the root-process is not alive,
-   * an empty tree will be returned.
+   * Update process-tree with latest state. If the root-process is not alive,
+   * tree will be empty.
    *
-   * @return the process-tree with latest state.
    */
   @Override
-  public ResourceCalculatorProcessTree getProcessTree() {
+  public void updateProcessTree() {
     if (!pid.equals(deadPid)) {
       // Get the list of processes
       List<String> processList = getProcessList();
@@ -172,7 +171,7 @@ public ResourceCalculatorProcessTree getProcessTree() {
       }
 
       if (me == null) {
-        return this;
+        return;
       }
 
       // Add each process to its parent.
@@ -214,7 +213,6 @@ public ResourceCalculatorProcessTree getProcessTree() {
         LOG.debug(this.toString());
       }
     }
-    return this;
   }
 
   /** Verify that the given process id is same as its process group id.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ResourceCalculatorPlugin.java

@@ -90,48 +90,6 @@ public abstract class ResourceCalculatorPlugin extends Configured {
    */
   public abstract float getCpuUsage();
 
-  /**
-   * Obtain resource status used by current process tree.
-   */
-  @InterfaceAudience.Private
-  @InterfaceStability.Unstable
-  public abstract ProcResourceValues getProcResourceValues();
-
-  public static class ProcResourceValues {
-    private final long cumulativeCpuTime;
-    private final long physicalMemorySize;
-    private final long virtualMemorySize;
-    public ProcResourceValues(long cumulativeCpuTime, long physicalMemorySize,
-                              long virtualMemorySize) {
-      this.cumulativeCpuTime = cumulativeCpuTime;
-      this.physicalMemorySize = physicalMemorySize;
-      this.virtualMemorySize = virtualMemorySize;
-    }
-    /**
-     * Obtain the physical memory size used by current process tree.
-     * @return physical memory size in bytes.
-     */
-    public long getPhysicalMemorySize() {
-      return physicalMemorySize;
-    }
-
-    /**
-     * Obtain the virtual memory size used by a current process tree.
-     * @return virtual memory size in bytes.
-     */
-    public long getVirtualMemorySize() {
-      return virtualMemorySize;
-    }
-
-    /**
-     * Obtain the cumulative CPU time used by a current process tree.
-     * @return cumulative CPU time in milliseconds
-     */
-    public long getCumulativeCpuTime() {
-      return cumulativeCpuTime;
-    }
-  }
-
   /**
    * Create the ResourceCalculatorPlugin from the class name and configure it. If
    * class name is null, this method will try and return a memory calculator

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/ResourceCalculatorProcessTree.java

@@ -43,16 +43,14 @@ public ResourceCalculatorProcessTree(String root) {
   }
 
   /**
-   * Get the process-tree with latest state. If the root-process is not alive,
-   * an empty tree will be returned.
+   * Update the process-tree with latest state.
    *
    * Each call to this function should increment the age of the running
    * processes that already exist in the process tree. Age is used other API's
    * of the interface.
    *
-   * @return the process-tree with latest state.
    */
-  public abstract ResourceCalculatorProcessTree getProcessTree();
+  public abstract void updateProcessTree();
 
   /**
    * Get a dump of the process-tree.