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 {
