hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acmur...@apache.org
Subject svn commit: r987585 - in /hadoop/mapreduce/trunk: ./ src/java/org/apache/hadoop/mapred/ src/java/org/apache/hadoop/mapreduce/ src/java/org/apache/hadoop/mapreduce/util/ src/test/mapred/org/apache/hadoop/mapred/
Date Fri, 20 Aug 2010 17:41:05 GMT
Author: acmurthy
Date: Fri Aug 20 17:41:05 2010
New Revision: 987585

URL: http://svn.apache.org/viewvc?rev=987585&view=rev
Log:
MAPREDUCE-220. Collect cpu and memory statistics per task. Contributed by Scott Chen.

Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/Task.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/TaskCounter.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/LinuxResourceCalculatorPlugin.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/ResourceCalculatorPlugin.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/DummyResourceCalculatorPlugin.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestTTResourceReporting.java

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Fri Aug 20 17:41:05 2010
@@ -12,6 +12,9 @@ Trunk (unreleased changes)
     MAPREDUCE-1804. Stress-test tool for HDFS introduced in HDFS-708.
     (Joshua Harlow via shv)
 
+    MAPREDUCE-220. Collect cpu and memory statistics per task. (Scott Chen via
+    acmurthy)
+
   IMPROVEMENTS
 
     MAPREDUCE-1546. Redirect all job pages to corresponding history page

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/Task.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/Task.java?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/Task.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/Task.java Fri Aug 20 17:41:05
2010
@@ -59,6 +59,8 @@ import org.apache.hadoop.mapreduce.JobSt
 import org.apache.hadoop.mapreduce.MRConfig;
 import org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer;
 import org.apache.hadoop.mapreduce.task.ReduceContextImpl;
+import org.apache.hadoop.mapreduce.util.ResourceCalculatorPlugin;
+import org.apache.hadoop.mapreduce.util.ResourceCalculatorPlugin.*;
 import org.apache.hadoop.mapreduce.server.tasktracker.TTConfig;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.Credentials;
@@ -77,7 +79,6 @@ abstract public class Task implements Wr
     LogFactory.getLog(Task.class);
 
   public static String MERGED_OUTPUT_PREFIX = ".merged";
-  
 
   /**
    * Counters to measure the usage of the different file systems.
@@ -133,7 +134,10 @@ abstract public class Task implements Wr
   private volatile long currentRecStartIndex; 
   private Iterator<Long> currentRecIndexIterator = 
     skipRanges.skipRangeIterator();
-  
+
+  private ResourceCalculatorPlugin resourceCalculator = null;
+  private long initCpuCumulativeTime = 0;
+
   protected JobConf conf;
   protected MapOutputFile mapOutputFile = new MapOutputFile();
   protected LocalDirAllocator lDirAlloc;
@@ -503,6 +507,16 @@ abstract public class Task implements Wr
       }
     }
     committer.setupTask(taskContext);
+    Class<? extends ResourceCalculatorPlugin> clazz =
+        conf.getClass(TTConfig.TT_RESOURCE_CALCULATOR_PLUGIN,
+            null, ResourceCalculatorPlugin.class);
+    resourceCalculator = ResourceCalculatorPlugin
+            .getResourceCalculatorPlugin(clazz, conf);
+    LOG.info(" Using ResourceCalculatorPlugin : " + resourceCalculator);
+    if (resourceCalculator != null) {
+      initCpuCumulativeTime =
+        resourceCalculator.getProcResourceValues().getCumulativeCpuTime();
+    }
   }
   
   @InterfaceAudience.Private
@@ -701,6 +715,24 @@ abstract public class Task implements Wr
   }
 
   /**
+   * Update resource information counters
+   */
+  void updateResourceCounters() {
+    if (resourceCalculator == null) {
+      return;
+    }
+    ProcResourceValues res = resourceCalculator.getProcResourceValues();
+    long cpuTime = res.getCumulativeCpuTime();
+    long pMem = res.getPhysicalMemorySize();
+    long vMem = res.getVirtualMemorySize();
+    // Remove the CPU time consumed previously by JVM reuse
+    cpuTime -= initCpuCumulativeTime;
+    counters.findCounter(TaskCounter.CPU_MILLISECONDS).setValue(cpuTime);
+    counters.findCounter(TaskCounter.PHYSICAL_MEMORY_BYTES).setValue(pMem);
+    counters.findCounter(TaskCounter.VIRTUAL_MEMORY_BYTES).setValue(vMem);
+  }
+
+  /**
    * An updater that tracks the amount of time this task has spent in GC.
    */
   class GcTimeUpdater {
@@ -799,6 +831,7 @@ abstract public class Task implements Wr
     }
 
     gcUpdater.incrementGcCounter();
