Merge pull request #7 from markhamstra/master-csd

SPY-287 Merging Apache 0.8.2 changes

Author: jhartlaub

assembly/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent</artifactId>
-    <version>0.8.1-csd-3-SNAPSHOT</version>
+    <version>0.8.2-candidate-csd-1-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

bagel/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent</artifactId>
-    <version>0.8.1-csd-3-SNAPSHOT</version>
+    <version>0.8.2-candidate-csd-1-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

core/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent</artifactId>
-    <version>0.8.1-csd-3-SNAPSHOT</version>
+    <version>0.8.2-candidate-csd-1-SNAPSHOT</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala

@@ -246,6 +246,17 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
     new java.util.ArrayList(arr)
   }
 
+  /**
+   * Return an array that contains all of the elements in a specific partition of this RDD.
+   */
+  def collectPartitions(partitionIds: Array[Int]): Array[JList[T]] = {
+    // This is useful for implementing `take` from other language frontends
+    // like Python where the data is serialized.
+    import scala.collection.JavaConversions._
+    val res = context.runJob(rdd, (it: Iterator[T]) => it.toArray, partitionIds, true)
+    res.map(x => new java.util.ArrayList(x.toSeq)).toArray
+  }
+
   /**
    * Reduces the elements of this RDD using the specified commutative and associative binary operator.
    */

core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala

@@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicLong
 
 import org.apache.spark._
 
