Added instrumentation for compute monotask time.

This commit adds a field to TaskMetrics to record the total time used
by ComputeMonotasks for each macrotask.

Conflicts:
	core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

core/src/main/scala/org/apache/spark/executor/TaskMetrics.scala

@@ -83,6 +83,14 @@ class TaskMetrics extends Serializable {
   def executorRunTime = _executorRunTime
   private[spark] def setExecutorRunTime(value: Long) = _executorRunTime = value
 
+  /**
+   * Total time consumed by compute monotasks for this macrotask. May be larger than executorRunTime
+   * if multiple compute monotasks ran simultaneously.
+   */
+  private var _computationNanos: Long = _
+  def computationNanos: Long = _computationNanos
+  private[spark] def setComputationNanos(value: Long) = _computationNanos = value
+
   /**
    * The number of bytes this task transmitted back to the driver as the TaskResult
    */

core/src/main/scala/org/apache/spark/monotasks/compute/ComputeMonotask.scala

@@ -35,6 +35,7 @@ private[spark] abstract class ComputeMonotask(context: TaskContextImpl)
 
   /** Runs the execute method and handles common exceptions thrown by ComputeMonotasks. */
   def executeAndHandleExceptions() {
+    val startTimeNanos = System.nanoTime()
     try {
       Accumulators.registeredAccumulables.set(context.accumulators)
       TaskContext.setTaskContext(context)
@@ -73,6 +74,8 @@ private[spark] abstract class ComputeMonotask(context: TaskContextImpl)
         val closureSerializer = context.env.closureSerializer.newInstance()
         context.localDagScheduler.handleTaskFailure(this, closureSerializer.serialize(reason))
       }
+    } finally {
+      context.taskMetrics.setComputationNanos(System.nanoTime() - startTimeNanos)
     }
   }
 }

core/src/main/scala/org/apache/spark/util/JsonProtocol.scala

@@ -306,6 +306,7 @@ private[spark] object JsonProtocol {
     ("Host Name" -> taskMetrics.hostname) ~
     ("Executor Deserialize Time" -> taskMetrics.executorDeserializeTime) ~
     ("Executor Run Time" -> taskMetrics.executorRunTime) ~
+    ("Computation Nanos" -> taskMetrics.computationNanos) ~
     ("Result Size" -> taskMetrics.resultSize) ~
     ("JVM GC Time" -> taskMetrics.jvmGCTime) ~
     ("Result Serialization Time" -> taskMetrics.resultSerializationTime) ~
@@ -721,6 +722,8 @@ private[spark] object JsonProtocol {
     metrics.setHostname((json \ "Host Name").extract[String])
     metrics.setExecutorDeserializeTime((json \ "Executor Deserialize Time").extract[Long])
     metrics.setExecutorRunTime((json \ "Executor Run Time").extract[Long])
+    metrics.setComputationNanos(
+      Utils.jsonOption(json \ "Computation Nanos").map(_.extract[Long]).getOrElse(0L))
     metrics.setResultSize((json \ "Result Size").extract[Long])
     metrics.setJvmGCTime((json \ "JVM GC Time").extract[Long])
     metrics.setResultSerializationTime((json \ "Result Serialization Time").extract[Long])