+    updateResourceCounters();
   }
 
   public void done(TaskUmbilicalProtocol umbilical,

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/TaskCounter.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/TaskCounter.java?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/TaskCounter.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/TaskCounter.java Fri Aug 20
17:41:05 2010
@@ -42,5 +42,8 @@ public enum TaskCounter {
   SHUFFLED_MAPS, 
   FAILED_SHUFFLE,
   MERGED_MAP_OUTPUTS,
-  GC_TIME_MILLIS
+  GC_TIME_MILLIS,
+  CPU_MILLISECONDS,
+  PHYSICAL_MEMORY_BYTES,
+  VIRTUAL_MEMORY_BYTES
 }

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/LinuxResourceCalculatorPlugin.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/LinuxResourceCalculatorPlugin.java?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/LinuxResourceCalculatorPlugin.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/LinuxResourceCalculatorPlugin.java
Fri Aug 20 17:41:05 2010
@@ -90,6 +90,7 @@ public class LinuxResourceCalculatorPlug
   private float cpuUsage = TaskTrackerStatus.UNAVAILABLE;
   private long sampleTime = TaskTrackerStatus.UNAVAILABLE;
   private long lastSampleTime = TaskTrackerStatus.UNAVAILABLE;
+  private ProcfsBasedProcessTree pTree = null;
 
   boolean readMemInfoFile = false;
   boolean readCpuInfoFile = false;
@@ -107,7 +108,8 @@ public class LinuxResourceCalculatorPlug
     procfsCpuFile = PROCFS_CPUINFO;
     procfsStatFile = PROCFS_STAT;
     jiffyLengthInMillis = ProcfsBasedProcessTree.JIFFY_LENGTH_IN_MILLIS;
-    
+    String pid = System.getenv().get("JVM_PID");
+    pTree = new ProcfsBasedProcessTree(pid);
   }
   
   /**
@@ -126,6 +128,8 @@ public class LinuxResourceCalculatorPlug
     this.procfsCpuFile = procfsCpuFile;
     this.procfsStatFile = procfsStatFile;
     this.jiffyLengthInMillis = jiffyLengthInMillis;
+    String pid = System.getenv().get("JVM_PID");
+    pTree = new ProcfsBasedProcessTree(pid);
   }
 
   /**
@@ -395,4 +399,13 @@ public class LinuxResourceCalculatorPlug
     }
     System.out.println("CPU usage % : " + plugin.getCpuUsage());
   }
-}
\ No newline at end of file
+
+  @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);
+  }
+}

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/ResourceCalculatorPlugin.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/ResourceCalculatorPlugin.java?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/ResourceCalculatorPlugin.java
(original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapreduce/util/ResourceCalculatorPlugin.java
Fri Aug 20 17:41:05 2010
@@ -91,6 +91,48 @@ public abstract class ResourceCalculator
   public abstract float getCpuUsage();
 
   /**
+   * Obtain resource status used by current process tree.
+   */
+  @InterfaceAudience.Private
+  @InterfaceStability.Unstable
+  public abstract ProcResourceValues getProcResourceValues();
+
+  public 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;
+    }
+  }
+
+  /**
    * Get the ResourceCalculatorPlugin from the class name and configure it. If
    * class name is null, this method will try and return a memory calculator
    * plugin available for this system.
@@ -120,4 +162,4 @@ public abstract class ResourceCalculator
     // Not supported on this system.
     return null;
   }
-}
\ No newline at end of file
+}

Modified: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/DummyResourceCalculatorPlugin.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/DummyResourceCalculatorPlugin.java?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/DummyResourceCalculatorPlugin.java
(original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/DummyResourceCalculatorPlugin.java
Fri Aug 20 17:41:05 2010
@@ -50,6 +50,15 @@ public class DummyResourceCalculatorPlug
   /** CPU usage percentage for testing */
   public static final String CPU_USAGE =
       "mapred.tasktracker.cpuusage.testing";