-abstract class Broadcast[T](private[spark] val id: Long) extends Serializable {
+abstract class Broadcast[T](val id: Long) extends Serializable {
   def value: T
 
   // We cannot have an abstract readObject here due to some weird issues with

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -409,7 +409,7 @@ private[spark] class Master(host: String, port: Int, webUiPort: Int) extends Act
     // There may be one or more refs to dead workers on this same node (w/ different ID's),
     // remove them.
     workers.filter { w =>
-      (w.host == host && w.port == port) && (w.state == WorkerState.DEAD)
+      (w.host == worker.host && w.port == worker.port) && (w.state == WorkerState.DEAD)
     }.foreach { w =>
       workers -= w
     }

core/src/main/scala/org/apache/spark/deploy/master/ZooKeeperPersistenceEngine.scala

@@ -77,7 +77,7 @@ class ZooKeeperPersistenceEngine(serialization: Serialization)
   }
 
   def deserializeFromFile[T <: Serializable](filename: String)(implicit m: Manifest[T]): T = {
-    val fileData = zk.getData("/spark/master_status/" + filename)
+    val fileData = zk.getData(WORKING_DIR + "/" + filename)
     val clazz = m.erasure.asInstanceOf[Class[T]]
     val serializer = serialization.serializerFor(clazz)
     serializer.fromBinary(fileData).asInstanceOf[T]

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

@@ -142,11 +142,6 @@ private[spark] class Executor(
     val tr = runningTasks.get(taskId)
     if (tr != null) {
       tr.kill()
-      // We remove the task also in the finally block in TaskRunner.run.
-      // The reason we need to remove it here is because killTask might be called before the task
-      // is even launched, and never reaching that finally block. ConcurrentHashMap's remove is
-      // idempotent.
-      runningTasks.remove(taskId)
     }
   }
 
@@ -168,6 +163,8 @@ private[spark] class Executor(
   class TaskRunner(execBackend: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer)
     extends Runnable {
 
+    object TaskKilledException extends Exception
+
     @volatile private var killed = false
     @volatile private var task: Task[Any] = _
 
@@ -201,9 +198,11 @@ private[spark] class Executor(
         // If this task has been killed before we deserialized it, let's quit now. Otherwise,
         // continue executing the task.
         if (killed) {
-          logInfo("Executor killed task " + taskId)
-          execBackend.statusUpdate(taskId, TaskState.KILLED, ser.serialize(TaskKilled))
-          return
+          // Throw an exception rather than returning, because returning within a try{} block
+          // causes a NonLocalReturnControl exception to be thrown. The NonLocalReturnControl
+          // exception will be caught by the catch block, leading to an incorrect ExceptionFailure
+          // for the task.
+          throw TaskKilledException
         }
 
         attemptedTask = Some(task)
@@ -217,9 +216,7 @@ private[spark] class Executor(
 
         // If the task has been killed, let's fail it.
         if (task.killed) {
-          logInfo("Executor killed task " + taskId)
-          execBackend.statusUpdate(taskId, TaskState.KILLED, ser.serialize(TaskKilled))
-          return
+          throw TaskKilledException
         }
 
         for (m <- task.metrics) {
@@ -257,6 +254,11 @@ private[spark] class Executor(
           execBackend.statusUpdate(taskId, TaskState.FAILED, ser.serialize(reason))
         }
 
+        case TaskKilledException => {
+          logInfo("Executor killed task " + taskId)
+          execBackend.statusUpdate(taskId, TaskState.KILLED, ser.serialize(TaskKilled))
+        }
+
         case t: Throwable => {
           val serviceTime = (System.currentTimeMillis() - taskStart).toInt
           val metrics = attemptedTask.flatMap(t => t.metrics)

core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala

@@ -287,7 +287,8 @@ private[spark] class ClusterScheduler(val sc: SparkContext)
               }
             }
           case None =>
-            logInfo("Ignoring update from TID " + tid + " because its task set is gone")
+            logInfo("Ignoring update with state %s from TID %s because its task set is gone"
+              .format(state, tid))
         }
       } catch {
         case e: Exception => logError("Exception in statusUpdate", e)

core/src/main/scala/org/apache/spark/scheduler/local/LocalTaskSetManager.scala

@@ -21,7 +21,8 @@ import java.nio.ByteBuffer
 import scala.collection.mutable.ArrayBuffer
 import scala.collection.mutable.HashMap
 
-import org.apache.spark.{ExceptionFailure, Logging, SparkEnv, SparkException, Success, TaskState}
+import org.apache.spark.{ExceptionFailure, Logging, SparkEnv, SparkException, Success,
+  TaskEndReason, TaskResultLost, TaskState}
 import org.apache.spark.TaskState.TaskState
 import org.apache.spark.scheduler.{DirectTaskResult, IndirectTaskResult, Pool, Schedulable, Task,
   TaskDescription, TaskInfo, TaskLocality, TaskResult, TaskSet, TaskSetManager}
@@ -144,7 +145,18 @@ private[spark] class LocalTaskSetManager(sched: LocalScheduler, val taskSet: Tas
     val result = ser.deserialize[TaskResult[_]](serializedData, getClass.getClassLoader) match {
       case directResult: DirectTaskResult[_] => directResult
       case IndirectTaskResult(blockId) => {
-        throw new SparkException("Expect only DirectTaskResults when using LocalScheduler")
+        logDebug("Fetching indirect task result for TID %s".format(tid))
+        val serializedTaskResult = env.blockManager.getRemoteBytes(blockId)
+        if (!serializedTaskResult.isDefined) {
+          /* We won't be able to get the task result if the block manager had to flush the
+           * result. */
+          taskFailed(tid, state, serializedData)
+          return
+        }
+        val deserializedResult = ser.deserialize[DirectTaskResult[_]](
+          serializedTaskResult.get)
+        env.blockManager.master.removeBlock(blockId)
+        deserializedResult
       }
     }
     result.metrics.resultSize = serializedData.limit()
@@ -164,18 +176,28 @@ private[spark] class LocalTaskSetManager(sched: LocalScheduler, val taskSet: Tas
     val task = taskSet.tasks(index)
     info.markFailed()
     decreaseRunningTasks(1)
-    val reason: ExceptionFailure = ser.deserialize[ExceptionFailure](
-      serializedData, getClass.getClassLoader)
-    sched.dagScheduler.taskEnded(task, reason, null, null, info, reason.metrics.getOrElse(null))
+    var failureReason = "unknown"
+    ser.deserialize[TaskEndReason](serializedData, getClass.getClassLoader) match {
+      case ef: ExceptionFailure =>
+        failureReason = "Exception failure: %s".format(ef.description)
+        val locs = ef.stackTrace.map(loc => "\tat %s".format(loc.toString))
+        logInfo("Task loss due to %s\n%s\n%s".format(
+          ef.className, ef.description, locs.mkString("\n")))
+        sched.dagScheduler.taskEnded(task, ef, null, null, info, ef.metrics.getOrElse(null))
+
+      case TaskResultLost =>
+        failureReason = "Lost result for TID %s".format(tid)
+        logWarning(failureReason)
+        sched.dagScheduler.taskEnded(task, TaskResultLost, null, null, info, null)
+
+      case _ => {}
+    }
     if (!finished(index)) {
       copiesRunning(index) -= 1
       numFailures(index) += 1
-      val locs = reason.stackTrace.map(loc => "\tat %s".format(loc.toString))
-      logInfo("Loss was due to %s\n%s\n%s".format(
-        reason.className, reason.description, locs.mkString("\n")))
       if (numFailures(index) > MAX_TASK_FAILURES) {
-        val errorMessage = "Task %s:%d failed more than %d times; aborting job %s".format(
-          taskSet.id, index, MAX_TASK_FAILURES, reason.description)
+        val errorMessage = ("Task %s:%d failed more than %d times; aborting job" +
+          "(most recent failure: %s").format(taskSet.id, index, MAX_TASK_FAILURES, failureReason)
         decreaseRunningTasks(runningTasks)
         sched.dagScheduler.taskSetFailed(taskSet, errorMessage)
         // need to delete failed Taskset from schedule queue