+  /** process cumulative CPU usage time for testing */
+  public static final String PROC_CUMULATIVE_CPU_TIME =
+      "mapred.tasktracker.proccumulativecputime.testing";
+  /** process pmem for testing*/
+  public static final String PROC_PMEM_TESTING_PROPERTY =
+      "mapred.tasktracker.procpmem.testing";
+  /** process vmem for testing*/
+  public static final String PROC_VMEM_TESTING_PROPERTY =
+      "mapred.tasktracker.procvmem.testing";
 
   /** {@inheritDoc} */
   @Override
@@ -98,4 +107,12 @@ public class DummyResourceCalculatorPlug
   public float getCpuUsage() {
     return getConf().getFloat(CPU_USAGE, -1);
   }
-}
\ No newline at end of file
+
+  @Override
+  public ProcResourceValues getProcResourceValues() {
+    long cpuTime = getConf().getLong(PROC_CUMULATIVE_CPU_TIME, -1);
+    long pMem = getConf().getLong(PROC_PMEM_TESTING_PROPERTY, -1);
+    long vMem = getConf().getLong(PROC_VMEM_TESTING_PROPERTY, -1);
+    return new ProcResourceValues(cpuTime, pMem, vMem);
+  }
+}

Modified: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestTTResourceReporting.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestTTResourceReporting.java?rev=987585&r1=987584&r2=987585&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestTTResourceReporting.java
(original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestTTResourceReporting.java
Fri Aug 20 17:41:05 2010
@@ -23,6 +23,7 @@ import java.util.List;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.mapreduce.SleepJob;
+import org.apache.hadoop.mapreduce.TaskCounter;
 import org.apache.hadoop.mapreduce.util.LinuxResourceCalculatorPlugin;
 import org.apache.hadoop.mapreduce.util.ResourceCalculatorPlugin;
 import org.apache.hadoop.util.ToolRunner;
@@ -154,9 +155,46 @@ public class TestTTResourceReporting ext
               + reportedNumProcessors + ","
               + reportedCpuUsage
                + ")";
-      hasPassed = true;
-      hasDynamicValuePassed = true;
       LOG.info(message);
+      hasDynamicValuePassed = true;
+      // Check task resource status in task reports
+      for (TaskStatus taskStatus : status.getTaskReports()) {
+        Counters counters = taskStatus.getCounters();
+        // This should be zero because the initial CPU time is subtracted.
+        long procCumulativeCpuTime = 0;
+        long procVirtualMemorySize =
+          getConf().getLong("procVirtualMemorySize", -1);
+        long procPhysicalMemorySize =
+          getConf().getLong("procPhysicalMemorySize", -1);
+        long reportedProcCumulativeCpuTime =
+          counters.findCounter(TaskCounter.CPU_MILLISECONDS).getValue();
+        long reportedProcVirtualMemorySize =
+          counters.findCounter(TaskCounter.VIRTUAL_MEMORY_BYTES).getValue();
+        long reportedProcPhysicalMemorySize =
+          counters.findCounter(TaskCounter.PHYSICAL_MEMORY_BYTES).getValue();
+        String procMessage =
+          "expected values : "
+              + "(procCumulativeCpuTime, procVirtualMemorySize,"
+              + " procPhysicalMemorySize) = ("
+              + procCumulativeCpuTime + ", "
+              + procVirtualMemorySize + ", "
+              + procPhysicalMemorySize + ")";
+        procMessage +=
+          "\nreported values : "
+              + "(procCumulativeCpuTime, procVirtualMemorySize,"
+              + " procPhysicalMemorySize) = ("
+              + reportedProcCumulativeCpuTime + ", "
+              + reportedProcVirtualMemorySize + ", "
+              + reportedProcPhysicalMemorySize + ")";
+        LOG.info(procMessage);
+        message += "\n" + procMessage;
+        if (procCumulativeCpuTime != reportedProcCumulativeCpuTime ||
+            procVirtualMemorySize != reportedProcVirtualMemorySize ||
+            procPhysicalMemorySize != reportedProcPhysicalMemorySize) {
+          hasDynamicValuePassed = false;
+        }
+      }
+      hasPassed = true;
       if (totalVirtualMemoryOnTT != reportedTotalVirtualMemoryOnTT
           || totalPhysicalMemoryOnTT != reportedTotalPhysicalMemoryOnTT
           || mapSlotMemorySize != reportedMapSlotMemorySize
@@ -192,7 +230,11 @@ public class TestTTResourceReporting ext
           org.apache.hadoop.mapreduce.server.tasktracker.TTConfig.TT_RESOURCE_CALCULATOR_PLUGIN,
      
           DummyResourceCalculatorPlugin.class, ResourceCalculatorPlugin.class);
       setUpCluster(conf);
-      runSleepJob(miniMRCluster.createJobConf());
+      JobConf jobConf = miniMRCluster.createJobConf();
+      jobConf.setClass(
+          org.apache.hadoop.mapreduce.server.tasktracker.TTConfig.TT_RESOURCE_CALCULATOR_PLUGIN,
+          DummyResourceCalculatorPlugin.class, ResourceCalculatorPlugin.class);
+      runSleepJob(jobConf);
       verifyTestResults();
     } finally {
       tearDownCluster();
@@ -218,6 +260,9 @@ public class TestTTResourceReporting ext
     conf.setLong("cpuFrequency", 2000000L);
     conf.setInt("numProcessors", 8);
     conf.setFloat("cpuUsage", 15.5F);
+    conf.setLong("procCumulativeCpuTime", 1000L);
+    conf.setLong("procVirtualMemorySize", 2 * 1024 * 1024 * 1024L);
+    conf.setLong("procPhysicalMemorySize", 1024 * 1024 * 1024L);
 
     conf.setClass(
         org.apache.hadoop.mapreduce.server.tasktracker.TTConfig.TT_RESOURCE_CALCULATOR_PLUGIN,
      
@@ -232,12 +277,19 @@ public class TestTTResourceReporting ext
     conf.setLong(DummyResourceCalculatorPlugin.CPU_FREQUENCY, 2000000L);
     conf.setInt(DummyResourceCalculatorPlugin.NUM_PROCESSORS, 8);
     conf.setFloat(DummyResourceCalculatorPlugin.CPU_USAGE, 15.5F);
-    
     try {
       setUpCluster(conf);
       JobConf jobConf = miniMRCluster.createJobConf();
       jobConf.setMemoryForMapTask(1 * 1024L);
       jobConf.setMemoryForReduceTask(2 * 1024L);
+      jobConf.setClass(
+        org.apache.hadoop.mapreduce.server.tasktracker.TTConfig.TT_RESOURCE_CALCULATOR_PLUGIN,
+        DummyResourceCalculatorPlugin.class, ResourceCalculatorPlugin.class);
+      jobConf.setLong(DummyResourceCalculatorPlugin.PROC_CUMULATIVE_CPU_TIME, 1000L);
+      jobConf.setLong(DummyResourceCalculatorPlugin.PROC_VMEM_TESTING_PROPERTY,
+                      2 * 1024 * 1024 * 1024L);
+      jobConf.setLong(DummyResourceCalculatorPlugin.PROC_PMEM_TESTING_PROPERTY,
+                      1024 * 1024 * 1024L);
       runSleepJob(jobConf);
       verifyTestResults();
     } finally {



Mime
View raw message